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..482dfff4 --- /dev/null +++ b/internal/core/runtime/executor_prepare_request.go @@ -0,0 +1,122 @@ +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 { + // 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) + } + + 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..84fad452 --- /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 atomic.Int32 + var otherOpens atomic.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) { + preferredOpens.Add(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) { + otherOpens.Add(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 preferredOpens.Load() == 0 { + t.Fatal("expected preferred backend to be opened; route preference was not honored by the planner") + } + if otherOpens.Load() > 0 { + t.Fatal("expected other backend to NOT be opened when preference is honored on first attempt") + } +}