Documentation
¶
Overview ¶
Package store persists RuntimePulse state in a single SQLite database.
Index ¶
- type IngestResult
- type RetentionPolicy
- type Store
- func (s *Store) AddRule(r core.Rule) (core.Rule, error)
- func (s *Store) AddWatch(w watch.Watch) (watch.Watch, error)
- func (s *Store) ClaimContinuation(id string) (bool, error)
- func (s *Store) Close() error
- func (s *Store) GetSession(id string) (core.Session, bool, error)
- func (s *Store) Ingest(ev core.Event) (IngestResult, error)
- func (s *Store) InsertEvent(ev core.Event) error
- func (s *Store) InsertManualContinuation(sessionID, prompt string) (core.Continuation, error)
- func (s *Store) ListContinuations(state string) ([]core.Continuation, error)
- func (s *Store) ListEvents(evType string, limit int) ([]core.Event, error)
- func (s *Store) ListRules(includeInactive bool) ([]core.Rule, error)
- func (s *Store) ListSessions() ([]core.Session, error)
- func (s *Store) ListWatches() ([]watch.Watch, error)
- func (s *Store) NextPendingForSession(sessionID string) (core.Continuation, bool, error)
- func (s *Store) PendingSessions() ([]string, error)
- func (s *Store) PruneEvents(before time.Time) (int64, error)
- func (s *Store) RegisterSession(sess core.Session) (core.Session, error)
- func (s *Store) RemoveRule(id string) error
- func (s *Store) RemoveWatch(id string) error
- func (s *Store) ResetRunningContinuations() (int64, error)
- func (s *Store) SetSessionState(id string, st core.SessionState) error
- func (s *Store) Sweep(now time.Time, p RetentionPolicy) (SweepResult, error)
- func (s *Store) UpdateContinuationResult(id string, state core.ContinuationState, command string, exitCode *int, ...) error
- type SweepResult
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type IngestResult ¶
type IngestResult struct {
Event core.Event `json:"event"`
Continuations []core.Continuation `json:"continuations"`
}
IngestResult reports what one event produced.
type RetentionPolicy ¶ added in v1.2.0
type RetentionPolicy struct {
SessionIdle time.Duration
ContinuationAge time.Duration
EventAge time.Duration
}
RetentionPolicy bounds how long pruned state lives. Windows are fixed constants in the daemon (owner decision: no flags, YAGNI).
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
func (*Store) AddWatch ¶
AddWatch persists a watch; an existing watch with the same (type, target) is returned unchanged (idempotent create, spec §7.2).
func (*Store) ClaimContinuation ¶
ClaimContinuation atomically moves a continuation pending → running. Returns false if it was not pending (already claimed or finished) — the DB row is the single source of truth for claims.
func (*Store) Ingest ¶
func (s *Store) Ingest(ev core.Event) (IngestResult, error)
Ingest is the outbox transaction (spec §5): store the event, match active rules, render prompts, create pending continuations, and consume oneShot rules — atomically. (rule_id, event_id) uniqueness makes re-delivery of the same event a no-op.
Everything runs on the tx: with SetMaxOpenConns(1), touching s.db while the tx is open would deadlock the connection pool.
func (*Store) InsertEvent ¶
InsertEvent stores an event; re-inserting the same id is a no-op.
func (*Store) InsertManualContinuation ¶
func (s *Store) InsertManualContinuation(sessionID, prompt string) (core.Continuation, error)
InsertManualContinuation creates a rule-less pending continuation (the CLI `continue` command / continuation.run RPC).
func (*Store) ListContinuations ¶
func (s *Store) ListContinuations(state string) ([]core.Continuation, error)
ListContinuations returns continuations, optionally filtered by state.
func (*Store) ListEvents ¶
ListEvents returns newest-first events, optionally filtered by type.
func (*Store) ListRules ¶
ListRules returns rules; includeInactive=false filters out consumed and expired.
func (*Store) NextPendingForSession ¶
NextPendingForSession returns the oldest pending continuation for a session (FIFO order: created_at, then id).
func (*Store) PendingSessions ¶
PendingSessions returns the distinct session ids that have pending work.
func (*Store) PruneEvents ¶
PruneEvents deletes events older than the cutoff, returning the count.
func (*Store) RegisterSession ¶
RegisterSession upserts a session; new sessions start as waiting. Re-registering preserves existing state and created_at; the returned session is re-read from the database so it never disagrees with it.
func (*Store) RemoveRule ¶
func (*Store) RemoveWatch ¶
func (*Store) ResetRunningContinuations ¶
ResetRunningContinuations moves orphaned running rows back to pending (daemon crashed mid-run). At-least-once: a continuation whose result was lost may run twice; that beats a lost wake-up.
func (*Store) SetSessionState ¶
func (s *Store) SetSessionState(id string, st core.SessionState) error
func (*Store) Sweep ¶ added in v1.2.0
func (s *Store) Sweep(now time.Time, p RetentionPolicy) (SweepResult, error)
Sweep garbage-collects expired state in ONE transaction:
- sessions idle past SessionIdle — but never one that is running or referenced by an active (unconsumed, unexpired) rule: an armed session must survive until its rule fires or expires;
- terminal (completed/failed) continuations older than ContinuationAge — pending/running are never touched;
- events older than EventAge.
The three prunes are independent by design: deleting a session does not cascade into its historical continuations (those age out on their own and are inert once terminal).
func (*Store) UpdateContinuationResult ¶
func (s *Store) UpdateContinuationResult(id string, state core.ContinuationState, command string, exitCode *int, outputSummary string) error
UpdateContinuationResult records a finished run.
type SweepResult ¶ added in v1.2.0
SweepResult reports what one retention sweep removed.
Source Files