Merge pull request #9 from matdev83/codex/fix-stream-race-ci

Fix race detector CI failures

Author: matdev83

.github/workflows/qa.yml

@@ -1,7 +1,6 @@
 name: QA
 
 on:
-  push:
   pull_request:
 
 jobs:

.golangci.yml

@@ -35,6 +35,3 @@ linters:
 formatters:
   enable:
     - gofumpt
-
-issues:
-  exclude-use-default: false

internal/core/auth/events.go

@@ -2,6 +2,7 @@ package auth
 
 import (
 	"context"
+	"sync"
 
 	sdkauth "github.com/matdev83/go-llm-interactive-proxy/pkg/lipsdk/auth"
 )
@@ -12,6 +13,7 @@ import (
 // Event DTO additions remain subject to non-secret classification; see [EventSink] and
 // [sdkauth.AuthDecisionEvent] / [sdkauth.SessionStartEvent] package docs.
 type EventDispatcher struct {
+	mu     sync.Mutex
 	sink   EventSink
 	policy EventFailurePolicy
 }
@@ -29,6 +31,8 @@ func (d *EventDispatcher) DispatchAuthDecision(ctx context.Context, ev sdkauth.A
 	}
 	ev2 := ev
 	ev2.ChallengeSummary = sdkauth.SanitizePublicChallengeSummary(ev.ChallengeSummary, "", sdkauth.PublicChallengeSummaryMaxRunes)
+	d.mu.Lock()
+	defer d.mu.Unlock()
 	err := d.sink.OnAuthDecision(ctx, ev2)
 	return d.handleSinkError(err)
 }
@@ -38,6 +42,8 @@ func (d *EventDispatcher) DispatchSessionStart(ctx context.Context, ev sdkauth.S
 	if d == nil || d.sink == nil {
 		return nil
 	}
+	d.mu.Lock()
+	defer d.mu.Unlock()
 	err := d.sink.OnSessionStart(ctx, ev)
 	return d.handleSinkError(err)
 }

internal/core/auth/events_test.go

@@ -3,6 +3,7 @@ package auth
 import (
 	"context"
 	"errors"
+	"sync"
 	"testing"
 	"time"
 
@@ -98,6 +99,30 @@ func TestEventDispatcher_failClosed_propagatesSinkError(t *testing.T) {
 	}
 }
 
+func TestEventDispatcher_serializesConcurrentSinkDelivery(t *testing.T) {
+	t.Parallel()
+	const calls = 64
+	sink := &fakeSink{}
+	d := NewEventDispatcher(sink, EventFailureBestEffort)
+	ctx := context.Background()
+
+	var wg sync.WaitGroup
+	wg.Add(calls)
+	for range calls {
+		go func() {
+			defer wg.Done()
+			if err := d.DispatchAuthDecision(ctx, sampleAuthEvent()); err != nil {
+				t.Errorf("DispatchAuthDecision: %v", err)
+			}
+		}()
+	}
+	wg.Wait()
+
+	if sink.authCalls != calls {
+		t.Fatalf("auth calls: got %d want %d", sink.authCalls, calls)
+	}
+}
+
 func sampleAuthEvent() sdkauth.AuthDecisionEvent {
 	return sdkauth.AuthDecisionEvent{
 		Time:       time.Unix(1700000000, 0).UTC(),

internal/core/auth/ports.go

@@ -45,7 +45,9 @@ const (
 )
 
 // EventSink receives non-secret auth and session events. Implementations are wired at the
-// composition root (e.g. structured logging). OnAuthDecision: do not treat [sdkauth.AuthDecisionEvent]
+// composition root (e.g. structured logging). [EventDispatcher] serializes calls per dispatcher
+// instance; sinks should still avoid long blocking work because delivery remains on request paths.
+// OnAuthDecision: do not treat [sdkauth.AuthDecisionEvent]
 // fields as proof of absence of secrets in upstream state; log only stable, operator-approved
 // attributes (the default JSON sink logs [sdkauth.AuthDecisionEvent.PrincipalSafeClaims] keys only,
 // not map values). New fields on event DTOs require explicit non-secret data classification before use;

internal/core/modelcatalog/eligibility.go

@@ -11,8 +11,8 @@ import (
 type EligibilityDecision struct {
 	IsEligible bool              `json:"eligible"`
 	Reason     EligibilityReason `json:"reason,omitempty"`
-	Facts      EffectiveFacts    `json:"facts,omitempty"`
-	Estimate   SizeEstimate      `json:"estimate,omitempty"`
+	Facts      EffectiveFacts    `json:"facts,omitzero"`
+	Estimate   SizeEstimate      `json:"estimate,omitzero"`
 }
 
 // EligibilityResolverImpl decides context-limit eligibility from already-resolved [EffectiveFacts]

internal/core/runtime/executor.go

@@ -77,6 +77,8 @@ type Executor struct {
 	ExtensionMetrics extensions.StageMetrics
 	// CompletionBufferLimits overrides completion-gate buffering bounds (tests). Zero MaxEvents uses SDK defaults.
 	CompletionBufferLimits completion.BufferLimits
+	// secureSessionMu guards lazy initialization of SecureSession in the test hook path.
+	secureSessionMu sync.Mutex
 
 	// SecureSession authorizes turns via BeginTurn before submit hooks; required for all executor prepares.
 	SecureSession *app.Manager
@@ -146,6 +148,8 @@ func (e *Executor) Execute(ctx context.Context, call *lipapi.Call) (_ lipapi.Eve
 	if e.RuntimeSnapshot != nil {
 		ctx = extensions.WithRequestRuntimeSnapshot(ctx, e.RuntimeSnapshot)
 	}
+	e.secureSessionMu.Lock()
+	defer e.secureSessionMu.Unlock()
 	if e.SecureSession == nil {
 		secureSessionTestPrepare(e)
 	}

internal/core/stream/recv_context_contract_test.go

@@ -16,6 +16,7 @@ type sdkLikeRecvBlocker struct {
 	mu          sync.Mutex
 	ch          chan struct{}
 	recvStarted chan struct{}
+	startedOnce sync.Once
 }
 
 func newSDKLikeRecvBlocker() *sdkLikeRecvBlocker {
@@ -29,11 +30,14 @@ func (s *sdkLikeRecvBlocker) Recv(ctx context.Context) (lipapi.Event, error) {
 	if ctx == nil {
 		return lipapi.Event{}, lipapi.ErrNilContext
 	}
-	select {
-	case s.recvStarted <- struct{}{}:
-	default:
+	s.startedOnce.Do(func() { close(s.recvStarted) })
+	s.mu.Lock()
+	ch := s.ch
+	s.mu.Unlock()
+	if ch == nil {
+		return lipapi.Event{}, context.Canceled
 	}
-	<-s.ch
+	<-ch
 	return lipapi.Event{}, context.Canceled
 }
 

internal/stdhttp/modelcatalog_diag_test.go

@@ -158,7 +158,7 @@ func TestModelCatalogDiagnostics_mount_stackHTTPHandler_wiring(t *testing.T) {
 	}
 	built := &runtimebundle.Built{CatalogRuntime: rt}
 	mux := http.NewServeMux()
-	mountModelCatalogDiagnostics(mux, cfg, testkit.DiscardLogger(), built, context.Background())
+	mountModelCatalogDiagnostics(context.Background(), mux, cfg, testkit.DiscardLogger(), built)
 	outer := stackHTTPHandler(stackHTTPInput{
 		Cfg: cfg, Log: testkit.DiscardLogger(), Built: built, TraceGen: diag.NewTraceIDGenerator(), Inner: mux, HTTPProm: nil,
 	})

internal/stdhttp/server.go

@@ -235,7 +235,7 @@ func RunWithRuntime(
 		mux.Handle("GET "+base, dh)
 		log.InfoContext(logCtx, "secure-session diagnostics mounted", "path", base)
 	}
-	mountModelCatalogDiagnostics(mux, cfg, log, built, logCtx)
+	mountModelCatalogDiagnostics(logCtx, mux, cfg, log, built)
 	maxBody := cfg.Server.EffectiveMaxRequestBodyBytes()
 	var trafficPorts traffic.PortBundle
 	if built.RuntimeSnapshot != nil {
@@ -327,7 +327,7 @@ func RunWithRuntime(
 	}
 }
 
-func mountModelCatalogDiagnostics(mux *http.ServeMux, cfg *config.Config, log *slog.Logger, built *runtimebundle.Built, logCtx context.Context) {
+func mountModelCatalogDiagnostics(logCtx context.Context, mux *http.ServeMux, cfg *config.Config, log *slog.Logger, built *runtimebundle.Built) {
 	if mux == nil || cfg == nil {
 		return
 	}