From 56c4b3aac88e174f5406f071dbb7af6a4b26334b Mon Sep 17 00:00:00 2001 From: "Jakub A. W" Date: Sat, 8 Aug 2026 16:15:49 +0200 Subject: [PATCH 1/3] fix(cache): stabilize compression session feedback --- ext/ext.go | 33 +++ internal/server/request_rewrite.go | 13 ++ internal/server/request_rewrite_test.go | 71 ++++++ internal/server/response_feedback.go | 221 ++++++++++++++++++ internal/server/response_feedback_test.go | 108 +++++++++ .../server/translated_inference_service.go | 44 +++- internal/session/detect.go | 53 ++++- internal/session/detect_test.go | 31 +++ 8 files changed, 566 insertions(+), 8 deletions(-) create mode 100644 internal/server/response_feedback.go create mode 100644 internal/server/response_feedback_test.go diff --git a/ext/ext.go b/ext/ext.go index dbc2fe94a..98343c730 100644 --- a/ext/ext.go +++ b/ext/ext.go @@ -81,6 +81,39 @@ type RequestRewriter interface { Rewrite(ctx context.Context, in Input) (*Result, error) } +// ResponseFeedbackObserver receives content-free feedback after a rewritten +// request successfully reaches a provider. It is an optional companion to +// RequestRewriter: core detects implementations structurally and invokes them +// for both ordinary and streaming responses. usageObserved distinguishes a +// confirmed zero from a provider/stream that returned no usage breakdown. +// Implementations must be safe for concurrent use and return promptly. +// +// The flat signature intentionally uses only long-standing extension types so +// extensions can implement the hook while supporting older core releases; +// older cores simply never call it. +type ResponseFeedbackObserver interface { + ObserveResponse( + ctx context.Context, + requestID string, + endpoint Endpoint, + sessionID string, + model string, + providerType string, + providerName string, + inputTokens int, + cachedInputTokens int, + cacheWriteInputTokens int, + usageObserved bool, + ) +} + +// ResponseFeedbackFilter lets an observer decline feedback per request. Core +// calls it after Rewrite and registers the observer only when it returns true. +// Observers without this optional interface receive every successful response. +type ResponseFeedbackFilter interface { + WantsResponseFeedback(in Input, result *Result) bool +} + // SettingOption is one allowed value for a dashboard-editable extension // setting. Label and Description are safe to expose in the admin UI. type SettingOption struct { diff --git a/internal/server/request_rewrite.go b/internal/server/request_rewrite.go index 3543b0bef..c9e9e0d9d 100644 --- a/internal/server/request_rewrite.go +++ b/internal/server/request_rewrite.go @@ -47,11 +47,21 @@ func RequestRewriteMiddleware(rewriters []ext.RequestRewriter, auditLogger audit changed := false tokensSaved := 0 + feedbackObservers := make([]ext.ResponseFeedbackObserver, 0, len(rewriters)) for _, rw := range rewriters { res, rwErr := rw.Rewrite(c.Request().Context(), in) if rwErr != nil { return handleError(c, rewriterGatewayError(rw.Name(), rwErr)) } + if observer, ok := rw.(ext.ResponseFeedbackObserver); ok { + wantsFeedback := true + if filter, filtered := rw.(ext.ResponseFeedbackFilter); filtered { + wantsFeedback = filter.WantsResponseFeedback(in, res) + } + if wantsFeedback { + feedbackObservers = append(feedbackObservers, observer) + } + } if res != nil { applyRewriteResponseHeaders(c, res.ResponseHeader) } @@ -78,6 +88,9 @@ func RequestRewriteMiddleware(rewriters []ext.RequestRewriter, auditLogger audit c.SetRequest(req.WithContext(core.WithRewriteTokensSaved(req.Context(), tokensSaved))) } } + if len(feedbackObservers) > 0 { + setResponseFeedbackObservers(c, feedbackObservers) + } return next(c) } } diff --git a/internal/server/request_rewrite_test.go b/internal/server/request_rewrite_test.go index d14fb16f6..9ec1ce3bc 100644 --- a/internal/server/request_rewrite_test.go +++ b/internal/server/request_rewrite_test.go @@ -15,6 +15,7 @@ import ( "github.com/enterpilot/gomodel/ext" "github.com/enterpilot/gomodel/internal/auditlog" "github.com/enterpilot/gomodel/internal/core" + "github.com/enterpilot/gomodel/internal/session" ) type stubRewriter struct { @@ -23,6 +24,20 @@ type stubRewriter struct { rewrite func(in ext.Input) (*ext.Result, error) } +type feedbackRewriter struct { + stubRewriter + feedbackCaptureObserver +} + +type filteredFeedbackRewriter struct { + feedbackRewriter + want bool +} + +func (r *filteredFeedbackRewriter) WantsResponseFeedback(ext.Input, *ext.Result) bool { + return r.want +} + func (r *stubRewriter) Name() string { return r.name } func (r *stubRewriter) Rewrite(_ context.Context, in ext.Input) (*ext.Result, error) { @@ -119,6 +134,62 @@ func TestRequestRewriteMiddlewareRewritesChatCompletions(t *testing.T) { } } +func TestRequestRewriteMiddlewareDeliversProviderFeedback(t *testing.T) { + provider := newRewriteTestProvider() + provider.response.Usage = core.Usage{ + PromptTokens: 2048, + PromptTokensDetails: &core.PromptTokensDetails{CachedTokens: 1536}, + } + rewriter := &feedbackRewriter{stubRewriter: stubRewriter{name: "feedback"}} + srv := New(provider, &Config{ + RequestRewriters: []ext.RequestRewriter{rewriter}, + SessionDetector: session.NewDetector(session.BuiltinRules(), false), + }) + + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", + strings.NewReader(`{"model":"gpt-4o-mini","messages":[{"role":"user","content":"hello"}]}`)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Session-Id", "feedback-session") + req.Header.Set("X-Request-Id", "feedback-request") + rec := httptest.NewRecorder() + srv.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d (%s)", rec.Code, rec.Body.String()) + } + if len(rewriter.feedback) != 1 { + t.Fatalf("feedback count = %d, want 1", len(rewriter.feedback)) + } + got := rewriter.feedback[0] + if got.requestID != "feedback-request" || got.sessionID != "feedback-session" || got.cacheRead != 1536 || !got.usageObserved { + t.Fatalf("feedback = %+v", got) + } +} + +func TestRequestRewriteMiddlewareHonorsResponseFeedbackFilter(t *testing.T) { + for _, want := range []bool{false, true} { + rewriter := &filteredFeedbackRewriter{ + feedbackRewriter: feedbackRewriter{stubRewriter: stubRewriter{name: "filtered"}}, + want: want, + } + var attached bool + next := func(c *echo.Context) error { + attached = hasResponseFeedbackObservers(c) + return c.NoContent(http.StatusOK) + } + e := echo.New() + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", + strings.NewReader(`{"model":"gpt-4o-mini","messages":[]}`)) + c := e.NewContext(req, httptest.NewRecorder()) + if err := RequestRewriteMiddleware([]ext.RequestRewriter{rewriter}, nil)(next)(c); err != nil { + t.Fatalf("want=%v: middleware: %v", want, err) + } + if attached != want { + t.Fatalf("want=%v: attached=%v", want, attached) + } + } +} + func TestRequestRewriteMiddlewareExposesSessionID(t *testing.T) { tests := []struct { name string diff --git a/internal/server/response_feedback.go b/internal/server/response_feedback.go new file mode 100644 index 000000000..f80edaa6f --- /dev/null +++ b/internal/server/response_feedback.go @@ -0,0 +1,221 @@ +package server + +import ( + "bytes" + "context" + "encoding/json" + "strconv" + + "github.com/labstack/echo/v5" + + "github.com/enterpilot/gomodel/ext" + "github.com/enterpilot/gomodel/internal/core" +) + +const responseFeedbackObserversKey = "gomodel.response-feedback-observers" + +func setResponseFeedbackObservers(c *echo.Context, observers []ext.ResponseFeedbackObserver) { + if c == nil || len(observers) == 0 { + return + } + c.Set(responseFeedbackObserversKey, append([]ext.ResponseFeedbackObserver(nil), observers...)) +} + +func responseFeedbackObservers(c *echo.Context) []ext.ResponseFeedbackObserver { + if c == nil { + return nil + } + observers, _ := c.Get(responseFeedbackObserversKey).([]ext.ResponseFeedbackObserver) + return observers +} + +func hasResponseFeedbackObservers(c *echo.Context) bool { + return len(responseFeedbackObservers(c)) > 0 +} + +type responseCacheUsage struct { + input int + read int + write int + observed bool +} + +func notifyChatResponseFeedback(c *echo.Context, endpoint ext.Endpoint, resp *core.ChatResponse, model, providerType, providerName string) { + observers := responseFeedbackObservers(c) + if len(observers) == 0 { + return + } + usage := responseCacheUsage{} + if resp != nil { + usage = cacheUsageFromCore(resp.Usage.PromptTokens, resp.Usage.PromptTokensDetails, resp.Usage.RawUsage) + if resp.Model != "" { + model = resp.Model + } + } + notifyResponseFeedback(c.Request().Context(), observers, core.GetRequestID(c.Request().Context()), core.SessionIDFromContext(c.Request().Context()), endpoint, model, providerType, providerName, usage) +} + +func notifyResponsesResponseFeedback(c *echo.Context, endpoint ext.Endpoint, resp *core.ResponsesResponse, model, providerType, providerName string) { + observers := responseFeedbackObservers(c) + if len(observers) == 0 { + return + } + usage := responseCacheUsage{} + if resp != nil { + if resp.Usage != nil { + usage = cacheUsageFromCore(resp.Usage.InputTokens, resp.Usage.PromptTokensDetails, resp.Usage.RawUsage) + } + if resp.Model != "" { + model = resp.Model + } + } + notifyResponseFeedback(c.Request().Context(), observers, core.GetRequestID(c.Request().Context()), core.SessionIDFromContext(c.Request().Context()), endpoint, model, providerType, providerName, usage) +} + +func notifyResponseFeedback( + ctx context.Context, + observers []ext.ResponseFeedbackObserver, + requestID, sessionID string, + endpoint ext.Endpoint, + model, providerType, providerName string, + usage responseCacheUsage, +) { + for _, observer := range observers { + if observer == nil { + continue + } + func() { + defer func() { _ = recover() }() + observer.ObserveResponse( + ctx, requestID, endpoint, sessionID, model, providerType, providerName, + usage.input, usage.read, usage.write, usage.observed, + ) + }() + } +} + +func cacheUsageFromCore(input int, details *core.PromptTokensDetails, raw map[string]any) responseCacheUsage { + usage := responseCacheUsage{input: input, observed: input > 0 || details != nil || raw != nil} + if details != nil { + usage.read = details.CachedTokens + } + read, write := cacheTokensFromMap(raw) + usage.read = max(usage.read, read) + usage.write = write + return usage +} + +// responseFeedbackStreamObserver observes only usage-bearing SSE events and +// emits one summary when the stream closes. The provider bytes remain +// untouched and no response content is retained. +type responseFeedbackStreamObserver struct { + ctx context.Context + observers []ext.ResponseFeedbackObserver + requestID string + sessionID string + endpoint ext.Endpoint + model string + providerType string + providerName string + usage responseCacheUsage +} + +func (o *responseFeedbackStreamObserver) WantsJSONEvent(raw []byte) bool { + return bytes.Contains(raw, []byte(`"usage"`)) +} + +func (o *responseFeedbackStreamObserver) OnJSONEvent(payload map[string]any) { + usageMap, ok := streamUsageMap(payload) + if !ok { + return + } + usage := responseCacheUsage{observed: true} + usage.input = firstNumericInt(usageMap, "prompt_tokens", "input_tokens") + usage.read, usage.write = cacheTokensFromMap(usageMap) + if details, ok := nestedMap(usageMap["prompt_tokens_details"]); ok { + usage.read = max(usage.read, firstNumericInt(details, "cached_tokens")) + } + if details, ok := nestedMap(usageMap["input_tokens_details"]); ok { + usage.read = max(usage.read, firstNumericInt(details, "cached_tokens")) + } + o.usage = usage +} + +func (o *responseFeedbackStreamObserver) OnStreamClose() { + if o == nil || o.ctx == nil { + return + } + notifyResponseFeedback( + context.WithoutCancel(o.ctx), + o.observers, + o.requestID, + o.sessionID, + o.endpoint, + o.model, + o.providerType, + o.providerName, + o.usage, + ) +} + +func streamUsageMap(payload map[string]any) (map[string]any, bool) { + if usage, ok := nestedMap(payload["usage"]); ok { + return usage, true + } + for _, key := range []string{"response", "message"} { + container, ok := nestedMap(payload[key]) + if !ok { + continue + } + if usage, ok := nestedMap(container["usage"]); ok { + return usage, true + } + } + return nil, false +} + +func cacheTokensFromMap(raw map[string]any) (read, write int) { + if len(raw) == 0 { + return 0, 0 + } + read = firstNumericInt(raw, "cache_read_input_tokens", "prompt_cached_tokens", "cached_tokens") + write = firstNumericInt(raw, "cache_creation_input_tokens", "cache_write_input_tokens") + if nested, ok := nestedMap(raw["raw_usage"]); ok { + nestedRead, nestedWrite := cacheTokensFromMap(nested) + read = max(read, nestedRead) + write = max(write, nestedWrite) + } + return read, write +} + +func firstNumericInt(values map[string]any, keys ...string) int { + for _, key := range keys { + if value, ok := numericInt(values[key]); ok { + return value + } + } + return 0 +} + +func numericInt(value any) (int, bool) { + switch typed := value.(type) { + case int: + return typed, true + case int32: + return int(typed), true + case int64: + return int(typed), true + case float64: + return int(typed), true + case json.Number: + parsed, err := strconv.ParseInt(string(typed), 10, 64) + return int(parsed), err == nil + default: + return 0, false + } +} + +func nestedMap(value any) (map[string]any, bool) { + typed, ok := value.(map[string]any) + return typed, ok +} diff --git a/internal/server/response_feedback_test.go b/internal/server/response_feedback_test.go new file mode 100644 index 000000000..53694c464 --- /dev/null +++ b/internal/server/response_feedback_test.go @@ -0,0 +1,108 @@ +package server + +import ( + "context" + "testing" + + "github.com/enterpilot/gomodel/ext" + "github.com/enterpilot/gomodel/internal/core" +) + +type capturedResponseFeedback struct { + requestID string + endpoint ext.Endpoint + sessionID string + model string + providerType string + providerName string + inputTokens int + cacheRead int + cacheWrite int + usageObserved bool +} + +type feedbackCaptureObserver struct { + feedback []capturedResponseFeedback +} + +func (o *feedbackCaptureObserver) ObserveResponse( + _ context.Context, + requestID string, + endpoint ext.Endpoint, + sessionID, model, providerType, providerName string, + inputTokens, cachedInputTokens, cacheWriteInputTokens int, + usageObserved bool, +) { + o.feedback = append(o.feedback, capturedResponseFeedback{ + requestID: requestID, endpoint: endpoint, sessionID: sessionID, + model: model, providerType: providerType, providerName: providerName, + inputTokens: inputTokens, cacheRead: cachedInputTokens, cacheWrite: cacheWriteInputTokens, + usageObserved: usageObserved, + }) +} + +func TestNotifyChatResponseFeedbackIncludesRouteAndCacheUsage(t *testing.T) { + observer := &feedbackCaptureObserver{} + ctx := core.WithRequestID(context.Background(), "req-1") + ctx = core.WithSessionID(ctx, "session-1") + resp := &core.ChatResponse{ + Model: "gpt-5.6", + Usage: core.Usage{ + PromptTokens: 2400, + PromptTokensDetails: &core.PromptTokensDetails{CachedTokens: 1800}, + RawUsage: map[string]any{"cache_creation_input_tokens": 300}, + }, + } + + usage := cacheUsageFromCore(resp.Usage.PromptTokens, resp.Usage.PromptTokensDetails, resp.Usage.RawUsage) + notifyResponseFeedback(ctx, []ext.ResponseFeedbackObserver{observer}, "req-1", "session-1", ext.EndpointChatCompletions, resp.Model, "anthropic", "primary", usage) + if len(observer.feedback) != 1 { + t.Fatalf("feedback count = %d, want 1", len(observer.feedback)) + } + got := observer.feedback[0] + if got.requestID != "req-1" || got.sessionID != "session-1" || got.model != "gpt-5.6" || + got.providerType != "anthropic" || got.providerName != "primary" || got.inputTokens != 2400 || + got.cacheRead != 1800 || got.cacheWrite != 300 || !got.usageObserved { + t.Fatalf("feedback = %+v", got) + } +} + +func TestResponseFeedbackStreamObserverUsesLatestUsageEvent(t *testing.T) { + observer := &feedbackCaptureObserver{} + ctx := core.WithRequestID(context.Background(), "req-stream") + ctx = core.WithSessionID(ctx, "session-stream") + streamObserver := &responseFeedbackStreamObserver{ + ctx: ctx, observers: []ext.ResponseFeedbackObserver{observer}, requestID: "req-stream", sessionID: "session-stream", + endpoint: ext.EndpointResponses, model: "claude", providerType: "anthropic", providerName: "primary", + } + streamObserver.OnJSONEvent(map[string]any{ + "type": "message_start", + "message": map[string]any{"usage": map[string]any{ + "input_tokens": float64(2000), "cache_read_input_tokens": float64(1600), + }}, + }) + streamObserver.OnJSONEvent(map[string]any{ + "type": "response.completed", + "response": map[string]any{"usage": map[string]any{ + "input_tokens": float64(2300), "cache_read_input_tokens": float64(1900), "cache_creation_input_tokens": float64(200), + }}, + }) + streamObserver.OnStreamClose() + + if len(observer.feedback) != 1 { + t.Fatalf("feedback count = %d, want 1", len(observer.feedback)) + } + got := observer.feedback[0] + if got.endpoint != ext.EndpointResponses || got.inputTokens != 2300 || got.cacheRead != 1900 || got.cacheWrite != 200 || !got.usageObserved { + t.Fatalf("feedback = %+v", got) + } +} + +func TestResponseFeedbackStreamObserverReportsUnknownUsage(t *testing.T) { + observer := &feedbackCaptureObserver{} + streamObserver := &responseFeedbackStreamObserver{ctx: context.Background(), observers: []ext.ResponseFeedbackObserver{observer}, endpoint: ext.EndpointChatCompletions} + streamObserver.OnStreamClose() + if len(observer.feedback) != 1 || observer.feedback[0].usageObserved { + t.Fatalf("feedback = %+v, want one unknown-usage observation", observer.feedback) + } +} diff --git a/internal/server/translated_inference_service.go b/internal/server/translated_inference_service.go index bb5948264..4f9ef61ab 100644 --- a/internal/server/translated_inference_service.go +++ b/internal/server/translated_inference_service.go @@ -14,6 +14,7 @@ import ( "github.com/labstack/echo/v5" + "github.com/enterpilot/gomodel/ext" "github.com/enterpilot/gomodel/internal/auditlog" "github.com/enterpilot/gomodel/internal/conversationstore" "github.com/enterpilot/gomodel/internal/core" @@ -105,7 +106,15 @@ func (s *translatedInferenceService) dispatchChatCompletion(c *echo.Context, req ctx = adm.dispatchContext(ctx) if req.Stream { - if len(s.inference().FailoverSelectors(workflow)) == 0 { + feedbackEnabled := hasResponseFeedbackObservers(c) + if feedbackEnabled { + req = gateway.CloneChatRequestForStreamUsage(req) + if req.StreamOptions == nil { + req.StreamOptions = &core.StreamOptions{} + } + req.StreamOptions.IncludeUsage = true + } + if !feedbackEnabled && len(s.inference().FailoverSelectors(workflow)) == 0 { if handled, err := s.tryFastPathStreamingChatPassthrough(c, workflow, req); handled { return err } @@ -144,6 +153,14 @@ func (s *translatedInferenceService) dispatchChatCompletion(c *echo.Context, req result.Meta.ProviderType, result.Meta.ProviderName, ) + notifyChatResponseFeedback( + c, + ext.Endpoint(c.Request().URL.Path), + result.Response, + result.Meta.Model, + result.Meta.ProviderType, + result.Meta.ProviderName, + ) return c.JSON(http.StatusOK, result.Response) } @@ -273,6 +290,9 @@ func (s *translatedInferenceService) dispatchResponses(c *echo.Context, req *cor ctx = adm.dispatchContext(ctx) if req.Stream { + if hasResponseFeedbackObservers(c) { + ctx = core.WithEnforceReturningUsageData(ctx, true) + } result, err := s.inference().StreamResponses(ctx, workflow, req) if err != nil { return handleStreamingDispatchError(c, err) @@ -311,6 +331,14 @@ func (s *translatedInferenceService) dispatchResponses(c *echo.Context, req *cor result.Meta.ProviderType, result.Meta.ProviderName, ) + notifyResponsesResponseFeedback( + c, + ext.Endpoint(c.Request().URL.Path), + result.Response, + result.Meta.Model, + result.Meta.ProviderType, + result.Meta.ProviderName, + ) if turn := conversationTurnFromContext(ctx); turn != nil { // Detach cancellation so a client disconnect after provider success @@ -550,7 +578,7 @@ func (s *translatedInferenceService) handleStreamingReadCloser( requestID := requestIDFromContextOrHeader(c.Request()) endpoint := c.Request().URL.Path - observers := make([]streaming.Observer, 0, 2) + observers := make([]streaming.Observer, 0, 3) if auditEnabled && streamEntry != nil { observers = append(observers, auditlog.NewStreamLogObserver(s.logger, streamEntry, endpoint)) } @@ -563,6 +591,18 @@ func (s *translatedInferenceService) handleStreamingReadCloser( observers = append(observers, usageObserver) } } + if hasResponseFeedbackObservers(c) { + observers = append(observers, &responseFeedbackStreamObserver{ + ctx: c.Request().Context(), + observers: responseFeedbackObservers(c), + requestID: requestID, + sessionID: core.SessionIDFromContext(c.Request().Context()), + endpoint: ext.Endpoint(endpoint), + model: model, + providerType: provider, + providerName: providerName, + }) + } wrappedStream := streaming.NewObservedSSEStream(stream, observers...) if outerWrap != nil { wrappedStream = outerWrap(wrappedStream) diff --git a/internal/session/detect.go b/internal/session/detect.go index 9182d3f2c..3ed71f415 100644 --- a/internal/session/detect.go +++ b/internal/session/detect.go @@ -1,9 +1,12 @@ package session import ( + "bytes" "crypto/sha256" "encoding/hex" "encoding/json" + "errors" + "io" "strings" "github.com/tidwall/gjson" @@ -134,17 +137,17 @@ func contentSessionID(snapshot *core.RequestSnapshot, body []byte, userPath stri root := gjson.ParseBytes(body) anchor := contentAnchor{ UserPath: userPath, - Model: rawSegment(root.Get("model")), - System: rawSegment(root.Get("system")), - Instructions: rawSegment(root.Get("instructions")), - Tools: rawSegment(root.Get("tools")), + Model: canonicalSegment(root.Get("model")), + System: canonicalSegment(root.Get("system")), + Instructions: canonicalSegment(root.Get("instructions")), + Tools: canonicalSegment(root.Get("tools")), } messages := root.Get("messages") if !messages.Exists() { messages = root.Get("input") } if messages.Type == gjson.String { - anchor.Opening = []json.RawMessage{rawSegment(messages)} + anchor.Opening = []json.RawMessage{canonicalSegment(messages)} } else { anchor.Opening = openingMessages(messages) } @@ -170,7 +173,7 @@ const maxOpeningMessages = 8 func openingMessages(messages gjson.Result) []json.RawMessage { var opening []json.RawMessage messages.ForEach(func(_, message gjson.Result) bool { - opening = append(opening, rawSegment(message)) + opening = append(opening, canonicalSegment(message)) if message.Get("role").Str == "user" || len(opening) >= maxOpeningMessages { return false } @@ -187,3 +190,41 @@ func rawSegment(result gjson.Result) json.RawMessage { } return json.RawMessage(strings.Clone(result.Raw)) } + +// canonicalSegment gives semantically equivalent JSON the same session +// anchor. In particular, object-key order, insignificant whitespace, and +// equivalent string escapes must not split one conversation into multiple +// auto-detected sessions. UseNumber preserves number spelling/precision while +// arrays retain their original order. +func canonicalSegment(result gjson.Result) json.RawMessage { + raw := rawSegment(result) + if len(raw) == 0 { + return nil + } + decoder := json.NewDecoder(bytes.NewReader(raw)) + decoder.UseNumber() + var value any + if err := decoder.Decode(&value); err != nil { + return raw + } + if err := ensureJSONEOF(decoder); err != nil { + return raw + } + canonical, err := json.Marshal(value) + if err != nil { + return raw + } + return canonical +} + +func ensureJSONEOF(decoder *json.Decoder) error { + var trailing any + err := decoder.Decode(&trailing) + if errors.Is(err, io.EOF) { + return nil + } + if err == nil { + return errors.New("multiple JSON values") + } + return err +} diff --git a/internal/session/detect_test.go b/internal/session/detect_test.go index ec5450a03..1ea8918a7 100644 --- a/internal/session/detect_test.go +++ b/internal/session/detect_test.go @@ -196,6 +196,37 @@ func TestDetectAutoStability(t *testing.T) { } } +func TestDetectAutoCanonicalizesStablePrefixJSON(t *testing.T) { + detector := newBuiltinDetector(true) + first := `{ + "model":"gpt-4o", + "tools":[{"type":"function","function":{"name":"read","parameters":{"type":"object","properties":{"path":{"type":"string"},"line":{"type":"integer"}}}}}], + "messages":[{"role":"user","content":[{"type":"text","text":"open\u0020file"}]}] + }` + reordered := `{"messages":[{"content":[{"text":"open file","type":"text"}],"role":"user"}],"tools":[{"function":{"parameters":{"properties":{"line":{"type":"integer"},"path":{"type":"string"}},"type":"object"},"name":"read"},"type":"function"}],"model":"gpt-4o"}` + + idFirst := detector.Detect(chatSnapshot(nil, first), "team") + idReordered := detector.Detect(chatSnapshot(nil, reordered), "team") + if idFirst == "" || idFirst != idReordered { + t.Fatalf("semantic JSON changes split auto session: %q vs %q", idFirst, idReordered) + } +} + +func TestDetectAutoPreservesArrayOrderAndValues(t *testing.T) { + detector := newBuiltinDetector(true) + base := `{"model":"gpt-4o","tools":[{"type":"function","function":{"name":"first"}},{"type":"function","function":{"name":"second"}}],"messages":[{"role":"user","content":"hello"}]}` + reorderedTools := `{"model":"gpt-4o","tools":[{"type":"function","function":{"name":"second"}},{"type":"function","function":{"name":"first"}}],"messages":[{"role":"user","content":"hello"}]}` + changedValue := `{"model":"gpt-4o","tools":[{"type":"function","function":{"name":"first"}},{"type":"function","function":{"name":"second"}}],"messages":[{"role":"user","content":"hello!"}]}` + + id := detector.Detect(chatSnapshot(nil, base), "") + if id == detector.Detect(chatSnapshot(nil, reorderedTools), "") { + t.Fatal("tool array order must remain part of the session anchor") + } + if id == detector.Detect(chatSnapshot(nil, changedValue), "") { + t.Fatal("changed message value must change the session anchor") + } +} + func TestDetectAutoSystemPromptShape(t *testing.T) { detector := newBuiltinDetector(true) first := `{"model":"gpt-4o","messages":[{"role":"system","content":"be brief"},{"role":"user","content":"opener A"}]}` From a70bca4e336e49b26a1c81f7b29039abb1dbcae9 Mon Sep 17 00:00:00 2001 From: "Jakub A. W" Date: Sat, 8 Aug 2026 16:49:29 +0200 Subject: [PATCH 2/3] fix(cache): address feedback review findings --- ext/ext.go | 22 ++++++---- internal/server/request_rewrite.go | 19 ++++++++- internal/server/request_rewrite_test.go | 28 +++++++++++- internal/server/response_feedback.go | 16 +++++-- internal/server/response_feedback_test.go | 52 +++++++++++++++++++++++ internal/session/detect_test.go | 14 ++++++ 6 files changed, 137 insertions(+), 14 deletions(-) diff --git a/ext/ext.go b/ext/ext.go index 98343c730..ef2bb92ea 100644 --- a/ext/ext.go +++ b/ext/ext.go @@ -82,11 +82,15 @@ type RequestRewriter interface { } // ResponseFeedbackObserver receives content-free feedback after a rewritten -// request successfully reaches a provider. It is an optional companion to -// RequestRewriter: core detects implementations structurally and invokes them -// for both ordinary and streaming responses. usageObserved distinguishes a -// confirmed zero from a provider/stream that returned no usage breakdown. -// Implementations must be safe for concurrent use and return promptly. +// request completes successfully at a provider. For ordinary responses the +// callback runs after the provider returns; for streaming responses it runs +// after the stream closes. Failed requests do not produce feedback. +// +// It is an optional companion to RequestRewriter: core detects implementations +// structurally. usageObserved distinguishes a provider-confirmed zero from a +// provider or stream that returned no usage breakdown. Implementations can be +// called concurrently, must therefore be concurrency-safe, and must return +// promptly so they do not delay completion of the client request. // // The flat signature intentionally uses only long-standing extension types so // extensions can implement the hook while supporting older core releases; @@ -107,9 +111,11 @@ type ResponseFeedbackObserver interface { ) } -// ResponseFeedbackFilter lets an observer decline feedback per request. Core -// calls it after Rewrite and registers the observer only when it returns true. -// Observers without this optional interface receive every successful response. +// ResponseFeedbackFilter lets an observer control registration per request. +// Core calls it after Rewrite and registers the observer only when it returns +// true. Observers without this optional interface receive feedback for every +// successful rewritten request. A panic is isolated and treated as false so +// this optional hook cannot abort inference. type ResponseFeedbackFilter interface { WantsResponseFeedback(in Input, result *Result) bool } diff --git a/internal/server/request_rewrite.go b/internal/server/request_rewrite.go index c9e9e0d9d..e49c653b6 100644 --- a/internal/server/request_rewrite.go +++ b/internal/server/request_rewrite.go @@ -56,7 +56,7 @@ func RequestRewriteMiddleware(rewriters []ext.RequestRewriter, auditLogger audit if observer, ok := rw.(ext.ResponseFeedbackObserver); ok { wantsFeedback := true if filter, filtered := rw.(ext.ResponseFeedbackFilter); filtered { - wantsFeedback = filter.WantsResponseFeedback(in, res) + wantsFeedback = safelyWantsResponseFeedback(rw.Name(), filter, in, res) } if wantsFeedback { feedbackObservers = append(feedbackObservers, observer) @@ -96,6 +96,23 @@ func RequestRewriteMiddleware(rewriters []ext.RequestRewriter, auditLogger audit } } +// safelyWantsResponseFeedback isolates an optional extension hook from the +// inference path. A broken filter declines feedback for this request but must +// never prevent the provider call itself. +func safelyWantsResponseFeedback(name string, filter ext.ResponseFeedbackFilter, in ext.Input, res *ext.Result) (wants bool) { + wants = true + defer func() { + if recovered := recover(); recovered != nil { + wants = false + slog.Warn("response feedback filter panicked; feedback disabled for request", + "rewriter", name, + "panic", recovered, + ) + } + }() + return filter.WantsResponseFeedback(in, res) +} + // redactCredentialHeaders clones the request headers with credential values // (Authorization, cookies, API keys, ...) masked. Rewriters run post-auth and // get UserPath for identity, so they never need raw credentials — and this diff --git a/internal/server/request_rewrite_test.go b/internal/server/request_rewrite_test.go index 9ec1ce3bc..2fc91576c 100644 --- a/internal/server/request_rewrite_test.go +++ b/internal/server/request_rewrite_test.go @@ -31,10 +31,14 @@ type feedbackRewriter struct { type filteredFeedbackRewriter struct { feedbackRewriter - want bool + want bool + panicFilter bool } func (r *filteredFeedbackRewriter) WantsResponseFeedback(ext.Input, *ext.Result) bool { + if r.panicFilter { + panic("feedback filter failed") + } return r.want } @@ -190,6 +194,28 @@ func TestRequestRewriteMiddlewareHonorsResponseFeedbackFilter(t *testing.T) { } } +func TestRequestRewriteMiddlewareIsolatesResponseFeedbackFilterPanic(t *testing.T) { + provider := newRewriteTestProvider() + rewriter := &filteredFeedbackRewriter{ + feedbackRewriter: feedbackRewriter{stubRewriter: stubRewriter{name: "panicking-filter"}}, + panicFilter: true, + } + srv := New(provider, &Config{RequestRewriters: []ext.RequestRewriter{rewriter}}) + + rec := postJSON(t, srv, "/v1/chat/completions", + `{"model":"gpt-4o-mini","messages":[{"role":"user","content":"hello"}]}`) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d (%s), want 200", rec.Code, rec.Body.String()) + } + if provider.capturedChatReq == nil { + t.Fatal("provider was not called after feedback filter panic") + } + if len(rewriter.feedback) != 0 { + t.Fatalf("feedback count = %d, want 0 after filter panic", len(rewriter.feedback)) + } +} + func TestRequestRewriteMiddlewareExposesSessionID(t *testing.T) { tests := []struct { name string diff --git a/internal/server/response_feedback.go b/internal/server/response_feedback.go index f80edaa6f..5ba5865a7 100644 --- a/internal/server/response_feedback.go +++ b/internal/server/response_feedback.go @@ -64,6 +64,7 @@ func notifyResponsesResponseFeedback(c *echo.Context, endpoint ext.Endpoint, res if resp != nil { if resp.Usage != nil { usage = cacheUsageFromCore(resp.Usage.InputTokens, resp.Usage.PromptTokensDetails, resp.Usage.RawUsage) + usage.observed = true } if resp.Model != "" { model = resp.Model @@ -204,17 +205,24 @@ func numericInt(value any) (int, bool) { case int32: return int(typed), true case int64: - return int(typed), true + return parseNumericInt(strconv.FormatInt(typed, 10)) case float64: - return int(typed), true + return parseNumericInt(strconv.FormatFloat(typed, 'f', -1, 64)) case json.Number: - parsed, err := strconv.ParseInt(string(typed), 10, 64) - return int(parsed), err == nil + return parseNumericInt(string(typed)) default: return 0, false } } +func parseNumericInt(value string) (int, bool) { + parsed, err := strconv.Atoi(value) + if err != nil { + return 0, false + } + return parsed, true +} + func nestedMap(value any) (map[string]any, bool) { typed, ok := value.(map[string]any) return typed, ok diff --git a/internal/server/response_feedback_test.go b/internal/server/response_feedback_test.go index 53694c464..ae3b8c21c 100644 --- a/internal/server/response_feedback_test.go +++ b/internal/server/response_feedback_test.go @@ -2,8 +2,13 @@ package server import ( "context" + "encoding/json" + "net/http" + "net/http/httptest" "testing" + "github.com/labstack/echo/v5" + "github.com/enterpilot/gomodel/ext" "github.com/enterpilot/gomodel/internal/core" ) @@ -67,6 +72,53 @@ func TestNotifyChatResponseFeedbackIncludesRouteAndCacheUsage(t *testing.T) { } } +func TestNotifyResponsesResponseFeedbackPreservesObservedZeroUsage(t *testing.T) { + observer := &feedbackCaptureObserver{} + e := echo.New() + req := httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + c := e.NewContext(req, httptest.NewRecorder()) + setResponseFeedbackObservers(c, []ext.ResponseFeedbackObserver{observer}) + + notifyResponsesResponseFeedback( + c, + ext.EndpointResponses, + &core.ResponsesResponse{Usage: &core.ResponsesUsage{}}, + "gpt-5.6", + "openai", + "primary", + ) + + if len(observer.feedback) != 1 { + t.Fatalf("feedback count = %d, want 1", len(observer.feedback)) + } + got := observer.feedback[0] + if got.inputTokens != 0 || got.cacheRead != 0 || got.cacheWrite != 0 || !got.usageObserved { + t.Fatalf("feedback = %+v, want confirmed zero usage", got) + } +} + +func TestNumericIntRejectsInvalidOrOutOfRangeValues(t *testing.T) { + tests := []struct { + name string + value any + want int + ok bool + }{ + {name: "json number", value: json.Number("2048"), want: 2048, ok: true}, + {name: "float integer", value: float64(1536), want: 1536, ok: true}, + {name: "fractional float", value: 1.5, ok: false}, + {name: "json number above int64", value: json.Number("9223372036854775808"), ok: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, ok := numericInt(tt.value) + if got != tt.want || ok != tt.ok { + t.Fatalf("numericInt(%v) = (%d, %v), want (%d, %v)", tt.value, got, ok, tt.want, tt.ok) + } + }) + } +} + func TestResponseFeedbackStreamObserverUsesLatestUsageEvent(t *testing.T) { observer := &feedbackCaptureObserver{} ctx := core.WithRequestID(context.Background(), "req-stream") diff --git a/internal/session/detect_test.go b/internal/session/detect_test.go index 1ea8918a7..365d53e46 100644 --- a/internal/session/detect_test.go +++ b/internal/session/detect_test.go @@ -4,6 +4,8 @@ import ( "strings" "testing" + "github.com/tidwall/gjson" + "github.com/enterpilot/gomodel/internal/core" ) @@ -212,6 +214,18 @@ func TestDetectAutoCanonicalizesStablePrefixJSON(t *testing.T) { } } +func TestCanonicalSegmentFallsBackToExactRawJSON(t *testing.T) { + for _, raw := range []string{ + `{"unterminated":`, + `{"first":1}{"second":2}`, + } { + result := gjson.Result{Type: gjson.JSON, Raw: raw} + if got := string(canonicalSegment(result)); got != raw { + t.Errorf("canonicalSegment(%q) = %q, want exact raw fallback", raw, got) + } + } +} + func TestDetectAutoPreservesArrayOrderAndValues(t *testing.T) { detector := newBuiltinDetector(true) base := `{"model":"gpt-4o","tools":[{"type":"function","function":{"name":"first"}},{"type":"function","function":{"name":"second"}}],"messages":[{"role":"user","content":"hello"}]}` From d77f441afc7a747b3c30b6899a02b8f7efa8798d Mon Sep 17 00:00:00 2001 From: "Jakub A. W" Date: Sat, 8 Aug 2026 16:54:33 +0200 Subject: [PATCH 3/3] fix(cache): avoid logging feedback panic values --- internal/server/request_rewrite.go | 1 - 1 file changed, 1 deletion(-) diff --git a/internal/server/request_rewrite.go b/internal/server/request_rewrite.go index e49c653b6..d5dfd45b3 100644 --- a/internal/server/request_rewrite.go +++ b/internal/server/request_rewrite.go @@ -106,7 +106,6 @@ func safelyWantsResponseFeedback(name string, filter ext.ResponseFeedbackFilter, wants = false slog.Warn("response feedback filter panicked; feedback disabled for request", "rewriter", name, - "panic", recovered, ) } }()