diff --git a/packages/core/chat/mutations.ts b/packages/core/chat/mutations.ts index f8d05ed87..d30b717f8 100644 --- a/packages/core/chat/mutations.ts +++ b/packages/core/chat/mutations.ts @@ -1,7 +1,7 @@ import { useMutation, useQueryClient } from "@tanstack/react-query"; import { api } from "../api"; import { useWorkspaceId } from "../hooks"; -import { chatKeys, sortChatSessions } from "./queries"; +import { chatKeys, sortChatSessions, QUICK_ACTIONS_PENDING_TIMEOUT_MS } from "./queries"; import { createLogger } from "../logger"; import type { ChatSession, @@ -398,7 +398,11 @@ export function useRegenerateChatQuickActions() { // matches on message_id). qc.setQueryData( chatKeys.quickActionsPending(sessionId), - { message_id: messageId, task_id: "" }, + { + message_id: messageId, + task_id: "", + expires_at: Date.now() + QUICK_ACTIONS_PENDING_TIMEOUT_MS, + }, ); return { previous, sessionId }; }, diff --git a/packages/core/chat/queries.ts b/packages/core/chat/queries.ts index 19ec5d392..19177af58 100644 --- a/packages/core/chat/queries.ts +++ b/packages/core/chat/queries.ts @@ -1,7 +1,20 @@ import { infiniteQueryOptions, queryOptions } from "@tanstack/react-query"; import { api } from "../api"; import type { TaskMessagePayload } from "../types/events"; -import type { ChatQuickActionsPendingState, ChatSession } from "../types/chat"; +import type { + ChatQuickActionsFailureState, + ChatQuickActionsPendingState, + ChatSession, +} from "../types/chat"; + +/** + * How long a quick-actions pending marker may stay unresolved before a client + * gives up and clears it (MUL-5149). Used to stamp the marker's absolute + * `expires_at` deadline at creation, and by `useQuickActionsPendingTimeout` to + * clear it — a shared deadline survives chat-surface switches instead of + * re-arming a fresh window on each remount. + */ +export const QUICK_ACTIONS_PENDING_TIMEOUT_MS = 30_000; // NOTE on workspace scoping: // `wsId` is used only as part of queryKey for cache isolation per workspace. @@ -25,6 +38,10 @@ export const chatKeys = { quickActionsPendingAll: () => ["chat", "quick-actions-pending"] as const, quickActionsPending: (sessionId: string) => [...chatKeys.quickActionsPendingAll(), sessionId] as const, + /** Client-only signal: this session's last refresh regeneration failed. */ + quickActionsFailureAll: () => ["chat", "quick-actions-failure"] as const, + quickActionsFailure: (sessionId: string) => + [...chatKeys.quickActionsFailureAll(), sessionId] as const, draftRestoresAll: () => ["chat", "draft-restores"] as const, /** Durable deferred-cancellation draft restores for a session (#5219). */ draftRestores: (sessionId: string) => [...chatKeys.draftRestoresAll(), sessionId] as const, @@ -246,3 +263,17 @@ export function chatQuickActionsPendingOptions(sessionId: string) { staleTime: Infinity, }); } + +/** + * Client-only cache entry mirroring {@link chatQuickActionsPendingOptions}: + * the realtime layer writes it off a failed chat:quick_actions, a view reads + * it once to toast "couldn't refresh," then clears it. Never fetched. + */ +export function chatQuickActionsFailureOptions(sessionId: string) { + return queryOptions({ + queryKey: chatKeys.quickActionsFailure(sessionId), + queryFn: async (): Promise => null, + enabled: false, + staleTime: Infinity, + }); +} diff --git a/packages/core/chat/use-quick-actions-pending-timeout.ts b/packages/core/chat/use-quick-actions-pending-timeout.ts index fe0734790..06ba94b21 100644 --- a/packages/core/chat/use-quick-actions-pending-timeout.ts +++ b/packages/core/chat/use-quick-actions-pending-timeout.ts @@ -1,26 +1,23 @@ import { useEffect } from "react"; import { useQueryClient } from "@tanstack/react-query"; -import { chatKeys } from "./queries"; +import { chatKeys, QUICK_ACTIONS_PENDING_TIMEOUT_MS } from "./queries"; import type { ChatQuickActionsPendingState } from "../types/chat"; -/** - * How long a quick-actions pending marker may stay unresolved before we give up - * and clear it. Normally a chat:quick_actions supplement resolves the marker - * well within this window; the timeout only covers the paths where that event - * never arrives (daemon died, or its supplement HTTP ultimately failed). - */ -export const QUICK_ACTIONS_PENDING_TIMEOUT_MS = 30_000; - /** * Clears a session's `quickActionsPending` marker from the query cache when no - * chat:quick_actions supplement resolves it in time (MUL-5149). Unlike a - * component-local timer, this drops the REAL cache state, so the pill spinner / - * skeleton stops AND a later refresh — even after closing and reopening chat — - * starts clean instead of re-reading a stuck marker. + * chat:quick_actions supplement resolves it before the marker's own deadline + * (MUL-5149). Unlike a component-local timer, this drops the REAL cache state, + * so the pill spinner / skeleton stops AND a later refresh — even after closing + * and reopening chat — starts clean instead of re-reading a stuck marker. + * + * The deadline is `marker.expires_at` (absolute), NOT a fresh window per mount: + * switching between the floating chat and the chat tab unmounts one host and + * mounts the other, and a per-mount timer would restart the full wait each time, + * letting a lost event strand the marker indefinitely. Reading the stored + * deadline means a remount waits only the remaining time. The final clear is + * message-scoped so it never wipes a newer turn's marker. * * Mount once per surface that reads the marker (chat page / floating window). - * The effect re-arms whenever the pending message changes, and the final clear - * is message-scoped so it never wipes a newer turn's marker. */ export function useQuickActionsPendingTimeout( sessionId: string | null, @@ -28,15 +25,22 @@ export function useQuickActionsPendingTimeout( ): void { const qc = useQueryClient(); const messageId = pending?.message_id ?? null; + const expiresAt = pending?.expires_at ?? null; useEffect(() => { if (!sessionId || !messageId) return; - const timer = setTimeout(() => { + const clear = () => qc.setQueryData( chatKeys.quickActionsPending(sessionId), (current) => current && current.message_id === messageId ? null : current, ); - }, QUICK_ACTIONS_PENDING_TIMEOUT_MS); + // Remaining time to this marker's absolute deadline; an old marker without + // one falls back to a full window. Already-past deadlines clear next tick. + const remaining = + expiresAt != null + ? expiresAt - Date.now() + : QUICK_ACTIONS_PENDING_TIMEOUT_MS; + const timer = setTimeout(clear, Math.max(0, remaining)); return () => clearTimeout(timer); - }, [qc, sessionId, messageId]); + }, [qc, sessionId, messageId, expiresAt]); } diff --git a/packages/core/realtime/use-realtime-sync.ts b/packages/core/realtime/use-realtime-sync.ts index a6a3ffdfa..6474776e1 100644 --- a/packages/core/realtime/use-realtime-sync.ts +++ b/packages/core/realtime/use-realtime-sync.ts @@ -48,7 +48,12 @@ import { type SystemNotificationPayload, } from "../platform/system-notification"; import type { Workspace } from "../types/workspace"; -import { chatKeys, mergeTaskMessagesBySeq, sortChatSessions } from "../chat/queries"; +import { + chatKeys, + mergeTaskMessagesBySeq, + sortChatSessions, + QUICK_ACTIONS_PENDING_TIMEOUT_MS, +} from "../chat/queries"; import { useChatStore } from "../chat"; import { resolvePostAuthDestination, useHasOnboarded } from "../paths"; import type { @@ -88,6 +93,7 @@ import type { ChatDonePayload, ChatQuickActionsPayload, ChatQuickActionsPendingState, + ChatQuickActionsFailureState, ChatCancelFinalizedPayload, ChatMessage, ChatPendingTask, @@ -177,7 +183,11 @@ export function applyChatDoneToCache( qc.setQueryData( chatKeys.quickActionsPending(sessionId), payload.quick_actions_pending === true && messageId - ? { message_id: messageId, task_id: taskId } + ? { + message_id: messageId, + task_id: taskId, + expires_at: Date.now() + QUICK_ACTIONS_PENDING_TIMEOUT_MS, + } : null, ); // Authoritative refetch reconciles redaction / migrations / clients @@ -191,6 +201,12 @@ export function applyChatDoneToCache( * message's quick_actions in both message caches and resolve the pending * placeholder. An empty/missing list is terminal ("no suggestions this * turn") — the placeholder still resolves. + * + * `payload.failed` marks a resolution whose regeneration failed: the carried + * actions are the turn's UNCHANGED prior pills, so the patch is a no-op, but a + * failure signal is raised for a view to toast — otherwise an explicit refresh + * that failed would look identical to one that succeeded with the same + * suggestions (MUL-5149 review). */ export async function applyChatQuickActionsToCache( qc: QueryClient, @@ -239,6 +255,15 @@ export async function applyChatQuickActionsToCache( (current) => current && current.message_id !== payload.message_id ? current : null, ); + // Raise a one-shot failure signal on an explicit refresh whose regeneration + // failed. A view consumes it to toast and clears it; the `at` nonce keeps a + // repeat failure on the same turn from being deduped away. + if (payload.failed === true) { + qc.setQueryData( + chatKeys.quickActionsFailure(sessionId), + { message_id: payload.message_id, at: Date.now() }, + ); + } } function patchLatestChatMessagePage( diff --git a/packages/core/types/chat.ts b/packages/core/types/chat.ts index 4f8ba577c..5d66e9245 100644 --- a/packages/core/types/chat.ts +++ b/packages/core/types/chat.ts @@ -35,6 +35,27 @@ export interface ChatQuickAction { export interface ChatQuickActionsPendingState { message_id: string; task_id: string; + /** + * Absolute epoch-ms deadline after which a client gives up waiting for the + * chat:quick_actions supplement and clears this marker. Stored on the marker + * (not as a component timer) so switching chat surfaces — which unmounts and + * remounts the timeout hook — resumes the SAME deadline instead of re-arming + * a fresh window each time (MUL-5149 review). + */ + expires_at: number; +} + +/** + * Client-only signal (never persisted, never fetched) that an explicit refresh's + * background regeneration FAILED — the daemon suggestion pass never completed, so + * the turn's pills are unchanged. Set by the realtime layer off a failed + * chat:quick_actions and consumed once by a view to toast "couldn't refresh" + * before clearing itself. `at` is a per-event nonce so a second failure re-fires + * even when the message_id repeats (MUL-5149 review). + */ +export interface ChatQuickActionsFailureState { + message_id: string; + at: number; } /** Preview of a session's most recent message, for the IM-style list. */ diff --git a/packages/core/types/events.ts b/packages/core/types/events.ts index 801f40b8c..bde2cf7cb 100644 --- a/packages/core/types/events.ts +++ b/packages/core/types/events.ts @@ -397,6 +397,14 @@ export interface ChatQuickActionsPayload { task_id: string; message_id: string; quick_actions?: import("./chat").ChatQuickAction[]; + /** + * The regeneration behind an explicit refresh failed: the actions carried + * here are the turn's UNCHANGED prior pills, sent only to resolve the pending + * spinner. Clients surface a "couldn't refresh" notice rather than treating + * them as freshly generated. Absent/false on the normal success path and on + * older servers. + */ + failed?: boolean; } /** diff --git a/packages/core/types/index.ts b/packages/core/types/index.ts index 4a97aa8c3..d69a98224 100644 --- a/packages/core/types/index.ts +++ b/packages/core/types/index.ts @@ -100,6 +100,7 @@ export type { ChatMessage, ChatQuickAction, ChatQuickActionsPendingState, + ChatQuickActionsFailureState, ChatMessagesPage, ChatPendingTask, PendingChatTaskItem, diff --git a/packages/views/chat/chat-page.tsx b/packages/views/chat/chat-page.tsx index 5f9392dc6..18b4e719d 100644 --- a/packages/views/chat/chat-page.tsx +++ b/packages/views/chat/chat-page.tsx @@ -16,6 +16,7 @@ import { useChatStore } from "@multica/core/chat"; import { chatQuickActionsPendingOptions } from "@multica/core/chat/queries"; import { useRegenerateChatQuickActions } from "@multica/core/chat/mutations"; import { useQuickActionsPendingTimeout } from "@multica/core/chat/use-quick-actions-pending-timeout"; +import { useQuickActionsFailureToast } from "./components/use-quick-actions-failure-toast"; import { useQuery } from "@tanstack/react-query"; import type { Agent, ChatSession } from "@multica/core/types"; import { PageHeader } from "../layout/page-header"; @@ -64,6 +65,8 @@ export function ChatPage() { // Drop a stuck pending marker (dead daemon / failed supplement) so the pill // spinner stops and a later refresh starts clean (MUL-5149). useQuickActionsPendingTimeout(c.activeSessionId ?? null, quickActionsPending); + // Toast when an accepted refresh later fails in the daemon (async half). + useQuickActionsFailureToast(c.activeSessionId ?? null); const regenerateQuickActions = useRegenerateChatQuickActions(); const urlSession = searchParams.get("session") || null; const urlAgent = searchParams.get("agent") || null; diff --git a/packages/views/chat/components/chat-window.tsx b/packages/views/chat/components/chat-window.tsx index fdb843287..bad1d7323 100644 --- a/packages/views/chat/components/chat-window.tsx +++ b/packages/views/chat/components/chat-window.tsx @@ -51,6 +51,7 @@ import { import { useChatStore } from "@multica/core/chat"; import { chatQuickActionsPendingOptions } from "@multica/core/chat/queries"; import { useQuickActionsPendingTimeout } from "@multica/core/chat/use-quick-actions-pending-timeout"; +import { useQuickActionsFailureToast } from "./use-quick-actions-failure-toast"; import { removeChatMessageFromCaches } from "@multica/core/realtime"; import { useChatDraftRestore } from "./use-chat-draft-restore"; import { useChatInputFocus } from "./use-chat-input-focus"; @@ -116,6 +117,8 @@ export function ChatWindow() { // Drop a stuck pending marker (dead daemon / failed supplement) so the pill // spinner stops and a later refresh starts clean (MUL-5149). useQuickActionsPendingTimeout(activeSessionId ?? null, quickActionsPending); + // Toast when an accepted refresh later fails in the daemon (async half). + useQuickActionsFailureToast(activeSessionId ?? null); const regenerateQuickActions = useRegenerateChatQuickActions(); const selectedAgentId = useChatStore((s) => s.selectedAgentId); const selectedProjectId = useChatStore((s) => s.selectedProjectId); diff --git a/packages/views/chat/components/use-quick-actions-failure-toast.ts b/packages/views/chat/components/use-quick-actions-failure-toast.ts new file mode 100644 index 000000000..250c52e0b --- /dev/null +++ b/packages/views/chat/components/use-quick-actions-failure-toast.ts @@ -0,0 +1,45 @@ +import { useEffect } from "react"; +import { useQuery, useQueryClient } from "@tanstack/react-query"; +import { + chatKeys, + chatQuickActionsFailureOptions, +} from "@multica/core/chat/queries"; +import type { ChatQuickActionsFailureState } from "@multica/core/types"; +import { toast } from "sonner"; +import { useT } from "../../i18n"; + +/** + * Surfaces a toast when a session's explicit quick-actions refresh fails + * asynchronously — the HTTP call was accepted but the daemon's regeneration + * never produced suggestions, so the turn's pills are unchanged (MUL-5149). The + * failure arrives over chat:quick_actions with failed=true, which + * applyChatQuickActionsToCache turns into a one-shot client-only signal; this + * hook consumes it and clears it. The synchronous HTTP-failure path (409 busy, + * network) already toasts from QuickActions.handleRegenerate, so this covers + * only the async half — without it, a failed refresh would look identical to a + * successful one whose suggestions happened not to change. + * + * A stable toast id keyed on the failure nonce collapses duplicates when both + * chat surfaces (floating window + chat tab) are mounted and each runs this + * hook against the same signal. Mount once per surface that reads the marker. + */ +export function useQuickActionsFailureToast(sessionId: string | null): void { + const qc = useQueryClient(); + const { t } = useT("chat"); + const { data: failure = null } = useQuery( + chatQuickActionsFailureOptions(sessionId ?? ""), + ); + + useEffect(() => { + if (!sessionId || !failure) return; + toast.error(t(($) => $.message_list.quick_actions_regenerate_failed), { + id: `qa-refresh-fail:${sessionId}:${failure.at}`, + }); + // Clear only this exact signal so a newer failure raised in the same tick + // is not swallowed. + qc.setQueryData( + chatKeys.quickActionsFailure(sessionId), + (current) => (current && current.at === failure.at ? null : current), + ); + }, [qc, t, sessionId, failure]); +} diff --git a/server/internal/daemon/chat_suggest.go b/server/internal/daemon/chat_suggest.go index e8291f25a..a5fe66190 100644 --- a/server/internal/daemon/chat_suggest.go +++ b/server/internal/daemon/chat_suggest.go @@ -65,7 +65,7 @@ func (d *Daemon) chatSuggestJob(task Task, backend agent.Backend, opts agent.Exe d.chatSuggestCancels.CompareAndDelete(task.ChatSessionID, handle) }() - raw, usage := d.runChatSuggestPass(ctx, backend, opts, providerSessionID, taskLog) + raw, usage, _ := d.runChatSuggestPass(ctx, backend, opts, providerSessionID, taskLog) if ctx.Err() != nil { // A newer turn on this session cancelled us: its user message // supersedes these suggestions. Skip the supplement — the client @@ -118,17 +118,11 @@ func (d *Daemon) runChatQuickActionsRegenerate(ctx context.Context, task Task, b "target_task", shortID(task.RegenerateQuickActionsFor), "resume", opts.ResumeSessionID != "", ) - raw, usage := d.runChatSuggestPass(ctx, backend, opts, opts.ResumeSessionID, taskLog) - - reportCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) - defer cancel() - supplementErr := d.client.SupplementTaskQuickActions(reportCtx, task.RegenerateQuickActionsFor, raw) - if supplementErr != nil { - taskLog.Warn("chat quick-actions regenerate supplement failed", "error", supplementErr) - } + raw, usage, ok := d.runChatSuggestPass(ctx, backend, opts, opts.ResumeSessionID, taskLog) // Report the pass's own token spend on this regenerate task (its own row — // no merge with a main turn, there isn't one). Same shape as chatSuggestJob. + // Reported on every outcome — a failed pass still burned tokens. var usageEntries []TaskUsageEntry for model, u := range usage { if u.InputTokens == 0 && u.OutputTokens == 0 && u.CacheReadTokens == 0 && u.CacheWriteTokens == 0 { @@ -143,12 +137,32 @@ func (d *Daemon) runChatQuickActionsRegenerate(ctx context.Context, task Task, b CacheWriteTokens: u.CacheWriteTokens, }) } - if supplementErr != nil { + + if !ok { + // The provider pass itself FAILED (didn't start / didn't complete / + // timed out) — distinct from a completed pass that produced no + // suggestions. Do NOT supplement: an empty raw would silently rebroadcast + // the old pills as if the refresh succeeded. Report failure so FailTask → + // resolveFailedRegenerateQuickActions broadcasts a FAILED + // chat:quick_actions and the user gets feedback (MUL-5149 review §3). + return TaskResult{ + Status: "blocked", + FailureReason: "quick_actions_generation_failed", + Usage: usageEntries, + }, nil + } + + // Generation succeeded (raw may be an empty list — a genuine "no new + // suggestions" that keeps the existing pills). Supplement the target turn. + reportCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := d.client.SupplementTaskQuickActions(reportCtx, task.RegenerateQuickActionsFor, raw); err != nil { // The supplement never landed, so no chat:quick_actions resolved the // client's refresh spinner. Report failure (never retried, writes no // visible turn — see FailTask's regenerate gates) so the server converges // the marker via resolveFailedRegenerateQuickActions instead of leaving // the client to its own timeout (MUL-5149). + taskLog.Warn("chat quick-actions regenerate supplement failed", "error", err) return TaskResult{ Status: "blocked", FailureReason: "quick_actions_supplement_failed", @@ -160,9 +174,11 @@ func (d *Daemon) runChatQuickActionsRegenerate(ctx context.Context, task Task, b // runChatSuggestPass runs one extra provider turn on the just-finished chat // session and returns its raw text output (the JSON array, hopefully — the -// server parses leniently and treats garbage as "no suggestions"). Best-effort -// by design: every failure path returns "" and the turn proceeds without -// suggestions. +// server parses leniently and treats garbage as "no suggestions"). ok is false +// only when the pass itself FAILED (didn't start, didn't complete, or timed +// out), distinct from a pass that completed and simply produced no suggestions +// (ok=true, possibly empty raw). The automatic pass is best-effort and ignores +// ok; the explicit-refresh caller uses it to surface failure feedback (MUL-5149). // // The pass deliberately reuses the main turn's backend and exec options so it // lands on the same provider, model, and workdir; only the resume pointer, @@ -173,7 +189,7 @@ func (d *Daemon) runChatQuickActionsRegenerate(ctx context.Context, task Task, b // new session file) keep the suggestion exchange in the session history the // next user turn resumes. The prompt's leading line instructs the model to // never surface it. -func (d *Daemon) runChatSuggestPass(ctx context.Context, backend agent.Backend, opts agent.ExecOptions, sessionID string, taskLog *slog.Logger) (raw string, usage map[string]agent.TokenUsage) { +func (d *Daemon) runChatSuggestPass(ctx context.Context, backend agent.Backend, opts agent.ExecOptions, sessionID string, taskLog *slog.Logger) (raw string, usage map[string]agent.TokenUsage, ok bool) { suggestCtx, cancel := context.WithTimeout(ctx, chatSuggestTimeout) defer cancel() @@ -186,7 +202,7 @@ func (d *Daemon) runChatSuggestPass(ctx context.Context, backend agent.Backend, session, err := backend.Execute(suggestCtx, chatSuggestPrompt, opts) if err != nil { taskLog.Warn("chat suggest pass failed to start", "error", err) - return "", nil + return "", nil, false } // Drain and discard the message stream: backends block on an undrained @@ -206,15 +222,15 @@ func (d *Daemon) runChatSuggestPass(ctx context.Context, backend agent.Backend, "status", result.Status, "error", result.Error, ) - return "", result.Usage + return "", result.Usage, false } taskLog.Debug("chat suggest pass finished", "duration", time.Since(start).Round(time.Second).String(), "output_bytes", len(result.Output), ) - return strings.TrimSpace(result.Output), result.Usage + return strings.TrimSpace(result.Output), result.Usage, true case <-suggestCtx.Done(): taskLog.Warn("chat suggest pass timed out or was cancelled", "timeout", chatSuggestTimeout.String()) - return "", nil + return "", nil, false } } diff --git a/server/internal/daemon/chat_suggest_test.go b/server/internal/daemon/chat_suggest_test.go index 0f4d4d520..1161aa999 100644 --- a/server/internal/daemon/chat_suggest_test.go +++ b/server/internal/daemon/chat_suggest_test.go @@ -19,7 +19,10 @@ func TestRunChatSuggestPassReturnsTrimmedOutputAndResumesSession(t *testing.T) { } d := &Daemon{logger: slog.Default()} - raw, usage := d.runChatSuggestPass(context.Background(), backend, agent.ExecOptions{Model: "m1"}, "sess-1", slog.Default()) + raw, usage, ok := d.runChatSuggestPass(context.Background(), backend, agent.ExecOptions{Model: "m1"}, "sess-1", slog.Default()) + if !ok { + t.Fatal("a completed pass must report ok=true") + } if raw != `[{"label":"Next","prompt":"Do the next thing"}]` { t.Fatalf("raw = %q", raw) } @@ -43,14 +46,14 @@ func TestRunChatSuggestPassSoftFailsOnErrorAndNonCompletion(t *testing.T) { d := &Daemon{logger: slog.Default()} failedStart := &fakeBackend{errors: []error{context.DeadlineExceeded}, results: []agent.Result{{}}} - if raw, _ := d.runChatSuggestPass(context.Background(), failedStart, agent.ExecOptions{}, "sess-1", slog.Default()); raw != "" { - t.Fatalf("start failure must degrade to no suggestions, got %q", raw) + if raw, _, ok := d.runChatSuggestPass(context.Background(), failedStart, agent.ExecOptions{}, "sess-1", slog.Default()); raw != "" || ok { + t.Fatalf("start failure must report ok=false and no suggestions, got raw=%q ok=%v", raw, ok) } notCompleted := &fakeBackend{results: []agent.Result{{Status: "timeout", Error: "slow", Usage: map[string]agent.TokenUsage{"m": {InputTokens: 1}}}}} - raw, usage := d.runChatSuggestPass(context.Background(), notCompleted, agent.ExecOptions{}, "sess-1", slog.Default()) - if raw != "" { - t.Fatalf("non-completed result must degrade to no suggestions, got %q", raw) + raw, usage, ok := d.runChatSuggestPass(context.Background(), notCompleted, agent.ExecOptions{}, "sess-1", slog.Default()) + if raw != "" || ok { + t.Fatalf("non-completed result must report ok=false and no suggestions, got raw=%q ok=%v", raw, ok) } if usage["m"].InputTokens != 1 { t.Fatalf("tokens burned by a failed pass must still be reported, got %+v", usage) diff --git a/server/internal/handler/chat.go b/server/internal/handler/chat.go index c3ee67406..df028c48f 100644 --- a/server/internal/handler/chat.go +++ b/server/internal/handler/chat.go @@ -973,6 +973,11 @@ func (h *Handler) RegenerateChatQuickActions(w http.ResponseWriter, r *http.Requ // view is stale (a newer reply arrived). 409 → the client rolls back // its optimistic marker and re-offers refresh on the new turn. writeError(w, http.StatusConflict, "a newer reply arrived — refresh it instead") + case errors.Is(err, service.ErrChatQuickActionsBusy): + // A turn or another refresh is already running for this session; the + // result would be stale or a duplicate. 409 → the client rolls back + // and can refresh once the session settles. + writeError(w, http.StatusConflict, "still working — try refreshing in a moment") case errors.Is(err, service.ErrChatQuickActionsNoTurn): writeError(w, http.StatusConflict, "no assistant reply to refresh yet") case errors.Is(err, service.ErrChatQuickActionsNotResumable): diff --git a/server/internal/handler/chat_input_ownership_test.go b/server/internal/handler/chat_input_ownership_test.go index 1498468d6..d52a2223b 100644 --- a/server/internal/handler/chat_input_ownership_test.go +++ b/server/internal/handler/chat_input_ownership_test.go @@ -565,7 +565,7 @@ func TestCompleteTask_ChatQuickActionsSupplement(t *testing.T) { } raw := "```json\n" + `[{"label":"From pass","prompt":"suggested prompt","primary":true}]` + "\n```" - if err := testHandler.TaskService.SupplementChatQuickActions(ctx, *task, raw); err != nil { + if err := testHandler.TaskService.SupplementChatQuickActions(ctx, *task, raw, false); err != nil { t.Fatalf("supplement quick actions: %v", err) } rows = assistantRows(t, ctx, sessionID) @@ -579,7 +579,7 @@ func TestCompleteTask_ChatQuickActionsSupplement(t *testing.T) { // An empty supplement must not clobber existing actions (it only resolves // the pending placeholder client-side). - if err := testHandler.TaskService.SupplementChatQuickActions(ctx, *task, ""); err != nil { + if err := testHandler.TaskService.SupplementChatQuickActions(ctx, *task, "", false); err != nil { t.Fatalf("empty supplement: %v", err) } rows = assistantRows(t, ctx, sessionID) @@ -655,13 +655,21 @@ func TestRegenerateChatQuickActions_StaleTargetRejected(t *testing.T) { m1 := rows[0].ID // Refreshing the current latest turn is accepted and targets exactly it. - target, _, err := testHandler.TaskService.RegenerateChatQuickActions(ctx, session, parseUUID(testUserID), m1) + target, regenTask, err := testHandler.TaskService.RegenerateChatQuickActions(ctx, session, parseUUID(testUserID), m1) if err != nil { t.Fatalf("regenerate latest turn: %v", err) } if target != m1 { t.Fatalf("regenerate target = %v, want the latest turn %v", target, m1) } + // That refresh enqueues a regen pass that occupies the session until the + // daemon finishes it. Settle it so the session is idle again — this test + // isolates the STALE check; the busy-rejection path has its own tests. + if _, err := testPool.Exec(ctx, + `UPDATE agent_task_queue SET status='completed', completed_at=now() WHERE id=$1`, + uuidToString(regenTask.ID)); err != nil { + t.Fatalf("settle regen task: %v", err) + } // Turn 2 lands, so m1 is no longer the latest. t2 := sendDirectChat(t, ctx, agentID, sessionID, "second question") @@ -687,3 +695,79 @@ func TestRegenerateChatQuickActions_StaleTargetRejected(t *testing.T) { t.Fatalf("regenerate new latest turn: %v", err) } } + +// TestRegenerateChatQuickActions_ActiveTurnRejected pins finding §1 of the +// MUL-5149 review: a newer reply that is queued/running but whose assistant row +// has not landed yet leaves the OLD turn as the latest-persisted one, so the +// stale check still passes on it. Regenerating then would resume the session +// after the newer turn advanced its provider state, attaching suggestions built +// from the newer context to the older turn. The in-flight task must refuse it. +func TestRegenerateChatQuickActions_ActiveTurnRejected(t *testing.T) { + if testHandler == nil { + t.Skip("database not available") + } + ctx := context.Background() + agentID, sessionID, _, _ := setupDirectChatSession(t, ctx, "regen busy chat") + + // Turn 1 completes with a resumable session → m1 is the latest assistant turn. + t1 := sendDirectChat(t, ctx, agentID, sessionID, "first") + markTaskRunning(t, ctx, t1) + if _, err := testHandler.TaskService.CompleteTask(ctx, parseUUID(t1), completeResult(t, "first reply"), "sess-1", "/tmp/wd", false, ""); err != nil { + t.Fatalf("complete turn 1: %v", err) + } + session, err := testHandler.Queries.GetChatSession(ctx, parseUUID(sessionID)) + if err != nil { + t.Fatalf("reload session: %v", err) + } + rows := assistantRows(t, ctx, sessionID) + if len(rows) != 1 { + t.Fatalf("expected one assistant turn, got %d", len(rows)) + } + m1 := rows[0].ID + + // Turn 2 is in flight (running, no assistant row yet) so m1 is STILL the + // latest-persisted turn — the stale check passes — but the session is busy. + t2 := sendDirectChat(t, ctx, agentID, sessionID, "second") + markTaskRunning(t, ctx, t2) + + if _, _, err := testHandler.TaskService.RegenerateChatQuickActions(ctx, session, parseUUID(testUserID), m1); !errors.Is(err, service.ErrChatQuickActionsBusy) { + t.Fatalf("busy-session regenerate error = %v, want ErrChatQuickActionsBusy", err) + } +} + +// TestRegenerateChatQuickActions_ConcurrentRefreshRejected pins finding §2 of +// the MUL-5149 review: a second refresh of the same turn while the first pass is +// still in flight is refused rather than enqueuing a duplicate, quota-spending +// pass. A per-client disabled button cannot stop a second client, so the guard +// lives under the session lock next to the enqueue. +func TestRegenerateChatQuickActions_ConcurrentRefreshRejected(t *testing.T) { + if testHandler == nil { + t.Skip("database not available") + } + ctx := context.Background() + agentID, sessionID, _, _ := setupDirectChatSession(t, ctx, "regen dedup chat") + + t1 := sendDirectChat(t, ctx, agentID, sessionID, "q") + markTaskRunning(t, ctx, t1) + if _, err := testHandler.TaskService.CompleteTask(ctx, parseUUID(t1), completeResult(t, "a"), "sess-1", "/tmp/wd", false, ""); err != nil { + t.Fatalf("complete turn 1: %v", err) + } + session, err := testHandler.Queries.GetChatSession(ctx, parseUUID(sessionID)) + if err != nil { + t.Fatalf("reload session: %v", err) + } + rows := assistantRows(t, ctx, sessionID) + if len(rows) != 1 { + t.Fatalf("expected one assistant turn, got %d", len(rows)) + } + m1 := rows[0].ID + + // First refresh is accepted and enqueues a regen pass that now occupies the session. + if _, _, err := testHandler.TaskService.RegenerateChatQuickActions(ctx, session, parseUUID(testUserID), m1); err != nil { + t.Fatalf("first refresh: %v", err) + } + // Second refresh of the same turn while the first pass is still queued is refused. + if _, _, err := testHandler.TaskService.RegenerateChatQuickActions(ctx, session, parseUUID(testUserID), m1); !errors.Is(err, service.ErrChatQuickActionsBusy) { + t.Fatalf("concurrent refresh error = %v, want ErrChatQuickActionsBusy", err) + } +} diff --git a/server/internal/handler/daemon.go b/server/internal/handler/daemon.go index b5a9915f9..cd0f07ca0 100644 --- a/server/internal/handler/daemon.go +++ b/server/internal/handler/daemon.go @@ -2959,7 +2959,7 @@ func (h *Handler) SupplementTaskQuickActions(w http.ResponseWriter, r *http.Requ writeError(w, http.StatusBadRequest, "invalid request body") return } - if err := h.TaskService.SupplementChatQuickActions(r.Context(), task, req.Raw); err != nil { + if err := h.TaskService.SupplementChatQuickActions(r.Context(), task, req.Raw, false); err != nil { slog.Warn("supplement task quick actions failed", "task_id", taskID, "error", err) writeError(w, http.StatusInternalServerError, err.Error()) return diff --git a/server/internal/service/chat_quick_actions.go b/server/internal/service/chat_quick_actions.go index 98a8711d6..f31e4c57d 100644 --- a/server/internal/service/chat_quick_actions.go +++ b/server/internal/service/chat_quick_actions.go @@ -208,7 +208,11 @@ func chatQuickActionsPending(result []byte, msg *db.ChatMessage) bool { // turn that never wrote an assistant row (no_response, channel empty-drop) // returns silently — no client is waiting in that case, because the pending // flag is only ever raised for a written ordinary message. -func (s *TaskService) SupplementChatQuickActions(ctx context.Context, task db.AgentTaskQueue, raw string) error { +// +// failed marks the broadcast as a failure resolution (an explicit refresh whose +// regeneration failed): the pills stay unchanged but the client shows a +// "couldn't refresh" notice instead of treating them as freshly generated. +func (s *TaskService) SupplementChatQuickActions(ctx context.Context, task db.AgentTaskQueue, raw string, failed bool) error { if !task.ChatSessionID.Valid { return nil } @@ -248,6 +252,7 @@ func (s *TaskService) SupplementChatQuickActions(ctx context.Context, task db.Ag TaskID: util.UUIDToString(task.ID), MessageID: util.UUIDToString(msg.ID), QuickActions: []protocol.ChatQuickAction{}, + Failed: failed, } _ = json.Unmarshal(msg.QuickActions, &payload.QuickActions) s.Bus.Publish(events.Event{ diff --git a/server/internal/service/task.go b/server/internal/service/task.go index c758f88a1..735cfd25b 100644 --- a/server/internal/service/task.go +++ b/server/internal/service/task.go @@ -1444,6 +1444,12 @@ var ErrChatQuickActionsNotResumable = errors.New("chat quick actions: session no // chat:quick_actions event will not match (MUL-5149). var ErrChatQuickActionsStale = errors.New("chat quick actions: refresh target is stale") +// ErrChatQuickActionsBusy signals the session already has work in flight — a +// running turn about to change the latest reply, or another regenerate pass — +// so a new refresh is refused to avoid a stale-target supplement or a duplicate +// quota-spending pass (MUL-5149). +var ErrChatQuickActionsBusy = errors.New("chat quick actions: session busy") + // EnqueueChatTask creates a task-owned input batch for a chat session. Channel // media makes the task deferred until binding completes or its durable fallback // deadline expires; other chat tasks are queued immediately. Unlike issue @@ -1680,6 +1686,24 @@ func (s *TaskService) RegenerateChatQuickActions(ctx context.Context, chatSessio if target.ID != expectedMessageID { return ErrChatQuickActionsStale } + // Refuse while the session already has work in flight (MUL-5149 review + // §1/§2). Two races this closes, both under the same session lock as the + // enqueue so neither can slip a sibling insert past it: + // 1. A normal turn is queued/running but its assistant row hasn't + // landed, so the latest-persisted check above still passes on the + // OLD turn. That turn will change the reply and advance the session + // pointer before this regen claims, leaving the suggestions built + // from the newer context but attached to the older turn. + // 2. A second concurrent refresh of the same turn would just enqueue a + // duplicate pass and double-spend quota — the button being disabled + // on one client can't stop another. + busy, berr := qtx.HasActiveChatTaskForSession(ctx, chatSession.ID) + if berr != nil { + return fmt.Errorf("check active chat task: %w", berr) + } + if busy { + return ErrChatQuickActionsBusy + } created, cerr := qtx.CreateChatTask(ctx, db.CreateChatTaskParams{ AgentID: chatSession.AgentID, RuntimeID: carrier.RuntimeID, @@ -3562,13 +3586,14 @@ func (s *TaskService) FailTask(ctx context.Context, taskID pgtype.UUID, errMsg, return &task, nil } -// resolveFailedRegenerateQuickActions clears a refresh's client-side pending -// spinner after the regeneration task itself failed before it could supplement -// (MUL-5149). It loads the target turn the refresh was aimed at and re-broadcasts -// that turn's current (unchanged) quick actions via chat:quick_actions, which -// resolves the pending marker on every client — leaving the old suggestions in -// place rather than stranding the user on a spinner. Best-effort: a lookup miss -// just leaves the client's own skeleton-timeout fallback to clear it. +// resolveFailedRegenerateQuickActions resolves a refresh whose regeneration +// task failed before it could deliver suggestions (MUL-5149). It loads the +// target turn the refresh was aimed at and broadcasts a FAILED +// chat:quick_actions carrying that turn's current (unchanged) pills — which +// resolves the pending marker on every client AND flags the failure so the UI +// shows a "couldn't refresh" notice instead of stopping the spinner on +// unchanged content as if it had succeeded. Best-effort: a lookup miss just +// leaves the client's own timeout fallback to clear the marker. func (s *TaskService) resolveFailedRegenerateQuickActions(ctx context.Context, task db.AgentTaskQueue) { if !task.RegenerateQuickActionsFor.Valid { return @@ -3581,8 +3606,8 @@ func (s *TaskService) resolveFailedRegenerateQuickActions(ctx context.Context, t } // Empty raw parses to no actions, so SupplementChatQuickActions re-broadcasts // the target's existing pills (GetChatMessageByTaskAssistant) and resolves the - // marker without overwriting anything. - if err := s.SupplementChatQuickActions(ctx, target, ""); err != nil { + // marker without overwriting anything; failed=true marks it a failure. + if err := s.SupplementChatQuickActions(ctx, target, "", true); err != nil { slog.Warn("regenerate failure: resolve pending marker failed", "task_id", util.UUIDToString(task.ID), "error", err) } diff --git a/server/pkg/db/generated/chat.sql.go b/server/pkg/db/generated/chat.sql.go index e6d4ab4e4..37df27cc7 100644 --- a/server/pkg/db/generated/chat.sql.go +++ b/server/pkg/db/generated/chat.sql.go @@ -869,6 +869,29 @@ func (q *Queries) GetPendingChatTask(ctx context.Context, chatSessionID pgtype.U return i, err } +const hasActiveChatTaskForSession = `-- name: HasActiveChatTaskForSession :one +SELECT EXISTS ( + SELECT 1 FROM agent_task_queue + WHERE chat_session_id = $1 + AND status IN ('queued', 'dispatched', 'running', 'waiting_local_directory') +) AS has_active +` + +// True while ANY task — a normal user turn OR a background quick-actions +// regenerate — is in flight for the session (contrast GetPendingChatTask, which +// hides regenerate passes from the UI). A quick-actions refresh is refused when +// this is true: a running turn is about to change the latest reply, so the +// target we'd resume is already stale even before its assistant row lands; and a +// second concurrent regenerate would double-spend quota on the same turn. Read +// inside the same session lock as the enqueue so it cannot race a sibling insert +// (MUL-5149 review §1/§2). +func (q *Queries) HasActiveChatTaskForSession(ctx context.Context, chatSessionID pgtype.UUID) (bool, error) { + row := q.db.QueryRow(ctx, hasActiveChatTaskForSession, chatSessionID) + var has_active bool + err := row.Scan(&has_active) + return has_active, err +} + const hasPendingChatTasksByCreator = `-- name: HasPendingChatTasksByCreator :one SELECT EXISTS ( SELECT 1 diff --git a/server/pkg/db/queries/chat.sql b/server/pkg/db/queries/chat.sql index a2164dda7..36a0defbc 100644 --- a/server/pkg/db/queries/chat.sql +++ b/server/pkg/db/queries/chat.sql @@ -456,6 +456,21 @@ WHERE session_id NOT IN (SELECT session_id FROM retired_sessions) ORDER BY completed_at DESC LIMIT 1; +-- name: HasActiveChatTaskForSession :one +-- True while ANY task — a normal user turn OR a background quick-actions +-- regenerate — is in flight for the session (contrast GetPendingChatTask, which +-- hides regenerate passes from the UI). A quick-actions refresh is refused when +-- this is true: a running turn is about to change the latest reply, so the +-- target we'd resume is already stale even before its assistant row lands; and a +-- second concurrent regenerate would double-spend quota on the same turn. Read +-- inside the same session lock as the enqueue so it cannot race a sibling insert +-- (MUL-5149 review §1/§2). +SELECT EXISTS ( + SELECT 1 FROM agent_task_queue + WHERE chat_session_id = $1 + AND status IN ('queued', 'dispatched', 'running', 'waiting_local_directory') +) AS has_active; + -- name: GetPendingChatTask :one -- Returns the most recent in-flight task for a chat session, if any. -- Used by the frontend to recover pending state after refresh / reopen. diff --git a/server/pkg/protocol/messages.go b/server/pkg/protocol/messages.go index 953089d88..820bd2086 100644 --- a/server/pkg/protocol/messages.go +++ b/server/pkg/protocol/messages.go @@ -141,6 +141,13 @@ type ChatQuickActionsPayload struct { TaskID string `json:"task_id"` MessageID string `json:"message_id"` QuickActions []ChatQuickAction `json:"quick_actions"` + // Failed marks a supplement that resolves the client's refresh spinner + // because the regeneration FAILED (the provider pass or its delivery), not + // because it produced new suggestions. QuickActions then carries the turn's + // unchanged pills; the client shows a "couldn't refresh" notice instead of + // treating unchanged content as a silent success (MUL-5149). Omitted (false) + // on the normal success path and for the automatic best-effort pass. + Failed bool `json:"failed,omitempty"` } // TaskMessagePayload represents a single agent execution message (tool call, text, etc.)