run

package
v0.61.0 Latest Latest
Warning

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

 Go to latest Published: Nov 10, 2026 License: Apache-2.0 Imports: 26 Imported by: 0

Documentation

Overview

Package run executes deterministic Programs against an append-only tree log. Store.Append acknowledges every Effect intent before dispatch and every settlement before another Step can consume it. A Checkpoint materializes its prefix: recovery starts from the newest one, re-executes only the Steps after it, and rejects code whose replayed Transition differs from the log. The Checkpoint is the authority for its prefix: recovery never re-executes the Steps it covers, so nondeterministic code is detected only in Steps after the newest Checkpoint. Retained inputs, successful outputs and unaddressed Signals must satisfy their bound Program's Descriptor before recovery acquires a writer Epoch. Wait answers belong to their Wait's schema; Definition.Restore owns only the opaque execution state.

A Runtime owns the host Build identity. Its Compiler capability lets strategy packages freeze declarations into Program references without exposing the Runtime's execution authority. Bind is the atomic construction API for custom Definitions. Revision identifies opaque callbacks and clients; changing Build, Revision, a declaration, or a child reference changes the Program reference.

Start is durable and detached from the request context after admission. Reuse the same StartID to retry submission; a conflicting declaration is rejected. Recover explicitly acquires a new writer Epoch. A duplicate Start never takes over another writer. Run instead attaches the foreground operation to its context and joins and releases its anonymous local writer before returning. An interrupted Run returns RuntimeError with the durable recovery identity. Await checks its observation context before reading and before returning a result, so concurrent completion cannot conceal caller cancellation. A later observation can still read the committed outcome. Close cancels and joins local work without declaring an uncertain external operation stopped; unresolved Effects remain in the durable log. ReleaseTree joins an exited root's local writer before relinquishing it; a writer already stopped by Close can be released even without a root Exit.

Process paths own ancestry. A child key cannot be reused within its parent's lifetime because that would also reuse Effect identities, so each parent keeps every key it has started with the LSN of that child's Started record. The set grows with every child started and is bounded only when TreeLimits.MaxChildren is. A child's state leaves the tree once its Exit is consumed and Reaped is appended; Runtime.Answered then reads only the history of that child's subtree, from its Started record to its Reaped record.

Completion carries a definite Effect outcome. An Unknown outcome stays pending until explicit resolution or a permitted replay under the same EffectID. Exit and Result are published only after all descendants and uncertain operations have drained, including after cancellation or failure. Yield declares local Wait values directly; the inbox never contains a wait-opening handshake. A WaitKey permanently identifies its first question, answer schema and byte limit. Withdrawal stops answer admission but retains that contract in the process and its Checkpoints; reintroduction must repeat it exactly. An answered key cannot reopen, so a new question requires a new key. Retained contracts grow with the process's distinct WaitKeys until reaping.

Store.Load serves recovery and Store.Read serves history, so the cost of each is bounded by what it returns rather than by the age of the tree. Observe reads history after an LSN and remains valid across recovery; a tree this Runtime writes wakes its observers on commit. Runtime.History reads the complete log for projections that need facts a Checkpoint no longer carries. Observation of an already exited tree ends at the head confirmed on its first pull, including recovery activations committed after the root's Exit. Hosts own log retention. Config.Deltas is the separate best-effort stream: channel saturation may drop increments but cannot change committed usage or outcomes. Keep the channel open until Close returns.

Index

Examples

Constants

View Source 
const MaxDescriptionBytes = 4096
View Source 
const MaxDiagnosticBytes = 4096
View Source 
const MaxPayloadBytes = 64 << 20

MaxPayloadBytes is the maximum encoded JSON size of an individual Agent input, output, Effect, Signal, or settlement payload. Both the supplied bytes and the canonical encoding must fit. Canonical payloads use RFC 7493 JSON, omit whitespace, sort object names by UTF-16 code units (RFC 8785 section 3.2.3), preserve number literals verbatim, and escape <, >, &, U+2028 and U+2029. Strings otherwise use their shortest JSON encoding. Digests over payloads and persisted payload bytes use this form; this is not RFC 8785 number canonicalization.

Variables

View Source 
var (
	ErrInvalidChildStart   = errors.New("run: invalid child process start")
	ErrInvalidChildControl = errors.New("run: invalid child control")
)
View Source 
var (
	ErrInvalidLog           = errors.New("run: invalid log")
	ErrCommitConflict       = errors.New("run: log position conflict")
	ErrEpochConflict        = errors.New("run: writer epoch conflict")
	ErrNondeterministicStep = errors.New("run: replayed step differs from committed transition")
)
View Source 
var (
	ErrInvalidMessage = errors.New("run: invalid message")
	ErrSignalRejected = errors.New("run: signal rejected")
	ErrSignalConflict = errors.New("run: signal identity conflict")
)
View Source 
var (
	ErrInvalidConfig = errors.New("run: invalid runtime configuration")
	ErrClosed        = errors.New("run: runtime is closed")
	// ErrWriterStopped is the work context's cancellation cause when its local
	// writer ends. It does not declare cancellation of the durable Process.
	ErrWriterStopped         = errors.New("run: local writer stopped")
	ErrProcessNotFound       = errors.New("run: process not found")
	ErrProcessNotActive      = errors.New("run: process has no active local writer")
	ErrProcessFinished       = errors.New("run: process has finished")
	ErrActiveProcesses       = errors.New("run: runtime has active processes")
	ErrTreeNotFound          = errors.New("run: tree not found")
	ErrEffectNotPending      = errors.New("run: effect does not require resolution")
	ErrEffectReplayForbidden = errors.New("run: effect cannot be replayed under the same identity")
	ErrEffectOutcomeUnknown  = errors.New("run: effect outcome remains unknown")
	ErrInvalidProcessControl = errors.New("run: invalid process control")
	ErrNilContext            = errors.New("run: nil context")
)
View Source 
var ErrCounterExhausted = NewClassifiedError(
	FailureKindExecution, failureCodeRuntimeCounterExhausted, "run: counter exhausted",
)

ErrCounterExhausted reports exhausted numeric identity or accounting space, independently of the Host's quota policy. Work stops before a counter wraps. Strategies exhausting their own counters report this same sentinel.

View Source 
var ErrInvalidCapability = errors.New("run: invalid capability")
View Source 
var ErrInvalidDescriptor = errors.New("run: invalid descriptor")
View Source 
var ErrInvalidDigest = errors.New("run: invalid digest")
View Source 
var ErrInvalidEffect = errors.New("run: invalid effect")
View Source 
var ErrInvalidExecutionState = errors.New("run: invalid execution state")
View Source 
var ErrInvalidFailure = errors.New("run: invalid failure")
View Source 
var ErrInvalidIdentity = errors.New("run: invalid identity")
View Source 
var ErrInvalidPayload = errors.New("run: invalid payload")
View Source 
var ErrInvalidProgram = errors.New("run: invalid program")
View Source 
var ErrInvalidProgramRef = errors.New("run: invalid program reference")
View Source 
var ErrInvalidSettlement = errors.New("run: invalid effect settlement")
View Source 
var ErrInvalidStart = errors.New("run: invalid start identity")
View Source 
var ErrInvalidStatus = errors.New("run: invalid status")
View Source 
var ErrInvalidTransition = errors.New("run: invalid transition")
View Source 
var ErrPayloadTooLarge = errors.New("run: payload exceeds the maximum encoded size")

ErrPayloadTooLarge reports an encoding over MaxPayloadBytes, letting a caller that reuses this encoding as an admission bound tell an undeliverable payload from an otherwise invalid one.

View Source 
var ErrProcessAdmissionRejected = errors.New("run: process admission rejected")
View Source 
var ErrResourceLimitExceeded = NewClassifiedError(FailureKindExecution, failureCodeRuntimeResourceLimitExceeded, "run: resource limit exceeded")

ErrResourceLimitExceeded reports a designed execution bound, not a Runtime defect.

Functions

func NormalizeDiagnostic 

func NormalizeDiagnostic(message string) string

NormalizeDiagnostic makes arbitrary external text non-empty, trimmed UTF-8 within MaxDiagnosticBytes. Callers remain responsible for excluding secrets.

func RequireContext 

func RequireContext(ctx context.Context) context.Context

func UsageFromRecord added in v0.61.0 

func UsageFromRecord(record Record) (ProcessID, Usage)

UsageFromRecord projects the contribution owned by one committed record. Live folding, historical totals, and observers use this same rule, including child exits that contribute to their parent after the child is reaped.

func ValidDescription 

func ValidDescription(description string) bool

func ValidDiagnostic 

func ValidDiagnostic(message string) bool

func ValidQualifiedName 

func ValidQualifiedName(name string) bool

Types

type Activated added in v0.60.0 

type Activated struct {
	Epoch Epoch `json:"epoch"`
}

type Batch added in v0.60.0 

type Batch struct {
	Root    ProcessID
	Epoch   Epoch
	After   LSN
	Records []Record
}

Batch is the unit a Store appends atomically: Records placed after After by the writer holding Epoch. Entries derives each record's coordinates, so a Store never assigns positions itself.

func (Batch) Admit added in v0.60.0 

func (b Batch) Admit(current Epoch, head LSN, existing []Entry) (next Epoch, retry bool, err error)

Admit is the one append decision every Store makes against its current Epoch and head, given the stored entries the batch would occupy. It returns the Epoch after the batch, or retry=true when an identical batch is already stored and the append must be acknowledged without writing.

func (Batch) Clone added in v0.60.0 

func (b Batch) Clone() (Batch, error)

func (Batch) Entries added in v0.60.0 

func (b Batch) Entries() []Entry

func (Batch) MarshalJSON added in v0.60.0 

func (b Batch) MarshalJSON() ([]byte, error)

func (*Batch) UnmarshalJSON added in v0.60.0 

func (b *Batch) UnmarshalJSON(data []byte) error

func (Batch) Validate added in v0.60.0 

func (b Batch) Validate() error

type BindingConfig 

type BindingConfig struct {
	Definition    Definition
	Dispatcher    Dispatcher
	Revision      string
	Configuration any
}

Bind hashes the declaration after freezing its canonical JSON. Opaque collaborators must be identified by Revision in their declaring configuration.

type Budget 

type Budget struct {
	Steps    Quota `json:"steps" jsonwire:"required"`
	Effects  Quota `json:"effects" jsonwire:"required"`
	Messages Quota `json:"messages" jsonwire:"required"`
}

Budget grants cumulative work authority. Zero quotas are unlimited. A finite parent permanently charges child grants, including unused child work. An unlimited parent grants either kind without a finite debit. Each dimension is independent; a finite parent cannot grant an unlimited child quota.

func (*Budget) UnmarshalJSON 

func (b *Budget) UnmarshalJSON(data []byte) error

type CallbackPanicError 

type CallbackPanicError struct {
	Operation string
	Value     any
}

CallbackPanicError identifies a panic contained at a Host callback boundary. Operation names the callback; Value is the recovered value, which Unwrap exposes when it is an error. The boundary still owns the outcome: execution failures use FailureKindPanic, uncertain dispatch remains unknown, and storage errors stop the writer through RuntimeError. It is an in-memory diagnostic and must not be serialized as a result.

func (*CallbackPanicError) Error 

func (c *CallbackPanicError) Error() string

func (*CallbackPanicError) Unwrap 

func (c *CallbackPanicError) Unwrap() error

type Capability 

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

Capability is one stable qualified authority name understood by a Program's dispatcher or another external boundary. The Framework only enforces possession and attenuation; it does not assign product meaning.

func ParseCapability 

func ParseCapability(name string) (Capability, error)

func (Capability) JSONSchemaModel 

func (Capability) JSONSchemaModel() any

func (Capability) MarshalText 

func (c Capability) MarshalText() ([]byte, error)

func (Capability) String 

func (c Capability) String() string

func (*Capability) UnmarshalText 

func (c *Capability) UnmarshalText(text []byte) error

func (Capability) Valid 

func (c Capability) Valid() bool

type CapabilitySet 

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

CapabilitySet is an immutable, sorted set of authority names. Its zero value is the valid empty set, encoded as an empty JSON array. JSON null is invalid.

func NewCapabilitySet 

func NewCapabilitySet(capabilities ...Capability) (CapabilitySet, error)

func (CapabilitySet) Allows 

func (c CapabilitySet) Allows(requested CapabilitySet) bool

func (CapabilitySet) Contains 

func (c CapabilitySet) Contains(capability Capability) bool

func (CapabilitySet) JSONSchemaModel 

func (CapabilitySet) JSONSchemaModel() any

func (CapabilitySet) MarshalJSON 

func (c CapabilitySet) MarshalJSON() ([]byte, error)

Construction and decode already guarantee valid, sorted, unique grants, and a nil slice encodes as the empty-set array the contract requires.

func (*CapabilitySet) UnmarshalJSON 

func (c *CapabilitySet) UnmarshalJSON(data []byte) error

UnmarshalJSON rejects duplicate grants and null; ordering is normalized. Construction accepts duplicates because its arguments describe a set.

func (CapabilitySet) Values 

func (c CapabilitySet) Values() []Capability

Values returns an independently owned, sorted capability slice.

type Checkpoint 

type Checkpoint struct {
	Processes        []ProcessCheckpoint `json:"processes"`
	StartedProcesses uint64              `json:"started_processes"`
}

Checkpoint materializes every acknowledged record before it. Recovery starts from the newest one without reading its prefix, so it is the authority for that prefix rather than a cache to be checked against it; it can only be written by the fenced writer that folded the prefix. Only this record carries opaque ExecutionState.

type ChildBinding added in v0.60.0 

type ChildBinding struct {
	Program      Program[Payload, Payload]
	Budget       Budget
	Capabilities CapabilitySet
}

ChildBinding fixes the program and delegated resources before a strategy chooses a particular child identity and input.

func (ChildBinding) Spec added in v0.60.0 

func (c ChildBinding) Spec(key ChildKey, input Payload) (ChildSpec, error)

func (ChildBinding) Valid added in v0.60.0 

func (c ChildBinding) Valid() bool

type ChildKey 

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

func ParseChildKey 

func ParseChildKey(value string) (ChildKey, error)

func (ChildKey) IsZero added in v0.60.0 

func (i ChildKey) IsZero() bool

func (ChildKey) JSONSchemaModel 

func (ChildKey) JSONSchemaModel() any

func (ChildKey) MarshalText 

func (i ChildKey) MarshalText() ([]byte, error)

func (ChildKey) String 

func (i ChildKey) String() string

func (*ChildKey) UnmarshalText 

func (c *ChildKey) UnmarshalText(data []byte) error

func (ChildKey) Valid 

func (i ChildKey) Valid() bool

type ChildSpec 

type ChildSpec struct {
	Key          ChildKey      `json:"key"`
	ProgramRef   ProgramRef    `json:"program_ref"`
	Input        Payload       `json:"input"`
	Budget       Budget        `json:"budget" jsonwire:"required"`
	Capabilities CapabilitySet `json:"capabilities" jsonwire:"required"`
}

func (*ChildSpec) UnmarshalJSON 

func (c *ChildSpec) UnmarshalJSON(data []byte) error

func (ChildSpec) Valid 

func (c ChildSpec) Valid() bool

type ClassifiedError 

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

ClassifiedError is a sentinel that owns the Failure kind and code persisted when an error wrapping it leaves Execution.Step, so no separate mapping can disagree with the declaration.

func NewClassifiedError 

func NewClassifiedError(kind FailureKind, code, message string) *ClassifiedError

NewClassifiedError declares one sentinel together with the Failure the Runtime persists when Execution.Step returns an error wrapping it; the persisted message is the complete wrapped diagnostic. An invalid kind, code, or message is a programming error and panics.

func (*ClassifiedError) Error 

func (c *ClassifiedError) Error() string

type Compiler added in v0.60.0 

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

type Completion added in v0.60.0 

type Completion struct {
	Effect  EffectID   `json:"effect"`
	Outcome Settlement `json:"outcome"`
}

func (Completion) Valid added in v0.60.0 

func (c Completion) Valid() bool

type Config 

type Config struct {
	Store           Store
	Build           string
	ProgramResolver ProgramResolver
	ProcessAdmitter ProcessAdmitter
	Deltas          chan<- Delta
	Budget          Budget
	TreeLimits      TreeLimits
	Capabilities    CapabilitySet
	// Tree-wide snapshots cannot read an Execution during its Step. At this
	// interval, new Steps wait for existing Steps to drain; controls and
	// settlements stay live.
	CheckpointInterval uint64
	StoreTimeout       time.Duration
}

type ControlKind added in v0.60.0 

type ControlKind string
const (
	ControlPause    ControlKind = "pause"
	ControlResume   ControlKind = "resume"
	ControlCancel   ControlKind = "cancel"
	ControlKill     ControlKind = "kill"
	ControlDeadline ControlKind = "deadline"
)

func (ControlKind) Valid added in v0.60.0 

func (c ControlKind) Valid() bool

type ControlOrigin added in v0.60.0 

type ControlOrigin string

ControlOrigin records who requested a control independently of its target. A parent's direct child request must retain its cause across replay. 

const (
	ControlOriginHost   ControlOrigin = "host"
	ControlOriginParent ControlOrigin = "parent"
)

func (ControlOrigin) Valid added in v0.60.0 

func (c ControlOrigin) Valid() bool

type ControlRequested added in v0.60.0 

type ControlRequested struct {
	Process ProcessID     `json:"process"`
	Kind    ControlKind   `json:"kind"`
	Origin  ControlOrigin `json:"origin"`
	Reason  string        `json:"reason"`
	At      time.Time     `json:"at"`
}

type Definition 

type Definition interface {
	Descriptor() Descriptor
	ChildPrograms() []Program[Payload, Payload]
	Start(input Payload) (Execution, error)
	Restore(ctx context.Context, state ExecutionState) (Execution, error)
}

Definition and Execution perform no external I/O. Their inputs contain every fact needed to replay a Step; clocks, randomness and Host context values must not influence a transition. Each restored transition is checked against its log.

type Delta 

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

func (Delta) EffectID 

func (d Delta) EffectID() EffectID

func (Delta) EffectSequence 

func (d Delta) EffectSequence() uint64

func (Delta) EmittedAt 

func (d Delta) EmittedAt() time.Time

func (Delta) Epoch added in v0.60.0 

func (d Delta) Epoch() Epoch

func (Delta) Payload 

func (d Delta) Payload() json.RawMessage

func (Delta) ProcessID 

func (d Delta) ProcessID() ProcessID

func (Delta) Valid 

func (d Delta) Valid() bool

type DeltaEmitter 

type DeltaEmitter func(json.RawMessage)

type Descriptor 

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

Descriptor is an immutable Definition contract. It contains no executable behavior or Program configuration.

func NewDescriptor 

func NewDescriptor(config DescriptorConfig) (Descriptor, error)

func (Descriptor) DecodeOutput 

func (d Descriptor) DecodeOutput[T any](output Payload) (T, error)

func (Descriptor) Description 

func (d Descriptor) Description() string

func (Descriptor) Digest 

func (d Descriptor) Digest() Digest

func (Descriptor) EncodeInput 

func (d Descriptor) EncodeInput[T any](value T) (Payload, error)

func (Descriptor) InputSchema 

func (d Descriptor) InputSchema() jsonschema.Schema

func (Descriptor) JSONSchemaModel 

func (Descriptor) JSONSchemaModel() any

func (Descriptor) MarshalJSON 

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

func (Descriptor) Name 

func (d Descriptor) Name() string

func (Descriptor) OutputSchema 

func (d Descriptor) OutputSchema() jsonschema.Schema

func (Descriptor) SignalSchema 

func (d Descriptor) SignalSchema() jsonschema.Schema

SignalSchema returns the unaddressed input contract; false rejects all input.

func (*Descriptor) UnmarshalJSON 

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

func (Descriptor) Valid 

func (d Descriptor) Valid() bool

func (Descriptor) ValidateInput 

func (d Descriptor) ValidateInput(input Payload) error

func (Descriptor) ValidateOutput 

func (d Descriptor) ValidateOutput(output Payload) error

func (Descriptor) ValidateSignal 

func (d Descriptor) ValidateSignal(input Payload) error

type DescriptorConfig 

type DescriptorConfig struct {
	Name string

	Description string

	InputSchema jsonschema.Schema

	OutputSchema jsonschema.Schema

	// SignalSchema declares unaddressed Host input. The zero value rejects all
	// unaddressed Signals. Addressed replies belong to their external wait;
	// Runtime-owned protocol frames never pass through this schema.
	SignalSchema jsonschema.Schema
}

DescriptorConfig contains the complete static contract of a Definition. Executable implementation and frozen configuration identity belong to a Program, not this contract.

type Digest 

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

Digest is a canonical SHA-256 content identity. Its zero value is invalid.

func ComputeDigest 

func ComputeDigest(data []byte) Digest

Hashes identify canonical bytes. Protocol owners must normalize equivalent JSON before hashing so representation details cannot create another identity.

func ParseDigest 

func ParseDigest(value string) (Digest, error)

func (Digest) IsZero added in v0.60.0 

func (i Digest) IsZero() bool

func (Digest) JSONSchemaModel 

func (Digest) JSONSchemaModel() any

func (Digest) MarshalText 

func (d Digest) MarshalText() ([]byte, error)

func (Digest) String 

func (i Digest) String() string

func (*Digest) UnmarshalText 

func (d *Digest) UnmarshalText(text []byte) error

func (Digest) Valid 

func (i Digest) Valid() bool

type Dispatcher 

type Dispatcher interface {
	Dispatch(context.Context, EffectRequest, DeltaEmitter) (Settlement, error)
	Policy(Effect) EffectPolicy
}

Policy is captured in EffectIntent before dispatch. Recovery uses that acknowledged declaration, never a newly queried replay decision. Dispatch must distinguish a definite result from interruption: an error leaves the outcome unknown. A work context canceled by local writer shutdown has ErrWriterStopped as its cause; this is not a durable business failure.

type Effect 

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

Each variant owns its operation; a payload is never inspected to guess a framework operation or to distinguish a completion from external input.

func NewChildCancelEffect 

func NewChildCancelEffect(child ProcessID, reason string) (Effect, error)

func NewChildSignalEffect 

func NewChildSignalEffect(child ProcessID, signal Signal) (Effect, error)

func NewChildStartEffect 

func NewChildStartEffect(spec ChildSpec) (Effect, error)

func NewDispatcherEffect 

func NewDispatcherEffect(payload json.RawMessage) (Effect, error)

func (Effect) ChildStart added in v0.60.0 

func (e Effect) ChildStart() (ChildSpec, bool)

ChildStart reports the child a framework Effect starts.

func (Effect) MarshalJSON 

func (e Effect) MarshalJSON() ([]byte, error)

func (Effect) Payload 

func (e Effect) Payload() Payload

func (Effect) Target 

func (e Effect) Target() EffectTarget

func (*Effect) UnmarshalJSON 

func (e *Effect) UnmarshalJSON(data []byte) error

func (Effect) Valid 

func (e Effect) Valid() bool

type EffectID 

type EffectID struct {
	Process ProcessID `json:"process"`
	Step    uint64    `json:"step"`
	Index   uint32    `json:"index" jsonwire:"required"`
}

func ParseEffectID 

func ParseEffectID(value string) (EffectID, error)

func (EffectID) IsZero added in v0.60.0 

func (e EffectID) IsZero() bool

func (EffectID) JSONSchemaModel 

func (EffectID) JSONSchemaModel() any

func (EffectID) MarshalJSON added in v0.60.0 

func (e EffectID) MarshalJSON() ([]byte, error)

func (EffectID) Precedes added in v0.60.0 

func (e EffectID) Precedes(other EffectID) bool

Precedes reports whether e was declared before other by the same process. Effects of different processes have no declaration order.

func (EffectID) String 

func (e EffectID) String() string

func (*EffectID) UnmarshalJSON added in v0.60.0 

func (e *EffectID) UnmarshalJSON(data []byte) error

func (EffectID) Valid 

func (e EffectID) Valid() bool

type EffectIntent added in v0.60.0 

type EffectIntent struct {
	ID     EffectID     `json:"id"`
	Effect Effect       `json:"effect"`
	Policy EffectPolicy `json:"policy"`
	At     time.Time    `json:"at"`
}

type EffectPolicy 

type EffectPolicy struct {
	Replay               ReplayPolicy  `json:"replay"`
	RequiredCapabilities CapabilitySet `json:"required_capabilities" jsonwire:"required"`
}

func (*EffectPolicy) UnmarshalJSON added in v0.60.0 

func (e *EffectPolicy) UnmarshalJSON(data []byte) error

func (EffectPolicy) Valid 

func (e EffectPolicy) Valid() bool

type EffectRequest 

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

func (EffectRequest) Effect 

func (e EffectRequest) Effect() Effect

func (EffectRequest) Epoch added in v0.60.0 

func (e EffectRequest) Epoch() Epoch

func (EffectRequest) ID 

func (e EffectRequest) ID() EffectID

func (EffectRequest) ProgramRef 

func (e EffectRequest) ProgramRef() ProgramRef

func (EffectRequest) Valid 

func (e EffectRequest) Valid() bool

type EffectSettled added in v0.60.0 

type EffectSettled struct {
	ID      EffectID   `json:"id"`
	Outcome Settlement `json:"outcome"`
	Failure *Failure   `json:"failure,omitzero"`
	At      time.Time  `json:"at"`
}

type EffectTarget 

type EffectTarget string
const (
	EffectTargetInvalid    EffectTarget = ""
	EffectTargetFramework  EffectTarget = "framework"
	EffectTargetDispatcher EffectTarget = "dispatcher"
)

func (EffectTarget) String 

func (e EffectTarget) String() string

func (EffectTarget) Valid 

func (e EffectTarget) Valid() bool

type Entry added in v0.60.0 

type Entry struct {
	LSN    LSN
	Epoch  Epoch
	Record Record
}

func (Entry) MarshalJSON added in v0.60.0 

func (e Entry) MarshalJSON() ([]byte, error)

func (*Entry) UnmarshalJSON added in v0.60.0 

func (e *Entry) UnmarshalJSON(data []byte) error

func (Entry) ValidateFor added in v0.61.0 

func (e Entry) ValidateFor(root ProcessID) error

ValidateFor admits one entry's structure and tree membership. A nonzero LSN is required, but the root's first Started legitimately has Epoch zero. Continuity, epoch progression, and replay consistency require the owning Log.

type Epoch added in v0.60.0 

type Epoch uint64

Epoch fences writers: each Activated advances it, and a Store rejects any batch appended under an older one.

type Execution 

type Execution interface {
	Step(ctx context.Context, turn Turn) (Transition, error)
	Snapshot() (ExecutionState, error)
}

type ExecutionFailed added in v0.60.0 

type ExecutionFailed struct {
	Process ProcessID `json:"process"`
	Failure Failure   `json:"failure"`
	At      time.Time `json:"at"`
}

type ExecutionState 

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

ExecutionState is an immutable envelope owned by one Execution Strategy. The Runtime persists and returns Payload without interpreting it. Its zero value is invalid.

func EncodeExecutionState 

func EncodeExecutionState[T any](kind string, value T) (ExecutionState, error)

EncodeExecutionState strictly encodes a typed strategy state and seals it with ParseExecutionState. Invalid UTF-8 and duplicate JSON names are rejected.

func ParseExecutionState 

func ParseExecutionState(kind string, payload json.RawMessage) (ExecutionState, error)

ParseExecutionState freezes an opaque payload under its owning Strategy kind.

func (ExecutionState) Decode 

func (e ExecutionState) Decode[T any](kind string) (T, error)

Decode checks the Strategy kind before decoding its payload into T. Unknown object members are rejected; the Strategy still validates its domain state.

func (ExecutionState) Kind 

func (e ExecutionState) Kind() string

func (ExecutionState) MarshalJSON 

func (e ExecutionState) MarshalJSON() ([]byte, error)

func (ExecutionState) Payload 

func (e ExecutionState) Payload() json.RawMessage

Payload returns an independently owned copy of the opaque Strategy state.

func (*ExecutionState) UnmarshalJSON 

func (e *ExecutionState) UnmarshalJSON(data []byte) error

func (ExecutionState) Valid 

func (e ExecutionState) Valid() bool

type Exit added in v0.60.0 

type Exit struct {
	Child       ProcessID   `json:"child"`
	Termination Termination `json:"termination"`
}

func (Exit) Valid added in v0.60.0 

func (e Exit) Valid() bool

type Exited added in v0.60.0 

type Exited struct {
	Process     ProcessID   `json:"process"`
	Termination Termination `json:"termination"`
	At          time.Time   `json:"at"`
}

type Failure 

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

Failure separates stable codes from diagnostic text so wording changes cannot alter control flow. Its bounded UTF-8 message must survive snapshot JSON and must exclude secrets.

func NewFailure 

func NewFailure(kind FailureKind, code, message string) (Failure, error)

NewFailure requires a trimmed UTF-8 message within MaxDiagnosticBytes.

func StepFailure 

func StepFailure(err error) (Failure, bool)

StepFailure reports the Failure the Runtime persists when Execution.Step returns an error wrapping a ClassifiedError. Its message includes the complete wrapped diagnostic. It reports false for an error the Runtime records as execution.step.failed, and for cancellation and contained panics, whose classifications the Runtime owns.

func (Failure) Classification 

func (f Failure) Classification() FailureClassification

func (Failure) Code 

func (f Failure) Code() string

func (Failure) JSONSchemaModel 

func (Failure) JSONSchemaModel() any

func (Failure) Kind 

func (f Failure) Kind() FailureKind

func (Failure) MarshalJSON 

func (f Failure) MarshalJSON() ([]byte, error)

func (Failure) Message 

func (f Failure) Message() string

func (*Failure) UnmarshalJSON 

func (f *Failure) UnmarshalJSON(data []byte) error

func (Failure) Valid 

func (f Failure) Valid() bool

type FailureClassification 

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

FailureClassification is the stable, diagnostic-free part of a Failure. Observations carry only this part, so they can be recorded and aggregated without exposing diagnostic text.

func (FailureClassification) Code 

func (f FailureClassification) Code() string

func (FailureClassification) Kind 

func (f FailureClassification) Kind() FailureKind

func (FailureClassification) Valid 

func (f FailureClassification) Valid() bool

type FailureKind 

type FailureKind string

FailureKind stays independent of retryability so classifying a failure cannot silently select a recovery or business policy. 

const (
	FailureKindInvalid FailureKind = ""
	// FailureKindExecution reports a Strategy error or an exhausted execution bound.
	FailureKindExecution FailureKind = "execution"
	// FailureKindContract reports an invalid protocol value or authority violation.
	FailureKindContract FailureKind = "contract"
	// FailureKindExternal reports a Host boundary error without a contained panic.
	FailureKindExternal FailureKind = "external"
	// FailureKindPanic reports a contained callback panic when that boundary produces
	// a Failure. Storage and API errors retain CallbackPanicError instead. This kind
	// does not establish an external operation's outcome or authorize replay.
	FailureKindPanic FailureKind = "panic"
)

func (FailureKind) String 

func (f FailureKind) String() string

func (FailureKind) Valid 

func (f FailureKind) Valid() bool

type Handle 

type Handle[O any] struct {
	// contains filtered or unexported fields
}

func (*Handle[O]) Answer 

func (h *Handle[O]) Answer[A any](ctx context.Context, wait WaitKey, value A) error

func (*Handle[O]) Await 

func (h *Handle[O]) Await(ctx context.Context) (Result[O], error)

func (*Handle[O]) Cancel 

func (h *Handle[O]) Cancel(ctx context.Context, reason string) error

func (*Handle[O]) DeliverSignals 

func (h *Handle[O]) DeliverSignals(ctx context.Context, signals ...Signal) (bool, error)

func (*Handle[O]) Epoch added in v0.60.0 

func (h *Handle[O]) Epoch() Epoch

func (*Handle[O]) ID 

func (h *Handle[O]) ID() ProcessID

func (*Handle[O]) Inspect added in v0.60.0 

func (h *Handle[O]) Inspect(ctx context.Context) (TreeInspection, error)

func (*Handle[O]) Kill 

func (h *Handle[O]) Kill(ctx context.Context, reason string) error

func (*Handle[O]) Observe 

func (h *Handle[O]) Observe(ctx context.Context, after LSN) iter.Seq2[Entry, error]

A terminal tree may acquire later writer epochs without another Exit. The first pull fixes its committed head as this observation's endpoint; further recovery cannot extend it. An active observation ends after the snapshot containing the root's exit. Local commits wake readers; other writers are polled, with each read starting after the last delivered position.

func (*Handle[O]) Pause 

func (h *Handle[O]) Pause(ctx context.Context, reason string) error

func (*Handle[O]) ProgramRef 

func (h *Handle[O]) ProgramRef() ProgramRef

func (*Handle[O]) ReplayUnknownEffect 

func (h *Handle[O]) ReplayUnknownEffect(ctx context.Context, id EffectID) error

func (*Handle[O]) ResolveUnknownEffect 

func (h *Handle[O]) ResolveUnknownEffect(ctx context.Context, id EffectID, outcome Settlement) error

func (*Handle[O]) Resume 

func (h *Handle[O]) Resume(ctx context.Context) error

type Inbox added in v0.60.0 

type Inbox []Message

func (Inbox) MarshalJSON added in v0.60.0 

func (i Inbox) MarshalJSON() ([]byte, error)

func (*Inbox) UnmarshalJSON added in v0.60.0 

func (i *Inbox) UnmarshalJSON(data []byte) error

type LSN added in v0.60.0 

type LSN uint64

LSN is a record's 1-based position in its tree's log, and therefore also a resumable observation cursor.

type Log added in v0.60.0 

type Log struct {
	Root    ProcessID `json:"root"`
	Entries []Entry   `json:"entries"`
}

Log is a contiguous segment of one tree's log that begins at the tree's Start or at a Checkpoint, so it can be folded without what precedes it.

func (Log) Clone added in v0.60.0 

func (l Log) Clone() (Log, error)

func (Log) Complete added in v0.60.0 

func (l Log) Complete() bool

Complete reports whether the log begins at the tree's Start rather than at a Checkpoint, which history projections require.

func (Log) Epoch added in v0.60.0 

func (l Log) Epoch() Epoch

func (Log) Head added in v0.60.0 

func (l Log) Head() LSN

func (Log) Inspect added in v0.60.0 

func (l Log) Inspect() (TreeInspection, error)

func (Log) Usage added in v0.60.0 

func (l Log) Usage() (Usage, error)

Usage requires the complete prefix because checkpoints omit reaped processes. A recovery cut cannot certify the tree's historical total.

func (Log) Validate added in v0.60.0 

func (l Log) Validate() error

type MemoryStore 

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

MemoryStore keeps each tree's log for the lifetime of the store value. It implements the complete Store contract, so it is a faithful stand-in for a durable backend in tests and single-process hosts.

func NewMemoryStore 

func NewMemoryStore() *MemoryStore

func (*MemoryStore) Append added in v0.60.0 

func (m *MemoryStore) Append(ctx context.Context, batch Batch) error

func (*MemoryStore) Load added in v0.60.0 

func (m *MemoryStore) Load(ctx context.Context, root ProcessID) (Log, bool, error)

func (*MemoryStore) Read added in v0.60.0 

func (m *MemoryStore) Read(ctx context.Context, root ProcessID, after LSN) iter.Seq2[Entry, error]

type Message added in v0.60.0 

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

type Payload 

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

Payload is an immutable canonical JSON value. Its zero value is invalid. ParsePayload and EncodePayload copy and normalize their input. Descriptor schemas define its role as Process input or final output; streamed Deltas never become final output implicitly.

func EncodePayload 

func EncodePayload[T any](value T) (Payload, error)

EncodePayload strictly encodes a typed value into an independently owned Payload. Invalid UTF-8 and duplicate JSON names are rejected, including custom codec output. Custom codecs are responsible for preserving their source values.

func ParsePayload 

func ParsePayload(data json.RawMessage) (Payload, error)

func (Payload) Decode 

func (p Payload) Decode[T any]() (T, error)

Decode strictly decodes p into a typed value. Unknown object fields are rejected when T is a struct.

func (Payload) Equal added in v0.61.0 

func (p Payload) Equal(other Payload) bool

Equal reports whether both payloads hold the same canonical JSON. Every constructor normalizes its input, so equal values have identical bytes; two absent payloads are equal.

func (Payload) IsZero 

func (p Payload) IsZero() bool

IsZero distinguishes an absent optional payload from an explicit JSON null.

func (Payload) JSON 

func (p Payload) JSON() json.RawMessage

JSON returns an independently owned JSON representation.

func (Payload) JSONSchemaModel 

func (Payload) JSONSchemaModel() any

func (Payload) MarshalJSON 

func (p Payload) MarshalJSON() ([]byte, error)

func (*Payload) UnmarshalJSON 

func (p *Payload) UnmarshalJSON(data []byte) error

func (Payload) Valid 

func (p Payload) Valid() bool

type PendingCheckpoint added in v0.60.0 

type PendingCheckpoint struct {
	Intent  EffectIntent `json:"intent"`
	Outcome *Settlement  `json:"outcome,omitzero"`
	Failure *Failure     `json:"failure,omitzero"`
}

type PendingEffect added in v0.60.0 

type PendingEffect struct {
	Request EffectRequest
	Policy  EffectPolicy
	Outcome Settlement
	Failure *Failure
}

type ProcessAdmission 

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

func (ProcessAdmission) Budget 

func (p ProcessAdmission) Budget() Budget

func (ProcessAdmission) Capabilities 

func (p ProcessAdmission) Capabilities() CapabilitySet

func (ProcessAdmission) Descriptor 

func (p ProcessAdmission) Descriptor() Descriptor

func (ProcessAdmission) ProcessID added in v0.60.0 

func (p ProcessAdmission) ProcessID() ProcessID

func (ProcessAdmission) ProgramRef 

func (p ProcessAdmission) ProgramRef() ProgramRef

func (ProcessAdmission) StartID 

func (p ProcessAdmission) StartID() (StartID, bool)

func (ProcessAdmission) Valid 

func (p ProcessAdmission) Valid() bool

type ProcessAdmitter 

type ProcessAdmitter interface {
	Admit(context.Context, ProcessAdmission) error
}

type ProcessAdmitterFunc 

type ProcessAdmitterFunc func(context.Context, ProcessAdmission) error

func (ProcessAdmitterFunc) Admit 

func (p ProcessAdmitterFunc) Admit(ctx context.Context, admission ProcessAdmission) error

type ProcessCheckpoint added in v0.60.0 

type ProcessCheckpoint struct {
	Started   Started             `json:"started"`
	State     ExecutionState      `json:"state,omitzero"`
	Inbox     Inbox               `json:"inbox"`
	Receipts  []SignalReceipt     `json:"receipts"`
	Children  []StartedChild      `json:"children"`
	Waits     []WaitCheckpoint    `json:"waits"`
	Pending   []PendingCheckpoint `json:"pending"`
	Usage     Usage               `json:"usage"`
	Allocated resourceAmounts     `json:"allocated"`
	Paused    bool                `json:"paused"`
	Yielded   bool                `json:"yielded"`
	Terminal  *Termination        `json:"terminal,omitzero"`
	Exited    *Exited             `json:"exited,omitzero"`
}

type ProcessID 

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

func ParseProcessID 

func ParseProcessID(value string) (ProcessID, error)

func (ProcessID) Child added in v0.60.0 

func (p ProcessID) Child(key ChildKey) (ProcessID, error)

func (ProcessID) Contains added in v0.60.0 

func (p ProcessID) Contains(other ProcessID) bool

func (ProcessID) Depth added in v0.60.0 

func (p ProcessID) Depth() uint32

func (ProcessID) IsZero added in v0.60.0 

func (p ProcessID) IsZero() bool

func (ProcessID) JSONSchemaModel 

func (ProcessID) JSONSchemaModel() any

func (ProcessID) Key added in v0.60.0 

func (p ProcessID) Key() (ChildKey, bool)

func (ProcessID) MarshalText 

func (p ProcessID) MarshalText() ([]byte, error)

func (ProcessID) Parent added in v0.60.0 

func (p ProcessID) Parent() (ProcessID, bool)

func (ProcessID) Root added in v0.60.0 

func (p ProcessID) Root() ProcessID

func (ProcessID) String 

func (p ProcessID) String() string

func (*ProcessID) UnmarshalText 

func (p *ProcessID) UnmarshalText(data []byte) error

func (ProcessID) Valid 

func (p ProcessID) Valid() bool

type ProcessInspection 

type ProcessInspection struct {
	ID             ProcessID
	Program        ProgramRef
	Status         Status
	Termination    Termination
	Usage          Usage
	StartedAt      time.Time
	FinishedAt     time.Time
	Waits          []Wait
	PendingEffects []PendingEffect
}

type Program 

type Program[I, O any] struct {
	// contains filtered or unexported fields
}

Program is an immutable binding of one Definition, its optional external Dispatcher, and an exact value reference used by Process logs.

func Bind 

func Bind[I, O any](compiler *Compiler, config BindingConfig) (Program[I, O], error)
Example 
package main

import (
	"context"
	"fmt"

	"github.com/Tangerg/scope/agent/run"
	"github.com/Tangerg/scope/core/jsonschema"
)

func main() {
	ctx := context.Background()
	runtime := exampleValue(run.New(ctx, run.Config{Build: "example", Store: run.NewMemoryStore()}))
	defer func() {
		if err := runtime.Close(ctx); err != nil {
			panic(err)
		}
	}()
	schema := exampleValue(jsonschema.For[string]())
	descriptor := exampleValue(run.NewDescriptor(run.DescriptorConfig{
		Name: "example.echo", Description: "Return the admitted input.", InputSchema: schema, OutputSchema: schema,
	}))
	program := exampleValue(run.Bind[string, string](runtime.Compiler(), run.BindingConfig{
		Definition: echoDefinition{descriptor: descriptor},
	}))
	output := exampleValue(runtime.Run(ctx, program, "hello"))
	fmt.Println(output)
}

func exampleValue[T any](value T, err error) T {
	if err != nil {
		panic(err)
	}
	return value
}

const echoStateKind = "example.echo"

type echoDefinition struct{ descriptor run.Descriptor }

func (e echoDefinition) Descriptor() run.Descriptor                           { return e.descriptor }
func (echoDefinition) ChildPrograms() []run.Program[run.Payload, run.Payload] { return nil }
func (echoDefinition) Start(input run.Payload) (run.Execution, error) {
	return echoExecution{input: input}, nil
}
func (echoDefinition) Restore(_ context.Context, state run.ExecutionState) (run.Execution, error) {
	input, err := state.Decode[string](echoStateKind)
	if err != nil {
		return nil, err
	}
	payload, err := run.EncodePayload(input)
	if err != nil {
		return nil, err
	}
	return echoExecution{input: payload}, nil
}

type echoExecution struct{ input run.Payload }

func (e echoExecution) Snapshot() (run.ExecutionState, error) {
	return run.ParseExecutionState(echoStateKind, e.input.JSON())
}
func (e echoExecution) Step(_ context.Context, turn run.Turn) (run.Transition, error) {
	return run.Complete(uint32(len(turn.Messages)), e.input)
}

Output:
hello

func (Program[I, O]) Definition 

func (p Program[I, O]) Definition() Definition

func (Program[I, O]) Descriptor 

func (p Program[I, O]) Descriptor() Descriptor

func (Program[I, O]) Erase 

func (p Program[I, O]) Erase() Program[Payload, Payload]

Erase preserves the same frozen binding for heterogeneous child graphs.

func (Program[I, O]) ProgramRef 

func (p Program[I, O]) ProgramRef() ProgramRef

func (Program[I, O]) Valid 

func (p Program[I, O]) Valid() bool

Valid checks the frozen binding without invoking user code. The Runtime checks the live Definition contract at startup and restoration boundaries.

type ProgramRef 

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

func (ProgramRef) Build added in v0.60.0 

func (p ProgramRef) Build() string

func (ProgramRef) Digest 

func (p ProgramRef) Digest() Digest

func (ProgramRef) JSONSchemaModel 

func (ProgramRef) JSONSchemaModel() any

func (ProgramRef) MarshalJSON 

func (p ProgramRef) MarshalJSON() ([]byte, error)

func (ProgramRef) Name 

func (p ProgramRef) Name() string

func (ProgramRef) String 

func (p ProgramRef) String() string

func (*ProgramRef) UnmarshalJSON 

func (p *ProgramRef) UnmarshalJSON(data []byte) error

func (ProgramRef) Valid 

func (p ProgramRef) Valid() bool

type ProgramResolver 

type ProgramResolver interface {
	Resolve(ProgramRef) (Program[Payload, Payload], error)
}

type QualifiedName 

type QualifiedName string

QualifiedName is a lowercase dotted identifier carried on the wire. Its value is validated by ValidQualifiedName; JSONSchemaPattern projects that same rule for schema generation from the same compiled grammar.

func (QualifiedName) JSONSchemaPattern 

func (QualifiedName) JSONSchemaPattern() string

type Quota 

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

Quota is an immutable optional upper bound. Its zero value is unlimited; NewQuota constructs a finite bound, including zero. A Quota is policy, never a usage counter or a reservation. Its JSON object preserves this distinction: {"maximum":null} is unlimited and {"maximum":0} permits no work.

Example 
unlimited := Quota{}
finite := NewQuota(3)
zero := NewQuota(0)
fmt.Println(unlimited.Allows(10001), finite.Allows(2, 1), finite.Allows(3, 1), zero.Allows(1))

Output:
true true false false

func NewQuota 

func NewQuota(maximum uint64) Quota

func (Quota) Allows 

func (q Quota) Allows(quantities ...uint64) bool

Allows reports whether all quantities fit together, without unsigned overflow. Unlimited quotas impose no bound; callers still own counter and identity overflow.

func (Quota) JSONSchemaModel 

func (Quota) JSONSchemaModel() any

func (Quota) MarshalJSON 

func (q Quota) MarshalJSON() ([]byte, error)

func (Quota) Maximum 

func (q Quota) Maximum() (uint64, bool)

Maximum returns the finite upper bound, or false for an unlimited quota.

func (*Quota) UnmarshalJSON 

func (q *Quota) UnmarshalJSON(data []byte) error

type Reaped added in v0.60.0 

type Reaped struct {
	Process ProcessID `json:"process"`
}

type Record added in v0.60.0 

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

Record is the closed set of facts a tree's log holds. Only this package implements it, so a Store can rely on EncodeRecord and DecodeRecord alone.

func DecodeRecord added in v0.60.0 

func DecodeRecord(kind RecordKind, data []byte) (Record, error)

type RecordKind added in v0.60.0 

type RecordKind string

RecordKind is the storage discriminator of a Record; a Store persists it beside the encoded body so DecodeRecord can restore the exact variant. 

const (
	RecordKindStarted          RecordKind = "started"
	RecordKindActivated        RecordKind = "activated"
	RecordKindSignalAdmitted   RecordKind = "signal_admitted"
	RecordKindStepped          RecordKind = "stepped"
	RecordKindEffectIntent     RecordKind = "effect_intent"
	RecordKindEffectSettled    RecordKind = "effect_settled"
	RecordKindExited           RecordKind = "exited"
	RecordKindReaped           RecordKind = "reaped"
	RecordKindCheckpoint       RecordKind = "checkpoint"
	RecordKindControlRequested RecordKind = "control_requested"
	RecordKindExecutionFailed  RecordKind = "execution_failed"
)

func EncodeRecord added in v0.60.0 

func EncodeRecord(record Record) (RecordKind, []byte, error)

type ReplayPolicy 

type ReplayPolicy string
const (
	ReplayPolicyInvalid      ReplayPolicy = ""
	ReplayPolicyNever        ReplayPolicy = "never"
	ReplayPolicySameIdentity ReplayPolicy = "same_identity"
)

func (ReplayPolicy) String 

func (r ReplayPolicy) String() string

func (ReplayPolicy) Valid 

func (r ReplayPolicy) Valid() bool

type Result 

type Result[O any] struct {
	// contains filtered or unexported fields
}

func (Result[O]) FinishedAt 

func (r Result[O]) FinishedAt() time.Time

func (Result[O]) Output 

func (r Result[O]) Output() (O, error)

func (Result[O]) ProcessID 

func (r Result[O]) ProcessID() ProcessID

func (Result[O]) StartedAt 

func (r Result[O]) StartedAt() time.Time

func (Result[O]) Termination 

func (r Result[O]) Termination() Termination

func (Result[O]) Usage 

func (r Result[O]) Usage() Usage

func (Result[O]) Valid 

func (r Result[O]) Valid() bool

type Runtime 

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

func New 

func New(ctx context.Context, config Config) (*Runtime, error)

func (*Runtime) Answered added in v0.60.0 

func (r *Runtime) Answered(ctx context.Context, process ProcessID, wait WaitKey, answer Payload) (bool, error)

Answered reports whether wait already admitted exactly this answer, so a caller can confirm a delivery whose acknowledgment was lost without sending it twice. A different answer to the same wait is ErrSignalConflict.

func (*Runtime) Close 

func (r *Runtime) Close(ctx context.Context) error

func (*Runtime) Compiler added in v0.60.0 

func (r *Runtime) Compiler() *Compiler

func (*Runtime) History added in v0.60.0 

func (r *Runtime) History(ctx context.Context, root ProcessID) (Log, error)

History reads the complete log of root's tree, for projections that need facts a checkpoint no longer carries, such as settled tool rounds.

func (*Runtime) Inspect added in v0.60.0 

func (r *Runtime) Inspect(ctx context.Context, root ProcessID) (TreeInspection, error)

func (*Runtime) Process 

func (r *Runtime) Process(id ProcessID) (*Handle[Payload], bool)

func (*Runtime) Recover 

func (r *Runtime) Recover[I, O any](ctx context.Context, program Program[I, O], root ProcessID) (*Handle[O], error)

func (*Runtime) ReleaseTree 

func (r *Runtime) ReleaseTree(ctx context.Context, root ProcessID) error

func (*Runtime) Run 

func (r *Runtime) Run[I, O any](ctx context.Context, program Program[I, O], input I) (O, error)

func (*Runtime) Start 

func (r *Runtime) Start[I, O any](ctx context.Context, program Program[I, O], request StartRequest[I]) (*Handle[O], error)

type RuntimeError 

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

func (*RuntimeError) After added in v0.60.0 

func (r *RuntimeError) After() LSN

func (*RuntimeError) Epoch added in v0.60.0 

func (r *RuntimeError) Epoch() Epoch

func (*RuntimeError) Error 

func (r *RuntimeError) Error() string

func (*RuntimeError) ProcessID 

func (r *RuntimeError) ProcessID() ProcessID

func (*RuntimeError) UnresolvedEffectIDs 

func (r *RuntimeError) UnresolvedEffectIDs() []EffectID

func (*RuntimeError) Unwrap 

func (r *RuntimeError) Unwrap() error

type Settlement 

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

Settlement is the immutable final fact for one Effect. The Effect it answers owns the EffectID, so a Settlement is addressed by where it is returned or recorded rather than by a copy of that identity. Status owns whether the Effect definitely succeeded, definitely failed, or remains unknown. Failure owns the diagnostic of a definite failure. Payload carries optional domain evidence independently, including work completed before a rejection. The Runtime delivers a definite settlement as a Completion message.

func NewFailedSettlement added in v0.60.0 

func NewFailedSettlement(failure Failure, details Payload) (Settlement, error)

func NewSettlement 

func NewSettlement(status SettlementStatus, payload json.RawMessage) (Settlement, error)

func (Settlement) Failure added in v0.60.0 

func (s Settlement) Failure() (Failure, bool)

func (Settlement) JSONSchemaModel added in v0.60.0 

func (Settlement) JSONSchemaModel() any

func (Settlement) MarshalJSON 

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

func (Settlement) Payload 

func (s Settlement) Payload() Payload

Payload is domain evidence. A failed settlement may omit it; its diagnostic remains available through Failure without interpreting the domain's schema.

func (Settlement) SpawnResult added in v0.60.0 

func (s Settlement) SpawnResult() (ProcessID, Failure, error)

SpawnResult reads the outcome of a Spawn: the started child, or the Failure that rejected the start. Exactly one of them is valid.

func (Settlement) Status 

func (s Settlement) Status() SettlementStatus

func (*Settlement) UnmarshalJSON 

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

func (Settlement) Valid 

func (s Settlement) Valid() bool

type SettlementStatus 

type SettlementStatus string

SettlementStatus records whether an Effect definitely succeeded, definitely failed, or has an unknown external result. Unknown never implies safe retry. 

const (
	SettlementStatusInvalid   SettlementStatus = ""
	SettlementStatusSucceeded SettlementStatus = "succeeded"
	SettlementStatusFailed    SettlementStatus = "failed"
	SettlementStatusUnknown   SettlementStatus = "unknown"
)

func (SettlementStatus) String 

func (s SettlementStatus) String() string

func (SettlementStatus) Valid 

func (s SettlementStatus) Valid() bool

type Signal 

type Signal struct {
	ID      SignalID `json:"id"`
	Wait    WaitKey  `json:"wait,omitzero"`
	Payload Payload  `json:"payload"`
}

func (Signal) Valid 

func (s Signal) Valid() bool

type SignalAdmitted added in v0.60.0 

type SignalAdmitted struct {
	Process ProcessID `json:"process"`
	Signal  Signal    `json:"signal"`
	At      time.Time `json:"at"`
}

type SignalID 

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

func ParseSignalID 

func ParseSignalID(value string) (SignalID, error)

func (SignalID) IsZero added in v0.60.0 

func (i SignalID) IsZero() bool

func (SignalID) JSONSchemaModel 

func (SignalID) JSONSchemaModel() any

func (SignalID) MarshalText 

func (i SignalID) MarshalText() ([]byte, error)

func (SignalID) String 

func (i SignalID) String() string

func (*SignalID) UnmarshalText 

func (s *SignalID) UnmarshalText(data []byte) error

func (SignalID) Valid 

func (i SignalID) Valid() bool

type SignalReceipt 

type SignalReceipt struct {
	ID     SignalID `json:"id"`
	Wait   WaitKey  `json:"wait,omitzero"`
	Digest Digest   `json:"digest"`
}

type StartConflictError 

type StartConflictError struct {
	Root   ProcessID
	Intent Digest
}

func (*StartConflictError) Error 

func (s *StartConflictError) Error() string

type StartID 

type StartID struct {
	Scope string `json:"scope"`
	Key   string `json:"key"`
}

func (StartID) Root added in v0.60.0 

func (s StartID) Root() ProcessID

func (StartID) Valid 

func (s StartID) Valid() bool

Valid reports whether a Host may submit the identity. The foreground scope belongs to Runtime.Run, whose random keys must not collide with Host keys.

type StartRequest 

type StartRequest[I any] struct {
	ID    StartID
	Input I
}

type Started added in v0.60.0 

type Started struct {
	Process      ProcessID     `json:"process"`
	Program      ProgramRef    `json:"program"`
	Input        Payload       `json:"input"`
	Budget       Budget        `json:"budget" jsonwire:"required"`
	Capabilities CapabilitySet `json:"capabilities" jsonwire:"required"`
	Limits       TreeLimits    `json:"limits,omitzero"`
	At           time.Time     `json:"at"`
}

func (Started) IntentDigest added in v0.60.0 

func (s Started) IntentDigest() (Digest, error)

IntentDigest is derived from the accepted declaration. Persisting a separately writable digest would let a stale cache accept a conflicting idempotent Start.

type StartedChild added in v0.60.0 

type StartedChild struct {
	Key     ChildKey `json:"key"`
	Started LSN      `json:"started"`
}

type Status 

type Status string

Status is the complete common lifecycle state of a Process. Strategy-specific conditions such as a Planning no-plan result do not add common statuses. 

const (
	StatusInvalid   Status = ""
	StatusRunning   Status = "running"
	StatusWaiting   Status = "waiting"
	StatusPaused    Status = "paused"
	StatusDraining  Status = "draining"
	StatusCompleted Status = "completed"
	StatusFailed    Status = "failed"
	StatusCanceled  Status = "canceled"
	StatusTimedOut  Status = "timed_out"
	StatusKilled    Status = "killed"
)

func (Status) JSONSchemaEnum added in v0.60.0 

func (Status) JSONSchemaEnum() []string

func (Status) MarshalText 

func (s Status) MarshalText() ([]byte, error)

func (Status) String 

func (s Status) String() string

func (Status) Terminal 

func (s Status) Terminal() bool

Terminal reports whether the Process may never transition again.

func (*Status) UnmarshalText 

func (s *Status) UnmarshalText(text []byte) error

func (Status) Valid 

func (s Status) Valid() bool

type Stepped added in v0.60.0 

type Stepped struct {
	Process    ProcessID  `json:"process"`
	Step       uint64     `json:"step"`
	Delivered  uint32     `json:"delivered" jsonwire:"required"`
	Transition Transition `json:"transition"`
}

type Store 

type Store interface {
	// Append is the only operation that advances a tree. It must atomically
	// compare the writer Epoch and the head position before extending the log,
	// and acknowledge a byte-identical retry of an applied batch.
	Append(ctx context.Context, batch Batch) error
	// Load returns the recovery segment: the newest Checkpoint entry and every
	// entry after it, or the whole log when it holds no Checkpoint.
	Load(ctx context.Context, root ProcessID) (Log, bool, error)
	// Read yields the committed entries after position in order. It reads one
	// consistent snapshot and ends at that snapshot's head.
	Read(ctx context.Context, root ProcessID, after LSN) iter.Seq2[Entry, error]
}

Store owns one append-only log per root tree. Recovery and history are separate reads so neither pays for the other's range: Load skips the prefix covered by the newest checkpoint, and Read is incremental.

type Termination 

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

Termination is the immutable result of applying the terminal priority matrix. Only the Runtime creates terminal facts from validated control intents and Step outcomes; callers obtain this observation from Result or decode it. Failure owns failed causes and diagnostics, Output owns completion, and the remaining control causes own their reason. Status is always derived from the cause.

func (Termination) Cause 

func (t Termination) Cause() TerminationCause

func (Termination) Failure 

func (t Termination) Failure() (Failure, bool)

func (Termination) JSONSchemaModel 

func (Termination) JSONSchemaModel() any

func (Termination) MarshalJSON 

func (t Termination) MarshalJSON() ([]byte, error)

func (Termination) Output 

func (t Termination) Output() (Payload, bool)

Output returns the final semantic result only for StatusCompleted. A present JSON null is a completed output.

func (Termination) Reason 

func (t Termination) Reason() string

Reason returns a bounded diagnostic reason. Completion has an empty reason.

func (Termination) Status 

func (t Termination) Status() Status

func (*Termination) UnmarshalJSON 

func (t *Termination) UnmarshalJSON(data []byte) error

func (Termination) Valid 

func (t Termination) Valid() bool

type TerminationCause 

type TerminationCause string
const (
	TerminationCauseInvalid            TerminationCause = ""
	TerminationCauseCompletion         TerminationCause = "completion"
	TerminationCauseRuntimeKill        TerminationCause = "runtime_kill"
	TerminationCauseParentDeadline     TerminationCause = "parent_deadline"
	TerminationCauseHostDeadline       TerminationCause = "host_deadline"
	TerminationCauseParentCancellation TerminationCause = "parent_cancellation"
	TerminationCauseHostCancellation   TerminationCause = "host_cancellation"
	TerminationCauseExecutionFailure   TerminationCause = "execution_failure"
	TerminationCauseContractFailure    TerminationCause = "contract_failure"
	TerminationCauseExternalFailure    TerminationCause = "external_failure"
	TerminationCausePanic              TerminationCause = "panic"
)

func (TerminationCause) String 

func (t TerminationCause) String() string

func (TerminationCause) Valid 

func (t TerminationCause) Valid() bool

type TerminationError 

type TerminationError struct {
	ID          ProcessID
	Termination Termination
}

func (*TerminationError) Error 

func (t *TerminationError) Error() string

type Transition 

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

func Complete 

func Complete(consumed uint32, output Payload) (Transition, error)

func Continue 

func Continue(consumed uint32, effects ...Effect) (Transition, error)

func Fail 

func Fail(consumed uint32, failure Failure) (Transition, error)

func Yield added in v0.60.0 

func Yield(consumed uint32, effects []Effect, waits ...Wait) (Transition, error)

func (Transition) Consumed added in v0.60.0 

func (t Transition) Consumed() uint32

func (Transition) Effects 

func (t Transition) Effects() []Effect

func (Transition) Failure 

func (t Transition) Failure() (Failure, bool)

func (Transition) Kind 

func (t Transition) Kind() TransitionKind

func (Transition) MarshalJSON 

func (t Transition) MarshalJSON() ([]byte, error)

func (Transition) Output 

func (t Transition) Output() (Payload, bool)

func (*Transition) UnmarshalJSON 

func (t *Transition) UnmarshalJSON(data []byte) error

func (Transition) Valid 

func (t Transition) Valid() bool

func (Transition) Waits added in v0.60.0 

func (t Transition) Waits() []Wait

type TransitionKind 

type TransitionKind string
const (
	TransitionKindInvalid  TransitionKind = ""
	TransitionKindContinue TransitionKind = "continue"
	TransitionKindYield    TransitionKind = "yield"
	TransitionKindComplete TransitionKind = "complete"
	TransitionKindFail     TransitionKind = "fail"
)

func (TransitionKind) String 

func (t TransitionKind) String() string

func (TransitionKind) Valid 

func (t TransitionKind) Valid() bool

type TreeInspection 

type TreeInspection struct {
	Root      ProcessID
	Head      LSN
	Epoch     Epoch
	Processes []ProcessInspection
}

func (TreeInspection) Process 

func (t TreeInspection) Process(id ProcessID) (ProcessInspection, bool)

type TreeLimits 

type TreeLimits struct {
	MaxRecordBytes     Quota  `json:"max_record_bytes" jsonwire:"required"`
	MaxCheckpointBytes Quota  `json:"max_checkpoint_bytes" jsonwire:"required"`
	MaxPendingMessages uint64 `json:"max_pending_messages" jsonwire:"required"`
	MaxDepth           uint32 `json:"max_depth" jsonwire:"required"`
	MaxChildren        Quota  `json:"max_children" jsonwire:"required"`
	MaxActiveChildren  uint32 `json:"max_active_children" jsonwire:"required"`
	MaxTreeProcesses   Quota  `json:"max_tree_processes" jsonwire:"required"`
}

Root Started owns the resolved policy. Child records never repeat these limits. Message capacity reserves Completions and child Exits before dispatch, so admitted work can finish even while external Signals arrive.

func DefaultTreeLimits 

func DefaultTreeLimits() TreeLimits

DefaultTreeLimits returns conservative structured-concurrency and mailbox bounds; cumulative counts and record bytes remain unlimited.

func (*TreeLimits) UnmarshalJSON 

func (t *TreeLimits) UnmarshalJSON(data []byte) error

func (TreeLimits) Valid 

func (t TreeLimits) Valid() bool

type Turn added in v0.60.0 

type Turn struct {
	Process  ProcessID
	Step     uint64
	Messages []Message
}

func (Turn) EffectID added in v0.60.0 

func (t Turn) EffectID(index uint32) EffectID

func (Turn) Valid added in v0.60.0 

func (t Turn) Valid() bool

type Usage 

type Usage struct {
	CommittedSteps   uint64 `json:"committed_steps" jsonwire:"required"`
	PreparedEffects  uint64 `json:"prepared_effects" jsonwire:"required"`
	AcceptedMessages uint64 `json:"accepted_messages" jsonwire:"required"`
}

Counters are projections of committed records; transient delta delivery cannot change durable usage. Each decoded counter must be present, including zero.

func (*Usage) UnmarshalJSON 

func (u *Usage) UnmarshalJSON(data []byte) error

type Wait 

type Wait struct {
	Key            WaitKey           `json:"key"`
	Question       Payload           `json:"question,omitzero"`
	AnswerSchema   jsonschema.Schema `json:"answer_schema,omitzero"`
	MaxAnswerBytes int               `json:"max_answer_bytes"`
}

func (Wait) Admits added in v0.61.0 

func (w Wait) Admits(payload Payload) error

Admits is the one admission rule for an answer to this Wait: the byte bound, where zero means MaxPayloadBytes, and then the answer schema. The Runtime applies it to every addressed Signal; Hosts may check an answer early with it.

func (Wait) Valid added in v0.60.0 

func (w Wait) Valid() bool

type WaitCheckpoint added in v0.61.0 

type WaitCheckpoint struct {
	Wait    Wait `json:"wait"`
	Pending bool `json:"pending"`
}

Retaining the first contract after withdrawal keeps a WaitKey immutable even when recovery starts after its original declaration.

type WaitKey 

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

func ParseWaitKey 

func ParseWaitKey(value string) (WaitKey, error)

func (WaitKey) IsZero added in v0.60.0 

func (i WaitKey) IsZero() bool

func (WaitKey) JSONSchemaModel 

func (WaitKey) JSONSchemaModel() any

func (WaitKey) MarshalText 

func (i WaitKey) MarshalText() ([]byte, error)

func (WaitKey) String 

func (i WaitKey) String() string

func (*WaitKey) UnmarshalText 

func (w *WaitKey) UnmarshalText(data []byte) error

func (WaitKey) Valid 

func (i WaitKey) Valid() bool

Source Files

View all Source files

Directories

Path Synopsis
postgres module
Package runtest exercises public execution and append-log contracts with deterministic fixtures.
 Package runtest exercises public execution and append-log contracts with deterministic fixtures.

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.