Files
multica/server/internal/handler/chat_input_ownership_test.go
Jiayuan Zhang f13969b996 refactor(chat): generate quick actions server-side via the LLM layer (MUL-5573) (#6214)
* refactor(chat): generate quick actions server-side via the LLM layer (MUL-5573)

Follow-up suggestions were produced by a second, full provider CLI invocation
per chat turn: the daemon resumed the just-finished session and ran a
suggestion-only pass. That pass inherited the main turn's exec options, so its
20s budget had to cover process spawn, every MCP handshake, session replay, and
model reasoning at the agent's own thinking level — typically 8-15s of visible
skeleton, and every turn paid two provider cold starts.

Generate them here instead, through the same pkg/llm layer that backs chat
auto-titling. Suggestions need no tools, workdir, or agent identity — only the
tail of the conversation — so a bounded 8s call on the deployment's small model
replaces the whole resumed turn.

Quality changes that came with the move:

  - The prompt now states the frame explicitly ("you write FOR THE USER"). The
    old pass ran inside the agent's session and inherited the runtime brief's
    identity, which drifted suggestions toward agent-operations actions.
  - Previously-offered labels are replayed as ALREADY SUGGESTED. The old
    architecture had the opposite effect: on providers that append on resume,
    each pass saw its predecessor's JSON and anchored on it.
  - A failed generation broadcasts failed=true. Before, a timeout delivered an
    empty array — indistinguishable from "nothing worth suggesting", so every
    slow pass read as a quality problem.
  - The in-band footer is still stripped from replies but its actions are now
    discarded, so a pre-upgrade session is not pinned to the retired
    suggestions with the replacing pass suppressed.

The refresh path no longer enqueues an agent task: it validates the target and
calls the same generator, which also drops the not-resumable refusal — a session
whose runtime was rebound can now be refreshed. Client contract is unchanged
(chat:done pending flag, chat:quick_actions supplement); the only frontend
change is the pending window, resized from 30s to 12s to match the new budget.

Also removes the daemon's TMPDIR-after-cleanup hazard by construction: the old
pass started after runTask's defers had already deleted the temp dir it was
still pointed at.

Co-authored-by: multica-agent <github@multica.ai>

* refactor(chat): drop the quick-actions opt-out setting (MUL-5573)

Suggestions are always on. The Settings → Chat toggle is removed along with
the whole per-turn opt-out path it fed: the persisted client preference, the
quick_actions_enabled send field, the quick_actions_disabled task stamp, and
the eligibility gate that read it.

The toggle predates server-side generation, when it could only hide chips a
provider pass had already paid for. Now that generation is a bounded call the
server decides on, an off switch buys nothing a user would miss, and it was
the last piece of UI implying the feature might be unavailable.

agent_task_queue.quick_actions_disabled is no longer written (dropped from
CreateChatTask's INSERT; the column keeps its false default). Left in place
alongside regenerate_quick_actions_for for a later drop migration — removing
columns an already-running binary still inserts would break mid-deploy.

Co-authored-by: multica-agent <github@multica.ai>

* fix(chat): address quick-actions review findings (MUL-5573)

Four defects from review of the server-side generation change.

1. Automatic failures were reported as refresh failures. The generator
   broadcast failed=true on any LLM error, but the client turns every
   failed=true into a "couldn't refresh" toast — so an automatic timeout
   popped a toast for an action the user never took. This also contradicted
   ChatQuickActionsPayload.Failed, which documents false for the automatic
   pass. The caller now passes its origin; only an explicit refresh reports.

2. Generation context was not bound to the target turn. The pass re-read the
   session's newest messages while always writing to the task it was handed,
   so a turn landing between the completion callback and the detached read
   supplied the context for a reply it did not belong to. Worse, a user
   typing a follow-up in the second after a reply left the window ending on
   a user row, which the old code treated as "nothing to build on" — that
   turn silently never got pills. The window is now anchored on the target
   assistant message and queried strictly before it.

3. No concurrency or idempotency bound on generation. Refresh stopped
   creating a task, so the busy check could not see a pass already running:
   two refreshes both returned 202, spent two upstream calls, and raced to
   write one row. Nothing bounded generation process-wide either. Adds a
   per-session in-flight guard (refresh now 409s on a duplicate) and a
   process-wide ceiling; a shed pass still resolves the client placeholder
   so no skeleton hangs on work that never started.

4. A new daemon could not safely talk to an older server. The refresh task
   discriminator was deleted, so a regenerate task from such a server fell
   through to the ordinary chat path: no user message, but the agent would
   answer anyway and the server would persist it as a real reply. The field
   is restored as a refusal marker only — the task completes empty, which is
   the shape the retired pass produced and which that server writes no row
   for. Not a restored execution path.

Co-authored-by: multica-agent <github@multica.ai>

---------

Co-authored-by: Lambda <lambda@multica.ai>
Co-authored-by: multica-agent <github@multica.ai>
2026-07-31 15:59:50 +08:00

841 lines
35 KiB
Go

package handler
import (
"context"
"encoding/json"
"errors"
"strings"
"testing"
"github.com/multica-ai/multica/server/internal/service"
db "github.com/multica-ai/multica/server/pkg/db/generated"
"github.com/multica-ai/multica/server/pkg/protocol"
)
// setupDirectChatSession creates a runtime-guard agent (with its registered
// runtime + daemon) and a non-intro chat session for direct (web/mobile) chat
// ownership tests.
func setupDirectChatSession(t *testing.T, ctx context.Context, title string) (agentID, sessionID, runtimeID, daemonID string) {
t.Helper()
agentID, runtimeID, daemonID = createRuntimeGuardAgent(t, ctx)
if err := testPool.QueryRow(ctx, `
INSERT INTO chat_session (workspace_id, agent_id, creator_id, title)
VALUES ($1, $2, $3, $4)
RETURNING id
`, testWorkspaceID, agentID, testUserID, title).Scan(&sessionID); err != nil {
t.Fatalf("setup: create chat session: %v", err)
}
t.Cleanup(func() { testPool.Exec(ctx, `DELETE FROM chat_session WHERE id = $1`, sessionID) })
return agentID, sessionID, runtimeID, daemonID
}
// sendDirectChat drives the transactional direct-send service path and returns
// the owning task id. The user message is created inside the same transaction
// with task_id = the new task, and the task owns its own input batch.
func sendDirectChat(t *testing.T, ctx context.Context, agentID, sessionID, content string) string {
t.Helper()
sess, err := testHandler.Queries.GetChatSession(ctx, parseUUID(sessionID))
if err != nil {
t.Fatalf("load chat session: %v", err)
}
ag, err := testHandler.Queries.GetAgent(ctx, parseUUID(agentID))
if err != nil {
t.Fatalf("load agent: %v", err)
}
res, err := testHandler.TaskService.SendDirectChatMessage(ctx, sess, ag, parseUUID(testUserID), content, nil, "member", parseUUID(testUserID))
if err != nil {
t.Fatalf("SendDirectChatMessage: %v", err)
}
return uuidToString(res.Task.ID)
}
func markTaskRunning(t *testing.T, ctx context.Context, taskID string) {
t.Helper()
if _, err := testPool.Exec(ctx, `
UPDATE agent_task_queue SET status = 'running', started_at = now(), dispatched_at = now()
WHERE id = $1
`, taskID); err != nil {
t.Fatalf("mark task running: %v", err)
}
}
func completeResult(t *testing.T, output string) []byte {
t.Helper()
b, err := json.Marshal(TaskCompleteRequest{Output: output})
if err != nil {
t.Fatalf("marshal complete result: %v", err)
}
return b
}
// TestDirectChat_TaskOwnsItsOwnInputBatch is the core input-boundary contract
// (MUL-4351): each direct send owns exactly the user message it created. When
// U1→T1 and U2→T2 are both queued, T1's claim must deliver ONLY U1 (never
// "U1\n\nU2" the way the trailing-message selector would), and after T1
// completes, T2 delivers ONLY U2.
func TestDirectChat_TaskOwnsItsOwnInputBatch(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
agentID, sessionID, runtimeID, daemonID := setupDirectChatSession(t, ctx, "ownership chat")
t1 := sendDirectChat(t, ctx, agentID, sessionID, "看上海天气")
t2 := sendDirectChat(t, ctx, agentID, sessionID, "还有青岛")
// Both tasks own their own input batch.
assertTaskInputOwner(t, ctx, t1, t1)
assertTaskInputOwner(t, ctx, t2, t2)
claimed := claimTaskForRuntimeGuard(t, runtimeID, daemonID)
if claimed.ChatMessage != "看上海天气" {
t.Fatalf("first claim must deliver ONLY its owned message; got %q", claimed.ChatMessage)
}
// Complete the first task (the older T1, claimed first by created_at order),
// then the next claim must deliver only the other message rather than a
// coalesced pair. The claim leaves T1 dispatched; move it to running so the
// completion CAS (WHERE status='running') applies.
markTaskRunning(t, ctx, t1)
if _, err := testHandler.TaskService.CompleteTask(ctx, parseUUID(t1), completeResult(t, "上海晴"), "", "", false, ""); err != nil {
t.Fatalf("complete first task: %v", err)
}
claimed2 := claimTaskForRuntimeGuard(t, runtimeID, daemonID)
if claimed2.ChatMessage != "还有青岛" {
t.Fatalf("second claim must deliver ONLY the second owned message; got %q", claimed2.ChatMessage)
}
}
// TestDirectChat_RunningTaskDoesNotAbsorbNewMessage pins the acceptance case:
// while T1 is running, a message sent from another surface lands on T2 and is
// never folded into T1's already-sealed input batch.
func TestDirectChat_RunningTaskDoesNotAbsorbNewMessage(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
agentID, sessionID, _, _ := setupDirectChatSession(t, ctx, "no-absorb chat")
t1 := sendDirectChat(t, ctx, agentID, sessionID, "first")
markTaskRunning(t, ctx, t1)
// A second message arrives mid-run and gets its own task.
t2 := sendDirectChat(t, ctx, agentID, sessionID, "second")
if t1 == t2 {
t.Fatal("second send must create a distinct task")
}
// T1's owned batch is still exactly {first}; the mid-run message belongs to T2.
owned1, err := testHandler.Queries.ListChatInputMessages(ctx, parseUUID(t1))
if err != nil {
t.Fatalf("list T1 owned input: %v", err)
}
if len(owned1) != 1 || owned1[0].Content != "first" {
t.Fatalf("T1 must own only its own message; got %+v", msgContents(owned1))
}
owned2, err := testHandler.Queries.ListChatInputMessages(ctx, parseUUID(t2))
if err != nil {
t.Fatalf("list T2 owned input: %v", err)
}
if len(owned2) != 1 || owned2[0].Content != "second" {
t.Fatalf("T2 must own the mid-run message; got %+v", msgContents(owned2))
}
}
// TestCompleteTask_ChatEmptyOutputWritesNoResponse: an empty final output is a
// visible, terminal no_response outcome — exactly one assistant row with
// message_kind='no_response' and a non-empty fallback body, task completed, and
// no auto-retry.
func TestCompleteTask_ChatEmptyOutputWritesNoResponse(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
agentID, sessionID, _, _ := setupDirectChatSession(t, ctx, "no-response chat")
taskID := sendDirectChat(t, ctx, agentID, sessionID, "do a tool-only thing")
markTaskRunning(t, ctx, taskID)
// Whitespace-only output trims to empty → no_response.
if _, err := testHandler.TaskService.CompleteTask(ctx, parseUUID(taskID), completeResult(t, " "), "", "", false, ""); err != nil {
t.Fatalf("complete task: %v", err)
}
rows := assistantRows(t, ctx, sessionID)
if len(rows) != 1 {
t.Fatalf("expected exactly one assistant outcome, got %d", len(rows))
}
if rows[0].MessageKind != protocol.ChatMessageKindNoResponse {
t.Fatalf("expected message_kind=no_response, got %q", rows[0].MessageKind)
}
if rows[0].Content == "" {
t.Fatal("no_response row must carry a non-empty fallback body for old clients")
}
assertTaskStatus(t, ctx, taskID, "completed")
assertNoRetryChild(t, ctx, taskID)
}
// TestCompleteTask_ChatNonEmptyOutputWritesMessage: a normal reply becomes a
// single ordinary assistant message.
func TestCompleteTask_ChatNonEmptyOutputWritesMessage(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
agentID, sessionID, _, _ := setupDirectChatSession(t, ctx, "reply chat")
taskID := sendDirectChat(t, ctx, agentID, sessionID, "hello")
markTaskRunning(t, ctx, taskID)
if _, err := testHandler.TaskService.CompleteTask(ctx, parseUUID(taskID), completeResult(t, "hi there"), "sess-1", "/tmp/wd", false, ""); err != nil {
t.Fatalf("complete task: %v", err)
}
rows := assistantRows(t, ctx, sessionID)
if len(rows) != 1 {
t.Fatalf("expected exactly one assistant message, got %d", len(rows))
}
if rows[0].MessageKind != protocol.ChatMessageKindMessage {
t.Fatalf("expected message_kind=message, got %q", rows[0].MessageKind)
}
if rows[0].Content != "hi there" {
t.Fatalf("expected content 'hi there', got %q", rows[0].Content)
}
}
// TestCompleteTask_ChatCallbackIdempotent: a replayed completion callback must
// not write a second assistant outcome.
func TestCompleteTask_ChatCallbackIdempotent(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
agentID, sessionID, _, _ := setupDirectChatSession(t, ctx, "idempotent chat")
taskID := sendDirectChat(t, ctx, agentID, sessionID, "hey")
markTaskRunning(t, ctx, taskID)
res := completeResult(t, "reply")
if _, err := testHandler.TaskService.CompleteTask(ctx, parseUUID(taskID), res, "", "", false, ""); err != nil {
t.Fatalf("first complete: %v", err)
}
// Replay: the status CAS fails, so this is an idempotent no-op success.
if _, err := testHandler.TaskService.CompleteTask(ctx, parseUUID(taskID), res, "", "", false, ""); err != nil {
t.Fatalf("replayed complete must be idempotent success, got %v", err)
}
if rows := assistantRows(t, ctx, sessionID); len(rows) != 1 {
t.Fatalf("expected exactly one assistant outcome after replay, got %d", len(rows))
}
}
// TestFailTask_ChatRetryInheritsInputOwnerAndPriority: a transient failure of a
// task-owned direct task creates a retry child that reuses the SAME input owner
// (so it reads the same user messages) and is queued at a bumped priority so it
// is claimed ahead of fresh chat tasks.
func TestFailTask_ChatRetryInheritsInputOwnerAndPriority(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
agentID, sessionID, _, _ := setupDirectChatSession(t, ctx, "retry chat")
rootID := sendDirectChat(t, ctx, agentID, sessionID, "root question")
markTaskRunning(t, ctx, rootID)
if _, err := testHandler.TaskService.FailTask(ctx, parseUUID(rootID), "runtime went away", "", "", "runtime_offline", false, ""); err != nil {
t.Fatalf("fail task: %v", err)
}
var childID, childOwner string
var childPriority, childAttempt int
var childStatus string
if err := testPool.QueryRow(ctx, `
SELECT id, chat_input_task_id, priority, status, attempt
FROM agent_task_queue
WHERE parent_task_id = $1
`, rootID).Scan(&childID, &childOwner, &childPriority, &childStatus, &childAttempt); err != nil {
t.Fatalf("expected a retry child, got: %v", err)
}
if childOwner != rootID {
t.Fatalf("retry child must inherit the root input owner %s, got %s", rootID, childOwner)
}
if childPriority < 3 {
t.Fatalf("chat retry must be bumped above fresh chat priority (2); got %d", childPriority)
}
if childStatus != "queued" {
t.Fatalf("retry child must be queued, got %q", childStatus)
}
// The root direct task starts at attempt 1, so its first retry is attempt 2.
var rootAttempt int
if err := testPool.QueryRow(ctx, `SELECT attempt FROM agent_task_queue WHERE id = $1`, rootID).Scan(&rootAttempt); err != nil {
t.Fatalf("read root attempt: %v", err)
}
if childAttempt != rootAttempt+1 {
t.Fatalf("retry child attempt must be root+1 (%d), got %d", rootAttempt+1, childAttempt)
}
// The child reads the same input batch: the root's user message.
owned, err := testHandler.Queries.ListChatInputMessages(ctx, parseUUID(childOwner))
if err != nil {
t.Fatalf("list child owned input: %v", err)
}
if len(owned) != 1 || owned[0].Content != "root question" {
t.Fatalf("retry child must read the root input batch; got %+v", msgContents(owned))
}
}
// ---- helpers ----
func msgContents(msgs []db.ChatMessage) []string {
out := make([]string, 0, len(msgs))
for _, m := range msgs {
out = append(out, m.Content)
}
return out
}
func assistantRows(t *testing.T, ctx context.Context, sessionID string) []db.ChatMessage {
t.Helper()
all, err := testHandler.Queries.ListChatMessages(ctx, parseUUID(sessionID))
if err != nil {
t.Fatalf("list chat messages: %v", err)
}
var out []db.ChatMessage
for _, m := range all {
if m.Role == "assistant" {
out = append(out, m)
}
}
return out
}
func assertTaskInputOwner(t *testing.T, ctx context.Context, taskID, wantOwner string) {
t.Helper()
var owner string
if err := testPool.QueryRow(ctx, `SELECT chat_input_task_id FROM agent_task_queue WHERE id = $1`, taskID).Scan(&owner); err != nil {
t.Fatalf("read chat_input_task_id: %v", err)
}
if owner != wantOwner {
t.Fatalf("task %s input owner = %s, want %s", taskID, owner, wantOwner)
}
}
func assertTaskStatus(t *testing.T, ctx context.Context, taskID, want string) {
t.Helper()
var status string
if err := testPool.QueryRow(ctx, `SELECT status FROM agent_task_queue WHERE id = $1`, taskID).Scan(&status); err != nil {
t.Fatalf("read task status: %v", err)
}
if status != want {
t.Fatalf("task %s status = %q, want %q", taskID, status, want)
}
}
func assertNoRetryChild(t *testing.T, ctx context.Context, taskID string) {
t.Helper()
var n int
if err := testPool.QueryRow(ctx, `SELECT count(*) FROM agent_task_queue WHERE parent_task_id = $1`, taskID).Scan(&n); err != nil {
t.Fatalf("count retry children: %v", err)
}
if n != 0 {
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
}
// insertSealedChannelChatTask creates a running channel-shaped chat task the
// way EnqueueChatTask now does: the task owns its input batch
// (chat_input_task_id = id) and the sealed user message carries the immutable
// channel_ingested stamp.
func insertSealedChannelChatTask(t *testing.T, ctx context.Context, agentID, runtimeID, sessionID, content 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 sealed channel chat task: %v", err)
}
if _, err := testPool.Exec(ctx, `
UPDATE agent_task_queue SET chat_input_task_id = id WHERE id = $1
`, taskID); err != nil {
t.Fatalf("setup: set input owner: %v", err)
}
if _, err := testPool.Exec(ctx, `
INSERT INTO chat_message (chat_session_id, role, content, task_id, channel_ingested)
VALUES ($1, 'user', $2, $3, TRUE)
`, sessionID, content, taskID); err != nil {
t.Fatalf("setup: seal channel user message: %v", err)
}
return taskID
}
// TestCompleteTask_ChannelEmptyOutputWritesNoRow pins the MUL-4351 review fix
// for the LEGACY channel shape (chat_input_task_id NULL): an empty completion
// 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, " "), "", "", false, ""); 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"), "", "", false, ""); 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)
}
}
// TestCompleteTask_SealedChannelEmptyOutputWritesNoRow pins the silent-drop
// contract for the NEW channel shape: sealed channel tasks own their input
// batch (chat_input_task_id = id), so the discriminator is the immutable
// channel_ingested stamp, not a NULL owner. An empty completion must write no
// assistant row — the outbound patcher forwards any non-empty content
// verbatim, so a no_response fallback row would be pushed to Feishu/Slack.
func TestCompleteTask_SealedChannelEmptyOutputWritesNoRow(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
agentID, sessionID, runtimeID, _ := setupDirectChatSession(t, ctx, "sealed channel chat")
emptyTask := insertSealedChannelChatTask(t, ctx, agentID, runtimeID, sessionID, "[Image]")
if _, err := testHandler.TaskService.CompleteTask(ctx, parseUUID(emptyTask), completeResult(t, " "), "", "", false, ""); err != nil {
t.Fatalf("complete sealed channel task (empty): %v", err)
}
if rows := assistantRows(t, ctx, sessionID); len(rows) != 0 {
t.Fatalf("sealed channel empty completion must write NO assistant row, got %d", len(rows))
}
// Non-empty output still writes one ordinary message.
textTask := insertSealedChannelChatTask(t, ctx, agentID, runtimeID, sessionID, "hello")
if _, err := testHandler.TaskService.CompleteTask(ctx, parseUUID(textTask), completeResult(t, "sealed channel reply"), "", "", false, ""); err != nil {
t.Fatalf("complete sealed channel task (text): %v", err)
}
rows := assistantRows(t, ctx, sessionID)
if len(rows) != 1 || rows[0].MessageKind != protocol.ChatMessageKindMessage || rows[0].Content != "sealed channel reply" {
t.Fatalf("sealed channel non-empty completion = %+v, want one ordinary message", rows)
}
}
// TestCompleteTask_SealedChannelRetryEmptyOutputWritesNoRow: an auto-retry
// clone inherits its parent's chat_input_task_id while the sealed messages
// stay tagged with the parent's id. The provenance check must follow the
// inherited owner — keying it off the retry's own id would misread the task
// as direct and push the no_response fallback to the external channel.
func TestCompleteTask_SealedChannelRetryEmptyOutputWritesNoRow(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
agentID, sessionID, runtimeID, _ := setupDirectChatSession(t, ctx, "sealed channel retry chat")
parentTask := insertSealedChannelChatTask(t, ctx, agentID, runtimeID, sessionID, "[Video]")
var retryTask string
if err := testPool.QueryRow(ctx, `
INSERT INTO agent_task_queue (agent_id, runtime_id, chat_session_id, status, priority, started_at, dispatched_at, parent_task_id, retry_of_task_id, chat_input_task_id)
VALUES ($1, $2, $3, 'running', 2, now(), now(), $4, $4, $4)
RETURNING id
`, agentID, runtimeID, sessionID, parentTask).Scan(&retryTask); err != nil {
t.Fatalf("setup: create retry clone: %v", err)
}
if _, err := testHandler.TaskService.CompleteTask(ctx, parseUUID(retryTask), completeResult(t, ""), "", "", false, ""); err != nil {
t.Fatalf("complete sealed channel retry (empty): %v", err)
}
if rows := assistantRows(t, ctx, sessionID); len(rows) != 0 {
t.Fatalf("sealed channel retry empty completion must write NO assistant row, got %d", len(rows))
}
}
func TestCompleteTask_ChatQuickActions(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
agentID, sessionID, _, _ := setupDirectChatSession(t, ctx, "quick-actions chat")
taskID := sendDirectChat(t, ctx, agentID, sessionID, "what next?")
markTaskRunning(t, ctx, taskID)
output := "Here is the plan.\n\n```quick-actions\n" +
`[{"label":"Draft it","prompt":"Draft the complete plan","primary":true},` +
`{"label":"Make a checklist","prompt":"Turn this into a checklist"}]` +
"\n```"
if _, err := testHandler.TaskService.CompleteTask(ctx, parseUUID(taskID), completeResult(t, output), "", "", false, ""); err != nil {
t.Fatalf("complete task: %v", err)
}
rows := assistantRows(t, ctx, sessionID)
if len(rows) != 1 {
t.Fatalf("expected one assistant message, got %d", len(rows))
}
if rows[0].Content != "Here is the plan." || rows[0].MessageKind != protocol.ChatMessageKindMessage {
t.Fatalf("persisted reply = kind %q content %q", rows[0].MessageKind, rows[0].Content)
}
// The footer's actions are DISCARDED, not persisted: suggestions come from
// the server-side pass now. Honouring an in-band footer would pin a
// pre-upgrade provider session to the retired, lower-quality suggestions and
// suppress the pass that replaces them.
var actions []protocol.ChatQuickAction
if err := json.Unmarshal(rows[0].QuickActions, &actions); err != nil {
t.Fatalf("decode quick actions: %v", err)
}
if len(actions) != 0 {
t.Fatalf("in-band footer actions must not be persisted, got %+v", actions)
}
// An actions-only turn (quick-actions footer with no visible text) must NOT
// become an empty-content message: older Desktop / mobile clients ignore
// quick_actions and would render an empty bubble. It falls through to the
// visible no_response fallback instead (MUL-4351).
actionsOnlyTask := sendDirectChat(t, ctx, agentID, sessionID, "give me options only")
markTaskRunning(t, ctx, actionsOnlyTask)
actionsOnly := "```quick-actions\n[{\"label\":\"Continue\",\"prompt\":\"Continue the plan\"}]\n```"
if _, err := testHandler.TaskService.CompleteTask(ctx, parseUUID(actionsOnlyTask), completeResult(t, actionsOnly), "", "", false, ""); err != nil {
t.Fatalf("complete actions-only task: %v", err)
}
rows = assistantRows(t, ctx, sessionID)
if len(rows) != 2 || rows[1].MessageKind != protocol.ChatMessageKindNoResponse {
t.Fatalf("actions-only outcome = %+v", rows)
}
if rows[1].Content == "" {
t.Fatal("actions-only no_response row must carry a non-empty fallback body for old clients")
}
var droppedActions []protocol.ChatQuickAction
if err := json.Unmarshal(rows[1].QuickActions, &droppedActions); err != nil {
t.Fatalf("decode quick actions: %v", err)
}
if len(droppedActions) != 0 {
t.Fatalf("actions-only no_response row must not carry quick actions, got %+v", droppedActions)
}
}
func TestCompleteTask_ChatQuickActionsSupplement(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
agentID, sessionID, _, _ := setupDirectChatSession(t, ctx, "suggest-pass chat")
taskID := sendDirectChat(t, ctx, agentID, sessionID, "what next?")
markTaskRunning(t, ctx, taskID)
output := "Main reply.\n\n```quick-actions\n" +
`[{"label":"From footer","prompt":"in-band fallback"}]` + "\n```"
req := TaskCompleteRequest{Output: output}
result, err := json.Marshal(req)
if err != nil {
t.Fatalf("marshal complete request: %v", err)
}
task, err := testHandler.TaskService.CompleteTask(ctx, parseUUID(taskID), result, "", "", false, "")
if err != nil {
t.Fatalf("complete task: %v", err)
}
rows := assistantRows(t, ctx, sessionID)
if len(rows) != 1 {
t.Fatalf("expected one assistant message, got %d", len(rows))
}
if rows[0].Content != "Main reply." {
t.Fatalf("footer must still be stripped from content, got %q", rows[0].Content)
}
raw := "```json\n" + `[{"label":"From pass","prompt":"suggested prompt","primary":true}]` + "\n```"
if err := testHandler.TaskService.SupplementChatQuickActions(ctx, *task, raw, false); err != nil {
t.Fatalf("supplement quick actions: %v", err)
}
rows = assistantRows(t, ctx, sessionID)
var actions []protocol.ChatQuickAction
if err := json.Unmarshal(rows[0].QuickActions, &actions); err != nil {
t.Fatalf("decode quick actions: %v", err)
}
if len(actions) != 1 || actions[0].Label != "From pass" || !actions[0].Primary {
t.Fatalf("supplement must attach the pass's actions, got %+v", actions)
}
// An empty supplement must not clobber existing actions (it only resolves
// the pending placeholder client-side).
if err := testHandler.TaskService.SupplementChatQuickActions(ctx, *task, "", false); err != nil {
t.Fatalf("empty supplement: %v", err)
}
rows = assistantRows(t, ctx, sessionID)
actions = nil
if err := json.Unmarshal(rows[0].QuickActions, &actions); err != nil {
t.Fatalf("decode quick actions after empty supplement: %v", err)
}
if len(actions) != 1 || actions[0].Label != "From pass" {
t.Fatalf("empty supplement must keep existing actions, got %+v", actions)
}
}
// stubChatQuickActionsLLM is an enabled ChatQuickActionsLLM that never dials an
// upstream, so the refresh gates below can be exercised on a deployment-shaped
// TaskService (RegenerateChatQuickActions refuses outright when the LLM layer
// is unconfigured).
type stubChatQuickActionsLLM struct {
// lastPrompt records the rendered conversation the pass was given, so a
// test can assert WHICH turn the suggestions were built from.
lastPrompt *string
}
func (stubChatQuickActionsLLM) Enabled() bool { return true }
func (s stubChatQuickActionsLLM) GenerateJSON(_ context.Context, _, _, userPrompt string, _ float64, _ int64) (string, error) {
if s.lastPrompt != nil {
*s.lastPrompt = userPrompt
}
return `{"actions":[{"label":"Next","prompt":"do the next thing","primary":true}]}`, nil
}
// TestChatQuickActions_ContextAnchorsOnTargetTurn pins the anchoring contract:
// generation runs on a detached goroutine, so by the time it executes the
// session may already carry a newer turn. The context it builds must end at the
// assistant turn the pills are written to — not at whatever is newest when the
// read happens.
//
// Without the anchor, a user who types a follow-up in the second after the
// reply lands leaves the window ending on a user row with no reply to build on,
// and that turn silently never gets pills.
func TestChatQuickActions_ContextAnchorsOnTargetTurn(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
agentID, sessionID, _, _ := setupDirectChatSession(t, ctx, "quick-actions anchor")
taskID := sendDirectChat(t, ctx, agentID, sessionID, "first question")
markTaskRunning(t, ctx, taskID)
task, err := testHandler.TaskService.CompleteTask(
ctx, parseUUID(taskID), completeResult(t, "ANCHOR REPLY"), "", "", false, "")
if err != nil {
t.Fatalf("complete turn 1: %v", err)
}
// The race: a newer user message lands before the pass reads the session.
sendDirectChat(t, ctx, agentID, sessionID, "NEWER USER MESSAGE")
var prompt string
prev := testHandler.TaskService.QuickActions
testHandler.TaskService.QuickActions = stubChatQuickActionsLLM{lastPrompt: &prompt}
defer func() { testHandler.TaskService.QuickActions = prev }()
if err := testHandler.TaskService.GenerateChatQuickActionsForTask(
ctx, *task, service.ChatQuickActionsAutomatic); err != nil {
t.Fatalf("generate quick actions: %v", err)
}
if !strings.Contains(prompt, "ANCHOR REPLY") {
t.Fatalf("context must end at the target reply, got:\n%s", prompt)
}
if strings.Contains(prompt, "NEWER USER MESSAGE") {
t.Fatalf("context must exclude turns newer than the target, got:\n%s", prompt)
}
// And the pills land on the target turn, not the newer one.
rows := assistantRows(t, ctx, sessionID)
if len(rows) != 1 {
t.Fatalf("expected one assistant row, got %d", len(rows))
}
var actions []protocol.ChatQuickAction
if err := json.Unmarshal(rows[0].QuickActions, &actions); err != nil {
t.Fatalf("decode quick actions: %v", err)
}
if len(actions) != 1 || actions[0].Label != "Next" {
t.Fatalf("target turn quick actions = %+v", actions)
}
}
// installStubQuickActions enables suggestion generation and returns the undo.
// Scoped tightly around the synchronous RegenerateChatQuickActions calls rather
// than installed for a whole test: CompleteTask starts a background pass when
// the feature is on, and a goroutine writing quick_actions mid-test would race
// these assertions.
func installStubQuickActions() func() {
prev := testHandler.TaskService.QuickActions
testHandler.TaskService.QuickActions = stubChatQuickActionsLLM{}
return func() { testHandler.TaskService.QuickActions = prev }
}
// TestRegenerateChatQuickActions_StaleTargetRejected pins the ack-alignment
// contract (MUL-5149 review): a refresh names the turn it is refreshing, and
// the server enqueues it only while that turn is STILL the session's latest.
// Once a newer reply lands, refreshing the older turn is refused, so the
// client's pending marker never points at a turn the resulting
// chat:quick_actions event could not match.
func TestRegenerateChatQuickActions_StaleTargetRejected(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
agentID, sessionID, _, _ := setupDirectChatSession(t, ctx, "regen stale chat")
// Turn 1: a resumable assistant reply (session_id + runtime bound) is latest.
t1 := sendDirectChat(t, ctx, agentID, sessionID, "first question")
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
// Refreshing the current latest turn is accepted and targets exactly it.
restore := installStubQuickActions()
target, _, err := testHandler.TaskService.RegenerateChatQuickActions(ctx, session, m1)
restore()
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)
}
// Turn 2 lands, so m1 is no longer the latest.
t2 := sendDirectChat(t, ctx, agentID, sessionID, "second question")
markTaskRunning(t, ctx, t2)
if _, err := testHandler.TaskService.CompleteTask(ctx, parseUUID(t2), completeResult(t, "second reply"), "sess-2", "/tmp/wd", false, ""); err != nil {
t.Fatalf("complete turn 2: %v", err)
}
session, err = testHandler.Queries.GetChatSession(ctx, parseUUID(sessionID))
if err != nil {
t.Fatalf("reload session after turn 2: %v", err)
}
rows = assistantRows(t, ctx, sessionID)
if len(rows) != 2 {
t.Fatalf("expected two assistant turns, got %d", len(rows))
}
m2 := rows[1].ID
// Refreshing the now-stale m1 is refused; refreshing the new latest m2 works.
restore = installStubQuickActions()
defer restore()
if _, _, err := testHandler.TaskService.RegenerateChatQuickActions(ctx, session, m1); !errors.Is(err, service.ErrChatQuickActionsStale) {
t.Fatalf("stale-target regenerate error = %v, want ErrChatQuickActionsStale", err)
}
if _, _, err := testHandler.TaskService.RegenerateChatQuickActions(ctx, session, m2); err != nil {
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)
restore := installStubQuickActions()
defer restore()
if _, _, err := testHandler.TaskService.RegenerateChatQuickActions(ctx, session, m1); !errors.Is(err, service.ErrChatQuickActionsBusy) {
t.Fatalf("busy-session regenerate error = %v, want ErrChatQuickActionsBusy", err)
}
}
// TestRegenerateChatQuickActions_DeferredActiveTurnRejected pins the re-review
// §1 gap: a chat auto-retry armed with a backoff fire_at is inserted 'deferred'
// (CreateRetryTask; provider_network's final attempt waits ~5s that way). During
// that window the failed turn has written no assistant row, so the old turn is
// still latest-persisted and its pills stay clickable — yet the session is about
// to advance when the retry fires. HasActiveChatTaskForSession must count
// 'deferred', else the refresh resumes a session the retry moves past and pins
// the new turn's suggestions onto the old one.
func TestRegenerateChatQuickActions_DeferredActiveTurnRejected(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
ctx := context.Background()
agentID, sessionID, _, _ := setupDirectChatSession(t, ctx, "regen deferred chat")
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 failed and is waiting out its retry backoff: a deferred task with a
// future fire_at and no assistant row yet, so m1 is still latest-persisted.
t2 := sendDirectChat(t, ctx, agentID, sessionID, "second")
if _, err := testPool.Exec(ctx,
`UPDATE agent_task_queue SET status='deferred', fire_at=now() + interval '5 seconds' WHERE id=$1`,
t2); err != nil {
t.Fatalf("defer turn 2: %v", err)
}
restore := installStubQuickActions()
defer restore()
if _, _, err := testHandler.TaskService.RegenerateChatQuickActions(ctx, session, m1); !errors.Is(err, service.ErrChatQuickActionsBusy) {
t.Fatalf("deferred-active regenerate error = %v, want ErrChatQuickActionsBusy", err)
}
}