refactor: share SSE parsing across audit and usage observers

AGENTS.md

@@ -21,3 +21,13 @@ Backward compatibility is not a primary constraint in the current development st
 
 1. Make small, focused changes.
 2. Run format/lint/tests relevant to the change.
+
+## Commit Format
+
+Use Conventional Commit format for commit subjects and PR titles:
+
+`type(scope): short summary`
+
+Allowed types: feat, fix, perf, docs, refactor, test, build, ci, chore, revert
+
+Squash merges should preserve the PR title as the resulting commit subject.

CLAUDE.md

@@ -44,6 +44,16 @@ make swagger           # Regenerate Swagger docs
 
 **Build tags:** E2E tests require `-tags=e2e`, integration tests require `-tags=integration`, contract tests require `-tags=contract`. The Makefile handles this automatically.
 
+## Commit And PR Title Format
+
+Use Conventional Commit format for commit subjects and PR titles:
+
+`type(scope): short summary`
+
+Allowed types: feat, fix, perf, docs, refactor, test, build, ci, chore, revert
+
+Prefer squash-and-merge to keep the merged commit subject aligned with the PR title.
+
 ## Error Handling
 
 - All errors returned to clients must be instances of `core.GatewayError`.

internal/auditlog/auditlog_test.go

@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"gomodel/internal/core"
+	"gomodel/internal/streaming"
 	"io"
 	"net/http"
 	"net/http/httptest"
@@ -705,7 +706,7 @@ func TestIsModelInteractionPath(t *testing.T) {
 	}
 }
 
-func TestStreamLogWrapper(t *testing.T) {
+func TestStreamLogObserver(t *testing.T) {
 	// Create a mock stream with content
 	streamContent := `data: {"id":"chatcmpl-123","choices":[{"delta":{"content":"Hello"}}]}
 
@@ -724,7 +725,6 @@ data: [DONE]
 		FlushInterval: 100 * time.Millisecond,
 	}
 	logger := NewLogger(store, cfg)
-	defer logger.Close()
 
 	entry := &LogEntry{
 		ID:        "test-entry",
@@ -733,44 +733,38 @@ data: [DONE]
 		Data:      &LogData{},
 	}
 
-	// Wrap the stream
-	wrapper := NewStreamLogWrapper(stream, logger, entry, "/v1/chat/completions")
+	observedStream := streaming.NewObservedSSEStream(
+		stream,
+		NewStreamLogObserver(logger, entry, "/v1/chat/completions"),
+	)
 
 	// Read all content
 	var buf bytes.Buffer
-	_, err := io.Copy(&buf, wrapper)
+	_, err := io.Copy(&buf, observedStream)
 	if err != nil {
 		t.Fatalf("failed to read stream: %v", err)
 	}
 
-	// Close wrapper to trigger logging
-	if err := wrapper.Close(); err != nil {
-		t.Fatalf("failed to close wrapper: %v", err)
+	// Close stream to trigger logging
+	if err := observedStream.Close(); err != nil {
+		t.Fatalf("failed to close stream: %v", err)
+	}
+	if err := logger.Close(); err != nil {
+		t.Fatalf("failed to close logger: %v", err)
 	}
-
-	// Wait for async write
-	time.Sleep(200 * time.Millisecond)
 
 	// Verify entry was logged
 	if len(store.getEntries()) != 1 {
 		t.Errorf("expected 1 entry, got %d", len(store.getEntries()))
 	}
 }
 
-func TestWrapStreamForLogging(t *testing.T) {
-	stream := io.NopCloser(strings.NewReader("test"))
-
-	// Test with nil logger
-	result := WrapStreamForLogging(stream, nil, nil, "/v1/chat/completions")
-	if result != stream {
-		t.Error("expected original stream with nil logger")
+func TestNewStreamLogObserverNilInputs(t *testing.T) {
+	if observer := NewStreamLogObserver(nil, &LogEntry{}, "/v1/chat/completions"); observer != nil {
+		t.Error("expected nil observer with nil logger")
 	}
-
-	// Test with disabled logger
-	noopLogger := &NoopLogger{}
-	result = WrapStreamForLogging(stream, noopLogger, &LogEntry{}, "/v1/chat/completions")
-	if result != stream {
-		t.Error("expected original stream with disabled logger")
+	if observer := NewStreamLogObserver(&NoopLogger{}, nil, "/v1/chat/completions"); observer != nil {
+		t.Error("expected nil observer with nil entry")
 	}
 }
 

internal/auditlog/constants.go

@@ -7,7 +7,7 @@ const (
 	MaxBodyCapture = 1024 * 1024
 
 	// MaxContentCapture is the maximum size of accumulated streaming content (1MB).
-	// Used by StreamLogWrapper to limit reconstructed response body size.
+	// Used by the stream observer to limit reconstructed response body size.
 	MaxContentCapture = 1024 * 1024
 
 	// BatchFlushThreshold is the number of entries that triggers an immediate flush.
@@ -27,6 +27,7 @@ const (
 	LogEntryKey contextKey = "auditlog_entry"
 
 	// LogEntryStreamingKey is the context key for marking a request as streaming.
-	// When true, the middleware skips logging (StreamLogWrapper handles it instead).
+	// When true, the middleware skips logging because the stream observer path
+	// handles streaming audit logging.
 	LogEntryStreamingKey contextKey = "auditlog_entry_streaming"
 )

internal/auditlog/middleware.go

@@ -149,7 +149,7 @@ func Middleware(logger LoggerInterface) echo.MiddlewareFunc {
 				}
 			}
 
-			// Write log entry asynchronously (skip if streaming - StreamLogWrapper handles it)
+			// Write log entry asynchronously (skip if streaming - the stream observer path handles it)
 			if !IsEntryMarkedAsStreaming(c) {
 				logger.Write(entry)
 			}
@@ -262,7 +262,7 @@ type responseBodyCapture struct {
 	body      *bytes.Buffer
 	truncated bool
 	// shouldCapture allows middleware to stop buffering once the request is
-	// known to be streaming. Streaming responses are handled by StreamLogWrapper.
+	// known to be streaming. Streaming responses are handled by the stream observer path.
 	shouldCapture func() bool
 }
 

internal/auditlog/stream_observer.go

@@ -0,0 +1,152 @@
+package auditlog
+
+import (
+	"strings"
+	"time"
+)
+
+// StreamLogObserver reconstructs stream metadata and optional response bodies
+// from parsed SSE JSON payloads.
+type StreamLogObserver struct {
+	logger    LoggerInterface
+	entry     *LogEntry
+	builder   *streamResponseBuilder
+	logBodies bool
+	closed    bool
+	startTime time.Time
+}
+
+func NewStreamLogObserver(logger LoggerInterface, entry *LogEntry, path string) *StreamLogObserver {
+	if logger == nil || entry == nil {
+		return nil
+	}
+
+	logBodies := logger.Config().LogBodies
+	var builder *streamResponseBuilder
+	if logBodies {
+		builder = &streamResponseBuilder{
+			IsResponsesAPI: strings.HasPrefix(path, "/v1/responses"),
+		}
+	}
+
+	return &StreamLogObserver{
+		logger:    logger,
+		entry:     entry,
+		builder:   builder,
+		logBodies: logBodies,
+		startTime: entry.Timestamp,
+	}
+}
+
+func (o *StreamLogObserver) OnJSONEvent(event map[string]interface{}) {
+	if !o.logBodies || o.builder == nil {
+		return
+	}
+	if o.builder.IsResponsesAPI {
+		o.parseResponsesAPIEvent(event)
+		return
+	}
+	o.parseChatCompletionEvent(event)
+}
+
+func (o *StreamLogObserver) OnStreamClose() {
+	if o.closed {
+		return
+	}
+	o.closed = true
+
+	if o.entry != nil && !o.startTime.IsZero() {
+		o.entry.DurationNs = time.Since(o.startTime).Nanoseconds()
+	}
+
+	if o.logBodies && o.builder != nil && o.entry != nil && o.entry.Data != nil {
+		if o.builder.IsResponsesAPI {
+			o.entry.Data.ResponseBody = o.builder.buildResponsesAPIResponse()
+		} else {
+			o.entry.Data.ResponseBody = o.builder.buildChatCompletionResponse()
+		}
+		o.entry.Data.ResponseBodyTooBigToHandle = o.builder.truncated
+	}
+
+	if o.logger != nil && o.entry != nil {
+		o.logger.Write(o.entry)
+	}
+}
+
+func (o *StreamLogObserver) parseChatCompletionEvent(event map[string]interface{}) {
+	if o.builder == nil {
+		return
+	}
+
+	if o.builder.ID == "" {
+		if id, ok := event["id"].(string); ok {
+			o.builder.ID = id
+		}
+		if model, ok := event["model"].(string); ok {
+			o.builder.Model = model
+		}
+		if created, ok := event["created"].(float64); ok {
+			o.builder.Created = int64(created)
+		}
+	}
+
+	if choices, ok := event["choices"].([]interface{}); ok && len(choices) > 0 {
+		if choice, ok := choices[0].(map[string]interface{}); ok {
+			if fr, ok := choice["finish_reason"].(string); ok && fr != "" {
+				o.builder.FinishReason = fr
+			}
+
+			if delta, ok := choice["delta"].(map[string]interface{}); ok {
+				if role, ok := delta["role"].(string); ok {
+					o.builder.Role = role
+				}
+				if content, ok := delta["content"].(string); ok && content != "" {
+					o.appendContent(content)
+				}
+			}
+		}
+	}
+}
+
+func (o *StreamLogObserver) parseResponsesAPIEvent(event map[string]interface{}) {
+	if o.builder == nil {
+		return
+	}
+
+	eventType, _ := event["type"].(string)
+	switch eventType {
+	case "response.created", "response.completed", "response.done":
+		if resp, ok := event["response"].(map[string]interface{}); ok {
+			if id, ok := resp["id"].(string); ok {
+				o.builder.ResponseID = id
+			}
+			if status, ok := resp["status"].(string); ok {
+				o.builder.Status = status
+			}
+			if model, ok := resp["model"].(string); ok {
+				o.builder.Model = model
+			}
+			if createdAt, ok := resp["created_at"].(float64); ok {
+				o.builder.CreatedAt = int64(createdAt)
+			}
+		}
+	case "response.output_text.delta":
+		if delta, ok := event["delta"].(string); ok && delta != "" {
+			o.appendContent(delta)
+		}
+	}
+}
+
+func (o *StreamLogObserver) appendContent(content string) {
+	if o.builder == nil || o.builder.truncated || o.builder.contentLen >= MaxContentCapture {
+		return
+	}
+
+	remaining := MaxContentCapture - o.builder.contentLen
+	if len(content) > remaining {
+		content = content[:remaining]
+		o.builder.truncated = true
+	}
+	o.builder.Content.WriteString(content)
+	o.builder.contentLen += len(content)
+}