diff --git a/server/internal/handler/chat_input_ownership_test.go b/server/internal/handler/chat_input_ownership_test.go index 7d52d53de4..93cfe4a344 100644 --- a/server/internal/handler/chat_input_ownership_test.go +++ b/server/internal/handler/chat_input_ownership_test.go @@ -332,3 +332,55 @@ func assertNoRetryChild(t *testing.T, ctx context.Context, taskID string) { t.Fatalf("expected no retry child for a completed no_response turn, got %d", n) } } + +// insertChannelChatTask creates a running chat task with chat_input_task_id NULL +// — the legacy/channel (Slack/Lark) shape — directly, bypassing the task-owned +// direct-send path. +func insertChannelChatTask(t *testing.T, ctx context.Context, agentID, runtimeID, sessionID string) string { + t.Helper() + var taskID string + if err := testPool.QueryRow(ctx, ` + INSERT INTO agent_task_queue (agent_id, runtime_id, chat_session_id, status, priority, started_at, dispatched_at) + VALUES ($1, $2, $3, 'running', 2, now(), now()) + RETURNING id + `, agentID, runtimeID, sessionID).Scan(&taskID); err != nil { + t.Fatalf("setup: create channel chat task: %v", err) + } + return taskID +} + +// TestCompleteTask_ChannelEmptyOutputWritesNoRow pins the MUL-4351 review fix: +// a legacy/channel task (chat_input_task_id NULL) that completes with empty +// output must NOT write an assistant row — so chat:done carries empty content +// and the Slack/Lark outbound keeps silently dropping it. The no_response +// fallback body must never reach an external channel. A non-empty channel +// completion still writes an ordinary message. +func TestCompleteTask_ChannelEmptyOutputWritesNoRow(t *testing.T) { + if testHandler == nil { + t.Skip("database not available") + } + ctx := context.Background() + agentID, sessionID, runtimeID, _ := setupDirectChatSession(t, ctx, "channel-like chat") + + // Empty output → no row at all. + emptyTask := insertChannelChatTask(t, ctx, agentID, runtimeID, sessionID) + if _, err := testHandler.TaskService.CompleteTask(ctx, parseUUID(emptyTask), completeResult(t, " "), "", ""); err != nil { + t.Fatalf("complete channel task (empty): %v", err) + } + if rows := assistantRows(t, ctx, sessionID); len(rows) != 0 { + t.Fatalf("channel empty completion must write NO assistant row (Slack/Lark silent-drop), got %d", len(rows)) + } + + // Non-empty output → one ordinary message (kind 'message', not no_response). + textTask := insertChannelChatTask(t, ctx, agentID, runtimeID, sessionID) + if _, err := testHandler.TaskService.CompleteTask(ctx, parseUUID(textTask), completeResult(t, "channel reply"), "", ""); err != nil { + t.Fatalf("complete channel task (text): %v", err) + } + rows := assistantRows(t, ctx, sessionID) + if len(rows) != 1 { + t.Fatalf("channel non-empty completion must write exactly one message, got %d", len(rows)) + } + if rows[0].MessageKind != protocol.ChatMessageKindMessage || rows[0].Content != "channel reply" { + t.Fatalf("channel message = kind %q content %q, want message/'channel reply'", rows[0].MessageKind, rows[0].Content) + } +} diff --git a/server/internal/handler/daemon.go b/server/internal/handler/daemon.go index 25711c9af8..51c72bc9ae 100644 --- a/server/internal/handler/daemon.go +++ b/server/internal/handler/daemon.go @@ -1774,38 +1774,38 @@ func (h *Handler) ClaimTaskByRuntime(w http.ResponseWriter, r *http.Request) { } } } - // Build the chat prompt from EVERY user message that has arrived - // since the agent's last reply — not just the most recent one. A - // short-window debounce (MUL-2968) can land several user messages - // before a single run fires; the agent resumes its prior session - // and only learns of new input through resp.ChatMessage, so - // delivering just the latest message would silently drop the - // earlier ones (e.g. "看上海天气" then "还有青岛" → only Qingdao - // answered). The unanswered set is the trailing run of user - // messages after the last assistant message (every completed or - // failed run writes an assistant row, so that anchor advances each - // turn). Attachments are collected from each included message so - // the agent can `multica attachment download ` — the markdown - // URL alone is signed and 30-min expiring on the private CDN. // Resolve the user-message input batch for this run. A task-owned // direct-chat task (chat_input_task_id set, MUL-4351) reads exactly // the user messages tagged with its own input owner, so a message // that arrived after this turn was sealed can never be absorbed here. // Legacy and channel (Slack/Lark) tasks carry a NULL owner and keep - // the trailing-message selector, so a rolling deploy never replays - // their history. + // the trailing-message selector — the run of user messages after the + // last assistant row, which also covers a debounced burst (MUL-2968: + // "看上海天气" then "还有青岛" must both be delivered) — so a rolling + // deploy never replays their history. Attachments are collected per + // included message so the agent can `multica attachment download ` + // (the inline markdown URL is signed + 30-min expiring on the CDN). var unanswered []db.ChatMessage + var inputLoadErr error if task.ChatInputTaskID.Valid { - if owned, err := h.Queries.ListChatInputMessages(r.Context(), task.ChatInputTaskID); err == nil { - unanswered = owned - } else { - slog.Warn("chat claim: load owned input messages failed", - "task_id", uuidToString(task.ID), - "chat_input_task_id", uuidToString(task.ChatInputTaskID), - "error", err) - } - } else if msgs, err := h.Queries.ListChatMessages(r.Context(), cs.ID); err == nil && len(msgs) > 0 { + unanswered, inputLoadErr = h.Queries.ListChatInputMessages(r.Context(), task.ChatInputTaskID) + } else if msgs, err := h.Queries.ListChatMessages(r.Context(), cs.ID); err == nil { unanswered = trailingUserMessages(msgs) + } else { + inputLoadErr = err + } + // A read failure must NOT masquerade as "zero input". Preserve the + // just-dispatched task (the stale-dispatched reclaim redelivers it) + // and reject the claim with 5xx, rather than cancelling a valid direct + // task on a transient DB error (MUL-4351 review). + if inputLoadErr != nil { + outcome = "error_chat_input_load" + slog.Error("chat claim: load chat input messages failed; preserving task for redelivery", + "task_id", uuidToString(task.ID), + "chat_session_id", uuidToString(cs.ID), + "error", inputLoadErr) + writeError(w, http.StatusInternalServerError, "failed to load chat input") + return } parts := make([]string, 0, len(unanswered)) diff --git a/server/internal/service/task.go b/server/internal/service/task.go index eb15a0b5e5..3e8cbcebdb 100644 --- a/server/internal/service/task.go +++ b/server/internal/service/task.go @@ -1769,18 +1769,18 @@ func (s *TaskService) CompleteTask(ctx context.Context, taskID pgtype.UUID, resu return fmt.Errorf("update chat session resume pointer: %w", err) } - // Write exactly one assistant outcome in the SAME transaction as the - // status flip and resume-pointer update (MUL-4351). A non-empty final - // output becomes an ordinary assistant message; an empty/whitespace - // output becomes a visible no_response outcome. Failing here rolls the - // whole completion back so the daemon retries the terminal callback — - // we never leave a completed task without its single outcome row, and - // the status CAS above guarantees a replay can't write a second one. + // Write the assistant outcome in the SAME transaction as the status + // flip and resume-pointer update (MUL-4351). For a task-owned direct + // task this is exactly one row (message or no_response); for a + // legacy/channel task an empty output writes no row (see + // writeChatCompletionOutcome). Failing here rolls the whole completion + // back so the daemon retries the terminal callback, and the status CAS + // above guarantees a replay can't write a second row. msg, err := s.writeChatCompletionOutcome(ctx, qtx, t, result) if err != nil { return fmt.Errorf("write chat assistant outcome: %w", err) } - chatAssistantMsg = &msg + chatAssistantMsg = msg } return nil }); err != nil { @@ -1898,14 +1898,24 @@ func (s *TaskService) CompleteTask(ctx context.Context, taskID pgtype.UUID, resu // message_kind still show this text instead of an empty bubble (MUL-4351). const chatNoResponseFallback = "The agent finished this turn without a text reply." -// writeChatCompletionOutcome writes exactly one assistant chat_message for a -// completed direct-chat task, inside the caller's completion transaction. A -// non-empty final output becomes an ordinary assistant message; an empty or -// whitespace-only output becomes a visible no_response outcome carrying a -// non-empty English fallback body. It never auto-retries: an empty output is a -// legitimate terminal result (a tool-only turn), and re-running it would repeat -// any external side effects the agent already performed. -func (s *TaskService) writeChatCompletionOutcome(ctx context.Context, qtx *db.Queries, task db.AgentTaskQueue, result []byte) (db.ChatMessage, error) { +// writeChatCompletionOutcome writes the assistant chat_message outcome for a +// completed chat task inside the caller's completion transaction, returning the +// row (nil when none is written). +// +// Task-owned direct (web/mobile) tasks — chat_input_task_id set — get the +// explicit single-outcome contract: a non-empty final output becomes an ordinary +// assistant message, and an empty/whitespace output becomes a visible +// no_response outcome carrying a non-empty English fallback body. It never +// auto-retries: an empty output is a legitimate terminal result (a tool-only +// turn) and re-running it would repeat side effects already performed. +// +// Legacy and channel (Slack/Lark) tasks — chat_input_task_id NULL — keep the +// prior behavior: a non-empty output writes an ordinary assistant message, but +// an EMPTY output writes NO row, so chat:done carries empty content and the +// channel outbound silently drops it. This preserves Slack/Lark empty-completion +// semantics unchanged (MUL-4351 review): the no_response fallback body must never +// be pushed to an external channel. +func (s *TaskService) writeChatCompletionOutcome(ctx context.Context, qtx *db.Queries, task db.AgentTaskQueue, result []byte) (*db.ChatMessage, error) { // result is the daemon request re-marshalled by the handler, so it is always // valid JSON; an empty Output is the only case this branch cares about. var payload protocol.TaskCompletedPayload @@ -1913,6 +1923,13 @@ func (s *TaskService) writeChatCompletionOutcome(ctx context.Context, qtx *db.Qu // Same unescape as the issue-comment path: literal `\n` from agent stdout // becomes a real newline so the chat panel renders paragraph breaks. body := util.UnescapeBackslashEscapes(payload.Output) + isEmpty := strings.TrimSpace(body) == "" + + // Channel/legacy empty completion: emit no assistant row, only an empty + // chat:done for typing/lifecycle. Keeps the Slack/Lark silent-drop path. + if isEmpty && !task.ChatInputTaskID.Valid { + return nil, nil + } params := db.CreateChatMessageParams{ ChatSessionID: task.ChatSessionID, @@ -1920,14 +1937,19 @@ func (s *TaskService) writeChatCompletionOutcome(ctx context.Context, qtx *db.Qu TaskID: task.ID, ElapsedMs: computeChatElapsedMs(task), } - if strings.TrimSpace(body) == "" { + if isEmpty { + // Task-owned direct task: explicit, visible no_response outcome. params.Content = chatNoResponseFallback params.MessageKind = pgtype.Text{String: protocol.ChatMessageKindNoResponse, Valid: true} } else { params.Content = redact.Text(body) // message_kind left NULL → COALESCE defaults to 'message'. } - return qtx.CreateChatMessage(ctx, params) + row, err := qtx.CreateChatMessage(ctx, params) + if err != nil { + return nil, err + } + return &row, nil } // FailTask marks a task as failed.