capturer

package
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 7, 2026 License: MIT Imports: 33 Imported by: 0

Documentation

Overview

Package capturer observes the router's existing client and provider seams without adding a listener or changing request and response bytes.

Index

Constants

View Source 
const AXReadOutputEnvironment = "MEKUGI_AX_OUTPUT"

AXReadOutputEnvironment opts executor-side readers into a local journal.

Variables

This section is empty.

Functions

func InspectProviderUsage 

func InspectProviderUsage(ctx context.Context, path string, filter UsageInspectionFilter) (map[string]UsageInspection, error)

InspectProviderUsage reads sanitized capture JSONL, never executable history.

func ObserveCompaction 

func ObserveCompaction(ctx context.Context, answer string, summaryBytes, changes, failures int)

func ObserveJournal 

func ObserveJournal(ctx context.Context, metrics *JournalMetrics)

func ObserveProjectedRequest 

func ObserveProjectedRequest(ctx context.Context, body []byte)

ObserveProjectedRequest supplies the actual request after replay and tool projection. Measurement is capture-owned; no raw body is retained. It is observation only and cannot fail or change the routed operation.

func ObserveProviderUsage 

func ObserveProviderUsage(ctx context.Context, usage ProviderUsage)

ObserveProviderUsage supplies the provider-authoritative usage parsed by the router for the current provider attempt. Requests outside a Recorder handler are ignored.

func ObserveRequestKind 

func ObserveRequestKind(ctx context.Context, kind string)

ObserveRequestKind records only the router's validated, content-free request kind.

func ObserveResponsesWebSocketControl 

func ObserveResponsesWebSocketControl(ctx context.Context, boundary ResponsesWebSocketControlBoundary, direction ResponsesWebSocketControlDirection, payload []byte)

ObserveResponsesWebSocketControl records one actual WebSocket control payload at either transport boundary. Call it once for the request payload sent toward a boundary and once for the response payload read from it.

func PrepareAXReadJournal 

func PrepareAXReadJournal(path string) error

PrepareAXReadJournal enables a debug bundle before the executor starts.

func RequestCorrelation 

func RequestCorrelation(ctx context.Context) (captureID string, requestSequence uint64)

RequestCorrelation exposes only the immutable local exchange identity. No credentials or caller headers are returned, and no transport header is added.

func ValidAXIdentity 

func ValidAXIdentity(value string) bool

ValidAXIdentity accepts opaque identifiers, not arbitrary attributes or paths.

Types

type AXCommandAccumulator 

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

func (*AXCommandAccumulator) Observe 

func (a *AXCommandAccumulator) Observe(kind, id, logicalCallID string, at time.Time, exitCode *int) error

func (*AXCommandAccumulator) ObserveCompleted 

func (a *AXCommandAccumulator) ObserveCompleted(id, logicalCallID string, start, end time.Time, exitCode *int) error

ObserveCompleted accepts execution endpoints persisted together by the producer. Reuse a matching older start event rather than counting its embedded copy again.

func (*AXCommandAccumulator) Result 

func (a *AXCommandAccumulator) Result() AXCommandMetrics

type AXCommandMetrics 

type AXCommandMetrics struct {
	State          string                 `json:"state"`
	Started        uint64                 `json:"started"`
	Completed      uint64                 `json:"completed"`
	UnpairedEvents uint64                 `json:"unpaired_events"`
	DurationMS     int64                  `json:"duration_ms"`
	GapMS          int64                  `json:"gap_ms"`
	Commands       []AXCommandObservation `json:"commands"`
	Coverage       string                 `json:"coverage"`
}

type AXCommandObservation 

type AXCommandObservation struct {
	ItemID        string     `json:"item_id"`
	LogicalCallID string     `json:"logical_call_id,omitempty"`
	StartedAt     *time.Time `json:"started_at,omitempty"`
	CompletedAt   *time.Time `json:"completed_at,omitempty"`
	DurationMS    *int64     `json:"duration_ms,omitempty"`
	GapBeforeMS   *int64     `json:"gap_before_ms,omitempty"`
	ExitCode      *int       `json:"exit_code,omitempty"`
}

Command timings come from executor events, never source scanning or response duration. A gap includes all intervening work; it is not attributed overhead.

type AXCompletionAccumulator 

type AXCompletionAccumulator struct {
	Metrics AXCompletionMetrics
	// contains filtered or unexported fields
}

AXCompletionAccumulator pairs actual task/turn events by identity. It does not infer completion from silence, a tool exit, or an assistant's prose.

func (*AXCompletionAccumulator) Observe 

func (accumulator *AXCompletionAccumulator) Observe(kind, id string, at time.Time)

func (*AXCompletionAccumulator) Result 

func (accumulator *AXCompletionAccumulator) Result() AXCompletionMetrics

type AXCompletionMetrics 

type AXCompletionMetrics struct {
	State          string `json:"state"`
	CompletedTurns uint64 `json:"completed_turns"`
	DurationMS     int64  `json:"duration_ms"`
	UnpairedEvents uint64 `json:"unpaired_events"`
}

type AXDefectAssessment 

type AXDefectAssessment struct {
	CallID   string `json:"call_id"`
	Verdict  string `json:"verdict"`
	Evidence string `json:"evidence"`
	SHA256   string `json:"sha256"`
}

type AXDefectMetrics 

type AXDefectMetrics struct {
	AssessedCalls uint64               `json:"assessed_calls"`
	Reported      uint64               `json:"reported_defects"`
	Unassessed    uint64               `json:"unassessed_calls"`
	Assessments   []AXDefectAssessment `json:"assessments"`
}

func ReadAXDefects 

func ReadAXDefects(path string, editCalls map[string]bool) (AXDefectMetrics, error)

ReadAXDefects requires a real evidence artifact for every explicit verdict. It records provenance, not a claim that a test failure proves edit causality.

type AXEditAccumulator 

type AXEditAccumulator struct {
	Metrics AXEditMetrics
	// contains filtered or unexported fields
}

AXEditAccumulator holds only the preceding original payload for exact, line-aligned repetition measurement, never a durable script history.

func (*AXEditAccumulator) Observe 

func (accumulator *AXEditAccumulator) Observe(script string, failed, unconfirmed bool)

type AXEditMetrics 

type AXEditMetrics struct {
	Calls              uint64 `json:"calls"`
	EmittedBytes       uint64 `json:"emitted_bytes"`
	ReEmittedLineBytes uint64 `json:"re_emitted_line_bytes"`
	Failed             uint64 `json:"failed"`
	Unconfirmed        uint64 `json:"unconfirmed"`
}

AXEditMetrics reports measured emissions separately from defect judgments.

type AXReadContext 

type AXReadContext struct {
	CallID  string `json:"call_id,omitempty"`
	ShellID string `json:"shell_id,omitempty"`
}

AXReadContext contains identities only, never command text or paths.

type AXReadFailure 

type AXReadFailure struct {
	AXReadContext
	ID         string    `json:"id"`
	Tool       string    `json:"tool"`
	At         time.Time `json:"at"`
	DurationNS int64     `json:"duration_ns"`
	Class      string    `json:"class"`
	ExitCode   *int      `json:"exit_code,omitempty"`
}

AXReadFailure joins a bounded failure sample back to the journal and rollout.

type AXReadJournal 

type AXReadJournal struct {
	Threads map[string]AXReadMetrics
}

AXReadJournal validates the complete journal before exposing any attribution. An empty thread key represents explicitly unattributed runtime evidence.

func ReadAXReadJournal 

func ReadAXReadJournal(ctx context.Context, path string) (AXReadJournal, error)

func (AXReadJournal) ForThread 

func (journal AXReadJournal) ForThread(threadID string) AXReadMetrics

type AXReadMetrics 

type AXReadMetrics struct {
	State               string            `json:"state"`
	Started             uint64            `json:"started"`
	Completed           uint64            `json:"completed"`
	Succeeded           uint64            `json:"succeeded"`
	Failed              uint64            `json:"failed"`
	Incomplete          uint64            `json:"incomplete"`
	DurationNS          uint64            `json:"duration_ns"`
	ByTool              map[string]uint64 `json:"by_tool"`
	FailuresByClass     map[string]uint64 `json:"failures_by_class"`
	Failures            []AXReadFailure   `json:"failures"`
	DroppedFailures     uint64            `json:"dropped_failure_details"`
	OtherThreadStarted  uint64            `json:"other_thread_started"`
	UnattributedStarted uint64            `json:"unattributed_started"`
	Coverage            string            `json:"coverage"`
}

AXReadMetrics counts only journal-observed invocations, not external reads.

func ReadAXReads 

func ReadAXReads(ctx context.Context, path, threadID string) (AXReadMetrics, error)

ReadAXReads retains the one-thread API, but no longer silently hides exclusions.

type AXReadObservation 

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

AXReadObservation records one actual invocation, not a parsed source command. It retains no command, file path, source text or output in its journal.

func StartAXReadWithContext 

func StartAXReadWithContext(path, threadID, tool string, identity AXReadContext) (*AXReadObservation, error)

StartAXReadWithContext is auxiliary to execution. Callers report failures separately and keep the original command outcome. Each event is one O_APPEND write.

func (*AXReadObservation) FinishResult 

func (observation *AXReadObservation) FinishResult(succeeded bool, failureClass string, exitCode *int) error

FinishResult closes the observation even if classification or writing fails. Only an allowlisted class and a process exit status may enter the journal.

type CompactionMetrics 

type CompactionMetrics struct {
	CompactionAnswer       string  `json:"compaction_answer,omitempty"`
	CompactionSummaryBytes *uint64 `json:"compaction_summary_bytes,omitempty"`
	CompactionChanges      *uint64 `json:"compaction_changes,omitempty"`
	CompactionFailures     *uint64 `json:"compaction_failures,omitempty"`
}

CompactionMetrics contains only routing provenance and content-free counts. Router answers have no provider attempt; zero usage is not a saving estimate.

type Config 

type Config struct {
	Writes *persistence.Counter
	Output string
	Mode   string
}

Config identifies the observed router behavior and optional durable JSONL destination. An empty Output retains records only for the process lifetime.

type JournalMetrics 

type JournalMetrics struct {
	StartedAt          string            `json:"started_at"`
	Sequence           uint64            `json:"sequence"`
	Operations         map[string]uint64 `json:"operations"`
	StandaloneRequests uint64            `json:"standalone_requests"`
	FinalAnswers       uint64            `json:"final_answers"`
	FinalAnswerBytes   uint64            `json:"final_answer_bytes"`
	EmptyOutcomes      uint64            `json:"empty_outcomes"`
	LastOutcomeEmpty   *bool             `json:"last_outcome_empty,omitempty"`
}

JournalMetrics is a cumulative, content-free snapshot for the owning thread. Consumers select its newest sequence, never sum snapshots across exchanges.

func (*JournalMetrics) Clone 

func (j *JournalMetrics) Clone() *JournalMetrics

type MetricsSnapshot 

type MetricsSnapshot struct {
	StorageWrites  *persistence.Snapshot    `json:"storage_writes,omitempty"`
	Schema         string                   `json:"schema"`
	Mode           string                   `json:"mode"`
	Requests       requestTotals            `json:"requests"`
	Usage          usageMetrics             `json:"usage"`
	Cache          cacheMetrics             `json:"cache"`
	Transport      transportMetrics         `json:"transport"`
	Semantic       semanticOutputMetrics    `json:"semantic"`
	ProviderTools  map[string]toolAggregate `json:"provider_tools"`
	DeliveredTools map[string]toolAggregate `json:"delivered_tools"`
	Exchanges      []exchangeMetrics        `json:"exchanges"`
	Capture        captureHealth            `json:"capture"`
}

MetricsSnapshot is a detached, capture-owned metrics view.

type OutputThroughput 

type OutputThroughput struct {
	OutputTokens     uint64 `json:"output_tokens"`
	DurationNanos    uint64 `json:"duration_ns"`
	MeasuredRequests uint64 `json:"measured_requests"`
}

OutputThroughput pairs authoritative output tokens with the elapsed provider request time for exactly the same measured requests. Zero duration is absent timing, not zero throughput. It excludes unmeasured requests from both sums.

func (*OutputThroughput) Add 

func (t *OutputThroughput) Add(next OutputThroughput)

func (OutputThroughput) Rate 

func (t OutputThroughput) Rate() (float64, bool)

type ProviderUsage 

type ProviderUsage struct {
	// Nil means legacy normalization did not retain field-completeness evidence.
	EvidenceComplete *bool            `json:"evidence_complete,omitempty"`
	InputTokens      uint64           `json:"input_tokens"`
	CachedTokens     uint64           `json:"cached_input_tokens"`
	OutputTokens     uint64           `json:"output_tokens"`
	ReasoningTokens  uint64           `json:"reasoning_tokens"`
	OutputThroughput OutputThroughput `json:"output_throughput,omitzero"`
}

ProviderUsage is the provider-authoritative token count parsed from one terminal Responses payload.

type Recorder 

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

Recorder owns correlation, sanitized measurement, durable capture, and derived metrics for one router process.

func New 

func New(config Config) (*Recorder, error)

New creates one in-process recorder. It never starts a server.

func (*Recorder) Close 

func (r *Recorder) Close() error

Close flushes the optional durable destination.

func (*Recorder) Handler 

func (r *Recorder) Handler(next http.Handler) http.Handler

Handler observes the existing Responses listener. POST requests are measured directly. GET requests receive a private WebSocket capture factory in their context, but the upgrade itself does not create a logical request.

func (*Recorder) ServeHTTP 

func (r *Recorder) ServeHTTP(writer http.ResponseWriter, _ *http.Request)

ServeHTTP exposes capture-owned calculations on the router's existing listener. It never reads or mutates router state.

func (*Recorder) Snapshot 

func (r *Recorder) Snapshot() MetricsSnapshot

Snapshot returns an independent view safe for presentation without holding recorder locks.

func (*Recorder) Transport 

func (r *Recorder) Transport(next http.RoundTripper) http.RoundTripper

Transport observes provider attempts made by the router's existing HTTP client. It preserves retry, cancellation, and response-body ownership.

func (*Recorder) WriteMetrics 

func (r *Recorder) WriteMetrics(writer io.Writer) error

WriteMetrics exports the same capture-owned snapshot used by the session dialog.

type ResponsesWebSocket 

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

ResponsesWebSocket observes one logical response delivered over the Codex-facing WebSocket. Its messages are the restored JSON payloads delivered to Codex, without WebSocket framing or provider-only representation.

func BeginResponsesWebSocket 

func BeginResponsesWebSocket(ctx context.Context, headers http.Header, payload []byte) (context.Context, *ResponsesWebSocket)

BeginResponsesWebSocket starts one logical Codex-facing exchange when ctx came from Recorder.Handler's GET /v1/responses handler. A nil payload is a valid automatic successor and is measured as zero request bytes.

func (*ResponsesWebSocket) Finish 

func (exchange *ResponsesWebSocket) Finish(err error)

Finish records the logical exchange after its terminal downstream message and correlated provider attempts have been observed.

func (*ResponsesWebSocket) Message 

func (exchange *ResponsesWebSocket) Message(payload []byte)

Message measures one restored JSON message delivered to Codex.

type ResponsesWebSocketControlBoundary 

type ResponsesWebSocketControlBoundary string
const (
	// ResponsesWebSocketControlCodex is the client-facing transport boundary.
	ResponsesWebSocketControlCodex ResponsesWebSocketControlBoundary = "codex"
	// ResponsesWebSocketControlProvider is the provider-facing transport boundary.
	ResponsesWebSocketControlProvider ResponsesWebSocketControlBoundary = "provider"
)

type ResponsesWebSocketControlDirection 

type ResponsesWebSocketControlDirection string
const (
	// ResponsesWebSocketControlRequest identifies a control message sent
	// toward a transport boundary.
	ResponsesWebSocketControlRequest ResponsesWebSocketControlDirection = "request"
	// ResponsesWebSocketControlResponse identifies a control message returned
	// from a transport boundary.
	ResponsesWebSocketControlResponse ResponsesWebSocketControlDirection = "response"
)

type UsageInspection 

type UsageInspection struct {
	UnknownCompletenessRecords uint64         `json:"unknown_completeness_records"`
	State                      string         `json:"state"`
	ObservedRecords            uint64         `json:"observed_records"`
	MissingRecords             uint64         `json:"missing_records"`
	Tokens                     *ProviderUsage `json:"tokens,omitempty"`
}

UsageInspection keeps provider-consumption evidence separate from local byte measurements. Missing/incomplete records never count as zero-token successes.

type UsageInspectionFilter 

type UsageInspectionFilter struct {
	Since, Until        time.Time
	Model, ExcludeModel string
	Threads             map[string]bool
}

type WebSocketAttempt 

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

WebSocketAttempt observes one response.create exchange on a possibly reused connection. Message records contain actual JSON payload bytes, not WebSocket frame headers, control frames, or the router's downstream SSE serialization. Like Transport, this is a transport-boundary observer, not a metrics callback. Its owner calls Finish after terminal usage observation and body consumption.

func BeginWebSocketAttempt 

func BeginWebSocketAttempt(ctx context.Context, payload []byte, headers, responseHeaders http.Header) *WebSocketAttempt

BeginWebSocketAttempt is a no-op outside a Recorder handler. Headers describe this exchange's routing, not a prior exchange's connection handshake.

func (*WebSocketAttempt) Finish 

func (attempt *WebSocketAttempt) Finish(err error)

func (*WebSocketAttempt) HTTPErrorBody 

func (attempt *WebSocketAttempt) HTTPErrorBody(body io.ReadCloser, status int, contentType, contentEncoding string) io.ReadCloser

HTTPErrorBody observes a rejected upgrade's actual HTTP body. No inference payload was sent, so its request measurement is empty rather than fabricated.

func (*WebSocketAttempt) Message 

func (attempt *WebSocketAttempt) Message(payload []byte)

Message measures the bytes read from one provider text message. Observation remains bounded even when the complete exchange exceeds the parsing budget.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.