fix(worker): pin the rune cut, answer unload honestly, drop the dead publisher

Review fix round 1. Seven non-blocking findings; the blocking one is a
merge gate for Task 4 rather than anything in this diff, and the report's
concern about it is corrected: until Task 4 lands, PingNode probes two
subjects no serve-backend worker subscribes to any more, so every healthy
worker reads as absent and is marked unhealthy on the scheduling path.

The rune-boundary cut in truncate was true behaviour with nothing
holding it: a byte-wise mutation survived all 201 specs. isRuneStart is
replaced by utf8.RuneStart, the same predicate the cluster package uses
for this rule, and two specs pin it, one with a rune straddling the
bound and one with a rune ending exactly on it so the fix cannot be
"always walk back".

unloadModel answered Success:true whatever Free did. That is the worker
saying "done" about work it did not do, and the frontend's only caller
is EvictLRU, so a false yes told the scheduler VRAM had been released
and let it place the next model on a node still holding the old one. It
now reports the failure, following stopModelExact, which is the honest
pattern already in this package. Still a 200: the worker answered, only
its verdict is negative. An address with nothing loaded still answers
success, which is a true answer rather than a claim about work done.

NewDebouncedInstallProgressPublisher had no production caller after the
last commit, only its own spec. Deleted rather than wired: wiring it
would publish every event on two carriers, which is what the carrier
decision exists to avoid. Its specs now run against the sink, plus one
that pins the identity stamped on each event, since the subject used to
carry the op and node id and now nothing but the body does.

The install progress wiring was exercised by no spec, because with no
gallery nothing ever invokes the download callback. The guard moves into
startProgress, shared by install and upgrade, which also emits one
resolving event before any gallery work. That is worth having on its
own: a cold install spends minutes on a manifest and a progress stream
with nothing on it is indistinguishable from a broken one. It also makes
the wiring observable end to end, and four specs now drive the real
installBackend and upgradeBackend over HTTP with no override.

model/stop and backend/stop keep taking Background rather than the
caller's context, and the sites now say why. model/stop is the
acknowledged stop path: it reserves the process, frees it, kills it,
waits for exit and releases the port, and abandoning that because the
caller hung up would leave a process marked stopping, a port not
returned to the allocator and a row nothing reconciles. In
stopBackendExact the Free is a courtesy before a kill that happens
anyway. model/unload differs because Free IS the operation there.

A route set with no prefix or no registrar is now a startup error rather
than a silent no-op: a server that comes up healthy while every route
the caller registered answers 404 is, through a tunnel, indistinguishable
from a version skew. And the AllPaths spec no longer claims to catch a
constant that was never added to the set, which it cannot; it asserts
the whole set instead, which catches a verb dropped from it.

Assisted-by: Claude Opus 5 [claude-code]
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

core/services/nodes/authenticated_routes_test.go

@@ -72,6 +72,26 @@ var _ = Describe("extra authenticated routes on the worker HTTP server", func()
 		Expect(get("/healthz", "").StatusCode).To(Equal(http.StatusOK))
 	})
 
+	It("refuses to start when a route set names no prefix", func() {
+		lis, err := net.Listen("tcp", "127.0.0.1:0")
+		Expect(err).NotTo(HaveOccurred())
+		DeferCleanup(func() { _ = lis.Close() })
+		dir := GinkgoT().TempDir()
+		_, err = StartFileTransferServerWithRoutes(lis, dir, dir, dir, token, 0, nil,
+			&AuthenticatedRoutes{Register: func(*http.ServeMux) {}})
+		Expect(err).To(MatchError(ContainSubstring("prefix")))
+	})
+
+	It("refuses to start when a route set names no registrar", func() {
+		lis, err := net.Listen("tcp", "127.0.0.1:0")
+		Expect(err).NotTo(HaveOccurred())
+		DeferCleanup(func() { _ = lis.Close() })
+		dir := GinkgoT().TempDir()
+		_, err = StartFileTransferServerWithRoutes(lis, dir, dir, dir, token, 0, nil,
+			&AuthenticatedRoutes{Prefix: "/v1/control/"})
+		Expect(err).To(MatchError(ContainSubstring("registrar")))
+	})
+
 	It("mounts nothing when no route set is given", func() {
 		lis, err := net.Listen("tcp", "127.0.0.1:0")
 		Expect(err).NotTo(HaveOccurred())

core/services/nodes/file_transfer_server.go

@@ -90,6 +90,19 @@ func StartFileTransferServerWithReadiness(lis net.Listener, stagingDir, modelsDi
 // StartFileTransferServerWithRoutes is StartFileTransferServerWithReadiness
 // plus an extra authenticated route set. See AuthenticatedRoutes.
 func StartFileTransferServerWithRoutes(lis net.Listener, stagingDir, modelsDir, dataDir, token string, maxUploadSize int64, readiness *WorkerReadiness, extra *AuthenticatedRoutes, logStore ...*model.BackendLogStore) (*http.Server, error) {
+	// Checked before anything is created. A route set that names no prefix or
+	// no registrar is a caller bug, and mounting nothing for it would be the
+	// worst possible answer: the server comes up healthy and every route the
+	// caller believes it registered answers 404, which through a tunnel is
+	// indistinguishable from a version skew.
+	if extra != nil {
+		if extra.Prefix == "" {
+			return nil, fmt.Errorf("extra routes were given no prefix to mount under")
+		}
+		if extra.Register == nil {
+			return nil, fmt.Errorf("extra routes under %q were given no registrar", extra.Prefix)
+		}
+	}
 	if err := os.MkdirAll(stagingDir, 0750); err != nil {
 		return nil, fmt.Errorf("creating staging dir %s: %w", stagingDir, err)
 	}
@@ -189,7 +202,7 @@ func StartFileTransferServerWithRoutes(lis net.Listener, stagingDir, modelsDir,
 	// Readiness: "can this worker actually accept work?" See WorkerReadiness.
 	mux.HandleFunc("/readyz", probe(readiness.Check))
 
-	if extra != nil && extra.Register != nil && extra.Prefix != "" {
+	if extra != nil {
 		extraMux := http.NewServeMux()
 		extra.Register(extraMux)
 		mux.Handle(extra.Prefix, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

core/services/nodes/install_progress_publisher.go

@@ -11,11 +11,10 @@ import (
 // and hands them to its emit sink at most once per `interval`. Always emits the
 // final event on Flush so the UI sees the terminal percentage.
 //
-// The sink is a function rather than a NATS subject because the same debounce
-// now bounds two carriers: the NATS progress subject an agent node still
-// publishes on, and a line written into the streaming HTTP response a worker
-// serves over its tunnel. Keeping one debouncer keeps the ~4/s tick bound a
-// single fact instead of one per carrier.
+// The sink is a function rather than a NATS subject because the events now ride
+// the streaming HTTP response the worker serves over its tunnel. Leaving the
+// debounce here, rather than at the sink, is what keeps the ~4/s tick bound a
+// property of install progress itself and not of whatever carries it.
 //
 // Behavior: leading-edge debounce. The first OnDownload after a quiet window
 // publishes immediately; subsequent ticks within `interval` only buffer the
@@ -38,21 +37,17 @@ type DebouncedInstallProgressPublisher struct {
 	timer           *time.Timer
 }
 
-// NewDebouncedInstallProgressPublisher constructs a publisher for one install
-// operation that publishes to the per-op NATS progress subject. interval is the
-// leading-edge debounce window (~250ms in production).
-func NewDebouncedInstallProgressPublisher(client messaging.MessagingClient, nodeID, opID, backend string, interval time.Duration) *DebouncedInstallProgressPublisher {
-	subject := messaging.SubjectNodeBackendInstallProgress(nodeID, opID)
-	return NewDebouncedInstallProgressSink(func(ev messaging.BackendInstallProgressEvent) {
-		_ = client.Publish(subject, ev)
-	}, nodeID, opID, backend, interval)
-}
-
 // NewDebouncedInstallProgressSink constructs a publisher for one install
 // operation that hands each debounced event to emit.
 //
 // emit is called with p.mu released, so a sink that blocks on a slow link
 // cannot stall the gallery download loop that feeds it.
+//
+// There was a sibling constructor that published to the per-op NATS progress
+// subject. It is gone rather than kept: a worker streams these events inside
+// the install response now, so the NATS publisher had no caller left, and a
+// constructor alive only for its own spec is a carrier a reader would believe
+// still runs.
 func NewDebouncedInstallProgressSink(emit func(messaging.BackendInstallProgressEvent), nodeID, opID, backend string, interval time.Duration) *DebouncedInstallProgressPublisher {
 	return &DebouncedInstallProgressPublisher{
 		emit:     emit,

core/services/nodes/install_progress_publisher_test.go

@@ -1,6 +1,7 @@
 package nodes
 
 import (
+	"sync"
 	"time"
 
 	. "github.com/onsi/ginkgo/v2"
@@ -9,10 +10,29 @@ import (
 	"github.com/mudler/LocalAI/core/services/messaging"
 )
 
+// collectingSink records every event the debouncer emits. Emits arrive from the
+// trailing timer's own goroutine as well as from OnDownload, so it locks.
+type collectingSink struct {
+	mu     sync.Mutex
+	events []messaging.BackendInstallProgressEvent
+}
+
+func (c *collectingSink) emit(ev messaging.BackendInstallProgressEvent) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.events = append(c.events, ev)
+}
+
+func (c *collectingSink) snapshot() []messaging.BackendInstallProgressEvent {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return append([]messaging.BackendInstallProgressEvent(nil), c.events...)
+}
+
 var _ = Describe("DebouncedInstallProgressPublisher", func() {
-	It("publishes the first event immediately and debounces subsequent ones within the window", func() {
-		mc := newScriptedMessagingClient()
-		pub := NewDebouncedInstallProgressPublisher(mc, "n1", "op1", "vllm", 50*time.Millisecond)
+	It("emits the first event immediately and debounces subsequent ones within the window", func() {
+		sink := &collectingSink{}
+		pub := NewDebouncedInstallProgressSink(sink.emit, "n1", "op1", "vllm", 50*time.Millisecond)
 
 		// Three rapid-fire ticks within the debounce window.
 		pub.OnDownload("vllm.tar.zst", "100 MB", "1 GB", 10.0)
@@ -20,29 +40,44 @@ var _ = Describe("DebouncedInstallProgressPublisher", func() {
 		pub.OnDownload("vllm.tar.zst", "300 MB", "1 GB", 30.0)
 		pub.Flush()
 
-		// First event publishes immediately; the others coalesce; Flush guarantees a final.
-		// So we expect at least 2 publishes and at most 4 (lead + final + any window-bounded).
-		Eventually(func() int {
-			return len(mc.publishCalls(messaging.SubjectNodeBackendInstallProgress("n1", "op1")))
-		}, "1s").Should(BeNumerically(">=", 2))
-		calls := mc.publishCalls(messaging.SubjectNodeBackendInstallProgress("n1", "op1"))
-		Expect(len(calls)).To(BeNumerically("<=", 4),
-			"three ticks within the debounce window should produce at most ~4 publishes")
+		// First event emits immediately; the others coalesce; Flush guarantees a final.
+		// So we expect at least 2 emits and at most 4 (lead + final + any window-bounded).
+		Eventually(func() int { return len(sink.snapshot()) }, "1s").Should(BeNumerically(">=", 2))
+		Expect(len(sink.snapshot())).To(BeNumerically("<=", 4),
+			"three ticks within the debounce window should produce at most ~4 emits")
 	})
 
-	It("publishes the final event after Flush with the latest percentage", func() {
-		mc := newScriptedMessagingClient()
-		pub := NewDebouncedInstallProgressPublisher(mc, "n1", "op1", "vllm", 50*time.Millisecond)
+	It("emits the final event after Flush with the latest percentage", func() {
+		sink := &collectingSink{}
+		pub := NewDebouncedInstallProgressSink(sink.emit, "n1", "op1", "vllm", 50*time.Millisecond)
 
 		pub.OnDownload("vllm.tar.zst", "1 GB", "1 GB", 100.0)
 		pub.Flush()
 
 		Eventually(func() float64 {
-			calls := mc.publishCalls(messaging.SubjectNodeBackendInstallProgress("n1", "op1"))
-			if len(calls) == 0 {
+			events := sink.snapshot()
+			if len(events) == 0 {
 				return -1
 			}
-			return calls[len(calls)-1].Percentage
+			return events[len(events)-1].Percentage
 		}, "1s").Should(Equal(100.0))
 	})
+
+	It("stamps every event with the identity the frontend correlates on", func() {
+		// The op id and node id are what a frontend matches a progress line to
+		// an operation with; the subject used to carry them and now nothing
+		// else does, so the event body has to.
+		sink := &collectingSink{}
+		pub := NewDebouncedInstallProgressSink(sink.emit, "node-7", "op-42", "vllm", time.Millisecond)
+		pub.OnDownload("vllm.tar.zst", "1 GB", "1 GB", 100.0)
+		pub.Flush()
+
+		Eventually(func() int { return len(sink.snapshot()) }, "1s").Should(BeNumerically(">=", 1))
+		ev := sink.snapshot()[0]
+		Expect(ev.NodeID).To(Equal("node-7"))
+		Expect(ev.OpID).To(Equal("op-42"))
+		Expect(ev.Backend).To(Equal("vllm"))
+		Expect(ev.FileName).To(Equal("vllm.tar.zst"))
+		Expect(ev.Phase).To(Equal(messaging.PhaseDownloading))
+	})
 })

core/services/worker/control_routes.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"net/http"
 	"sync"
+	"unicode/utf8"
 
 	"github.com/mudler/LocalAI/core/services/messaging"
 	"github.com/mudler/LocalAI/core/services/workerctl"
@@ -117,6 +118,14 @@ func (s *backendSupervisor) RegisterControlRoutes(mux *http.ServeMux) {
 		return messaging.ModelsRunningReply{Models: s.runningModels()}, nil
 	})
 
+	// model.stop deliberately does NOT take the caller's context, and the
+	// asymmetry with model.unload below is the point. This is the ACKNOWLEDGED
+	// stop path: it reserves the process, frees it, kills it, waits for it to
+	// exit and releases its port. Abandoning that half way because the caller
+	// stopped listening would leave a process the worker has marked stopping,
+	// a port not returned to the allocator, and a controller row nothing ever
+	// reconciles. The stop has to finish whether or not anyone reads the
+	// answer; its own bounds are internal and already in place.
 	post(workerctl.PathModelStop, func(_ context.Context, body []byte) (any, error) {
 		var req messaging.ModelStopRequest
 		if err := json.Unmarshal(body, &req); err != nil {
@@ -137,6 +146,10 @@ func (s *backendSupervisor) RegisterControlRoutes(mux *http.ServeMux) {
 		return s.deleteBackend(req), nil
 	})
 
+	// model.unload is the one verb that DOES take the caller's context. Free is
+	// the whole operation here rather than a courtesy before a kill: nothing is
+	// terminated, no port is released, and abandoning it leaves the worker
+	// exactly as it was. So the caller's budget is the operation's budget.
 	post(workerctl.PathModelUnload, func(ctx context.Context, body []byte) (any, error) {
 		var req messaging.ModelUnloadRequest
 		if err := json.Unmarshal(body, &req); err != nil {
@@ -153,6 +166,9 @@ func (s *backendSupervisor) RegisterControlRoutes(mux *http.ServeMux) {
 		return s.deleteModel(req), nil
 	})
 
+	// backend.stop drops the caller's context for the same reason model.stop
+	// does, and more plainly: it is fire-and-forget, so there is no answer for
+	// the caller to still be waiting on.
 	post(workerctl.PathBackendStop, func(_ context.Context, body []byte) (any, error) {
 		req, stopAll, err := decodeBackendStopRequest(body)
 		if err != nil {
@@ -207,24 +223,24 @@ func readControlBody(w http.ResponseWriter, r *http.Request) ([]byte, bool) {
 }
 
 // truncate bounds a caller-controlled string that is about to be echoed.
-// It cuts on a rune boundary: a byte-wise cut can split a multi-byte rune, and
-// the half-rune then travels as a replacement character through every log and
-// UI that reads it.
+//
+// It cuts on a rune boundary. A byte-wise cut can split a multi-byte rune, and
+// the half rune then travels as a replacement character through every log and
+// UI that reads it; phase 2 shipped exactly that defect on a refusal reason and
+// pinned the rule afterwards. utf8.RuneStart is the same predicate the cluster
+// package uses for it, so the two are one rule rather than two hand-rolled
+// copies that can drift.
 func truncate(s string, max int) string {
 	if len(s) <= max {
 		return s
 	}
 	cut := max
-	for cut > 0 && !isRuneStart(s[cut]) {
+	for cut > 0 && !utf8.RuneStart(s[cut]) {
 		cut--
 	}
 	return s[:cut] + "…"
 }
 
-// isRuneStart reports whether b begins a UTF-8 rune (i.e. is not a
-// continuation byte).
-func isRuneStart(b byte) bool { return b&0xC0 != 0x80 }
-
 // ndjsonStream writes the Envelope lines of one streaming control response.
 //
 // The mutex is not optional. A progress line can be emitted from the debounce

core/services/worker/control_routes_test.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"io"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -13,6 +14,7 @@ import (
 	"strings"
 	"sync"
 	"syscall"
+	"unicode/utf8"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -20,6 +22,7 @@ import (
 	"github.com/mudler/LocalAI/core/services/messaging"
 	"github.com/mudler/LocalAI/core/services/nodes"
 	"github.com/mudler/LocalAI/core/services/workerctl"
+	"github.com/mudler/LocalAI/pkg/model"
 	"github.com/mudler/LocalAI/pkg/system"
 )
 
@@ -136,6 +139,39 @@ var _ = Describe("worker control routes", func() {
 		Expect(string(body)).To(ContainSubstring("control"))
 	})
 
+	It("cuts the echoed path on a rune boundary, so no half rune reaches a log", func() {
+		// The multi-byte rune is placed so that it STRADDLES the cut: a
+		// byte-wise cut lands inside it and the body carries a replacement
+		// character. Phase 2 shipped this exact defect on a refusal reason.
+		// "a" is one byte and "€" is three. The echoed string is the WHOLE
+		// path, prefix included, so the padding is sized against the prefix to
+		// put the rune across bytes max-1, max and max+1.
+		lead := maxEchoedPathBytes - len(workerctl.Prefix) - 1
+		straddle := strings.Repeat("a", lead) + "€" + strings.Repeat("b", 64)
+		resp := post(workerctl.Prefix+straddle, struct{}{})
+		Expect(resp.StatusCode).To(Equal(http.StatusNotFound))
+		body, err := io.ReadAll(resp.Body)
+		Expect(err).NotTo(HaveOccurred())
+
+		Expect(utf8.ValidString(string(body))).To(BeTrue(), "the 404 body must stay valid UTF-8")
+		Expect(string(body)).NotTo(ContainSubstring("\uFFFD"), "the cut split a rune")
+		// And the whole rune is dropped rather than kept past the bound.
+		Expect(string(body)).NotTo(ContainSubstring("€"))
+	})
+
+	It("keeps a rune that ends exactly on the bound, so the cut is not off by one", func() {
+		// Here the rune's last byte is at maxEchoedPathBytes-1, so it fits
+		// entirely and must survive. A cut that walked back unconditionally
+		// would drop it and this spec would catch that.
+		lead := maxEchoedPathBytes - len(workerctl.Prefix) - 3
+		exact := strings.Repeat("a", lead) + "€" + strings.Repeat("b", 64)
+		resp := post(workerctl.Prefix+exact, struct{}{})
+		body, err := io.ReadAll(resp.Body)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(utf8.ValidString(string(body))).To(BeTrue())
+		Expect(string(body)).To(ContainSubstring("€"))
+	})
+
 	It("bounds the unknown path it echoes back, so a long URL cannot be reflected wholesale", func() {
 		long := strings.Repeat("a", 4096)
 		resp := post(workerctl.Prefix+long, struct{}{})
@@ -196,6 +232,8 @@ var _ = Describe("worker control routes", func() {
 	})
 
 	It("answers model.unload with the worker's own reply rather than a transport error", func() {
+		// Nothing is loaded, so there is nothing to free and the true answer is
+		// success.
 		resp := post(workerctl.PathModelUnload, messaging.ModelUnloadRequest{ModelName: "m"})
 		Expect(resp.StatusCode).To(Equal(http.StatusOK))
 		var reply messaging.ModelUnloadReply
@@ -203,6 +241,26 @@ var _ = Describe("worker control routes", func() {
 		Expect(reply.Success).To(BeTrue())
 	})
 
+	It("reports a failed Free as a failed unload, not as success", func() {
+		// A worker that answers "done" about work it did not do is the fourth
+		// kind of answer in this programme's taxonomy, and it is the one acted
+		// on: the frontend's only caller of unload is EvictLRU, so a false yes
+		// tells the scheduler VRAM was released and lets it place the next
+		// model on a node still holding the old one.
+		dead, err := net.Listen("tcp", "127.0.0.1:0")
+		Expect(err).NotTo(HaveOccurred())
+		addr := dead.Addr().String()
+		Expect(dead.Close()).To(Succeed())
+
+		resp := post(workerctl.PathModelUnload, messaging.ModelUnloadRequest{ModelName: "m", Address: addr})
+		// Still a 200: the WORKER answered. Only the verdict inside is negative.
+		Expect(resp.StatusCode).To(Equal(http.StatusOK))
+		var reply messaging.ModelUnloadReply
+		Expect(json.NewDecoder(resp.Body).Decode(&reply)).To(Succeed())
+		Expect(reply.Success).To(BeFalse())
+		Expect(reply.Error).To(ContainSubstring(addr))
+	})
+
 	It("answers model.stop for an unknown process with the worker's verdict, not a 5xx", func() {
 		// The whole invariant of the phase: "that backend is not there" is the
 		// WORKER answering, and it must never arrive as the status code a
@@ -522,3 +580,137 @@ var _ = Describe("the NDJSON control stream", func() {
 		}
 	})
 })
+
+// These specs drive the REAL installBackend and upgradeBackend, with no
+// installFn override, so the progress wiring inside them is exercised end to
+// end through the carrier. Every other install spec scripts installFn, which
+// leaves the one line that decides whether a caller sees any progress at all
+// pinned by nothing.
+var _ = Describe("the real install progress wiring", func() {
+	var (
+		sup *backendSupervisor
+		srv *httptest.Server
+	)
+
+	BeforeEach(func() {
+		dir := GinkgoT().TempDir()
+		st, err := system.GetSystemState(
+			system.WithModelPath(dir),
+			system.WithBackendPath(filepath.Join(dir, "backends")),
+			system.WithBackendSystemPath(filepath.Join(dir, "backends-system")),
+		)
+		Expect(err).NotTo(HaveOccurred())
+		sup = &backendSupervisor{
+			cfg:         &Config{ModelsPath: dir, BackendsPath: filepath.Join(dir, "backends")},
+			systemState: st,
+			ml:          model.NewModelLoader(st),
+			nodeID:      "node-under-test",
+			sigCh:       make(chan os.Signal, 1),
+			processes:   map[string]*backendProcess{},
+			// Empty on purpose. The install cannot succeed, which is the point:
+			// what is under test is that the caller is told what the worker is
+			// doing and then told it failed, not that a download works.
+			galleries: nil,
+		}
+		mux := http.NewServeMux()
+		sup.RegisterControlRoutes(mux)
+		srv = httptest.NewServer(mux)
+		DeferCleanup(srv.Close)
+	})
+
+	postJSON := func(path string, body any) *http.Response {
+		GinkgoHelper()
+		buf, err := json.Marshal(body)
+		Expect(err).NotTo(HaveOccurred())
+		resp, err := srv.Client().Post(srv.URL+path, "application/json", bytes.NewReader(buf))
+		Expect(err).NotTo(HaveOccurred())
+		DeferCleanup(func() { _ = resp.Body.Close() })
+		return resp
+	}
+
+	// envelopesOf decodes a whole NDJSON body into its envelopes.
+	envelopesOf := func(body io.Reader) []workerctl.Envelope {
+		GinkgoHelper()
+		var out []workerctl.Envelope
+		dec := json.NewDecoder(body)
+		for {
+			var env workerctl.Envelope
+			err := dec.Decode(&env)
+			if errors.Is(err, io.EOF) {
+				break
+			}
+			Expect(err).NotTo(HaveOccurred())
+			out = append(out, env)
+		}
+		return out
+	}
+
+	It("streams a resolving line before the failure, from the real installBackend", func() {
+		resp := postJSON(workerctl.PathBackendInstall,
+			messaging.BackendInstallRequest{Backend: "no-such-backend", OpID: "op-real"})
+		Expect(resp.StatusCode).To(Equal(http.StatusOK))
+
+		envs := envelopesOf(resp.Body)
+		Expect(envs).NotTo(BeEmpty())
+		Expect(envs[0].Progress).NotTo(BeNil(), "the caller must see a line before the gallery work starts")
+
+		var ev messaging.BackendInstallProgressEvent
+		Expect(json.Unmarshal(envs[0].Progress, &ev)).To(Succeed())
+		Expect(ev.Phase).To(Equal(messaging.PhaseResolving))
+		Expect(ev.OpID).To(Equal("op-real"))
+		Expect(ev.NodeID).To(Equal("node-under-test"))
+		Expect(ev.Backend).To(Equal("no-such-backend"))
+
+		last := envs[len(envs)-1]
+		Expect(last.Reply).NotTo(BeNil())
+		var reply messaging.BackendInstallReply
+		Expect(json.Unmarshal(last.Reply, &reply)).To(Succeed())
+		Expect(reply.Success).To(BeFalse())
+	})
+
+	It("streams nothing but the reply when the caller asked for no progress", func() {
+		// An empty OpID is a reconciler-driven retry. It must not be given a
+		// progress stream, and it must still get its terminal line.
+		resp := postJSON(workerctl.PathBackendInstall,
+			messaging.BackendInstallRequest{Backend: "no-such-backend"})
+		Expect(envelopeKindsOf(resp.Body)).To(Equal([]string{"reply"}))
+	})
+
+	It("streams a resolving line from the real upgradeBackend too", func() {
+		resp := postJSON(workerctl.PathBackendUpgrade,
+			messaging.BackendUpgradeRequest{Backend: "no-such-backend", OpID: "op-real-upgrade"})
+		envs := envelopesOf(resp.Body)
+		Expect(envs).NotTo(BeEmpty())
+		Expect(envs[0].Progress).NotTo(BeNil())
+
+		var ev messaging.BackendInstallProgressEvent
+		Expect(json.Unmarshal(envs[0].Progress, &ev)).To(Succeed())
+		Expect(ev.Phase).To(Equal(messaging.PhaseResolving))
+		Expect(ev.OpID).To(Equal("op-real-upgrade"))
+
+		var reply messaging.BackendUpgradeReply
+		Expect(json.Unmarshal(envs[len(envs)-1].Reply, &reply)).To(Succeed())
+		Expect(reply.Success).To(BeFalse())
+	})
+
+	It("hands the gallery no download callback when either half of the guard is missing", func() {
+		// A nil callback is what puts the gallery on its silent path, and both
+		// halves have to be checked: a caller with an OpID but no sink is the
+		// shape that would nil-panic on the resolving emit.
+		collected := func(messaging.BackendInstallProgressEvent) {}
+
+		cb, flush := sup.startProgress("", "b", collected)
+		Expect(cb).To(BeNil())
+		Expect(flush).NotTo(BeNil())
+		flush()
+
+		cb, flush = sup.startProgress("op", "b", nil)
+		Expect(cb).To(BeNil())
+		Expect(flush).NotTo(BeNil())
+		flush()
+
+		cb, flush = sup.startProgress("op", "b", collected)
+		Expect(cb).NotTo(BeNil())
+		flush()
+	})
+})

core/services/worker/install.go

@@ -134,19 +134,11 @@ func (s *backendSupervisor) installBackend(ctx context.Context, req messaging.Ba
 		galleries = reqGalleries
 	}
 
-	// When the caller tagged this install with an OpID and is listening for
-	// progress, stream the gallery download ticks back on the response the
-	// caller is already reading. Callers that omit OpID stay on the silent path.
-	// The publisher releases its mutex before every emit so a slow link never
-	// stalls the download loop, and the deferred Flush guarantees a
-	// terminal-percentage event reaches the caller even when the install errors
-	// out.
-	var downloadCb func(file, current, total string, percentage float64)
-	if req.OpID != "" && onProgress != nil {
-		publisher := nodes.NewDebouncedInstallProgressSink(onProgress, s.nodeID, req.OpID, req.Backend, installProgressDebounce)
-		downloadCb = publisher.OnDownload
-		defer publisher.Flush()
-	}
+	// Gallery download ticks go back on the response the caller is already
+	// reading. See startProgress for the guard and for why the flush is
+	// deferred.
+	downloadCb, flushProgress := s.startProgress(req.OpID, req.Backend, onProgress)
+	defer flushProgress()
 
 	// On upgrade, run the gallery install path even if the binary already
 	// exists on disk: findBackend would otherwise short-circuit and we'd
@@ -235,17 +227,10 @@ func (s *backendSupervisor) upgradeBackend(ctx context.Context, req messaging.Ba
 		galleries = reqGalleries
 	}
 
-	// When the caller tagged this upgrade with an OpID, stream gallery download
-	// progress back on the same sink install uses — an upgrade IS a
-	// force-reinstall. Callers that omit OpID stay on the silent path. The
-	// deferred Flush guarantees a terminal-percentage event even if the upgrade
-	// errors out, so the caller's per-node bar never hangs mid-download.
-	var downloadCb func(file, current, total string, percentage float64)
-	if req.OpID != "" && onProgress != nil {
-		publisher := nodes.NewDebouncedInstallProgressSink(onProgress, s.nodeID, req.OpID, req.Backend, installProgressDebounce)
-		downloadCb = publisher.OnDownload
-		defer publisher.Flush()
-	}
+	// The same sink install uses: an upgrade IS a force-reinstall, so its
+	// progress is install progress.
+	downloadCb, flushProgress := s.startProgress(req.OpID, req.Backend, onProgress)
+	defer flushProgress()
 
 	if req.URI != "" {
 		xlog.Info("Upgrading backend from external URI", "backend", req.Backend, "uri", req.URI)
@@ -309,3 +294,34 @@ func (s *backendSupervisor) lockBackend(name string) func() {
 	m.Lock()
 	return m.Unlock
 }
+
+// startProgress wires one install or upgrade to its caller's progress sink.
+//
+// It returns the download callback the gallery takes, and a flush the caller
+// must defer: the debouncer buffers within its window, and the deferred flush
+// is what gets the terminal percentage to the caller even when the install
+// errors out.
+//
+// Both halves of the guard matter. An empty OpID means the caller is a
+// reconciler-driven retry that asked for no progress, and a nil sink means the
+// caller is not reading a stream at all; either way the gallery gets a nil
+// callback and takes its silent path.
+//
+// The resolving event is emitted here, before any gallery work, so the caller
+// sees a line as soon as the worker has accepted the job rather than only once
+// bytes start moving. A cold install spends minutes resolving a manifest, and
+// during that time a progress stream with nothing on it is indistinguishable
+// from one that is broken.
+func (s *backendSupervisor) startProgress(opID, backend string, onProgress func(messaging.BackendInstallProgressEvent)) (func(file, current, total string, percentage float64), func()) {
+	if opID == "" || onProgress == nil {
+		return nil, func() {}
+	}
+	publisher := nodes.NewDebouncedInstallProgressSink(onProgress, s.nodeID, opID, backend, installProgressDebounce)
+	onProgress(messaging.BackendInstallProgressEvent{
+		OpID:    opID,
+		NodeID:  s.nodeID,
+		Backend: backend,
+		Phase:   messaging.PhaseResolving,
+	})
+	return publisher.OnDownload, publisher.Flush
+}

core/services/worker/lifecycle.go

@@ -183,18 +183,34 @@ func (s *backendSupervisor) unloadModel(ctx context.Context, req messaging.Model
 		s.mu.Unlock()
 	}
 
-	if targetAddr != "" {
-		// Best-effort bounded gRPC Free(). A model.unload request must not
-		// occupy the handler forever when a backend is wedged. The bound is
-		// derived from the caller's own budget where it has one, so a caller
-		// that allowed less than workerBackendFreeTimeout is not made to wait
-		// longer than it asked for.
-		client := grpc.NewClientWithToken(targetAddr, false, nil, false, s.cfg.RegistrationToken)
-		freeCtx, cancel := context.WithTimeout(ctx, workerBackendFreeTimeout)
-		if err := client.Free(freeCtx); err != nil {
-			xlog.Warn("Free() failed during model.unload", "error", err, "addr", targetAddr)
+	if targetAddr == "" {
+		// Nothing is loaded here, so there is nothing to free. That is the
+		// worker's own answer and it is a true one, not a claim about work it
+		// performed.
+		return messaging.ModelUnloadReply{Success: true}
+	}
+
+	// Bounded gRPC Free(). A model.unload request must not occupy the handler
+	// forever when a backend is wedged. The bound is derived from the caller's
+	// own budget, so a caller that allowed less than workerBackendFreeTimeout
+	// is not made to wait longer than it asked for.
+	client := grpc.NewClientWithToken(targetAddr, false, nil, false, s.cfg.RegistrationToken)
+	freeCtx, cancel := context.WithTimeout(ctx, workerBackendFreeTimeout)
+	err := client.Free(freeCtx)
+	cancel()
+	if err != nil {
+		// Reported, not swallowed. This used to answer Success:true whatever
+		// Free did, which is the worker saying "done" about something it did
+		// not do: the frontend's only caller is EvictLRU, so a false yes told
+		// the scheduler VRAM had been released and let it place the next model
+		// on a node still holding the old one. A failure here is the worker's
+		// own verdict about one Free call and nothing more; it is not a
+		// statement that the node or the model is gone.
+		xlog.Warn("Free() failed during model.unload", "error", err, "addr", targetAddr)
+		return messaging.ModelUnloadReply{
+			Success: false,
+			Error:   fmt.Sprintf("freeing model on %s: %v", targetAddr, err),
 		}
-		cancel()
 	}
 
 	return messaging.ModelUnloadReply{Success: true}
@@ -215,7 +231,7 @@ func (s *backendSupervisor) deleteModel(req messaging.ModelDeleteRequest) messag
 // signalNodeStop serves node.stop: it triggers the normal shutdown path via
 // sigCh so deferred cleanup runs, rather than exiting the process here.
 func (s *backendSupervisor) signalNodeStop() {
-	xlog.Info("Serving node.stop — signaling shutdown")
+	xlog.Info("Serving node.stop, signaling shutdown")
 	select {
 	case s.sigCh <- syscall.SIGTERM:
 	default:

core/services/worker/supervisor.go

@@ -849,6 +849,12 @@ func (s *backendSupervisor) stopBackendExact(key string, force bool) error {
 	}
 
 	if !force {
+		// Background and not a caller's context on purpose. This Free is a
+		// courtesy before the process is killed anyway, and the stop that
+		// follows must complete for the port to be released, so binding it to
+		// a caller that may have gone would buy nothing and could abandon a
+		// half-finished stop. model.unload, where Free IS the operation, takes
+		// the caller's budget instead; see RegisterControlRoutes.
 		client := grpc.NewClientWithToken(bp.addr, false, nil, false, s.cfg.RegistrationToken)
 		freeCtx, cancel := context.WithTimeout(context.Background(), workerBackendFreeTimeout)
 		xlog.Debug("Calling bounded Free() before stopping backend", "backend", key, "timeout", workerBackendFreeTimeout)
@@ -895,6 +901,8 @@ func (s *backendSupervisor) stopModelExact(req messaging.ModelStopRequest) messa
 	s.mu.Unlock()
 
 	if !req.Force {
+		// Background, for the reason given in stopBackendExact: this is the
+		// acknowledged stop and it has to run to completion.
 		client := grpc.NewClientWithToken(bp.addr, false, nil, false, s.cfg.RegistrationToken)
 		freeCtx, cancel := context.WithTimeout(context.Background(), workerBackendFreeTimeout)
 		freeErr := client.Free(freeCtx)

core/services/workerctl/paths_test.go

@@ -40,10 +40,24 @@ var _ = Describe("control plane paths on the wire", func() {
 		}
 	})
 
-	It("enumerates every verb, so a new one cannot be added without the prefix check seeing it", func() {
-		Expect(workerctl.AllPaths()).To(HaveLen(10))
-		Expect(workerctl.AllPaths()).To(ContainElement(workerctl.PathBackendInstall))
-		Expect(workerctl.AllPaths()).To(ContainElement(workerctl.PathNodeStop))
+	It("lists every verb this package names, so none can be dropped from the set", func() {
+		// The claim is bounded on purpose. Go constants are not enumerable, so
+		// nothing here can see a NEW constant that was never added to AllPaths;
+		// what this catches is an EXISTING verb going missing from it, which
+		// matters because the prefix check above and the worker's mounting spec
+		// both iterate AllPaths and would silently stop covering it.
+		Expect(workerctl.AllPaths()).To(ConsistOf(
+			workerctl.PathBackendInstall,
+			workerctl.PathBackendUpgrade,
+			workerctl.PathBackendList,
+			workerctl.PathBackendStop,
+			workerctl.PathBackendDelete,
+			workerctl.PathModelStop,
+			workerctl.PathModelUnload,
+			workerctl.PathModelDelete,
+			workerctl.PathModelsRunning,
+			workerctl.PathNodeStop,
+		))
 	})
 
 	It("gives each verb a distinct path", func() {

docs/content/features/distributed-mode.md

@@ -186,7 +186,11 @@ Two of the routes stream. `POST /v1/control/backend/install` and
 `/v1/control/backend/upgrade` answer with `application/x-ndjson`: zero or more
 `{"progress":{...}}` lines carrying the same download-progress payload the
 per-op NATS subject carried, followed by exactly one `{"reply":{...}}` line,
-which is always the last line on the body. An install that FAILS is still a
+which is always the last line on the body. When the request carries an `op_id`,
+the first progress line has phase `resolving` and is written before any gallery
+work begins, so a cold install that spends minutes on a manifest is
+distinguishable from a stream that is broken. Nothing publishes install progress
+over NATS any more. An install that FAILS is still a
 `200` with a reply whose `success` is `false`. That is deliberate, and it is the
 same distinction the refusal table above draws: a non-2xx means the frontend
 could not get the request to the worker, which nothing may act on, while the