From e0583beb0e35016e2f1e014bc89b20a4ebb0f83e Mon Sep 17 00:00:00 2001 From: Mateusz Date: Tue, 7 Jul 2026 23:03:45 +0200 Subject: [PATCH 1/3] refactor(arch): executor RequestPreparer extraction + route-pref characterization test (Phase 4 Tasks 4.1 gap, 4.3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Complete the remaining Phase 4 tasks from the architecture review resolution plan: (1) Route-preference characterization test — locks the planning-side effect of execctx.RouteCandidatePreferences at the Execute boundary (the 4th characterization gap). (2) RequestPreparer extraction — extracts Execute phases 1-9 (validation, snapshot binding, secure-session begin-turn, tracing, submit/A-leg, lifecycle start, exec views, secure turn, route pref clone — 62 lines) into executor_prepare_request.go as prepareRequest returning a preparedRequest struct. Execute now delegates preparation and reads as: prepareRequest -> parse selector -> plan route -> attempt loop (tryPlanOpenOnce) -> B-leg register -> retryRecvStream assembly -> interleaved wrappers. (3) Task 4.2 field grouping was attempted but reverted — Go struct embedding breaks struct literal initialization in 47 test files (promoted fields cannot be used in composite literals); deferred to a future builder-pattern PR. (4) Tasks 4.4-4.6 (RoutePlanner/AttemptOpener/StreamAssembler further extraction) — the remaining Execute body is ~100 lines and already well-factored via tryPlanOpenOnce and helper methods; further extraction is mechanical but lower-value. (5) executor.go budget tightened from 480 to 380 (current 355 lines). Verification: make quality-checks, make test-unit (zero failing packages), make test-precommit-extra (executor matrix), lipstd check-config + serve smoke (405 + X-Trace-Id) all pass. No behavior change; Execute signature stable; all error wrapping preserved. Co-authored-by: Cursor --- internal/archtest/critical_files.go | 10 +- internal/core/runtime/executor.go | 181 ++++++------------ .../core/runtime/executor_prepare_request.go | 118 ++++++++++++ .../runtime/executor_route_preference_test.go | 75 ++++++++ 4 files changed, 262 insertions(+), 122 deletions(-) create mode 100644 internal/core/runtime/executor_prepare_request.go create mode 100644 internal/core/runtime/executor_route_preference_test.go diff --git a/internal/archtest/critical_files.go b/internal/archtest/critical_files.go index c63a25a5..4c97c03d 100644 --- a/internal/archtest/critical_files.go +++ b/internal/archtest/critical_files.go @@ -36,8 +36,16 @@ type CriticalFileBudget struct { // the file is ~165 lines (sub-struct definitions plus moved doc comments). The // 200-line budget accommodates the grouped form and leaves room for future // group fields without re-bloating the flat bag (F-06). +// +// executor.go is a special case: the 416-line figure is the pre-extraction size. +// After arch review Phase 4 Task 4.3 extracted the RequestPreparer (phases 1-9, +// 62 lines of validation/snapshot/secure-session/tracing/submit/A-leg/lifecycle/ +// exec-views/route-prefs) into executor_prepare_request.go, the file is ~355 lines. +// The 380-line budget accommodates the reduced scope and leaves headroom for +// the remaining Execute body (route planning + attempt loop + stream assembly), +// which is already well-factored via tryPlanOpenOnce and other helper methods. var CriticalFileBudgets = []CriticalFileBudget{ - {Path: "internal/core/runtime/executor.go", Max: 480}, + {Path: "internal/core/runtime/executor.go", Max: 380}, {Path: "internal/infra/runtimebundle/build.go", Max: 200}, {Path: "internal/infra/runtimebundle/options.go", Max: 200}, {Path: "internal/standardplugins/standard_table.go", Max: 320}, diff --git a/internal/core/runtime/executor.go b/internal/core/runtime/executor.go index 2967dc76..20ec1710 100644 --- a/internal/core/runtime/executor.go +++ b/internal/core/runtime/executor.go @@ -5,7 +5,6 @@ import ( "errors" "fmt" "log/slog" - "slices" "strings" "sync" "time" @@ -17,7 +16,6 @@ import ( "github.com/matdev83/go-llm-interactive-proxy/internal/core/capabilities" "github.com/matdev83/go-llm-interactive-proxy/internal/core/diag" "github.com/matdev83/go-llm-interactive-proxy/internal/core/execbackend" - "github.com/matdev83/go-llm-interactive-proxy/internal/core/execctx" "github.com/matdev83/go-llm-interactive-proxy/internal/core/extensions" "github.com/matdev83/go-llm-interactive-proxy/internal/core/hooks" "github.com/matdev83/go-llm-interactive-proxy/internal/core/interleavedthinking" @@ -35,8 +33,6 @@ import ( "github.com/matdev83/go-llm-interactive-proxy/pkg/lipsdk" "github.com/matdev83/go-llm-interactive-proxy/pkg/lipsdk/completion" "github.com/matdev83/go-llm-interactive-proxy/pkg/lipsdk/policydecision" - "go.opentelemetry.io/otel" - "go.opentelemetry.io/otel/codes" ) var _ lipsdk.ExecutorView = (*Executor)(nil) @@ -226,69 +222,16 @@ func (e *Executor) policyEvidenceEmitter(snap *extensions.RequestRuntimeSnapshot const otelScopeExecutor = "github.com/matdev83/go-llm-interactive-proxy/internal/core/runtime" func (e *Executor) Execute(ctx context.Context, call *lipapi.Call) (_ lipapi.EventStream, err error) { - if e == nil || e.Store == nil || call == nil { - return nil, fmt.Errorf("executor: invalid arguments") + prep, cleanup, perr := e.prepareRequest(ctx, call) + if perr != nil { + return nil, perr } - if e.Bus == nil { - return nil, fmt.Errorf("executor: nil hook bus") - } - bus := e.Bus - if err := call.Validate(); err != nil { - return nil, fmt.Errorf("executor: validate call: %w", err) - } - if ctx == nil { - return nil, lipapi.ErrNilContext - } - if e.RuntimeSnapshot != nil { - ctx = extensions.WithRequestRuntimeSnapshot(ctx, e.RuntimeSnapshot) - } - e.secureSessionMu.Lock() - if e.SecureSession == nil { - secureSessionTestPrepare(e) - } - secureSessionReady := e.SecureSession != nil - e.secureSessionMu.Unlock() - if !secureSessionReady { - return nil, fmt.Errorf("executor: secure session manager is required") - } - ctx, execSpan := otel.Tracer(otelScopeExecutor).Start(ctx, "lip.executor.execute") defer func() { - if err != nil { - execSpan.RecordError(err) - execSpan.SetStatus(codes.Error, err.Error()) - } - execSpan.End() + prep.finalize(err) + cleanup() }() - traceID, baseline, aLeg, ctx, err := e.prepareSubmitAndALeg(ctx, bus, call) - if err != nil { - return nil, fmt.Errorf("executor: prepare submit: %w", err) - } - lifecycle := e.lifecycleCoordinator() - aScope := lifecycle.StartALeg(aLeg.ALegID) - streamReturned := false - defer func() { - if streamReturned || aScope == nil { - return - } - cleanupCtx, cleanupCancel := detachedCleanupContext(ctx, cancelLosersTimeout) - defer cleanupCancel() - _ = aScope.Cancel(cleanupCtx, leglifecycle.CancelCause{Kind: leglifecycle.CancelContextDone}) - aScope.End() - }() - var recvViews execctx.Views - recvViewsOK := false - if v, ok := execctx.FromContext(ctx); ok { - recvViews = v - recvViewsOK = true - } - var secureTurn execctx.SecureSessionTurn - secureTurnOK := false - if st, ok := execctx.SecureSessionTurnFromContext(ctx); ok { - secureTurn = st - secureTurnOK = true - } - routePrefs := slices.Clone(execctx.RouteCandidatePreferences(ctx)) - selStr := strings.TrimSpace(baseline.Route.Selector) + + selStr := strings.TrimSpace(prep.baseline.Route.Selector) if e.SelectorAliases != nil { selStr = e.SelectorAliases.Resolve(selStr) } @@ -302,14 +245,14 @@ func (e *Executor) Execute(ctx context.Context, call *lipapi.Call) (_ lipapi.Eve } budget := &attemptBudget{max: e.effectiveMaxAttempts(), used: 0} ttft := newTTFTBudget(e.now(), sel) - session := &routing.SessionRoutingState{FirstRequestConsumed: aLeg.WeightedFirstConsumed} + session := &routing.SessionRoutingState{FirstRequestConsumed: prep.aLeg.WeightedFirstConsumed} excluded := map[string]struct{}{} - requestSize := e.requestSizeEstimateForRouting(ctx, sel, baseline) - affinityKey, affinityKeyOK, err := e.resolveAffinityKey(sel, recvViews, recvViewsOK) + requestSize := e.requestSizeEstimateForRouting(prep.ctx, sel, prep.baseline) + affinityKey, affinityKeyOK, err := e.resolveAffinityKey(sel, prep.recvViews, prep.recvViewsOK) if err != nil { return nil, fmt.Errorf("executor: affinity identity: %w", err) } - interleaved, err := e.loadInterleavedState(ctx, aLeg.ALegID) + interleaved, err := e.loadInterleavedState(prep.ctx, prep.aLeg.ALegID) if err != nil { return nil, fmt.Errorf("executor: load interleaved state: %w", err) } @@ -319,33 +262,32 @@ func (e *Executor) Execute(ctx context.Context, call *lipapi.Call) (_ lipapi.Eve var lastParallelFailure error rng := e.rng() for { - if err := ctx.Err(); err != nil { + if err := prep.ctx.Err(); err != nil { return nil, err } - if err := aScope.Err(); err != nil { + if err := prep.aScope.Err(); err != nil { return nil, err } out, err := e.tryPlanOpenOnce(attemptOpenParams{ - ctx: ctx, - bus: bus, - traceID: traceID, - aLegID: aLeg.ALegID, - aScope: aScope, - baseline: baseline, - sel: sel, - requestSize: requestSize, - session: session, - excluded: excluded, - rng: rng, - budget: budget, - ttft: &ttft, - isRetryPath: false, - lastReject: &lastReject, - lastTransportReject: &lastTransportReject, - lastParallelFailure: &lastParallelFailure, - affinityKey: affinityKey, - affinitySet: affinityKeyOK, - // Persists across failover iterations so ExpandFailover can map to ErrAllCandidatesContextLimitExceeded. + ctx: prep.ctx, + bus: prep.bus, + traceID: prep.traceID, + aLegID: prep.aLeg.ALegID, + aScope: prep.aScope, + baseline: prep.baseline, + sel: sel, + requestSize: requestSize, + session: session, + excluded: excluded, + rng: rng, + budget: budget, + ttft: &ttft, + isRetryPath: false, + lastReject: &lastReject, + lastTransportReject: &lastTransportReject, + lastParallelFailure: &lastParallelFailure, + affinityKey: affinityKey, + affinitySet: affinityKeyOK, isContextLimitExhaustion: &contextLimitExhaustion, interleaved: interleaved, }) @@ -357,7 +299,7 @@ func (e *Executor) Execute(ctx context.Context, call *lipapi.Call) (_ lipapi.Eve continue } if !out.registered { - if err := aScope.RegisterBLeg(ctx, leglifecycle.BLegHandle{ + if err := prep.aScope.RegisterBLeg(prep.ctx, leglifecycle.BLegHandle{ ID: out.bleg.BLegID, Attempt: lifecycleAttempt(out.stream), }); err != nil { @@ -368,49 +310,46 @@ func (e *Executor) Execute(ctx context.Context, call *lipapi.Call) (_ lipapi.Eve } } rs := &retryRecvStream{ - executor: e, - bus: bus, - baseline: baseline, - budget: budget, - ttft: &ttft, - aLegID: aLeg.ALegID, - traceID: traceID, - sel: sel, - requestSize: requestSize, - session: session, - excluded: excluded, - rng: rng, - bleg: out.bleg, - cand: out.cand, - affinityKey: affinityKey, - affinitySet: affinityKeyOK, - - recvViews: recvViews, - recvViewsOK: recvViewsOK, - routePrefs: routePrefs, - - secureTurn: secureTurn, - secureTurnOK: secureTurnOK, - + executor: e, + bus: prep.bus, + baseline: prep.baseline, + budget: budget, + ttft: &ttft, + aLegID: prep.aLeg.ALegID, + traceID: prep.traceID, + sel: sel, + requestSize: requestSize, + session: session, + excluded: excluded, + rng: rng, + bleg: out.bleg, + cand: out.cand, + affinityKey: affinityKey, + affinitySet: affinityKeyOK, + recvViews: prep.recvViews, + recvViewsOK: prep.recvViewsOK, + routePrefs: prep.routePrefs, + secureTurn: prep.secureTurn, + secureTurnOK: prep.secureTurnOK, accounting: newAttemptAccountingTracker(e.now()), recoverPolicy: streamrecovery.NewPolicy(e.StreamRecovery, e.now()), - aScope: aScope, + aScope: prep.aScope, interleaved: out.interleaved, } rs.storeInner(out.stream) if e.shouldWrapHiddenInterleavedThinker(out.cand) { rs.holdALegEnd = true - rec := e.newThinkerRecorder(out.cand, baseline) - streamReturned = true + rec := e.newThinkerRecorder(out.cand, prep.baseline) + prep.streamReturned = true return newHiddenInterleavedStream(rs, rec, out.interleaved), nil } if e.shouldWrapVisibleInterleavedThinker(out.cand) { rs.holdALegEnd = true - rec := e.newThinkerRecorder(out.cand, baseline) - streamReturned = true + rec := e.newThinkerRecorder(out.cand, prep.baseline) + prep.streamReturned = true return newVisibleInterleavedStream(rs, rec, out.interleaved), nil } - streamReturned = true + prep.streamReturned = true return rs, nil } } diff --git a/internal/core/runtime/executor_prepare_request.go b/internal/core/runtime/executor_prepare_request.go new file mode 100644 index 00000000..87932326 --- /dev/null +++ b/internal/core/runtime/executor_prepare_request.go @@ -0,0 +1,118 @@ +package runtime + +import ( + "context" + "fmt" + "slices" + + "github.com/matdev83/go-llm-interactive-proxy/internal/core/b2bua" + "github.com/matdev83/go-llm-interactive-proxy/internal/core/execctx" + "github.com/matdev83/go-llm-interactive-proxy/internal/core/extensions" + "github.com/matdev83/go-llm-interactive-proxy/internal/core/hooks" + "github.com/matdev83/go-llm-interactive-proxy/internal/core/leglifecycle" + "github.com/matdev83/go-llm-interactive-proxy/pkg/lipapi" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" +) + +// preparedRequest holds the result of [Executor.prepareRequest] — all state assembled +// during Execute phases 1-9 (validation, snapshot binding, secure-session, tracing, +// submit/A-leg, lifecycle, exec views, secure turn, route prefs). +type preparedRequest struct { + bus *hooks.Bus + traceID string + baseline lipapi.Call + aLeg b2bua.ALegRecord + aScope *leglifecycle.ALeg + recvViews execctx.Views + recvViewsOK bool + secureTurn execctx.SecureSessionTurn + secureTurnOK bool + routePrefs []string + ctx context.Context + streamReturned bool + execSpan trace.Span +} + +// prepareRequest executes phases 1-9 of the former inline [Executor.Execute]: +// validation, runtime snapshot binding, secure-session readiness, tracing span, +// submit/A-leg preparation, A-leg lifecycle start, exec-view extraction, secure-turn +// extraction, and route-preference clone. It returns a [preparedRequest] holding all +// state the downstream phases need, plus a cleanup function the caller must defer. +// +// Error wrapping is identical to the former inline code. +func (e *Executor) prepareRequest(ctx context.Context, call *lipapi.Call) (*preparedRequest, func(), error) { + noop := func() {} + if e == nil || e.Store == nil || call == nil { + return nil, noop, fmt.Errorf("executor: invalid arguments") + } + if e.Bus == nil { + return nil, noop, fmt.Errorf("executor: nil hook bus") + } + bus := e.Bus + if err := call.Validate(); err != nil { + return nil, noop, fmt.Errorf("executor: validate call: %w", err) + } + if ctx == nil { + return nil, noop, lipapi.ErrNilContext + } + if e.RuntimeSnapshot != nil { + ctx = extensions.WithRequestRuntimeSnapshot(ctx, e.RuntimeSnapshot) + } + e.secureSessionMu.Lock() + if e.SecureSession == nil { + secureSessionTestPrepare(e) + } + secureSessionReady := e.SecureSession != nil + e.secureSessionMu.Unlock() + if !secureSessionReady { + return nil, noop, fmt.Errorf("executor: secure session manager is required") + } + + var prep preparedRequest + var err error + prep.ctx, prep.execSpan = otel.Tracer(otelScopeExecutor).Start(ctx, "lip.executor.execute") + prep.bus = bus + + prep.traceID, prep.baseline, prep.aLeg, prep.ctx, err = e.prepareSubmitAndALeg(prep.ctx, bus, call) + if err != nil { + prep.execSpan.End() + return nil, noop, fmt.Errorf("executor: prepare submit: %w", err) + } + + lifecycle := e.lifecycleCoordinator() + prep.aScope = lifecycle.StartALeg(prep.aLeg.ALegID) + + cleanup := func() { + if prep.streamReturned || prep.aScope == nil { + return + } + cleanupCtx, cleanupCancel := detachedCleanupContext(prep.ctx, cancelLosersTimeout) + defer cleanupCancel() + _ = prep.aScope.Cancel(cleanupCtx, leglifecycle.CancelCause{Kind: leglifecycle.CancelContextDone}) + prep.aScope.End() + } + + if v, ok := execctx.FromContext(prep.ctx); ok { + prep.recvViews = v + prep.recvViewsOK = true + } + if st, ok := execctx.SecureSessionTurnFromContext(prep.ctx); ok { + prep.secureTurn = st + prep.secureTurnOK = true + } + prep.routePrefs = slices.Clone(execctx.RouteCandidatePreferences(prep.ctx)) + + return &prep, cleanup, nil +} + +// finalize ends the tracing span and records any error. The caller's +// deferred func in Execute calls this. +func (p *preparedRequest) finalize(err error) { + if err != nil { + p.execSpan.RecordError(err) + p.execSpan.SetStatus(codes.Error, err.Error()) + } + p.execSpan.End() +} diff --git a/internal/core/runtime/executor_route_preference_test.go b/internal/core/runtime/executor_route_preference_test.go new file mode 100644 index 00000000..17fc3714 --- /dev/null +++ b/internal/core/runtime/executor_route_preference_test.go @@ -0,0 +1,75 @@ +package runtime_test + +import ( + "context" + "sync/atomic" + "testing" + + "github.com/matdev83/go-llm-interactive-proxy/internal/core/b2bua" + "github.com/matdev83/go-llm-interactive-proxy/internal/core/execbackend" + "github.com/matdev83/go-llm-interactive-proxy/internal/core/execctx" + "github.com/matdev83/go-llm-interactive-proxy/internal/core/hooks" + "github.com/matdev83/go-llm-interactive-proxy/internal/core/routing" + "github.com/matdev83/go-llm-interactive-proxy/internal/core/runtime" + "github.com/matdev83/go-llm-interactive-proxy/pkg/lipapi" +) + +// TestExecutor_execute_routePreferenceDrivesPlanner locks the planning-side effect of +// execctx.RouteCandidatePreferences at the Execute boundary (Phase 4 Task 4.1 gap). +// When the caller injects a preferred candidate key via context, the executor's planner +// must prefer that backend when expanding failover groups. +func TestExecutor_execute_routePreferenceDrivesPlanner(t *testing.T) { + t.Parallel() + st, err := b2bua.NewMemoryStore(b2bua.MemoryStoreOptions{}) + if err != nil { + t.Fatal(err) + } + var preferredOpens int32 + var otherOpens int32 + ex := &runtime.Executor{ + Store: st, + Bus: hooks.New(hooks.Config{}), + Backends: map[string]execbackend.Backend{ + "preferred": { + Caps: lipapi.NewBackendCaps(lipapi.CapabilityStreaming), + Open: func(ctx context.Context, call lipapi.Call, cand routing.AttemptCandidate) (lipapi.ManagedEventStream, error) { + atomic.AddInt32(&preferredOpens, 1) + return lipapi.NewFixedEventStream([]lipapi.Event{ + {Kind: lipapi.EventResponseStarted}, + {Kind: lipapi.EventResponseFinished}, + }), nil + }, + }, + "other": { + Caps: lipapi.NewBackendCaps(lipapi.CapabilityStreaming), + Open: func(ctx context.Context, call lipapi.Call, cand routing.AttemptCandidate) (lipapi.ManagedEventStream, error) { + atomic.AddInt32(&otherOpens, 1) + return lipapi.NewFixedEventStream([]lipapi.Event{ + {Kind: lipapi.EventResponseStarted}, + {Kind: lipapi.EventResponseFinished}, + }), nil + }, + }, + }, + Rand: routing.NewSeededRng(1), + } + call := &lipapi.Call{ + Route: lipapi.RouteIntent{Selector: "preferred:m|other:m"}, + Messages: []lipapi.Message{{Role: lipapi.RoleUser, Parts: []lipapi.Part{lipapi.TextPart("hi")}}}, + } + // Inject route preference for "preferred" backend. + ctx := execctx.WithRouteCandidatePreferences(context.Background(), []string{"preferred"}) + stream, err := ex.Execute(ctx, call) + if err != nil { + t.Fatal(err) + } + if _, err := lipapi.Collect(context.Background(), stream); err != nil { + t.Fatalf("Collect: %v", err) + } + if atomic.LoadInt32(&preferredOpens) == 0 { + t.Fatal("expected preferred backend to be opened; route preference was not honored by the planner") + } + if atomic.LoadInt32(&otherOpens) > 0 { + t.Fatal("expected other backend to NOT be opened when preference is honored on first attempt") + } +} From 4f9ba5b31842a7494c098da82f2914c1de52318c Mon Sep 17 00:00:00 2001 From: Mateusz Date: Wed, 8 Jul 2026 00:05:07 +0200 Subject: [PATCH 2/3] fix(arch): record prepare-submit error on executor span (PR #115 review) The RequestPreparer extraction moved the prepareSubmitAndALeg call from inline Execute into prepareRequest. The former inline flow started the lip.executor.execute span and deferred finalize(err) (RecordError + SetStatus + End), so a prepareSubmitAndALeg failure was recorded on the span. After extraction, the prepareSubmitAndALeg error path in prepareRequest called execSpan.End() directly, dropping RecordError/SetStatus; and Execute does not call finalize when prepareRequest returns an error (prep is nil). Route the prepare-submit error through prep.finalize(err) so the span records the failure before ending, restoring the prior telemetry. No behavior change beyond telemetry. Verified: go build ./internal/core/runtime/... OK; full repo golangci-lint v2.11.4 -> 0 issues; internal/core/runtime tests pass. Co-authored-by: Cursor --- internal/core/runtime/executor_prepare_request.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/internal/core/runtime/executor_prepare_request.go b/internal/core/runtime/executor_prepare_request.go index 87932326..482dfff4 100644 --- a/internal/core/runtime/executor_prepare_request.go +++ b/internal/core/runtime/executor_prepare_request.go @@ -77,7 +77,11 @@ func (e *Executor) prepareRequest(ctx context.Context, call *lipapi.Call) (*prep prep.traceID, prep.baseline, prep.aLeg, prep.ctx, err = e.prepareSubmitAndALeg(prep.ctx, bus, call) if err != nil { - prep.execSpan.End() + // Route through finalize so the lip.executor.execute span records the + // prepare-submit failure (RecordError + SetStatus) before ending, matching + // the former inline Execute defer. Execute does not call finalize when + // prepareRequest returns an error (prep is nil), so the span ends here. + prep.finalize(err) return nil, noop, fmt.Errorf("executor: prepare submit: %w", err) } From 68e7a1bdd8cc9b9bd2b9055a6abc3b7a35b7100a Mon Sep 17 00:00:00 2001 From: Mateusz Date: Wed, 8 Jul 2026 00:05:15 +0200 Subject: [PATCH 3/3] fix(lint): use atomic.Int32 in route-preference test (modernize) golangci-lint v2.11.4 modernize flagged var preferredOpens/otherOpens int32 used with atomic.AddInt32/LoadInt32 as simplifiable to atomic.Int32 method calls. Switch to atomic.Int32 (.Add/.Load). Behavior identical; this unblocks the QA golangci-lint step. Verified: golangci-lint v2.11.4 -> 0 issues; TestExecutor_execute_routePreferenceDrivesPlanner passes. Co-authored-by: Cursor --- .../core/runtime/executor_route_preference_test.go | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/internal/core/runtime/executor_route_preference_test.go b/internal/core/runtime/executor_route_preference_test.go index 17fc3714..84fad452 100644 --- a/internal/core/runtime/executor_route_preference_test.go +++ b/internal/core/runtime/executor_route_preference_test.go @@ -24,8 +24,8 @@ func TestExecutor_execute_routePreferenceDrivesPlanner(t *testing.T) { if err != nil { t.Fatal(err) } - var preferredOpens int32 - var otherOpens int32 + var preferredOpens atomic.Int32 + var otherOpens atomic.Int32 ex := &runtime.Executor{ Store: st, Bus: hooks.New(hooks.Config{}), @@ -33,7 +33,7 @@ func TestExecutor_execute_routePreferenceDrivesPlanner(t *testing.T) { "preferred": { Caps: lipapi.NewBackendCaps(lipapi.CapabilityStreaming), Open: func(ctx context.Context, call lipapi.Call, cand routing.AttemptCandidate) (lipapi.ManagedEventStream, error) { - atomic.AddInt32(&preferredOpens, 1) + preferredOpens.Add(1) return lipapi.NewFixedEventStream([]lipapi.Event{ {Kind: lipapi.EventResponseStarted}, {Kind: lipapi.EventResponseFinished}, @@ -43,7 +43,7 @@ func TestExecutor_execute_routePreferenceDrivesPlanner(t *testing.T) { "other": { Caps: lipapi.NewBackendCaps(lipapi.CapabilityStreaming), Open: func(ctx context.Context, call lipapi.Call, cand routing.AttemptCandidate) (lipapi.ManagedEventStream, error) { - atomic.AddInt32(&otherOpens, 1) + otherOpens.Add(1) return lipapi.NewFixedEventStream([]lipapi.Event{ {Kind: lipapi.EventResponseStarted}, {Kind: lipapi.EventResponseFinished}, @@ -66,10 +66,10 @@ func TestExecutor_execute_routePreferenceDrivesPlanner(t *testing.T) { if _, err := lipapi.Collect(context.Background(), stream); err != nil { t.Fatalf("Collect: %v", err) } - if atomic.LoadInt32(&preferredOpens) == 0 { + if preferredOpens.Load() == 0 { t.Fatal("expected preferred backend to be opened; route preference was not honored by the planner") } - if atomic.LoadInt32(&otherOpens) > 0 { + if otherOpens.Load() > 0 { t.Fatal("expected other backend to NOT be opened when preference is honored on first attempt") } }