sync

Documentation

Overview

Package sync provides conversation sync to observability backends.

Index

• func GenerateMessageHash(entry *jsonl.RawEntry) string
• type Backend
• type Mapper
  • func NewMapper(filePath string) *Mapper
  • func (m *Mapper) MapEntry(entry *jsonl.RawEntry, lineNum int) (*Span, error)
• type Span
• type SpanBatch
• type SpanKind
• type SyncResult
• type Syncer
  • func NewSyncer(opts SyncerOptions) (*Syncer, error)
  • func (s *Syncer) Close() error
  • func (s *Syncer) Stats() (*syncdb.Stats, error)
  • func (s *Syncer) SyncAll(ctx context.Context) (*SyncResult, error)
  • func (s *Syncer) SyncFile(ctx context.Context, path string) (int, error)
• type SyncerOptions

Functions

func GenerateMessageHash

func GenerateMessageHash(entry *jsonl.RawEntry) string

GenerateMessageHash creates a hash for deduplication.

Types

type Backend

type Backend interface {
	// Name returns the backend identifier.
	Name() string

	// SendSpan sends a single span to the backend.
	SendSpan(ctx context.Context, span *Span) error

	// SendBatch sends a batch of spans to the backend.
	SendBatch(ctx context.Context, batch *SpanBatch) error

	// Flush ensures all pending spans are sent.
	Flush(ctx context.Context) error

	// Close releases backend resources.
	Close() error
}

Backend defines the interface for sync backends.

type Mapper

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

Mapper converts JSONL entries to spans.

func NewMapper

func NewMapper(filePath string) *Mapper

NewMapper creates a new span mapper for a file.

func (*Mapper) MapEntry

func (m *Mapper) MapEntry(entry *jsonl.RawEntry, lineNum int) (*Span, error)

MapEntry converts a JSONL entry to a span. Returns nil if the entry should not produce a span.

type Span

type Span struct {
	// Identity
	ID       string `json:"id"`        // Unique span ID (derived from entry UUID or hash)
	TraceID  string `json:"trace_id"`  // Trace ID (conversation session ID)
	ParentID string `json:"parent_id"` // Parent span ID (for nested spans)

	// Timing
	StartTime time.Time `json:"start_time"`
	EndTime   time.Time `json:"end_time"`

	// Classification
	Kind SpanKind `json:"kind"`
	Name string   `json:"name"` // e.g., "user-message", "assistant-generation", "tool-Read"

	// Content
	Input  string `json:"input,omitempty"`  // User message or tool input
	Output string `json:"output,omitempty"` // Assistant response or tool output

	// Metadata
	Metadata map[string]interface{} `json:"metadata,omitempty"`

	// LLM-specific (for generation spans)
	Model       string `json:"model,omitempty"`
	TokensIn    int    `json:"tokens_in,omitempty"`
	TokensOut   int    `json:"tokens_out,omitempty"`
	IsStreaming bool   `json:"is_streaming,omitempty"`

	// Tool-specific
	ToolName   string `json:"tool_name,omitempty"`
	ToolResult string `json:"tool_result,omitempty"`
	IsError    bool   `json:"is_error,omitempty"`

	// Source info
	SourceFile string `json:"source_file"` // Original JSONL file path
	SourceLine int    `json:"source_line"` // Line number in JSONL file
}

Span represents a telemetry span for export.

type SpanBatch

type SpanBatch struct {
	TraceID   string    `json:"trace_id"`
	SessionID string    `json:"session_id"`
	Project   string    `json:"project"`
	Spans     []*Span   `json:"spans"`
	CreatedAt time.Time `json:"created_at"`
}

SpanBatch represents a batch of spans to be sent to backend.

type SpanKind

type SpanKind string

SpanKind indicates the type of span.

const (
	SpanKindTrace      SpanKind = "trace"      // Root span for a conversation
	SpanKindGeneration SpanKind = "generation" // LLM generation (assistant message)
	SpanKindSpan       SpanKind = "span"       // User message, tool call, etc.
)

type SyncResult

type SyncResult struct {
	FilesScanned int
	FilesUpdated int
	SpansSynced  int
	Errors       []error
	Duration     time.Duration
}

SyncResult holds the result of a sync operation.

type Syncer

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

Syncer coordinates the sync process.

func NewSyncer

func NewSyncer(opts SyncerOptions) (*Syncer, error)

NewSyncer creates a new syncer.

func (*Syncer) Close

func (s *Syncer) Close() error

Close releases syncer resources.

func (*Syncer) Stats

func (s *Syncer) Stats() (*syncdb.Stats, error)

Stats returns sync database statistics.

func (*Syncer) SyncAll

func (s *Syncer) SyncAll(ctx context.Context) (*SyncResult, error)

SyncAll syncs all conversation files.

func (*Syncer) SyncFile

func (s *Syncer) SyncFile(ctx context.Context, path string) (int, error)

SyncFile syncs a single file.

type SyncerOptions

type SyncerOptions struct {
	DBPath      string
	Backend     Backend
	ProjectsDir string
	Workers     int
	DryRun      bool
}

SyncerOptions configures the syncer.