obs

type ActionType ¶

type ActionType = string

ActionType represents the type of action performed

const (
	ActionAddProvider    ActionType = "add_provider"
	ActionDeleteProvider ActionType = "delete_provider"
	ActionUpdateProvider ActionType = "update_provider"
	ActionStartServer    ActionType = "start_server"
	ActionStopServer     ActionType = "stop_server"
	ActionRestartServer  ActionType = "restart_server"
	ActionGenerateToken  ActionType = "generate_token"
	ActionUpdateDefaults ActionType = "update_defaults"
	ActionFetchModels    ActionType = "fetch_models"
)

type BatchProcessor ¶ added in v0.260514.1

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

BatchProcessor batches Records and forwards them to a RecordExporter on a periodic ticker or when the batch reaches maxBatch size.

Emit is non-blocking: if the internal queue is full the record is dropped and a counter is incremented.

func NewBatchProcessor(exporter RecordExporter, opts BatchProcessorOptions) *BatchProcessor

func (bp *BatchProcessor) Emit(r *Record)

func (bp *BatchProcessor) ForceFlush(ctx context.Context) error

func (bp *BatchProcessor) Shutdown(ctx context.Context) error

type BatchProcessorOptions ¶ added in v0.260514.1

type BatchProcessorOptions struct {
	// QueueSize is the capacity of the internal channel (default 1024).
	QueueSize int
	// MaxBatch is the maximum number of records per Export call (default 256).
	MaxBatch int
	// FlushInterval is how often the worker drains the queue (default 5s).
	FlushInterval time.Duration
}

BatchProcessorOptions configures a BatchProcessor.

type CASFileExporter ¶ added in v0.260514.1

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

CASFileExporter implements RecordExporter with content-addressed storage:

1. Slim-ifies every Record (large sub-values replaced by $ref pointers).
2. Writes new blobs atomically (tmp+rename) under {baseDir}/blobs/.
3. Appends SlimRecords to per-session JSONL files.

CASFileExporter is NOT goroutine-safe; it must be driven by a single goroutine (the BatchProcessor worker).

func NewCASFileExporter(baseDir string) (*CASFileExporter, error)

func (e *CASFileExporter) Export(_ context.Context, records []*Record) error

func (e *CASFileExporter) Shutdown(_ context.Context) error

type GzipFileExporter ¶ added in v0.260514.1

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

GzipFileExporter implements RecordExporter by appending one gzip member per batch to a per-session .jsonl.gz file. Records are emitted with all bodies inlined (no $ref / blob extraction).

On-disk layout:

{baseDir}/{scenario}/sessions/{YYYY-MM-DD}/{session}.jsonl.gz

Multiple gzip members concatenated in the same file remain a valid gzip stream; standard tools (zcat, gzip -d, gzip.Reader with Multistream(true)) decompress them transparently.

func NewGzipFileExporter(baseDir string) *GzipFileExporter

func (e *GzipFileExporter) Export(_ context.Context, records []*Record) error

func (e *GzipFileExporter) Shutdown(_ context.Context) error

type MultiExporter ¶ added in v0.260514.1

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

MultiExporter fans Export and Shutdown calls out to multiple RecordExporters. One exporter failing does not prevent the others from running; the first non-nil error is returned.

func NewMultiExporter(exporters ...RecordExporter) *MultiExporter

func (m *MultiExporter) Export(ctx context.Context, records []*Record) error

func (m *MultiExporter) Shutdown(ctx context.Context) error

type Record ¶ added in v0.260514.1

type Record struct {
	Timestamp  time.Time
	RequestID  string
	SessionID  string // sha256(raw session value)[:16], empty when unknown
	SessionSrc string // "user" | "hdr" | "ip" | ""
	Provider   string
	Scenario   string
	Model      string

	// Provider connection details for debugging
	APIStyle string `json:"api_style,omitempty"` // Provider API style (e.g., "openai", "anthropic")
	BaseURL  string `json:"base_url,omitempty"`  // Provider base URL

	OriginalRequest    *RecordRequest
	TransformedRequest *RecordRequest
	ProviderResponse   *RecordResponse
	FinalResponse      *RecordResponse

	Duration time.Duration
	Err      string
	Steps    []string
}

Record is the canonical data model for one LLM request/response cycle. Construct it on the hot path and pass to Sink.Emit; the only cost is a non-blocking channel send.

type RecordExporter ¶ added in v0.260514.1

type RecordExporter interface {
	Export(ctx context.Context, records []*Record) error
	Shutdown(ctx context.Context) error
}

RecordExporter is called by BatchProcessor with a slice of Records to persist.

type RecordMode ¶

type RecordMode string

RecordMode defines which fields are captured by the Sink.

const (
	RecordModeRequestOnly           RecordMode = "request"                 // Record transformed request only
	RecordModeRequestResponse       RecordMode = "request_response"        // Record transformed request + final response
	RecordModeStagedRequestResponse RecordMode = "staged_request_response" // Record original request + transformed request + final response
)

type RecordProcessor ¶ added in v0.260514.1

type RecordProcessor interface {
	Emit(r *Record)
	ForceFlush(ctx context.Context) error
	Shutdown(ctx context.Context) error
}

RecordProcessor is the OTel-shaped hot-path interface. Emit must be non-blocking; callers on the request goroutine must not be delayed.

type RecordRequest ¶

type RecordRequest struct {
	Method  string                 `json:"method"`
	URL     string                 `json:"url"`
	Headers map[string]string      `json:"headers"`
	Body    map[string]interface{} `json:"body,omitempty"`
}

RecordRequest represents the HTTP request details. Kept for callers that build RecordRequest directly (e.g. client/record_roundtripper.go).

type RecordResponse ¶

type RecordResponse struct {
	StatusCode   int                    `json:"status_code"`
	Headers      map[string]string      `json:"headers"`
	Body         map[string]interface{} `json:"body,omitempty"`
	IsStreaming  bool                   `json:"is_streaming,omitempty"`
	StreamChunks []string               `json:"-"`
}

RecordResponse represents the HTTP response details.

type ResolveOpts ¶ added in v0.260514.1

type ResolveOpts struct {
	// AllowMissingBlobs replaces missing blob refs with {"_missing_blob":"sha256:..."}
	// instead of returning an error.
	AllowMissingBlobs bool
}

ResolveOpts controls how missing blobs are handled during resolution.

type SessionIterator ¶ added in v0.260514.1

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

SessionIterator iterates over SlimRecords in a session JSONL file.

func WalkSession(baseDir, scenario, sessionShort string) (*SessionIterator, error)

func (it *SessionIterator) Next() (*SlimRecord, error)

type Sink ¶

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

Sink manages recording of LLM request/response cycles. All writes are batched asynchronously; Emit is non-blocking.

func NewSink(baseDir string, mode RecordMode, opts ...SinkOption) *Sink

func (s *Sink) Close()

func (s *Sink) Emit(r *Record)

func (s *Sink) ForceFlush(ctx context.Context) error

func (s *Sink) GetBaseDir() string

func (s *Sink) GetMode() RecordMode

func (s *Sink) IsEnabled() bool

func (s *Sink) RecordWithScenario(provider, model, scenario string, req *RecordRequest, resp *RecordResponse, duration time.Duration, err error)

type SinkOption ¶ added in v0.260514.1

type SinkOption func(*sinkConfig)

SinkOption customises Sink construction. Apply via NewSink(baseDir, mode, opts...).

func WithCASExporter() SinkOption

func WithExporters(exporters ...RecordExporter) SinkOption

type SlimHTTPData ¶ added in v0.260514.1

type SlimHTTPData struct {
	Method      string            `json:"method,omitempty"`
	URL         string            `json:"url,omitempty"`
	Headers     map[string]string `json:"headers,omitempty"`
	StatusCode  int               `json:"status_code,omitempty"`
	Body        interface{}       `json:"body,omitempty"`
	IsStreaming bool              `json:"is_streaming,omitempty"`
}

SlimHTTPData mirrors RecordRequest / RecordResponse with a body that may contain {"$ref":"sha256:<hex>"} markers instead of large inline values.

type SlimRecord ¶ added in v0.260514.1

type SlimRecord struct {
	V         int    `json:"v"` // schema version = 3
	Timestamp string `json:"ts"`
	RequestID string `json:"rid"`

	SessionID  string `json:"sid,omitempty"`
	SessionSrc string `json:"sid_src,omitempty"`

	Provider string `json:"provider,omitempty"`
	Scenario string `json:"scenario,omitempty"`
	Model    string `json:"model,omitempty"`

	OriginalRequest    *SlimHTTPData `json:"original_request,omitempty"`
	TransformedRequest *SlimHTTPData `json:"transformed_request,omitempty"`
	ProviderResponse   *SlimHTTPData `json:"provider_response,omitempty"`
	FinalResponse      *SlimHTTPData `json:"final_response,omitempty"`

	DurationMs int64  `json:"duration_ms"`
	Error      string `json:"error,omitempty"`

	Steps []string `json:"transform_steps,omitempty"`
}

SlimRecord is the JSON-serializable slim form stored in session JSONL files. Large values are replaced by {"$ref":"sha256:<hex>"} pointers into the blob store.

func FullRecord(r *Record) *SlimRecord

func ResolveRecord(baseDir string, slim *SlimRecord, opts ResolveOpts) (*SlimRecord, error)

func SlimifyRecord(r *Record, knownBlobs map[string]struct{}) (*SlimRecord, map[string][]byte)