diff --git a/db/settings.go b/db/settings.go index f8fafcc..5c3727c 100644 --- a/db/settings.go +++ b/db/settings.go @@ -3,6 +3,7 @@ package db import ( "context" "database/sql" + "sort" ) // Settings is a tiny key-value store for global app config the UI toggles at @@ -33,6 +34,32 @@ ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()`, key return err } +// SetSettingsAtomic upserts several settings in one transaction. Keys are +// sorted so concurrent callers acquire row locks in a deterministic order. +func (d *DB) SetSettingsAtomic(ctx context.Context, values map[string]string) error { + if len(values) == 0 { + return nil + } + tx, err := d.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() //nolint:errcheck // commit makes this a no-op + keys := make([]string, 0, len(values)) + for key := range values { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + if _, err := tx.ExecContext(ctx, ` +INSERT INTO settings(key, value) VALUES ($1, $2) +ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()`, key, values[key]); err != nil { + return err + } + } + return tx.Commit() +} + // InsertSettingIfAbsent 只在 key 尚未存在时写入，已存在则原样保留并返回 inserted=false。 // 给 auth.password_hash 这类"只允许首次设置"的键用：判定落在数据库的主键约束上， // 调用方先 GetSetting 再写的那种检查就只是快速失败路径——读出错或并发撞车时， diff --git a/db/settings_test.go b/db/settings_test.go new file mode 100644 index 0000000..9de68a9 --- /dev/null +++ b/db/settings_test.go @@ -0,0 +1,28 @@ +package db + +import ( + "context" + "fmt" + "testing" + "time" +) + +func TestSetSettingsAtomic(t *testing.T) { + d, err := Open(testDSN(t)) + if err != nil { + t.Skipf("postgres unavailable (%v)", err) + } + defer d.Close() + prefix := fmt.Sprintf("test.atomic_settings.%d", time.Now().UnixNano()) + defer func() { _, _ = d.Exec(`DELETE FROM settings WHERE key LIKE $1`, prefix+".%") }() + values := map[string]string{prefix + ".worker_pools": "[]", prefix + ".workers": "3"} + if err := d.SetSettingsAtomic(context.Background(), values); err != nil { + t.Fatal(err) + } + for key, want := range values { + got, ok, err := d.GetSetting(key) + if err != nil || !ok || got != want { + t.Fatalf("key %q: value=%q ok=%v err=%v, want %q", key, got, ok, err, want) + } + } +} diff --git a/server/engine.go b/server/engine.go index 9a8c07b..be30998 100644 --- a/server/engine.go +++ b/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 diff --git a/server/manager.go b/server/manager.go index 76f72ae..b2ff3c7 100644 --- a/server/manager.go +++ b/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. diff --git a/server/server.go b/server/server.go index fc3e0b5..71cf168 100644 --- a/server/server.go +++ b/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 diff --git a/server/worker_pools_test.go b/server/worker_pools_test.go index 36f40aa..4facf4a 100644 --- a/server/worker_pools_test.go +++ b/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) + } +} diff --git a/web/src/app/(main)/system/settings/page.tsx b/web/src/app/(main)/system/settings/page.tsx index 84bfe45..535e9a0 100644 --- a/web/src/app/(main)/system/settings/page.tsx +++ b/web/src/app/(main)/system/settings/page.tsx @@ -40,6 +40,7 @@ export default function SystemSettingsPage() { const [savingGlobalProxy, setSavingGlobalProxy] = React.useState(false); const [testing, setTesting] = React.useState(false); const [loaded, setLoaded] = React.useState(false); + const [settingsLoaded, setSettingsLoaded] = React.useState(false); const [saving, setSaving] = React.useState(false); const [savingKey, setSavingKey] = React.useState(false); const [pyInterp, setPyInterp] = React.useState(""); @@ -69,13 +70,17 @@ export default function SystemSettingsPage() { setGlobalProxyInput(s.global_proxy ?? ""); setPyInterp(s.python_interpreter ?? ""); setWorkers(String(s.workers ?? 3)); - setWorkerPools(s.worker_pools); + setWorkerPools(s.worker_pools ?? []); setInjectPlanner(s.constraints_inject_planner !== false); setInjectWorker(s.constraints_inject_worker !== false); setNoaCompaction(!!s.noa_compaction); }, []); const saveWorkers = () => { + if (!settingsLoaded) { + toast.error("设置加载失败，请刷新后再保存 Worker 配置"); + return; + } const n = Number(workers); if (workerPools.length > 0) { const total = workerPools.reduce((sum, pool) => sum + Number(pool.workers), 0); @@ -150,8 +155,14 @@ export default function SystemSettingsPage() { React.useEffect(() => { api .settings() - .then(apply) - .catch(() => undefined) + .then((s) => { + apply(s); + setSettingsLoaded(true); + }) + .catch(() => { + setSettingsLoaded(false); + toast.error("设置加载失败，Worker 配置已锁定，避免覆盖服务器上的模型池"); + }) .finally(() => setLoaded(true)); }, [apply]); @@ -804,7 +815,12 @@ export default function SystemSettingsPage() {
-
@@ -817,6 +833,7 @@ export default function SystemSettingsPage() {