state

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Aug 20, 2026 License: MIT Imports: 15 Imported by: 0

Documentation

Overview

Package state provides structured graph state, paths, contracts, patches, and snapshots.

Index

Constants

View Source 
const (
	// SectionShared stores graph-wide business state.
	SectionShared = "shared"
	// SectionScopes stores state namespaced by node or agent scope.
	SectionScopes = "scopes"
	// SectionInternal is reserved for framework internals.
	SectionInternal = "internal"
	// SectionRuntime stores checkpoint/runtime metadata.
	SectionRuntime = "runtime"
)
View Source 
const DefaultSnapshotVersion = "state-v2"

Variables

This section is empty.

Functions

func Append 

func Append[T any](writer Writer, ref Ref[[]T], value T) error

Append records an append operation for a slice ref while preserving the concrete []T type when the path is initially missing.

func CloneValue 

func CloneValue(value any) (any, error)

CloneValue returns a cycle-safe best-effort copy of value. Pointer and map identity, plus repeated references to the same slice view, are preserved within the copied object graph, including cycles. Distinct overlapping slice views are copied independently and need not share backing storage. The returned error means full isolation could not be proven because the graph contains an opaque reference value. In that case the returned copy may retain aliases and must not be exposed as an isolated view.

func CountKeys 

func CountKeys(current *State) int

func CountScopes 

func CountScopes(current *State) int

func Delete 

func Delete[T any](writer Writer, ref Ref[T]) error

Delete records a delete operation for ref.

func DeletePath 

func DeletePath(current *State, path string) error

func Get 

func Get[T any](reader Reader, ref Ref[T]) (T, error)

Get returns the value at ref or a descriptive error that distinguishes missing paths from type mismatches.

func IsNilReducer 

func IsNilReducer(reducer Reducer) bool

func Merge 

func Merge(writer Writer, ref Ref[map[string]any], value map[string]any) error

Merge records an object merge operation for a map ref.

func MergePath 

func MergePath(current *State, path string, value map[string]any) error

func Read 

func Read[T any](reader Reader, ref Ref[T]) (T, bool)

Read returns the value at ref and false when the path is missing or the value has a different Go type. JSON checkpoint restore preserves JSON-compatible shapes, but does not reconstruct arbitrary Go slice, map, or struct types; capability views that expose typed values should convert restored JSON shapes explicitly.

func ReadPath 

func ReadPath(current *State, path string) (any, bool)

func ReadRequired 

func ReadRequired[T any](reader Reader, ref Ref[T]) (T, error)

ReadRequired is the error-returning form of Read kept for call-site clarity.

func Reduce 

func Reduce[T any](writer Writer, ref Ref[T], value T) error

func Replace 

func Replace[T any](writer Writer, ref Ref[T], value T) error

Replace records a replace operation for ref.

func ResolveStateValue 

func ResolveStateValue(root any, segments []string) (any, bool)

func SetPath 

func SetPath(current *State, path string, value any) error

func SplitStatePath 

func SplitStatePath(path string) []string

func SummaryFields 

func SummaryFields(current *State) []zap.Field

func ValidateJSONSchemaDefinition 

func ValidateJSONSchemaDefinition(schema JSONSchema) error

Types

type Access 

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

Access is the node-facing state API. Read-only access wraps a State directly; editing access records structured patch ops and applies them to a working copy so later reads observe earlier writes.

func NewAccess 

func NewAccess(current *State) *Access

NewAccess returns a read-only copy of state for inspection or condition evaluation. Mutating methods on the returned access fail.

func NewEditingAccess 

func NewEditingAccess(current *State) *Access

NewEditingAccess returns a copy-on-write view over state. Mutations update the working copy and are captured as a Patch for replay or parallel merging.

func NewEditingAccessWithReducers 

func NewEditingAccessWithReducers(current *State, reducers map[string]Reducer) *Access

func (*Access) AppendAny 

func (a *Access) AppendAny(path Path, value any) error

AppendAny records an append operation at path. Prefer the typed Append helper when appending to a Ref[[]T].

func (*Access) Delete 

func (a *Access) Delete(path Path) error

Delete records a delete operation at path.

func (*Access) MergeAny 

func (a *Access) MergeAny(path Path, value map[string]any) error

MergeAny records an object merge operation at path.

func (*Access) Patch 

func (a *Access) Patch() Patch

Patch returns the structured mutations recorded by editing access.

func (*Access) ReadAny 

func (a *Access) ReadAny(path Path) (any, bool)

ReadAny reads a cloned value at path. The returned value can be mutated by the caller without changing state-owned data.

func (*Access) ReduceAny 

func (a *Access) ReduceAny(path Path, reducer string, value any) error

func (*Access) SetAny 

func (a *Access) SetAny(path Path, value any) error

SetAny records a replace operation at path.

func (*Access) State 

func (a *Access) State() *State

State returns the current working state as a clone.

type AccessMode 

type AccessMode string

AccessMode describes how a node uses a state field. 

const (
	// AccessRead declares that a node reads the field.
	AccessRead AccessMode = "read"
	// AccessWrite declares that a node writes the field.
	AccessWrite AccessMode = "write"
	// AccessReadWrite declares that a node both reads and writes the field.
	AccessReadWrite AccessMode = "read_write"
)

type ArtifactRef 

type ArtifactRef struct {
	ID           string    `json:"id"`
	RunID        string    `json:"run_id,omitempty"`
	StepID       string    `json:"step_id,omitempty"`
	NodeID       string    `json:"node_id,omitempty"`
	OperationKey string    `json:"operation_key,omitempty"`
	ParentRunID  string    `json:"parent_run_id,omitempty"`
	ParentStepID string    `json:"parent_step_id,omitempty"`
	ParentTaskID string    `json:"parent_task_id,omitempty"`
	RootRunID    string    `json:"root_run_id,omitempty"`
	RunPath      []string  `json:"run_path,omitempty"`
	Namespace    string    `json:"namespace,omitempty"`
	Type         string    `json:"type,omitempty"`
	MIMEType     string    `json:"mime_type,omitempty"`
	Location     string    `json:"location,omitempty"`
	CreatedAt    time.Time `json:"created_at,omitempty"`
}

func CloneArtifactRefs 

func CloneArtifactRefs(artifacts []ArtifactRef) []ArtifactRef

type BranchPatch 

type BranchPatch struct {
	TaskID string
	NodeID string
	Order  int
	Patch  Patch
}

BranchPatch is one fan-in branch's state mutation.

Order must be stable for the graph wave. The merger uses Order, NodeID, then input order to make append and merge application deterministic.

type BreakpointHit 

type BreakpointHit struct {
	BreakpointID string    `json:"breakpoint_id"`
	NodeID       string    `json:"node_id"`
	Stage        string    `json:"stage"`
	HitAt        time.Time `json:"hit_at"`
}

type Change 

type Change struct {
	Path   string `json:"path"`
	Before any    `json:"before,omitempty"`
	After  any    `json:"after,omitempty"`
}

func DiffSnapshots 

func DiffSnapshots(before, after Snapshot) ([]Change, error)

type Codec 

type Codec interface {
	Name() string
	Version() string
	Encode(snapshot Snapshot) ([]byte, error)
	Decode(data []byte) (Snapshot, error)
	Diff(before, after Snapshot) ([]Change, error)
}

type Contract 

type Contract struct {
	Fields        []FieldAccess
	WildcardRead  bool
	WildcardWrite bool
}

Contract declares the state paths a node may read or write. It is used for input projection, write validation, and parallel merge conflict handling.

func NewContract 

func NewContract(fields ...FieldAccess) Contract

NewContract constructs a contract from cloned field declarations.

func (Contract) Clone 

func (c Contract) Clone() Contract

Clone returns an independent copy of the contract.

func (Contract) ReadPaths 

func (c Contract) ReadPaths() []Path

ReadPaths returns unique paths declared readable by the contract.

func (Contract) WritePaths 

func (c Contract) WritePaths() []Path

WritePaths returns unique paths declared writable by the contract.

type Editor 

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

Editor applies patch operations to a working state while recording them for later replay. Most node code should use Access instead.

func NewEditor 

func NewEditor(base *State) *Editor

NewEditor creates an editor over a cloned copy of base.

func NewEditorWithReducers 

func NewEditorWithReducers(base *State, reducers map[string]Reducer) *Editor

func (*Editor) AppendAny 

func (e *Editor) AppendAny(path Path, value any) error

AppendAny applies and records an append operation.

func (*Editor) Delete 

func (e *Editor) Delete(path Path) error

Delete applies and records a delete operation.

func (*Editor) MergeAny 

func (e *Editor) MergeAny(path Path, value map[string]any) error

MergeAny applies and records an object merge operation.

func (*Editor) Patch 

func (e *Editor) Patch() Patch

Patch returns a clone of the operations recorded so far.

func (*Editor) ReadAny 

func (e *Editor) ReadAny(path Path) (any, bool)

ReadAny reads from the editor's working state.

func (*Editor) ReduceAny 

func (e *Editor) ReduceAny(path Path, reducer string, value any) error

func (*Editor) SetAny 

func (e *Editor) SetAny(path Path, value any) error

SetAny applies and records a replace operation.

func (*Editor) State 

func (e *Editor) State() *State

State returns a clone of the editor's working state.

type FieldAccess 

type FieldAccess struct {
	Path        Path
	Mode        AccessMode
	Required    bool
	Merge       MergeStrategy
	Reducer     string
	Type        string
	Schema      JSONSchema
	Description string
}

FieldAccess declares one path in a node contract.

type JSONSchema 

type JSONSchema map[string]any

func (JSONSchema) Clone 

func (schema JSONSchema) Clone() JSONSchema

func (JSONSchema) Type 

func (schema JSONSchema) Type() string

type JSONStateCodec 

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

func NewJSONStateCodec 

func NewJSONStateCodec(version string) *JSONStateCodec

func (*JSONStateCodec) Decode 

func (c *JSONStateCodec) Decode(data []byte) (Snapshot, error)

func (*JSONStateCodec) Diff 

func (c *JSONStateCodec) Diff(before, after Snapshot) ([]Change, error)

func (*JSONStateCodec) Encode 

func (c *JSONStateCodec) Encode(snapshot Snapshot) ([]byte, error)

func (*JSONStateCodec) Name 

func (c *JSONStateCodec) Name() string

func (*JSONStateCodec) Version 

func (c *JSONStateCodec) Version() string

type MaxReducer 

type MaxReducer struct{}

func (MaxReducer) Reduce 

func (MaxReducer) Reduce(current, incoming any) (any, error)

type MergeStrategy 

type MergeStrategy string

MergeStrategy describes how concurrent branch writes should be reconciled. 

const (
	// MergeDefault lets the merge engine infer behavior from patch operations.
	MergeDefault MergeStrategy = ""
	// MergeReplace allows only identical concurrent replacements.
	MergeReplace MergeStrategy = "replace"
	// MergeMerge allows object merge operations with disjoint keys.
	MergeMerge MergeStrategy = "merge"
	// MergeAppend allows append operations to be ordered deterministically.
	MergeAppend MergeStrategy = "append"
)

type MessagesReducer 

type MessagesReducer struct{}

func (MessagesReducer) Reduce 

func (MessagesReducer) Reduce(current, incoming any) (any, error)

type ParallelMergeOptions 

type ParallelMergeOptions struct {
	Contracts map[string]Contract
	Schemas   map[string]JSONSchema
	Reducers  map[string]Reducer
}

type Patch 

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

Patch is an ordered list of state mutations. It is immutable from callers' perspective: constructors and readers clone operation values.

func NewPatch 

func NewPatch(ops ...PatchOp) Patch

NewPatch constructs a patch from cloned operations.

func (Patch) Apply 

func (p Patch) Apply(base *State) (*State, error)

Apply replays the patch against a clone of base and returns the resulting state.

func (Patch) ApplyWithReducers 

func (p Patch) ApplyWithReducers(base *State, reducers map[string]Reducer) (*State, error)

func (Patch) Empty 

func (p Patch) Empty() bool

Empty reports whether the patch has no operations.

func (Patch) MarshalJSON 

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

func (Patch) Ops 

func (p Patch) Ops() []PatchOp

Ops returns a cloned copy of patch operations.

func (*Patch) UnmarshalJSON 

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

type PatchOp 

type PatchOp struct {
	Kind    PatchOpKind `json:"kind"`
	Path    Path        `json:"path"`
	Value   any         `json:"value,omitempty"`
	Reducer string      `json:"reducer,omitempty"`
}

PatchOp is one state mutation recorded by an Editor.

type PatchOpKind 

type PatchOpKind string

PatchOpKind identifies how a patch operation updates its path. 

const (
	// OpSet replaces the value at Path.
	OpSet PatchOpKind = "set"
	// OpDelete removes the value at Path.
	OpDelete PatchOpKind = "delete"
	// OpMerge deep-merges an object value into Path.
	OpMerge PatchOpKind = "merge"
	// OpAppend appends a value or slice of values to Path.
	OpAppend PatchOpKind = "append"
	OpReduce PatchOpKind = "reduce"
)

type Path 

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

Path is the canonical address format used by state. Business code should construct paths through typed refs or capability views instead of parsing strings at call sites.

func Internal 

func Internal(namespace string, segments ...string) Path

Internal constructs a path in the internal section under namespace.

func MustPath 

func MustPath(section string, segments ...string) Path

MustPath is NewPath for package-level declarations; it panics on invalid input.

func NewPath 

func NewPath(section string, segments ...string) (Path, error)

NewPath constructs a validated state path from a known section and clean segments. Segments may not contain dots.

func ParsePath 

func ParsePath(text string) (Path, error)

ParsePath parses the dotted form used in DSL and event payloads, for example "shared.request.input".

func Runtime 

func Runtime(segments ...string) Path

Runtime constructs a path in the runtime metadata section.

func Scope 

func Scope(scope string, segments ...string) Path

Scope constructs a path in the scoped section under scope.

func Shared 

func Shared(segments ...string) Path

Shared constructs a path in the shared business state section.

func (Path) Child 

func (p Path) Child(segments ...string) (Path, error)

Child appends segments to an existing path and validates them.

func (Path) Empty 

func (p Path) Empty() bool

Empty reports whether path has no section and cannot address state.

func (Path) MarshalJSON 

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

MarshalJSON encodes a path as its canonical dotted string.

func (Path) MustChild 

func (p Path) MustChild(segments ...string) Path

MustChild is Child for package-level ref declarations; it panics on invalid input.

func (Path) Section 

func (p Path) Section() string

Section returns the root section of the path.

func (Path) Segments 

func (p Path) Segments() []string

Segments returns a copy of path segments after the section.

func (Path) String 

func (p Path) String() string

String returns the canonical dotted path.

func (*Path) UnmarshalJSON 

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

type Reader 

type Reader interface {
	ReadAny(path Path) (any, bool)
}

Reader is the minimal read-only state surface accepted by typed refs.

type Reducer 

type Reducer interface {
	Reduce(current, incoming any) (any, error)
}

type Ref 

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

Ref is a typed handle to one state path. It is intended to be declared once by a path-bound capability view and reused by business code.

func NewRef 

func NewRef[T any](path Path) Ref[T]

NewRef creates a typed reference to path.

func (Ref[T]) Path 

func (r Ref[T]) Path() Path

Path returns the state path addressed by the ref.

func (Ref[T]) ReadField 

func (r Ref[T]) ReadField() FieldAccess

ReadField converts the ref into a read-only contract field.

func (Ref[T]) ReadWriteField 

func (r Ref[T]) ReadWriteField() FieldAccess

ReadWriteField converts the ref into a read-write contract field.

func (Ref[T]) Required 

func (r Ref[T]) Required() Ref[T]

Required marks the field as required when the ref is converted into a contract field.

func (Ref[T]) WithDescription 

func (r Ref[T]) WithDescription(description string) Ref[T]

WithDescription sets human-readable contract documentation.

func (Ref[T]) WithMerge 

func (r Ref[T]) WithMerge(strategy MergeStrategy) Ref[T]

WithMerge sets the parallel merge strategy emitted in contract fields.

func (Ref[T]) WithReducer 

func (r Ref[T]) WithReducer(identifier string) Ref[T]

func (Ref[T]) WriteField 

func (r Ref[T]) WriteField() FieldAccess

WriteField converts the ref into a write-only contract field.

type RestoredStateSnapshot 

type RestoredStateSnapshot struct {
	Snapshot  Snapshot      `json:"snapshot"`
	Business  *State        `json:"business"`
	Runtime   RuntimeState  `json:"runtime"`
	Artifacts []ArtifactRef `json:"artifacts,omitempty"`
}

func RestoreStateSnapshot 

func RestoreStateSnapshot(snapshot Snapshot) (RestoredStateSnapshot, error)

type RuntimeState 

type RuntimeState struct {
	RunID           string         `json:"run_id,omitempty"`
	CurrentStepID   string         `json:"current_step_id,omitempty"`
	CurrentTaskID   string         `json:"current_task_id,omitempty"`
	CurrentNodeID   string         `json:"current_node_id,omitempty"`
	CurrentNodeIDs  []string       `json:"current_node_ids,omitempty"`
	CurrentStepIDs  []string       `json:"current_step_ids,omitempty"`
	NextNodeIDs     []string       `json:"next_node_ids,omitempty"`
	ParallelWaveID  string         `json:"parallel_wave_id,omitempty"`
	WaveID          string         `json:"wave_id,omitempty"`
	Status          string         `json:"status,omitempty"`
	RetryCount      int            `json:"retry_count,omitempty"`
	PauseRequested  bool           `json:"pause_requested,omitempty"`
	CancelRequested bool           `json:"cancel_requested,omitempty"`
	BreakpointHit   *BreakpointHit `json:"breakpoint_hit,omitempty"`
}

type Snapshot 

type Snapshot struct {
	Version  string         `json:"version"`
	Shared   map[string]any `json:"shared,omitempty"`
	Scopes   map[string]any `json:"scopes,omitempty"`
	Internal map[string]any `json:"internal,omitempty"`
	Runtime  map[string]any `json:"runtime,omitempty"`
}

func SnapshotFromState 

func SnapshotFromState(current *State) (Snapshot, error)

func SnapshotFromStateWithRuntime 

func SnapshotFromStateWithRuntime(current *State, runtime RuntimeState, artifacts []ArtifactRef) (Snapshot, error)

type State 

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

State is the private storage envelope for state. External packages should use Access, typed Ref values, or registered capability views instead of mutating this structure directly.

func FromMap 

func FromMap(input map[string]any) *State

FromMap creates state from an exported envelope. Missing root sections are initialized.

func FromShared 

func FromShared(shared map[string]any) *State

FromShared creates state with best-effort copies placed under the shared section. Opaque values that cannot be safely cloned may retain aliases.

func FromSnapshot 

func FromSnapshot(snapshot Snapshot) (*State, error)

func MergeParallelPatches 

func MergeParallelPatches(base *State, branches []BranchPatch, options ParallelMergeOptions) (*State, error)

MergeParallelPatches applies fan-in branch patches to a shared base state in a deterministic order and rejects ambiguous overlapping writes.

func MergeResumeInput 

func MergeResumeInput(base *State, input *State) (*State, error)

func NewState 

func NewState() *State

NewState creates an empty state envelope with all root sections present.

func PrepareContinuationState 

func PrepareContinuationState(base *State, input *State) (*State, error)

func ProjectStateByContract 

func ProjectStateByContract(full *State, contract Contract) *State

func (*State) Clone 

func (s *State) Clone() *State

Clone returns a cycle-safe best-effort copy of state. Callers that require proven isolation should use CloneStrict.

func (*State) CloneStrict 

func (s *State) CloneStrict() (*State, error)

CloneStrict returns an isolated copy or an error when state contains an opaque mutable value that cannot be safely cloned.

func (*State) Export 

func (s *State) Export() map[string]any

Export returns a cycle-safe best-effort copy of the state envelope.

func (*State) MarshalJSON 

func (s *State) MarshalJSON() ([]byte, error)

MarshalJSON encodes the exported state envelope.

func (*State) SetSection 

func (s *State) SetSection(section string, values map[string]any) error

SetSection replaces one root section. The value must be an object.

func (*State) UnmarshalJSON 

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

UnmarshalJSON decodes an exported state envelope and initializes missing root sections.

type SumReducer 

type SumReducer struct{}

func (SumReducer) Reduce 

func (SumReducer) Reduce(current, incoming any) (any, error)

type ValidationError 

type ValidationError struct {
	Boundary string            `json:"boundary"`
	Issues   []ValidationIssue `json:"issues"`
}

func NewValidationError 

func NewValidationError(boundary string, issues []ValidationIssue) *ValidationError

func (*ValidationError) Error 

func (validationErr *ValidationError) Error() string

type ValidationIssue 

type ValidationIssue struct {
	Path    string `json:"path,omitempty"`
	Kind    string `json:"kind"`
	Message string `json:"message"`
}

func ValidateAppliedInputPatchByContract 

func ValidateAppliedInputPatchByContract(inputState *State, patch Patch, contract Contract) []ValidationIssue

func ValidateAppliedPatchResultByContract 

func ValidateAppliedPatchResultByContract(resultState *State, patch Patch, contract Contract) []ValidationIssue

func ValidateContract 

func ValidateContract(contract Contract) []ValidationIssue

func ValidateInputPatch 

func ValidateInputPatch(patch Patch) []ValidationIssue

func ValidateInputPatchByContract 

func ValidateInputPatchByContract(patch Patch, contract Contract) []ValidationIssue

func ValidateInputPatchByContractWithReducers 

func ValidateInputPatchByContractWithReducers(base *State, patch Patch, contract Contract, reducers map[string]Reducer) []ValidationIssue

func ValidateJSONSchemaValue 

func ValidateJSONSchemaValue(value any, schema JSONSchema, basePath string) []ValidationIssue

func ValidateNodePatch 

func ValidateNodePatch(patch Patch) []ValidationIssue

func ValidatePatch 

func ValidatePatch(patch Patch) []ValidationIssue

func ValidatePatchByContract 

func ValidatePatchByContract(patch Patch, contract Contract) []ValidationIssue

func ValidatePatchResultByContract 

func ValidatePatchResultByContract(base *State, patch Patch, contract Contract) []ValidationIssue

func ValidatePatchResultByContractWithReducers 

func ValidatePatchResultByContractWithReducers(base *State, patch Patch, contract Contract, reducers map[string]Reducer) []ValidationIssue

func ValidateRequiredReads 

func ValidateRequiredReads(current *State, contract Contract) []ValidationIssue

func ValidateStateBySchemas 

func ValidateStateBySchemas(currentState *State, schemas map[string]JSONSchema) []ValidationIssue

type Writer 

type Writer interface {
	SetAny(path Path, value any) error
	Delete(path Path) error
	MergeAny(path Path, value map[string]any) error
	AppendAny(path Path, value any) error
	ReduceAny(path Path, reducer string, value any) error
}

Writer is the minimal mutation surface used by typed refs and capability views. Implementations are expected to record explicit patch operations.

Source Files

View all Source files

Jump to

Keyboard shortcuts

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