Documentation
¶
Overview ¶
Package ingest provides an HTTP server that accepts completed LLM conversation turns and appends them to the immutable raw-turn capture log. This enables "sidecar mode" where an external gateway (e.g., Envoy AI Gateway) handles upstream LLM traffic and tapes only captures the turns for the deriver. Embeddings are written downstream by the derive worker family (pkg/spanembed), never at ingest time.
Index ¶
- Constants
- Variables
- func CompileOpenAPI(ctx context.Context, docs oas.TypeDocs) (*oas.CompiledDoc, error)
- func NewOpenAPIParser(docs oas.TypeDocs) *oas.Parser
- type Config
- type Metrics
- func (m *Metrics) Handler() http.Handler
- func (m *Metrics) ObserveDAGLatency(provider string, seconds float64)
- func (m *Metrics) ObserveRawOnlyStamp(provider string, field StampField, source StampSource)
- func (m *Metrics) ObserveWrite(provider string, result Result, bodyBytes int)
- func (m *Metrics) Registry() *prometheus.Registry
- func (m *Metrics) SetQueueDepth(depth int)
- type Result
- type Server
- type StampField
- type StampSource
- type TranscriptPayload
- type TurnMeta
- type TurnPayload
Constants ¶
const ( // HeaderPaperAuthOrgID carries the verified org claim. HeaderPaperAuthOrgID = "x-paper-auth-org-id" // HeaderPaperAuthSubject carries the verified `sub` claim. HeaderPaperAuthSubject = "x-paper-auth-subject" )
Server-trusted identity headers, populated by the upstream gateway from validated JWT claims. This is the same contract the wire-capture path consumes (the extproc/headers package reads the identical names into the session envelope): clients are not permitted to send these themselves, and the gateway is responsible for stripping inbound values so only edge-verified identity reaches the handler. When the gateway is not configured to populate them, the headers are absent and the payload envelope's own identity fields stand.
const MaxIngestBodyBytes = MaxRawResponseBytes*4/3 + 4<<20
MaxIngestBodyBytes is the request body limit, derived from MaxRawResponseBytes rather than chosen independently: raw_response travels base64-encoded (4/3 expansion), alongside the raw request, the reduced response, and the meta block. Fiber's 4 MiB default would reject a body carrying a raw response well under the cap, which would make the cap unreachable and its drop-and-mark path dead code — the limit that actually bit would be an unrelated framework default, and turns would fail at the transport with no fidelity marker recorded anywhere.
const MaxRawResponseBytes = 8 << 20
MaxRawResponseBytes caps the verbatim response bytes ingest will store on a single turn. Beyond it the bytes are dropped and the row is marked (raw_response_dropped), rather than the write being refused: the reduced response, the raw request, and the session attribution are all still worth keeping, and a turn that vanishes entirely is a worse outcome than one whose verbatim bytes are known to be missing.
8 MiB is well above a normal turn — it is a backstop against a pathological response, not a working limit. Reduction happens before the cap is applied, so an oversize turn still lands with a usable reduced response.
Variables ¶
var ( // ErrEnvelope means the POST body could not be decoded as a TurnPayload. // Returned as 400 Bad Request. ErrEnvelope = errors.New("invalid envelope") // ErrUnprocessable covers validation / parse failures inside a well-formed // envelope: unknown provider, unparseable provider-specific request / // response body, etc. Returned as 422 Unprocessable Entity. ErrUnprocessable = errors.New("unprocessable turn") // ErrDownstream covers failures that originate below the handler: worker // pool saturation, DAG write errors, storage unavailability. Returned as // 502 Bad Gateway. ErrDownstream = errors.New("downstream failure") )
Ingest error classes. Each maps to a distinct HTTP status so operators can tell malformed envelopes from unknown providers from downstream outages without tailing logs.
Functions ¶
func CompileOpenAPI ¶ added in v0.30.0
CompileOpenAPI builds the ingest write surface's published contract.
Like the read API's, it constructs a server purely to make it register its routes, then compiles what that registration produced. The server is never started; it is given no driver and no worker pool it will use.
func NewOpenAPIParser ¶ added in v0.30.0
NewOpenAPIParser returns a parser configured for the ingest contract.
docs is non-nil only under `tapes dev openapi --docs-root`, which supplies the repository's doc comments so the compiled document carries the prose that documents each envelope field in the source. A deployed binary has no source tree, so what it serves describes every field's shape but not its meaning.
Types ¶
type Config ¶
type Config struct {
// ListenAddr is the address to listen on (e.g., ":8082")
ListenAddr string
// Project is the git repository or project name to tag on captured turns.
Project string
}
Config is the ingest server configuration.
type Metrics ¶ added in v0.5.2
type Metrics struct {
// contains filtered or unexported fields
}
Metrics enumerates the Prometheus counters and histograms emitted by the ingest server. Metric names are fixed so dashboards and alerts reference stable identifiers.
func NewMetrics ¶ added in v0.5.2
func NewMetrics() *Metrics
NewMetrics builds a fresh registry and registers the ingest metric set on it. Each Server owns its own registry so tests don't leak counters across suite runs (the default prometheus registry is global state).
func (*Metrics) Handler ¶ added in v0.5.2
Handler returns an http.Handler that serves the Prometheus scrape endpoint backed by this Metrics' registry.
func (*Metrics) ObserveDAGLatency ¶ added in v0.5.2
ObserveDAGLatency records how long it took to enqueue a turn into the worker pool. Latency is a cheap proxy for back-pressure so we graph it even though enqueue is nominally O(1) — a slow enqueue hints at queue saturation.
func (*Metrics) ObserveRawOnlyStamp ¶ added in v0.31.0
func (m *Metrics) ObserveRawOnlyStamp(provider string, field StampField, source StampSource)
ObserveRawOnlyStamp records one capture-side field restored (or not) on a server-side raw-only reduction.
func (*Metrics) ObserveWrite ¶ added in v0.5.2
ObserveWrite increments the writes counter for a given provider/result. A zero-length provider label becomes "unknown" so scrapes don't drop rows.
func (*Metrics) Registry ¶ added in v0.5.2
func (m *Metrics) Registry() *prometheus.Registry
Registry exposes the backing *prometheus.Registry so callers can mount a scrape handler or assert on the metric state in tests.
func (*Metrics) SetQueueDepth ¶ added in v0.5.2
SetQueueDepth updates the worker queue depth gauge.
type Result ¶ added in v0.5.2
type Result string
Result enumerates the status-label values emitted on the writes counter. Closed enumeration keeps dashboards safe against label typos.
const ( ResultAccepted Result = "accepted" ResultRejectEnv Result = "reject_envelope" ResultRejectParse Result = "reject_parse" ResultUnknownProv Result = "unknown_provider" ResultQueueFull Result = "queue_full" ResultDownstreamErr Result = "downstream_error" // ResultInternalErr covers a failure inside the handler itself (e.g. a // server-side marshal that should never fail) — a 500, distinct from a bad // payload or a downstream outage — so no handler exit is invisible to the // writes counter. ResultInternalErr Result = "internal_error" )
type Server ¶
type Server struct {
// contains filtered or unexported fields
}
Server is an HTTP server that accepts completed LLM conversation turns for async capture to the raw_turns log.
func (*Server) Close ¶
Close gracefully shuts down the server and waits for the worker pool to drain.
func (*Server) Metrics ¶ added in v0.5.2
Metrics exposes the ingest metrics so tests and health checks can scrape the registry programmatically.
func (*Server) OpenAPIParser ¶ added in v0.30.0
OpenAPIParser returns the live parser this server's routes registered into.
type StampField ¶ added in v0.31.0
type StampField string
StampField enumerates the capture-side fields ingest restores onto a server-side raw-only reduction. Closed enumeration keeps dashboards safe against label typos.
const ( StampFieldDuration StampField = "duration" StampFieldCreatedAt StampField = "created_at" )
type StampSource ¶ added in v0.31.0
type StampSource string
StampSource names the meta field that supplied a stamped value, or the fallback taken when none did. Naming the field rather than just "envelope" is what lets a scrape answer which producers have been upgraded, per field, without correlating against deploy history.
StampSourceFallback means the envelope carried nothing usable. What that leaves behind is field-specific: an unstamped duration for StampFieldDuration, and whatever the reducer produced — ingest's own clock for providers whose wire format carries no timestamp — for StampFieldCreatedAt.
The fallback bucket is the point of this metric. Both fields degrade silently otherwise: a NULL duration and an ingest-time CreatedAt are indistinguishable downstream from a turn that genuinely had them.
const ( // StampSourceElapsed is meta.elapsed_seconds, for the duration. StampSourceElapsed StampSource = "elapsed_seconds" // StampSourceCapturedAt is meta.captured_at — the turn's completion // instant, exactly what CreatedAt denotes. StampSourceCapturedAt StampSource = "captured_at" // StampSourceTsRequest is meta.ts_request, the turn's request instant, // offset by elapsed_seconds when the envelope carries one. StampSourceTsRequest StampSource = "ts_request" // StampSourceFallback means no capture-side source was available. StampSourceFallback StampSource = "fallback" )
type TranscriptPayload ¶ added in v0.16.0
type TranscriptPayload struct {
// Session identifies the harness session the transcript belongs to.
Session *sessions.IngestEnvelope `json:"session"`
// AgentID is empty for the main transcript, or the subagent id for
// subagents/agent-<id>.jsonl files.
AgentID string `json:"agent_id,omitempty"`
// AgentType / Description / ToolUseID mirror the harness's
// subagent meta.json: ToolUseID is the Task tool_use that forked
// this agent — the causal fork edge the deriver attaches.
AgentType string `json:"agent_type,omitempty"`
Description string `json:"description,omitempty"`
ToolUseID string `json:"tool_use_id,omitempty"`
// Kind qualifies Codex sub_agent_activity anchor rows:
// "interacted" marks a re-entry record (send_message /
// followup_task, targeting AgentID with ToolUseID = the triggering
// call), banked for future rendering and ignored by derivation.
// Absent/empty means spawn evidence — the legacy default.
Kind string `json:"kind,omitempty"`
// Records is the transcript's JSONL content as a JSON array,
// verbatim.
Records json.RawMessage `json:"records"`
}
TranscriptPayload is the ingest body for one harness transcript file — the main session transcript or one subagent's. The records land in the immutable raw layer verbatim (source: transcript); the deriver reconciles them against the wire capture to recover the causal/fork skeleton. No node-path processing happens at ingest time.
type TurnMeta ¶ added in v0.16.0
type TurnMeta struct {
RequestID string `json:"request_id,omitempty"`
ContentType string `json:"content_type,omitempty"`
// ThreadID is the harness sub-thread id resolved by the capture
// adapter (extproc headers.ThreadID); "" for main-thread calls.
ThreadID string `json:"thread_id,omitempty"`
Method string `json:"method,omitempty"`
Path string `json:"path,omitempty"`
Endpoint string `json:"endpoint,omitempty"`
Model string `json:"model,omitempty"`
ModelFamily string `json:"model_family,omitempty"`
Stream string `json:"stream,omitempty"`
ContentEncoding string `json:"content_encoding,omitempty"`
UpstreamStatus int `json:"upstream_status,omitempty"`
UpstreamStatusClass string `json:"upstream_status_class,omitempty"`
RequestBytes int `json:"request_bytes,omitempty"`
ResponseBytes int `json:"response_bytes,omitempty"`
ElapsedSeconds float64 `json:"elapsed_seconds,omitempty"`
// TsRequest is the capture-side instant the turn's request went
// upstream, RFC 3339. It is not new: pkg/backfill emits it and
// derive.CapturedAt already reads it as a row's original capture
// time, which is where every derived span's StartedAt comes from.
// Ingest parses it here so a server-side reduction can date itself
// from the same source the deriver uses, rather than a second,
// silently different clock.
TsRequest string `json:"ts_request,omitempty"`
// CapturedAt is the capture-side instant the turn COMPLETED
// upstream, RFC 3339 — the quantity CreatedAt actually means, and
// the one a producer that reduces live records as time.Now().
//
// Distinct from TsRequest by exactly the call's duration. Preferred
// over it because it needs no arithmetic to be exact; optional, and
// no released producer sends it yet. See stampCaptureTime for the
// precedence and what happens when neither field is present.
CapturedAt string `json:"captured_at,omitempty"`
}
TurnMeta mirrors the capture adapter's meta block (tapes-extproc TurnMeta). Every field is optional; adapters that predate a field simply omit it. Ingest only reads RequestID directly (raw-turn dedup) — the rest ride along verbatim in the raw layer and become queryable post-derive.
type TurnPayload ¶
type TurnPayload struct {
// Provider type: "openai", "anthropic", "ollama"
Provider string `json:"provider"`
// AgentName optionally tags the turn (same as X-Tapes-Agent-Name header)
AgentName string `json:"agent_name,omitempty"`
// RawRequest is the original request body sent to the LLM provider.
RawRequest json.RawMessage `json:"request"`
// Response is the already reduced, provider-agnostic response for the turn.
Response llm.ChatResponse `json:"response"`
// RawResponse is the upstream response body exactly as it arrived on the
// wire, base64-encoded in the JSON envelope. Optional and independent of
// Response: an adapter may send both (reduction plus the bytes it reduced
// from), only the reduction (the historical shape), or only the bytes.
//
// Raw-only is the interesting case. Reduction is lossy and adapter-
// specific, so an adapter that ships only the bytes lets ingest perform
// the reduction with the shared pkg/capture reducers — which is what makes
// two capture paths produce identical rows for identical upstream traffic
// instead of two subtly different ones.
RawResponse []byte `json:"raw_response,omitempty"`
// RawResponseEncoding is the Content-Encoding of RawResponse ("identity",
// "gzip", …). Empty means identity. The bytes are stored under this
// encoding rather than decompressed, so the stored column stays literally
// what the upstream sent.
RawResponseEncoding string `json:"raw_response_encoding,omitempty"`
// RawResponseWithheld says the producer captured verbatim bytes and
// deliberately did not send them — almost always because including them
// would have pushed the envelope past MaxIngestBodyBytes, so the choice
// was between a turn without its bytes and no turn at all.
//
// Without it that turn is indistinguishable from one produced by an
// adapter that never captured raw bytes in the first place: both arrive
// with raw_response absent. Those are opposite facts. The first is a
// limit that bit and wants tuning; the second is a deployment fact about
// which producers are running. A fidelity report that cannot separate
// them reports the wrong one — silently, and in the reassuring
// direction, since "this producer doesn't send raw" reads as expected
// where "we lost bytes we had" does not.
//
// Deliberately NOT named to match the raw_response_dropped column it
// feeds. The column is the union of two causes — the producer withheld,
// or ingest capped — and this field is only one of them. A shared name
// would imply the producer sets the column, when it contributes to it.
//
// Absent means "no claim", which is exactly the pre-existing behavior:
// every producer shipped before this field omits it and keeps reading as
// it always did.
RawResponseWithheld bool `json:"raw_response_withheld,omitempty"`
// Meta is the capture adapter's metadata block. Parsed for the
// fields ingest promotes (request_id for raw-turn dedup); the
// verbatim JSON is persisted alongside the raw turn so fields
// unknown to this build survive.
Meta TurnMeta `json:"meta"`
// Session is the optional session-tracking envelope. When present,
// ingest UPSERTs a `sessions` row keyed by
// (org_id, harness_id, harness_session_id), resolves the
// parent_session_id FK (placeholder-inserting when needed), and
// rolls up turn counters — all in the same transaction as the
// nodes insert. When absent, ingest treats the turn as
// harness_id="unknown" and derives a synthetic harness_session_id
// from the captured turn's Merkle root prefix.
//
// The type lives in pkg/sessions to avoid an import cycle
// (proxy/worker depends on it too).
Session *sessions.IngestEnvelope `json:"session,omitempty"`
}
TurnPayload is the ingest request body for a single completed conversation turn. It carries the raw provider request plus an already-reduced response. Capture adapters such as tapes-extproc own protocol-specific stream reduction; ingest owns request parsing, validation, and durable storage.