From 141984be8585b70cac26a8b5fe1d8a5bd2a129cb Mon Sep 17 00:00:00 2001 From: "Jakub A. W" Date: Sun, 26 Jul 2026 22:03:11 +0200 Subject: [PATCH 1/2] fix(auditlog): keep request revisions on streamed entries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CreateStreamEntry rebuilds LogData with a field whitelist, and the stream observer completes and persists that copy — the base entry is never written. RequestRevisions was missing from the whitelist, so the ingress rewrite chain recorded by EnrichEntryWithRequestRevision was discarded on every streamed request. The audit UI's "Rewritten" pane therefore never appeared for successful streams, while surviving on the non-streamed error path. Measured on a live gateway: of 72 requests where a rewriter reported token savings into the usage table, 0 retained a revision snapshot; all 6 non-streamed failures in the same window kept theirs. RequestBodyTooBigToHandle was dropped the same way — also a request-side fact known before the stream opens. Response-side fields (ResponseBody, ErrorMessage, ErrorCode, ResponseBodyTooBigToHandle) stay omitted on purpose: the observer fills them once the stream closes. Adds a reflection-based drift guard that walks LogData and fails when any request-side field does not survive the copy, since the whitelist will keep attracting this bug otherwise. Co-Authored-By: Claude Opus 5 (1M context) --- .../stream_entry_request_fields_test.go | 128 ++++++++++++++++++ internal/auditlog/stream_wrapper.go | 38 ++++-- 2 files changed, 157 insertions(+), 9 deletions(-) create mode 100644 internal/auditlog/stream_entry_request_fields_test.go diff --git a/internal/auditlog/stream_entry_request_fields_test.go b/internal/auditlog/stream_entry_request_fields_test.go new file mode 100644 index 000000000..9ab14ac86 --- /dev/null +++ b/internal/auditlog/stream_entry_request_fields_test.go @@ -0,0 +1,128 @@ +package auditlog + +import ( + "reflect" + "testing" +) + +// responseSideLogDataFields are the LogData fields CreateStreamEntry may +// legitimately leave unset: they are not known when the stream entry is +// created and are filled in by the stream observer once the stream closes. +// Every other field describes the request and must survive the copy. +var responseSideLogDataFields = map[string]bool{ + "ResponseBody": true, + "ResponseBodyTooBigToHandle": true, + "ErrorMessage": true, + "ErrorCode": true, +} + +// A streamed request is persisted from the CreateStreamEntry copy — the base +// entry is never written — so an ingress rewrite chain recorded by +// EnrichEntryWithRequestRevision is lost unless the copy carries it. This +// regression covers request rewriters (e.g. pro token compression) whose +// "Rewritten" audit pane vanished on every successful streamed request while +// surviving on the non-streamed error path. +func TestCreateStreamEntryPreservesRequestRevisions(t *testing.T) { + base := &LogEntry{ + ID: "entry-1", + Path: "/v1/chat/completions", + Data: &LogData{ + RequestRevisions: []RequestRevisionSnapshot{{ + Seq: 1, + Rewriter: "pro-token-compression", + BytesBefore: 65209, + BytesAfter: 64418, + TokensSaved: 189, + Detail: map[string]any{"chars_removed": 757}, + }}, + RequestBodyTooBigToHandle: true, + }, + } + + streamEntry := CreateStreamEntry(base) + if streamEntry == nil || streamEntry.Data == nil { + t.Fatal("expected a stream entry with data") + } + + got := streamEntry.Data.RequestRevisions + if len(got) != 1 { + t.Fatalf("RequestRevisions dropped: got %d revisions, want 1", len(got)) + } + if got[0].Rewriter != "pro-token-compression" || got[0].TokensSaved != 189 { + t.Fatalf("revision not copied faithfully: %+v", got[0]) + } + if got[0].BytesBefore != 65209 || got[0].BytesAfter != 64418 { + t.Fatalf("revision byte counts not copied: %+v", got[0]) + } + if !streamEntry.Data.RequestBodyTooBigToHandle { + t.Error("RequestBodyTooBigToHandle dropped") + } + + // The copy must own its slice, so later appends to the base entry cannot + // reach into the entry the observer is writing. + base.Data.RequestRevisions = append(base.Data.RequestRevisions, RequestRevisionSnapshot{Seq: 2}) + if len(streamEntry.Data.RequestRevisions) != 1 { + t.Error("stream entry shares its revision backing array with the base entry") + } +} + +// CreateStreamEntry builds LogData with a field whitelist, so any request-side +// field added to LogData later is silently dropped until someone remembers to +// extend that literal. This walks LogData by reflection and fails when a +// request-side field does not survive, which is how RequestRevisions went +// missing in the first place. +func TestCreateStreamEntryCopiesEveryRequestSideField(t *testing.T) { + populated := &LogData{} + v := reflect.ValueOf(populated).Elem() + typ := v.Type() + + for i := range typ.NumField() { + field := typ.Field(i) + if !v.Field(i).CanSet() { + continue + } + if !setRecognizableValue(v.Field(i)) { + t.Fatalf("test needs a sample value for LogData.%s (%s)", field.Name, field.Type) + } + } + + streamEntry := CreateStreamEntry(&LogEntry{ID: "entry-1", Data: populated}) + if streamEntry == nil || streamEntry.Data == nil { + t.Fatal("expected a stream entry with data") + } + + copied := reflect.ValueOf(streamEntry.Data).Elem() + for i := range typ.NumField() { + name := typ.Field(i).Name + if responseSideLogDataFields[name] { + continue + } + if copied.Field(i).IsZero() { + t.Errorf("LogData.%s is a request-side field but CreateStreamEntry dropped it", name) + } + } +} + +// setRecognizableValue fills one field with a non-zero value so a dropped +// field shows up as the zero value on the other side of the copy. +func setRecognizableValue(field reflect.Value) bool { + switch field.Kind() { + case reflect.String: + field.SetString("x") + case reflect.Bool: + field.SetBool(true) + case reflect.Map: + m := reflect.MakeMap(field.Type()) + m.SetMapIndex(reflect.ValueOf("k"), reflect.ValueOf("v")) + field.Set(m) + case reflect.Slice: + field.Set(reflect.MakeSlice(field.Type(), 1, 1)) + case reflect.Pointer: + field.Set(reflect.New(field.Type().Elem())) + case reflect.Interface: + field.Set(reflect.ValueOf(map[string]any{"k": "v"})) + default: + return false + } + return true +} diff --git a/internal/auditlog/stream_wrapper.go b/internal/auditlog/stream_wrapper.go index 9b5a4698d..9108a0355 100644 --- a/internal/auditlog/stream_wrapper.go +++ b/internal/auditlog/stream_wrapper.go @@ -2,6 +2,7 @@ package auditlog import ( "maps" + "slices" "sort" "strings" ) @@ -234,17 +235,25 @@ func CreateStreamEntry(baseEntry *LogEntry) *LogEntry { Stream: true, // Mark as streaming } + // This is a whitelist copy, so every request-side field of LogData must be + // listed here or it is silently lost: the stream observer completes and + // writes THIS copy, and the base entry is never persisted. Only the + // response-side fields may be omitted — ResponseBody, ErrorMessage, + // ErrorCode and ResponseBodyTooBigToHandle are all filled in later, once + // the stream closes. Anything already known at ingress belongs below. if baseEntry.Data != nil { entryCopy.Data = &LogData{ - UserAgent: baseEntry.Data.UserAgent, - APIKeyHash: baseEntry.Data.APIKeyHash, - Labels: baseEntry.Data.Labels, - Temperature: baseEntry.Data.Temperature, - MaxTokens: baseEntry.Data.MaxTokens, - RequestHeaders: copyMap(baseEntry.Data.RequestHeaders), - ResponseHeaders: copyMap(baseEntry.Data.ResponseHeaders), - RequestBody: baseEntry.Data.RequestBody, - Attempts: normalizeAttemptSnapshots(baseEntry.Data.Attempts), + UserAgent: baseEntry.Data.UserAgent, + APIKeyHash: baseEntry.Data.APIKeyHash, + Labels: baseEntry.Data.Labels, + Temperature: baseEntry.Data.Temperature, + MaxTokens: baseEntry.Data.MaxTokens, + RequestHeaders: copyMap(baseEntry.Data.RequestHeaders), + ResponseHeaders: copyMap(baseEntry.Data.ResponseHeaders), + RequestBody: baseEntry.Data.RequestBody, + RequestBodyTooBigToHandle: baseEntry.Data.RequestBodyTooBigToHandle, + RequestRevisions: copyRequestRevisions(baseEntry.Data.RequestRevisions), + Attempts: normalizeAttemptSnapshots(baseEntry.Data.Attempts), } if baseEntry.Data.WorkflowFeatures != nil { snapshot := *baseEntry.Data.WorkflowFeatures @@ -259,6 +268,17 @@ func CreateStreamEntry(baseEntry *LogEntry) *LogEntry { return entryCopy } +// copyRequestRevisions copies the ingress rewrite chain so the streamed entry +// owns its own slice. The chain is complete before the stream opens, but the +// copy keeps this consistent with the map copying above and leaves no shared +// backing array between the base entry and the entry the observer writes. +func copyRequestRevisions(revisions []RequestRevisionSnapshot) []RequestRevisionSnapshot { + if revisions == nil { + return nil + } + return slices.Clone(revisions) +} + // copyMap creates a shallow copy of a string map func copyMap(m map[string]string) map[string]string { if m == nil { From 16cf4e84dd8b001b78b3075ce87ffe472245fa00 Mon Sep 17 00:00:00 2001 From: "Jakub A. W" Date: Sun, 26 Jul 2026 22:46:30 +0200 Subject: [PATCH 2/2] test(auditlog): give the slice-ownership assertion teeth MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The append-based check could not fail: the source literal has no spare capacity, so append reallocates and leaves the copy untouched whether or not the two share a backing array. Verified by swapping the clone for a shallow assignment — the test still passed. Writing through an existing element does distinguish them, and now fails against that same shallow assignment. Also covers the absent-revisions case, which is what most streamed requests hit: it must stay nil rather than becoming an empty slice. Co-Authored-By: Claude Opus 5 (1M context) --- .../stream_entry_request_fields_test.go | 26 ++++++++++++++++--- 1 file changed, 22 insertions(+), 4 deletions(-) diff --git a/internal/auditlog/stream_entry_request_fields_test.go b/internal/auditlog/stream_entry_request_fields_test.go index 9ab14ac86..e546b5a34 100644 --- a/internal/auditlog/stream_entry_request_fields_test.go +++ b/internal/auditlog/stream_entry_request_fields_test.go @@ -58,14 +58,32 @@ func TestCreateStreamEntryPreservesRequestRevisions(t *testing.T) { t.Error("RequestBodyTooBigToHandle dropped") } - // The copy must own its slice, so later appends to the base entry cannot - // reach into the entry the observer is writing. - base.Data.RequestRevisions = append(base.Data.RequestRevisions, RequestRevisionSnapshot{Seq: 2}) - if len(streamEntry.Data.RequestRevisions) != 1 { + // The copy must own its slice. Appending to the base entry would not show + // that: the source literal has no spare capacity, so append reallocates and + // leaves the copy alone whether or not the two share an array. Writing + // through an existing element is what actually distinguishes them. + base.Data.RequestRevisions[0].Rewriter = "mutated" + if streamEntry.Data.RequestRevisions[0].Rewriter != "pro-token-compression" { t.Error("stream entry shares its revision backing array with the base entry") } } +// The nil case is the one the stream observer sees most often — most requests +// carry no rewriter — and it must stay nil rather than becoming an empty slice, +// so a streamed entry without rewrites serializes the same as it always did. +func TestCreateStreamEntryLeavesAbsentRequestRevisionsNil(t *testing.T) { + streamEntry := CreateStreamEntry(&LogEntry{ID: "entry-1", Data: &LogData{UserAgent: "curl/8"}}) + if streamEntry == nil || streamEntry.Data == nil { + t.Fatal("expected a stream entry with data") + } + if streamEntry.Data.RequestRevisions != nil { + t.Errorf("RequestRevisions = %#v, want nil", streamEntry.Data.RequestRevisions) + } + if streamEntry.Data.UserAgent != "curl/8" { + t.Errorf("UserAgent = %q, want %q", streamEntry.Data.UserAgent, "curl/8") + } +} + // CreateStreamEntry builds LogData with a field whitelist, so any request-side // field added to LogData later is silently dropped until someone remembers to // extend that literal. This walks LogData by reflection and fails when a