Documentation
¶
Index ¶
- Constants
- type CheckpointHistory
- func (h *CheckpointHistory) Append(ctx context.Context, snapshot runstate.RunSnapshot) error
- func (h *CheckpointHistory) Delete(ctx context.Context, runID string) error
- func (h *CheckpointHistory) List(ctx context.Context, runID string, limit int) ([]runstate.CheckpointSummary, error)
- func (h *CheckpointHistory) Load(ctx context.Context, runID string, version int64) (runstate.RunSnapshot, error)
- type CheckpointHistoryOption
- type Option
- type Repository
- func (r *Repository) Delete(ctx context.Context, runID string) error
- func (r *Repository) DeleteOutboxForRun(ctx context.Context, runID string) (int64, error)
- func (r *Repository) FetchUnpublishedOutbox(ctx context.Context, limit int) ([]runstate.OutboxEvent, error)
- func (r *Repository) List(ctx context.Context, filter runstate.ListFilter) ([]runstate.RunSnapshot, error)
- func (r *Repository) ListStale(ctx context.Context, filter runstate.ListFilter, grace time.Duration) ([]runstate.RunSnapshot, error)
- func (r *Repository) Load(ctx context.Context, runID string) (runstate.RunSnapshot, error)
- func (r *Repository) MarkOutboxPublished(ctx context.Context, id int64) error
- func (r *Repository) PurgeOutboxPublishedBefore(ctx context.Context, cutoff time.Time) (int64, error)
- func (r *Repository) Save(ctx context.Context, snapshot *runstate.RunSnapshot, expectedVersion int64) error
- func (r *Repository) SaveFenced(ctx context.Context, snapshot *runstate.RunSnapshot, expectedVersion int64, ...) error
- func (r *Repository) SaveWithEvents(ctx context.Context, snapshot *runstate.RunSnapshot, expectedVersion int64, ...) error
- func (r *Repository) SchemaVersion(ctx context.Context) (int, error)
Constants ¶
const DefaultCheckpointHistoryTable = "agentflow_run_checkpoint_history"
const DefaultEventTableName = "agentflow_runtime_events"
DefaultEventTableName is the durable runtime-event table. SaveWithEvents reads its per-run max sequence (under the same advisory lock the event store's Append takes) so outbox rows continue the run's sequence without colliding with directly appended events.
const DefaultOutboxTableName = "agentflow_outbox"
DefaultOutboxTableName is the transactional event outbox written by SaveWithEvents and drained by the framework relay (migration 0005).
const DefaultTableName = "agentflow_run_snapshots"
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type CheckpointHistory ¶ added in v0.2.0
type CheckpointHistory struct {
// contains filtered or unexported fields
}
func NewCheckpointHistory ¶ added in v0.2.0
func NewCheckpointHistory(db *sql.DB, opts ...CheckpointHistoryOption) (*CheckpointHistory, error)
func (*CheckpointHistory) Append ¶ added in v0.2.0
func (h *CheckpointHistory) Append(ctx context.Context, snapshot runstate.RunSnapshot) error
func (*CheckpointHistory) Delete ¶ added in v0.3.0
func (h *CheckpointHistory) Delete(ctx context.Context, runID string) error
Delete drops every recorded revision for a run; a missing run is a no-op.
func (*CheckpointHistory) List ¶ added in v0.2.0
func (h *CheckpointHistory) List(ctx context.Context, runID string, limit int) ([]runstate.CheckpointSummary, error)
func (*CheckpointHistory) Load ¶ added in v0.2.0
func (h *CheckpointHistory) Load(ctx context.Context, runID string, version int64) (runstate.RunSnapshot, error)
type CheckpointHistoryOption ¶ added in v0.2.0
type CheckpointHistoryOption func(*CheckpointHistory) error
func WithCheckpointHistoryTable ¶ added in v0.2.0
func WithCheckpointHistoryTable(name string) CheckpointHistoryOption
type Option ¶
type Option func(*Repository) error
func WithEventTableName ¶ added in v0.3.1
WithEventTableName tells SaveWithEvents which runtime-event table sequences must stay compatible with. It must match the observability event store's table; the default is the store's own default.
func WithOutboxTableName ¶ added in v0.3.1
WithOutboxTableName overrides the event outbox table. It must match the table the framework relay and the observability outbox sink use.
func WithTableName ¶
type Repository ¶
type Repository struct {
// contains filtered or unexported fields
}
func NewRepository ¶
func NewRepository(db *sql.DB, opts ...Option) (*Repository, error)
func (*Repository) DeleteOutboxForRun ¶ added in v0.3.1
DeleteOutboxForRun implements runstate.OutboxRepository.
func (*Repository) FetchUnpublishedOutbox ¶ added in v0.3.1
func (r *Repository) FetchUnpublishedOutbox(ctx context.Context, limit int) ([]runstate.OutboxEvent, error)
FetchUnpublishedOutbox implements runstate.OutboxRepository.
func (*Repository) List ¶ added in v0.1.1
func (r *Repository) List(ctx context.Context, filter runstate.ListFilter) ([]runstate.RunSnapshot, error)
func (*Repository) ListStale ¶ added in v0.3.1
func (r *Repository) ListStale(ctx context.Context, filter runstate.ListFilter, grace time.Duration) ([]runstate.RunSnapshot, error)
ListStale implements runstate.StaleRepository. The staleness comparison runs in SQL against NOW(), the database clock, so application-clock skew on any worker cannot shrink or stretch the grace window.
func (*Repository) Load ¶
func (r *Repository) Load(ctx context.Context, runID string) (runstate.RunSnapshot, error)
func (*Repository) MarkOutboxPublished ¶ added in v0.3.1
func (r *Repository) MarkOutboxPublished(ctx context.Context, id int64) error
MarkOutboxPublished implements runstate.OutboxRepository. The conditional update makes concurrent relays on several nodes safe: only the first marker affects a row, and an already-published row is not an error.
func (*Repository) PurgeOutboxPublishedBefore ¶ added in v0.3.1
func (r *Repository) PurgeOutboxPublishedBefore(ctx context.Context, cutoff time.Time) (int64, error)
PurgeOutboxPublishedBefore implements runstate.OutboxRepository. Only published rows in the caller's tenant scope are removed; unpublished rows are undelivered events.
func (*Repository) Save ¶
func (r *Repository) Save(ctx context.Context, snapshot *runstate.RunSnapshot, expectedVersion int64) error
func (*Repository) SaveFenced ¶ added in v0.3.1
func (r *Repository) SaveFenced(ctx context.Context, snapshot *runstate.RunSnapshot, expectedVersion int64, fenceToken uint64) error
SaveFenced implements runstate.FencedRepository. The update carries fence_token <= $token in its WHERE clause, atomically with the version CAS, so a superseded lease holder cannot overwrite a newer holder's snapshot even if it still passes the version check. When no row matches, a follow-up read distinguishes the cause: version moved -> ErrStaleSnapshot; a higher fence token is recorded -> ErrStaleFence.
func (*Repository) SaveWithEvents ¶ added in v0.3.1
func (r *Repository) SaveWithEvents(ctx context.Context, snapshot *runstate.RunSnapshot, expectedVersion int64, events []core.Event, fenceToken uint64) error
SaveWithEvents implements runstate.EventOutboxRepository: the snapshot save (version CAS plus, for a non-zero fenceToken, the same fence validation as SaveFenced) and the outbox inserts commit in ONE transaction, so a run's state and its parked lifecycle events either both land or both roll back — a fence conflict or stale version leaves no orphan outbox rows behind.
Each event's per-run sequence is minted inside the transaction under the same per-run advisory lock the observability event store's Append takes, continuing from the max sequence present in the event table and in still-unpublished outbox rows. The relay later inserts with exactly this sequence, so the event store's UNIQUE (run_id, sequence) constraint deduplicates redelivery.
func (*Repository) SchemaVersion ¶ added in v0.3.1
func (r *Repository) SchemaVersion(ctx context.Context) (int, error)
SchemaVersion reports the highest applied agentflow schema migration version visible to this repository's database. A database without the version table reports 0 (never migrated) instead of an error, so startup validation can demand migrations with a clear message.
Source Files