Skip to content

Commit 017d89b

Browse files
committed
fix(distributed): thread request context through backend Load + cover ctx propagation
Five non-OpenAI backend helpers were silently using app.Context instead of the request context for the gRPC backend call: transcription, TTS, image generation, rerank, VAD. Effect: distributedhdr.Stamp in the router callback was a silent no-op for these paths, AND client cancellation didn't propagate to in-flight inference. Thread c.Request().Context() (or the equivalent input.Context after the request middleware has installed the correlation-ID derived context) through each helper and into ModelOptions via model.WithContext(ctx). ImageGeneration's signature gains a leading ctx parameter; in-tree callers (openai image, openai inpainting, openai inpainting_test) are updated to match. ModelEmbedding gains a leading ctx parameter for the same reason; the openai and ollama embedding handlers pass the request context through. chat_stream_workers.go defers the initial role=assistant chunk emission until the first token callback so the wrapper's lazy X-LocalAI-Node lookup against the loader runs AFTER ml.Load has stamped the per-modelID node ID; semantically identical for clients (role still arrives before any text). Regression test core/backend/ctx_propagation_test.go pins ctx propagation for all five helpers. Docs updated to enumerate the full endpoint coverage of the --expose-node-header flag. Assisted-by: Claude:claude-opus-4-7[1m] Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
1 parent 48b08c0 commit 017d89b

15 files changed

Lines changed: 236 additions & 31 deletions

File tree

Lines changed: 169 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,169 @@
1+
package backend_test
2+
3+
// Regression spec for X-LocalAI-Node coverage on audio/image/TTS/rerank/VAD.
4+
//
5+
// The X-LocalAI-Node middleware (core/http/middleware.ExposeNodeHeader)
6+
// works end-to-end only if the per-request holder attached to the HTTP
7+
// request context reaches the SmartRouter via ml.Load(opts...). The chain
8+
// is:
9+
//
10+
// handler -> backend.Foo(ctx, ...) -> ModelOptions(cfg, app, WithContext(ctx))
11+
// -> ml.Load(opts...) -> grpcModel(..., o.context) -> modelRouter(ctx, ...)
12+
// -> SmartRouter -> distributedhdr.Stamp(ctx, nodeID)
13+
//
14+
// If any backend helper drops `ctx` and lets ModelOptions fall back to the
15+
// app context, the router never sees the per-request holder and the
16+
// header silently stays empty for that endpoint. These specs pin the
17+
// request-context-reaches-router contract for the five backend helpers
18+
// that were previously dropping ctx between the handler and Load.
19+
20+
import (
21+
"context"
22+
"sync/atomic"
23+
24+
"github.com/mudler/LocalAI/core/backend"
25+
"github.com/mudler/LocalAI/core/config"
26+
"github.com/mudler/LocalAI/core/schema"
27+
pbproto "github.com/mudler/LocalAI/pkg/grpc/proto"
28+
"github.com/mudler/LocalAI/pkg/distributedhdr"
29+
"github.com/mudler/LocalAI/pkg/model"
30+
"github.com/mudler/LocalAI/pkg/system"
31+
32+
. "github.com/onsi/ginkgo/v2"
33+
. "github.com/onsi/gomega"
34+
)
35+
36+
// newCapturingLoader returns a ModelLoader wired with a stub model router
37+
// that captures the context it receives and then short-circuits with a
38+
// sentinel error. The router callback is the exact seam where the
39+
// SmartRouter would call distributedhdr.Stamp in production, so observing
40+
// the holder here is equivalent to observing it at the real router.
41+
func newCapturingLoader() (*model.ModelLoader, *atomic.Value, func() context.Context) {
42+
loader := model.NewModelLoader(&system.SystemState{})
43+
var captured atomic.Value
44+
loader.SetModelRouter(func(ctx context.Context, _ string, _, _, _ string, _ *pbproto.ModelOptions, _ bool) (*model.Model, error) {
45+
captured.Store(ctx)
46+
// Return an error so the backend short-circuits before trying to
47+
// dial gRPC. We only care about the context-arrival contract.
48+
return nil, errRouterShortCircuit
49+
})
50+
get := func() context.Context {
51+
v, _ := captured.Load().(context.Context)
52+
return v
53+
}
54+
return loader, &captured, get
55+
}
56+
57+
var errRouterShortCircuit = sentinelErr("router short-circuit (test)")
58+
59+
type sentinelErr string
60+
61+
func (s sentinelErr) Error() string { return string(s) }
62+
63+
func newAppCfg() *config.ApplicationConfig {
64+
return config.NewApplicationConfig(config.WithSystemState(&system.SystemState{}))
65+
}
66+
67+
func newModelCfg() config.ModelConfig {
68+
threads := 1
69+
cfg := config.ModelConfig{
70+
Name: "test-model",
71+
Backend: "stub-backend",
72+
Threads: &threads,
73+
}
74+
cfg.Model = "test.bin"
75+
return cfg
76+
}
77+
78+
var _ = Describe("X-LocalAI-Node ctx propagation contract", func() {
79+
const fakeNodeID = "node-ctx-propagation-7"
80+
81+
var (
82+
appCfg *config.ApplicationConfig
83+
modelCfg config.ModelConfig
84+
loader *model.ModelLoader
85+
routerCtxOf func() context.Context
86+
holder *atomic.Value
87+
reqCtx context.Context
88+
)
89+
90+
BeforeEach(func() {
91+
appCfg = newAppCfg()
92+
modelCfg = newModelCfg()
93+
loader, _, routerCtxOf = newCapturingLoader()
94+
holder = distributedhdr.NewHolder()
95+
reqCtx = distributedhdr.WithHolder(context.Background(), holder)
96+
})
97+
98+
// stampViaRouterCtx asserts the captured router context carries the
99+
// SAME holder that was attached to the request. We verify by stamping
100+
// through the router-side ctx and observing the value via the
101+
// request-side holder; if the holders were different objects the load
102+
// would return "".
103+
stampViaRouterCtx := func() {
104+
routerCtx := routerCtxOf()
105+
Expect(routerCtx).ToNot(BeNil(), "router callback must have been invoked")
106+
distributedhdr.Stamp(routerCtx, fakeNodeID)
107+
Expect(distributedhdr.Load(holder)).To(Equal(fakeNodeID),
108+
"stamp via router-side ctx must be observable via the request-side holder")
109+
}
110+
111+
It("Rerank forwards the request context to the SmartRouter", func() {
112+
_, err := backend.Rerank(reqCtx, &pbproto.RerankRequest{Query: "q"}, loader, appCfg, modelCfg)
113+
Expect(err).To(HaveOccurred())
114+
Expect(err.Error()).To(ContainSubstring("router short-circuit (test)"))
115+
stampViaRouterCtx()
116+
})
117+
118+
It("VAD forwards the request context to the SmartRouter", func() {
119+
_, err := backend.VAD(&schema.VADRequest{}, reqCtx, loader, appCfg, modelCfg)
120+
Expect(err).To(HaveOccurred())
121+
Expect(err.Error()).To(ContainSubstring("router short-circuit (test)"))
122+
stampViaRouterCtx()
123+
})
124+
125+
It("ModelTTS forwards the request context to the SmartRouter", func() {
126+
_, _, err := backend.ModelTTS(reqCtx, "hello", "", "", loader, appCfg, modelCfg)
127+
Expect(err).To(HaveOccurred())
128+
Expect(err.Error()).To(ContainSubstring("router short-circuit (test)"))
129+
stampViaRouterCtx()
130+
})
131+
132+
It("ModelTTSStream forwards the request context to the SmartRouter", func() {
133+
err := backend.ModelTTSStream(reqCtx, "hello", "", "", loader, appCfg, modelCfg, func([]byte) error { return nil })
134+
Expect(err).To(HaveOccurred())
135+
Expect(err.Error()).To(ContainSubstring("router short-circuit (test)"))
136+
stampViaRouterCtx()
137+
})
138+
139+
It("ModelTranscriptionWithOptions forwards the request context to the SmartRouter", func() {
140+
_, err := backend.ModelTranscriptionWithOptions(reqCtx, backend.TranscriptionRequest{Audio: "x.wav"}, loader, modelCfg, appCfg)
141+
Expect(err).To(HaveOccurred())
142+
Expect(err.Error()).To(ContainSubstring("router short-circuit (test)"))
143+
stampViaRouterCtx()
144+
})
145+
146+
It("ModelTranscriptionStream forwards the request context to the SmartRouter", func() {
147+
err := backend.ModelTranscriptionStream(reqCtx, backend.TranscriptionRequest{Audio: "x.wav"}, loader, modelCfg, appCfg, func(backend.TranscriptionStreamChunk) {})
148+
Expect(err).To(HaveOccurred())
149+
Expect(err.Error()).To(ContainSubstring("router short-circuit (test)"))
150+
stampViaRouterCtx()
151+
})
152+
153+
It("ImageGeneration forwards the request context to the SmartRouter", func() {
154+
_, err := backend.ImageGeneration(reqCtx, 64, 64, 1, 0, "p", "", "", "/tmp/out.png", loader, modelCfg, appCfg, nil)
155+
Expect(err).To(HaveOccurred())
156+
Expect(err.Error()).To(ContainSubstring("router short-circuit (test)"))
157+
stampViaRouterCtx()
158+
})
159+
160+
It("does NOT leak the holder when the app context is used instead", func() {
161+
// Sanity: the bug being fixed manifests as the router getting
162+
// appCfg.Context (no holder) instead of reqCtx (holder). A direct
163+
// call with context.Background() must not see the holder via the
164+
// app context surface.
165+
appCtxOnly := appCfg.Context
166+
Expect(distributedhdr.Holder(appCtxOnly)).To(BeNil(),
167+
"the app context must not be the carrier of per-request holders")
168+
})
169+
})

core/backend/embeddings.go

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -30,17 +30,20 @@ type modelEmbedder struct {
3030
appConfig *config.ApplicationConfig
3131
}
3232

33-
func (e *modelEmbedder) Embed(_ context.Context, text string) ([]float32, error) {
34-
fn, err := ModelEmbedding(text, nil, e.loader, e.modelConfig, e.appConfig)
33+
func (e *modelEmbedder) Embed(ctx context.Context, text string) ([]float32, error) {
34+
fn, err := ModelEmbedding(ctx, text, nil, e.loader, e.modelConfig, e.appConfig)
3535
if err != nil {
3636
return nil, err
3737
}
3838
return fn()
3939
}
4040

41-
func ModelEmbedding(s string, tokens []int, loader *model.ModelLoader, modelConfig config.ModelConfig, appConfig *config.ApplicationConfig) (func() ([]float32, error), error) {
41+
func ModelEmbedding(ctx context.Context, s string, tokens []int, loader *model.ModelLoader, modelConfig config.ModelConfig, appConfig *config.ApplicationConfig) (func() ([]float32, error), error) {
4242

43-
opts := ModelOptions(modelConfig, appConfig)
43+
// model.WithContext(ctx) overrides the app-context default set in
44+
// ModelOptions so distributed routing decisions reach the request's
45+
// X-LocalAI-Node holder via distributedhdr.Stamp.
46+
opts := ModelOptions(modelConfig, appConfig, model.WithContext(ctx))
4447

4548
inferenceModel, err := loader.Load(opts...)
4649
if err != nil {

core/backend/image.go

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package backend
22

33
import (
4+
"context"
45
"time"
56

67
"github.com/mudler/LocalAI/core/config"
@@ -10,9 +11,12 @@ import (
1011
model "github.com/mudler/LocalAI/pkg/model"
1112
)
1213

13-
func ImageGeneration(height, width, step, seed int, positive_prompt, negative_prompt, src, dst string, loader *model.ModelLoader, modelConfig config.ModelConfig, appConfig *config.ApplicationConfig, refImages []string) (func() error, error) {
14+
func ImageGeneration(ctx context.Context, height, width, step, seed int, positive_prompt, negative_prompt, src, dst string, loader *model.ModelLoader, modelConfig config.ModelConfig, appConfig *config.ApplicationConfig, refImages []string) (func() error, error) {
1415

15-
opts := ModelOptions(modelConfig, appConfig)
16+
// model.WithContext(ctx) overrides the app-context default set in
17+
// ModelOptions so distributed routing decisions reach the request's
18+
// X-LocalAI-Node holder via distributedhdr.Stamp.
19+
opts := ModelOptions(modelConfig, appConfig, model.WithContext(ctx))
1620
inferenceModel, err := loader.Load(
1721
opts...,
1822
)
@@ -23,7 +27,7 @@ func ImageGeneration(height, width, step, seed int, positive_prompt, negative_pr
2327

2428
fn := func() error {
2529
_, err := inferenceModel.GenerateImage(
26-
appConfig.Context,
30+
ctx,
2731
&proto.GenerateImageRequest{
2832
Height: int32(height),
2933
Width: int32(width),

core/backend/llm.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -94,7 +94,7 @@ func ModelInference(ctx context.Context, s string, messages schema.Messages, ima
9494
}
9595
}
9696

97-
opts := ModelOptions(*c, o)
97+
opts := ModelOptions(*c, o, model.WithContext(ctx))
9898
inferenceModel, err := loader.Load(opts...)
9999
if err != nil {
100100
recordModelLoadFailure(o, c.Name, c.Backend, err, map[string]any{"model_file": modelFile})

core/backend/rerank.go

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@ type RerankResult struct {
2020
}
2121

2222
// Reranker scores a list of candidate documents against a query.
23-
// Returns one RerankResult per input document (no top-N truncation
23+
// Returns one RerankResult per input document (no top-N truncation -
2424
// callers that need it can sort and slice).
2525
type Reranker interface {
2626
Rerank(ctx context.Context, query string, documents []string) ([]RerankResult, error)
@@ -41,7 +41,7 @@ func (r *modelReranker) Rerank(ctx context.Context, query string, documents []st
4141
req := &proto.RerankRequest{
4242
Query: query,
4343
Documents: documents,
44-
// TopN=0 backend returns scores for every document. Truncating
44+
// TopN=0: backend returns scores for every document. Truncating
4545
// here would silently zero out labels the reranker considered
4646
// unlikely, which the router classifier needs.
4747
}
@@ -57,7 +57,10 @@ func (r *modelReranker) Rerank(ctx context.Context, query string, documents []st
5757
}
5858

5959
func Rerank(ctx context.Context, request *proto.RerankRequest, loader *model.ModelLoader, appConfig *config.ApplicationConfig, modelConfig config.ModelConfig) (*proto.RerankResult, error) {
60-
opts := ModelOptions(modelConfig, appConfig)
60+
// model.WithContext(ctx) overrides the app-context default set in
61+
// ModelOptions so distributed routing decisions reach the request's
62+
// X-LocalAI-Node holder via distributedhdr.Stamp.
63+
opts := ModelOptions(modelConfig, appConfig, model.WithContext(ctx))
6164
rerankModel, err := loader.Load(opts...)
6265
if err != nil {
6366
recordModelLoadFailure(appConfig, modelConfig.Name, modelConfig.Backend, err, nil)

core/backend/transcript.go

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -41,11 +41,14 @@ func (r *TranscriptionRequest) toProto(threads uint32) *proto.TranscriptRequest
4141
}
4242
}
4343

44-
func loadTranscriptionModel(ml *model.ModelLoader, modelConfig config.ModelConfig, appConfig *config.ApplicationConfig) (grpcPkg.Backend, error) {
44+
func loadTranscriptionModel(ctx context.Context, ml *model.ModelLoader, modelConfig config.ModelConfig, appConfig *config.ApplicationConfig) (grpcPkg.Backend, error) {
4545
if modelConfig.Backend == "" {
4646
modelConfig.Backend = model.WhisperBackend
4747
}
48-
opts := ModelOptions(modelConfig, appConfig)
48+
// model.WithContext(ctx) overrides the app-context default set in
49+
// ModelOptions so distributed routing decisions reach the request's
50+
// X-LocalAI-Node holder via distributedhdr.Stamp.
51+
opts := ModelOptions(modelConfig, appConfig, model.WithContext(ctx))
4952
transcriptionModel, err := ml.Load(opts...)
5053
if err != nil {
5154
recordModelLoadFailure(appConfig, modelConfig.Name, modelConfig.Backend, err, nil)
@@ -68,7 +71,7 @@ func ModelTranscription(ctx context.Context, audio, language string, translate,
6871
}
6972

7073
func ModelTranscriptionWithOptions(ctx context.Context, req TranscriptionRequest, ml *model.ModelLoader, modelConfig config.ModelConfig, appConfig *config.ApplicationConfig) (*schema.TranscriptionResult, error) {
71-
transcriptionModel, err := loadTranscriptionModel(ml, modelConfig, appConfig)
74+
transcriptionModel, err := loadTranscriptionModel(ctx, ml, modelConfig, appConfig)
7275
if err != nil {
7376
return nil, err
7477
}
@@ -150,7 +153,7 @@ type TranscriptionStreamChunk struct {
150153
// support real streaming should still emit one terminal event with Final set,
151154
// which the HTTP layer turns into a single delta + done SSE pair.
152155
func ModelTranscriptionStream(ctx context.Context, req TranscriptionRequest, ml *model.ModelLoader, modelConfig config.ModelConfig, appConfig *config.ApplicationConfig, onChunk func(TranscriptionStreamChunk)) error {
153-
transcriptionModel, err := loadTranscriptionModel(ml, modelConfig, appConfig)
156+
transcriptionModel, err := loadTranscriptionModel(ctx, ml, modelConfig, appConfig)
154157
if err != nil {
155158
return err
156159
}

core/backend/tts.go

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,10 @@ func ModelTTS(
2929
appConfig *config.ApplicationConfig,
3030
modelConfig config.ModelConfig,
3131
) (string, *proto.Result, error) {
32-
opts := ModelOptions(modelConfig, appConfig)
32+
// model.WithContext(ctx) overrides the app-context default set in
33+
// ModelOptions so distributed routing decisions reach the request's
34+
// X-LocalAI-Node holder via distributedhdr.Stamp.
35+
opts := ModelOptions(modelConfig, appConfig, model.WithContext(ctx))
3336
ttsModel, err := loader.Load(opts...)
3437
if err != nil {
3538
recordModelLoadFailure(appConfig, modelConfig.Name, modelConfig.Backend, err, nil)
@@ -131,7 +134,7 @@ func ModelTTSStream(
131134
modelConfig config.ModelConfig,
132135
audioCallback func([]byte) error,
133136
) error {
134-
opts := ModelOptions(modelConfig, appConfig)
137+
opts := ModelOptions(modelConfig, appConfig, model.WithContext(ctx))
135138
ttsModel, err := loader.Load(opts...)
136139
if err != nil {
137140
recordModelLoadFailure(appConfig, modelConfig.Name, modelConfig.Backend, err, nil)

core/backend/vad.go

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,10 @@ func VAD(request *schema.VADRequest,
1414
ml *model.ModelLoader,
1515
appConfig *config.ApplicationConfig,
1616
modelConfig config.ModelConfig) (*schema.VADResponse, error) {
17-
opts := ModelOptions(modelConfig, appConfig)
17+
// model.WithContext(ctx) overrides the app-context default set in
18+
// ModelOptions so distributed routing decisions reach the request's
19+
// X-LocalAI-Node holder via distributedhdr.Stamp.
20+
opts := ModelOptions(modelConfig, appConfig, model.WithContext(ctx))
1821
vadModel, err := ml.Load(opts...)
1922
if err != nil {
2023
recordModelLoadFailure(appConfig, modelConfig.Name, modelConfig.Backend, err, nil)

core/http/endpoints/ollama/embed.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ func EmbedEndpoint(cl *config.ModelConfigLoader, ml *model.ModelLoader, appConfi
3636
promptEvalCount := 0
3737

3838
for _, s := range inputStrings {
39-
embedFn, err := backend.ModelEmbedding(s, []int{}, ml, *cfg, appConfig)
39+
embedFn, err := backend.ModelEmbedding(c.Request().Context(), s, []int{}, ml, *cfg, appConfig)
4040
if err != nil {
4141
xlog.Error("Ollama embed failed", "error", err)
4242
return ollamaError(c, 500, fmt.Sprintf("embedding failed: %v", err))

core/http/endpoints/openai/chat_stream_workers.go

Lines changed: 22 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,13 @@ import (
2121
// The caller owns the `responses` channel and is expected to read from
2222
// it while this function runs; processStream closes the channel before
2323
// returning.
24+
//
25+
// X-LocalAI-Node attribution (when --expose-node-header is on) is
26+
// handled by middleware.ExposeNodeHeader at the response writer wrapper
27+
// layer; no in-band signal from the worker is needed. The initial
28+
// role=assistant chunk is still emitted from the first token callback
29+
// rather than eagerly here, so the wrapper's lazy lookup against the
30+
// loader runs AFTER ml.Load has stamped the per-modelID node ID.
2431
func processStream(
2532
s string,
2633
req *schema.OpenAIRequest,
@@ -32,13 +39,7 @@ func processStream(
3239
id string,
3340
created int,
3441
) (backend.TokenUsage, error) {
35-
responses <- schema.OpenAIResponse{
36-
ID: id,
37-
Created: created,
38-
Model: req.Model, // we have to return what the user sent here, due to OpenAI spec.
39-
Choices: []schema.Choice{{Delta: &schema.Message{Role: "assistant"}, Index: 0, FinishReason: nil}},
40-
Object: "chat.completion.chunk",
41-
}
42+
sentInitialRole := false
4243

4344
// Detect if thinking token is already in prompt or template
4445
// When UseTokenizerTemplate is enabled, predInput is empty, so we check the template
@@ -70,6 +71,17 @@ func processStream(
7071
contentDelta = goContent
7172
}
7273

74+
if !sentInitialRole {
75+
sentInitialRole = true
76+
responses <- schema.OpenAIResponse{
77+
ID: id,
78+
Created: created,
79+
Model: req.Model, // we have to return what the user sent here, due to OpenAI spec.
80+
Choices: []schema.Choice{{Delta: &schema.Message{Role: "assistant"}, Index: 0, FinishReason: nil}},
81+
Object: "chat.completion.chunk",
82+
}
83+
}
84+
7385
delta := &schema.Message{}
7486
if contentDelta != "" {
7587
delta.Content = &contentDelta
@@ -130,6 +142,9 @@ func processStreamWithTools(
130142
hasChatDeltaToolCalls := false
131143
hasChatDeltaContent := false
132144

145+
// X-LocalAI-Node attribution is handled by middleware.ExposeNodeHeader
146+
// at the wrapper layer; no in-band signalling from this worker.
147+
133148
_, finalUsage, chatDeltas, err := ComputeChoices(req, prompt, cfg, cl, startupOptions, loader, func(s string, c *[]schema.Choice) {}, func(s string, usage backend.TokenUsage) bool {
134149
result += s
135150

0 commit comments

Comments
 (0)