fix(addons): do not panic on an unknown upgrade phase

upgradeTransition ended in panic for a phase or event it did not know.
The phase of a submitted operation is read from the journal on disk, so a
corrupted file would crash a background goroutine. The reducer now
returns errInvalidUpgradeTransition, and submit marks the operation as
failed and returns ErrOperationFailed. The current call paths only pass
known phases, so this is a guard, not a fix of a known crash.

upgradeViews also grew without bound, and every persist walked it. Views
of finished upgrades are now kept for the newest 256 only; the journal
still answers for older ones. The journal itself is rewritten on each
persist, so the walk stays proportional to what is written anyway.

Assisted-by: Claude Code:claude-sonnet-5-5
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

core/services/addons/manager.go

@@ -102,6 +102,7 @@ type Manager struct {
 	mu                            sync.Mutex
 	upgradeMu                     sync.Mutex
 	upgradeViews                  map[string]upgradeSnapshot
+	finishedUpgrades              []string // oldest first; guarded by upgradeMu
 	upgradeCtx                    context.Context
 	upgradeCancel                 context.CancelFunc
 	upgradeWG                     sync.WaitGroup
@@ -260,7 +261,7 @@ func NewManager(ctx context.Context, cfg OwnerConfig, c catalog.Catalog, i *pack
 		_ = release()
 		return nil, e
 	}
-	m.refreshUpgradeViews()
+	m.syncUpgradeViews(true)
 	return m, nil
 }
 func (m *Manager) Close() error {
@@ -465,7 +466,19 @@ func (m *Manager) submit(ctx context.Context, actor contract.Actor, cmd Lifecycl
 		return OperationStatus{}, ErrConflict
 	}
 	if accepted {
-		phase, _ := upgradeTransition(lifecyclePhase(op.Status.Phase), upgradeStart)
+		phase, _, err := upgradeTransition(lifecyclePhase(op.Status.Phase), upgradeStart)
+		if err != nil {
+			// A phase the reducer does not know cannot be run. End the
+			// operation as failed instead of leaving it half-owned.
+			xlog.Warn("Addon operation has an unknown phase", "operationId", cmd.OperationID, "phase", op.Status.Phase)
+			op.Status.Phase = string(lifecycleFailed)
+			op.Status.FailureCode = "upgrade_failed"
+			m.journal.Operations[cmd.OperationID] = op
+			if e := m.persist(string(lifecycleFailed)); e != nil {
+				return op.Status, e
+			}
+			return op.Status, ErrOperationFailed
+		}
 		op.Status.Phase = string(phase)
 		m.journal.Operations[cmd.OperationID] = op
 		if e := m.persist(string(lifecycleRunning)); e != nil {

core/services/addons/upgrade_operation.go

@@ -6,10 +6,14 @@ import (
 	"errors"
 	"github.com/mudler/LocalAI/pkg/addons/contract"
 	"github.com/mudler/xlog"
+	"slices"
 )
 
 type upgradeExecutionKey struct{}
 
+// maxFinishedUpgradeViews bounds the views kept for finished upgrades.
+const maxFinishedUpgradeViews = 256
+
 type upgradeSnapshot struct {
 	Command LifecycleCommand
 	Actor   contract.Actor
@@ -99,8 +103,7 @@ func (m *Manager) SubmitUpgrade(ctx context.Context, actor contract.Actor, cmd L
 		m.mu.Lock()
 		current := m.journal.Operations[cmd.OperationID]
 		if err != nil && current.Status.Phase != string(lifecycleCompleted) && current.Status.Phase != string(lifecycleFailed) && !m.uncertain {
-			phase, _ := upgradeTransition(lifecycleRunning, upgradeFail)
-			current.Status.Phase = string(phase)
+			current.Status.Phase = string(lifecycleFailed)
 			current.Status.FailureCode = "upgrade_failed"
 			if errors.Is(err, context.Canceled) {
 				current.Status.FailureCode = "interrupted"
@@ -124,17 +127,39 @@ func (m *Manager) SubmitUpgrade(ctx context.Context, actor contract.Actor, cmd L
 
 // refreshUpgradeViews is called only while mu is held. Values contain no
 // mutable slices or pointers and may be returned without the execution lock.
-func (m *Manager) refreshUpgradeViews() {
+//
+// Finished operations keep a view only while they are among the newest
+// maxFinishedUpgradeViews; the journal still answers for older ones.
+func (m *Manager) refreshUpgradeViews() { m.syncUpgradeViews(false) }
+
+// syncUpgradeViews projects the journal. At start (loadFinished) the finished
+// operations of earlier processes get a view too, so a status read keeps its
+// failure code; afterwards only operations that already have one do.
+func (m *Manager) syncUpgradeViews(loadFinished bool) {
 	m.upgradeMu.Lock()
 	defer m.upgradeMu.Unlock()
 	for id, op := range m.journal.Operations {
-		if op.Async {
-			status := op.Status
-			if m.uncertain || m.runtimeRecoveryRequired {
-				status.RuntimeState = string(lifecycleRecovery)
-			}
-			m.upgradeViews[id] = upgradeSnapshot{op.Command, op.Actor, status}
+		if !op.Async {
+			continue
 		}
+		_, viewed := m.upgradeViews[id]
+		if !isActive(op.Status.Phase) {
+			if !viewed && !loadFinished {
+				continue
+			}
+			if !slices.Contains(m.finishedUpgrades, id) {
+				m.finishedUpgrades = append(m.finishedUpgrades, id)
+			}
+		}
+		status := op.Status
+		if m.uncertain || m.runtimeRecoveryRequired {
+			status.RuntimeState = string(lifecycleRecovery)
+		}
+		m.upgradeViews[id] = upgradeSnapshot{op.Command, op.Actor, status}
+	}
+	for len(m.finishedUpgrades) > maxFinishedUpgradeViews {
+		delete(m.upgradeViews, m.finishedUpgrades[0])
+		m.finishedUpgrades = m.finishedUpgrades[1:]
 	}
 }
 

core/services/addons/upgrade_state.go

@@ -1,6 +1,13 @@
 // SPDX-License-Identifier: MIT
 package addons
 
+import "errors"
+
+// errInvalidUpgradeTransition means the phase or the event is not one the
+// reducer knows. The phase can come from a journal on disk, so it is an error
+// for the caller to handle, not a reason to crash a worker.
+var errInvalidUpgradeTransition = errors.New("invalid upgrade state or event")
+
 // upgradeTransition is the owner's observation reducer. Effects remain in the
 // existing lifecycle and recovery journals; observer detach never emits work.
 // Every terminal result is sticky, including after late progress or a retry.
@@ -16,25 +23,31 @@ const (
 	upgradeRetry     upgradeEvent = "retry"
 )
 
-func upgradeTransition(phase lifecyclePhase, event upgradeEvent) (next lifecyclePhase, rule string) {
+func upgradeTransition(phase lifecyclePhase, event upgradeEvent) (next lifecyclePhase, rule string, err error) {
 	switch phase {
 	case lifecycleCompleted, lifecycleFailed, lifecycleRecovery:
-		return phase, "U0-terminal"
+		return phase, "U0-terminal", nil
 	case lifecycleAccepted, lifecycleRunning:
 		switch event {
 		case upgradeDetach, upgradeRetry:
-			return phase, "U1-observe"
+			return phase, "U1-observe", nil
 		case upgradeStart:
-			return lifecycleRunning, "U2-start"
+			return lifecycleRunning, "U2-start", nil
 		case upgradeComplete:
-			return lifecycleCompleted, "U3-complete"
+			return lifecycleCompleted, "U3-complete", nil
 		case upgradeFail:
-			return lifecycleFailed, "U4-fail"
+			return lifecycleFailed, "U4-fail", nil
 		case upgradeInterrupt:
-			return lifecycleFailed, "U5-interrupted"
+			return lifecycleFailed, "U5-interrupted", nil
 		case upgradeRecover:
-			return lifecycleRecovery, "U6-recovery"
+			return lifecycleRecovery, "U6-recovery", nil
 		}
 	}
-	panic("invalid upgrade state or event")
+	return phase, "", errInvalidUpgradeTransition
+}
+
+// isActive reports whether an operation in this phase still blocks its add-on:
+// anything but completed or failed, including recovery-required.
+func isActive(phase string) bool {
+	return phase != string(lifecycleCompleted) && phase != string(lifecycleFailed)
 }

core/services/addons/upgrade_state_test.go

@@ -2,6 +2,8 @@
 package addons
 
 import (
+	"fmt"
+
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 )
@@ -22,7 +24,7 @@ var _ = Describe("upgrade observation reducer", func() {
 			phase := lifecycleAccepted
 			for n, event := range trace.events {
 				var rule string
-				phase, rule = upgradeTransition(phase, event)
+				phase, rule, _ = upgradeTransition(phase, event)
 				Expect(rule).To(Equal(trace.rules[n]))
 			}
 			Expect(phase).To(Equal(trace.end))
@@ -32,21 +34,49 @@ var _ = Describe("upgrade observation reducer", func() {
 		events := []upgradeEvent{upgradeStart, upgradeRetry, upgradeDetach, upgradeComplete, upgradeFail, upgradeInterrupt, upgradeRecover}
 		for _, phase := range []lifecyclePhase{lifecycleAccepted, lifecycleRunning, lifecycleCompleted, lifecycleFailed, lifecycleRecovery} {
 			for _, event := range []upgradeEvent{upgradeStart, upgradeComplete, upgradeFail, upgradeInterrupt, upgradeRecover, upgradeDetach, upgradeRetry} {
-				next, rule := upgradeTransition(phase, event)
+				next, rule, _ := upgradeTransition(phase, event)
 				Expect(next).NotTo(BeEmpty())
 				Expect(rule).NotTo(BeEmpty())
 			}
 		}
 		for _, phase := range []lifecyclePhase{lifecycleCompleted, lifecycleFailed, lifecycleRecovery} {
 			for _, event := range events {
-				next, _ := upgradeTransition(phase, event)
+				next, _, _ := upgradeTransition(phase, event)
 				Expect(next).To(Equal(phase))
 			}
 		}
 		for _, phase := range []lifecyclePhase{lifecycleAccepted, lifecycleRunning} {
-			next, rule := upgradeTransition(phase, upgradeInterrupt)
+			next, rule, _ := upgradeTransition(phase, upgradeInterrupt)
 			Expect(next).To(Equal(lifecycleFailed))
 			Expect(rule).To(Equal("U5-interrupted"))
 		}
 	})
+	It("refuses an unknown phase or event with an error instead of panicking", func() {
+		for _, phase := range []lifecyclePhase{"", "garbage", lifecycleIntent} {
+			next, _, err := upgradeTransition(phase, upgradeStart)
+			Expect(err).To(MatchError(errInvalidUpgradeTransition))
+			Expect(next).To(Equal(phase))
+		}
+		_, _, err := upgradeTransition(lifecycleAccepted, upgradeEvent("bogus"))
+		Expect(err).To(MatchError(errInvalidUpgradeTransition))
+	})
+	It("drops the oldest finished views once too many pile up", func() {
+		m := &Manager{upgradeViews: map[string]upgradeSnapshot{}, journal: journal{Operations: map[string]journalOperation{}}}
+		for i := 0; i < maxFinishedUpgradeViews+10; i++ {
+			id := fmt.Sprintf("op-%04d", i)
+			m.journal.Operations[id] = journalOperation{Async: true, Status: OperationStatus{OperationID: id, Phase: string(lifecycleAccepted)}}
+			m.refreshUpgradeViews()
+			op := m.journal.Operations[id]
+			op.Status.Phase = string(lifecycleCompleted)
+			m.journal.Operations[id] = op
+			m.refreshUpgradeViews()
+		}
+		m.journal.Operations["live"] = journalOperation{Async: true, Status: OperationStatus{OperationID: "live", Phase: string(lifecycleRunning)}}
+		m.refreshUpgradeViews()
+		Expect(m.upgradeViews).To(HaveLen(maxFinishedUpgradeViews + 1))
+		Expect(m.upgradeViews).To(HaveKey("live"))
+		Expect(m.upgradeViews).NotTo(HaveKey("op-0000"))
+		Expect(m.upgradeViews).To(HaveKey(fmt.Sprintf("op-%04d", maxFinishedUpgradeViews+9)))
+
+	})
 })