fix(quantization): record job progress without an attached listener (#11874)

[Anai-Guo]

Fixes #11874.

The bug

A quantization job that runs with no client attached to its progress stream stays queued forever — in the API and in state.json — while the finished artifact sits on disk.

state.json is written once, by saveJobState in StartJob, and the only code that ever advanced a job afterwards lived inside the stream callback of StreamProgress:

// core/services/quantization/service.go, inside StreamProgress
j.Status = update.Status

StartJob kicks the backend off over gRPC and returns. Nothing consumes the backend's progress stream until an SSE client calls StreamProgress, so with nobody watching, no transition is ever recorded. Job state depended on an observer.

Why the obvious fix needs care

The issue suggests having the job runner record transitions and letting StreamProgress be a pure reader. That is the right shape, but simply starting a second consumer alongside StreamProgress would break the case that works today.

The backend's progress stream is single-consumer and destructive. In backend/python/llama-cpp-quantization/backend.py, each job owns one queue.Queue:

class ActiveJob:
    def __init__(self, job_id):
        self.progress_queue = queue.Queue()

_send_progress does job.progress_queue.put(update) and QuantizationProgress does job.progress_queue.get(timeout=1.0) — it pops. Two readers of that stream do not each see every update; they split them. So a background watcher running next to a live SSE reader would steal half the events from the UI.

That rules out "add a watcher and leave StreamProgress as-is". The stream has to be opened exactly once per job.

The change

core/services/quantization/service.go only:

• StartJob starts watchProgress on the application context (not the request context, which is done the moment the HTTP handler returns). That goroutine is the single reader of the backend stream for the job's lifetime.
• applyProgressUpdate — the state-recording half of the old StreamProgress callback, unchanged in behaviour (terminal statuses still win over late updates) — now runs from the watcher, so transitions land in the SyncedMap and in state.json whether or not anyone is attached.
• StreamProgress becomes a pure reader: it subscribes to an in-process fan-out and returns when a terminal event arrives or ctx is done. It no longer loads a backend or opens a gRPC stream.
• A client that attaches to a job which already finished gets one final event built from the stored job instead of blocking. This also covers jobs hydrated from disk after a restart, which loadJobsFromDisk marks stopped and which have no watcher.
• A subscriber that cannot keep up drops events (buffer 64) rather than stalling the reader that is recording state for everyone else.

Two paths that used to end a client's stream by breaking the gRPC connection now need an explicit release, since the client is no longer holding that connection:

• StopJob kills the backend process, so no terminal update will ever reach the watcher — it publishes the stopped event itself.
• If the watcher's stream ends without a terminal update the backend is gone, so the job is recorded as failed and that is published, rather than leaving it in a running state forever — the same failure mode this change exists to prevent. Skipped when ctx is already done, so a shutdown still leaves jobs for loadJobsFromDisk to report as stopped.

No schema, API, or route changes; the SSE payload is byte-for-byte the same QuantizationProgressEvent.

Tests

Seven specs added to core/services/quantization/service_test.go (white-box, no backend needed — the existing suite already drives the service with a nil model loader):

• advances job state and rewrites state.json with no subscriber attached — the reported failure, asserted with progressSubs empty: GetJob reports completed and the on-disk state.json is rewritten with the status and output file.
• does not let a late update overwrite a terminal status — pins the existing terminal-state guard that moved with the code.
• delivers one update to every attached subscriber — the fan-out that replaces per-client gRPC streams.
• unsubscribing removes the job's entry once the last client leaves — no map growth per finished job.
• returns a final event immediately for a job that already finished — the attach-after-completion / post-restart path.
• releases an attached client when the job is stopped — the stop path that no longer ends by breaking a gRPC connection.
• streams published events to a client until a terminal status arrives — end of stream on terminal status.

$ make protogen-go && go test -race -count=2 ./core/services/quantization/
ok  	github.com/mudler/LocalAI/core/services/quantization

gofmt -l and go vet on the package are clean.

Scope of verification: the specs above and the package build/vet are what I actually ran. I did not re-run the end-to-end reproduction against a live llama-cpp-quantization backend — the single-consumer queue behaviour is read from backend.py as quoted above, and the failure mode is the one in the issue report.

🤖 Generated with Claude Code