store

package
v0.13.0 Latest Latest
Warning

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

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

Documentation

Index

Constants

View Source 
const (
	CanonicalCommandPending   = "pending"
	CanonicalCommandCompleted = "completed"
	CanonicalCommandFailed    = "failed"
	// CanonicalCommandUnknown means the Host stopped after durable admission
	// but before recording a result. It must never be re-executed implicitly:
	// the provider side effect may already have happened.
	CanonicalCommandUnknown = "unknown"
)

Variables

View Source 
var (
	ErrCanonicalSequenceConflict = errors.New("canonical agent event sequence conflict")
	ErrCanonicalEventConflict    = errors.New("canonical agent event identity conflict")
	ErrCanonicalHistoryBoundary  = errors.New("canonical agent history boundary")
	ErrCanonicalCommandConflict  = errors.New("canonical agent command identity conflict")
	ErrCanonicalCommandPending   = errors.New("canonical agent command is pending")
	ErrCanonicalCommandUnknown   = errors.New("canonical agent command outcome is unknown")
)

Functions

func NewID 

func NewID() string

Types

type AgentEventStore added in v0.12.0 

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

AgentEventStore is the durable Host-owned canonical Agent journal.

Protocol 4 starts a clean data boundary: canonical stream rows and command admission records are the only persisted Agent state.

func OpenAgentEventStore added in v0.12.0 

func OpenAgentEventStore(dbPath string) (*AgentEventStore, error)

OpenAgentEventStore opens the canonical Agent journal database.

func (*AgentEventStore) AppendCanonicalEvents added in v0.12.0 

func (s *AgentEventStore) AppendCanonicalEvents(
	ctx context.Context,
	streamID, executionID string,
	events []api.CanonicalAgentEvent,
) ([]api.CanonicalAgentEvent, error)

AppendCanonicalEvents appends one batch atomically. A zero sequence is assigned by the Host; explicit sequences are accepted only when they match an existing immutable row. Replayed event IDs and positions are idempotent, while a different payload at either identity is a hard integrity error.

func (*AgentEventStore) AppendCanonicalEventsWithCheckpoint added in v0.12.0 

func (s *AgentEventStore) AppendCanonicalEventsWithCheckpoint(
	ctx context.Context,
	streamID, executionID string,
	events []api.CanonicalAgentEvent,
	checkpoint map[string]any,
) ([]api.CanonicalAgentEvent, error)

AppendCanonicalEventsWithCheckpoint commits the immutable event batch and the replaceable projection checkpoint in the same SQLite transaction. A checkpoint is only written when the argument is non-nil; callers that do not have a projection update retain the last durable checkpoint.

func (*AgentEventStore) BeginCanonicalCommand added in v0.12.0 

func (s *AgentEventStore) BeginCanonicalCommand(
	ctx context.Context,
	executionID, commandID, fingerprint string,
) (CanonicalCommandRecord, bool, error)

BeginCanonicalCommand atomically admits a command. The caller becomes the leader only when it inserted a new pending row. A completed or failed row is returned for idempotent replay; a pending row belongs to another attempt.

func (*AgentEventStore) CanonicalCheckpoint added in v0.12.0 

func (s *AgentEventStore) CanonicalCheckpoint(
	ctx context.Context,
	streamID string,
) (api.AgentProjectionCheckpoint, bool, error)

CanonicalCheckpoint returns the last checkpoint committed with a stream. It is deliberately separate from the event page so subscription code can establish a coherent snapshot while holding its broadcast fence.

func (*AgentEventStore) CanonicalExecution added in v0.12.0 

func (s *AgentEventStore) CanonicalExecution(ctx context.Context, streamID string) (api.AgentExecution, bool, error)

func (*AgentEventStore) Close added in v0.12.0 

func (s *AgentEventStore) Close() error

Close closes the underlying database.

func (*AgentEventStore) CompleteCanonicalCommand added in v0.12.0 

func (s *AgentEventStore) CompleteCanonicalCommand(
	ctx context.Context,
	executionID, commandID, fingerprint string,
	result any,
	callErr error,
) error

CompleteCanonicalCommand records the immutable result of an admitted command. Completion is intentionally independent from the request context: a disconnected client must not leave a successful provider call pending.

func (*AgentEventStore) GetCanonicalCommand added in v0.12.0 

func (s *AgentEventStore) GetCanonicalCommand(
	ctx context.Context,
	executionID, commandID string,
) (CanonicalCommandRecord, bool, error)

GetCanonicalCommand returns the durable admission record for one command. The boolean is false when the command has never been admitted.

func (*AgentEventStore) QueryCanonicalEvents added in v0.12.0 

func (s *AgentEventStore) QueryCanonicalEvents(
	ctx context.Context,
	streamID string,
	afterSequence, beforeSequence uint64,
	limit int,
) (api.AgentEventsHistoryResult, error)

QueryCanonicalEvents returns an ascending page. afterSequence is exclusive; beforeSequence is exclusive. With neither bound the newest page is returned.

func (*AgentEventStore) ReconcilePendingCanonicalCommands added in v0.12.0 

func (s *AgentEventStore) ReconcilePendingCanonicalCommands(
	ctx context.Context,
	now time.Time,
	maxAge time.Duration,
) (int64, error)

ReconcilePendingCanonicalCommands marks commands left pending by an older Host process as indeterminate. Admission is durable before the provider is called, so replaying such a row could duplicate a turn or an attachment; callers must surface the state and ask the client to issue a new command ID. Only rows older than maxAge are touched, so a long-running command in the current process is never converted while it is still executing.

type CanonicalCommandRecord added in v0.12.0 

type CanonicalCommandRecord struct {
	ExecutionID string
	CommandID   string
	Fingerprint string
	Status      string
	Result      any
	Error       string
	CreatedAt   int64
}

CanonicalCommandRecord is the durable admission record for one mutation. Result is decoded as generic JSON because command result types belong to the wire method, not to the journal. A pending row means another Host process may still be executing the command; it is never re-run implicitly.

type CanonicalHistoryBoundary added in v0.12.0 

type CanonicalHistoryBoundary struct {
	StreamID             string
	RetainedFromSequence uint64
	HeadSequence         uint64
	CheckpointSequence   uint64
	Checkpoint           map[string]any
}

CanonicalHistoryBoundary carries the first sequence still retained by the Host. Callers should return this as a structured history_boundary error and install a replacement checkpoint rather than guessing a cursor.

func (*CanonicalHistoryBoundary) Error added in v0.12.0 

func (e *CanonicalHistoryBoundary) Error() string

type StateResetError added in v0.12.0 

type StateResetError struct {
	Path           string
	FoundSchema    int
	RequiredSchema int
}

StateResetError is returned for unknown or future state files. Known older schemas are upgraded in place because their fields remain compatible.

func (*StateResetError) Error added in v0.12.0 

func (e *StateResetError) Error() string

type Store 

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

func Open 

func Open(path, hostName string) (*Store, error)

func (*Store) ChangesSince 

func (s *Store) ChangesSince(revision uint64) <-chan struct{}

func (*Store) Snapshot 

func (s *Store) Snapshot() api.State

func (*Store) SnapshotVersion 

func (s *Store) SnapshotVersion() (api.State, uint64)

func (*Store) Update 

func (s *Store) Update(fn func(*api.State) error) error

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.