database

Documentation

Overview

Package database provides the storage layer for Oculo.

It implements the Store interface using SQLite with WAL mode, FTS5 full-text search, and optimized indexes for time-series trace data. The DBService struct is the primary entry point for all database operations.

Index

• type DBService
  • func NewDBService(path string) (*DBService, error)
  • func (s *DBService) BatchInsertMemoryEvents(events []*MemoryEvent) error
  • func (s *DBService) BatchInsertSpans(spans []*Span) error
  • func (s *DBService) Close() error
  • func (s *DBService) CommitPendingPayload(writeID int64) error
  • func (s *DBService) GetMemoryDiffs(spanID string) ([]*MemoryEvent, error)
  • func (s *DBService) GetMemoryTimeline(key string, namespace string) ([]*MemoryEvent, error)
  • func (s *DBService) GetPendingPayloads() ([]PendingWrite, error)
  • func (s *DBService) GetTraceStats(traceID string) (*TraceStats, error)
  • func (s *DBService) InsertMemoryEvent(event *MemoryEvent) error
  • func (s *DBService) InsertSpan(span *Span) error
  • func (s *DBService) InsertToolCall(call *ToolCall) error
  • func (s *DBService) InsertTrace(trace *Trace) error
  • func (s *DBService) QueryTimeline(traceID string) ([]*Span, error)
  • func (s *DBService) QueryTraces(filter TraceFilter) ([]*Trace, error)
  • func (s *DBService) SearchContent(query string, limit int) ([]*Span, error)
  • func (s *DBService) WritePendingPayload(payload []byte) (int64, error)
• type MemoryEvent
• type PendingWrite
• type Span
• type Store
• type ToolCall
• type Trace
• type TraceFilter
• type TraceStats

Types

type DBService

type DBService struct {
	// contains filtered or unexported fields
}

DBService implements the Store interface using SQLite. It manages the database connection pool, prepared statements, and ensures thread-safe access through a read-write mutex.

func NewDBService

func NewDBService(path string) (*DBService, error)

NewDBService creates a new database service, initializes the schema, and prepares frequently-used statements.

The path parameter specifies the SQLite database file location. Use ":memory:" for in-memory databases (useful for testing).

func (*DBService) BatchInsertMemoryEvents

func (s *DBService) BatchInsertMemoryEvents(events []*MemoryEvent) error

BatchInsertMemoryEvents inserts multiple memory events within a single transaction for improved throughput.

func (*DBService) BatchInsertSpans

func (s *DBService) BatchInsertSpans(spans []*Span) error

BatchInsertSpans inserts multiple spans within a single transaction for improved throughput during batch ingestion.

func (*DBService) Close

func (s *DBService) Close() error

Close gracefully shuts down the database, closing all prepared statements and the underlying connection pool.

func (*DBService) CommitPendingPayload

func (s *DBService) CommitPendingPayload(writeID int64) error

CommitPendingPayload marks a pending write as committed.

func (*DBService) GetMemoryDiffs

func (s *DBService) GetMemoryDiffs(spanID string) ([]*MemoryEvent, error)

GetMemoryDiffs returns all memory events for a given span, ordered by timestamp. This powers the bottom diff pane in the TUI.

func (*DBService) GetMemoryTimeline

func (s *DBService) GetMemoryTimeline(key string, namespace string) ([]*MemoryEvent, error)

GetMemoryTimeline returns the full mutation history for a specific memory key within a namespace. This lets users answer: "When did the agent start believing X?"

func (*DBService) GetPendingPayloads

func (s *DBService) GetPendingPayloads() ([]PendingWrite, error)

GetPendingPayloads returns all uncommitted payloads for crash recovery.

func (*DBService) GetTraceStats

func (s *DBService) GetTraceStats(traceID string) (*TraceStats, error)

GetTraceStats returns aggregated statistics for a trace. Used by the TUI detail pane and the analysis engine.

func (*DBService) InsertMemoryEvent

func (s *DBService) InsertMemoryEvent(event *MemoryEvent) error

InsertMemoryEvent persists a memory mutation event.

func (*DBService) InsertSpan

func (s *DBService) InsertSpan(span *Span) error

InsertSpan persists a new span within an existing trace. If a span with the same ID already exists, it updates duration, completion, tokens, and status.

func (*DBService) InsertToolCall

func (s *DBService) InsertToolCall(call *ToolCall) error

InsertToolCall persists a tool call record.

func (*DBService) InsertTrace

func (s *DBService) InsertTrace(trace *Trace) error

InsertTrace persists a new trace record. If a trace with the same ID already exists, it updates the end_time, status, and metadata.

func (*DBService) QueryTimeline

func (s *DBService) QueryTimeline(traceID string) ([]*Span, error)

QueryTimeline returns all spans for a given trace, ordered by start_time. This is the primary query for the TUI timeline view.

func (*DBService) QueryTraces

func (s *DBService) QueryTraces(filter TraceFilter) ([]*Trace, error)

QueryTraces returns traces matching the given filter criteria. Results are ordered by start_time descending (most recent first).

func (*DBService) SearchContent

func (s *DBService) SearchContent(query string, limit int) ([]*Span, error)

SearchContent performs full-text search over prompt and completion content using the FTS5 index. Returns matching spans with BM25 relevance ranking.

func (*DBService) WritePendingPayload

func (s *DBService) WritePendingPayload(payload []byte) (int64, error)

WritePendingPayload stores a raw payload in the pending_writes table for crash recovery. Returns the write ID for later commitment.

type MemoryEvent

type MemoryEvent struct {
	EventID   string  `json:"event_id"`
	SpanID    string  `json:"span_id"`
	Timestamp int64   `json:"timestamp"`
	Operation string  `json:"operation"`
	Key       string  `json:"key"`
	OldValue  *string `json:"old_value,omitempty"`
	NewValue  *string `json:"new_value,omitempty"`
	Namespace string  `json:"namespace"`
}

MemoryEvent captures a single mutation to the agent's memory.

type PendingWrite

type PendingWrite struct {
	WriteID   int64  `json:"write_id"`
	Payload   []byte `json:"payload"`
	Status    string `json:"status"`
	CreatedAt int64  `json:"created_at"`
}

PendingWrite represents an uncommitted ingestion payload.

type Span

type Span struct {
	SpanID           string   `json:"span_id"`
	TraceID          string   `json:"trace_id"`
	ParentSpanID     *string  `json:"parent_span_id,omitempty"`
	OperationType    string   `json:"operation_type"`
	OperationName    string   `json:"operation_name"`
	StartTime        int64    `json:"start_time"`
	DurationMs       int64    `json:"duration_ms"`
	Prompt           *string  `json:"prompt,omitempty"`
	Completion       *string  `json:"completion,omitempty"`
	PromptTokens     int      `json:"prompt_tokens"`
	CompletionTokens int      `json:"completion_tokens"`
	Model            *string  `json:"model,omitempty"`
	Temperature      *float64 `json:"temperature,omitempty"`
	Metadata         *string  `json:"metadata,omitempty"`
	Status           string   `json:"status"`
	ErrorMessage     *string  `json:"error_message,omitempty"`
}

Span represents a single operation within a trace.

type Store

type Store interface {
	// InsertTrace persists a new trace record.
	InsertTrace(trace *Trace) error
	// InsertSpan persists a new span within an existing trace.
	InsertSpan(span *Span) error
	// InsertMemoryEvent persists a memory mutation event.
	InsertMemoryEvent(event *MemoryEvent) error
	// InsertToolCall persists a tool call record.
	InsertToolCall(call *ToolCall) error

	// BatchInsertSpans inserts multiple spans in a single transaction.
	BatchInsertSpans(spans []*Span) error
	// BatchInsertMemoryEvents inserts multiple memory events in a single transaction.
	BatchInsertMemoryEvents(events []*MemoryEvent) error

	// QueryTraces returns traces matching the given filter, ordered by start_time DESC.
	QueryTraces(filter TraceFilter) ([]*Trace, error)
	// QueryTimeline returns all spans for a trace, ordered by start_time.
	QueryTimeline(traceID string) ([]*Span, error)
	// GetMemoryDiffs returns all memory events for a span, ordered by timestamp.
	GetMemoryDiffs(spanID string) ([]*MemoryEvent, error)
	// GetMemoryTimeline returns the full mutation history for a memory key.
	GetMemoryTimeline(key string, namespace string) ([]*MemoryEvent, error)
	// SearchContent performs full-text search over prompt/completion content.
	SearchContent(query string, limit int) ([]*Span, error)
	// GetTraceStats returns aggregated statistics for a trace.
	GetTraceStats(traceID string) (*TraceStats, error)

	// WritePendingPayload stores a raw payload for crash recovery.
	WritePendingPayload(payload []byte) (int64, error)
	// CommitPendingPayload marks a pending write as committed.
	CommitPendingPayload(writeID int64) error
	// GetPendingPayloads returns all payloads that haven't been committed.
	GetPendingPayloads() ([]PendingWrite, error)

	// Close gracefully shuts down the database connection.
	Close() error
}

Store defines the interface for trace data persistence. This abstraction allows for mocking in tests and potential future backends beyond SQLite.

type ToolCall

type ToolCall struct {
	CallID        int64   `json:"call_id"`
	SpanID        string  `json:"span_id"`
	ToolName      string  `json:"tool_name"`
	ArgumentsJSON *string `json:"arguments_json,omitempty"`
	ResultJSON    *string `json:"result_json,omitempty"`
	Success       bool    `json:"success"`
	LatencyMs     int64   `json:"latency_ms"`
}

ToolCall captures an external tool invocation.

type Trace

type Trace struct {
	TraceID   string            `json:"trace_id"`
	AgentName string            `json:"agent_name"`
	StartTime int64             `json:"start_time"`
	EndTime   *int64            `json:"end_time,omitempty"`
	Status    string            `json:"status"`
	Metadata  map[string]string `json:"metadata,omitempty"`
}

Trace represents a complete execution trace of an AI agent.

type TraceFilter

type TraceFilter struct {
	AgentName *string `json:"agent_name,omitempty"`
	Status    *string `json:"status,omitempty"`
	Since     *int64  `json:"since,omitempty"` // Unix nanoseconds
	Until     *int64  `json:"until,omitempty"` // Unix nanoseconds
	Limit     int     `json:"limit"`
	Offset    int     `json:"offset"`
}

TraceFilter defines query parameters for trace listing.

type TraceStats

type TraceStats struct {
	TraceID               string `json:"trace_id"`
	TotalSpans            int    `json:"total_spans"`
	LLMCalls              int    `json:"llm_calls"`
	ToolCalls             int    `json:"tool_calls"`
	MemoryOps             int    `json:"memory_ops"`
	TotalPromptTokens     int    `json:"total_prompt_tokens"`
	TotalCompletionTokens int    `json:"total_completion_tokens"`
	TotalDurationMs       int64  `json:"total_duration_ms"`
	MemoryEventCount      int    `json:"memory_event_count"`
}

TraceStats holds aggregated statistics for a single trace.