diff --git a/server/cmd/server/comment_at_least_once_test.go b/server/cmd/server/comment_at_least_once_test.go index 8683d17b5e..680ea3edff 100644 --- a/server/cmd/server/comment_at_least_once_test.go +++ b/server/cmd/server/comment_at_least_once_test.go @@ -308,3 +308,103 @@ func taskOriginator(t *testing.T, issueID, agentID string) string { } return originator } + +// TestMergeCommentIntoPendingTask_TargetsQueuedNotDeferred is the MUL-4195 +// round-4 regression test. When a `(issue, agent)` pair has BOTH an older +// queued task (the run about to be claimed) and a newer deferred +// assignee-fallback task, a new comment's merge must land on the QUEUED task — +// the one that will actually run next — not the newer deferred fallback. An +// earlier `status IN ('queued','deferred') ORDER BY created_at DESC` target +// picked the deferred row, so the comment missed the imminent run and the +// deferred fallback could later promote into a duplicate. The merge now matches +// `status = 'queued'` only; the deferred row is left to its own escalation +// lifecycle. +func TestMergeCommentIntoPendingTask_TargetsQueuedNotDeferred(t *testing.T) { + if testPool == nil { + t.Skip("no database connection") + } + ctx := context.Background() + queries := db.New(testPool) + + agentID := getAgentID(t) + issueID := createIssueAssignedToAgent(t, "Merge target queued-vs-deferred test", agentID) + clearTasks(t, issueID) + t.Cleanup(func() { + clearTasks(t, issueID) + resp := authRequest(t, "DELETE", "/api/issues/"+issueID, nil) + resp.Body.Close() + }) + + now := time.Now() + cidQueued := insertCommentAt(t, issueID, "member", testUserID, "queued task trigger", now.Add(-3*time.Minute)) + cidDeferred := insertCommentAt(t, issueID, "member", testUserID, "deferred fallback trigger", now.Add(-2*time.Minute)) + cidNew := insertCommentAt(t, issueID, "member", testUserID, "new comment, must fold into queued", now.Add(-1*time.Minute)) + + var runtimeID string + if err := testPool.QueryRow(ctx, `SELECT runtime_id FROM agent WHERE id = $1`, agentID).Scan(&runtimeID); err != nil { + t.Fatalf("load runtime: %v", err) + } + + // Older queued task (the imminent run) ... + var queuedTaskID string + if err := testPool.QueryRow(ctx, ` + INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, trigger_comment_id, status, priority, created_at) + VALUES ($1, $2, $3, $4, 'queued', 0, now() - interval '3 minutes') + RETURNING id + `, agentID, runtimeID, issueID, cidQueued).Scan(&queuedTaskID); err != nil { + t.Fatalf("seed queued task: %v", err) + } + // ... and a NEWER deferred assignee-fallback task for the same (issue, agent). + var deferredTaskID string + if err := testPool.QueryRow(ctx, ` + INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, trigger_comment_id, status, priority, created_at, fire_at) + VALUES ($1, $2, $3, $4, 'deferred', 0, now() - interval '2 minutes', now() + interval '5 minutes') + RETURNING id + `, agentID, runtimeID, issueID, cidDeferred).Scan(&deferredTaskID); err != nil { + t.Fatalf("seed deferred task: %v", err) + } + + row, err := queries.MergeCommentIntoPendingTask(ctx, db.MergeCommentIntoPendingTaskParams{ + IssueID: toPgUUID(t, issueID), + AgentID: toPgUUID(t, agentID), + NewTriggerCommentID: toPgUUID(t, cidNew), + NewOriginatorUserID: toPgUUID(t, testUserID), + NewTriggerSummary: pgtype.Text{String: "new comment", Valid: true}, + }) + if err != nil { + t.Fatalf("merge should target the queued task, got %v", err) + } + // The merge must have hit the QUEUED task, not the deferred one. + if got := pgUUIDToText(row.ID); got != queuedTaskID { + t.Fatalf("merge must target the queued task %s, got %s", queuedTaskID, got) + } + + // Queued task: trigger repointed to the new comment, old trigger coalesced. + var queuedTrigger string + var queuedCoalesced []string + if err := testPool.QueryRow(ctx, ` + SELECT trigger_comment_id::text, coalesced_comment_ids::text[] FROM agent_task_queue WHERE id = $1 + `, queuedTaskID).Scan(&queuedTrigger, &queuedCoalesced); err != nil { + t.Fatalf("read queued task: %v", err) + } + if queuedTrigger != cidNew { + t.Errorf("queued task trigger must be repointed to %s, got %s", cidNew, queuedTrigger) + } + if !containsID(queuedCoalesced, cidQueued) { + t.Errorf("queued task must coalesce its old trigger %s, got %v", cidQueued, queuedCoalesced) + } + + // Deferred fallback must be UNTOUCHED (still its own trigger, still deferred). + var deferredTrigger, deferredStatus string + if err := testPool.QueryRow(ctx, ` + SELECT trigger_comment_id::text, status FROM agent_task_queue WHERE id = $1 + `, deferredTaskID).Scan(&deferredTrigger, &deferredStatus); err != nil { + t.Fatalf("read deferred task: %v", err) + } + if deferredTrigger != cidDeferred { + t.Errorf("deferred fallback trigger must be untouched (%s), got %s", cidDeferred, deferredTrigger) + } + if deferredStatus != "deferred" { + t.Errorf("deferred fallback status must stay 'deferred', got %s", deferredStatus) + } +} diff --git a/server/internal/handler/comment.go b/server/internal/handler/comment.go index ecfc764913..10534a1980 100644 --- a/server/internal/handler/comment.go +++ b/server/internal/handler/comment.go @@ -1399,16 +1399,16 @@ func (h *Handler) enqueueCommentAgentTriggers(ctx context.Context, issue db.Issu if trigger.AlreadyPending { // MUL-4195: a queued/dispatched task for this (issue, agent) // already exists. Historically we DROPPED the comment here, losing - // the user's follow-up instruction. Instead try to fold it into a - // not-yet-claimed (queued/deferred) task so a single run still - // covers every comment, re-stamping the run's originator/overlay to - // the new comment (mergeCommentIntoPendingTask). + // the user's follow-up instruction. Instead try to fold it into the + // queued (not-yet-claimed) task so a single run still covers every + // comment, re-stamping the run's originator/overlay to the new + // comment (mergeCommentIntoPendingTask). if h.mergeCommentIntoPendingTask(ctx, issue, trigger, triggerCommentID) { continue } - // The merge found no pre-claim task to fold into: the existing task - // is already dispatched/running (its claim response is built), or a - // mismatched pre-claim row was just claimed. We must NOT enqueue a + // The merge found no queued task to fold into: the existing task + // is already dispatched/running (its claim response is built), or + // the queued row was just claimed. We must NOT enqueue a // fresh queued task in that case — a dispatched sibling would trip // the idx_one_pending_task_per_issue_agent unique index (dropping // the comment again) and even where the index allows it we'd risk a @@ -1453,11 +1453,11 @@ func (h *Handler) hasActiveTaskForIssueAndAgent(ctx context.Context, issueID, ag // fresh task and the deliberate comment is never lost. // // The merge is GATED on the originator being unchanged (MUL-4195 review -// mergeCommentIntoPendingTask folds a newly-arrived comment into an existing -// NOT-YET-CLAIMED (queued/deferred) task for (issue, agent) instead of dropping -// it (MUL-4195). Returns true when the comment was handled (merged, or a +// mergeCommentIntoPendingTask folds a newly-arrived comment into the existing +// QUEUED (not-yet-claimed) task for (issue, agent) instead of dropping it +// (MUL-4195). Returns true when the comment was handled (merged, or a // non-fatal DB error we deliberately do not turn into a duplicate). Returns -// false only when no pre-claim task exists to merge into (pgx.ErrNoRows) — the +// false only when no queued task exists to merge into (pgx.ErrNoRows) — the // existing task is already dispatched/running, or was just claimed — in which // case the caller decides between deferring to completion reconcile and a fresh // enqueue. diff --git a/server/pkg/db/generated/agent.sql.go b/server/pkg/db/generated/agent.sql.go index 8e9f2ac9ab..ee41864fb6 100644 --- a/server/pkg/db/generated/agent.sql.go +++ b/server/pkg/db/generated/agent.sql.go @@ -3194,7 +3194,7 @@ WHERE id = ( SELECT t.id FROM agent_task_queue t WHERE t.issue_id = $6 AND t.agent_id = $7 - AND t.status IN ('queued', 'deferred') + AND t.status = 'queued' ORDER BY t.created_at DESC LIMIT 1 ) @@ -3224,16 +3224,22 @@ type MergeCommentIntoPendingTaskRow struct { // the latest deliberate instruction while the single run is still told to // address every folded comment. // -// Target is restricted to PRE-CLAIM states — 'queued' and 'deferred' — on -// purpose (MUL-4195 review must-fix #2). A 'dispatched' / 'waiting_local_directory' -// / 'running' task has already had its claim response (with its -// coalesced_comment_ids) built and shipped to the daemon; folding a comment in -// after that point would add it to coalesced_comment_ids WITHOUT the daemon ever -// seeing it, making an undelivered comment look delivered. Those post-claim -// comments are handled by completion reconciliation instead, which anchors on -// dispatched_at and sweeps every member comment the run did not deliver. Because -// merges only ever touch pre-claim rows, coalesced_comment_ids == the set the -// run actually received. +// Target is restricted to the single 'queued' task on purpose (MUL-4195 review +// rounds 2–4). This merge is only reached when HasPendingTaskForIssueAndAgent +// matched a 'queued'/'dispatched' task, and 'dispatched' is deliberately NOT a +// target: a dispatched / waiting_local_directory / running task has already had +// its claim response (with its coalesced_comment_ids) built and shipped, so +// folding a comment in afterward would mark an undelivered comment as delivered; +// those are handled by completion reconciliation instead. 'deferred' is also NOT +// a target: a deferred row is an assignee-fallback escalation with its own +// fire_at/promotion lifecycle, and it never sets AlreadyPending +// (HasPendingTaskForIssueAndAgent only looks at queued/dispatched). If a newer +// deferred fallback and an older queued task coexisted, a status-IN target would +// pick the deferred one by created_at and steal the coalescing target away from +// the queued run that is actually about to be claimed — so we match ONLY the +// queued row (the idx_one_pending_task_per_issue_agent unique index guarantees +// at most one). Because merges only ever touch that pre-claim queued row, +// coalesced_comment_ids == the set the run actually received. // // Recompute-on-merge (MUL-4195 review must-fix #1): originator_user_id, // runtime_mcp_overlay and runtime_connected_apps are re-stamped to the NEW @@ -3248,7 +3254,7 @@ type MergeCommentIntoPendingTaskRow struct { // index allows only one queued/dispatched task per (issue, agent)) and therefore // silently dropped the mismatched-originator comment. // -// Returns pgx.ErrNoRows when no pre-claim task exists (it was claimed/started +// Returns pgx.ErrNoRows when no queued task exists (it was claimed/started // between the dedup check and this call, or the only task is already // dispatched/running). The caller must NOT blindly enqueue a fresh task in that // case — a dispatched sibling would trip the unique index — it defers to diff --git a/server/pkg/db/queries/agent.sql b/server/pkg/db/queries/agent.sql index b1a0615196..d37faee61b 100644 --- a/server/pkg/db/queries/agent.sql +++ b/server/pkg/db/queries/agent.sql @@ -739,16 +739,22 @@ WHERE issue_id = @issue_id -- the latest deliberate instruction while the single run is still told to -- address every folded comment. -- --- Target is restricted to PRE-CLAIM states — 'queued' and 'deferred' — on --- purpose (MUL-4195 review must-fix #2). A 'dispatched' / 'waiting_local_directory' --- / 'running' task has already had its claim response (with its --- coalesced_comment_ids) built and shipped to the daemon; folding a comment in --- after that point would add it to coalesced_comment_ids WITHOUT the daemon ever --- seeing it, making an undelivered comment look delivered. Those post-claim --- comments are handled by completion reconciliation instead, which anchors on --- dispatched_at and sweeps every member comment the run did not deliver. Because --- merges only ever touch pre-claim rows, coalesced_comment_ids == the set the --- run actually received. +-- Target is restricted to the single 'queued' task on purpose (MUL-4195 review +-- rounds 2–4). This merge is only reached when HasPendingTaskForIssueAndAgent +-- matched a 'queued'/'dispatched' task, and 'dispatched' is deliberately NOT a +-- target: a dispatched / waiting_local_directory / running task has already had +-- its claim response (with its coalesced_comment_ids) built and shipped, so +-- folding a comment in afterward would mark an undelivered comment as delivered; +-- those are handled by completion reconciliation instead. 'deferred' is also NOT +-- a target: a deferred row is an assignee-fallback escalation with its own +-- fire_at/promotion lifecycle, and it never sets AlreadyPending +-- (HasPendingTaskForIssueAndAgent only looks at queued/dispatched). If a newer +-- deferred fallback and an older queued task coexisted, a status-IN target would +-- pick the deferred one by created_at and steal the coalescing target away from +-- the queued run that is actually about to be claimed — so we match ONLY the +-- queued row (the idx_one_pending_task_per_issue_agent unique index guarantees +-- at most one). Because merges only ever touch that pre-claim queued row, +-- coalesced_comment_ids == the set the run actually received. -- -- Recompute-on-merge (MUL-4195 review must-fix #1): originator_user_id, -- runtime_mcp_overlay and runtime_connected_apps are re-stamped to the NEW @@ -763,7 +769,7 @@ WHERE issue_id = @issue_id -- index allows only one queued/dispatched task per (issue, agent)) and therefore -- silently dropped the mismatched-originator comment. -- --- Returns pgx.ErrNoRows when no pre-claim task exists (it was claimed/started +-- Returns pgx.ErrNoRows when no queued task exists (it was claimed/started -- between the dedup check and this call, or the only task is already -- dispatched/running). The caller must NOT blindly enqueue a fresh task in that -- case — a dispatched sibling would trip the unique index — it defers to @@ -783,7 +789,7 @@ WHERE id = ( SELECT t.id FROM agent_task_queue t WHERE t.issue_id = @issue_id AND t.agent_id = @agent_id - AND t.status IN ('queued', 'deferred') + AND t.status = 'queued' ORDER BY t.created_at DESC LIMIT 1 )