store

package
v0.14.0 Latest Latest
Warning

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

 Go to latest Published: Sep 14, 2026 License: Apache-2.0 Imports: 18 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) LatestUsageIntervalDay added in v0.14.0 

func (s *AgentEventStore) LatestUsageIntervalDay(ctx context.Context, fromDay, toDay string) (string, error)

LatestUsageIntervalDay returns the most recent local day with five-minute rows in the inclusive range, or "" when none. It lets the Service choose the default detail day without loading every bucket in the range.

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) QueryUsageDaily added in v0.14.0 

func (s *AgentEventStore) QueryUsageDaily(ctx context.Context, fromDay, toDay string) ([]UsageDailyRow, error)

QueryUsageDaily returns rows for the inclusive local-day range, ascending by day. Both bounds are YYYY-MM-DD; an empty bound is unconstrained.

func (*AgentEventStore) QueryUsageIntervals added in v0.14.0 

func (s *AgentEventStore) QueryUsageIntervals(ctx context.Context, fromDay, toDay string) ([]UsageIntervalRow, error)

QueryUsageIntervals returns the canonical 5-minute rows for an inclusive local-day range, ordered by day and bucket. Both bounds are YYYY-MM-DD; an empty bound is unconstrained.

func (*AgentEventStore) RebuildUsageRollups added in v0.14.0 

func (s *AgentEventStore) RebuildUsageRollups(ctx context.Context) (UsageRebuildResult, error)

RebuildUsageRollups replaces the daily and intraday Usage projections from every canonical event retained in the Agent journal. It is an explicit maintenance operation rather than a normal write path: callers should warn the user that the current Usage aggregates are discarded, while the journal and every other Host database remain unchanged.

The whole replacement is one SQLite transaction. A failed or cancelled rebuild therefore leaves the previous Usage projections intact, and running it again is idempotent because the source journal is immutable.

func (*AgentEventStore) RebuildUsageRollupsFromObservations added in v0.14.0 

func (s *AgentEventStore) RebuildUsageRollupsFromObservations(
	ctx context.Context,
	observations []UsageRebuildObservation,
) (UsageRebuildResult, error)

RebuildUsageRollupsFromObservations replaces Usage projections from parser output collected from historical transcript files. Only the derived Usage tables are touched; the canonical journal and every other database remain unchanged.

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.

func (*AgentEventStore) RepriceUsageDaily added in v0.14.0 

func (s *AgentEventStore) RepriceUsageDaily(ctx context.Context, table *usage.PriceTable) (int, error)

RepriceUsageDaily recomputes the cached cost of every rollup row from the supplied price table and returns how many rows changed.

Cost is recomputed rather than accrued per call because unit prices change. The tokens are the durable fact; the money is a projection of them, so a price correction has to be able to restate history rather than only affect new spend. Rows keep their tokens either way.

priced_calls records how many of a row's calls had a fully known price. When it trails calls, the row's cost is a lower bound and every aggregate built from it must be presented as such.

func (*AgentEventStore) SetUsageAttributionResolver added in v0.14.0 

func (s *AgentEventStore) SetUsageAttributionResolver(resolver UsageAttributionResolver)

SetUsageAttributionResolver installs the project attribution lookup. Usage accumulation is inert until a resolver is set, which keeps stores constructed by tests and embedders free of a host-state dependency.

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

type UsageAttribution added in v0.14.0 

type UsageAttribution struct {
	// ProjectID is empty when the stream cannot be attributed to a project.
	// An empty value is filed as unattributed rather than dropped, because
	// dropping it would make the panel total disagree with reality.
	ProjectID string
}

UsageAttribution is the ownership context a spend row is filed under. The journal does not know about projects, so the Service supplies this through SetUsageAttributionResolver rather than the store reaching into host state.

type UsageAttributionResolver added in v0.14.0 

type UsageAttributionResolver func(streamID string) UsageAttribution

UsageAttributionResolver maps a canonical stream to its ownership context.

type UsageDailyRow added in v0.14.0 

type UsageDailyRow struct {
	LocalDay    string `json:"localDay"`
	Provider    string `json:"provider"`
	Model       string `json:"model"`
	ModelRaw    string `json:"modelRaw"`
	ProjectID   string `json:"projectId,omitempty"`
	Calls       int64  `json:"calls"`
	FreshInput  int64  `json:"freshInput"`
	CacheWrite  int64  `json:"cacheWrite"`
	CacheRead   int64  `json:"cacheRead"`
	Output      int64  `json:"output"`
	Reasoning   int64  `json:"reasoning,omitempty"`
	CostNanoUSD int64  `json:"costNanoUsd"`
	PricedCalls int64  `json:"pricedCalls"`
}

UsageDailyRow is one aggregated spend row.

type UsageIntervalRow added in v0.14.0 

type UsageIntervalRow struct {
	LocalDay       string `json:"localDay"`
	BucketStartMin int    `json:"bucketStartMin"`
	Provider       string `json:"provider"`
	Model          string `json:"model"`
	ModelRaw       string `json:"modelRaw"`
	ProjectID      string `json:"projectId,omitempty"`
	Calls          int64  `json:"calls"`
	FreshInput     int64  `json:"freshInput"`
	CacheWrite     int64  `json:"cacheWrite"`
	CacheRead      int64  `json:"cacheRead"`
	Output         int64  `json:"output"`
	Reasoning      int64  `json:"reasoning,omitempty"`
	CostNanoUSD    int64  `json:"costNanoUsd"`
	PricedCalls    int64  `json:"pricedCalls"`
}

UsageIntervalRow is one 5-minute intraday aggregate. BucketStartMin is the bucket's local-minute offset from midnight, always a multiple of 5.

type UsageRebuildObservation added in v0.14.0 

type UsageRebuildObservation struct {
	Provider   string
	Model      string
	ProjectID  string
	OccurredAt time.Time
	Usage      api.AgentUsage
}

UsageRebuildObservation is one provider usage observation read directly from a historical transcript. It is deliberately separate from the canonical journal event: rebuilding Usage must be able to correct an old parser's duplicate rows without rewriting that immutable journal.

type UsageRebuildResult added in v0.14.0 

type UsageRebuildResult struct {
	Events int64
	Calls  int64
	Days   int64
}

UsageRebuildResult describes a completed replacement of the Usage projections. The canonical Agent journal is intentionally left untouched; only the two derived Usage tables are cleared and rebuilt.

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.