Documentation
¶
Overview ¶
Package capture holds the shared streaming-reduction library consumed by both the local tapes proxy and the cloud tapes-extproc sidecar. A Reducer turns the raw bytes of a single LLM turn — streaming (SSE, NDJSON) or one-shot (application/json) — into a canonical *llm.ChatResponse suitable for handing to tapes-ingest.
Callers construct reducers explicitly via NewXxxReducer constructors and dispatch them by provider name themselves; the package deliberately holds no global registry so import-order and init() side effects stay out of the call graph.
Index ¶
Constants ¶
const MaxDecompressedBytes = 48 << 20
MaxDecompressedBytes caps one decompressed body, per layer. It bounds a decompression bomb: the compressed bytes are already capped on the way in, but a few KiB of zstd expands to gigabytes if nothing stops it.
48 MiB must stay ≥ extproc's requestCaptureBudget (MaxIngestBodyBytes − rawLaneEnvelopeReserve, ≈45.67 MiB). A request whose wire bytes fit but decode just past the budget must reach the over-budget shed and drop as request_over_budget; if this cap were lower, readCapped would error first and route the turn down the decode-error path instead. Read/decode side ≥ producer side is the ordering that keeps that shed reachable.
const ProviderAnthropic = "anthropic"
ProviderAnthropic is the canonical provider-name string consumers use to key the Anthropic reducer in their dispatch maps.
const ProviderOpenAI = "openai"
ProviderOpenAI is the canonical provider-name string consumers use to key the OpenAI Responses reducer in their dispatch maps.
Variables ¶
This section is empty.
Functions ¶
func IsPolicyDropReason ¶ added in v0.35.0
func IsPolicyDropReason(reason DropReason) bool
IsPolicyDropReason reports whether reason is one of the shared capture-policy reasons rather than a deployment's own transport or runtime reason.
Types ¶
type DecodeStats ¶ added in v0.32.0
type DecodeStats struct {
// Truncated marks a body recovered from a stream that ended early —
// the decode succeeded only because partial output was accepted. The
// caller decides what that is worth; it is surfaced rather than
// swallowed so a salvaged reduction is never silently indistinguishable
// from a clean one.
Truncated bool
}
DecodeStats reports what had to be tolerated to produce the decoded bytes.
func DecodeContentEncoding ¶ added in v0.32.0
func DecodeContentEncoding(body []byte, encoding string) ([]byte, DecodeStats, error)
DecodeContentEncoding returns body decoded per encoding, for handing to a reducer. Callers store the ENCODED bytes and decode only for reduction: re-compression is not byte-identical, so a column that promises "verbatim" has to keep what arrived.
An empty body under any non-identity coding is an error, uniformly across codings — see refuseEmpty. Callers must not reach here for a request that simply had no body.
An unrecognized encoding is an error rather than a pass-through. Handing compressed bytes to a reducer that expects text yields a parse failure well away from the actual cause, and the bytes are still stored either way — so erroring here loses nothing and names the real problem.
Truncated streams are salvaged rather than refused when they yielded any output at all, and the salvage is reported in DecodeStats. A capture that lost its tail is still most of a turn; refusing it would discard everything the stream did deliver in exchange for nothing. This mirrors extproc's response-side behavior — and it must, because extproc forwards the truncated compressed bytes as they arrived, so these are exactly the bytes ingest is handed.
type DropReason ¶ added in v0.35.0
type DropReason string
DropReason is one answer to "why was this turn not captured". Values are the wire-visible strings: metric label values and log fields, so they are part of the contract rather than an internal detail.
const ( // DropUpstreamStatus: the upstream did not return a success status. // A turn is a completed exchange with a provider; an error response is // a record of one failing to happen. Capturing them would put failed // requests in the same log as conversations and leave every consumer to // re-derive the difference. DropUpstreamStatus DropReason = "upstream_status" // DropNonTurnRequest: the request is not a turn. Adjacent endpoints on // the same host are not conversation (token counting, model listing), // and neither is a non-POST method on a turn path — a health probe // against the chat endpoint is still a probe. DropNonTurnRequest DropReason = "non_turn_request" // DropRequestDecode: the request body could not be decoded to the bytes // a reducer would parse. The decode policy itself is specified // separately, in fixtures/content-encoding/; this is the reason a turn // carries when that policy refuses it. DropRequestDecode DropReason = "request_decode" // DropEmptyResponse: the response phase completed with zero body bytes. // There is nothing to reduce, no encoding to undo, and no preview that // would tell a reader anything. Distinct from a reduction that came out // empty, which had bytes and lost them. DropEmptyResponse DropReason = "empty_response" // DropUnknownProvider: no reducer claims this provider and endpoint. // The set of shapes capture can read is a property of the build, and a // turn refused for this reason is a coverage gap rather than a defect // in the traffic. DropUnknownProvider DropReason = "unknown_provider" // DropResponseDecode: the response body could not be decoded. Same // policy and same separation as DropRequestDecode, one side over. DropResponseDecode DropReason = "response_decode" // DropReducerError: the bytes decoded but the reducer refused them. // The last policy gate: everything upstream of it said the turn was // capturable in principle, and the content said otherwise. DropReducerError DropReason = "reducer_error" )
The capture-policy reasons, in the order a turn meets them.
The order is part of the specification: a turn can satisfy several of these at once — a HEAD probe that returned 500 with no body satisfies three — and two implementations that report different reasons for the same turn have produced two different answers to the same question, even though both correctly declined to capture it.
func PolicyDropReasons ¶ added in v0.35.0
func PolicyDropReasons() []DropReason
PolicyDropReasons enumerates the reasons above, in precedence order.
It exists so a consumer can assert its own vocabulary against this one exhaustively rather than reason by reason: a policy reason added here and nowhere else, or added somewhere else and not here, is then a test failure instead of a discovery.
type Reducer ¶
type Reducer interface {
// Reduce consumes the raw request body and the raw response body and
// produces a canonical ChatResponse. The reqBody is supplied for context
// (the reducer may enrich the response with metadata the wire format
// doesn't carry); passing nil is valid when no request-side context is
// available. contentType is the upstream response's Content-Type, used to
// disambiguate streaming (text/event-stream, application/x-ndjson) from
// non-streaming (application/json) reduction paths.
//
// Errors are reserved for unrecoverable parse failures or malformed
// envelopes. Partial captures (mid-stream errors, EOF before terminal
// frame, per-block parse failures) should return a ChatResponse with
// diagnostic metadata in Extra rather than an error — callers want to
// see what happened instead of silence.
//
// Per-turn memory is not bounded inside the reducer. Content grows with
// the upstream's output, ceiling'd in practice by the caller's
// max_tokens. Callers that need a hard ceiling should impose one
// themselves; sidecar deployments track turn size via metrics rather
// than enforcing a cap so the full tape is preserved.
Reduce(ctx context.Context, reqBody, respBody io.Reader, contentType string) (*llm.ChatResponse, error)
}
Reducer produces a canonical *llm.ChatResponse from the raw request and response bodies for a single turn. Implementations must be stateless at the package level; per-turn state is held in local variables for the duration of a single Reduce call.
func NewAnthropicReducer ¶
func NewAnthropicReducer() Reducer
NewAnthropicReducer returns an Anthropic reducer. The value is stateless at the package level; per-turn state lives inside Reduce.
func NewOpenAIResponsesReducer ¶ added in v0.14.0
func NewOpenAIResponsesReducer() Reducer
NewOpenAIResponsesReducer returns an OpenAI Responses reducer. The value is stateless at the package level; per-turn state lives inside Reduce.
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
Package fixtures exposes the capture reducer test fixtures as an embed.FS so every consumer exercises the SAME bytes instead of hand-copied inline duplicates.
|
Package fixtures exposes the capture reducer test fixtures as an embed.FS so every consumer exercises the SAME bytes instead of hand-copied inline duplicates. |