Documentation
¶
Index ¶
- Constants
- type Config
- type OutboxSink
- type OutboxSinkConfig
- type Store
- func (store *Store) Append(ctx context.Context, event core.Event) (obspkg.EventRecord, error)
- func (store *Store) AppendSequenced(ctx context.Context, sequence int64, event core.Event) (obspkg.EventRecord, error)
- func (store *Store) DeleteEventsForRun(ctx context.Context, runID string) (int64, error)
- func (store *Store) EnsureSchema(ctx context.Context) error
- func (store *Store) ListEvents(ctx context.Context, runID string, query obspkg.EventQuery) ([]obspkg.EventRecord, error)
- func (store *Store) ListRuns(ctx context.Context, query obspkg.RunQuery) ([]obspkg.RunSummary, error)
- func (store *Store) ListScopedEvents(ctx context.Context, query obspkg.ScopedEventQuery) ([]obspkg.EventRecord, error)
- func (store *Store) PurgeEventsBefore(ctx context.Context, cutoff time.Time) (int64, error)
Constants ¶
const DefaultOutboxTableName = "agentflow_outbox"
DefaultOutboxTableName is the transactional event outbox drained by the framework relay (migration 0005). It must match the runstate repository's outbox table when both are customized.
const DefaultTableName = "agentflow_runtime_events"
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type OutboxSink ¶ added in v0.3.1
type OutboxSink struct {
// contains filtered or unexported fields
}
OutboxSink is a core.EventSink that keeps the durable event store and the run-state outbox consistent without touching the engine's emit path. Emit first tries Store.Append (the historical behavior, including live publication); when the durable append fails, the event is parked in the run-state outbox in the same database as the run snapshots — a single INSERT that survives the transient store failure — and Emit returns nil so the runtime's lifecycle retry does not re-append a duplicate. The framework's outbox relay (WithOutboxRelay) later redelivers parked events with their minted sequences, and the event store's UNIQUE (run_id, sequence) constraint deduplicates retries.
Live publication is skipped for parked events, matching the pre-existing behavior of a failed append (EventStoreSink publishes only after a successful store write).
func NewOutboxSink ¶ added in v0.3.1
func NewOutboxSink(config OutboxSinkConfig) (*OutboxSink, error)
type OutboxSinkConfig ¶ added in v0.3.1
type OutboxSinkConfig struct {
// Store is the durable event store this sink writes to first.
Store *Store
// OutboxTableName overrides the fallback outbox table; defaults to
// DefaultOutboxTableName.
OutboxTableName string
// Publishers receive every successfully stored event (same semantics as
// observability.EventStoreSink, e.g. an EventHub for live push).
Publishers []obspkg.EventPublisher
}
OutboxSinkConfig configures NewOutboxSink.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
func (*Store) AppendSequenced ¶ added in v0.3.1
func (store *Store) AppendSequenced(ctx context.Context, sequence int64, event core.Event) (obspkg.EventRecord, error)
AppendSequenced implements observability.SequencedEventStore for the framework outbox relay: the event is inserted with a sequence minted earlier (when it was parked in the outbox), continuing the run's regular sequence space. A row that already exists at (run_id, sequence) means the event was delivered before — by an earlier relay attempt that crashed before marking, or by a concurrent relay — so it is read back and returned with a nil error instead of being duplicated.
func (*Store) DeleteEventsForRun ¶ added in v0.3.1
DeleteEventsForRun implements observability.EventRunPurger for the retention cascade: when a run is deleted, its event history goes with it.
func (*Store) ListEvents ¶
func (store *Store) ListEvents(ctx context.Context, runID string, query obspkg.EventQuery) ([]obspkg.EventRecord, error)
func (*Store) ListScopedEvents ¶ added in v0.3.0
func (store *Store) ListScopedEvents(ctx context.Context, query obspkg.ScopedEventQuery) ([]obspkg.EventRecord, error)
func (*Store) PurgeEventsBefore ¶ added in v0.3.1
PurgeEventsBefore implements observability.EventStorePurger: removes events that occurred before cutoff, bounding the append-only event table. Note this also trims history of runs still alive past cutoff; callers pick the cutoff to match their run retention.
Source Files