fix: Worker 池评审修复（系统会话可见报错、暂停意图续跑保持绑定模型、加载失败不清空池、原子写入）

Code authored by Codex CLI gpt-6.1-sol (addresses Grok review #1).

server/engine.go

@@ -150,6 +150,7 @@ type Engine struct {
 	resolveAuthoritative bool
 	workerResolve        func(t *Task, slot WorkerSlot) (*agent.Worker, error)
 	slotState            sync.Map // task/slot -> slotAvailability
+	slotPlans            sync.Map // task -> map[slot name]profile ID; immutable for a run
 	// readiness reports whether a global LLM provider is configured — the signal behind
 	// Ready()/the llm_configured indicator. Wired once at startup; nil → not ready.
 	readiness func() bool
@@ -292,6 +293,7 @@ func (e *Engine) StopTask(taskID string) {
 	e.stamped.Delete(taskID)
 	e.inflight.Delete(taskID)
 	e.coordStarted.Delete(taskID)
+	e.slotPlans.Delete(taskID)
 	e.slotState.Range(func(key, _ any) bool {
 		keyText, ok := key.(string)
 		if ok && strings.HasPrefix(keyText, taskID+"/") {
@@ -741,7 +743,7 @@ func (e *Engine) reportSlotAvailability(t *Task, slot WorkerSlot, err error) {
 	if err == nil {
 		if loaded && !state.available {
 			e.slotState.Store(key, slotAvailability{available: true, lastEmit: now})
-			e.emitActivity(t, db.Activity{Worker: slot.Name, Kind: "text", Summary: slot.Name + " 绑定的 LLM 配置已恢复，可以继续领取任务"})
+			e.emitActivity(t, db.Activity{Worker: "system", Kind: "text", Summary: fmt.Sprintf("%s 绑定的 LLM 配置 #%d 已恢复，可以继续领取任务", slot.Name, slot.ProfileID)})
 		}
 		return
 	}
@@ -756,7 +758,7 @@ func (e *Engine) reportSlotAvailability(t *Task, slot WorkerSlot, err error) {
 		e.slotState.Store(key, state)
 	}
 	summary := fmt.Sprintf("%s 绑定的 LLM 配置 #%d 不可用（已删除或配置无效），该会话暂停领取任务，不会回退到默认模型", slot.Name, slot.ProfileID)
-	e.emitActivity(t, db.Activity{Worker: slot.Name, Kind: "text", IsError: true, Summary: summary, Detail: err.Error()})
+	e.emitActivity(t, db.Activity{Worker: "system", Kind: "text", IsError: true, Summary: summary, Detail: err.Error()})
 	log.Printf("[worker %s] task %s pinned profile #%d unavailable: %v", slot.Name, t.ID, slot.ProfileID, err)
 }
 
@@ -807,6 +809,11 @@ func (e *Engine) Run(ctx context.Context, t *Task) {
 		return
 	}
 	rt := e.registerTaskRoutines(ctx, t.ID, 1+len(slots))
+	plan := make(map[string]int64, len(slots))
+	for _, slot := range slots {
+		plan[slot.Name] = slot.ProfileID
+	}
+	e.slotPlans.Store(t.ID, plan)
 	e.deleteMu.RUnlock()
 	e.touch(t.ID)
 	runTaskRoutine(rt, func(loopCtx context.Context) { e.plannerLoop(loopCtx, t) })
@@ -1245,10 +1252,6 @@ func (e *Engine) runDetachedIntent(ctx context.Context, t *Task, intentID int64,
 			e.decInflight(t.ID)
 		}
 	}()
-	_, worker := e.snapshotFor(t)
-	if worker == nil {
-		return fmt.Errorf("worker 尚未就绪")
-	}
 	node, err := t.Store.GetNode(intentID)
 	if err != nil {
 		return err
@@ -1256,6 +1259,30 @@ func (e *Engine) runDetachedIntent(ctx context.Context, t *Task, intentID int64,
 	if node == nil || node.Kind != db.KindIntent {
 		return fmt.Errorf("intent not found")
 	}
+	owner := node.Owner
+	var worker *agent.Worker
+	if value, ok := e.slotPlans.Load(t.ID); ok {
+		if plan, ok := value.(map[string]int64); ok {
+			if profileID, pinned := plan[owner]; pinned && profileID > 0 {
+				if e.workerResolve == nil {
+					return fmt.Errorf("意图所属的 LLM 配置 #%d 不可用：pinned worker resolver unavailable", profileID)
+				}
+				worker, err = e.workerResolve(t, WorkerSlot{Name: owner, ProfileID: profileID})
+				if err != nil || worker == nil {
+					if err == nil {
+						err = errors.New("worker resolver returned no worker")
+					}
+					return fmt.Errorf("意图所属的 LLM 配置 #%d 不可用：%w", profileID, err)
+				}
+			}
+		}
+	}
+	if worker == nil {
+		_, worker = e.snapshotFor(t)
+		if worker == nil {
+			return fmt.Errorf("worker 尚未就绪")
+		}
+	}
 	changed, err := t.Store.CompareAndSetIntentState(intentID, "paused", "running")
 	if err != nil {
 		return err

server/manager.go

@@ -1,6 +1,7 @@
 package server
 
 import (
+	"context"
 	"database/sql"
 	"encoding/json"
 	"errors"
@@ -444,9 +445,10 @@ func (m *Manager) WorkerPools() []WorkerPool {
 	return pools
 }
 
-// SetWorkerPools validates profile references at save time and synchronizes the
-// legacy workers setting with the pool total. An empty slice restores legacy mode.
-func (m *Manager) SetWorkerPools(pools []WorkerPool) error {
+// SetWorkerPools validates profile references at save time and atomically
+// synchronizes the legacy workers setting with the pool total. An empty slice
+// restores legacy mode; an optional legacy count is used when clearing pools.
+func (m *Manager) SetWorkerPools(pools []WorkerPool, legacyWorkers ...int) error {
 	if pools == nil {
 		pools = []WorkerPool{}
 	}
@@ -468,17 +470,22 @@ func (m *Manager) SetWorkerPools(pools []WorkerPool) error {
 	if err != nil {
 		return err
 	}
-	if err := m.pg.SetSetting(settingWorkerPools, string(b)); err != nil {
-		return err
-	}
+	workerCount := m.Workers()
 	if len(pools) > 0 {
-		total := 0
+		workerCount = 0
 		for _, pool := range pools {
-			total += pool.Workers
+			workerCount += pool.Workers
 		}
-		return m.pg.SetSetting(settingWorkers, strconv.Itoa(total))
+	} else if len(legacyWorkers) > 0 {
+		if legacyWorkers[0] <= 0 {
+			return fmt.Errorf("workers 必须 >0")
+		}
+		workerCount = legacyWorkers[0]
 	}
-	return nil
+	return m.pg.SetSettingsAtomic(context.Background(), map[string]string{
+		settingWorkerPools: string(b),
+		settingWorkers:     strconv.Itoa(workerCount),
+	})
 }
 
 // SetWorkers persists the concurrent work-agent count. Values <=0 are rejected.

server/server.go

@@ -3593,16 +3593,14 @@ func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) {
 			writeErr(w, 400, "workers 必须 >0")
 			return
 		}
-		if err := s.m.SetWorkerPools(*req.WorkerPools); err != nil {
+		legacyWorkers := []int(nil)
+		if req.Workers != nil {
+			legacyWorkers = []int{*req.Workers}
+		}
+		if err := s.m.SetWorkerPools(*req.WorkerPools, legacyWorkers...); err != nil {
 			writeErr(w, 400, err.Error())
 			return
 		}
-		if len(*req.WorkerPools) == 0 && req.Workers != nil {
-			if err := s.m.SetWorkers(*req.Workers); err != nil {
-				writeErr(w, 400, err.Error())
-				return
-			}
-		}
 	} else if req.Workers != nil {
 		// An omitted worker_pools field leaves the saved mode unchanged. If a
 		// non-empty pool configuration is already saved, its total remains the

server/worker_pools_test.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/Autumn-27/artex/agent"
 	"github.com/Autumn-27/artex/db"
+	"github.com/Autumn-27/norma/llm"
 )
 
 func TestExpandWorkerSlotsLegacy(t *testing.T) {
@@ -210,6 +211,51 @@ func TestPinnedRuntimeMissingProfileDoesNotFallback(t *testing.T) {
 	}
 }
 
+func TestPinnedRuntimeQuotaDoesNotAdvanceTaskFailover(t *testing.T) {
+	m, err := NewManager(t.TempDir(), "")
+	if err != nil {
+		t.Skipf("postgres unavailable: %v", err)
+	}
+	defer m.Close()
+	task, err := m.CreateTask("pinned quota", "test", nil, 0, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer m.DeleteTask(task.ID, DeleteTaskOptions{})
+	profileID, err := m.pg.SaveProfile(&db.LLMProfile{Name: "pinned-quota", Format: "openai", Model: "pinned", APIKey: "key"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer m.pg.DeleteProfile(profileID)
+	if _, err := m.ReplaceTaskLLMProfiles(task.ID, []int64{profileID}, profileID); err != nil {
+		t.Fatal(err)
+	}
+	before, err := m.pg.GetTask(taskIDInt(t, task.ID))
+	if err != nil || before == nil || before.ActiveLLMProfileID == nil {
+		t.Fatalf("load task failover state: task=%+v err=%v", before, err)
+	}
+	provider := &scriptedLLMProvider{err: errors.New("insufficient_quota")}
+	s := &Server{m: m, provByProfile: map[int64]*provEntry{profileID: {prov: provider, cfg: agent.Config{}}}}
+	id := profileID
+	runtime := &taskLLMRuntime{s: s, taskID: task.ID, agentKey: "worker", pinProfileID: &id}
+	if _, _, _, err := runtime.Complete(context.Background(), llm.CompletionRequest{}); err == nil {
+		t.Fatal("pinned quota error unexpectedly succeeded")
+	}
+	after, err := m.pg.GetTask(taskIDInt(t, task.ID))
+	if err != nil || after == nil || after.LLMChainRevision != before.LLMChainRevision || after.ActiveLLMProfileID == nil || *after.ActiveLLMProfileID != *before.ActiveLLMProfileID {
+		t.Fatalf("pinned quota changed task failover state: before=%+v after=%+v err=%v", before, after, err)
+	}
+}
+
+func taskIDInt(t *testing.T, id string) int64 {
+	t.Helper()
+	n, err := strconv.ParseInt(id, 10, 64)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return n
+}
+
 func TestPinnedWorkerLoopDoesNotConsultTaskRouterAndReportsFailureOnce(t *testing.T) {
 	m, err := NewManager(t.TempDir(), "")
 	if err != nil {
@@ -253,7 +299,7 @@ func TestPinnedWorkerLoopDoesNotConsultTaskRouterAndReportsFailureOnce(t *testin
 	}
 	var failures int
 	for _, activity := range activities {
-		if activity.Worker == "work#4" && activity.IsError {
+		if activity.Worker == "system" && activity.IsError {
 			failures++
 		}
 	}
@@ -261,3 +307,87 @@ func TestPinnedWorkerLoopDoesNotConsultTaskRouterAndReportsFailureOnce(t *testin
 		t.Fatalf("pinned failure activities=%d, want one throttled entry: %+v", failures, activities)
 	}
 }
+
+func TestPinnedWorkerAvailabilityRecoveryUsesSystemSession(t *testing.T) {
+	m, err := NewManager(t.TempDir(), "")
+	if err != nil {
+		t.Skipf("postgres unavailable: %v", err)
+	}
+	defer m.Close()
+	task, err := m.CreateTask("pinned availability", "test", nil, 0, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer m.DeleteTask(task.ID, DeleteTaskOptions{})
+	e := NewEngine(m)
+	slot := WorkerSlot{Name: "work#4", ProfileID: 12}
+	e.reportSlotAvailability(task, slot, errors.New("profile unavailable"))
+	e.reportSlotAvailability(task, slot, nil)
+	activities, _, err := task.Store.ActivityList(nil, 0, 20)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var failure, recovery int
+	for _, activity := range activities {
+		if activity.Worker != "system" {
+			continue
+		}
+		if activity.IsError && strings.Contains(activity.Summary, "work#4") && strings.Contains(activity.Summary, "#12") {
+			failure++
+		}
+		if !activity.IsError && strings.Contains(activity.Summary, "work#4") && strings.Contains(activity.Summary, "#12") {
+			recovery++
+		}
+	}
+	if failure != 1 || recovery != 1 {
+		t.Fatalf("system availability activities failure=%d recovery=%d: %+v", failure, recovery, activities)
+	}
+}
+
+func TestDetachedIntentPinnedOwnerUsesPinnedResolver(t *testing.T) {
+	m, err := NewManager(t.TempDir(), "")
+	if err != nil {
+		t.Skipf("postgres unavailable: %v", err)
+	}
+	defer m.Close()
+	task, err := m.CreateTask("detached pinned owner", "test", nil, 0, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer m.DeleteTask(task.ID, DeleteTaskOptions{})
+	intentID, err := task.Store.AddIntent(map[string]any{"text": "continue"}, 0, nil, "test")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if ok, err := task.Store.ClaimIntent(intentID, "work#4"); err != nil || !ok {
+		t.Fatalf("claim intent: ok=%v err=%v", ok, err)
+	}
+	if changed, err := task.Store.CompareAndSetIntentState(intentID, "running", "paused"); err != nil || !changed {
+		t.Fatalf("pause intent: changed=%v err=%v", changed, err)
+	}
+	e := NewEngine(m)
+	e.slotPlans.Store(task.ID, map[string]int64{"work#4": 12})
+	var pinnedCalls, snapshotCalls atomic.Int32
+	e.SetWorkerResolver(func(_ *Task, slot WorkerSlot) (*agent.Worker, error) {
+		pinnedCalls.Add(1)
+		if slot.Name != "work#4" || slot.ProfileID != 12 {
+			t.Fatalf("unexpected pinned slot: %+v", slot)
+		}
+		return nil, errors.New("profile unavailable")
+	})
+	e.SetAuthoritativeAgentResolver(func(*Task) (*agent.Planner, *agent.Worker) {
+		snapshotCalls.Add(1)
+		return nil, nil
+	})
+	err = e.runDetachedIntent(context.Background(), task, intentID, "request", "human", "agent")
+	if err == nil || !strings.Contains(err.Error(), "LLM 配置 #12 不可用") {
+		t.Fatalf("runDetachedIntent err=%v", err)
+	}
+	if pinnedCalls.Load() != 1 || snapshotCalls.Load() != 0 {
+		t.Fatalf("pinned calls=%d snapshot calls=%d", pinnedCalls.Load(), snapshotCalls.Load())
+	}
+	node, err := task.Store.GetNode(intentID)
+	if err != nil || node == nil || node.State != "paused" {
+		t.Fatalf("intent changed after unavailable resolver: node=%+v err=%v", node, err)
+	}
+}