Documentation
¶
Overview ¶
Package state persists per-run state.json + events.jsonl under `<BaseDir>/workflows/<id>/runs/<run-id>/`. Atomic writes via the shared internal/agents/storage helpers. In-memory variant available for tests.
Index ¶
- func EvictIndex(indexDir string)
- type FileStore
- func (s *FileStore) AppendEvent(id, runID string, ev workflow.RunEvent) error
- func (s *FileStore) Delete(id, runID string) error
- func (s *FileStore) IndexAppend(id string, entry IndexEntry) error
- func (s *FileStore) IndexList(id string, page, pageSize int) ([]IndexEntry, bool, error)
- func (s *FileStore) ListEvents(id, runID string) ([]workflow.RunEvent, error)
- func (s *FileStore) ListEventsTail(id, runID string, limit int) ([]workflow.RunEvent, int, error)
- func (s *FileStore) ListRuns(id string) ([]string, error)
- func (s *FileStore) Load(id, runID string) (workflow.RunState, error)
- func (s *FileStore) Save(id, runID string, st workflow.RunState) error
- type IndexEntry
- type LokiPusher
- type Store
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func EvictIndex ¶ added in v0.15.1
func EvictIndex(indexDir string)
EvictIndex drops the cached Store for an index dir after its workflow is deleted, so the global cache stops pinning a dead directory.
Types ¶
type FileStore ¶
FileStore writes state.json + events.jsonl per run.
func (*FileStore) AppendEvent ¶
AppendEvent appends one line to events.jsonl atomically.
func (*FileStore) Delete ¶ added in v0.15.1
Delete removes a run's on-disk folder (state.json + events.jsonl) and drops its row from the sharded index.
func (*FileStore) IndexAppend ¶
func (s *FileStore) IndexAppend(id string, entry IndexEntry) error
IndexAppend persists one summary row to the id's sharded index. Constant-time regardless of total run history (touches only the current shard).
func (*FileStore) IndexList ¶
IndexList returns one page of summaries, newest first. pageSize defaults to shardedlog.DefaultShardMax. hasMore=true when older pages exist.
func (*FileStore) ListEvents ¶
ListEvents streams the full events.jsonl. Absent file → nil; corrupt lines are skipped so one bad row never blanks the history.
func (*FileStore) ListEventsTail ¶ added in v0.15.1
ListEventsTail returns the most recent `limit` events (chronological) plus the total valid count. limit <= 0 returns all; memory is O(limit).
type IndexEntry ¶
type IndexEntry struct {
ID string `json:"id"`
Status string `json:"status,omitempty"`
StartedAt time.Time `json:"at"`
EndedAt *time.Time `json:"end,omitempty"`
DurationMs int64 `json:"ms,omitempty"`
// Source tags how the run was kicked off so the editor can
// distinguish manual vs automation vs test fires in the runs
// list. Persisted as a short slug; the FE maps it to a coloured
// pill. Empty = legacy / unknown.
Source string `json:"src,omitempty"`
// TriggerID is the workflow.Trigger that fired, when one was
// identified. Empty for runs that fell back to graph.entry.
TriggerID string `json:"trig,omitempty"`
// TriggerType mirrors workflow.TriggerType for the firing
// trigger ("manual" / "cron" / "channel" / …). Kept separate
// from Source so the FE can show "automation" buckets without
// re-loading each run's state.json.
TriggerType string `json:"tt,omitempty"`
}
IndexEntry is the row shape stored in the per-id index. Kept lean so a 100-row shard stays well under 10KB.
type LokiPusher ¶
type LokiPusher struct {
// contains filtered or unexported fields
}
LokiPusher batches workflow RunEvents and pushes them to a Loki HTTP endpoint asynchronously. Events are queued into a channel; a background goroutine flushes every 5 seconds or whenever 100 entries accumulate.
Payload is Loki-compatible push body per https://grafana.com/docs/loki/latest/reference/loki-http-api/#push-log-entries-to-loki
Label scheme:
- wick_workflow = workflow id
- wick_run = run ID
- wick_event = event type (node_started, node_completed, …)
- any extra labels from ExtraLabels
Input/output JSON body is intentionally excluded from the pushed payload to avoid Loki size limits — only event metadata is sent. The full payload remains on disk in events.jsonl.
func NewLokiPusher ¶
func NewLokiPusher(lokiURL, extraLabelStr string) *LokiPusher
NewLokiPusher creates a pusher targeting lokiURL. extraLabelStr is an optional "key=value,key2=value2" string. Returns nil if lokiURL is empty.
func (*LokiPusher) Push ¶
func (p *LokiPusher) Push(id, runID string, ev workflow.RunEvent)
Push enqueues one event. Non-blocking — drops silently if queue full (Loki is a best-effort sink; disk events.jsonl is the source of truth).
func (*LokiPusher) Stop ¶
func (p *LokiPusher) Stop()
Stop drains pending entries and shuts down the background goroutine.
type Store ¶
type Store interface {
Save(id, runID string, st workflow.RunState) error
Load(id, runID string) (workflow.RunState, error)
AppendEvent(id, runID string, ev workflow.RunEvent) error
ListEvents(id, runID string) ([]workflow.RunEvent, error)
ListEventsTail(id, runID string, limit int) ([]workflow.RunEvent, int, error)
ListRuns(id string) ([]string, error)
Delete(id, runID string) error
IndexAppend(id string, entry IndexEntry) error
IndexList(id string, page, pageSize int) ([]IndexEntry, bool, error)
}
Store persists RunState + appends RunEvent for one workflow's runs/ folder.
Source Files