From 4e11e4aa76696d2d731a4e83af06b917f733ed39 Mon Sep 17 00:00:00 2001 From: jack Date: Fri, 14 Aug 2026 14:24:34 +0800 Subject: [PATCH 1/2] fix(web): keep control plane available when selected model is unavailable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the selected model could not be constructed at startup (e.g. a managed account whose token expired), the web server aborted before the settings UI, /api/health, and provider-auth recovery endpoints were reachable — leaving the desktop shell with no way to recover. - runWebServer no longer aborts on agent construction failure: it keeps the selected provider/model and logs, so the control plane stays up for reauthentication and model changes. - Engine.ensureAgentAvailable lazily retries agent construction before the next send. It is fail-closed: it never falls back to a different model silently, and rebuilds are serialized with the normal model/mode switch paths. - submitMessage returns an error instead of assuming success; chat, set-goal, automation runs, and SubmitMessage now surface HTTP 503 with an actionable message and release the running claim on failure, so a degraded engine cannot wedge the session behind "processing". --- internal/command/web.go | 18 +-- .../command/web_unavailable_model_test.go | 126 ++++++++++++++++++ internal/web/approval.go | 7 +- internal/web/automation_run.go | 5 +- internal/web/chat.go | 23 +++- internal/web/engine.go | 40 ++++++ internal/web/engine_availability_test.go | 44 ++++++ 7 files changed, 247 insertions(+), 16 deletions(-) create mode 100644 internal/command/web_unavailable_model_test.go create mode 100644 internal/web/engine_availability_test.go diff --git a/internal/command/web.go b/internal/command/web.go index a7b5da91..c41a3b94 100644 --- a/internal/command/web.go +++ b/internal/command/web.go @@ -1080,14 +1080,16 @@ func runWebServer(parent context.Context, port int, host string, openBrowser boo var err error ag, err = createAgent(providerName, modelName) if err != nil { - if _, isCloud := cloud.ParseCloudProviderRef(providerName); !isCloud { - return nil, fmt.Errorf("error creating agent: %w", err) - } - // Keep the Desktop control plane usable when a previously - // selected Cloud model is temporarily unavailable. We never - // silently execute on a local model; the empty agent blocks - // sends until the user reconnects or explicitly selects one. - config.Logger().Printf("[cloud] selected model unavailable at startup: %v", err) + // Model availability must not gate the Desktop control plane. In + // particular, managed accounts that require reauthentication need + // the settings UI served by this same process in order to recover. + // Keep the selected provider/model and let the Engine lazily retry + // agent creation before the next send; it never falls back to a + // different model silently. + config.Logger().Printf( + "[web] selected model %s/%s unavailable while building task engine; control plane remains available: %v", + providerName, modelName, err, + ) } } diff --git a/internal/command/web_unavailable_model_test.go b/internal/command/web_unavailable_model_test.go new file mode 100644 index 00000000..e6823cac --- /dev/null +++ b/internal/command/web_unavailable_model_test.go @@ -0,0 +1,126 @@ +package command + +import ( + "bytes" + "context" + "fmt" + "io" + "net" + "net/http" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func managedReauthHome(t *testing.T) string { + t.Helper() + home := t.TempDir() + configDir := filepath.Join(home, ".jcode") + if err := os.MkdirAll(configDir, 0o700); err != nil { + t.Fatalf("mkdir config: %v", err) + } + configJSON := `{ + "model": "xai/grok-4.6", + "providers": { + "xai": { + "auth": {"method": "xai_oauth", "account_id": "account-1"}, + "custom_models": [ + {"id": "grok-4.6", "managed": true, "protocol": "responses"} + ] + } + }, + "memory": {"enabled": false} +}` + if err := os.WriteFile(filepath.Join(configDir, "config.json"), []byte(configJSON), 0o600); err != nil { + t.Fatalf("write config: %v", err) + } + authJSON := `{ + "version": 1, + "methods": { + "xai_oauth": { + "accounts": { + "account-1": { + "id": "account-1", + "login": "reauth@example.test", + "secret": "expired-refresh-token", + "authenticated_at": "2026-08-09T00:00:00Z", + "requires_reauth": true + } + }, + "default_account_id": "account-1" + } + } +}` + if err := os.WriteFile(filepath.Join(configDir, "provider-auth.json"), []byte(authJSON), 0o600); err != nil { + t.Fatalf("write provider auth: %v", err) + } + return home +} + +// Regression: a selected managed account that needs reauthentication must not +// abort the sidecar before /api/health and the provider-auth recovery endpoints +// are reachable. Chat remains fail-closed until reauthentication succeeds. +func TestRunWebServerManagedReauthKeepsControlPlaneAvailable(t *testing.T) { + t.Setenv("HOME", managedReauthHome(t)) + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("reserve port: %v", err) + } + port := ln.Addr().(*net.TCPAddr).Port + _ = ln.Close() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + errCh := make(chan error, 1) + go func() { errCh <- runWebServer(ctx, port, "127.0.0.1", false, "") }() + + baseURL := fmt.Sprintf("http://127.0.0.1:%d", port) + requireHealthOK(t, baseURL, errCh) + + statusResp, err := http.Get(baseURL + "/api/provider-auth/xai_oauth") //nolint:gosec // loopback test server + if err != nil { + t.Fatalf("get provider auth status: %v", err) + } + statusBody, readErr := io.ReadAll(io.LimitReader(statusResp.Body, 1<<20)) + _ = statusResp.Body.Close() + if readErr != nil { + t.Fatalf("read provider auth status: %v", readErr) + } + if statusResp.StatusCode != http.StatusOK || !bytes.Contains(statusBody, []byte(`"requires_reauth":true`)) { + t.Fatalf("provider auth status=%d body=%s", statusResp.StatusCode, statusBody) + } + + // A failed lazy rebuild must release the task's running claim. The second + // request should retry authentication and return the same actionable error, + // not get stuck behind a false "already processing" conflict. + for attempt := 1; attempt <= 2; attempt++ { + chatResp, err := http.Post( //nolint:gosec // loopback test server + baseURL+"/api/chat", "application/json", strings.NewReader(`{"message":"hello"}`), + ) + if err != nil { + t.Fatalf("post chat attempt %d: %v", attempt, err) + } + chatBody, readErr := io.ReadAll(io.LimitReader(chatResp.Body, 1<<20)) + _ = chatResp.Body.Close() + if readErr != nil { + t.Fatalf("read chat response attempt %d: %v", attempt, readErr) + } + if chatResp.StatusCode != http.StatusServiceUnavailable || + !bytes.Contains(chatBody, []byte("requires reauthentication")) { + t.Fatalf("chat attempt %d status=%d body=%s", attempt, chatResp.StatusCode, chatBody) + } + } + + cancel() + select { + case runErr := <-errCh: + if runErr != nil { + t.Fatalf("runWebServer returned error: %v", runErr) + } + case <-time.After(15 * time.Second): + t.Fatal("web server did not shut down after context cancel") + } +} diff --git a/internal/web/approval.go b/internal/web/approval.go index a744c68f..55797af8 100644 --- a/internal/web/approval.go +++ b/internal/web/approval.go @@ -61,7 +61,12 @@ func (s *Server) handleSetGoal(w http.ResponseWriter, r *http.Request) { // will pick the goal up after the current run finishes. Targets the active // task. if eng.running.CompareAndSwap(false, true) { - s.submitMessage(eng, tools.GoalKickoffPrompt(objective), eng.curMode(), req.Source, req.TaskID, nil) + if _, err := s.submitMessage( + eng, tools.GoalKickoffPrompt(objective), eng.curMode(), req.Source, req.TaskID, nil, + ); err != nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": err.Error()}) + return + } } } writeJSON(w, http.StatusOK, g) diff --git a/internal/web/automation_run.go b/internal/web/automation_run.go index c4785b92..a93b9363 100644 --- a/internal/web/automation_run.go +++ b/internal/web/automation_run.go @@ -121,7 +121,10 @@ func (s *Server) runAutomation(ctx context.Context, a *automation.Automation, ki s.deleteEngine(sid) return sid, fmt.Errorf("engine busy") } - _ = s.submitMessage(eng, a.Prompt, mode, "automation", sid, nil) + if _, err := s.submitMessage(eng, a.Prompt, mode, "automation", sid, nil); err != nil { + s.deleteEngine(sid) + return sid, err + } s.stampAutomationMeta(sid, a, kind) var runErr error diff --git a/internal/web/chat.go b/internal/web/chat.go index d0b8c166..d72d6a35 100644 --- a/internal/web/chat.go +++ b/internal/web/chat.go @@ -62,7 +62,11 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) { return } - sessionID := s.submitMessage(eng, req.Message, modeStr, req.Source, req.SessionID, req.Images) + sessionID, err := s.submitMessage(eng, req.Message, modeStr, req.Source, req.SessionID, req.Images) + if err != nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": err.Error()}) + return + } writeJSON(w, http.StatusAccepted, map[string]string{"status": "processing", "session_id": sessionID}) } @@ -154,8 +158,8 @@ func (s *Server) SubmitMessage(message, source string) bool { if !eng.running.CompareAndSwap(false, true) { return false } - s.submitMessage(eng, message, eng.curMode(), source, "", nil) - return true + _, err := s.submitMessage(eng, message, eng.curMode(), source, "", nil) + return err == nil } // submitMessage is the shared implementation for starting an agent run. @@ -165,8 +169,15 @@ func (s *Server) SubmitMessage(message, source string) bool { // correct session instead of creating a new one. // images is an optional list of base64-encoded images to include in the message. // The caller must have already set eng.running to true (via CompareAndSwap). -// Returns the session_id of the recorder used. -func (s *Server) submitMessage(eng *Engine, message, mode, source, sessionID string, images []chatImage) string { +// Returns the session_id of the recorder used. If a degraded Engine is +// still unavailable, agent construction is retried before any history or +// recorder mutation and the caller's running claim is released. +func (s *Server) submitMessage(eng *Engine, message, mode, source, sessionID string, images []chatImage) (string, error) { + if err := eng.ensureAgentAvailable(); err != nil { + eng.running.Store(false) + return "", err + } + // Slash command rewrite: if the original message starts with "/", check for // skill slash commands and rewrite to load_skill instruction (same pattern as // ACP/TUI). This must happen BEFORE the plan-mode prefix is applied, otherwise @@ -381,7 +392,7 @@ func (s *Server) submitMessage(eng *Engine, message, mode, source, sessionID str } }() - return recorder.UUID() + return recorder.UUID(), nil } // --- Stop handler --- diff --git a/internal/web/engine.go b/internal/web/engine.go index 063fec61..b2e765fc 100644 --- a/internal/web/engine.go +++ b/internal/web/engine.go @@ -260,6 +260,46 @@ func (e *Engine) modelSnapshot() (provider, model, modeStr string) { return e.providerName, e.modelName, e.mode } +// ensureAgentAvailable lazily repairs an Engine whose selected model could not +// be constructed when the task engine was built. This keeps the control plane +// reachable for account reauthentication and provider/model changes without +// ever executing on a silent fallback. Rebuilds are serialized with the normal +// model/mode paths so a successful recovery cannot overwrite a concurrent user +// selection. +func (e *Engine) ensureAgentAvailable() error { + e.emu.Lock() + if e.agent != nil { + e.emu.Unlock() + return nil + } + e.emu.Unlock() + + if e.createAgent == nil { + return fmt.Errorf("selected model agent is unavailable") + } + + e.rebuildMu.Lock() + defer e.rebuildMu.Unlock() + + e.emu.Lock() + if e.agent != nil { + e.emu.Unlock() + return nil + } + provider, modelName := e.providerName, e.modelName + e.emu.Unlock() + + ag, err := e.createAgent(provider, modelName) + if err != nil { + return fmt.Errorf("create agent for selected model %s/%s: %w", provider, modelName, err) + } + if ag == nil { + return fmt.Errorf("create agent for selected model %s/%s returned nil", provider, modelName) + } + e.applyModelSwitch(ag, provider, modelName) + return nil +} + // agentBuildSnapshot captures the inputs and revision for an asynchronous // agent rebuild. Call installAgentIfRevision with the returned revision: a // concurrent model/mode/skill change must win instead of being overwritten by diff --git a/internal/web/engine_availability_test.go b/internal/web/engine_availability_test.go new file mode 100644 index 00000000..80154616 --- /dev/null +++ b/internal/web/engine_availability_test.go @@ -0,0 +1,44 @@ +package web + +import ( + "errors" + "testing" + + "github.com/cloudwego/eino/adk" +) + +func TestEnsureAgentAvailableRetriesAfterProviderRecovery(t *testing.T) { + wantErr := errors.New("provider needs reauthentication") + recovered := &adk.ChatModelAgent{} + attempts := 0 + eng := newEngine(&EngineConfig{ + ProviderName: "xai", + ModelName: "grok-4.6", + CreateAgent: func(provider, model string) (*adk.ChatModelAgent, error) { + attempts++ + if provider != "xai" || model != "grok-4.6" { + t.Fatalf("create agent called with %s/%s", provider, model) + } + if attempts == 1 { + return nil, wantErr + } + return recovered, nil + }, + }) + + if err := eng.ensureAgentAvailable(); !errors.Is(err, wantErr) { + t.Fatalf("first recovery error = %v, want %v", err, wantErr) + } + if err := eng.ensureAgentAvailable(); err != nil { + t.Fatalf("second recovery: %v", err) + } + if eng.agent != recovered { + t.Fatalf("recovered agent = %p, want %p", eng.agent, recovered) + } + if err := eng.ensureAgentAvailable(); err != nil { + t.Fatalf("available agent check: %v", err) + } + if attempts != 2 { + t.Fatalf("create agent attempts = %d, want 2", attempts) + } +} From 9000389820d4d8461dabf94b719b75972b8fb258 Mon Sep 17 00:00:00 2001 From: jack Date: Fri, 14 Aug 2026 18:09:50 +0800 Subject: [PATCH 2/2] fix(web): surface degraded channel submission errors --- internal/command/web.go | 11 +++- .../command/web_unavailable_model_test.go | 5 +- internal/web/chat.go | 64 +++++++++++++------ internal/web/engine_availability_test.go | 23 +++++++ 4 files changed, 80 insertions(+), 23 deletions(-) diff --git a/internal/command/web.go b/internal/command/web.go index c41a3b94..366a8994 100644 --- a/internal/command/web.go +++ b/internal/command/web.go @@ -1275,8 +1275,15 @@ func startWebServer(runtime webServerRuntime) error { return } config.Logger().Printf("[wechat] inbound message from %s: %s", from, text) - if !srv.SubmitMessage(text, "wechat") { - _ = runtime.wechatClient.SendText(channel.BusyMessage()) + accepted, submitErr := srv.SubmitMessage(text, "wechat") + if submitErr != nil { + if sendErr := runtime.wechatClient.SendText(channel.DoneMessage("", submitErr)); sendErr != nil { + config.Logger().Printf("[wechat] failed to send submission error: %v", sendErr) + } + } else if !accepted { + if sendErr := runtime.wechatClient.SendText(channel.BusyMessage()); sendErr != nil { + config.Logger().Printf("[wechat] failed to send busy response: %v", sendErr) + } } }) defer func() { diff --git a/internal/command/web_unavailable_model_test.go b/internal/command/web_unavailable_model_test.go index e6823cac..bb65a2b7 100644 --- a/internal/command/web_unavailable_model_test.go +++ b/internal/command/web_unavailable_model_test.go @@ -79,8 +79,9 @@ func TestRunWebServerManagedReauthKeepsControlPlaneAvailable(t *testing.T) { baseURL := fmt.Sprintf("http://127.0.0.1:%d", port) requireHealthOK(t, baseURL, errCh) + client := &http.Client{Timeout: 10 * time.Second} - statusResp, err := http.Get(baseURL + "/api/provider-auth/xai_oauth") //nolint:gosec // loopback test server + statusResp, err := client.Get(baseURL + "/api/provider-auth/xai_oauth") //nolint:gosec // loopback test server if err != nil { t.Fatalf("get provider auth status: %v", err) } @@ -97,7 +98,7 @@ func TestRunWebServerManagedReauthKeepsControlPlaneAvailable(t *testing.T) { // request should retry authentication and return the same actionable error, // not get stuck behind a false "already processing" conflict. for attempt := 1; attempt <= 2; attempt++ { - chatResp, err := http.Post( //nolint:gosec // loopback test server + chatResp, err := client.Post( //nolint:gosec // loopback test server baseURL+"/api/chat", "application/json", strings.NewReader(`{"message":"hello"}`), ) if err != nil { diff --git a/internal/web/chat.go b/internal/web/chat.go index d72d6a35..52ab40e1 100644 --- a/internal/web/chat.go +++ b/internal/web/chat.go @@ -149,17 +149,22 @@ type chatImage struct { } // SubmitMessage submits a message for agent processing from an external source -// (e.g. WeChat inbound message). Returns false if the agent is busy. -func (s *Server) SubmitMessage(message, source string) bool { +// (e.g. WeChat inbound message). accepted is false only when no active engine is +// available or the targeted engine is already busy; agent recovery failures are +// returned separately so channel callers can surface the real error. +func (s *Server) SubmitMessage(message, source string) (accepted bool, err error) { eng := s.activeEngine() if eng == nil { - return false + return false, nil } if !eng.running.CompareAndSwap(false, true) { - return false + return false, nil } - _, err := s.submitMessage(eng, message, eng.curMode(), source, "", nil) - return err == nil + _, err = s.submitMessage(eng, message, eng.curMode(), source, "", nil) + if err != nil { + return false, err + } + return true, nil } // submitMessage is the shared implementation for starting an agent run. @@ -249,31 +254,52 @@ func (s *Server) submitMessage(eng *Engine, message, mode, source, sessionID str eng.toolOverrideMu.Lock() eng.emu.Lock() if eng.recorder == nil { - rec, _ := session.NewRecorder(eng.pwd, eng.providerName, eng.modelName) - if rec != nil { - rec.SetAgent(eng.agentRole) + rec, err := session.NewRecorder(eng.pwd, eng.providerName, eng.modelName) + if err != nil { + eng.emu.Unlock() + eng.toolOverrideMu.Unlock() + eng.running.Store(false) + return "", fmt.Errorf("create session recorder: %w", err) } - if rec != nil && sessionID != "" { + if rec == nil { + eng.emu.Unlock() + eng.toolOverrideMu.Unlock() + eng.running.Store(false) + return "", fmt.Errorf("create session recorder: returned nil recorder") + } + rec.SetAgent(eng.agentRole) + if sessionID != "" { rec.SetUUID(sessionID) } - if rec != nil && eng.recorderInit != nil { + if eng.recorderInit != nil { eng.recorderInit(rec) } eng.recorder = rec // sessionID == "" means the recorder just minted a new UUID. - stampNew = rec != nil && sessionID == "" + stampNew = sessionID == "" } else if sessionID != "" && eng.recorder.UUID() != sessionID { // Client is continuing a session that doesn't match the current recorder. - // Resume the client's session to keep all messages together. - eng.recorder.Close() - rec, _ := session.NewRecorder(eng.pwd, eng.providerName, eng.modelName) - if rec != nil { - rec.SetAgent(eng.agentRole) - rec.SetUUID(sessionID) + // Build the replacement before closing the current recorder so a creation + // failure cannot discard the live session. + rec, err := session.NewRecorder(eng.pwd, eng.providerName, eng.modelName) + if err != nil { + eng.emu.Unlock() + eng.toolOverrideMu.Unlock() + eng.running.Store(false) + return "", fmt.Errorf("create recorder for session %s: %w", sessionID, err) } - if rec != nil && eng.recorderInit != nil { + if rec == nil { + eng.emu.Unlock() + eng.toolOverrideMu.Unlock() + eng.running.Store(false) + return "", fmt.Errorf("create recorder for session %s: returned nil recorder", sessionID) + } + rec.SetAgent(eng.agentRole) + rec.SetUUID(sessionID) + if eng.recorderInit != nil { eng.recorderInit(rec) } + eng.recorder.Close() eng.recorder = rec } recorder := eng.recorder diff --git a/internal/web/engine_availability_test.go b/internal/web/engine_availability_test.go index 80154616..44523745 100644 --- a/internal/web/engine_availability_test.go +++ b/internal/web/engine_availability_test.go @@ -42,3 +42,26 @@ func TestEnsureAgentAvailableRetriesAfterProviderRecovery(t *testing.T) { t.Fatalf("create agent attempts = %d, want 2", attempts) } } + +func TestSubmitMessageReturnsLazyRecoveryError(t *testing.T) { + wantErr := errors.New("provider needs reauthentication") + eng := newEngine(&EngineConfig{ + ProviderName: "xai", + ModelName: "grok-4.6", + CreateAgent: func(_, _ string) (*adk.ChatModelAgent, error) { + return nil, wantErr + }, + }) + s := &Server{Engine: eng} + + accepted, err := s.SubmitMessage("hello", "wechat") + if accepted { + t.Fatal("SubmitMessage accepted message despite recovery failure") + } + if !errors.Is(err, wantErr) { + t.Fatalf("SubmitMessage error = %v, want %v", err, wantErr) + } + if eng.running.Load() { + t.Fatal("SubmitMessage left engine running after recovery failure") + } +}