runtime

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Oct 3, 2026 License: MIT Imports: 26 Imported by: 0

Documentation

Overview

Package runtime owns application-neutral workflow execution mechanics, including node lifecycle state, readiness, retries, cancellation, and replay.

The StateStore contract persists workflow facts and compare-and-swap claims without exposing SQL or host records. Lifecycle, named readiness rules, durable ready-queue coordination, plan-to-run input binding, completed workflow output binding, classification-safe rendering/event masking, retention hook coordination, generic durable wait coordination, and atomic wait timeout are core-owned. Memoized and caller-pinned outputs use the same typed ValueSetRef data plane as executed/replayed outcomes: memo keys are evaluated under compiler visibility, pins are installed before admission, and reuse completes a claimed node atomically without a synthetic attempt. Host scheduling policy, principal and grant resolution, scheduler adapters, secret authorities, cleanup execution, and concrete storage remain separate concerns.

Package inmemory provides a concurrency-safe process-lifetime implementation for embedded execution and qualification. Hosts that promise crash recovery must bind a durable StateStore implementation instead.

It is extraction-ready engine core. Hadron internal packages and concrete persistence implementations are forbidden.

Index

Constants

View Source 
const (
	// CodeInvalidRunBinding identifies an incomplete plan, run identity, or
	// persistence-independent binding request.
	CodeInvalidRunBinding diagnostic.Code = "HADR-VALUE-010"
	// CodeUnknownWorkflowInput identifies caller data absent from the plan's
	// declared workflow inputs.
	CodeUnknownWorkflowInput diagnostic.Code = "HADR-VALUE-011"
	// CodeMissingWorkflowInput identifies an unsatisfied required declaration.
	CodeMissingWorkflowInput diagnostic.Code = "HADR-VALUE-012"
	// CodeInvalidWorkflowDefault identifies a default that cannot be bound as a
	// literal typed value.
	CodeInvalidWorkflowDefault diagnostic.Code = "HADR-VALUE-013"
	// CodeInvalidWorkflowInput identifies caller/default data that is not a
	// losslessly normalized native JSON value.
	CodeInvalidWorkflowInput diagnostic.Code = "HADR-VALUE-014"
	// CodeWorkflowInputSchema identifies an invalid input schema or a value that
	// violates it.
	CodeWorkflowInputSchema diagnostic.Code = "HADR-VALUE-015"
	// CodeGraphNotComplete identifies output binding attempted before every
	// declared graph node has a canonical terminal observation.
	CodeGraphNotComplete diagnostic.Code = "HADR-VALUE-016"
	// CodeWorkflowOutputBinding identifies an absent, malformed, or failed
	// workflow output binding.
	CodeWorkflowOutputBinding diagnostic.Code = "HADR-VALUE-017"
	// CodeWorkflowOutputSchema identifies an invalid output schema or a bound
	// output that violates it.
	CodeWorkflowOutputSchema diagnostic.Code = "HADR-VALUE-018"
)
View Source 
const (
	EventRunCancellationRequested = "run.cancellation_requested"
	EventCancellationResolved     = "cancellation.resolved"
)
View Source 
const (
	EventCompensationEligible  = "compensation.entry_eligible"
	EventCompensationFrozen    = "compensation.frozen"
	EventCompensationReady     = "compensation.entry_ready"
	EventCompensationFinished  = "compensation.entry_finished"
	EventCompensationCompleted = "compensation.completed"
	EventCompensationCanceled  = "compensation.canceled"

	CompensationHandlerChildRunIDInput      = "compensation.child.run_id"
	CompensationHandlerChildResolutionInput = "compensation.child.resolution"
)
View Source 
const (
	EventSwitchDecided  = "control.switch_decided"
	EventCatchDecided   = "control.catch_decided"
	EventTerminalIntent = "run.terminal_intent_recorded"
)
View Source 
const (
	EventExternalOperationSuspended       = "external_operation.suspended"
	EventExternalOperationObserved        = "external_operation.observed"
	EventExternalOperationCancelRequested = "external_operation.cancel_requested"
)
View Source 
const (
	EventFanOutExpanded  = "fanout.expanded"
	EventFanOutCompleted = "fanout.completed"
)
View Source 
const (
	EventRunStatusChanged    = "run.status_changed"
	EventNodeStatusChanged   = "node.status_changed"
	EventNodeAttemptStarted  = "node.attempt_started"
	EventNodeAttemptFinished = "node.attempt_finished"
)
View Source 
const (
	EventNodeOutcomeReused = "node.outcome_reused"

	CodeMemoMiss     diagnostic.Code = "HADR-RUNTIME-041"
	CodeMemoExpired  diagnostic.Code = "HADR-RUNTIME-042"
	CodeMemoRejected diagnostic.Code = "HADR-RUNTIME-043"
	CodePinRejected  diagnostic.Code = "HADR-RUNTIME-044"
)
View Source 
const (
	// ReasonReadinessWaiting identifies a dependency set that is not resolved.
	ReasonReadinessWaiting = "readiness_waiting"
	// ReasonReadinessUnsatisfied identifies a rule that can no longer become true.
	ReasonReadinessUnsatisfied = "readiness_unsatisfied"
	// ReasonPredicateFalse identifies a ready node skipped by its if predicate.
	ReasonPredicateFalse = "predicate_false"
)
View Source 
const (
	EventRetryScheduled = "retry.scheduled"
	EventRetryActivated = "retry.activated"
	EventRetryCanceled  = "retry.canceled"
)
View Source 
const (
	RetryReasonEligible           = "eligible"
	RetryReasonNoPolicy           = "no_policy"
	RetryReasonAttemptsExhausted  = "attempts_exhausted"
	RetryReasonClassNotSelected   = "class_not_selected"
	RetryReasonPermanentFailure   = "permanent_failure"
	RetryReasonCanceled           = "canceled"
	RetryReasonKindUnsupported    = "kind_unsupported"
	RetryReasonIdempotencyMissing = "idempotency_missing"
	RetryReasonEffectDenied       = "effect_denied"
)
View Source 
const (
	EventServiceSuspended = "service.suspended"
	EventServiceReady     = "service.ready"
	EventServiceStopping  = "service.stopping"
	EventServiceStopped   = "service.stopped"
	EventServiceFailed    = "service.failed"
)
View Source 
const (
	WaitOpen     = workflowwait.StatusOpen
	WaitResumed  = workflowwait.StatusResumed
	WaitTimedOut = workflowwait.StatusTimedOut
	WaitCanceled = workflowwait.StatusCanceled
)
View Source 
const (
	EventWaitSuspended = "wait.suspended"
	EventWaitResumed   = "wait.resumed"
	ResumeValueName    = "resume"
	ActivationWaitWake = "wait_wake"
)
View Source 
const ContinueAsNewExtensionVersion = "reactor/v1"
View Source 
const EventCrashReconciled = "node.crash_reconciled"
View Source 
const (
	EventNodeFinalizeWarning = "node.finalize_warning"
)
View Source 
const EventNodeInputsBound = "node.inputs_bound"
View Source 
const (
	EventNodeVerificationCompleted = "node.verification_completed"
)
View Source 
const EventReplayCreated = "run.replay_created"
View Source 
const EventRunFailFastTriggered = "run.fail_fast_triggered"
View Source 
const EventWaitTimedOut = "wait.timed_out"

EventWaitTimedOut records the wait identity and persisted deadline that caused an invocation attempt to time out.

View Source 
const MaximumReactorRecoveryLimit = 1000
View Source 
const MaximumRunQueryLimit = 1000

Variables

View Source 
var (
	ErrInvalidCancellation     = errors.New("invalid workflow cancellation")
	ErrCancellationUnsupported = errors.New("workflow cancellation is unsupported by executor")
)
View Source 
var (
	ErrInvalidCompensation  = errors.New("invalid compensation request")
	ErrCompensationPending  = errors.New("compensation is pending")
	ErrCompensationConflict = errors.New("compensation state conflict")
)
View Source 
var (
	ErrInvalidControlFlow = errors.New("invalid workflow control flow")
	ErrControlFlowPending = errors.New("workflow control flow is pending")
	// ErrRunOutputsPending reports that cleanup is complete but declared run
	// outputs must be bound before the successful terminal transition.
	ErrRunOutputsPending   = errors.New("workflow run outputs are pending")
	ErrControlFlowConflict = errors.New("workflow control flow conflict")
)
View Source 
var (
	ErrInvalidDispatch = errors.New("invalid step dispatch request")
	ErrStepValidation  = errors.New("step invocation validation failed")
	ErrStepExecution   = errors.New("step invocation execution failed")
)
View Source 
var (
	ErrInvalidFanOut    = errors.New("invalid workflow fan-out")
	ErrFanOutIncomplete = errors.New("workflow fan-out is incomplete")
	ErrFanOutLimit      = errors.New("workflow fan-out concurrency limit reached")
)
View Source 
var (
	ErrInvalidTransition  = errors.New("invalid workflow lifecycle transition")
	ErrTransitionConflict = errors.New("workflow lifecycle transition conflict")
	ErrAttemptConflict    = errors.New("workflow attempt lifecycle conflict")
)
View Source 
var (
	ErrInvalidReuse = errors.New("invalid workflow value reuse")
	ErrReuseDenied  = errors.New("workflow value reuse denied")
)
View Source 
var (
	ErrReactorRolling  = errors.New("workflow reactor is rolling to a new generation")
	ErrReactorBusy     = errors.New("workflow reactor has an applying delivery")
	ErrReactorTerminal = errors.New("workflow reactor is terminal")
)
View Source 
var (
	ErrInvalidRecovery  = errors.New("invalid workflow recovery request")
	ErrRecoveryConflict = errors.New("workflow recovery conflict")
	ErrRecoveryUnsafe   = errors.New("workflow recovery repetition is unsafe")
)
View Source 
var (
	ErrInvalidRetryPolicy = errors.New("invalid workflow retry policy")
	ErrRetryNotDue        = errors.New("workflow retry activation is not due")
	ErrRetryDenied        = errors.New("workflow retry is not permitted")
)
View Source 
var (
	ErrSignalNotOpen   = errors.New("named workflow signal is not open")
	ErrSignalAmbiguous = errors.New("named workflow signal is ambiguous")
)
View Source 
var (
	ErrInvalidService = errors.New("invalid workflow service")
	ErrServicePending = errors.New("workflow service is not terminal")
)
View Source 
var (
	ErrNotFound            = errors.New("workflow state not found")
	ErrAlreadyExists       = errors.New("workflow state already exists")
	ErrInvalidRecord       = errors.New("invalid workflow state record")
	ErrCASMismatch         = errors.New("workflow state compare-and-swap mismatch")
	ErrIdempotencyConflict = errors.New("workflow idempotency conflict")
	ErrClaimMismatch       = errors.New("workflow claim token or generation mismatch")
	ErrLeaseExpired        = errors.New("workflow claim lease expired")
)
View Source 
var (
	ErrInvalidResumeToken = errors.New("invalid workflow wait resume token")
	ErrWaitClosed         = errors.New("workflow wait is closed")
	ErrWaitUnresumable    = errors.New("workflow wait cannot be resumed")
	ErrWaitRecovery       = errors.New("workflow wait recovery failed")
	ErrWaitWakeNotDue     = errors.New("workflow timer wake is not due")
	ErrWaitWakePending    = errors.New("workflow timer wake precedes timeout")
)
View Source 
var ErrInvalidReadiness = errors.New("invalid workflow readiness input")

ErrInvalidReadiness identifies malformed readiness or progression input.

View Source 
var ErrInvalidReplay = errors.New("invalid workflow replay request")
View Source 
var ErrInvalidRunPolicy = errors.New("invalid workflow run policy")
View Source 
var (
	ErrInvalidSchedulerResource = errors.New("invalid workflow scheduler resource")
)
View Source 
var ErrOutputConflict = errors.New("workflow run output finalization conflict")
View Source 
var ErrRetentionHook = errors.New("workflow retention hook failed")

ErrRetentionHook marks host retention preparation or recording failure.

View Source 
var ErrVerificationConflict = errors.New("durable verification report conflicts with replay")
View Source 
var ErrWaitTimeoutNotDue = errors.New("workflow wait timeout is not due")

ErrWaitTimeoutNotDue identifies a deadline that has not arrived.

Functions

func BuildExpressionContext 

func BuildExpressionContext(ctx context.Context, store StateStore, control ControlFlowStore, workflow graph.Graph, runID RunID) (values.ExpressionContext, error)

BuildExpressionContext reconstructs typed step outputs and errors entirely from durable state. It is safe to call after restart and never unwraps an ArtifactRef or SecretRef into the returned context.

func CalculateBackoff 

func CalculateBackoff(policy graph.BackoffPolicy, failedAttempt int) (time.Duration, error)

CalculateBackoff returns the delay after failedAttempt. It uses checked duration arithmetic and caps the result at max_delay when supplied.

func CatchBinding 

func CatchBinding(ctx context.Context, store StateStore, control ControlFlowStore, id ControlDecisionID) (string, values.ValueSet, error)

CatchBinding loads the exact typed error selected by a catch decision. The returned set is suitable for ExpressionContext.Locals and is defensively copied by StateStore.

func CompensationDependencies 

func CompensationDependencies(workflow graph.Graph) map[string][]string

CompensationDependencies exposes the deterministic reverse dependency projection to production hosts that initiate an authorized manual rollback.

func CompensationEntryID 

func CompensationEntryID(attempt AttemptID) string

func CompensationEvidenceDigest 

func CompensationEvidenceDigest(evidence stepkind.ReversibilityEvidence) (string, error)

func CompensationHandlerInputName 

func CompensationHandlerInputName(namespace CompensationHandlerInputNamespace, sourceName string) (string, error)

CompensationHandlerInputName returns the stable injective property name a handler InputSchema uses for an exact source ValueSet name.

func CompensationHandlers 

func CompensationHandlers(workflow graph.Graph) map[string]struct{}

func CompensationRequestDigest 

func CompensationRequestDigest(input any) (string, error)

CompensationRequestDigest binds idempotency to semantic intent while excluding ephemeral CAS generations and wall-clock timestamps. Those remain mandatory first-apply fences but cannot make a lost-response replay differ.

func DigestSchema 

func DigestSchema(schema graph.Schema) (string, error)

DigestSchema validates and hashes canonical JSON Schema bytes.

func EffectiveDurability 

func EffectiveDurability(workflow graph.Graph) graph.DurabilityMode

EffectiveDurability preserves the durable default for existing plans.

func EncodeAttemptIdentity 

func EncodeAttemptIdentity(id AttemptID) (string, error)

EncodeAttemptIdentity returns a collision-free, transport-safe durable key for an attempt whose opaque identity components may contain delimiters. The encoding is stable and reversible by component, but callers must treat the result as an opaque storage key rather than parse it themselves.

func FanOutFailFastAdmissionAllowed 

func FanOutFailFastAdmissionAllowed(fanOut FanOutSnapshot, children []NodeInvocationSnapshot) (bool, error)

FanOutFailFastAdmissionAllowed reports whether a first attempt may be admitted under the immutable fan-out snapshot. Stores call it while holding their claim transaction/lock so no post-failure first attempt slips through.

func FanOutFailuresTolerated 

func FanOutFailuresTolerated(policy *graph.ToleratedFailurePolicy, failures, total int) (bool, error)

FanOutFailuresTolerated applies count or percentage policy inclusively. Nil policy and an explicit zero tolerance both reject any hard item failure.

func FanOutIteration 

func FanOutIteration(index int) string

FanOutIteration produces a stable identity whose lexical order equals index order for every supported nonnegative int value.

func NewFailureValue 

func NewFailureValue(origin NodeInvocationID, attempt *AttemptID, status NodeStatus, timeout TimeoutKind, failure Failure) (values.Value, error)

NewFailureValue creates the typed, private/run-retained error exposed at steps.<id>.error and fan-out items[].error. Exact json.Number use is retained for the attempt number. Failure is already a persistence-boundary record; callers must not pass raw causes or unredacted secret material in it.

func NewRunFailureValue 

func NewRunFailureValue(runID RunID, status RunStatus, failure Failure) (values.Value, error)

NewRunFailureValue creates the typed originating error retained by an unsuccessful terminal intent that is not attributable to one node attempt, such as an explicit run cancellation.

func ReplayCompensationRiskDigest 

func ReplayCompensationRiskDigest(ctx context.Context, store StateStore, replay ReplayStore, runID RunID) (string, error)

ReplayCompensationRiskDigest returns a stable digest for terminal compensable attempts whose receipt contract left application indeterminate. An empty result proves that no such durable failure fact exists.

func RetryActivation 

func RetryActivation(snapshot RetryActivationSnapshot) workflowwait.Activation

RetryActivation converts the durable retry record to the host scheduler's application-neutral activation envelope.

func SaveValuesWithRetention 

func SaveValuesWithRetention(
	ctx context.Context,
	store StateStore,
	hook RetentionHook,
	request SaveValuesRequest,
) (values.ValueSetRef, error)

SaveValuesWithRetention validates persistence classification, invokes an optional host hook, and performs one StateStore write. A nil hook is valid and means no host lifecycle callback. An AfterPersist failure returns the successfully written immutable reference in RetentionHookError.Ref.

func SemanticallyEqualBindPinRequest 

func SemanticallyEqualBindPinRequest(left, right BindPinRequest) bool

SemanticallyEqualBindPinRequest ignores the target generation because it is an application fence, not immutable pin intent.

func SemanticallyEqualNodeInputRequest 

func SemanticallyEqualNodeInputRequest(left, right BindNodeInputsRequest) bool

SemanticallyEqualNodeInputRequest compares the immutable binding intent; generation and timestamp are fencing/application facts and are ignored on replay after the first transaction advanced the node.

func SemanticallyEqualReuseRequest 

func SemanticallyEqualReuseRequest(left, right ReuseNodeOutputsRequest) bool

SemanticallyEqualReuseRequest ignores generation and At, which fence the first application. Exact replay returns the originally persisted outcome.

func StartBoundRun 

func StartBoundRun(ctx context.Context, store StateStore, run BoundRun, startIdempotencyKey string) (RunSnapshot, IdempotencyOutcome, error)

StartBoundRun delegates exact start replay and conflict decisions to StateStore.CreateRun. Callers must reuse the same BoundRun for retries.

func TimerActivation 

func TimerActivation(snapshot WaitSnapshot) workflowwait.Activation

TimerActivation converts a durable successful timer into the host scheduler envelope. Identity depends only on immutable wait ID plus activation kind.

func ValidateCompensationChildRunID 

func ValidateCompensationChildRunID(runID RunID) error

func ValidateCompensationEntryDependencies 

func ValidateCompensationEntryDependencies(entries []CompensationEntrySnapshot) error

ValidateCompensationEntryDependencies rejects malformed durable prerequisite graphs supplied directly through CompensationStore, independently of graph compiler validation.

func ValidateCompensationHandlerNodeID 

func ValidateCompensationHandlerNodeID(nodeID string) error

func ValidateCompensationTerminalEvidence 

func ValidateCompensationTerminalEvidence(status NodeStatus, originalError values.ValueSet) error

ValidateCompensationTerminalEvidence binds the original typed error evidence to the exact terminal result of the forward effect attempt.

func ValidateControlErrorValues 

func ValidateControlErrorValues(input values.ValueSet) error

ValidateControlErrorValues validates the exact typed error envelope accepted by ControlFlowStore. It is exported so extraction hosts and persistence adapters reject corrupt or caller-crafted transport records consistently.

func ValidateFanOutTolerance 

func ValidateFanOutTolerance(policy *graph.ToleratedFailurePolicy) error

ValidateFanOutTolerance requires a single count or percentage tolerance.

func ValidateMemoizableValueSet 

func ValidateMemoizableValueSet(set values.ValueSet) error

ValidateMemoizableValueSet rejects classifications or retention modes that cannot outlive the source attempt as a reusable cache publication.

func ValidateNodeControlErrorValues 

func ValidateNodeControlErrorValues(input values.ValueSet, origin NodeInvocationID, attempt *AttemptID, status NodeStatus, expectedFailure *Failure) error

ValidateNodeControlErrorValues binds a structurally valid typed error to the exact durable node fact that owns it. Stores call this after loading the source so a caller cannot attach another node's otherwise-valid error.

func ValidateNodeStatusTransition 

func ValidateNodeStatusTransition(from, to NodeStatus) error

ValidateNodeStatusTransition validates an applied generic node status edge. Execution-only edges are owned by StartNodeAttempt and FinishNodeAttempt.

func ValidatePinnableValueSet 

func ValidatePinnableValueSet(set values.ValueSet, sameRun bool) error

ValidatePinnableValueSet enforces retention at the durable pin boundary. Run-retained outputs may only be pinned inside their source run.

func ValidateRetryPolicy 

func ValidateRetryPolicy(policy graph.RetryPolicy) error

ValidateRetryPolicy rejects malformed attempts, class filters, and duration arithmetic before a failure can mutate durable state.

func ValidateRunControlErrorValues 

func ValidateRunControlErrorValues(input values.ValueSet, runID RunID, status RunStatus) error

ValidateRunControlErrorValues binds a typed run error to the exact terminal outcome requested by a durable terminal intent.

func ValidateRunStatusTransition 

func ValidateRunStatusTransition(from, to RunStatus) error

ValidateRunStatusTransition validates an applied run status edge. Same-state semantic replay is evaluated by the store after CAS.

Types

type ActivateCompensationEntryRequest 

type ActivateCompensationEntryRequest struct {
	RunID                    RunID
	EntryID                  string
	ExpectedLedgerGeneration uint64
	ExpectedEntryGeneration  uint64
	Inputs                   values.ValueSet
	ChildResolution          CompensationChildResolution
	At                       time.Time
}

type ActivateCompensationEntryResult 

type ActivateCompensationEntryResult struct {
	Ledger CompensationLedgerSnapshot
	Entry  CompensationEntrySnapshot
	Node   NodeInvocationSnapshot
}

type ActivateNodeRetryRequest 

type ActivateNodeRetryRequest struct {
	ActivationID                 string
	ExpectedActivationGeneration uint64
	ExpectedNodeGeneration       uint64
	IdempotencyKey               string
	Now                          time.Time
}

ActivateNodeRetryRequest fires a due activation exactly once. IdempotencyKey distinguishes same-intent replay from conflicting activation attempts.

func (ActivateNodeRetryRequest) Validate 

func (r ActivateNodeRetryRequest) Validate() error

type ActivateNodeRetryResult 

type ActivateNodeRetryResult struct {
	Outcome    IdempotencyOutcome
	Activation RetryActivationSnapshot
	Node       NodeInvocationSnapshot
	Event      *Event
}

type AdmitNodeRequest 

type AdmitNodeRequest struct {
	Claim        ClaimNodeRequest               `json:"claim"`
	Requirements []SchedulerResourceRequirement `json:"requirements"`
	Priority     int                            `json:"priority,omitempty"`
	EnqueuedAt   time.Time                      `json:"enqueued_at"`
}

AdmitNodeRequest atomically acquires a node claim and all requirements, or records a diagnostic waiter without acquiring either. Claim idempotency and requirements are one immutable request.

func (AdmitNodeRequest) Validate 

func (r AdmitNodeRequest) Validate() error

type AdmitNodeResult 

type AdmitNodeResult struct {
	Claim   ClaimResult           `json:"claim"`
	Blocked []SchedulerResourceID `json:"blocked,omitempty"`
}

type AppendEventRequest 

type AppendEventRequest struct {
	RunID      RunID
	Invocation *NodeInvocationID
	Attempt    *AttemptID
	Type       string
	OccurredAt time.Time
	Attributes map[string]string
	Values     *values.ValueSetRef
	Redaction  values.RedactionClass
	Retention  values.RetentionClass
}

AppendEventRequest supplies an event without a store-assigned sequence.

func MaskEventRequest 

func MaskEventRequest(request AppendEventRequest, redactor *values.Redactor) AppendEventRequest

MaskEventRequest defensively copies an event request and masks all attribute values with known resolved material before persistence. No redactor or secret material is retained in the result.

type ApplyExternalOperationRequest 

type ApplyExternalOperationRequest struct {
	Attempt                     AttemptID
	ExpectedOperationGeneration uint64
	ExpectedNodeGeneration      uint64
	ExpectedAttemptGeneration   uint64
	Status                      stepkind.ObservationState
	Progress                    map[string]string
	Outputs                     *values.ValueSetRef
	Failure                     *Failure
	Verification                *ExternalVerificationCompletion
	NextNodeStatus              NodeStatus
	ObservedAt                  time.Time
	HeartbeatAt                 time.Time
	At                          time.Time
}

ApplyExternalOperationRequest atomically records one observation. Pending observations update only operational metadata. Terminal observations also close the unfinished attempt and set the aggregate node terminal or ready for an injected retry policy.

func (ApplyExternalOperationRequest) Validate 

func (r ApplyExternalOperationRequest) Validate() error

type ApplyExternalOperationResult 

type ApplyExternalOperationResult struct {
	Operation ExternalOperationSnapshot
	Node      NodeInvocationSnapshot
	Attempt   AttemptSnapshot
	Events    []Event
}

type ApplyRunFailurePolicyRequest 

type ApplyRunFailurePolicyRequest struct {
	RunID                    RunID            `json:"run_id"`
	ExpectedRunGeneration    uint64           `json:"expected_run_generation"`
	Trigger                  NodeInvocationID `json:"trigger"`
	ExpectedSourceGeneration uint64           `json:"expected_source_generation"`
	IntendedStatus           RunStatus        `json:"intended_status"`
	Reason                   Failure          `json:"reason"`
	ErrorValues              values.ValueSet  `json:"error_values"`
	CompensationRequired     bool             `json:"compensation_required,omitempty"`
	IdempotencyKey           string           `json:"idempotency_key"`
	Finalizers               []FinalizerScope `json:"finalizers"`
	At                       time.Time        `json:"at"`
}

func (ApplyRunFailurePolicyRequest) Validate 

func (r ApplyRunFailurePolicyRequest) Validate() error

type ApplyRunFailurePolicyResult 

type ApplyRunFailurePolicyResult struct {
	Disposition RunFailureDisposition        `json:"disposition"`
	Decision    RunPolicyDecisionSnapshot    `json:"decision"`
	Run         RunSnapshot                  `json:"run"`
	Intent      TerminalIntentSnapshot       `json:"intent"`
	Nodes       []NodeInvocationSnapshot     `json:"nodes,omitempty"`
	Intents     []CancellationIntentSnapshot `json:"cancellation_intents,omitempty"`
	Events      []Event                      `json:"events,omitempty"`
}

type ApplyServiceHeartbeatRequest 

type ApplyServiceHeartbeatRequest struct {
	Start                     NodeInvocationID
	ExpectedServiceGeneration uint64
	ObservedAt                time.Time
	HeartbeatAt               time.Time
	Failure                   *Failure
	At                        time.Time
}

ApplyServiceHeartbeatRequest records ready-service liveness or an unexpected terminal provider failure without rewriting the succeeded start node. The returned failed service remains visible for finalizer recovery.

type ApplyServiceReadyRequest 

type ApplyServiceReadyRequest struct {
	Start                     AttemptID
	ExpectedServiceGeneration uint64
	ExpectedNodeGeneration    uint64
	ExpectedAttemptGeneration uint64
	Ready                     bool
	Outputs                   *values.ValueSetRef
	Failure                   *Failure
	ObservedAt                time.Time
	HeartbeatAt               time.Time
	At                        time.Time
}

ApplyServiceReadyRequest records either another starting observation or the readiness milestone. Outputs is required only when Ready is true.

type ApplyServiceReadyResult 

type ApplyServiceReadyResult struct {
	Service ServiceSnapshot
	Node    NodeInvocationSnapshot
	Attempt AttemptSnapshot
	Events  []Event
}

type ApplyServiceStopRequest 

type ApplyServiceStopRequest struct {
	Start                     NodeInvocationID
	ExpectedServiceGeneration uint64
	ExpectedNodeGeneration    uint64
	ExpectedAttemptGeneration uint64
	Stopped                   bool
	Failure                   *Failure
	ObservedAt                time.Time
	HeartbeatAt               time.Time
	At                        time.Time
}

ApplyServiceStopRequest records another pending stop observation or closes the generated teardown attempt and service record.

type ApplyServiceStopResult 

type ApplyServiceStopResult struct {
	Service ServiceSnapshot
	Node    NodeInvocationSnapshot
	Attempt AttemptSnapshot
	Events  []Event
}

type AttemptCanceler 

type AttemptCanceler interface {
	CancelAttempt(context.Context, AttemptSnapshot) error
}

AttemptCanceler signals an in-process or host-managed context execution. Success means execution ownership has stopped and the durable attempt may be closed. Calls happen outside StateStore transactions.

type AttemptConflictError 

type AttemptConflictError struct {
	Invocation NodeInvocationID
	Attempt    int
	Reason     string
}

AttemptConflictError describes an incoherent attempt history or operation.

func (*AttemptConflictError) Error 

func (e *AttemptConflictError) Error() string

func (*AttemptConflictError) Unwrap 

func (e *AttemptConflictError) Unwrap() error

Unwrap supports errors.Is(err, ErrAttemptConflict).

type AttemptID 

type AttemptID struct {
	Invocation NodeInvocationID `json:"invocation"`
	Number     int              `json:"number"`
}

AttemptID identifies one numbered execution attempt.

func (AttemptID) Validate 

func (id AttemptID) Validate() error

Validate reports malformed attempt identity.

type AttemptSnapshot 

type AttemptSnapshot struct {
	ID         AttemptID           `json:"id"`
	Status     NodeStatus          `json:"status"`
	Executor   ExecutorMetadata    `json:"executor"`
	Inputs     *values.ValueSetRef `json:"inputs,omitempty"`
	Outputs    *values.ValueSetRef `json:"outputs,omitempty"`
	Failure    *Failure            `json:"failure,omitempty"`
	StartedAt  time.Time           `json:"started_at"`
	FinishedAt time.Time           `json:"finished_at,omitempty"`
	Generation uint64              `json:"generation"`
	CreatedAt  time.Time           `json:"created_at"`
	UpdatedAt  time.Time           `json:"updated_at"`
}

AttemptSnapshot records one execution attempt without defining which status transitions are legal.

func (AttemptSnapshot) Validate 

func (s AttemptSnapshot) Validate() error

Validate checks attempt record integrity.

type BeginManualCompensationRequest 

type BeginManualCompensationRequest struct {
	RunID                 RunID
	PlanDigest            string
	ExpectedRunGeneration uint64
	OriginalStatus        RunStatus
	Dependencies          map[string][]string
	IdempotencyKey        string
	Authorization         string
	At                    time.Time
}

BeginManualCompensationRequest freezes a collecting ledger after the original run has already reached an immutable terminal state. Authorization is a non-secret durable policy attestation supplied by the host.

type BeginReactorDeliveryRequest 

type BeginReactorDeliveryRequest struct {
	Identity            ReactorIdentity        `json:"identity"`
	InitialRunID        RunID                  `json:"initial_run_id"`
	ContinueAfterEvents uint64                 `json:"continue_after_events"`
	Delivery            ReactorDeliveryRequest `json:"delivery"`
	At                  time.Time              `json:"at"`
}

type BeginReplayRequest 

type BeginReplayRequest struct {
	Provenance               ReplayProvenance      `json:"provenance"`
	Plan                     PlanRef               `json:"plan"`
	Inputs                   *values.ValueSetRef   `json:"inputs,omitempty"`
	ExpectedSourceGeneration uint64                `json:"expected_source_generation"`
	Nodes                    []ReplayNodeBinding   `json:"nodes"`
	FanOuts                  []ReplayFanOutBinding `json:"fan_outs,omitempty"`
}

func (BeginReplayRequest) Validate 

func (r BeginReplayRequest) Validate() error

type BeginReplayResult 

type BeginReplayResult struct {
	Outcome    IdempotencyOutcome       `json:"outcome"`
	Run        RunSnapshot              `json:"run"`
	Provenance ReplayProvenance         `json:"provenance"`
	Nodes      []NodeInvocationSnapshot `json:"nodes"`
	Event      Event                    `json:"event"`
}

type BeginRunUpdateRequest 

type BeginRunUpdateRequest struct {
	IdempotencyKey string                 `json:"idempotency_key"`
	Selector       SignalSelector         `json:"selector"`
	WaitID         WaitID                 `json:"wait_id"`
	Responder      workflowwait.Responder `json:"responder"`
	Payload        values.Value           `json:"payload"`
	ReceivedAt     time.Time              `json:"received_at"`
}

func (BeginRunUpdateRequest) Validate 

func (r BeginRunUpdateRequest) Validate() error

type BeginTerminalIntentRequest 

type BeginTerminalIntentRequest struct {
	RunID                 RunID
	ExpectedRunGeneration uint64
	IntendedStatus        RunStatus
	// SuccessOutputsRequired must be derived from the exact graph and may be
	// true only for a successful intent with declared workflow outputs.
	SuccessOutputsRequired bool
	CompensationRequired   bool
	Reason                 *Failure
	// ErrorValues is atomically persisted and bound for unsuccessful intent.
	ErrorValues    values.ValueSet
	IdempotencyKey string
	Finalizers     []FinalizerScope
	At             time.Time
}

type BeginTerminalIntentResult 

type BeginTerminalIntentResult struct {
	Outcome IdempotencyOutcome
	Run     RunSnapshot
	Intent  TerminalIntentSnapshot
	Event   *Event
}

type BindNodeInputsRequest 

type BindNodeInputsRequest struct {
	InvocationID       NodeInvocationID `json:"invocation_id"`
	ExpectedGeneration uint64           `json:"expected_generation"`
	IdempotencyKey     string           `json:"idempotency_key"`
	Values             values.ValueSet  `json:"values"`
	MemoKeyDigest      string           `json:"memo_key_digest,omitempty"`
	At                 time.Time        `json:"at"`
}

BindNodeInputsRequest atomically persists one evaluated typed input set and attaches its immutable reference to a still-unclaimable invocation. At and ExpectedGeneration fence the first application; exact replay is identified by IdempotencyKey and the complete typed value-set digest.

func (BindNodeInputsRequest) Validate 

func (r BindNodeInputsRequest) Validate() error

type BindNodeInputsResult 

type BindNodeInputsResult struct {
	Outcome IdempotencyOutcome     `json:"outcome"`
	Node    NodeInvocationSnapshot `json:"node"`
	Inputs  values.ValueSetRef     `json:"inputs"`
	Event   Event                  `json:"event"`
}

type BindPinRequest 

type BindPinRequest struct {
	Binding            PinBinding `json:"binding"`
	ExpectedGeneration uint64     `json:"expected_generation"`
	IdempotencyKey     string     `json:"idempotency_key"`
}

func (BindPinRequest) Validate 

func (r BindPinRequest) Validate() error

type BindPinResult 

type BindPinResult struct {
	Outcome IdempotencyOutcome     `json:"outcome"`
	Binding PinBinding             `json:"binding"`
	Node    NodeInvocationSnapshot `json:"node"`
}

type BindRunRequest 

type BindRunRequest struct {
	ID            RunID
	Plan          *compile.ExecutionPlan
	Inputs        map[string]any
	CreatedAt     time.Time
	RetentionHook RetentionHook
}

BindRunRequest supplies immutable plan identity and caller data. Inputs are losslessly normalized native JSON values; strings are never parsed into numbers, booleans, null, arrays, or objects. RetentionHook is optional.

type BindRunResult 

type BindRunResult struct {
	Run         *BoundRun
	Diagnostics []diagnostic.Diagnostic
}

BindRunResult contains a BoundRun only when all diagnostics are absent and the complete input set was saved successfully.

func BindRun 

func BindRun(ctx context.Context, store StateStore, request BindRunRequest) (BindRunResult, error)

BindRun validates all caller inputs and defaults before performing exactly one SaveValues call. SaveValues has no idempotency contract: rebinding is a new release operation and may create another unreferenced immutable value set. A start retry must reuse the returned BoundRun and InputsRef.

type BlockedReason 

type BlockedReason struct {
	Code         string             `json:"code"`
	Message      string             `json:"message"`
	Dependencies []NodeInvocationID `json:"dependencies,omitempty"`
	Details      map[string]string  `json:"details,omitempty"`
}

BlockedReason preserves explainable readiness failure without overloading a pending status.

func (BlockedReason) Validate 

func (r BlockedReason) Validate() error

Validate reports malformed structured blocked metadata.

type BoundRun 

type BoundRun struct {
	ID         RunID              `json:"id"`
	Plan       PlanRef            `json:"plan"`
	InputsRef  values.ValueSetRef `json:"inputs_ref"`
	CreatedAt  time.Time          `json:"created_at"`
	Provenance graph.Provenance   `json:"provenance"`
}

BoundRun is an immutable release artifact between an ExecutionPlan and a persisted RunSnapshot. InputsRef names one validated, complete value set; its existence alone does not mean the run has started. Plan retains the exact plan ID, version, digest, and serialized schema version used at bind time. Provenance is copied from that immutable plan.

func (BoundRun) Validate 

func (r BoundRun) Validate() error

Validate defensively checks the extraction-safe BoundRun envelope. It does not resolve, load, or start the run.

type CASMismatchError 

type CASMismatchError struct {
	Resource string
	Expected uint64
	Actual   uint64
}

CASMismatchError reports the expected and current record generation.

func (*CASMismatchError) Error 

func (e *CASMismatchError) Error() string

func (*CASMismatchError) Unwrap 

func (e *CASMismatchError) Unwrap() error

Unwrap supports errors.Is(err, ErrCASMismatch).

type CacheEntry 

type CacheEntry struct {
	Key         string             `json:"key"`
	PlanDigest  string             `json:"plan_digest"`
	NodeID      string             `json:"node_id"`
	InputDigest string             `json:"input_digest"`
	Outputs     values.ValueSetRef `json:"outputs"`
	CreatedAt   time.Time          `json:"created_at"`
	ExpiresAt   time.Time          `json:"expires_at,omitempty"`
}

CacheEntry references cached node outputs without embedding values.

func (CacheEntry) Validate 

func (e CacheEntry) Validate() error

Validate reports malformed cache metadata.

type CancelCompensationRequest 

type CancelCompensationRequest struct {
	RunID                    RunID
	ExpectedLedgerGeneration uint64
	IdempotencyKey           string
	Reason                   string
	At                       time.Time
}

type CancellationCoordinator 

type CancellationCoordinator struct {
	Store    StateStore
	Registry stepkind.Registry
	External *ExternalOperationCoordinator
	Attempts AttemptCanceler
	Children ChildRunCanceler
	Now      func() time.Time
}

CancellationCoordinator replays durable cancellation work without keeping correctness-critical process state.

func (CancellationCoordinator) Recover 

func (c CancellationCoordinator) Recover(ctx context.Context, query CancellationIntentQuery) ([]CancellationIntentSnapshot, []error, error)

Recover replays pending I/O intents in deterministic store order.

func (CancellationCoordinator) Request 

func (c CancellationCoordinator) Request(ctx context.Context, request RequestRunCancellationRequest) (RequestRunCancellationResult, []error, error)

Request commits run cancellation before performing any host or adapter I/O.

type CancellationDescendantGraph 

type CancellationDescendantGraph struct {
	Run   RunSnapshot
	Graph graph.Graph
}

CancellationDescendantGraph supplies the exact stored graph and current run snapshot used to plan one direct-cancel descendant outside the store transaction. The store independently validates the complete reachable set.

type CancellationDescendantPlan 

type CancellationDescendantPlan struct {
	RunID                 RunID
	ExpectedRunGeneration uint64
	IdempotencyKey        string
	Finalizers            []FinalizerScope
	CompensationRequired  bool
	ErrorValues           values.ValueSet
}

CancellationDescendantPlan is one locally owned direct-cancel descendant in an atomic cancellation tree. ErrorValues is required exactly when Finalizers is non-empty and is store-bound to that descendant's terminal intent.

func (CancellationDescendantPlan) Validate 

func (p CancellationDescendantPlan) Validate(reason Failure) error

Validate reports malformed descendant plan metadata independently of the store-owned reachable-child and generation checks.

type CancellationIntentKind 

type CancellationIntentKind string

CancellationIntentKind describes durable work that cannot be completed inside the run-cancellation transaction because it requires adapter or host I/O. 

const (
	CancellationRunningAttempt    CancellationIntentKind = "running_attempt"
	CancellationExternalOperation CancellationIntentKind = "external_operation"
	CancellationChildRun          CancellationIntentKind = "child_run"
)

func (CancellationIntentKind) Valid 

func (k CancellationIntentKind) Valid() bool

type CancellationIntentQuery 

type CancellationIntentQuery struct {
	RunID RunID
	Limit int
}

type CancellationIntentSnapshot 

type CancellationIntentSnapshot struct {
	ID          string                   `json:"id"`
	RunID       RunID                    `json:"run_id"`
	Kind        CancellationIntentKind   `json:"kind"`
	Attempt     *AttemptID               `json:"attempt,omitempty"`
	ChildRunID  RunID                    `json:"child_run_id,omitempty"`
	ChildPolicy graph.ParentClosePolicy  `json:"child_policy,omitempty"`
	Status      CancellationIntentStatus `json:"status"`
	Generation  uint64                   `json:"generation"`
	RequestedAt time.Time                `json:"requested_at"`
	ResolvedAt  time.Time                `json:"resolved_at,omitempty"`
	UpdatedAt   time.Time                `json:"updated_at"`
}

CancellationIntentSnapshot is restart-durable and remains pending while an explicit or unsupported remote cancellation is unresolved. Attempt and ChildRunID are mutually exclusive according to Kind.

func (CancellationIntentSnapshot) Validate 

func (s CancellationIntentSnapshot) Validate() error

type CancellationIntentStatus 

type CancellationIntentStatus string
const (
	CancellationPending  CancellationIntentStatus = "pending"
	CancellationResolved CancellationIntentStatus = "resolved"
)

func (CancellationIntentStatus) Valid 

func (s CancellationIntentStatus) Valid() bool

type CancellationStore 

type CancellationStore interface {
	RecordChildRun(context.Context, ChildRunLink) error
	ListChildRuns(context.Context, RunID) ([]ChildRunLink, error)
	RequestRunCancellation(context.Context, RequestRunCancellationRequest) (RequestRunCancellationResult, error)
	ResolveCancellationIntent(context.Context, ResolveCancellationIntentRequest) (ResolveCancellationIntentResult, error)
	RecoverCancellationIntents(context.Context, CancellationIntentQuery) ([]CancellationIntentSnapshot, error)
}

CancellationStore is the atomic durable cancellation surface.

type ChildRunCanceler 

type ChildRunCanceler interface {
	RequestChildRunCancellation(context.Context, ChildRunLink) error
}

ChildRunCanceler requests cancellation from a separately hosted child run. It is used only for request_cancel links; direct cancel links are propagated inside the owning StateStore. 

type ChildRunLink struct {
	ParentRunID RunID                   `json:"parent_run_id"`
	Invocation  NodeInvocationID        `json:"invocation"`
	ChildRunID  RunID                   `json:"child_run_id"`
	Policy      graph.ParentClosePolicy `json:"policy"`
	CreatedAt   time.Time               `json:"created_at"`
}

ChildRunLink is the immutable relation used to propagate parent close.

func (ChildRunLink) Validate 

func (l ChildRunLink) Validate() error

type ChildTerminalWait 

type ChildTerminalWait struct {
	Link  ChildRunLink `json:"link"`
	Child RunSnapshot  `json:"child"`
	Wait  WaitSnapshot `json:"wait"`
}

ChildTerminalWait is a bounded recovery candidate for a terminal child and its still-open parent wait. The host turns it into the adapter's typed terminal envelope and uses WaitCoordinator.Resume; storage owns no adapter payload or alternate completion state machine.

func (ChildTerminalWait) Validate 

func (c ChildTerminalWait) Validate() error

type ChildTerminalWaitStore 

type ChildTerminalWaitStore interface {
	RecoverChildTerminalWaits(context.Context, int) ([]ChildTerminalWait, error)
}

ChildTerminalWaitStore exposes only eligible terminal-child/open-wait pairs. Limit zero means the store's bounded default/all-current page.

type ClaimLease 

type ClaimLease struct {
	Owner      string    `json:"owner"`
	Token      string    `json:"token"`
	Generation uint64    `json:"generation"`
	ExpiresAt  time.Time `json:"expires_at"`
}

ClaimLease is an opaque token plus monotonic generation proving current ownership until ExpiresAt. Tokens are never returned to contending claimants.

func (ClaimLease) Validate 

func (l ClaimLease) Validate() error

Validate reports malformed lease metadata.

type ClaimNodeRequest 

type ClaimNodeRequest struct {
	InvocationID            NodeInvocationID
	ExpectedClaimGeneration uint64
	Owner                   string
	Token                   string
	IdempotencyKey          string
	Now                     time.Time
	LeaseUntil              time.Time
}

ClaimNodeRequest atomically attempts to acquire a ready node only when ExpectedClaimGeneration matches. A matching request against a non-ready node or a live lease durably returns Acquired false; an expired ready lease may be replaced. Now is caller-supplied for deterministic storage adapters and tests; LeaseUntil must be later than Now.

type ClaimProof 

type ClaimProof struct {
	Owner      string
	Token      string
	Generation uint64
}

ClaimProof fences lifecycle mutations to the current lease owner.

func (ClaimProof) Validate 

func (p ClaimProof) Validate() error

Validate reports incomplete claim proof.

type ClaimReactorDeliveryRequest 

type ClaimReactorDeliveryRequest struct {
	ReactorID          string    `json:"reactor_id"`
	IdempotencyKey     string    `json:"idempotency_key"`
	ExpectedGeneration uint64    `json:"expected_generation"`
	WaitID             WaitID    `json:"wait_id,omitempty"`
	At                 time.Time `json:"at"`
}

type ClaimResult 

type ClaimResult struct {
	Acquired bool
	Replayed bool
	Lease    *ClaimLease
}

ClaimResult never exposes another claimant's lease.

type CompensationChildResolution 

type CompensationChildResolution string

CompensationChildResolution is the immutable outcome observed before a parent-side handler is admitted. NoLedger is distinct from a child ledger that ran and succeeded. 

const (
	CompensationChildNoLedger  CompensationChildResolution = "no_ledger"
	CompensationChildSucceeded CompensationChildResolution = "succeeded"
	CompensationChildPartial   CompensationChildResolution = "partial"
	CompensationChildFailed    CompensationChildResolution = "failed"
	CompensationChildCanceled  CompensationChildResolution = "canceled"
)

func (CompensationChildResolution) Valid 

func (r CompensationChildResolution) Valid() bool

type CompensationCoordinator 

type CompensationCoordinator struct {
	Store        StateStore
	Compensation CompensationStore
	Plans        RecoveryPlanSource
}

CompensationCoordinator reconstructs rollback scheduling solely from the durable ledger. It never executes handlers itself and therefore preserves the ordinary claim, retry, policy, timeout, and executor path.

func (CompensationCoordinator) Progress 

func (c CompensationCoordinator) Progress(ctx context.Context, runID RunID, at time.Time) (CompensationProgressResult, error)

Progress seals terminal handlers, then activates every reverse-dependency eligible entry. Independent entries become Ready in the same pass.

type CompensationCycle 

type CompensationCycle struct {
	Number       int                 `json:"number"`
	Attestation  string              `json:"attestation,omitempty"`
	CancelReason string              `json:"cancel_reason,omitempty"`
	Outcome      CompensationOutcome `json:"outcome,omitempty"`
	StartedAt    time.Time           `json:"started_at"`
	CompletedAt  time.Time           `json:"completed_at,omitempty"`
}

type CompensationEligibility 

type CompensationEligibility struct {
	PlanDigest      string
	HandlerNodeID   string
	Evidence        stepkind.ReversibilityEvidence
	Receipt         stepkind.CompensationReceipt
	OriginalOutputs values.ValueSet
	OriginalError   values.ValueSet
	ChildRunID      RunID
}

type CompensationEntryHistory 

type CompensationEntryHistory struct {
	Cycle           int                         `json:"cycle"`
	Handler         NodeInvocationID            `json:"handler"`
	Status          CompensationEntryStatus     `json:"status"`
	ChildResolution CompensationChildResolution `json:"child_resolution,omitempty"`
	Outputs         *values.ValueSetRef         `json:"outputs,omitempty"`
	Failure         *Failure                    `json:"failure,omitempty"`
	CompletedAt     time.Time                   `json:"completed_at"`
}

type CompensationEntrySnapshot 

type CompensationEntrySnapshot struct {
	ID              string                      `json:"id"`
	RunID           RunID                       `json:"run_id"`
	PlanDigest      string                      `json:"plan_digest"`
	Source          NodeInvocationID            `json:"source"`
	SourceAttempt   AttemptID                   `json:"source_attempt"`
	Handler         NodeInvocationID            `json:"handler"`
	Status          CompensationEntryStatus     `json:"status"`
	Operation       string                      `json:"operation"`
	EvidenceDigest  string                      `json:"evidence_digest"`
	OriginalInputs  *values.ValueSetRef         `json:"original_inputs,omitempty"`
	OriginalOutputs *values.ValueSetRef         `json:"original_outputs,omitempty"`
	OriginalError   *values.ValueSetRef         `json:"original_error,omitempty"`
	Receipt         values.ValueSetRef          `json:"receipt"`
	Prerequisites   []string                    `json:"prerequisites,omitempty"`
	HandlerOutputs  *values.ValueSetRef         `json:"handler_outputs,omitempty"`
	HandlerFailure  *Failure                    `json:"handler_failure,omitempty"`
	ChildRunID      RunID                       `json:"child_run_id,omitempty"`
	ChildResolution CompensationChildResolution `json:"child_resolution,omitempty"`
	History         []CompensationEntryHistory  `json:"history,omitempty"`
	Generation      uint64                      `json:"generation"`
	CreatedAt       time.Time                   `json:"created_at"`
	UpdatedAt       time.Time                   `json:"updated_at"`
	CompletedAt     time.Time                   `json:"completed_at,omitempty"`
}

CompensationEntrySnapshot binds one actual applied forward effect to one dormant handler invocation and immutable typed evidence.

func (CompensationEntrySnapshot) Validate 

func (s CompensationEntrySnapshot) Validate() error

type CompensationEntryStatus 

type CompensationEntryStatus string
const (
	CompensationEligible  CompensationEntryStatus = "eligible"
	CompensationPending   CompensationEntryStatus = "pending"
	CompensationActive    CompensationEntryStatus = "active"
	CompensationSucceeded CompensationEntryStatus = "succeeded"
	CompensationPartial   CompensationEntryStatus = "partial"
	CompensationFailed    CompensationEntryStatus = "failed"
	CompensationCanceled  CompensationEntryStatus = "canceled"
)

func (CompensationEntryStatus) Terminal 

func (s CompensationEntryStatus) Terminal() bool

func (CompensationEntryStatus) Valid 

func (s CompensationEntryStatus) Valid() bool

type CompensationHandlerInputNamespace 

type CompensationHandlerInputNamespace string

CompensationHandlerInputNamespace identifies one typed, durable evidence envelope made available to a compensation handler. 

const (
	CompensationHandlerOriginalInputs  CompensationHandlerInputNamespace = "original.inputs"
	CompensationHandlerOriginalOutputs CompensationHandlerInputNamespace = "original.outputs"
	CompensationHandlerOriginalError   CompensationHandlerInputNamespace = "original.error"
	CompensationHandlerReceipt         CompensationHandlerInputNamespace = "receipt"
)

type CompensationLedgerSnapshot 

type CompensationLedgerSnapshot struct {
	RunID           RunID                     `json:"run_id"`
	PlanDigest      string                    `json:"plan_digest"`
	Status          CompensationLedgerStatus  `json:"status"`
	Outcome         CompensationOutcome       `json:"outcome,omitempty"`
	Trigger         graph.CompensationTrigger `json:"trigger,omitempty"`
	OriginalStatus  RunStatus                 `json:"original_status,omitempty"`
	OriginalFailure *values.ValueSetRef       `json:"original_failure,omitempty"`
	CancelReason    string                    `json:"cancel_reason,omitempty"`
	Cycles          []CompensationCycle       `json:"cycles,omitempty"`
	Generation      uint64                    `json:"generation"`
	CreatedAt       time.Time                 `json:"created_at"`
	UpdatedAt       time.Time                 `json:"updated_at"`
	CompletedAt     time.Time                 `json:"completed_at,omitempty"`
}

CompensationLedgerSnapshot is a separate run-owned rollback projection. OriginalStatus/Failure preserve the forward terminal intent and are never rewritten by handler outcomes.

func (CompensationLedgerSnapshot) Validate 

func (s CompensationLedgerSnapshot) Validate() error

type CompensationLedgerStatus 

type CompensationLedgerStatus string
const (
	CompensationCollecting CompensationLedgerStatus = "collecting"
	CompensationFrozen     CompensationLedgerStatus = "frozen"
	CompensationRunning    CompensationLedgerStatus = "running"
	CompensationTerminal   CompensationLedgerStatus = "terminal"
)

func (CompensationLedgerStatus) Valid 

func (s CompensationLedgerStatus) Valid() bool

type CompensationOutcome 

type CompensationOutcome string
const (
	CompensationOutcomeSucceeded CompensationOutcome = "succeeded"
	CompensationOutcomePartial   CompensationOutcome = "partial"
	CompensationOutcomeFailed    CompensationOutcome = "failed"
	CompensationOutcomeCanceled  CompensationOutcome = "canceled"
)

func (CompensationOutcome) Valid 

func (o CompensationOutcome) Valid() bool

type CompensationProgressResult 

type CompensationProgressResult struct {
	Ledger    CompensationLedgerSnapshot        `json:"ledger"`
	Activated []ActivateCompensationEntryResult `json:"activated,omitempty"`
	Sealed    []SealCompensationEntryResult     `json:"sealed,omitempty"`
}

CompensationProgressResult reports one bounded scheduling pass. Activated handlers are ordinary durable node invocations; sealed entries retain their result separately from the immutable forward outcome.

type CompensationStore 

type CompensationStore interface {
	FinishCompensableAttempt(context.Context, FinishCompensableAttemptRequest) (FinishCompensableAttemptResult, error)
	FreezeCompensation(context.Context, FreezeCompensationRequest) (FreezeCompensationResult, error)
	BeginManualCompensation(context.Context, BeginManualCompensationRequest) (FreezeCompensationResult, error)
	LoadCompensationLedger(context.Context, RunID) (CompensationLedgerSnapshot, error)
	ListCompensationEntries(context.Context, RunID) ([]CompensationEntrySnapshot, error)
	LoadCompensationEntryByHandler(context.Context, NodeInvocationID) (CompensationEntrySnapshot, error)
	ActivateCompensationEntry(context.Context, ActivateCompensationEntryRequest) (ActivateCompensationEntryResult, error)
	FailCompensationEntry(context.Context, FailCompensationEntryRequest) (SealCompensationEntryResult, error)
	SealCompensationEntry(context.Context, SealCompensationEntryRequest) (SealCompensationEntryResult, error)
	CancelCompensation(context.Context, CancelCompensationRequest) (CompensationLedgerSnapshot, error)
	RetryCompensation(context.Context, RetryCompensationRequest) (CompensationLedgerSnapshot, error)
	RecoverCompensation(context.Context, int) ([]CompensationLedgerSnapshot, error)
}

CompensationStore is an optional extension: StateStore and CompleteHost are unchanged. Implementations own the atomic finish+eligibility and ledger+handler materialization transactions.

type CompleteFanOutRequest 

type CompleteFanOutRequest struct {
	Parent                   NodeInvocationID
	ExpectedParentGeneration uint64
	ExpectedFanOutGeneration uint64
	ExpectedChildGenerations map[NodeInvocationID]uint64
	Status                   FanOutStatus
	Outputs                  values.ValueSetRef
	Failure                  *Failure
	At                       time.Time
}

func (CompleteFanOutRequest) Validate 

func (r CompleteFanOutRequest) Validate() error

type CompleteFanOutResult 

type CompleteFanOutResult struct {
	FanOut FanOutSnapshot
	Parent NodeInvocationSnapshot
	Event  Event
}

type CompleteReactorDeliveryRequest 

type CompleteReactorDeliveryRequest struct {
	ReactorID          string                 `json:"reactor_id"`
	IdempotencyKey     string                 `json:"idempotency_key"`
	ExpectedGeneration uint64                 `json:"expected_generation"`
	Status             ReactorDeliveryStatus  `json:"status"`
	Receipt            ReactorDeliveryReceipt `json:"receipt"`
	At                 time.Time              `json:"at"`
}

type CompleteRunUpdateRequest 

type CompleteRunUpdateRequest struct {
	IdempotencyKey     string           `json:"idempotency_key"`
	ExpectedGeneration uint64           `json:"expected_generation"`
	Status             RunUpdateStatus  `json:"status"`
	Receipt            RunUpdateReceipt `json:"receipt"`
	At                 time.Time        `json:"at"`
}

type CompleteTerminalIntentRequest 

type CompleteTerminalIntentRequest struct {
	RunID                    RunID
	ExpectedRunGeneration    uint64
	ExpectedIntentGeneration uint64
	// Outputs names the complete, digest-bound run-outputs record to publish
	// atomically with successful intent completion. Every other outcome forbids it.
	Outputs *values.ValueSetRef
	At      time.Time
}

type CompleteTerminalIntentResult 

type CompleteTerminalIntentResult struct {
	Run    RunSnapshot
	Intent TerminalIntentSnapshot
	Event  Event
}

type ContinueAsNewPolicy 

type ContinueAsNewPolicy struct {
	MaxEvents int      `json:"max_events"`
	Carry     []string `json:"carry,omitempty"`
}

ContinueAsNewPolicy is the typed runtime interpretation of the graph's versioned durability extension. Carry names exact same-named output/input values handed to the next generation.

func ParseContinueAsNew 

func ParseContinueAsNew(workflow graph.Graph) (ContinueAsNewPolicy, bool, error)

ParseContinueAsNew returns the typed extension without accepting unknown or partially canonical config.

type ControlDecisionID 

type ControlDecisionID struct {
	Source NodeInvocationID    `json:"source"`
	Kind   ControlDecisionKind `json:"kind"`
}

ControlDecisionID is collision-free because stores persist its structured fields rather than a delimiter-concatenated primary key.

func (ControlDecisionID) Validate 

func (id ControlDecisionID) Validate() error

type ControlDecisionKind 

type ControlDecisionKind string

ControlDecisionKind identifies one immutable routing fact for an invocation. 

const (
	ControlSwitch ControlDecisionKind = "switch"
	ControlCatch  ControlDecisionKind = "catch"
)

func (ControlDecisionKind) Valid 

func (k ControlDecisionKind) Valid() bool

type ControlDecisionOutcome 

type ControlDecisionOutcome string

ControlDecisionOutcome describes how an ordered route resolved. 

const (
	ControlSelected  ControlDecisionOutcome = "selected"
	ControlDefault   ControlDecisionOutcome = "default"
	ControlUnmatched ControlDecisionOutcome = "unmatched"
	ControlContinued ControlDecisionOutcome = "continued"
)

func (ControlDecisionOutcome) Valid 

func (o ControlDecisionOutcome) Valid() bool

type ControlDecisionSnapshot 

type ControlDecisionSnapshot struct {
	ID               ControlDecisionID      `json:"id"`
	Outcome          ControlDecisionOutcome `json:"outcome"`
	RuleIndex        *int                   `json:"rule_index,omitempty"`
	Targets          []NodeInvocationID     `json:"targets,omitempty"`
	BindAs           string                 `json:"bind_as,omitempty"`
	Error            *values.ValueSetRef    `json:"error,omitempty"`
	SourceGeneration uint64                 `json:"source_generation"`
	Generation       uint64                 `json:"generation"`
	CreatedAt        time.Time              `json:"created_at"`
}

ControlDecisionSnapshot is the immutable, restart-durable result of one ordered switch or catch evaluation. Error is a digest-bound typed ValueSet reference; it never embeds resolved secret material.

func RebindReplayControlDecision 

func RebindReplayControlDecision(decision ControlDecisionSnapshot, targetRun RunID, at time.Time) (ControlDecisionSnapshot, error)

RebindReplayControlDecision preserves an immutable selected-route/error fact while moving only its run identity to the replay. The source decision remains authoritative history; this creates the exact replay-local routing fact.

func (ControlDecisionSnapshot) Validate 

func (s ControlDecisionSnapshot) Validate() error

type ControlErrorOrigin 

type ControlErrorOrigin struct {
	Invocation *NodeInvocationID
	Attempt    *AttemptID
}

ControlErrorOrigin identifies the durable node/attempt behind a terminal error. A run-level error leaves both pointers nil.

func ValidateTerminalControlErrorValues 

func ValidateTerminalControlErrorValues(input values.ValueSet, runID RunID, status RunStatus, reason *Failure) (ControlErrorOrigin, error)

ValidateTerminalControlErrorValues binds an unsuccessful terminal intent's typed error to its run and intended status. The retained error may be a run-level fact (for example explicit cancellation) or the exact originating node failure selected during terminal accounting.

type ControlFlowCoordinator 

type ControlFlowCoordinator struct {
	Store        StateStore
	Control      ControlFlowStore
	Evaluator    PredicateEvaluator
	Compensation CompensationStore
}

ControlFlowCoordinator evaluates expressions outside storage transactions and persists only their deterministic selected-route facts.

func NewControlFlowCoordinator 

func NewControlFlowCoordinator(store StateStore, control ControlFlowStore, evaluator PredicateEvaluator) *ControlFlowCoordinator

func (*ControlFlowCoordinator) DecideCatch 

func (c *ControlFlowCoordinator) DecideCatch(ctx context.Context, request DecideCatchRequest) (RecordControlDecisionResult, error)

func (*ControlFlowCoordinator) DecideSwitch 

func (c *ControlFlowCoordinator) DecideSwitch(ctx context.Context, request DecideSwitchRequest) (RecordControlDecisionResult, error)

func (*ControlFlowCoordinator) PreviewControlNode 

func (c *ControlFlowCoordinator) PreviewControlNode(ctx context.Context, request ProgressControlNodeRequest) (ControlNodePreview, error)

PreviewControlNode reconstructs the authoritative route decision set and catch locals without changing lifecycle state. Drivers use it to determine eligibility before persisting node-local inputs.

func (*ControlFlowCoordinator) PreviewFinally 

func (c *ControlFlowCoordinator) PreviewFinally(ctx context.Context, workflow graph.Graph, invocation NodeInvocationID, expressionContext values.ExpressionContext) (FinalizerPreview, error)

PreviewFinally validates terminal-intent ownership, nested ordering, and cancellation reconciliation without mutating state.

func (*ControlFlowCoordinator) ProgressControlNode 

func (c *ControlFlowCoordinator) ProgressControlNode(ctx context.Context, request ProgressControlNodeRequest) (ProgressNodeResult, error)

ProgressControlNode applies a persisted switch/catch selection before normal readiness. An unselected route is durably skipped; a selected catch source is success-equivalent only for its handler and retains its typed error binding.

func (*ControlFlowCoordinator) ProgressFinally 

func (c *ControlFlowCoordinator) ProgressFinally(ctx context.Context, workflow graph.Graph, invocation NodeInvocationID, expressionContext values.ExpressionContext, expressionOptions values.ExpressionOptions, at time.Time) (ProgressNodeResult, error)

ProgressFinally makes one finalizer eligible only after its durable scope is terminal and all remote cancellation intents are resolved.

func (*ControlFlowCoordinator) ReconcileRunCompletion 

func (c *ControlFlowCoordinator) ReconcileRunCompletion(ctx context.Context, workflow graph.Graph, runID RunID, idempotencyKey string, at time.Time) (RunSnapshot, *TerminalIntentSnapshot, error)

ReconcileRunCompletion records an intended outcome before cleanup and closes it after every finalizer is terminal. Cleanup failure always wins and yields RunFailed while retaining the original intent as typed context.

func (*ControlFlowCoordinator) RequestRunCancellationTree 

func (c *ControlFlowCoordinator) RequestRunCancellationTree(ctx context.Context, workflow graph.Graph, request RequestRunCancellationRequest, descendants []CancellationDescendantGraph) (RequestRunCancellationWithFinalizersResult, error)

RequestRunCancellationTree plans cleanup for the root and every locally owned ParentCloseCancel descendant, then asks the store to validate and apply the complete tree atomically. Descendants may explicitly have no finalizers; zero-finalizer entries keep reachability validation inside the same atomic operation, closing the child-start versus cancellation planning race.

func (*ControlFlowCoordinator) RequestRunCancellationWithFinalizers 

func (c *ControlFlowCoordinator) RequestRunCancellationWithFinalizers(ctx context.Context, workflow graph.Graph, request RequestRunCancellationRequest) (RequestRunCancellationWithFinalizersResult, error)

RequestRunCancellationWithFinalizers plans cleanup and atomically retains a typed run-level cancellation error with the terminal intent and ordinary-work fence. The store assigns one immutable error reference even under contention.

type ControlFlowStore 

type ControlFlowStore interface {
	LoadControlDecision(context.Context, ControlDecisionID) (ControlDecisionSnapshot, error)
	RecordControlDecision(context.Context, RecordControlDecisionRequest) (RecordControlDecisionResult, error)
	LoadTerminalIntent(context.Context, RunID) (TerminalIntentSnapshot, error)
	BeginTerminalIntent(context.Context, BeginTerminalIntentRequest) (BeginTerminalIntentResult, error)
	RequestRunCancellationWithFinalizers(context.Context, RequestRunCancellationWithFinalizersRequest) (RequestRunCancellationWithFinalizersResult, error)
	CompleteTerminalIntent(context.Context, CompleteTerminalIntentRequest) (CompleteTerminalIntentResult, error)
}

ControlFlowStore is deliberately separate from StateStore while the Hadron SQLite persistence lane is serialized. Implementations must make each method atomic with its event and any ErrorValues persistence, assign those immutable ValueSet references themselves, and apply the same run/admission fencing as StateStore. A failed operation must leave no orphan error values or events.

type ControlNodePreview 

type ControlNodePreview struct {
	Target       graph.Node
	Dependencies []DependencyRef
	Context      values.ExpressionContext
	RouteOwned   bool
	Selected     bool
	OwnerIDs     []NodeInvocationID
}

type CrashRecoveryAction 

type CrashRecoveryAction string

CrashRecoveryAction chooses the aggregate node state after atomically closing an interrupted attempt as crashed. 

const (
	CrashRetry    CrashRecoveryAction = "retry_waiting"
	CrashTerminal CrashRecoveryAction = "crashed"
)

type CrashRecoveryDecision 

type CrashRecoveryDecision struct {
	Action CrashRecoveryAction  `json:"action"`
	Policy RepeatPolicyDecision `json:"policy"`
	Retry  *RetryDecision       `json:"retry,omitempty"`
}

func (CrashRecoveryDecision) Validate 

func (d CrashRecoveryDecision) Validate() error

type CreateNodeInvocationRequest 

type CreateNodeInvocationRequest struct {
	Snapshot NodeInvocationSnapshot
}

CreateNodeInvocationRequest persists a new pending invocation snapshot. The store assigns generation one; attempt and claim state must initially be empty.

type CreateRunRequest 

type CreateRunRequest struct {
	ID                  RunID
	Plan                PlanRef
	Status              RunStatus
	Inputs              *values.ValueSetRef
	StartIdempotencyKey string
	CreatedAt           time.Time
}

CreateRunRequest starts persistence for one already validated pending run. It deliberately does not embed the W02-T03 BoundRun contract.

type DecideCatchRequest 

type DecideCatchRequest struct {
	Source            NodeInvocationID
	Node              graph.Node
	Failure           *Failure
	Timeout           TimeoutKind
	ExpressionContext values.ExpressionContext
	ExpressionOptions values.ExpressionOptions
	At                time.Time
}

type DecideSwitchRequest 

type DecideSwitchRequest struct {
	Source            NodeInvocationID
	Node              graph.Node
	ExpressionContext values.ExpressionContext
	ExpressionOptions values.ExpressionOptions
	At                time.Time
}

type DependencyRef 

type DependencyRef struct {
	InvocationID   NodeInvocationID
	FailureHandled bool
}

DependencyRef identifies an upstream invocation for ProgressNode. Status is always loaded from the durable StateStore. Refs are trusted plan-derived identities: callers must supply the complete normalized dependency set, including expanded invocation IDs, because StateStore does not expose graph edge or fan-out reconstruction queries.

type DependencyState 

type DependencyState struct {
	InvocationID   NodeInvocationID
	Status         NodeStatus
	FailureHandled bool
}

DependencyState is one persisted upstream outcome considered by a readiness rule. FailureHandled is a narrow propagation signal: it makes an otherwise hard-failed dependency success-equivalent without implementing catch routes.

type DispatchError 

type DispatchError struct {
	Stage      DispatchStage
	Invocation NodeInvocationID
	Kind       string
	Version    string
	Cause      error
}

DispatchError preserves the typed lifecycle stage and underlying cause.

func (*DispatchError) Error 

func (e *DispatchError) Error() string

Error implements error.

func (*DispatchError) Unwrap 

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

Unwrap preserves both the stable dispatch category and the underlying cause.

type DispatchRequest 

type DispatchRequest struct {
	Claim              ReadyClaim
	Node               graph.Node
	CallLineage        []graph.DefinitionRef
	IdempotencyKey     string
	Target             string
	ExecutorAttributes map[string]string
}

DispatchRequest binds one acquired ready claim to an exact graph node and adapter invocation. Target and ExecutorAttributes describe the selected host execution target without exposing host types.

type DispatchResult 

type DispatchResult struct {
	Node         NodeInvocationSnapshot
	Attempt      AttemptSnapshot
	Result       *stepkind.StepResult
	Outputs      *values.ValueSetRef
	Wait         *WaitSnapshot
	External     *ExternalOperationSnapshot
	Service      *ServiceSnapshot
	Verification *VerificationRecord
	Diagnostics  []diagnostic.Diagnostic
	Warnings     []DispatchWarning
	// contains filtered or unexported fields
}

DispatchResult contains durable attempt state even when Error is returned after a failed invocation. Result and Outputs are populated only on success.

type DispatchStage 

type DispatchStage string

DispatchStage identifies the lifecycle boundary at which dispatch failed. 

const (
	DispatchResolve        DispatchStage = "resolve"
	DispatchValidateConfig DispatchStage = "validate_config"
	DispatchValidateInput  DispatchStage = "validate_input"
	DispatchStartAttempt   DispatchStage = "start_attempt"
	DispatchPrepare        DispatchStage = "prepare"
	DispatchExecute        DispatchStage = "execute"
	DispatchValidateOutput DispatchStage = "validate_output"
	DispatchVerify         DispatchStage = "verify"
	DispatchPersistVerify  DispatchStage = "persist_verification"
	DispatchPersistOutput  DispatchStage = "persist_output"
	DispatchReuseOutput    DispatchStage = "reuse_output"
	DispatchPublishMemo    DispatchStage = "publish_memo"
	DispatchSuspend        DispatchStage = "suspend"
	DispatchHeartbeat      DispatchStage = "heartbeat"
	DispatchObserve        DispatchStage = "observe"
	DispatchCancel         DispatchStage = "cancel"
	DispatchFinishAttempt  DispatchStage = "finish_attempt"
	DispatchFinalize       DispatchStage = "finalize"
)

type DispatchWarning 

type DispatchWarning struct {
	Stage   DispatchStage
	Failure Failure
	Event   *Event
	Cause   error
}

DispatchWarning is a non-reversing lifecycle warning produced after a durable terminal attempt, including finalization and memo publication.

type DispatcherOptions 

type DispatcherOptions struct {
	Store              StateStore
	Registry           stepkind.Registry
	Now                func() time.Time
	FailureDisposition FailureDisposition
	RetentionHook      RetentionHook
	Redactor           *values.Redactor
	WaitCoordinator    *WaitCoordinator
	RetryCoordinator   *RetryCoordinator
	// Verifiers defaults to the deterministic core registry when nil. A
	// supplied typed-nil registry is rejected rather than silently defaulted.
	Verifiers verification.Registry
	// ReuseAuthorizer is required for pins, private cache entries, and every
	// materialize-effect memoization decision.
	ReuseAuthorizer ReuseAuthorizer
}

DispatcherOptions supplies extraction-safe dispatcher collaborators.

type DriveNodeRequest 

type DriveNodeRequest struct {
	Run               RunSnapshot
	Plan              RecoveryPlan
	InvocationID      NodeInvocationID
	Node              graph.Node
	ExpressionContext values.ExpressionContext
	ExpressionOptions values.ExpressionOptions
	At                time.Time
}

type DriveNodeResult 

type DriveNodeResult struct {
	Binding    *BindNodeInputsResult `json:"binding,omitempty"`
	Progressed ProgressNodeResult    `json:"progressed"`
}

type Event 

type Event struct {
	Sequence   uint64                `json:"sequence"`
	RunID      RunID                 `json:"run_id"`
	Invocation *NodeInvocationID     `json:"invocation,omitempty"`
	Attempt    *AttemptID            `json:"attempt,omitempty"`
	Type       string                `json:"type"`
	OccurredAt time.Time             `json:"occurred_at"`
	Attributes map[string]string     `json:"attributes,omitempty"`
	Values     *values.ValueSetRef   `json:"values,omitempty"`
	Redaction  values.RedactionClass `json:"redaction"`
	Retention  values.RetentionClass `json:"retention"`
}

Event is an immutable, append-only operational fact. Sequence is assigned by the store and orders events within one run.

func AppendMaskedEvent 

func AppendMaskedEvent(
	ctx context.Context,
	store StateStore,
	request AppendEventRequest,
	redactor *values.Redactor,
) (Event, error)

AppendMaskedEvent applies adapter-boundary masking before StateStore sees the request. Graph-native command/prompt/message/HTTP/MCP adapters can share this path without adding their transport types to runtime.

func (Event) Validate 

func (e Event) Validate() error

Validate reports malformed event metadata.

type EventQuery 

type EventQuery struct {
	RunID         RunID
	AfterSequence uint64
	Limit         int
}

EventQuery reads immutable events after an exclusive per-run sequence.

type ExecutorMetadata 

type ExecutorMetadata struct {
	Kind       string            `json:"kind"`
	Version    string            `json:"version"`
	Target     string            `json:"target,omitempty"`
	Attributes map[string]string `json:"attributes,omitempty"`
}

ExecutorMetadata identifies the application-neutral executor contract used by one attempt. It is immutable for the lifetime of that attempt.

func (ExecutorMetadata) Validate 

func (m ExecutorMetadata) Validate() error

Validate reports malformed persisted executor identity.

type ExpandFanOutRequest 

type ExpandFanOutRequest struct {
	FanOut                   FanOutSnapshot
	ExpectedParentGeneration uint64
	Priority                 int
	At                       time.Time
}

func (ExpandFanOutRequest) Validate 

func (r ExpandFanOutRequest) Validate() error

type ExpandFanOutResult 

type ExpandFanOutResult struct {
	FanOut   FanOutSnapshot
	Parent   NodeInvocationSnapshot
	Children []NodeInvocationSnapshot
	Event    Event
}

type ExplainService 

type ExplainService struct {
	Store   StateStore
	Control ControlFlowStore
	Replay  ReplayStore
	Plans   RecoveryPlanSource
}

func (ExplainService) Explain 

func (s ExplainService) Explain(ctx context.Context, runID RunID, now time.Time) (RunExplanation, error)

type ExternalActivationRequest 

type ExternalActivationRequest struct {
	ActivationID   string
	IdempotencyKey string
	RequestedRunID RunID
	Plan           PlanRef
	Inputs         *values.ValueSetRef
	OccurredAt     time.Time
}

ExternalActivationRequest records one externally initiated run request.

type ExternalActivationSnapshot 

type ExternalActivationSnapshot struct {
	ActivationID   string              `json:"activation_id"`
	IdempotencyKey string              `json:"idempotency_key"`
	RequestedRunID RunID               `json:"requested_run_id"`
	Plan           PlanRef             `json:"plan"`
	Inputs         *values.ValueSetRef `json:"inputs,omitempty"`
	OccurredAt     time.Time           `json:"occurred_at"`
}

ExternalActivationSnapshot is the immutable accepted activation request.

type ExternalOperationCoordinator 

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

ExternalOperationCoordinator reconciles adapter-owned work without a worker claim. Suspension has already released the execution lease; W03-T06 may schedule and retry these idempotent observation calls, while durable refs and store CAS generations fence competing observers.

func NewExternalOperationCoordinator 

func NewExternalOperationCoordinator(options ExternalOperationOptions) (*ExternalOperationCoordinator, error)

NewExternalOperationCoordinator constructs recovery-aware external work reconciliation. Adapter calls are never made inside StateStore transactions.

func (*ExternalOperationCoordinator) Reconcile 

func (c *ExternalOperationCoordinator) Reconcile(ctx context.Context, id AttemptID) (ExternalOperationResult, error)

Reconcile heartbeats and observes one pending durable operation. Transient heartbeat or observation I/O leaves the durable operation pending for recovery. Adapter-reported terminal states and irrecoverable contract mismatches close failed by default, or ready when disposition selects retry.

func (*ExternalOperationCoordinator) Recover 

func (c *ExternalOperationCoordinator) Recover(ctx context.Context, query ExternalOperationQuery) ([]ExternalOperationSnapshot, error)

Recover returns pending durable operations in store order. It deliberately performs no adapter I/O or scheduling policy.

func (*ExternalOperationCoordinator) RequestCancel 

func (c *ExternalOperationCoordinator) RequestCancel(ctx context.Context, id AttemptID) (ExternalOperationResult, error)

RequestCancel persists cancellation intent before calling the adapter, then reconciles the operation. A crash between those steps leaves discoverable intent for recovery to retry.

type ExternalOperationOptions 

type ExternalOperationOptions struct {
	Store              StateStore
	Registry           stepkind.Registry
	Now                func() time.Time
	FailureDisposition FailureDisposition
	RetentionHook      RetentionHook
	Redactor           *values.Redactor
	// Verifiers must match the dispatcher catalog. Nil selects the core
	// deterministic catalog; construction freezes the exact snapshot.
	Verifiers verification.Registry
}

ExternalOperationOptions supplies extraction-safe reconciliation collaborators. A nil failure disposition closes failed observations terminally; W03 retry policy may inject NodeReady.

type ExternalOperationQuery 

type ExternalOperationQuery struct {
	RunID RunID
	Limit int
}

ExternalOperationQuery selects pending recovery work in deterministic updated-time then attempt-identity order. Limit zero means unlimited.

type ExternalOperationResult 

type ExternalOperationResult struct {
	Operation    ExternalOperationSnapshot
	Node         NodeInvocationSnapshot
	Attempt      AttemptSnapshot
	Result       *stepkind.StepResult
	Outputs      *values.ValueSetRef
	Verification *VerificationRecord
	Warnings     []DispatchWarning
}

ExternalOperationResult is one durable reconciliation outcome. A non-nil error may accompany a terminal result or a post-commit finalization warning; callers must treat the returned snapshots as authoritative.

type ExternalOperationSnapshot 

type ExternalOperationSnapshot struct {
	Attempt           AttemptID                     `json:"attempt"`
	Ref               stepkind.ExternalOperationRef `json:"ref"`
	Invocation        stepkind.Invocation           `json:"invocation"`
	Status            stepkind.ObservationState     `json:"status"`
	Progress          map[string]string             `json:"progress,omitempty"`
	Outputs           *values.ValueSetRef           `json:"outputs,omitempty"`
	Failure           *Failure                      `json:"failure,omitempty"`
	CancelRequestedAt time.Time                     `json:"cancel_requested_at,omitempty"`
	LastObservedAt    time.Time                     `json:"last_observed_at"`
	LastHeartbeatAt   time.Time                     `json:"last_heartbeat_at,omitempty"`
	Generation        uint64                        `json:"generation"`
	CreatedAt         time.Time                     `json:"created_at"`
	UpdatedAt         time.Time                     `json:"updated_at"`
}

ExternalOperationSnapshot is the durable one-per-attempt binding between an unfinished attempt and adapter-owned work. Attempt, Ref, and Invocation are immutable after creation. Status, progress, observation/heartbeat times, cancellation intent, and the terminal outcome evolve only through CAS store operations.

func (ExternalOperationSnapshot) Validate 

func (s ExternalOperationSnapshot) Validate() error

Validate reports malformed durable external-operation state without assigning lifecycle transition policy.

type ExternalOperationStore 

type ExternalOperationStore interface {
	LoadExternalOperation(context.Context, AttemptID) (ExternalOperationSnapshot, error)
	SuspendExternalOperation(context.Context, SuspendExternalOperationRequest) (SuspendExternalOperationResult, error)
	RequestExternalOperationCancel(context.Context, RequestExternalOperationCancelRequest) (RequestExternalOperationCancelResult, error)
	ApplyExternalOperation(context.Context, ApplyExternalOperationRequest) (ApplyExternalOperationResult, error)
	RecoverExternalOperations(context.Context, ExternalOperationQuery) ([]ExternalOperationSnapshot, error)
}

ExternalOperationStore is the recovery-facing persistence surface. Adapter calls happen outside these atomic mutations; their immutable ref plus CAS generations make retries and competing observers fail closed.

type ExternalVerificationCompletion 

type ExternalVerificationCompletion struct {
	Values     values.ValueSetRef
	Attributes map[string]string
}

ExternalVerificationCompletion links one already-saved immutable report to a terminal external observation. The store appends the attempt-bound event inside the same CAS transaction that closes the operation, so competing observers cannot publish duplicate verification histories. A CAS loser may leave an unlinked value set for retention/repair, but never a durable claim that verification was applied.

func (ExternalVerificationCompletion) Validate 

func (c ExternalVerificationCompletion) Validate() error

type FailCompensationEntryRequest 

type FailCompensationEntryRequest struct {
	RunID                    RunID
	EntryID                  string
	ExpectedLedgerGeneration uint64
	ExpectedEntryGeneration  uint64
	Failure                  Failure
	ChildResolution          CompensationChildResolution
	At                       time.Time
}

FailCompensationEntryRequest durably terminalizes a pending entry when its pinned handler inputs cannot be constructed. No handler invocation is materialized because there is no schema-valid input envelope to execute.

type FailReactorRequest 

type FailReactorRequest struct {
	ReactorID          string    `json:"reactor_id"`
	ExpectedGeneration uint64    `json:"expected_generation"`
	RunID              RunID     `json:"run_id"`
	RunStatus          RunStatus `json:"run_status"`
	At                 time.Time `json:"at"`
}

FailReactorRequest durably fences a reactor whose current run ended without a successful continuation boundary. Pending deliveries are closed against that exact terminal run in the same atomic transition.

func (FailReactorRequest) Validate 

func (r FailReactorRequest) Validate() error

type Failure 

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

Failure is a persisted application-neutral execution failure.

func TimeoutFailure 

func TimeoutFailure(kind TimeoutKind, deadline time.Time) Failure

TimeoutFailure produces the stable failure used by retry and lifecycle policy. The exact category remains searchable in Details.

func WaitTimeoutFailure 

func WaitTimeoutFailure(deadline time.Time) Failure

WaitTimeoutFailure returns the canonical attempt failure for a deadline.

func (Failure) Validate 

func (f Failure) Validate() error

Validate reports malformed failure metadata.

type FailureDisposition 

type FailureDisposition interface {
	NextNodeStatus(context.Context, FailureDispositionRequest) (NodeStatus, error)
}

FailureDisposition selects the aggregate node state after an unsuccessful attempt without implementing retry limits or backoff in the dispatcher.

type FailureDispositionFunc 

type FailureDispositionFunc func(context.Context, FailureDispositionRequest) (NodeStatus, error)

FailureDispositionFunc adapts a function to FailureDisposition.

func (FailureDispositionFunc) NextNodeStatus 

func (f FailureDispositionFunc) NextNodeStatus(ctx context.Context, request FailureDispositionRequest) (NodeStatus, error)

NextNodeStatus implements FailureDisposition.

type FailureDispositionRequest 

type FailureDispositionRequest struct {
	Spec    stepkind.StepKindSpec
	Attempt AttemptSnapshot
	Failure Failure
	Status  NodeStatus
}

FailureDispositionRequest supplies a finished execution failure to injected retry policy. W03-T04 may return NodeReady; no policy defaults to the terminal attempt status.

type FanOutCoordinator 

type FanOutCoordinator struct {
	Store     StateStore
	Evaluator FanOutExpressionEvaluator
	Retention RetentionHook
}

func (FanOutCoordinator) Collect 

func (c FanOutCoordinator) Collect(ctx context.Context, parent NodeInvocationID, at time.Time) (CompleteFanOutResult, values.ValueSet, []FanOutItemResult, error)

Collect loads every terminal child in stable index order, publishes an aggregate items value containing typed output references and structured failures, then atomically finalizes the aggregate parent.

func (FanOutCoordinator) Expand 

func (c FanOutCoordinator) Expand(ctx context.Context, command FanOutExpandCommand) (ExpandFanOutResult, error)

func (FanOutCoordinator) ReconcileFailFast 

func (c FanOutCoordinator) ReconcileFailFast(ctx context.Context, parent NodeInvocationID, at time.Time) ([]NodeInvocationSnapshot, error)

ReconcileFailFast fences admission and cancels fan-out items that have not started after failures exceed the declared tolerance. The transition uses ordinary node CAS and is therefore safe to replay and race with a worker claim; already-started items are never interrupted by this policy.

type FanOutExpandCommand 

type FanOutExpandCommand struct {
	Parent                   NodeInvocationID
	ExpectedParentGeneration uint64
	Spec                     graph.ForEachSpec
	// InputBindings is the exact compiled node-local binding set. Each child
	// receives the synthetic item/index values plus these evaluated inputs in
	// one immutable ValueSet before it becomes ready. Nil preserves the legacy
	// low-level fan-out contract that stores only item/index.
	InputBindings     map[string]graph.Binding
	ExpressionContext values.ExpressionContext
	ExpressionOptions values.ExpressionOptions
	Priority          int
	At                time.Time
}

FanOutExpandCommand evaluates one static for_each declaration and persists the immutable runtime expansion. Saved item input sets may remain unreferenced if the final atomic expansion loses CAS; they never create partial children.

type FanOutExpressionEvaluator 

type FanOutExpressionEvaluator interface {
	EvaluateRaw(graph.Expression, values.ExpressionContext, values.ExpressionOptions) (any, error)
}

FanOutExpressionEvaluator is the narrow typed expression seam.

type FanOutItemBinding 

type FanOutItemBinding struct {
	Index      int                `json:"index"`
	Iteration  string             `json:"iteration"`
	Invocation NodeInvocationID   `json:"invocation"`
	Inputs     values.ValueSetRef `json:"inputs"`
}

FanOutItemBinding is the immutable relationship between an evaluated item, its stable zero-based index, typed local input set, and child invocation.

func (FanOutItemBinding) Validate 

func (b FanOutItemBinding) Validate(parent NodeInvocationID) error

type FanOutItemResult 

type FanOutItemResult struct {
	Index        int                 `json:"index"`
	Iteration    string              `json:"iteration"`
	Invocation   NodeInvocationID    `json:"invocation"`
	Status       NodeStatus          `json:"status"`
	Outputs      *values.ValueSetRef `json:"outputs,omitempty"`
	OutputValues values.ValueSet     `json:"output_values,omitempty"`
	Failure      *Failure            `json:"failure,omitempty"`
}

FanOutItemResult is the lossless recovery projection for one item. Outputs is the immutable durable reference and OutputValues is its defensively loaded typed ValueSet. Parent aggregate values intentionally contain only the reference because ValueSet has no nested typed-envelope container.

type FanOutSnapshot 

type FanOutSnapshot struct {
	Parent         NodeInvocationID              `json:"parent"`
	ItemName       string                        `json:"item_name"`
	IndexName      string                        `json:"index_name"`
	MaxConcurrency int                           `json:"max_concurrency,omitempty"`
	FailFast       bool                          `json:"fail_fast,omitempty"`
	Tolerate       *graph.ToleratedFailurePolicy `json:"tolerate,omitempty"`
	Status         FanOutStatus                  `json:"status"`
	Items          []FanOutItemBinding           `json:"items"`
	Outputs        *values.ValueSetRef           `json:"outputs,omitempty"`
	Failure        *Failure                      `json:"failure,omitempty"`
	Generation     uint64                        `json:"generation"`
	CreatedAt      time.Time                     `json:"created_at"`
	UpdatedAt      time.Time                     `json:"updated_at"`
}

FanOutSnapshot is one durable runtime expansion. MaxConcurrency counts started nonterminal child invocations, including retry delays, generic waits, and pending external operations; it is deliberately independent of leases.

func (FanOutSnapshot) Validate 

func (s FanOutSnapshot) Validate() error

type FanOutStatus 

type FanOutStatus string

FanOutStatus is the aggregate lifecycle stored independently from worker leases and individual item invocations. 

const (
	FanOutActive    FanOutStatus = "active"
	FanOutSucceeded FanOutStatus = "succeeded"
	FanOutFailed    FanOutStatus = "failed"
	FanOutCanceled  FanOutStatus = "canceled"
)

func (FanOutStatus) Valid 

func (s FanOutStatus) Valid() bool

type FanOutStore 

type FanOutStore interface {
	LoadFanOut(context.Context, NodeInvocationID) (FanOutSnapshot, error)
	LoadFanOutItemResults(context.Context, NodeInvocationID) ([]FanOutItemResult, error)
	ExpandFanOut(context.Context, ExpandFanOutRequest) (ExpandFanOutResult, error)
	CompleteFanOut(context.Context, CompleteFanOutRequest) (CompleteFanOutResult, error)
}

FanOutStore persists expansion identity and performs atomic aggregate creation/completion. Implementations also enforce MaxConcurrency when a child's first attempt starts.

type FinalizeRunRequest 

type FinalizeRunRequest struct {
	BoundRun      BoundRun
	Run           RunSnapshot
	Plan          *compile.ExecutionPlan
	Context       values.ExpressionContext
	BaseOptions   values.ExpressionOptions
	RetentionHook RetentionHook
	Control       ControlFlowStore
	At            time.Time
}

FinalizeRunRequest binds declared workflow outputs after graph completion. Context may contain more steps than the plan, but output evaluation sees only declared plan nodes. BaseOptions retains host policy such as AllowEnv. RetentionHook is optional and is not consulted during replay without writes. Control is required when a pending terminal intent owns finalizer completion; when omitted, FinalizeRunOutputs uses store if it implements ControlFlowStore.

type FinalizeRunResult 

type FinalizeRunResult struct {
	Run         RunSnapshot
	Outputs     values.ValueSet
	OutputsRef  values.ValueSetRef
	Outcome     OutputFinalizationOutcome
	Diagnostics []diagnostic.Diagnostic
}

FinalizeRunResult is present only after outputs have been published or an already-published semantically identical set has been confirmed.

func FinalizeRunOutputs 

func FinalizeRunOutputs(ctx context.Context, store StateStore, request FinalizeRunRequest) (FinalizeRunResult, error)

FinalizeRunOutputs evaluates and validates the entire declared output set, saves it once, and publishes only the complete reference with the succeeded run transition. A failed transition can leave the complete saved set unreferenced, but never publishes partial outputs. Routine replay loads and compares existing outputs before any SaveValues call.

type FinalizerPreview 

type FinalizerPreview struct {
	Definition   graph.Node
	Context      values.ExpressionContext
	Dependencies []DependencyRef
}

FinalizerPreview is the read-only, durable eligibility/context projection shared by input binding and finalizer progression. It prevents recovery from evaluating or persisting cleanup inputs before the exact scope is eligible.

type FinalizerScope 

type FinalizerScope struct {
	Invocation NodeInvocationID   `json:"invocation"`
	Scope      []NodeInvocationID `json:"scope"`
	Order      int                `json:"order"`
}

FinalizerScope is one cleanup invocation and its complete ordered terminal prerequisite set. Order is an inner-to-outer topological layer.

func PlanFinalizerScopes 

func PlanFinalizerScopes(workflow graph.Graph, runID RunID) ([]FinalizerScope, error)

PlanFinalizerScopes derives deterministic inner-to-outer durable scopes from one immutable graph. An empty declared scope means the whole workflow.

func (FinalizerScope) Validate 

func (s FinalizerScope) Validate(runID RunID) error

type FinishCompensableAttemptRequest 

type FinishCompensableAttemptRequest struct {
	Finish      FinishNodeAttemptRequest
	Eligibility CompensationEligibility
}

type FinishCompensableAttemptResult 

type FinishCompensableAttemptResult struct {
	Finish FinishNodeAttemptResult
	Ledger CompensationLedgerSnapshot
	Entry  CompensationEntrySnapshot
}

type FinishNodeAttemptRequest 

type FinishNodeAttemptRequest struct {
	InvocationID              NodeInvocationID
	AttemptNumber             int
	ExpectedNodeGeneration    uint64
	ExpectedAttemptGeneration uint64
	Claim                     ClaimProof
	AttemptStatus             NodeStatus
	NextNodeStatus            NodeStatus
	Outputs                   *values.ValueSetRef
	Failure                   *Failure
	At                        time.Time
}

FinishNodeAttemptRequest atomically closes the latest unfinished running attempt. AttemptStatus is the durable attempt outcome. NextNodeStatus is either that terminal outcome or ready for a future retry/recovery decision.

type FinishNodeAttemptResult 

type FinishNodeAttemptResult struct {
	Node    NodeInvocationSnapshot
	Attempt AttemptSnapshot
	Event   Event
}

FinishNodeAttemptResult contains the atomically persisted aggregate node, finished attempt, and attempt-finished event.

type FreezeCompensationRequest 

type FreezeCompensationRequest struct {
	RunID                    RunID
	PlanDigest               string
	ExpectedRunGeneration    uint64
	ExpectedIntentGeneration uint64
	Trigger                  graph.CompensationTrigger
	OriginalStatus           RunStatus
	OriginalFailure          *values.ValueSetRef
	Dependencies             map[string][]string
	IdempotencyKey           string
	At                       time.Time
}

type FreezeCompensationResult 

type FreezeCompensationResult struct {
	Outcome IdempotencyOutcome
	Ledger  CompensationLedgerSnapshot
	Entries []CompensationEntrySnapshot
}

type HandleRunFailureRequest 

type HandleRunFailureRequest struct {
	Workflow       graph.Graph
	Source         NodeInvocationID
	Failure        *Failure
	IdempotencyKey string
	At             time.Time
}

HandleRunFailureRequest carries the exact durable failure when a node reaches a hard terminal state before an attempt exists (for example, a queue timeout). Attempt-backed failures are always loaded from StateStore; supplying a divergent Failure is rejected by the control-flow contract.

type IdempotencyConflictError 

type IdempotencyConflictError struct {
	Operation string
	Key       string
}

IdempotencyConflictError reports reuse of a key for a different request.

func (*IdempotencyConflictError) Error 

func (e *IdempotencyConflictError) Error() string

func (*IdempotencyConflictError) Unwrap 

func (e *IdempotencyConflictError) Unwrap() error

Unwrap supports errors.Is(err, ErrIdempotencyConflict).

type IdempotencyOutcome 

type IdempotencyOutcome string

IdempotencyOutcome distinguishes a new mutation from an exact replay. 

const (
	IdempotencyApplied  IdempotencyOutcome = "applied"
	IdempotencyReplayed IdempotencyOutcome = "replayed"
)

type InvocationExplanation 

type InvocationExplanation struct {
	Node     NodeInvocationSnapshot `json:"node"`
	Attempts []AttemptSnapshot      `json:"attempts"`
}

InvocationExplanation combines one authoritative node snapshot with its complete durable attempt history. It never infers state from logs.

type InvocationOrigin 

type InvocationOrigin string

InvocationOrigin identifies how an invocation reached its durable outcome. 

const (
	OriginExecuted InvocationOrigin = "executed"
	OriginMemoized InvocationOrigin = "memoized"
	OriginReplayed InvocationOrigin = "replayed"
	OriginPinned   InvocationOrigin = "pinned"
)

func (InvocationOrigin) Valid 

func (o InvocationOrigin) Valid() bool

type InvocationPhase 

type InvocationPhase string

InvocationPhase is the durable lifecycle lane for an invocation. Empty is the ordinary forward lane for source compatibility; compensation is created only by the saga store after the terminal intent fence is established. 

const (
	InvocationForward      InvocationPhase = ""
	InvocationCompensation InvocationPhase = "compensation"
)

func (InvocationPhase) Valid 

func (p InvocationPhase) Valid() bool

type MemoEntry 

type MemoEntry struct {
	Key                string              `json:"key"`
	PlanDigest         string              `json:"plan_digest"`
	NodeID             string              `json:"node_id"`
	Kind               string              `json:"kind"`
	KindVersion        string              `json:"kind_version"`
	MemoKeyDigest      string              `json:"memo_key_digest"`
	InputDigest        string              `json:"input_digest"`
	OutputSchemaDigest string              `json:"output_schema_digest"`
	OutputDigest       string              `json:"output_digest"`
	Outputs            values.ValueSetRef  `json:"outputs"`
	Source             NodeInvocationID    `json:"source"`
	SourceAttempt      AttemptID           `json:"source_attempt"`
	SourceOrigin       InvocationOrigin    `json:"source_origin"`
	Effects            graph.EffectSet     `json:"effects"`
	Policy             ReusePolicyDecision `json:"policy"`
	CreatedAt          time.Time           `json:"created_at"`
	ExpiresAt          time.Time           `json:"expires_at"`
}

MemoEntry is an immutable cache publication with its exact source facts.

func (MemoEntry) FreshAt 

func (e MemoEntry) FreshAt(now time.Time, maxAge time.Duration) error

FreshAt applies both the publication expiry and the caller's current max-age policy. This prevents an older, longer-lived declaration from widening a later invocation's freshness contract.

func (MemoEntry) Validate 

func (e MemoEntry) Validate() error

type MemoStore 

type MemoStore interface {
	RecordMemoEntry(context.Context, MemoEntry) (MemoEntry, IdempotencyOutcome, error)
	LoadMemoEntry(context.Context, string) (MemoEntry, error)
}

MemoStore is the new runtime truth for append-only memo publications. Legacy StateStore cache CRUD remains compatibility scaffolding and is not consulted by the dispatcher.

type NodeDriver 

type NodeDriver struct {
	Store     StateStore
	Inputs    NodeInputStore
	Control   ControlFlowStore
	Registry  stepkind.Registry
	Evaluator PredicateEvaluator
}

NodeDriver binds one node's exact plan inputs before applying existing readiness/control-flow semantics. A node cannot become claimable until the binding transaction has committed.

func (*NodeDriver) Drive 

func (d *NodeDriver) Drive(ctx context.Context, request DriveNodeRequest) (DriveNodeResult, error)

func (*NodeDriver) DriveFinally 

func (d *NodeDriver) DriveFinally(ctx context.Context, request DriveNodeRequest) (DriveNodeResult, error)

DriveFinally binds one eligible finalizer with the same compiler visibility and atomic input seam used for ordinary work. Scope/cancellation ordering is previewed before expressions run, so an ineligible cleanup never publishes input values or becomes claimable.

type NodeInputStore 

type NodeInputStore interface {
	BindNodeInputs(context.Context, BindNodeInputsRequest) (BindNodeInputsResult, error)
}

NodeInputStore is the atomic persistence seam shared by ordinary execution drivers and crash recovery. It deliberately does not widen StateStore.

type NodeInvocationID 

type NodeInvocationID struct {
	RunID     RunID  `json:"run_id"`
	NodeID    string `json:"node_id"`
	Iteration string `json:"iteration,omitempty"`
}

NodeInvocationID identifies one node expansion inside a run. Iteration is empty for nodes that are not expanded by for_each.

func CompensationHandlerID 

func CompensationHandlerID(runID RunID, nodeID, entryID string) NodeInvocationID

func (NodeInvocationID) Validate 

func (id NodeInvocationID) Validate() error

Validate reports malformed invocation identity.

type NodeInvocationSnapshot 

type NodeInvocationSnapshot struct {
	ID      NodeInvocationID    `json:"id"`
	Phase   InvocationPhase     `json:"phase,omitempty"`
	Status  NodeStatus          `json:"status"`
	Blocked *BlockedReason      `json:"blocked,omitempty"`
	Inputs  *values.ValueSetRef `json:"inputs,omitempty"`
	Outputs *values.ValueSetRef `json:"outputs,omitempty"`
	// Origin identifies how this invocation reached its durable outcome. It is
	// empty while the invocation has no outcome.
	Origin InvocationOrigin `json:"origin,omitempty"`
	// MemoKeyDigest is the digest of the compiler-scoped memo key evaluation.
	// The raw key is never persisted because it may contain sensitive data.
	MemoKeyDigest   string      `json:"memo_key_digest,omitempty"`
	Wait            *WaitRef    `json:"wait,omitempty"`
	LatestAttempt   int         `json:"latest_attempt,omitempty"`
	Priority        int         `json:"priority,omitempty"`
	ClaimGeneration uint64      `json:"claim_generation"`
	Lease           *ClaimLease `json:"lease,omitempty"`
	Generation      uint64      `json:"generation"`
	CreatedAt       time.Time   `json:"created_at"`
	UpdatedAt       time.Time   `json:"updated_at"`
}

NodeInvocationSnapshot is one persisted graph-node expansion. Generation is the record CAS revision; ClaimGeneration changes only on successful claims.

func (NodeInvocationSnapshot) Validate 

func (s NodeInvocationSnapshot) Validate() error

Validate checks record integrity without enforcing lifecycle transitions.

type NodeOutputValidationError 

type NodeOutputValidationError struct {
	Output string
	Source *graph.SourceRef
	Cause  error
}

NodeOutputValidationError preserves the graph source of one declared output while unwrapping the canonical values-layer schema or shape error.

func (*NodeOutputValidationError) Error 

func (e *NodeOutputValidationError) Error() string

func (*NodeOutputValidationError) Unwrap 

func (e *NodeOutputValidationError) Unwrap() error

type NodeStatus 

type NodeStatus string

NodeStatus is the persisted state of one node invocation. 

const (
	NodePending   NodeStatus = "pending"
	NodeReady     NodeStatus = "ready"
	NodeRunning   NodeStatus = "running"
	NodeWaiting   NodeStatus = "waiting"
	NodeSucceeded NodeStatus = "succeeded"
	NodeFailed    NodeStatus = "failed"
	NodeSkipped   NodeStatus = "skipped"
	NodeCanceled  NodeStatus = "canceled"
	NodeTimedOut  NodeStatus = "timed_out"
	NodeCrashed   NodeStatus = "crashed"
	NodeBlocked   NodeStatus = "blocked"
)

func (NodeStatus) CanTransitionTo 

func (s NodeStatus) CanTransitionTo(next NodeStatus) bool

CanTransitionTo reports whether the generic node lifecycle permits an applied status change. Attempt start/finish operations own execution-only transitions, including ready-to-running and running-to-terminal/ready.

func (NodeStatus) Terminal 

func (s NodeStatus) Terminal() bool

Terminal reports whether an aggregate node status cannot be reopened.

func (NodeStatus) Valid 

func (s NodeStatus) Valid() bool

Valid reports whether s is a supported persisted node status.

type NodeTransitionRequest 

type NodeTransitionRequest struct {
	InvocationID       NodeInvocationID
	ExpectedGeneration uint64
	To                 NodeStatus
	Blocked            *BlockedReason
	// Explanation is an application-neutral reason persisted in the atomic
	// status event for a transition to skipped. Stores must compare it for an
	// exact skipped replay and reject a different explanation. It is never
	// stored on the terminal node snapshot; Blocked remains exclusive to
	// NodeBlocked.
	Explanation *BlockedReason
	Claim       *ClaimProof
	At          time.Time
}

NodeTransitionRequest changes a non-attempt node lifecycle edge under CAS. Claim is required when the node currently has a lease. An exact same-state request is a no-op. NodeBlocked to NodeBlocked with a changed valid Blocked reason and strictly later At is an applied diagnostic refresh; every other changed same-state request conflicts.

type NodeTransitionResult 

type NodeTransitionResult struct {
	Snapshot NodeInvocationSnapshot
	Outcome  TransitionOutcome
	Event    *Event
}

NodeTransitionResult is the atomic node snapshot and event outcome. Event is nil only for an exact no-op.

type OpenWaitQuery 

type OpenWaitQuery struct {
	RunID RunID
	Limit int
}

OpenWaitQuery supports deterministic durable recovery. Limit zero means unlimited; results are ordered by earliest WakeAt/Deadline, creation time, then wait ID.

type OutputFinalizationOutcome 

type OutputFinalizationOutcome string

OutputFinalizationOutcome distinguishes a newly published output set from a semantically identical replay against an already succeeded run. 

const (
	OutputFinalizationApplied  OutputFinalizationOutcome = "applied"
	OutputFinalizationReplayed OutputFinalizationOutcome = "replayed"
)

type OutputReuseStore 

type OutputReuseStore interface {
	ReuseNodeOutputs(context.Context, ReuseNodeOutputsRequest) (ReuseNodeOutputsResult, error)
}

OutputReuseStore atomically completes one claimed, unattempted node from an immutable memo publication or pin binding.

type PinBinding 

type PinBinding struct {
	Target             NodeInvocationID    `json:"target"`
	PlanDigest         string              `json:"plan_digest"`
	Outputs            values.ValueSetRef  `json:"outputs"`
	OutputSchemaDigest string              `json:"output_schema_digest"`
	Source             NodeInvocationID    `json:"source"`
	SourcePlanDigest   string              `json:"source_plan_digest"`
	SourceOrigin       InvocationOrigin    `json:"source_origin"`
	Authority          ReuseAuthority      `json:"authority"`
	Policy             ReusePolicyDecision `json:"policy"`
	BoundAt            time.Time           `json:"bound_at"`
}

PinBinding durably binds one target invocation to trusted source outputs.

func (PinBinding) Validate 

func (b PinBinding) Validate() error

type PinCoordinator 

type PinCoordinator struct {
	Store      StateStore
	Pins       PinStore
	Values     ValueRecordStore
	Plans      RecoveryPlanSource
	Registry   stepkind.Registry
	Authorizer ReuseAuthorizer
}

PinCoordinator validates a transport-neutral node=<value-ref> request and atomically installs the immutable run-scoped binding before admission.

func (PinCoordinator) Bind 

func (c PinCoordinator) Bind(ctx context.Context, request PinNodeRequest) (BindPinResult, error)

type PinNodeRequest 

type PinNodeRequest struct {
	Target         NodeInvocationID   `json:"target"`
	Outputs        values.ValueSetRef `json:"outputs"`
	Authority      ReuseAuthority     `json:"authority"`
	IdempotencyKey string             `json:"idempotency_key"`
	At             time.Time          `json:"at"`
}

type PinStore 

type PinStore interface {
	BindPin(context.Context, BindPinRequest) (BindPinResult, error)
	LoadPin(context.Context, NodeInvocationID) (PinBinding, error)
}

PinStore atomically installs immutable run-scoped pins before admission.

type PinnedValue 

type PinnedValue struct {
	Key       string    `json:"key"`
	Value     ValueRef  `json:"value"`
	PinnedAt  time.Time `json:"pinned_at"`
	ExpiresAt time.Time `json:"expires_at,omitempty"`
}

PinnedValue gives a stable host-independent name to a persisted value.

func (PinnedValue) Validate 

func (p PinnedValue) Validate() error

Validate reports malformed pin metadata.

type PlanRef 

type PlanRef struct {
	ID            string `json:"id"`
	Version       string `json:"version"`
	Digest        string `json:"digest"`
	SchemaVersion string `json:"schema_version"`
}

PlanRef identifies an immutable compiled execution plan without embedding it in mutable runtime records.

func (PlanRef) Validate 

func (r PlanRef) Validate() error

Validate reports malformed plan identity or digest metadata.

type PostCommitError 

type PostCommitError struct {
	Operation string
	Err       error
}

PostCommitError reports an operational adapter failure after the durable result was committed. Callers must inspect and retain the returned result.

func (*PostCommitError) Error 

func (e *PostCommitError) Error() string

func (*PostCommitError) Unwrap 

func (e *PostCommitError) Unwrap() error

type PredicateEvaluator 

type PredicateEvaluator interface {
	EvaluateBool(graph.Expression, values.ExpressionContext, values.ExpressionOptions) (bool, error)
}

PredicateEvaluator is the narrow expression seam used after readiness is satisfied. values.ExpressionEngine implements this interface.

type PrepareServiceStartRequest 

type PrepareServiceStartRequest struct {
	Service                   ServiceSnapshot
	ExpectedNodeGeneration    uint64
	ExpectedAttemptGeneration uint64
	At                        time.Time
}

PrepareServiceStartRequest persists acquisition intent after the attempt is durable and immediately before provider Start. Recovery may safely invoke Start again with the immutable invocation idempotency key.

type ProgressControlNodeRequest 

type ProgressControlNodeRequest struct {
	Graph             graph.Graph
	InvocationID      NodeInvocationID
	Dependencies      []DependencyRef
	ExpressionContext values.ExpressionContext
	ExpressionOptions values.ExpressionOptions
	At                time.Time
}

type ProgressNodeRequest 

type ProgressNodeRequest struct {
	InvocationID      NodeInvocationID
	Dependencies      []DependencyRef
	Rule              graph.ReadyRule
	Predicate         *graph.Expression
	ExpressionContext values.ExpressionContext
	ExpressionOptions values.ExpressionOptions
	At                time.Time
}

ProgressNodeRequest supplies durable dependency identities and the typed expression context for one pending or blocked invocation.

type ProgressNodeResult 

type ProgressNodeResult struct {
	Disposition        ReadinessDisposition
	Snapshot           NodeInvocationSnapshot
	Reason             *BlockedReason
	Event              *Event
	PredicateEvaluated bool
	PredicateResult    bool
}

ProgressNodeResult contains the applied or observed progression outcome. Event is nil for a durable no-op.

type ProgressionCoordinator 

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

ProgressionCoordinator evaluates and persists readiness without retaining correctness-critical state between calls.

func NewProgressionCoordinator 

func NewProgressionCoordinator(store StateStore, evaluator PredicateEvaluator) *ProgressionCoordinator

NewProgressionCoordinator constructs a progression coordinator. A nil evaluator uses a fresh typed values.ExpressionEngine.

func (*ProgressionCoordinator) ProgressNode 

func (c *ProgressionCoordinator) ProgressNode(ctx context.Context, request ProgressNodeRequest) (ProgressNodeResult, error)

ProgressNode loads authoritative dependency state, evaluates the readiness rule, then evaluates if only when ready. It persists blocked, ready, or skipped through lifecycle CAS and returns expression failures without a transition.

type ReactorContinuationRequest 

type ReactorContinuationRequest struct {
	IdempotencyKey     string          `json:"idempotency_key"`
	ReactorID          string          `json:"reactor_id"`
	ExpectedGeneration uint64          `json:"expected_generation"`
	FromGeneration     uint64          `json:"from_generation"`
	FromRunID          RunID           `json:"from_run_id"`
	ToRunID            RunID           `json:"to_run_id"`
	State              values.ValueSet `json:"state,omitempty"`
	At                 time.Time       `json:"at"`
}

func (ReactorContinuationRequest) Validate 

func (r ReactorContinuationRequest) Validate() error

type ReactorContinuationSnapshot 

type ReactorContinuationSnapshot struct {
	Request    ReactorContinuationRequest `json:"request"`
	Status     ReactorContinuationStatus  `json:"status"`
	Generation uint64                     `json:"generation"`
	CreatedAt  time.Time                  `json:"created_at"`
	UpdatedAt  time.Time                  `json:"updated_at"`
}

func (ReactorContinuationSnapshot) Validate 

func (s ReactorContinuationSnapshot) Validate() error

type ReactorContinuationStatus 

type ReactorContinuationStatus string
const (
	ReactorContinuationPending   ReactorContinuationStatus = "pending"
	ReactorContinuationStarted   ReactorContinuationStatus = "started"
	ReactorContinuationCompleted ReactorContinuationStatus = "completed"
)

func (ReactorContinuationStatus) Valid 

func (s ReactorContinuationStatus) Valid() bool

type ReactorDeliveryReceipt 

type ReactorDeliveryReceipt struct {
	Kind        ReactorDeliveryReceiptKind `json:"kind"`
	RunID       RunID                      `json:"run_id"`
	Update      *RunUpdateReceipt          `json:"update,omitempty"`
	RunStatus   RunStatus                  `json:"run_status,omitempty"`
	ProcessedAt time.Time                  `json:"processed_at"`
}

ReactorDeliveryReceipt records the one canonical consumption path. The first delivery is consumed by Host.StartRun; later deliveries are consumed only through the canonical tracked wait update.

func (ReactorDeliveryReceipt) Validate 

func (r ReactorDeliveryReceipt) Validate() error

type ReactorDeliveryReceiptKind 

type ReactorDeliveryReceiptKind string
const (
	ReactorDeliveryStartedRun  ReactorDeliveryReceiptKind = "started_run"
	ReactorDeliveryResumedWait ReactorDeliveryReceiptKind = "resumed_wait"
	ReactorDeliveryTerminalRun ReactorDeliveryReceiptKind = "terminal_run"
)

type ReactorDeliveryRequest 

type ReactorDeliveryRequest struct {
	ReactorID      string                 `json:"reactor_id"`
	IdempotencyKey string                 `json:"idempotency_key"`
	SignalName     string                 `json:"signal_name"`
	Payload        values.Value           `json:"payload"`
	Responder      workflowwait.Responder `json:"responder"`
	OccurredAt     time.Time              `json:"occurred_at"`
	ReceivedAt     time.Time              `json:"received_at"`
}

func (ReactorDeliveryRequest) Validate 

func (r ReactorDeliveryRequest) Validate() error

type ReactorDeliverySnapshot 

type ReactorDeliverySnapshot struct {
	Request           ReactorDeliveryRequest `json:"request"`
	ReactorGeneration uint64                 `json:"reactor_generation"`
	RunID             RunID                  `json:"run_id"`
	// StartsGeneration is store-derived only for the delivery that created
	// generation one. Its payload is consumed by ordinary Host.StartRun and
	// must not also be resumed into a wait.
	StartsGeneration bool                    `json:"starts_generation,omitempty"`
	ClaimedWaitID    WaitID                  `json:"claimed_wait_id,omitempty"`
	Status           ReactorDeliveryStatus   `json:"status"`
	Generation       uint64                  `json:"generation"`
	Receipt          *ReactorDeliveryReceipt `json:"receipt,omitempty"`
	CreatedAt        time.Time               `json:"created_at"`
	UpdatedAt        time.Time               `json:"updated_at"`
}

func (ReactorDeliverySnapshot) Validate 

func (s ReactorDeliverySnapshot) Validate() error

type ReactorDeliveryStatus 

type ReactorDeliveryStatus string
const (
	ReactorDeliveryPending  ReactorDeliveryStatus = "pending"
	ReactorDeliveryApplying ReactorDeliveryStatus = "applying"
	ReactorDeliveryApplied  ReactorDeliveryStatus = "applied"
	ReactorDeliveryClosed   ReactorDeliveryStatus = "closed"
)

func (ReactorDeliveryStatus) Valid 

func (s ReactorDeliveryStatus) Valid() bool

type ReactorIdentity 

type ReactorIdentity struct {
	ID                     string              `json:"id"`
	RegistrationID         string              `json:"registration_id"`
	RegistrationGeneration uint64              `json:"registration_generation"`
	Correlation            string              `json:"correlation"`
	Definition             graph.DefinitionRef `json:"definition"`
	Plan                   PlanRef             `json:"plan"`
	Provenance             graph.Provenance    `json:"provenance"`
}

ReactorIdentity is derived by a host from one exact source-owned activation registration. Callers never choose its plan, provenance, or logical ID.

func (ReactorIdentity) Validate 

func (i ReactorIdentity) Validate() error

type ReactorSnapshot 

type ReactorSnapshot struct {
	Identity            ReactorIdentity `json:"identity"`
	CurrentGeneration   uint64          `json:"current_generation"`
	CurrentRunID        RunID           `json:"current_run_id"`
	ContinueAfterEvents uint64          `json:"continue_after_events"`
	Status              ReactorStatus   `json:"status"`
	Generation          uint64          `json:"generation"`
	EventCount          uint64          `json:"event_count"`
	CreatedAt           time.Time       `json:"created_at"`
	UpdatedAt           time.Time       `json:"updated_at"`
}

func (ReactorSnapshot) Validate 

func (s ReactorSnapshot) Validate() error

type ReactorStatus 

type ReactorStatus string
const (
	ReactorStarting ReactorStatus = "starting"
	ReactorWaiting  ReactorStatus = "waiting"
	ReactorRolling  ReactorStatus = "rolling"
	ReactorFailed   ReactorStatus = "failed"
	ReactorClosed   ReactorStatus = "closed"
)

func (ReactorStatus) Valid 

func (s ReactorStatus) Valid() bool

type ReactorStore 

type ReactorStore interface {
	BeginReactorDelivery(context.Context, BeginReactorDeliveryRequest) (ReactorSnapshot, ReactorDeliverySnapshot, IdempotencyOutcome, error)
	LoadReactor(context.Context, string) (ReactorSnapshot, error)
	MarkReactorWaiting(context.Context, string, uint64, time.Time) (ReactorSnapshot, error)
	LoadReactorDelivery(context.Context, string, string) (ReactorDeliverySnapshot, error)
	ClaimReactorDelivery(context.Context, ClaimReactorDeliveryRequest) (ReactorDeliverySnapshot, error)
	ReleaseReactorDelivery(context.Context, ReleaseReactorDeliveryRequest) (ReactorDeliverySnapshot, error)
	CompleteReactorDelivery(context.Context, CompleteReactorDeliveryRequest) (ReactorSnapshot, ReactorDeliverySnapshot, error)
	FailReactor(context.Context, FailReactorRequest) (ReactorSnapshot, error)
	RecoverReactorDeliveries(context.Context, int) ([]ReactorDeliverySnapshot, error)
	RecoverReactors(context.Context, int) ([]ReactorSnapshot, error)
	BeginReactorContinuation(context.Context, ReactorContinuationRequest) (ReactorSnapshot, ReactorContinuationSnapshot, IdempotencyOutcome, error)
	CompleteReactorContinuation(context.Context, string, uint64, time.Time) (ReactorSnapshot, ReactorContinuationSnapshot, error)
	RecoverReactorContinuations(context.Context, int) ([]ReactorContinuationSnapshot, error)
}

type ReadinessDisposition 

type ReadinessDisposition string

ReadinessDisposition distinguishes an unresolved dependency set from work that may run and work that must be skipped. 

const (
	// ReadinessWaiting leaves the invocation durably blocked.
	ReadinessWaiting ReadinessDisposition = "waiting"
	// ReadinessReady permits dispatch after any predicate succeeds.
	ReadinessReady ReadinessDisposition = "ready"
	// ReadinessSkip durably prevents dispatch.
	ReadinessSkip ReadinessDisposition = "skip"
)

type ReadinessEvaluation 

type ReadinessEvaluation struct {
	Rule         graph.ReadyRule
	Disposition  ReadinessDisposition
	Dependencies []DependencyState
	Reason       *BlockedReason
}

ReadinessEvaluation is the deterministic result of one named rule.

func EvaluateReadiness 

func EvaluateReadiness(rule graph.ReadyRule, dependencies []DependencyState) (ReadinessEvaluation, error)

EvaluateReadiness applies the D10 Airflow-style rules. A handled hard failure is success-equivalent for rule evaluation, while skipped remains a distinct terminal outcome. The exact truth table is:

  • all_success: ready when all dependencies succeeded or were handled; skip as soon as any other terminal outcome appears; otherwise wait.
  • all_done: ready when every dependency is terminal; otherwise wait.
  • one_failed: ready as soon as one unhandled hard failure appears; skip once all dependencies are terminal without one; otherwise wait.
  • all_failed: ready only for a non-empty, all-unhandled-failure set; skip as soon as a non-failure terminal outcome appears; otherwise wait.
  • none_failed: skip as soon as an unhandled hard failure appears; ready once every dependency is terminal without one; otherwise wait.
  • always: ready immediately without waiting for dependency outcomes.

Empty dependency sets are ready for all_success, all_done, none_failed, and always, and skip for one_failed and all_failed. Hard failures are failed, timed_out, canceled, and crashed. This is progression only: FailureHandled is caller-supplied route state and does not implement catch/error routing.

type ReadyCandidate 

type ReadyCandidate struct {
	InvocationID NodeInvocationID
	Priority     int
	CreatedAt    time.Time
}

ReadyCandidate is the immutable scheduling metadata exposed to a host policy. Claim generations and leases remain owned by the StateStore and cannot be rewritten by policy code.

type ReadyClaim 

type ReadyClaim struct {
	Candidate ReadyCandidate
	Lease     ClaimLease
	Replayed  bool
}

ReadyClaim identifies the candidate and lease acquired by ClaimNext.

type ReadyClaimRequest 

type ReadyClaimRequest struct {
	RunID          RunID
	Owner          string
	Token          string
	IdempotencyKey string
	Now            time.Time
	LeaseUntil     time.Time
}

ReadyClaimRequest supplies host-owned claim identity and lease timing for one persisted ready-queue scan. RunID optionally restricts the scan. IdempotencyKey is scoped deterministically to each candidate before the underlying StateStore is called. Retrying the same request may replay its live acquired lease. A new acquisition after release or expiry requires a new IdempotencyKey.

type ReadyQueueCoordinator 

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

ReadyQueueCoordinator discovers and claims work exclusively through a StateStore. It retains no candidate or ownership state between calls.

func NewReadyQueueCoordinator 

func NewReadyQueueCoordinator(store StateStore, policy ReadyQueuePolicy) *ReadyQueueCoordinator

NewReadyQueueCoordinator constructs a stateless durable-ready-queue coordinator. A nil policy preserves FIFO ordering.

func NewResourceReadyQueueCoordinator 

func NewResourceReadyQueueCoordinator(store StateStore, policy ReadyQueuePolicy, admission SchedulerAdmissionPolicy) *ReadyQueueCoordinator

NewResourceReadyQueueCoordinator constructs the bounded cross-process admission path. Store must implement SchedulerResourceStore; validation is deferred to ClaimNext so typed-nil collaborators fail as structured errors.

func (*ReadyQueueCoordinator) ClaimNext 

func (c *ReadyQueueCoordinator) ClaimNext(ctx context.Context, request ReadyClaimRequest) (ReadyClaim, bool, error)

ClaimNext discovers persisted ready candidates, applies FIFO and optional host selection, then claims at most one candidate under claim-generation CAS. Expected lost CAS and negative claim races are skipped; all other store errors are returned. The boolean is false when no acquisition or live exact replay is available, in which case the returned ReadyClaim is empty.

func (*ReadyQueueCoordinator) Release 

func (c *ReadyQueueCoordinator) Release(ctx context.Context, request ReleaseClaimRequest) error

Release clears a matching lease through the durable store.

func (*ReadyQueueCoordinator) Renew 

func (c *ReadyQueueCoordinator) Renew(ctx context.Context, request RenewLeaseRequest) (ClaimLease, error)

Renew extends a matching live lease through the durable store.

type ReadyQueuePolicy 

type ReadyQueuePolicy interface {
	SelectReady(context.Context, []ReadyCandidate) ([]NodeInvocationID, error)
}

ReadyQueuePolicy may reorder or select from the FIFO candidate list. Every returned identity must occur exactly once in candidates. Returning no identities elects not to acquire new work from the current persisted snapshot. A matching live exact idempotency replay is prioritized ahead of returned identities because it is already owned work, not a new scheduling decision.

type ReadyQueuePolicyFunc 

type ReadyQueuePolicyFunc func(context.Context, []ReadyCandidate) ([]NodeInvocationID, error)

ReadyQueuePolicyFunc adapts a function to ReadyQueuePolicy.

func (ReadyQueuePolicyFunc) SelectReady 

func (f ReadyQueuePolicyFunc) SelectReady(ctx context.Context, candidates []ReadyCandidate) ([]NodeInvocationID, error)

SelectReady implements ReadyQueuePolicy.

type ReconcileCrashedAttemptRequest 

type ReconcileCrashedAttemptRequest struct {
	Attempt                   AttemptID             `json:"attempt"`
	ExpectedNodeGeneration    uint64                `json:"expected_node_generation"`
	ExpectedAttemptGeneration uint64                `json:"expected_attempt_generation"`
	IdempotencyKey            string                `json:"idempotency_key"`
	Decision                  CrashRecoveryDecision `json:"decision"`
	At                        time.Time             `json:"at"`
}

func (ReconcileCrashedAttemptRequest) Validate 

func (r ReconcileCrashedAttemptRequest) Validate() error

type ReconcileCrashedAttemptResult 

type ReconcileCrashedAttemptResult struct {
	Outcome    IdempotencyOutcome       `json:"outcome"`
	Node       NodeInvocationSnapshot   `json:"node"`
	Attempt    AttemptSnapshot          `json:"attempt"`
	Event      Event                    `json:"event"`
	Activation *RetryActivationSnapshot `json:"activation,omitempty"`
}

type RecordControlDecisionRequest 

type RecordControlDecisionRequest struct {
	Decision ControlDecisionSnapshot
	// ErrorValues is atomically persisted and bound to catch decisions by the
	// store. Callers never preallocate or choose the durable ValueSet identity.
	ErrorValues              values.ValueSet
	ExpectedSourceGeneration uint64
	At                       time.Time
}

type RecordControlDecisionResult 

type RecordControlDecisionResult struct {
	Outcome  IdempotencyOutcome
	Decision ControlDecisionSnapshot
	Event    *Event
}

type RecoverServiceStartRequest 

type RecoverServiceStartRequest struct {
	Start                     AttemptID
	Ref                       stepkind.ExternalOperationRef
	ExpectedServiceGeneration uint64
	ExpectedNodeGeneration    uint64
	ExpectedAttemptGeneration uint64
	At                        time.Time
}

RecoverServiceStartRequest attaches the exact reacquired provider reference after a crash. No worker claim is required; generations fence live workers.

type RecoveryCoordinator 

type RecoveryCoordinator struct {
	Store           StateStore
	Recovery        RecoveryStore
	Inputs          NodeInputStore
	Control         ControlFlowStore
	Plans           RecoveryPlanSource
	Registry        stepkind.Registry
	Policy          RepeatPolicy
	RetryAuthorizer RetryAuthorizer
	Policies        RunPolicyStore
	Waits           *WaitCoordinator
	Evaluator       PredicateEvaluator
}

RecoveryCoordinator reconstructs work exclusively from durable facts. Its order is: immutable control facts, interrupted attempts and their run-policy fences, waits/timers, due retries, then ordinary readiness. A host must run pinned child materialization and recursive cancellation-tree recovery before invoking it; Hadron's RecoveryHook boundary provides that ordering.

func (*RecoveryCoordinator) Recover 

func (c *RecoveryCoordinator) Recover(ctx context.Context, request RecoveryRequest) (RecoveryResult, error)

type RecoveryPlan 

type RecoveryPlan struct {
	Ref        PlanRef                     `json:"ref"`
	Plan       compile.ExecutionPlan       `json:"plan"`
	Visibility compile.ValueVisibilityPlan `json:"visibility"`
}

RecoveryPlan is the exact immutable graph bound to a persisted PlanRef. Implementations must load pinned source/plan material, never re-resolve a movable definition.

func (RecoveryPlan) Validate 

func (p RecoveryPlan) Validate() error

type RecoveryPlanSource 

type RecoveryPlanSource interface {
	LoadRecoveryPlan(context.Context, RunSnapshot) (RecoveryPlan, error)
}

RecoveryPlanSource loads exact, restart-safe plan material.

type RecoveryQuery 

type RecoveryQuery struct {
	RunID RunID
	Now   time.Time
	Limit int
}

RecoveryQuery selects deterministic recovery candidates. Limit applies independently to each result category; zero means unlimited.

type RecoveryRequest 

type RecoveryRequest struct {
	Now               time.Time
	Limit             int
	ExpressionOptions values.ExpressionOptions
}

RecoveryRequest supplies one deterministic recovery pass. Limit is applied independently to persisted query categories. ExpressionOptions retain the same evaluator semantics used during ordinary progression.

type RecoveryResult 

type RecoveryResult struct {
	Snapshot         RecoverySnapshot
	Services         []ServiceReconcileResult
	Waits            WaitRecoveryResults
	RetryActivations []ActivateNodeRetryResult
	Crashes          []ReconcileCrashedAttemptResult
	ControlDecisions []ControlDecisionSnapshot
	Progressed       []ProgressNodeResult
	FanOuts          []CompleteFanOutResult
	FanOutCanceled   []NodeInvocationSnapshot
	Compensations    []CompensationProgressResult
}

type RecoverySnapshot 

type RecoverySnapshot struct {
	ActiveRuns    []RunSnapshot
	Ready         []NodeInvocationSnapshot
	Running       []NodeInvocationSnapshot
	Waiting       []NodeInvocationSnapshot
	Leased        []NodeInvocationSnapshot
	ExpiredLeases []NodeInvocationSnapshot
	DueTimers     []WaitSnapshot
}

RecoverySnapshot contains only persisted candidates; category membership may overlap. Ready is a storage candidate set whose order is not a scheduling contract. Recovery does not schedule or transition records.

type RecoveryStore 

type RecoveryStore interface {
	ReconcileCrashedAttempt(context.Context, ReconcileCrashedAttemptRequest) (ReconcileCrashedAttemptResult, error)
	ListRunInvocations(context.Context, RunID) ([]NodeInvocationSnapshot, error)
	ReplayStore
}

RecoveryStore owns the atomic crash and replay persistence operations added by W03-T06 without widening the foundational StateStore interface.

type ReleaseClaimRequest 

type ReleaseClaimRequest struct {
	InvocationID NodeInvocationID
	Owner        string
	Token        string
	Generation   uint64
	Now          time.Time
}

ReleaseClaimRequest clears the matching lease at Now. The next claimant must use the released claim generation as its expected generation.

type ReleaseReactorDeliveryRequest 

type ReleaseReactorDeliveryRequest struct {
	ReactorID          string    `json:"reactor_id"`
	IdempotencyKey     string    `json:"idempotency_key"`
	ExpectedGeneration uint64    `json:"expected_generation"`
	At                 time.Time `json:"at"`
}

type RenderedEvent 

type RenderedEvent struct {
	Sequence   uint64                `json:"sequence"`
	RunID      RunID                 `json:"run_id"`
	Invocation *NodeInvocationID     `json:"invocation,omitempty"`
	Attempt    *AttemptID            `json:"attempt,omitempty"`
	Type       string                `json:"type"`
	OccurredAt time.Time             `json:"occurred_at"`
	Attributes map[string]string     `json:"attributes,omitempty"`
	Values     *values.ValueSetRef   `json:"values,omitempty"`
	Redaction  values.RedactionClass `json:"redaction"`
	Retention  values.RetentionClass `json:"retention"`
	Masked     bool                  `json:"masked"`
}

RenderedEvent is the stable display projection shared by CLI, HTTP, MCP, A2A, and UI callers. Value references contain no payload and remain visible.

func RenderEvent 

func RenderEvent(event Event, policy values.DisplayPolicy) (RenderedEvent, error)

RenderEvent applies the same private policy as value rendering. Secret event attributes are always masked; private attributes require explicit reveal.

type RenewLeaseRequest 

type RenewLeaseRequest struct {
	InvocationID NodeInvocationID
	Owner        string
	Token        string
	Generation   uint64
	Now          time.Time
	LeaseUntil   time.Time
}

RenewLeaseRequest extends the matching, unexpired lease without changing its claim generation or the invocation's semantic generation/timestamp.

type RepeatCandidate 

type RepeatCandidate struct {
	Operation  RepeatOperation        `json:"operation"`
	Run        RunSnapshot            `json:"run"`
	Node       NodeInvocationSnapshot `json:"node"`
	Attempt    *AttemptSnapshot       `json:"attempt,omitempty"`
	Definition graph.Node             `json:"definition"`
	Spec       stepkind.StepKindSpec  `json:"spec"`
	Effects    graph.EffectSet        `json:"effects"`
	// IdempotencyKey is the exact compiler-scoped durable evaluation of a
	// keyed node declaration. It is empty for intrinsic/unsupported modes.
	IdempotencyKey string `json:"idempotency_key,omitempty"`
}

RepeatCandidate binds every policy decision to the pinned graph node, the frozen registered kind spec, and the authoritative durable attempt.

type RepeatOperation 

type RepeatOperation string

RepeatOperation distinguishes crash uncertainty from an explicit replay of a completed durable invocation. 

const (
	RepeatCrashRecovery RepeatOperation = "crash_recovery"
	RepeatReplay        RepeatOperation = "replay"
)

type RepeatPolicy 

type RepeatPolicy interface {
	EvaluateRepeat(context.Context, RepeatCandidate) (RepeatPolicyDecision, error)
}

RepeatPolicy is the host authorization seam for consequential recovery and replay. Core retry-safety and idempotency floors remain non-overridable.

type RepeatPolicyDecision 

type RepeatPolicyDecision struct {
	Allow      bool              `json:"allow"`
	Code       string            `json:"code"`
	Reason     string            `json:"reason"`
	Attributes map[string]string `json:"attributes,omitempty"`
}

RepeatPolicyDecision is a persistence-safe policy fact. Code, Reason, and Attributes must not contain resolved secrets.

func (RepeatPolicyDecision) Validate 

func (d RepeatPolicyDecision) Validate() error

type RepeatPolicyFunc 

type RepeatPolicyFunc func(context.Context, RepeatCandidate) (RepeatPolicyDecision, error)

RepeatPolicyFunc adapts a function to RepeatPolicy.

func (RepeatPolicyFunc) EvaluateRepeat 

func (f RepeatPolicyFunc) EvaluateRepeat(ctx context.Context, candidate RepeatCandidate) (RepeatPolicyDecision, error)

type ReplayCompensationAuthorization 

type ReplayCompensationAuthorization struct {
	LedgerGeneration uint64              `json:"ledger_generation"`
	LedgerOutcome    CompensationOutcome `json:"ledger_outcome,omitempty"`
	RiskDigest       string              `json:"risk_digest,omitempty"`
	Digest           string              `json:"digest"`
}

ReplayCompensationAuthorization is the bounded host-issued proof for replaying a source whose rollback is indeterminate or did not succeed. It binds the exact immutable source-ledger observation without retaining the caller's free-form justification.

func (ReplayCompensationAuthorization) Validate 

func (a ReplayCompensationAuthorization) Validate() error

type ReplayFanOutBinding 

type ReplayFanOutBinding struct {
	Source  FanOutSnapshot     `json:"source"`
	Target  FanOutSnapshot     `json:"target"`
	Results []FanOutItemResult `json:"results"`
}

func (ReplayFanOutBinding) Validate 

func (b ReplayFanOutBinding) Validate(source, target RunID) error

type ReplayNodeBinding 

type ReplayNodeBinding struct {
	Source   NodeInvocationSnapshot    `json:"source"`
	Target   NodeInvocationID          `json:"target"`
	Reuse    bool                      `json:"reuse"`
	Attempts []AttemptSnapshot         `json:"attempts,omitempty"`
	Control  []ControlDecisionSnapshot `json:"control,omitempty"`
}

ReplayNodeBinding binds one exact immutable source invocation to its new run identity. Reuse means the terminal status and value references are journaled into an unclaimable replay node; otherwise the target starts pending with no attempt. Attempts remain claim/dispatch-owned, so "fresh attempts" means a fresh invocation history beginning at LatestAttempt zero.

func (ReplayNodeBinding) Validate 

func (b ReplayNodeBinding) Validate(source, target RunID) error

type ReplayNodePolicy 

type ReplayNodePolicy struct {
	Invocation NodeInvocationID     `json:"invocation"`
	Attempt    *AttemptID           `json:"attempt,omitempty"`
	Decision   RepeatPolicyDecision `json:"decision"`
}

type ReplayProvenance 

type ReplayProvenance struct {
	RunID                     RunID                            `json:"run_id"`
	SourceRunID               RunID                            `json:"source_run_id"`
	FromNodeID                string                           `json:"from_node_id"`
	PlanDigest                string                           `json:"plan_digest"`
	IdempotencyKey            string                           `json:"idempotency_key"`
	Policy                    []ReplayNodePolicy               `json:"policy"`
	CompensationAuthorization *ReplayCompensationAuthorization `json:"compensation_authorization,omitempty"`
	CreatedAt                 time.Time                        `json:"created_at"`
}

ReplayProvenance is the immutable durable explanation of one replay run.

func (ReplayProvenance) Validate 

func (p ReplayProvenance) Validate() error

type ReplayRequest 

type ReplayRequest struct {
	SourceRunID               RunID
	RunID                     RunID
	FromNodeID                string
	IdempotencyKey            string
	CompensationAuthorization *ReplayCompensationAuthorization
	At                        time.Time
}

type ReplayService 

type ReplayService struct {
	Store     StateStore
	Replay    ReplayStore
	Inputs    NodeInputStore
	Control   ControlFlowStore
	Plans     RecoveryPlanSource
	Registry  stepkind.Registry
	Policy    RepeatPolicy
	Evaluator PredicateEvaluator
}

ReplayService validates the complete downstream effect surface before the atomic store mutation and then rebuilds ready state from the pinned graph.

func (*ReplayService) Rerun 

func (s *ReplayService) Rerun(ctx context.Context, request ReplayRequest) (BeginReplayResult, error)

type ReplayStore 

type ReplayStore interface {
	BeginReplay(context.Context, BeginReplayRequest) (BeginReplayResult, error)
	LoadReplayProvenance(context.Context, RunID) (ReplayProvenance, error)
	ListRunInvocations(context.Context, RunID) ([]NodeInvocationSnapshot, error)
}

ReplayStore atomically binds provenance and creates the new run/node state. Failed or losing operations leave no run, event, or provenance residue.

type RequestExternalOperationCancelRequest 

type RequestExternalOperationCancelRequest struct {
	Attempt                     AttemptID
	ExpectedOperationGeneration uint64
	At                          time.Time
}

RequestExternalOperationCancelRequest durably records cancellation intent before the adapter call. Recovery retries pending intent after a crash.

func (RequestExternalOperationCancelRequest) Validate 

func (r RequestExternalOperationCancelRequest) Validate() error

type RequestExternalOperationCancelResult 

type RequestExternalOperationCancelResult struct {
	Operation ExternalOperationSnapshot
	Event     *Event
}

type RequestRunCancellationRequest 

type RequestRunCancellationRequest struct {
	RunID              RunID
	ExpectedGeneration uint64
	IdempotencyKey     string
	Reason             Failure
	At                 time.Time
}

RequestRunCancellationRequest atomically begins cancellation and is idempotent by key. Reason is safe-to-persist structured metadata.

func (RequestRunCancellationRequest) Validate 

func (r RequestRunCancellationRequest) Validate() error

type RequestRunCancellationResult 

type RequestRunCancellationResult struct {
	Outcome IdempotencyOutcome
	Run     RunSnapshot
	Nodes   []NodeInvocationSnapshot
	Intents []CancellationIntentSnapshot
	Events  []Event
}

type RequestRunCancellationWithFinalizersRequest 

type RequestRunCancellationWithFinalizersRequest struct {
	Cancellation         RequestRunCancellationRequest
	Finalizers           []FinalizerScope
	CompensationRequired bool
	ErrorValues          values.ValueSet
	// Descendants is the canonical RunID-ordered plan for every recursively
	// reachable ParentCloseCancel child. Entries with no finalizers are
	// explicit so the store can validate the complete local cancellation tree
	// before mutating any run.
	Descendants []CancellationDescendantPlan
}

func (RequestRunCancellationWithFinalizersRequest) Validate 

func (r RequestRunCancellationWithFinalizersRequest) Validate() error

Validate reports malformed cancellation-tree transport. State stores also compare Descendants with the exact reachable ParentCloseCancel closure.

type RequestRunCancellationWithFinalizersResult 

type RequestRunCancellationWithFinalizersResult struct {
	Cancellation RequestRunCancellationResult
	// Intent retains the root intent for the common root-finalizer case. It is
	// zero when only descendants declare cleanup.
	Intent          TerminalIntentSnapshot
	TerminalIntents []TerminalIntentSnapshot
}

type ResolveCancellationIntentRequest 

type ResolveCancellationIntentRequest struct {
	IntentID           string
	ExpectedGeneration uint64
	At                 time.Time
}

func (ResolveCancellationIntentRequest) Validate 

func (r ResolveCancellationIntentRequest) Validate() error

type ResolveCancellationIntentResult 

type ResolveCancellationIntentResult struct {
	Intent  CancellationIntentSnapshot
	Node    *NodeInvocationSnapshot
	Attempt *AttemptSnapshot
	Event   *Event
}

type ResumeCommand 

type ResumeCommand struct {
	WaitID         WaitID
	Correlation    string
	Token          string
	WakeSource     workflowwait.WakeSource
	Responder      workflowwait.Responder
	Payload        values.Value
	IdempotencyKey string
	ReceivedAt     time.Time
}

ResumeCommand is the adapter-facing raw response. Token is hashed before it reaches persistence.

func (ResumeCommand) Validate 

func (c ResumeCommand) Validate() error

type ResumeNodeWaitRequest 

type ResumeNodeWaitRequest struct {
	WaitID                    WaitID
	ExpectedWaitGeneration    uint64
	ExpectedNodeGeneration    uint64
	ExpectedAttemptGeneration uint64
	Correlation               string
	PresentedTokenDigest      string
	WakeSource                workflowwait.WakeSource
	Responder                 workflowwait.Responder
	Payload                   values.Value
	IdempotencyKey            string
	ReceivedAt                time.Time
}

ResumeNodeWaitRequest is the fully validated input to an atomic store. It carries only the presented token digest, never the raw token.

func (ResumeNodeWaitRequest) Validate 

func (r ResumeNodeWaitRequest) Validate() error

type ResumeOutcome 

type ResumeOutcome string
const (
	ResumeApplied        ResumeOutcome = "applied"
	ResumeReplayed       ResumeOutcome = "replayed"
	ResumeAlreadyResumed ResumeOutcome = "already_resumed"
	ResumeClosed         ResumeOutcome = "closed"
)

type ResumeWaitResult 

type ResumeWaitResult struct {
	Outcome ResumeOutcome
	Wait    WaitSnapshot
	Node    NodeInvocationSnapshot
	Attempt AttemptSnapshot
	Values  values.ValueSetRef
	Events  []Event
}

type RetentionGroup 

type RetentionGroup struct {
	Class values.RetentionClass `json:"class"`
	Names []string              `json:"names"`
}

RetentionGroup lists names sharing one retention class. Names are sorted; groups use the closed class order run, project, external.

type RetentionHook 

type RetentionHook interface {
	BeforePersist(context.Context, RetentionPlan) error
	AfterPersist(context.Context, RetentionRecord) error
}

RetentionHook lets a host prepare and record lifecycle ownership without entering the core StateStore contract. Run values are eligible for run-end cleanup; project values follow project policy; external payload lifetimes remain externally owned and must not be deleted by the workflow store.

type RetentionHookError 

type RetentionHookError struct {
	Phase string
	Ref   *values.ValueSetRef
	Cause error
}

RetentionHookError identifies the failed phase. Ref is present for an after failure so a host can find an unreferenced immutable set for cleanup.

func (*RetentionHookError) Error 

func (e *RetentionHookError) Error() string

func (*RetentionHookError) Unwrap 

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

Unwrap exposes the stable sentinel and the callback cause.

type RetentionHookFuncs 

type RetentionHookFuncs struct {
	Before func(context.Context, RetentionPlan) error
	After  func(context.Context, RetentionRecord) error
}

RetentionHookFuncs adapts optional callbacks to RetentionHook.

func (RetentionHookFuncs) AfterPersist 

func (f RetentionHookFuncs) AfterPersist(ctx context.Context, record RetentionRecord) error

AfterPersist calls the optional callback.

func (RetentionHookFuncs) BeforePersist 

func (f RetentionHookFuncs) BeforePersist(ctx context.Context, plan RetentionPlan) error

BeforePersist calls the optional callback.

type RetentionPlan 

type RetentionPlan struct {
	Owner  ValueOwner       `json:"owner"`
	Groups []RetentionGroup `json:"groups"`
}

RetentionPlan describes classification before a value set is persisted. Mixed retained classes are intentionally supported.

type RetentionRecord 

type RetentionRecord struct {
	Plan RetentionPlan      `json:"plan"`
	Ref  values.ValueSetRef `json:"ref"`
}

RetentionRecord adds the immutable persisted reference to its plan.

type RetryActivationQuery 

type RetryActivationQuery struct {
	RunID     RunID
	DueBefore time.Time
	Limit     int
}

type RetryActivationSnapshot 

type RetryActivationSnapshot struct {
	ID         string                `json:"id"`
	Attempt    AttemptID             `json:"attempt"`
	Failure    Failure               `json:"failure"`
	FireAt     time.Time             `json:"fire_at"`
	Status     RetryActivationStatus `json:"status"`
	Generation uint64                `json:"generation"`
	CreatedAt  time.Time             `json:"created_at"`
	UpdatedAt  time.Time             `json:"updated_at"`
}

RetryActivationSnapshot is the restart-durable timer produced when one unsuccessful attempt is closed for a later retry. Attempt names the closed attempt; activation creates exactly Attempt.Number+1 when the node is next claimed and dispatched.

func (RetryActivationSnapshot) Validate 

func (s RetryActivationSnapshot) Validate() error

Validate checks durable retry state without deciding retry eligibility.

type RetryActivationStatus 

type RetryActivationStatus string

RetryActivationStatus is the closed durable lifecycle for one retry timer. 

const (
	RetryScheduled RetryActivationStatus = "scheduled"
	RetryActivated RetryActivationStatus = "activated"
	RetryCanceled  RetryActivationStatus = "canceled"
)

func (RetryActivationStatus) Valid 

func (s RetryActivationStatus) Valid() bool

Valid reports whether s is a supported retry-activation state.

type RetryAuthorizationRequest 

type RetryAuthorizationRequest struct {
	Node           graph.Node
	Spec           stepkind.StepKindSpec
	AttemptNumber  int
	Failure        Failure
	AttemptStatus  NodeStatus
	IdempotencyKey string
}

RetryAuthorizationRequest is the narrow host-policy seam for retrying externally consequential effects. The host returns nil to grant the retry.

type RetryAuthorizer 

type RetryAuthorizer interface {
	AuthorizeRetry(context.Context, RetryAuthorizationRequest) error
}

RetryAuthorizer may grant an otherwise policy-gated mutate, materialize, or destructive retry. It performs no persistence or adapter I/O.

type RetryCompensationRequest 

type RetryCompensationRequest struct {
	RunID                    RunID
	ExpectedLedgerGeneration uint64
	IdempotencyKey           string
	Attestation              string
	At                       time.Time
}

type RetryCoordinator 

type RetryCoordinator struct {
	Store     RetryStore
	Evaluator RetryEvaluator
	Scheduler workflowwait.ActivationScheduler
}

RetryCoordinator evaluates an unsuccessful attempt, atomically closes it into a durable retry wait, and only then notifies the optional host timer. A scheduler failure cannot undo the persisted activation; recovery remains authoritative and may schedule it again idempotently.

func (RetryCoordinator) Schedule 

func (c RetryCoordinator) Schedule(ctx context.Context, command ScheduleRetryCommand) (ScheduleNodeRetryResult, RetryDecision, error)

Schedule returns a non-retry decision without mutating state. A true retry decision always accompanies a durable result, even if host scheduling then returns an operational error.

type RetryDecision 

type RetryDecision struct {
	Retry        bool
	Reason       string
	MatchedClass string
	Delay        time.Duration
	FireAt       time.Time
}

RetryDecision is deterministic for a request and policy decision. Retry is false for normal policy exhaustion as well as denied unsafe retries; Reason is a stable searchable explanation.

type RetryEvaluationRequest 

type RetryEvaluationRequest struct {
	Node           graph.Node
	Spec           stepkind.StepKindSpec
	AttemptNumber  int
	Failure        Failure
	AttemptStatus  NodeStatus
	Timeout        TimeoutKind
	IdempotencyKey string
	FailedAt       time.Time
}

RetryEvaluationRequest contains all immutable facts used to decide whether another attempt may be scheduled.

type RetryEvaluator 

type RetryEvaluator struct {
	Authorizer RetryAuthorizer
}

RetryEvaluator implements graph retry policy without retaining state.

func (RetryEvaluator) Evaluate 

func (e RetryEvaluator) Evaluate(ctx context.Context, request RetryEvaluationRequest) (RetryDecision, error)

Evaluate validates the policy and returns an effect-aware retry decision.

type RetryStore 

type RetryStore interface {
	LoadRetryActivation(context.Context, string) (RetryActivationSnapshot, error)
	ScheduleNodeRetry(context.Context, ScheduleNodeRetryRequest) (ScheduleNodeRetryResult, error)
	ActivateNodeRetry(context.Context, ActivateNodeRetryRequest) (ActivateNodeRetryResult, error)
	RecoverRetryActivations(context.Context, RetryActivationQuery) ([]RetryActivationSnapshot, error)
}

RetryStore is the atomic persistence surface for retry timers.

type ReuseAuthority 

type ReuseAuthority struct {
	Principal  string            `json:"principal"`
	Scope      string            `json:"scope,omitempty"`
	Attributes map[string]string `json:"attributes,omitempty"`
}

ReuseAuthority is the application-neutral caller identity presented to host policy. Attributes must never contain credentials or resolved secrets.

func (ReuseAuthority) Validate 

func (a ReuseAuthority) Validate() error

type ReuseAuthorizer 

type ReuseAuthorizer interface {
	AuthorizeReuse(context.Context, ReuseCandidate) (ReusePolicyDecision, error)
}

ReuseAuthorizer is the host policy seam. Read/compute memoization does not require it; materialize memoization and every caller-selected pin do.

type ReuseAuthorizerFunc 

type ReuseAuthorizerFunc func(context.Context, ReuseCandidate) (ReusePolicyDecision, error)

func (ReuseAuthorizerFunc) AuthorizeReuse 

func (f ReuseAuthorizerFunc) AuthorizeReuse(ctx context.Context, c ReuseCandidate) (ReusePolicyDecision, error)

type ReuseCandidate 

type ReuseCandidate struct {
	Origin             InvocationOrigin      `json:"origin"`
	Target             NodeInvocationID      `json:"target"`
	PlanDigest         string                `json:"plan_digest"`
	Definition         graph.Node            `json:"definition"`
	Spec               stepkind.StepKindSpec `json:"spec"`
	Effects            graph.EffectSet       `json:"effects"`
	Outputs            values.ValueSetRef    `json:"outputs"`
	OutputSchemaDigest string                `json:"output_schema_digest"`
	Source             *NodeInvocationID     `json:"source,omitempty"`
	Authority          ReuseAuthority        `json:"authority"`
}

ReuseCandidate is the exact non-secret policy input for memo/pin reuse.

type ReuseNodeOutputsRequest 

type ReuseNodeOutputsRequest struct {
	InvocationID       NodeInvocationID   `json:"invocation_id"`
	ExpectedGeneration uint64             `json:"expected_generation"`
	Claim              ClaimProof         `json:"claim"`
	Origin             InvocationOrigin   `json:"origin"`
	Outputs            values.ValueSetRef `json:"outputs"`
	Source             NodeInvocationID   `json:"source"`
	// MemoEntryKey and SourceAttempt bind memoized reuse to the exact immutable
	// cache publication selected by the dispatcher. They are empty for pins,
	// which are instead fenced by the target's durable PinBinding.
	MemoEntryKey   string              `json:"memo_entry_key,omitempty"`
	SourceAttempt  *AttemptID          `json:"source_attempt,omitempty"`
	SourceOrigin   InvocationOrigin    `json:"source_origin"`
	PlanDigest     string              `json:"plan_digest"`
	Policy         ReusePolicyDecision `json:"policy"`
	IdempotencyKey string              `json:"idempotency_key"`
	At             time.Time           `json:"at"`
}

func (ReuseNodeOutputsRequest) Validate 

func (r ReuseNodeOutputsRequest) Validate() error

type ReuseNodeOutputsResult 

type ReuseNodeOutputsResult struct {
	Outcome IdempotencyOutcome     `json:"outcome"`
	Node    NodeInvocationSnapshot `json:"node"`
	Event   Event                  `json:"event"`
}

type ReusePolicyDecision 

type ReusePolicyDecision struct {
	Allow      bool              `json:"allow"`
	Code       string            `json:"code"`
	Reason     string            `json:"reason"`
	Attributes map[string]string `json:"attributes,omitempty"`
}

ReusePolicyDecision is an append-only, persistence-safe authorization fact.

func (ReusePolicyDecision) Validate 

func (d ReusePolicyDecision) Validate() error

type RunControlStore 

type RunControlStore interface {
	QueryRunState(context.Context, RunStateQuery) (RunStateView, error)
	FindOpenSignalWait(context.Context, SignalSelector) (WaitSnapshot, error)
	FindSignalWait(context.Context, SignalSelector, string) (WaitSnapshot, error)
	BeginRunUpdate(context.Context, BeginRunUpdateRequest) (RunUpdateSnapshot, IdempotencyOutcome, error)
	CompleteRunUpdate(context.Context, CompleteRunUpdateRequest) (RunUpdateSnapshot, error)
	LoadRunUpdate(context.Context, string) (RunUpdateSnapshot, error)
	RecoverRunUpdates(context.Context, int) ([]RunUpdateSnapshot, error)
}

RunControlStore adds no alternate lifecycle mutation. Its only write records and seals tracked update intent around WaitStore.ResumeNodeWait.

type RunExplanation 

type RunExplanation struct {
	Run            RunSnapshot               `json:"run"`
	Invocations    []InvocationExplanation   `json:"invocations"`
	Events         []Event                   `json:"events"`
	Decisions      []ControlDecisionSnapshot `json:"decisions,omitempty"`
	TerminalIntent *TerminalIntentSnapshot   `json:"terminal_intent,omitempty"`
	Replay         *ReplayProvenance         `json:"replay,omitempty"`
	Recovery       RecoverySnapshot          `json:"recovery"`
}

RunExplanation is a restart-stable projection of state plus append-only facts. Events explain chronology; snapshots remain lifecycle authority.

type RunFailureDisposition 

type RunFailureDisposition string
const (
	RunFailureHandled        RunFailureDisposition = "handled"
	RunFailureContinue       RunFailureDisposition = "continue"
	RunFailureFailFast       RunFailureDisposition = "fail_fast"
	RunFailureAlreadyDecided RunFailureDisposition = "already_decided"
)

type RunID 

type RunID string

RunID is an opaque host-independent run identifier.

type RunPolicyCoordinator 

type RunPolicyCoordinator struct {
	Store    StateStore
	Control  ControlFlowStore
	Policies RunPolicyStore
}

RunPolicyCoordinator applies explicit graph completion policy after an unhandled aggregate node failure. Fan-out children are intentionally ignored until their durable aggregate applies the configured tolerance. Locally hosted request-cancel children reuse durable cancellation intents; exact propagation to separately hosted direct-cancel children remains a host recovery concern because this extraction-safe store has no child graph port.

func NewRunPolicyCoordinator 

func NewRunPolicyCoordinator(store StateStore, control ControlFlowStore, policies RunPolicyStore) *RunPolicyCoordinator

func (*RunPolicyCoordinator) HandleFailure 

func (c *RunPolicyCoordinator) HandleFailure(ctx context.Context, workflow graph.Graph, source NodeInvocationID, idempotencyKey string, at time.Time) (ApplyRunFailurePolicyResult, error)

func (*RunPolicyCoordinator) HandleRunFailure 

func (c *RunPolicyCoordinator) HandleRunFailure(ctx context.Context, request HandleRunFailureRequest) (ApplyRunFailurePolicyResult, error)

HandleRunFailure applies graph completion policy to one durable aggregate failure. Fan-out item invocations are rejected until their aggregate has applied its tolerance policy.

type RunPolicyDecisionSnapshot 

type RunPolicyDecisionSnapshot struct {
	RunID            RunID                   `json:"run_id"`
	Mode             graph.RunCompletionMode `json:"mode"`
	Trigger          NodeInvocationID        `json:"trigger"`
	SourceGeneration uint64                  `json:"source_generation"`
	IntendedStatus   RunStatus               `json:"intended_status"`
	IdempotencyKey   string                  `json:"idempotency_key"`
	Generation       uint64                  `json:"generation"`
	CreatedAt        time.Time               `json:"created_at"`
}

RunPolicyDecisionSnapshot is the immutable winning fail-fast trigger. The existing TerminalIntentSnapshot is the only run fence and cleanup owner; this record exists for inspection and normal concurrent-failure convergence.

func (RunPolicyDecisionSnapshot) Validate 

func (s RunPolicyDecisionSnapshot) Validate() error

type RunPolicyStore 

type RunPolicyStore interface {
	LoadRunPolicyDecision(context.Context, RunID) (RunPolicyDecisionSnapshot, error)
	ApplyRunFailurePolicy(context.Context, ApplyRunFailurePolicyRequest) (ApplyRunFailurePolicyResult, error)
}

type RunSnapshot 

type RunSnapshot struct {
	ID         RunID               `json:"id"`
	Plan       PlanRef             `json:"plan"`
	Status     RunStatus           `json:"status"`
	Inputs     *values.ValueSetRef `json:"inputs,omitempty"`
	Outputs    *values.ValueSetRef `json:"outputs,omitempty"`
	Generation uint64              `json:"generation"`
	CreatedAt  time.Time           `json:"created_at"`
	UpdatedAt  time.Time           `json:"updated_at"`
}

RunSnapshot is one persisted run view. Generation is a store-managed CAS revision, not a lifecycle state-machine version.

func (RunSnapshot) Validate 

func (s RunSnapshot) Validate() error

Validate checks record integrity without deciding whether a transition into Status is legal.

type RunStateQuery 

type RunStateQuery struct {
	RunID         RunID  `json:"run_id"`
	AfterSequence uint64 `json:"after_sequence,omitempty"`
	Limit         int    `json:"limit"`
}

RunStateQuery is a strictly bounded, read-only operational projection. Limit applies independently to node, attempt, wait, and event histories.

func (RunStateQuery) Validate 

func (q RunStateQuery) Validate() error

type RunStateView 

type RunStateView struct {
	Run      RunSnapshot              `json:"run"`
	Nodes    []NodeInvocationSnapshot `json:"nodes,omitempty"`
	Attempts []AttemptSnapshot        `json:"attempts,omitempty"`
	Waits    []WaitSnapshot           `json:"waits,omitempty"`
	Events   []Event                  `json:"events,omitempty"`
}

type RunStatus 

type RunStatus string

RunStatus is the persisted state of a workflow run. 

const (
	RunPending   RunStatus = "pending"
	RunRunning   RunStatus = "running"
	RunWaiting   RunStatus = "waiting"
	RunSucceeded RunStatus = "succeeded"
	RunFailed    RunStatus = "failed"
	RunCanceled  RunStatus = "canceled"
	RunTimedOut  RunStatus = "timed_out"
	RunCrashed   RunStatus = "crashed"
)

func (RunStatus) Active 

func (s RunStatus) Active() bool

Active reports whether a run belongs in active-run recovery queries.

func (RunStatus) CanTransitionTo 

func (s RunStatus) CanTransitionTo(next RunStatus) bool

CanTransitionTo reports whether the run lifecycle permits an applied status change. Same-state idempotency is evaluated separately against state data.

func (RunStatus) Terminal 

func (s RunStatus) Terminal() bool

Terminal reports whether a run status cannot transition to another status.

func (RunStatus) Valid 

func (s RunStatus) Valid() bool

Valid reports whether s is a supported persisted run status.

type RunTransitionRequest 

type RunTransitionRequest struct {
	RunID              RunID
	ExpectedGeneration uint64
	To                 RunStatus
	Outputs            *values.ValueSetRef
	At                 time.Time
}

RunTransitionRequest changes only lifecycle-owned run fields under CAS.

type RunTransitionResult 

type RunTransitionResult struct {
	Snapshot RunSnapshot
	Outcome  TransitionOutcome
	Event    *Event
}

RunTransitionResult is the atomic run snapshot and event outcome. Event is nil only for an exact no-op.

type RunUpdateReceipt 

type RunUpdateReceipt struct {
	Outcome    ResumeOutcome       `json:"outcome"`
	WaitID     WaitID              `json:"wait_id"`
	WaitStatus WaitStatus          `json:"wait_status"`
	ResolvedAt time.Time           `json:"resolved_at,omitempty"`
	Values     *values.ValueSetRef `json:"values,omitempty"`
}

RunUpdateReceipt is intentionally compact: detailed node/attempt/event history remains available only through the bounded query path.

func (RunUpdateReceipt) Validate 

func (r RunUpdateReceipt) Validate() error

type RunUpdateSnapshot 

type RunUpdateSnapshot struct {
	Request    BeginRunUpdateRequest `json:"request"`
	Status     RunUpdateStatus       `json:"status"`
	Receipt    *RunUpdateReceipt     `json:"receipt,omitempty"`
	Generation uint64                `json:"generation"`
	CreatedAt  time.Time             `json:"created_at"`
	UpdatedAt  time.Time             `json:"updated_at"`
}

func (RunUpdateSnapshot) Validate 

func (s RunUpdateSnapshot) Validate() error

type RunUpdateStatus 

type RunUpdateStatus string

RunUpdateStatus is the durable tracked-operation lifecycle around the canonical wait-resume mutation. 

const (
	RunUpdatePending RunUpdateStatus = "pending"
	RunUpdateApplied RunUpdateStatus = "applied"
	RunUpdateClosed  RunUpdateStatus = "closed"
)

func (RunUpdateStatus) Valid 

func (s RunUpdateStatus) Valid() bool

type SaveNodeInvocationRequest 

type SaveNodeInvocationRequest struct {
	Snapshot           NodeInvocationSnapshot
	ExpectedGeneration uint64
}

SaveNodeInvocationRequest stores non-lifecycle invocation fields under record CAS. Status, attempt, and claim fields have dedicated atomic methods.

type SaveRunRequest 

type SaveRunRequest struct {
	Snapshot           RunSnapshot
	ExpectedGeneration uint64
}

SaveRunRequest stores non-lifecycle run fields when ExpectedGeneration matches. Status changes must use TransitionRun.

type SaveValuesRequest 

type SaveValuesRequest struct {
	Owner  ValueOwner
	Values values.ValueSet
}

SaveValuesRequest persists a defensively copied value set for an owner.

type ScheduleNodeRetryRequest 

type ScheduleNodeRetryRequest struct {
	Activation                RetryActivationSnapshot
	ExpectedNodeGeneration    uint64
	ExpectedAttemptGeneration uint64
	Claim                     ClaimProof
	AttemptStatus             NodeStatus
	At                        time.Time
}

ScheduleNodeRetryRequest atomically closes the current attempt, releases its claim, sets the aggregate node waiting, and inserts one scheduled timer.

func (ScheduleNodeRetryRequest) Validate 

func (r ScheduleNodeRetryRequest) Validate() error

type ScheduleNodeRetryResult 

type ScheduleNodeRetryResult struct {
	Activation RetryActivationSnapshot
	Node       NodeInvocationSnapshot
	Attempt    AttemptSnapshot
	Events     []Event
}

type ScheduleRetryCommand 

type ScheduleRetryCommand struct {
	Node           graph.Node
	Spec           stepkind.StepKindSpec
	NodeSnapshot   NodeInvocationSnapshot
	Attempt        AttemptSnapshot
	Claim          ClaimProof
	Failure        Failure
	AttemptStatus  NodeStatus
	Timeout        TimeoutKind
	IdempotencyKey string
	At             time.Time
}

ScheduleRetryCommand binds a policy evaluation to the currently fenced running attempt.

type SchedulerAdmissionPolicy 

type SchedulerAdmissionPolicy interface {
	Requirements(context.Context, ReadyCandidate) ([]SchedulerResourceRequirement, error)
}

SchedulerAdmissionPolicy supplies immutable resource requirements for one persisted candidate. Implementations may load pinned graph/spec metadata but must not perform side effects.

type SchedulerAdmissionPolicyFunc 

type SchedulerAdmissionPolicyFunc func(context.Context, ReadyCandidate) ([]SchedulerResourceRequirement, error)

func (SchedulerAdmissionPolicyFunc) Requirements 

func (f SchedulerAdmissionPolicyFunc) Requirements(ctx context.Context, candidate ReadyCandidate) ([]SchedulerResourceRequirement, error)

type SchedulerDemand 

type SchedulerDemand struct {
	Effects      graph.EffectSet          `json:"effects,omitempty"`
	Capabilities []string                 `json:"capabilities,omitempty"`
	Concurrency  []graph.ConcurrencyClaim `json:"concurrency,omitempty"`
}

SchedulerDemand is the trusted execution metadata for one invocation. Effects and capabilities must come from the registered StepKindSpec (plus host policy), never from an untrusted graph declaration that narrows it.

type SchedulerLimits 

type SchedulerLimits struct {
	Workers      int                  `json:"workers"`
	PerRun       int                  `json:"per_run,omitempty"`
	Effects      map[graph.Effect]int `json:"effects,omitempty"`
	Capabilities map[string]int       `json:"capabilities,omitempty"`
	Named        map[string]int       `json:"named,omitempty"`
}

SchedulerLimits is host-owned initial admission configuration. The first durable admission fixes each resource's limit for that database; a later request with a different limit fails closed because safe live resizing needs a separately versioned, quiescence-aware reconfiguration contract. A zero optional limit disables that dimension. Workers must be positive. Named resources used by a node must have an explicit positive definition.

type SchedulerResourceHolder 

type SchedulerResourceHolder struct {
	Resource        SchedulerResourceID `json:"resource"`
	Invocation      NodeInvocationID    `json:"invocation"`
	Units           int                 `json:"units"`
	ClaimGeneration uint64              `json:"claim_generation"`
	Owner           string              `json:"owner"`
	AcquiredAt      time.Time           `json:"acquired_at"`
	ExpiresAt       time.Time           `json:"expires_at"`
}

func (SchedulerResourceHolder) Validate 

func (h SchedulerResourceHolder) Validate() error

type SchedulerResourceID 

type SchedulerResourceID struct {
	Kind   SchedulerResourceKind `json:"kind"`
	Name   string                `json:"name,omitempty"`
	RunID  RunID                 `json:"run_id,omitempty"`
	NodeID string                `json:"node_id,omitempty"`
}

SchedulerResourceID is structured so opaque run identities and graph IDs cannot collide through delimiter concatenation. RunID is set only for run and fan-out resources; NodeID is set only for fan-out resources.

func (SchedulerResourceID) Validate 

func (id SchedulerResourceID) Validate() error

type SchedulerResourceKind 

type SchedulerResourceKind string

SchedulerResourceKind identifies one independently enforced admission dimension. Fan-out occupancy is persisted by FanOutStore and is reported through the same diagnostic vocabulary, but is not a worker lease resource. 

const (
	SchedulerResourceWorker     SchedulerResourceKind = "worker"
	SchedulerResourceRun        SchedulerResourceKind = "run"
	SchedulerResourceEffect     SchedulerResourceKind = "effect"
	SchedulerResourceCapability SchedulerResourceKind = "capability"
	SchedulerResourceKey        SchedulerResourceKind = "concurrency_key"
	SchedulerResourceFanOut     SchedulerResourceKind = "fan_out"
)

func (SchedulerResourceKind) Valid 

func (k SchedulerResourceKind) Valid() bool

type SchedulerResourceQuery 

type SchedulerResourceQuery struct {
	RunID RunID     `json:"run_id,omitempty"`
	Now   time.Time `json:"now"`
	Limit int       `json:"limit,omitempty"`
}

type SchedulerResourceRequirement 

type SchedulerResourceRequirement struct {
	Resource SchedulerResourceID `json:"resource"`
	Units    int                 `json:"units"`
	Limit    int                 `json:"limit"`
}

SchedulerResourceRequirement requests Units from one persisted semaphore. Limit is part of the durable definition: another process cannot silently claim the same resource using a different configured limit.

func BuildSchedulerRequirements 

func BuildSchedulerRequirements(runID RunID, limits SchedulerLimits, demand SchedulerDemand) ([]SchedulerResourceRequirement, error)

BuildSchedulerRequirements validates and canonicalizes independently enforced worker, run, effect, capability, and named-key requirements.

func (SchedulerResourceRequirement) Validate 

func (r SchedulerResourceRequirement) Validate() error

type SchedulerResourceState 

type SchedulerResourceState struct {
	Holders []SchedulerResourceHolder `json:"holders"`
	Waiters []SchedulerResourceWaiter `json:"waiters"`
}

type SchedulerResourceStore 

type SchedulerResourceStore interface {
	AdmitNode(context.Context, AdmitNodeRequest) (AdmitNodeResult, error)
	InspectSchedulerResources(context.Context, SchedulerResourceQuery) (SchedulerResourceState, error)
}

SchedulerResourceStore owns cross-process admission and diagnostics.

type SchedulerResourceWaiter 

type SchedulerResourceWaiter struct {
	Invocation   NodeInvocationID               `json:"invocation"`
	Requirements []SchedulerResourceRequirement `json:"requirements"`
	Blocked      []SchedulerResourceID          `json:"blocked"`
	Priority     int                            `json:"priority,omitempty"`
	EnqueuedAt   time.Time                      `json:"enqueued_at"`
	UpdatedAt    time.Time                      `json:"updated_at"`
}

func (SchedulerResourceWaiter) Validate 

func (w SchedulerResourceWaiter) Validate() error

type SealCompensationEntryRequest 

type SealCompensationEntryRequest struct {
	RunID                    RunID
	EntryID                  string
	ExpectedLedgerGeneration uint64
	ExpectedEntryGeneration  uint64
	ExpectedNodeGeneration   uint64
	At                       time.Time
}

type SealCompensationEntryResult 

type SealCompensationEntryResult struct {
	Ledger CompensationLedgerSnapshot
	Entry  CompensationEntrySnapshot
}

type ServiceCoordinator 

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

ServiceCoordinator observes readiness and liveness and executes durable stop intent. It never retains a process handle; recovery reconstructs every call from ServiceSnapshot and the frozen registry.

func NewServiceCoordinator 

func NewServiceCoordinator(options ServiceCoordinatorOptions) (*ServiceCoordinator, error)

func (*ServiceCoordinator) Reconcile 

func (c *ServiceCoordinator) Reconcile(ctx context.Context, start NodeInvocationID) (ServiceReconcileResult, error)

Reconcile advances one durable service. Host stop is invoked only after the generated teardown attempt has atomically recorded ServiceStopping.

func (*ServiceCoordinator) Recover 

func (c *ServiceCoordinator) Recover(ctx context.Context, query ServiceQuery) ([]ServiceReconcileResult, error)

Recover reconciles active services in deterministic store order.

type ServiceCoordinatorOptions 

type ServiceCoordinatorOptions struct {
	Store         ServiceStore
	State         StateStore
	Registry      stepkind.Registry
	Plans         RecoveryPlanSource
	Control       ControlFlowStore
	Verifiers     verification.Registry
	RetentionHook RetentionHook
	Now           func() time.Time
}

ServiceCoordinatorOptions supplies only extraction-safe collaborators.

type ServiceQuery 

type ServiceQuery struct {
	RunID RunID
	Limit int
}

ServiceQuery selects active records in updated-time and identity order.

type ServiceReconcileResult 

type ServiceReconcileResult struct {
	Service  ServiceSnapshot
	Node     *NodeInvocationSnapshot
	Attempt  *AttemptSnapshot
	Outputs  *values.ValueSetRef
	Warnings []DispatchWarning
}

ServiceReconcileResult reports the durable record after one provider call.

type ServiceSnapshot 

type ServiceSnapshot struct {
	Start              AttemptID                     `json:"start"`
	Invocation         stepkind.Invocation           `json:"invocation"`
	Ref                stepkind.ExternalOperationRef `json:"ref"`
	Status             ServiceStatus                 `json:"status"`
	HeartbeatTimeout   graph.Duration                `json:"heartbeat_timeout,omitempty"`
	ReadyCheck         *graph.VerificationSpec       `json:"ready_check,omitempty"`
	Outputs            *values.ValueSetRef           `json:"outputs,omitempty"`
	Failure            *Failure                      `json:"failure,omitempty"`
	Teardown           *AttemptID                    `json:"teardown,omitempty"`
	TeardownInvocation *stepkind.Invocation          `json:"teardown_invocation,omitempty"`
	StopRequestedAt    time.Time                     `json:"stop_requested_at,omitempty"`
	LastObservedAt     time.Time                     `json:"last_observed_at,omitempty"`
	LastHeartbeatAt    time.Time                     `json:"last_heartbeat_at,omitempty"`
	ReadyAt            time.Time                     `json:"ready_at,omitempty"`
	Generation         uint64                        `json:"generation"`
	CreatedAt          time.Time                     `json:"created_at"`
	UpdatedAt          time.Time                     `json:"updated_at"`
}

ServiceSnapshot is the restart-safe binding from an immutable service-start attempt to its provider reference and optional generated teardown attempt. Provider handles remain opaque inside Ref; raw credentials are forbidden by stepkind.ExternalOperationRef validation.

func (ServiceSnapshot) Validate 

func (s ServiceSnapshot) Validate() error

type ServiceStatus 

type ServiceStatus string

ServiceStatus is the closed, durable lifecycle of one service-start invocation. Ready is intentionally active: dependents may run, but recovery must continue heartbeat observation until the generated finalizer stops it. 

const (
	ServiceStarting  ServiceStatus = "starting"
	ServiceLaunching ServiceStatus = "launching"
	ServiceReady     ServiceStatus = "ready"
	ServiceStopping  ServiceStatus = "stopping"
	ServiceStopped   ServiceStatus = "stopped"
	ServiceFailed    ServiceStatus = "failed"
)

type ServiceStore 

type ServiceStore interface {
	LoadService(context.Context, NodeInvocationID) (ServiceSnapshot, error)
	PrepareServiceStart(context.Context, PrepareServiceStartRequest) (ServiceSnapshot, error)
	RecoverServiceStart(context.Context, RecoverServiceStartRequest) (SuspendServiceStartResult, error)
	SuspendServiceStart(context.Context, SuspendServiceStartRequest) (SuspendServiceStartResult, error)
	ApplyServiceReady(context.Context, ApplyServiceReadyRequest) (ApplyServiceReadyResult, error)
	ApplyServiceHeartbeat(context.Context, ApplyServiceHeartbeatRequest) (ServiceSnapshot, error)
	SuspendServiceTeardown(context.Context, SuspendServiceTeardownRequest) (SuspendServiceTeardownResult, error)
	ApplyServiceStop(context.Context, ApplyServiceStopRequest) (ApplyServiceStopResult, error)
	RecoverServices(context.Context, ServiceQuery) ([]ServiceSnapshot, error)
}

ServiceStore is deliberately separate from StateStore so hosts that cannot execute service nodes are not forced to advertise incomplete durability.

type SignalSelector 

type SignalSelector struct {
	RunID       RunID  `json:"run_id"`
	Name        string `json:"name"`
	Correlation string `json:"correlation"`
}

SignalSelector identifies one named signal wait without treating the caller-visible run or wait ID as authorization.

func (SignalSelector) Validate 

func (s SignalSelector) Validate() error

type StartNodeAttemptRequest 

type StartNodeAttemptRequest struct {
	InvocationID           NodeInvocationID
	ExpectedNodeGeneration uint64
	Claim                  ClaimProof
	Executor               ExecutorMetadata
	Inputs                 *values.ValueSetRef
	At                     time.Time
}

StartNodeAttemptRequest atomically creates LatestAttempt+1 and moves a ready node to running. Attempt numbers are assigned by the store.

type StartNodeAttemptResult 

type StartNodeAttemptResult struct {
	Node    NodeInvocationSnapshot
	Attempt AttemptSnapshot
	Event   Event
}

StartNodeAttemptResult contains the atomically persisted node, attempt, and attempt-started event.

type StateStore 

type StateStore interface {
	ExternalOperationStore
	RetryStore
	FanOutStore
	CancellationStore

	CreateRun(context.Context, CreateRunRequest) (RunSnapshot, IdempotencyOutcome, error)
	LoadRun(context.Context, RunID) (RunSnapshot, error)
	SaveRun(context.Context, SaveRunRequest) (RunSnapshot, error)
	TransitionRun(context.Context, RunTransitionRequest) (RunTransitionResult, error)

	CreateNodeInvocation(context.Context, CreateNodeInvocationRequest) (NodeInvocationSnapshot, error)
	LoadNodeInvocation(context.Context, NodeInvocationID) (NodeInvocationSnapshot, error)
	SaveNodeInvocation(context.Context, SaveNodeInvocationRequest) (NodeInvocationSnapshot, error)
	TransitionNode(context.Context, NodeTransitionRequest) (NodeTransitionResult, error)

	StartNodeAttempt(context.Context, StartNodeAttemptRequest) (StartNodeAttemptResult, error)
	FinishNodeAttempt(context.Context, FinishNodeAttemptRequest) (FinishNodeAttemptResult, error)
	LoadAttempt(context.Context, AttemptID) (AttemptSnapshot, error)
	ListAttempts(context.Context, NodeInvocationID) ([]AttemptSnapshot, error)

	LoadWait(context.Context, WaitID) (WaitSnapshot, error)

	SaveValues(context.Context, SaveValuesRequest) (values.ValueSetRef, error)
	LoadValues(context.Context, values.ValueSetRef) (values.ValueSet, error)

	RecordPlan(context.Context, PlanRef) error
	LoadPlan(context.Context, string) (PlanRef, error)

	AppendEvent(context.Context, AppendEventRequest) (Event, error)
	ListEvents(context.Context, EventQuery) ([]Event, error)

	ClaimNode(context.Context, ClaimNodeRequest) (ClaimResult, error)
	RenewNodeLease(context.Context, RenewLeaseRequest) (ClaimLease, error)
	ReleaseNodeClaim(context.Context, ReleaseClaimRequest) error

	PutCacheEntry(context.Context, CacheEntry) error
	GetCacheEntry(context.Context, string, time.Time) (CacheEntry, bool, error)
	PutPinnedValue(context.Context, PinnedValue) error
	GetPinnedValue(context.Context, string, time.Time) (PinnedValue, bool, error)
	ListPinnedValues(context.Context, time.Time) ([]PinnedValue, error)

	RecordExternalActivation(context.Context, ExternalActivationRequest) (ExternalActivationSnapshot, IdempotencyOutcome, error)
	Recovery(context.Context, RecoveryQuery) (RecoverySnapshot, error)
}

StateStore is the high-level, extraction-ready runtime persistence contract. Implementations must defensively isolate mutable envelopes and preserve CAS, idempotency, append-only event, and recovery-query semantics.

type StepDispatcher 

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

StepDispatcher executes claimed nodes exclusively through a step-kind registry and the atomic StateStore attempt lifecycle.

func NewStepDispatcher 

func NewStepDispatcher(options DispatcherOptions) (*StepDispatcher, error)

NewStepDispatcher constructs a registry-driven dispatcher. A nil Now uses the UTC wall clock. A nil disposition terminates failed attempts.

func (*StepDispatcher) Dispatch 

func (d *StepDispatcher) Dispatch(ctx context.Context, request DispatchRequest) (DispatchResult, error)

Dispatch resolves, validates, starts, executes, and durably finishes one claimed invocation. Every adapter-side failure after StartNodeAttempt is converted into an unsuccessful FinishNodeAttempt before Dispatch returns.

type SuspendCommand 

type SuspendCommand struct {
	Request     SuspendNodeWaitRequest
	ResumeToken string
}

SuspendCommand keeps the one-time raw resume token outside the persistence request while allowing the coordinator to bind it to the stored digest and reject token-bearing URL paths.

type SuspendExternalOperationRequest 

type SuspendExternalOperationRequest struct {
	Operation                 ExternalOperationSnapshot
	ExpectedNodeGeneration    uint64
	ExpectedAttemptGeneration uint64
	Claim                     ClaimProof
	At                        time.Time
}

SuspendExternalOperationRequest atomically binds adapter work to the running attempt, sets the aggregate node waiting, and releases its claim.

func (SuspendExternalOperationRequest) Validate 

func (r SuspendExternalOperationRequest) Validate() error

type SuspendExternalOperationResult 

type SuspendExternalOperationResult struct {
	Operation ExternalOperationSnapshot
	Node      NodeInvocationSnapshot
	Attempt   AttemptSnapshot
	Events    []Event
}

type SuspendNodeWaitRequest 

type SuspendNodeWaitRequest struct {
	Wait                      WaitSnapshot
	ExpectedNodeGeneration    uint64
	ExpectedAttemptGeneration uint64
	Claim                     ClaimProof
	At                        time.Time
}

SuspendNodeWaitRequest atomically creates a wait and releases its running worker lease while preserving the unfinished attempt for continuation.

func (SuspendNodeWaitRequest) Validate 

func (r SuspendNodeWaitRequest) Validate() error

type SuspendServiceStartRequest 

type SuspendServiceStartRequest struct {
	Service                   ServiceSnapshot
	ExpectedNodeGeneration    uint64
	ExpectedAttemptGeneration uint64
	Claim                     ClaimProof
	At                        time.Time
}

SuspendServiceStartRequest atomically persists provider work and releases the running service-start attempt while it awaits readiness.

type SuspendServiceStartResult 

type SuspendServiceStartResult struct {
	Service ServiceSnapshot
	Node    NodeInvocationSnapshot
	Attempt AttemptSnapshot
	Events  []Event
}

type SuspendServiceTeardownRequest 

type SuspendServiceTeardownRequest struct {
	Start                     NodeInvocationID
	Teardown                  AttemptID
	Invocation                stepkind.Invocation
	ExpectedServiceGeneration uint64
	ExpectedNodeGeneration    uint64
	ExpectedAttemptGeneration uint64
	Claim                     ClaimProof
	At                        time.Time
}

SuspendServiceTeardownRequest binds the generated finally attempt to the exact start record and durably records stop intent before any host call.

type SuspendServiceTeardownResult 

type SuspendServiceTeardownResult struct {
	Service ServiceSnapshot
	Node    NodeInvocationSnapshot
	Attempt AttemptSnapshot
	Events  []Event
}

type SuspendWaitResult 

type SuspendWaitResult struct {
	Outcome IdempotencyOutcome
	Wait    WaitSnapshot
	Node    NodeInvocationSnapshot
	Attempt AttemptSnapshot
	Events  []Event
}

type TerminalIntentSnapshot 

type TerminalIntentSnapshot struct {
	RunID          RunID     `json:"run_id"`
	IntendedStatus RunStatus `json:"intended_status"`
	// SuccessOutputsRequired is derived from the exact graph when the intent
	// is created and is immutable across cleanup, recovery, and replay.
	SuccessOutputsRequired bool `json:"success_outputs_required,omitempty"`
	// CompensationRequired is graph-derived and fences terminal completion
	// until an exact plan-bound ledger reaches a terminal summary.
	CompensationRequired bool                 `json:"compensation_required,omitempty"`
	Reason               *Failure             `json:"reason,omitempty"`
	Error                *values.ValueSetRef  `json:"error,omitempty"`
	IdempotencyKey       string               `json:"idempotency_key"`
	Finalizers           []FinalizerScope     `json:"finalizers"`
	Status               TerminalIntentStatus `json:"status"`
	Generation           uint64               `json:"generation"`
	CreatedAt            time.Time            `json:"created_at"`
	UpdatedAt            time.Time            `json:"updated_at"`
	CompletedAt          time.Time            `json:"completed_at,omitempty"`
}

TerminalIntentSnapshot is the immutable intended outcome plus mutable CAS completion state. Finalizer ordering and typed originating error are fixed at creation and never inferred from events or timestamps. A zero-finalizer snapshot is reserved for the atomic fail-fast policy operation; the public BeginTerminalIntent contract continues to require declared cleanup.

func (TerminalIntentSnapshot) Validate 

func (s TerminalIntentSnapshot) Validate() error

type TerminalIntentStatus 

type TerminalIntentStatus string

TerminalIntentStatus is distinct from RunStatus: a run remains internally active while ordinary cleanup executes, but callers can observe the intended terminal outcome explicitly. 

const (
	TerminalIntentPending   TerminalIntentStatus = "pending"
	TerminalIntentCompleted TerminalIntentStatus = "completed"
)

type TimeoutAnchor 

type TimeoutAnchor struct {
	ScheduledAt     time.Time
	StartedAt       time.Time
	ExternalAt      time.Time
	LastHeartbeatAt time.Time
}

TimeoutAnchor contains durable timestamps used by timeout evaluation. Zero timestamps mean the corresponding executor phase has not begun.

type TimeoutDeadline 

type TimeoutDeadline struct {
	Kind     TimeoutKind
	Deadline time.Time
}

TimeoutDeadline is one configured operational deadline.

type TimeoutEvaluation 

type TimeoutEvaluation struct {
	Deadlines []TimeoutDeadline
	Due       *TimeoutDeadline
}

TimeoutEvaluation lists all configured deadlines in chronological then kind order and identifies the first deadline due at Now.

func EvaluateTimeouts 

func EvaluateTimeouts(policy *graph.TimeoutPolicy, anchor TimeoutAnchor, now time.Time) (TimeoutEvaluation, error)

EvaluateTimeouts resolves the graph duration vocabulary against durable phase anchors. Generic wait deadlines remain owned by WaitCoordinator.

type TimeoutKind 

type TimeoutKind string

TimeoutKind distinguishes operational deadlines that have different retry and cancellation semantics. 

const (
	TimeoutQueue           TimeoutKind = "queue"
	TimeoutExecution       TimeoutKind = "execution"
	TimeoutExternalWait    TimeoutKind = "external_wait"
	TimeoutHeartbeat       TimeoutKind = "heartbeat"
	TimeoutScheduleToClose TimeoutKind = "schedule_to_close"
)

func (TimeoutKind) Valid 

func (k TimeoutKind) Valid() bool

Valid reports whether k is a supported timeout category.

type TimeoutWaitRequest 

type TimeoutWaitRequest struct {
	WaitID                 WaitID
	ExpectedWaitGeneration uint64
	ExpectedNodeGeneration uint64
	IdempotencyKey         string
	Deadline               time.Time
	Now                    time.Time
}

TimeoutWaitRequest atomically closes one open wait and its waiting node attempt under wait/node CAS. Deadline must exactly match the persisted wait. An already-terminal wait returns its current durable wait and node without applying a mutation; expected generations fence only the open mutation.

func (TimeoutWaitRequest) Validate 

func (r TimeoutWaitRequest) Validate() error

Validate reports malformed timeout input. A valid request may still be not due when Now precedes Deadline.

type TimerWakeCommand 

type TimerWakeCommand struct {
	WaitID  WaitID
	FiredAt time.Time
}

TimerWakeCommand is the credential-free host handoff for a scheduled timer. Correlation, payload, responder, semantic resolution time, and idempotency identity are derived exclusively from the durable wait record.

func (TimerWakeCommand) Validate 

func (c TimerWakeCommand) Validate() error

Validate reports malformed timer delivery input. Whether the timer is due is checked against the persisted WakeAt after the record is loaded.

type TransitionConflictError 

type TransitionConflictError struct {
	Entity string
	ID     string
	Status string
	Reason string
}

TransitionConflictError describes a same-status request whose state data is not an exact semantic replay.

func (*TransitionConflictError) Error 

func (e *TransitionConflictError) Error() string

func (*TransitionConflictError) Unwrap 

func (e *TransitionConflictError) Unwrap() error

Unwrap supports errors.Is(err, ErrTransitionConflict).

type TransitionError 

type TransitionError struct {
	Entity string
	ID     string
	From   string
	To     string
	Reason string
}

TransitionError describes a disallowed status edge without requiring callers to parse the presentation string.

func (*TransitionError) Error 

func (e *TransitionError) Error() string

func (*TransitionError) Unwrap 

func (e *TransitionError) Unwrap() error

Unwrap supports errors.Is(err, ErrInvalidTransition).

type TransitionOutcome 

type TransitionOutcome string

TransitionOutcome distinguishes a persisted state change from an exact same-state semantic replay. 

const (
	TransitionApplied TransitionOutcome = "applied"
	TransitionNoOp    TransitionOutcome = "no_op"
)

type ValueOwner 

type ValueOwner struct {
	Kind       string            `json:"kind"`
	RunID      RunID             `json:"run_id"`
	Invocation *NodeInvocationID `json:"invocation,omitempty"`
	Attempt    *AttemptID        `json:"attempt,omitempty"`
}

ValueOwner describes why a value set is persisted without prescribing a table or host record type.

func (ValueOwner) Validate 

func (o ValueOwner) Validate() error

Validate reports malformed value ownership metadata.

type ValueRecord 

type ValueRecord struct {
	Ref    values.ValueSetRef `json:"ref"`
	Owner  ValueOwner         `json:"owner"`
	Values values.ValueSet    `json:"values"`
}

ValueRecord exposes immutable value ownership for pin validation.

func (ValueRecord) Validate 

func (r ValueRecord) Validate() error

Validate proves that a loaded value record is internally self-consistent. Host stores are untrusted boundaries: the reference, owner, and content digest must agree before source lookup or reuse authorization begins.

type ValueRecordStore 

type ValueRecordStore interface {
	LoadValueRecord(context.Context, values.ValueSetRef) (ValueRecord, error)
}

type ValueRef 

type ValueRef struct {
	Set  values.ValueSetRef `json:"set"`
	Name string             `json:"name"`
}

ValueRef identifies one named value inside a persisted value set.

func (ValueRef) Validate 

func (r ValueRef) Validate() error

Validate reports malformed value identity.

type VerificationPersistenceError 

type VerificationPersistenceError struct {
	Ref   *values.ValueSetRef
	Cause error
}

VerificationPersistenceError reports a report value set that was written but could not be linked through its attempt event. Ref lets retention/repair tooling find the immutable orphan without treating verification as applied.

func (*VerificationPersistenceError) Error 

func (e *VerificationPersistenceError) Error() string

func (*VerificationPersistenceError) Unwrap 

func (e *VerificationPersistenceError) Unwrap() error

type VerificationRecord 

type VerificationRecord struct {
	Report   verification.Report `json:"report"`
	Ref      values.ValueSetRef  `json:"ref"`
	Event    Event               `json:"event"`
	Replayed bool                `json:"replayed,omitempty"`
}

VerificationRecord is the durable, attempt-bound verification outcome returned by dispatch. Event.Values is the authoritative report reference.

type WaitClosedError 

type WaitClosedError struct {
	WaitID     WaitID
	Status     WaitStatus
	ResolvedAt time.Time
}

WaitClosedError reports a response that lost to a durable terminal outcome.

func (*WaitClosedError) Error 

func (e *WaitClosedError) Error() string

func (*WaitClosedError) Unwrap 

func (e *WaitClosedError) Unwrap() error

type WaitCoordinator 

type WaitCoordinator struct {
	Store        WaitStore
	Scheduler    workflowwait.ActivationScheduler
	Materializer workflowwait.Materializer
	Authorizer   workflowwait.ResponderAuthorizer
}

WaitCoordinator owns generic suspension/resume orchestration. Store state is authoritative; Scheduler, Materializer, and Authorizer are optional host adapters.

func (WaitCoordinator) Recover 

func (c WaitCoordinator) Recover(ctx context.Context, query OpenWaitQuery, now time.Time) ([]WaitTimeoutResult, error)

Recover reconstructs open-wait scheduling solely from durable records.

func (WaitCoordinator) RecoverWaits 

func (c WaitCoordinator) RecoverWaits(ctx context.Context, query OpenWaitQuery, now time.Time) (WaitRecoveryResults, error)

RecoverWaits deterministically fires due successful timers, times out due deadlines, and re-materializes future waits. Store recovery ordering is by the earliest applicable WakeAt or Deadline and respects query Limit.

func (WaitCoordinator) Resume 

func (c WaitCoordinator) Resume(ctx context.Context, command ResumeCommand) (ResumeWaitResult, error)

func (WaitCoordinator) Suspend 

func (c WaitCoordinator) Suspend(ctx context.Context, command SuspendCommand) (SuspendWaitResult, error)

func (WaitCoordinator) WakeTimer 

func (c WaitCoordinator) WakeTimer(ctx context.Context, command TimerWakeCommand) (ResumeWaitResult, error)

WakeTimer resolves a due successful timer through the same authorized, atomic, idempotent resume path as every external wake. Late delivery uses persisted WakeAt as the semantic resolution time, so a scheduled wake that precedes a later Deadline wins deterministically after recovery.

type WaitID 

type WaitID string

WaitID is an opaque host-independent wait identifier.

type WaitRecoveryResults 

type WaitRecoveryResults struct {
	Woken    []ResumeWaitResult
	TimedOut []WaitTimeoutResult
}

WaitRecoveryResults contains every durable transition applied while recovering open waits. Future waits are only re-materialized.

type WaitRef 

type WaitRef struct {
	ID WaitID `json:"id"`
}

WaitRef is the compact wait identity stored on a node invocation.

func (WaitRef) Validate 

func (r WaitRef) Validate() error

Validate reports malformed wait identity.

type WaitSnapshot 

type WaitSnapshot struct {
	Ref        WaitRef          `json:"ref"`
	Invocation NodeInvocationID `json:"invocation"`
	workflowwait.Record
	Generation uint64    `json:"generation"`
	CreatedAt  time.Time `json:"created_at"`
	UpdatedAt  time.Time `json:"updated_at"`
	ResolvedAt time.Time `json:"resolved_at,omitempty"`
}

WaitSnapshot adds runtime identity, CAS, and storage chronology to the canonical workflow/wait Record. The anonymous record produces one flat JSON envelope; it does not duplicate status or resume fields.

func (WaitSnapshot) Validate 

func (s WaitSnapshot) Validate() error

Validate checks minimal wait persistence integrity.

type WaitStatus 

type WaitStatus = workflowwait.Status

WaitStatus is retained as a source-compatible alias. workflow/wait owns the generic wait state model.

type WaitStore 

type WaitStore interface {
	LoadWait(context.Context, WaitID) (WaitSnapshot, error)
	LoadNodeInvocation(context.Context, NodeInvocationID) (NodeInvocationSnapshot, error)
	LoadAttempt(context.Context, AttemptID) (AttemptSnapshot, error)
	LoadWaitContinuation(context.Context, AttemptID) (WaitSnapshot, error)
	SuspendNodeWait(context.Context, SuspendNodeWaitRequest) (SuspendWaitResult, error)
	ResumeNodeWait(context.Context, ResumeNodeWaitRequest) (ResumeWaitResult, error)
	TimeoutWait(context.Context, TimeoutWaitRequest) (WaitTimeoutResult, error)
	RecoverOpenWaits(context.Context, OpenWaitQuery) ([]WaitSnapshot, error)
}

WaitStore is the only production mutation surface for generic waits. The unsafe wait CRUD methods deliberately are not part of StateStore.

type WaitTimeoutNotDueError 

type WaitTimeoutNotDueError struct {
	Now      time.Time
	Deadline time.Time
}

WaitTimeoutNotDueError reports a persisted deadline that has not arrived.

func (*WaitTimeoutNotDueError) Error 

func (e *WaitTimeoutNotDueError) Error() string

func (*WaitTimeoutNotDueError) Unwrap 

func (e *WaitTimeoutNotDueError) Unwrap() error

type WaitTimeoutResult 

type WaitTimeoutResult struct {
	Applied  bool
	Replayed bool
	Wait     WaitSnapshot
	Node     NodeInvocationSnapshot
	Attempt  AttemptSnapshot
	Events   []Event
}

WaitTimeoutResult is the store-neutral atomic outcome. Applied is false only when the wait was already resolved. Replayed marks an exact durable retry. Events contains the attempt-finished, node-status, and wait-timeout facts in durable sequence order for an applied timeout.

type WaitTimeoutStore 

type WaitTimeoutStore interface {
	TimeoutWait(context.Context, TimeoutWaitRequest) (WaitTimeoutResult, error)
}

WaitTimeoutStore is the timeout portion of the atomic WaitStore contract.

type WaitWakeNotDueError 

type WaitWakeNotDueError struct {
	FiredAt time.Time
	WakeAt  time.Time
}

WaitWakeNotDueError reports an early host timer delivery without mutating durable state.

func (*WaitWakeNotDueError) Error 

func (e *WaitWakeNotDueError) Error() string

func (*WaitWakeNotDueError) Unwrap 

func (e *WaitWakeNotDueError) Unwrap() error

Source Files

View all Source files

Directories

Path Synopsis
Package inmemory provides the public concurrency-safe, process-lifetime implementation of the workflow runtime storage contracts.
 Package inmemory provides the public concurrency-safe, process-lifetime implementation of the workflow runtime storage contracts.
Package runtimetest preserves the original test-support import path for the public process-lifetime in-memory runtime store.
 Package runtimetest preserves the original test-support import path for the public process-lifetime in-memory runtime store.

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.