multica/server/internal/handler/comment_duplicate_enqueue_race_test.go

package handler

import (
	"bytes"
	"context"
	"errors"
	"log/slog"
	"strings"
	"testing"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgtype"
	"github.com/multica-ai/multica/server/internal/service"
	"github.com/multica-ai/multica/server/internal/testutil"
	"github.com/multica-ai/multica/server/internal/util"
	db "github.com/multica-ai/multica/server/pkg/db/generated"
)

const (
	dupRaceHeadA = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
	dupRaceHeadB = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
)

// dupRaceFixture creates a workspace-invocable agent and an issue assigned to it.
func dupRaceFixture(t *testing.T, agentName string, issueNumber int) (agentID, issueID, runtimeID string) {
	t.Helper()
	ctx := context.Background()
	agentID = createHandlerTestAgent(t, agentName, nil)
	if err := testPool.QueryRow(ctx, `SELECT runtime_id FROM agent WHERE id = $1`, agentID).Scan(&runtimeID); err != nil {
		t.Fatalf("load runtime: %v", err)
	}
	if err := testPool.QueryRow(ctx, `
		INSERT INTO issue (workspace_id, title, status, priority, creator_id, creator_type, number, position, assignee_type, assignee_id)
		VALUES ($1, 'dup-enqueue-race fixture', 'in_progress', 'none', $2, 'member', $3, 0, 'agent', $4)
		RETURNING id
	`, testWorkspaceID, testUserID, issueNumber, agentID).Scan(&issueID); err != nil {
		t.Fatalf("create issue: %v", err)
	}
	t.Cleanup(func() {
		testPool.Exec(context.Background(), `DELETE FROM agent_task_queue WHERE issue_id = $1`, issueID)
	})
	t.Cleanup(func() { testPool.Exec(context.Background(), `DELETE FROM comment WHERE issue_id = $1`, issueID) })
	t.Cleanup(func() { testPool.Exec(context.Background(), `DELETE FROM issue WHERE id = $1`, issueID) })
	return agentID, issueID, runtimeID
}

func insertDupRaceComment(t *testing.T, issueID, content, age string) string {
	t.Helper()
	// These race cases compete for the same thread slot.
	var parentID *string
	dbfx.QueryRow(t, `SELECT (SELECT id::text FROM comment WHERE issue_id=$1 AND parent_id IS NULL ORDER BY created_at, id LIMIT 1)`, issueID).Scan(&parentID)
	id := dbfx.Comment(t, issueID, content, testutil.Cols{"parent_id": parentID})
	dbfx.Exec(t, `UPDATE comment SET created_at=now()-$2::interval WHERE id=$1`, id, age)
	return id
}

func commentCovered(t *testing.T, issueID, agentID, commentID, statusFilter string) bool {
	t.Helper()
	var ok bool
	if err := testPool.QueryRow(context.Background(), `
		SELECT $3::uuid = trigger_comment_id OR $3::uuid = ANY(coalesced_comment_ids)
		FROM agent_task_queue
		WHERE issue_id = $1 AND agent_id = $2 AND status = $4
	`, issueID, agentID, commentID, statusFilter).Scan(&ok); err != nil {
		t.Fatalf("check covered %s: %v", commentID, err)
	}
	return ok
}

// TestCommentEnqueueRaceQueuedWinnerFoldsLoser: when the lost-race winner is
// still QUEUED and shares the reviewed head, the losing comment is folded into
// it (the merge makes the newer comment the trigger and pushes the prior trigger
// into coalesced), so the single run covers both. Coalesced outcome, one pending
// task, and no warning / constraint-name leak.
//
// Run once per trigger source that can lose this race on an issue whose agent is
// the assignee. The mention source has been covered since #5958; the
// ISSUE-ASSIGNEE source is the half #5914 left behind, and it is not redundant:
// its enqueue surfaced the RAW unique violation, so the loser came back
// blocked/internal_error instead of coalescing. That dropped the comment's
// instruction outright — the comment-creation path has no obligation hand-off
// (unlike completion reconcile), and a losing comment usually PREDATES the
// winning task row, so completion reconcile's created_at window cannot pick it
// up either.
func TestCommentEnqueueRaceQueuedWinnerFoldsLoser(t *testing.T) {
	if testHandler == nil {
		t.Skip("database not available")
	}
	cases := []struct {
		name          string
		agentName     string
		issueNumber   int
		source        commentAgentTriggerSource
		enqueueWinner func(ctx context.Context, issue db.Issue, agentID, commentID pgtype.UUID) error
	}{
		{
			name:        "mention",
			agentName:   "dup-race-queued",
			issueNumber: 999311,
			source:      commentTriggerSourceMentionAgent,
			enqueueWinner: func(ctx context.Context, issue db.Issue, agentID, commentID pgtype.UUID) error {
				_, err := testHandler.TaskService.EnqueueTaskForMention(ctx, issue, agentID, commentID, service.OriginNamed)
				return err
			},
		},
		{
			name:        "issue_assignee",
			agentName:   "dup-race-assignee",
			issueNumber: 999316,
			source:      commentTriggerSourceIssueAssignee,
			enqueueWinner: func(ctx context.Context, issue db.Issue, _, commentID pgtype.UUID) error {
				_, err := testHandler.TaskService.EnqueueTaskForIssue(ctx, issue, commentID)
				return err
			},
		},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			ctx := context.Background()
			agentID, issueID, _ := dupRaceFixture(t, tc.agentName, tc.issueNumber)
			agentUUID := util.MustParseUUID(agentID)
			issue, err := testHandler.Queries.GetIssue(ctx, util.MustParseUUID(issueID))
			if err != nil {
				t.Fatalf("load issue: %v", err)
			}
			agent, err := testHandler.Queries.GetAgent(ctx, agentUUID)
			if err != nil {
				t.Fatalf("load agent: %v", err)
			}

			winnerCommentID := insertDupRaceComment(t, issueID, "first instruction", "6 minutes")
			if err := tc.enqueueWinner(ctx, issue, agentUUID, util.MustParseUUID(winnerCommentID)); err != nil {
				t.Fatalf("enqueue winning task: %v", err)
			}
			loserCommentID := insertDupRaceComment(t, issueID, "second distinct instruction", "1 minute")

			var logs bytes.Buffer
			prev := slog.Default()
			slog.SetDefault(slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelInfo})))
			t.Cleanup(func() { slog.SetDefault(prev) })

			trigger := commentAgentTrigger{Agent: agent, Source: tc.source}
			results := testHandler.enqueueCommentAgentTriggers(ctx, issue, util.MustParseUUID(loserCommentID), []commentAgentTrigger{trigger})

			// Guards the service hunk: the lost race must coalesce, not block.
			if res := results[agentID]; res.status != DispatchCoalesced {
				t.Fatalf("queued-winner race: got status %q reason %q, want coalesced", res.status, res.reason)
			}
			if !commentCovered(t, issueID, agentID, loserCommentID, "queued") {
				t.Fatal("losing comment was NOT folded into the queued winner — its instruction would be dropped")
			}
			if !commentCovered(t, issueID, agentID, winnerCommentID, "queued") {
				t.Fatal("winner comment is no longer covered after the fold")
			}
			if n := pendingTaskCountForAgentIssue(t, issueID, agentID); n != 1 {
				t.Fatalf("pending task count = %d, want exactly 1", n)
			}
			// Guards the handler hunk: a benign race must not surface as WARN/ERROR.
			for _, leak := range []string{"idx_one_pending_task_per_issue_agent", "level=WARN", "level=ERROR"} {
				if strings.Contains(logs.String(), leak) {
					t.Fatalf("benign enqueue race leaked %q into logs:\n%s", leak, logs.String())
				}
			}
		})
	}
}

// TestCommentEnqueueRaceDispatchedWinnerDurablyCoversLoser is the regression for
// Elon round-2 must-fix 1: the losing comment PRECEDES a winner that is already
// DISPATCHED before the merge retry, so completion reconcile's created_at window
// cannot see it. The race path must durably register it as a planned (but NOT
// delivered) input on the active winner, and a follow-up must actually cover it
// after the winner completes — never a bare coalesced success that drops it.
func TestCommentEnqueueRaceDispatchedWinnerDurablyCoversLoser(t *testing.T) {
	if testHandler == nil {
		t.Skip("database not available")
	}
	ctx := context.Background()
	agentID, issueID, runtimeID := dupRaceFixture(t, "dup-race-dispatched", 999312)
	agentUUID := util.MustParseUUID(agentID)
	issue, err := testHandler.Queries.GetIssue(ctx, util.MustParseUUID(issueID))
	if err != nil {
		t.Fatalf("load issue: %v", err)
	}
	agent, err := testHandler.Queries.GetAgent(ctx, agentUUID)
	if err != nil {
		t.Fatalf("load agent: %v", err)
	}

	// Losing comment PREDATES the winner task; winner is already dispatched with
	// its claim receipt (delivered = its own trigger only).
	loserCommentID := insertDupRaceComment(t, issueID, "losing instruction (predates winner)", "10 minutes")
	winnerCommentID := insertDupRaceComment(t, issueID, "winning instruction", "6 minutes")
	var winnerTaskID string
	if err := testPool.QueryRow(ctx, `
		INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, trigger_comment_id, delivered_comment_ids, status, priority, created_at, dispatched_at)
		VALUES ($1, $2, $3, $4, ARRAY[$4::uuid], 'dispatched', 0, now() - interval '5 minutes', now() - interval '4 minutes')
		RETURNING id
	`, agentID, runtimeID, issueID, winnerCommentID).Scan(&winnerTaskID); err != nil {
		t.Fatalf("insert dispatched winner: %v", err)
	}

	trigger := commentAgentTrigger{Agent: agent, Source: commentTriggerSourceMentionAgent}
	results := testHandler.enqueueCommentAgentTriggers(ctx, issue, util.MustParseUUID(loserCommentID), []commentAgentTrigger{trigger})

	// Truthful outcome: deferred (a follow-up will cover it), NOT coalesced.
	if res := results[agentID]; res.status != DispatchDeferred {
		t.Fatalf("dispatched-winner race: got status %q reason %q, want deferred", res.status, res.reason)
	}
	// Durably planned on the winner, but NOT faked as delivered.
	var plannedHasLoser, deliveredHasLoser bool
	if err := testPool.QueryRow(ctx, `
		SELECT $2::uuid = ANY(coalesced_comment_ids), $2::uuid = ANY(delivered_comment_ids)
		FROM agent_task_queue WHERE id = $1
	`, winnerTaskID, loserCommentID).Scan(&plannedHasLoser, &deliveredHasLoser); err != nil {
		t.Fatalf("read winner planned/delivered: %v", err)
	}
	if !plannedHasLoser {
		t.Fatal("losing comment was NOT registered as a planned input on the dispatched winner")
	}
	if deliveredHasLoser {
		t.Fatal("losing comment must NOT be faked as delivered on the dispatched winner")
	}
	if n := pendingTaskCountForAgentIssue(t, issueID, agentID); n != 1 {
		t.Fatalf("pending task count = %d, want exactly 1 (the dispatched winner)", n)
	}

	// The winner completes; reconcile must replay the planned-but-undelivered
	// loser as a bounded follow-up — proving the drop is actually closed.
	if _, err := testPool.Exec(ctx, `UPDATE agent_task_queue SET status = 'running', started_at = now() - interval '1 minute' WHERE id = $1`, winnerTaskID); err != nil {
		t.Fatalf("advance winner to running: %v", err)
	}
	if w := completeTaskViaHandler(t, winnerTaskID, "done"); w.Code != 200 {
		t.Fatalf("complete winner: got %d: %s", w.Code, w.Body.String())
	}
	if n := queuedTaskCountForAgentIssue(t, issueID, agentID); n != 1 {
		t.Fatalf("expected exactly 1 follow-up covering the loser after completion, got %d", n)
	}
	if !commentCovered(t, issueID, agentID, loserCommentID, "queued") {
		t.Fatal("follow-up does not cover the losing comment — the instruction was dropped")
	}
}

// TestCommentEnqueueRaceDifferentHeadNotCoalesced is the regression for Elon
// round-2 must-fix 2 + round-7: a comment for a NEW head (PR updated A→B) collides
// with the still-pending head-A task. It must NOT be folded into the head-A run
// (TEN-356), and — per round 7 — it must NOT be reported as a fabricated deferred
// either, since a point-in-time "an older task exists" cannot durably promise
// coverage. The honest outcome is a non-success; the head-A task's completion
// reconcile still covers the head-B comment best-effort, earning head-B coverage.
func TestCommentEnqueueRaceDifferentHeadNotCoalesced(t *testing.T) {
	if testHandler == nil {
		t.Skip("database not available")
	}
	ctx := context.Background()
	agentID, issueID, runtimeID := dupRaceFixture(t, "dup-race-headshift", 999313)
	agentUUID := util.MustParseUUID(agentID)

	// Link a PR at head B so ResolveIssueReviewSHA returns B for new enqueues.
	var prID string
	if err := testPool.QueryRow(ctx, `
		INSERT INTO github_pull_request (workspace_id, installation_id, repo_owner, repo_name, pr_number, title, state, html_url, pr_created_at, pr_updated_at, head_sha)
		VALUES ($1, 1, 'multica-ai', 'multica', 999313, 'review PR', 'open', 'https://example.test/pr', now(), now(), $2)
		RETURNING id
	`, testWorkspaceID, dupRaceHeadB).Scan(&prID); err != nil {
		t.Fatalf("seed PR: %v", err)
	}
	t.Cleanup(func() {
		testPool.Exec(context.Background(), `DELETE FROM issue_pull_request WHERE pull_request_id = $1`, prID)
	})
	t.Cleanup(func() { testPool.Exec(context.Background(), `DELETE FROM github_pull_request WHERE id = $1`, prID) })
	if _, err := testPool.Exec(ctx, `INSERT INTO issue_pull_request (issue_id, pull_request_id) VALUES ($1, $2)`, issueID, prID); err != nil {
		t.Fatalf("link PR: %v", err)
	}

	issue, err := testHandler.Queries.GetIssue(ctx, util.MustParseUUID(issueID))
	if err != nil {
		t.Fatalf("load issue: %v", err)
	}
	agent, err := testHandler.Queries.GetAgent(ctx, agentUUID)
	if err != nil {
		t.Fatalf("load agent: %v", err)
	}

	// Pending winner stamped for the OLD head A (queued), and a NEWER head-B comment.
	headAcommentID := insertDupRaceComment(t, issueID, "review head A", "6 minutes")
	var winnerTaskID string
	if err := testPool.QueryRow(ctx, `
		INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, trigger_comment_id, delivered_comment_ids, status, priority, created_at, context)
		VALUES ($1, $2, $3, $4, ARRAY[$4::uuid], 'queued', 0, now() - interval '5 minutes', jsonb_build_object('head_sha', $5::text))
		RETURNING id
	`, agentID, runtimeID, issueID, headAcommentID, dupRaceHeadA).Scan(&winnerTaskID); err != nil {
		t.Fatalf("insert head-A winner: %v", err)
	}
	headBcommentID := insertDupRaceComment(t, issueID, "please review head B", "1 minute")

	trigger := commentAgentTrigger{Agent: agent, Source: commentTriggerSourceMentionAgent}
	results := testHandler.enqueueCommentAgentTriggers(ctx, issue, util.MustParseUUID(headBcommentID), []commentAgentTrigger{trigger})

	// Not coalesced into the head-A run, and NOT a fabricated deferred either — a
	// snapshot cannot durably promise the head-A task's reconcile will still cover
	// this comment (Elon round 7). The honest outcome is a non-success; the head-A
	// reconcile still covers it best-effort below.
	if res := results[agentID]; res.status == DispatchDeferred || res.status == DispatchCoalesced || res.status == DispatchQueued {
		t.Fatalf("different-head race: got success-shaped %q/%q, want a truthful non-success", res.status, res.reason)
	}
	if res := results[agentID]; res.status != DispatchBlocked || res.reason != ReasonInternalError {
		t.Fatalf("different-head race: got %q/%q, want blocked/internal_error", res.status, res.reason)
	}
	// The head-A task is untouched: the head-B comment was NOT folded in, and the
	// task still carries head A.
	var headBinA bool
	var winnerHead string
	if err := testPool.QueryRow(ctx, `
		SELECT $2::uuid = ANY(coalesced_comment_ids), COALESCE(context->>'head_sha','')
		FROM agent_task_queue WHERE id = $1
	`, winnerTaskID, headBcommentID).Scan(&headBinA, &winnerHead); err != nil {
		t.Fatalf("read head-A winner: %v", err)
	}
	if headBinA {
		t.Fatal("head-B comment was wrongly folded into the head-A run (TEN-356 violation)")
	}
	if winnerHead != dupRaceHeadA {
		t.Fatalf("head-A winner head_sha changed to %q, want unchanged %q", winnerHead, dupRaceHeadA)
	}

	// After the head-A task completes, reconcile enqueues fresh coverage for the
	// newer head-B comment — and that follow-up is stamped for head B.
	if _, err := testPool.Exec(ctx, `UPDATE agent_task_queue SET status = 'running', started_at = now() - interval '1 minute' WHERE id = $1`, winnerTaskID); err != nil {
		t.Fatalf("advance head-A winner to running: %v", err)
	}
	if w := completeTaskViaHandler(t, winnerTaskID, "done"); w.Code != 200 {
		t.Fatalf("complete head-A winner: got %d: %s", w.Code, w.Body.String())
	}
	var followupHead string
	if err := testPool.QueryRow(ctx, `
		SELECT COALESCE(context->>'head_sha','')
		FROM agent_task_queue
		WHERE issue_id = $1 AND agent_id = $2 AND status = 'queued'
		ORDER BY created_at DESC LIMIT 1
	`, issueID, agentID).Scan(&followupHead); err != nil {
		t.Fatalf("read follow-up head: %v", err)
	}
	if followupHead != dupRaceHeadB {
		t.Fatalf("follow-up head_sha = %q, want head B %q (head B must earn its own coverage)", followupHead, dupRaceHeadB)
	}
}

// TestRegisterPlannedCommentForActiveTaskExcludesQueued is the regression for
// Elon round-3 must-fix 2: a planned-only append must never target a QUEUED task
// (it has no claim receipt, so the append would be delivered at claim time and
// bypass the atomic re-attribution a queued fold requires — MUL-4302). Only
// claim-receipt statuses (dispatched/running/waiting_local_directory) are valid
// planned-id targets; a queued task must miss so the caller routes it to the
// atomic merge instead.
func TestRegisterPlannedCommentForActiveTaskExcludesQueued(t *testing.T) {
	if testHandler == nil {
		t.Skip("database not available")
	}
	ctx := context.Background()
	agentID, issueID, runtimeID := dupRaceFixture(t, "dup-race-register-scope", 999314)
	commentID := insertDupRaceComment(t, issueID, "planned candidate", "1 minute")
	triggerID := insertDupRaceComment(t, issueID, "winner trigger", "6 minutes")

	var taskID 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 '5 minutes')
		RETURNING id
	`, agentID, runtimeID, issueID, triggerID).Scan(&taskID); err != nil {
		t.Fatalf("insert queued task: %v", err)
	}

	params := db.RegisterPlannedCommentForActiveTaskParams{
		CommentID: util.MustParseUUID(commentID),
		IssueID:   util.MustParseUUID(issueID),
		AgentID:   util.MustParseUUID(agentID),
	}

	// QUEUED target must MISS.
	if _, err := testHandler.Queries.RegisterPlannedCommentForActiveTask(ctx, params); !errors.Is(err, pgx.ErrNoRows) {
		t.Fatalf("register against a QUEUED task: err = %v, want pgx.ErrNoRows (queued excluded)", err)
	}

	// Flip to dispatched: now the claim receipt exists and the append is valid.
	if _, err := testPool.Exec(ctx, `UPDATE agent_task_queue SET status = 'dispatched', dispatched_at = now() WHERE id = $1`, taskID); err != nil {
		t.Fatalf("flip to dispatched: %v", err)
	}
	row, err := testHandler.Queries.RegisterPlannedCommentForActiveTask(ctx, params)
	if err != nil {
		t.Fatalf("register against a DISPATCHED task: %v", err)
	}
	var found bool
	for _, id := range row.CoalescedCommentIds {
		if uuidToString(id) == commentID {
			found = true
		}
	}
	if !found {
		t.Fatalf("dispatched register did not add the planned comment: %v", row.CoalescedCommentIds)
	}
}

// TestCommentEnqueueRaceQueuedWinnerReattributesOriginator is the regression for
// Elon round-3 must-fix 2: when the lost-race winner is a same-head QUEUED task,
// the losing comment must fold through the ATOMIC merge, which re-stamps the run
// to the NEW comment's originator — never a bare planned append that would leave
// a second member's comment executing under the first member's identity. Two
// different members: the winner is attributed to M1, the losing comment is M2's,
// and after the race the run must be re-attributed to M2.
func TestCommentEnqueueRaceQueuedWinnerReattributesOriginator(t *testing.T) {
	if testHandler == nil {
		t.Skip("database not available")
	}
	ctx := context.Background()
	agentID, issueID, _ := dupRaceFixture(t, "dup-race-reattr", 999315)
	agentUUID := util.MustParseUUID(agentID)
	issue, err := testHandler.Queries.GetIssue(ctx, util.MustParseUUID(issueID))
	if err != nil {
		t.Fatalf("load issue: %v", err)
	}
	agent, err := testHandler.Queries.GetAgent(ctx, agentUUID)
	if err != nil {
		t.Fatalf("load agent: %v", err)
	}

	// A second member M2 authors the losing comment.
	var m2 string
	if err := testPool.QueryRow(ctx, `INSERT INTO "user" (name, email) VALUES ('Race M2', '[email protected]') RETURNING id`).Scan(&m2); err != nil {
		t.Fatalf("create M2 user: %v", err)
	}
	t.Cleanup(func() { testPool.Exec(context.Background(), `DELETE FROM "user" WHERE id = $1`, m2) })
	if _, err := testPool.Exec(ctx, `INSERT INTO member (workspace_id, user_id, role) VALUES ($1, $2, 'member')`, testWorkspaceID, m2); err != nil {
		t.Fatalf("create M2 member: %v", err)
	}
	t.Cleanup(func() { testPool.Exec(context.Background(), `DELETE FROM member WHERE user_id = $1`, m2) })

	// Winner: a queued task attributed to M1 (testUserID) via its own comment.
	winnerCommentID := insertDupRaceComment(t, issueID, "M1 instruction", "6 minutes")
	if _, err := testHandler.TaskService.EnqueueTaskForMention(ctx, issue, agentUUID, util.MustParseUUID(winnerCommentID), service.OriginNamed); err != nil {
		t.Fatalf("enqueue winning task: %v", err)
	}
	// Losing comment authored by M2 in the same thread.
	var loserCommentID string
	if err := testPool.QueryRow(ctx, `
		INSERT INTO comment (issue_id, workspace_id, author_type, author_id, content, type, created_at)
		VALUES ($1, $2, 'member', $3, 'M2 instruction', 'comment', now() - interval '1 minute')
		RETURNING id
	`, issueID, testWorkspaceID, m2).Scan(&loserCommentID); err != nil {
		t.Fatalf("insert M2 loser comment: %v", err)
	}

	dbfx.Exec(t, `UPDATE comment SET parent_id=$2 WHERE id=$1`, loserCommentID, winnerCommentID)

	trigger := commentAgentTrigger{Agent: agent, Source: commentTriggerSourceMentionAgent}
	results := testHandler.enqueueCommentAgentTriggers(ctx, issue, util.MustParseUUID(loserCommentID), []commentAgentTrigger{trigger})
	if res := results[agentID]; res.status != DispatchCoalesced {
		t.Fatalf("queued-winner reattribution race: got status %q reason %q, want coalesced", res.status, res.reason)
	}

	// The run is now attributed to M2 (the folded comment's author) and triggered
	// by M2's comment — proving the atomic merge ran, not a bare planned append.
	var trig, orig string
	if err := testPool.QueryRow(ctx, `
		SELECT COALESCE(trigger_comment_id::text,''), COALESCE(originator_user_id::text,'')
		FROM agent_task_queue WHERE issue_id = $1 AND agent_id = $2 AND status = 'queued'
	`, issueID, agentID).Scan(&trig, &orig); err != nil {
		t.Fatalf("read winner attribution: %v", err)
	}
	if trig != loserCommentID {
		t.Fatalf("trigger_comment_id = %s, want repointed to M2's comment %s", trig, loserCommentID)
	}
	if orig != m2 {
		t.Fatalf("originator_user_id = %s, want re-attributed to M2 %s (bare append would leave M1)", orig, m2)
	}
}

// TestCommentEnqueueRaceNewerDifferentHeadNotDeferred is the regression for Elon
// round-5 must-fix: when the different-head task holding the slot is NEWER than
// the losing comment (and the comment is not in its planned ids), completion
// reconcile's `created_at > since` window never matches it — so the resolver must
// NOT fabricate a deferred off "head differs". It returns a truthful non-success
// (internal_error) instead, never a success-shaped outcome that drops the comment.
func TestCommentEnqueueRaceNewerDifferentHeadNotDeferred(t *testing.T) {
	if testHandler == nil {
		t.Skip("database not available")
	}
	ctx := context.Background()
	agentID, issueID, runtimeID := dupRaceFixture(t, "dup-race-newer-head", 999317)
	agentUUID := util.MustParseUUID(agentID)

	// Link a PR at head B so the request resolves head B.
	var prID string
	if err := testPool.QueryRow(ctx, `
		INSERT INTO github_pull_request (workspace_id, installation_id, repo_owner, repo_name, pr_number, title, state, html_url, pr_created_at, pr_updated_at, head_sha)
		VALUES ($1, 1, 'multica-ai', 'multica', 999317, 'review PR', 'open', 'https://example.test/pr', now(), now(), $2)
		RETURNING id
	`, testWorkspaceID, dupRaceHeadB).Scan(&prID); err != nil {
		t.Fatalf("seed PR: %v", err)
	}
	t.Cleanup(func() {
		testPool.Exec(context.Background(), `DELETE FROM issue_pull_request WHERE pull_request_id = $1`, prID)
	})
	t.Cleanup(func() { testPool.Exec(context.Background(), `DELETE FROM github_pull_request WHERE id = $1`, prID) })
	if _, err := testPool.Exec(ctx, `INSERT INTO issue_pull_request (issue_id, pull_request_id) VALUES ($1, $2)`, issueID, prID); err != nil {
		t.Fatalf("link PR: %v", err)
	}

	issue, err := testHandler.Queries.GetIssue(ctx, util.MustParseUUID(issueID))
	if err != nil {
		t.Fatalf("load issue: %v", err)
	}
	agent, err := testHandler.Queries.GetAgent(ctx, agentUUID)
	if err != nil {
		t.Fatalf("load agent: %v", err)
	}

	// Losing comment C persisted first (older); a head-A winner is created LATER
	// (newer than C), so reconcile's created_at window can never replay C.
	loserCommentID := insertDupRaceComment(t, issueID, "older losing comment", "5 minutes")
	winnerCommentID := insertDupRaceComment(t, issueID, "head A trigger", "4 minutes")
	if _, err := testPool.Exec(ctx, `
		INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, trigger_comment_id, delivered_comment_ids, status, priority, created_at, context)
		VALUES ($1, $2, $3, $4, ARRAY[$4::uuid], 'queued', 0, now() - interval '1 minute', jsonb_build_object('head_sha', $5::text))
	`, agentID, runtimeID, issueID, winnerCommentID, dupRaceHeadA); err != nil {
		t.Fatalf("insert newer head-A winner: %v", err)
	}

	trigger := commentAgentTrigger{Agent: agent, Source: commentTriggerSourceMentionAgent}
	results := testHandler.enqueueCommentAgentTriggers(ctx, issue, util.MustParseUUID(loserCommentID), []commentAgentTrigger{trigger})

	// Must NOT be a fabricated deferred — the newer head-A task cannot cover C.
	res := results[agentID]
	if res.status == DispatchDeferred || res.status == DispatchCoalesced || res.status == DispatchQueued {
		t.Fatalf("newer different-head: got success-shaped %q/%q, want a truthful non-success (not deferred/coalesced/queued)", res.status, res.reason)
	}
	if res.status != DispatchBlocked || res.reason != ReasonInternalError {
		t.Fatalf("newer different-head: got %q/%q, want blocked/internal_error", res.status, res.reason)
	}
	// The head-A winner was not mutated to fake coverage.
	var folded bool
	if err := testPool.QueryRow(ctx, `
		SELECT $2::uuid = ANY(coalesced_comment_ids) FROM agent_task_queue
		WHERE issue_id = $1 AND agent_id = $3 AND context->>'head_sha' = $4
	`, issueID, loserCommentID, agentID, dupRaceHeadA).Scan(&folded); err != nil {
		t.Fatalf("read winner coalesced: %v", err)
	}
	if folded {
		t.Fatal("losing comment was wrongly attached to the newer different-head task")
	}
}

// TestCommentEnqueueRaceMixedCoveringAndNewerNotDeferred covers Elon rounds 6+9:
// an OLDER different-head task and a NEWER one can coexist (running sits outside
// the unique index). At decision time nothing has attached the comment, so the
// resolver must return a truthful non-success rather than a fabricated deferred
// (round 6). Draining the chain then pins the boundary of the hand-off: the newer
// blocker is at a DIFFERENT head, so the obligation is NOT folded into it — that
// would let an old-head run consume a current-head request (TEN-356, round 9).
// This is the one documented best-effort shape; correctness rests on the fact
// that nothing ever promised coverage here.
func TestCommentEnqueueRaceMixedCoveringAndNewerNotDeferred(t *testing.T) {
	if testHandler == nil {
		t.Skip("database not available")
	}
	ctx := context.Background()
	agentID, issueID, runtimeID := dupRaceFixture(t, "dup-race-mixed", 999318)
	agentUUID := util.MustParseUUID(agentID)

	// PR at head B so the request resolves head B; both tasks are at head A.
	var prID string
	if err := testPool.QueryRow(ctx, `
		INSERT INTO github_pull_request (workspace_id, installation_id, repo_owner, repo_name, pr_number, title, state, html_url, pr_created_at, pr_updated_at, head_sha)
		VALUES ($1, 1, 'multica-ai', 'multica', 999318, 'review PR', 'open', 'https://example.test/pr', now(), now(), $2)
		RETURNING id
	`, testWorkspaceID, dupRaceHeadB).Scan(&prID); err != nil {
		t.Fatalf("seed PR: %v", err)
	}
	t.Cleanup(func() {
		testPool.Exec(context.Background(), `DELETE FROM issue_pull_request WHERE pull_request_id = $1`, prID)
	})
	t.Cleanup(func() { testPool.Exec(context.Background(), `DELETE FROM github_pull_request WHERE id = $1`, prID) })
	if _, err := testPool.Exec(ctx, `INSERT INTO issue_pull_request (issue_id, pull_request_id) VALUES ($1, $2)`, issueID, prID); err != nil {
		t.Fatalf("link PR: %v", err)
	}

	issue, err := testHandler.Queries.GetIssue(ctx, util.MustParseUUID(issueID))
	if err != nil {
		t.Fatalf("load issue: %v", err)
	}
	agent, err := testHandler.Queries.GetAgent(ctx, agentUUID)
	if err != nil {
		t.Fatalf("load agent: %v", err)
	}

	// T0: A (head-A) running, OLDER than the comment. T1: comment C. T2: B
	// (head-A) queued, NEWER than C — it occupies the unique slot.
	aTrigger := insertDupRaceComment(t, issueID, "A trigger", "11 minutes")
	var taskA string
	if err := testPool.QueryRow(ctx, `
		INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, trigger_comment_id, delivered_comment_ids, status, priority, created_at, started_at, context)
		VALUES ($1, $2, $3, $4, ARRAY[$4::uuid], 'running', 0, now() - interval '10 minutes', now() - interval '9 minutes', jsonb_build_object('head_sha', $5::text))
		RETURNING id
	`, agentID, runtimeID, issueID, aTrigger, dupRaceHeadA).Scan(&taskA); err != nil {
		t.Fatalf("insert running task A: %v", err)
	}
	loserCommentID := insertDupRaceComment(t, issueID, "losing comment C", "5 minutes")
	bTrigger := insertDupRaceComment(t, issueID, "B trigger", "2 minutes")
	var taskB string
	if err := testPool.QueryRow(ctx, `
		INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, trigger_comment_id, delivered_comment_ids, status, priority, created_at, context)
		VALUES ($1, $2, $3, $4, ARRAY[$4::uuid], 'queued', 0, now() - interval '1 minute', jsonb_build_object('head_sha', $5::text))
		RETURNING id
	`, agentID, runtimeID, issueID, bTrigger, dupRaceHeadA).Scan(&taskB); err != nil {
		t.Fatalf("insert queued task B: %v", err)
	}

	trigger := commentAgentTrigger{Agent: agent, Source: commentTriggerSourceMentionAgent}
	results := testHandler.enqueueCommentAgentTriggers(ctx, issue, util.MustParseUUID(loserCommentID), []commentAgentTrigger{trigger})

	// Even though a covering (older) task exists, the newer one forbids deferral.
	res := results[agentID]
	if res.status == DispatchDeferred || res.status == DispatchCoalesced || res.status == DispatchQueued {
		t.Fatalf("mixed covering+newer: got success-shaped %q/%q, want a truthful non-success", res.status, res.reason)
	}
	if res.status != DispatchBlocked || res.reason != ReasonInternalError {
		t.Fatalf("mixed covering+newer: got %q/%q, want blocked/internal_error", res.status, res.reason)
	}

	// Draining the chain: A completes and its replay of C collides with the still-
	// queued newer B, which is at a DIFFERENT head. The hand-off correctly declines
	// to fold C there (that would let an old-head run consume it — TEN-356), so this
	// is the one documented best-effort shape: C stays uncovered and the failure is
	// logged at error. What matters for correctness is that the initial call never
	// PROMISED coverage — it returned a non-success above.
	if w := completeTaskViaHandler(t, taskA, "done"); w.Code != 200 {
		t.Fatalf("complete A: %d: %s", w.Code, w.Body.String())
	}
	if commentCovered(t, issueID, agentID, loserCommentID, "queued") {
		t.Fatal("C was folded into the different-head queued blocker (TEN-356 violation)")
	}
}

// commentCoveredByAnyActiveTask reports whether ANY active task for (issue, agent)
// has the comment as its trigger or in its planned (coalesced) batch.
func commentCoveredByAnyActiveTask(t *testing.T, issueID, agentID, commentID string) bool {
	t.Helper()
	var ok bool
	if err := testPool.QueryRow(context.Background(), `
		SELECT EXISTS (
			SELECT 1 FROM agent_task_queue
			WHERE issue_id = $1 AND agent_id = $2
			  AND status IN ('queued','dispatched','running','waiting_local_directory')
			  AND ($3::uuid = trigger_comment_id OR $3::uuid = ANY(coalesced_comment_ids))
		)
	`, issueID, agentID, commentID).Scan(&ok); err != nil {
		t.Fatalf("check coverage: %v", err)
	}
	return ok
}

// coveringTaskHead returns the head_sha of the active task that covers the comment
// (empty when none does), so a test can assert WHICH head finally covers it.
func coveringTaskHead(t *testing.T, issueID, agentID, commentID string) string {
	t.Helper()
	var head string
	if err := testPool.QueryRow(context.Background(), `
		SELECT COALESCE(context->>'head_sha','')
		FROM agent_task_queue
		WHERE issue_id = $1 AND agent_id = $2
		  AND status IN ('queued','dispatched','running','waiting_local_directory')
		  AND ($3::uuid = trigger_comment_id OR $3::uuid = ANY(coalesced_comment_ids))
		ORDER BY created_at DESC LIMIT 1
	`, issueID, agentID, commentID).Scan(&head); err != nil {
		t.Fatalf("read covering task head: %v", err)
	}
	return head
}

// seedDupRacePR links a PR at head B so the reviewed head resolves to head B.
func seedDupRacePR(t *testing.T, issueID string, prNumber int) {
	t.Helper()
	ctx := context.Background()
	var prID string
	if err := testPool.QueryRow(ctx, `
		INSERT INTO github_pull_request (workspace_id, installation_id, repo_owner, repo_name, pr_number, title, state, html_url, pr_created_at, pr_updated_at, head_sha)
		VALUES ($1, 1, 'multica-ai', 'multica', $2, 'review PR', 'open', 'https://example.test/pr', now(), now(), $3)
		RETURNING id
	`, testWorkspaceID, prNumber, dupRaceHeadB).Scan(&prID); err != nil {
		t.Fatalf("seed PR: %v", err)
	}
	t.Cleanup(func() {
		testPool.Exec(context.Background(), `DELETE FROM issue_pull_request WHERE pull_request_id = $1`, prID)
	})
	t.Cleanup(func() { testPool.Exec(context.Background(), `DELETE FROM github_pull_request WHERE id = $1`, prID) })
	if _, err := testPool.Exec(ctx, `INSERT INTO issue_pull_request (issue_id, pull_request_id) VALUES ($1, $2)`, issueID, prID); err != nil {
		t.Fatalf("link PR: %v", err)
	}
}

// TestReconcileBlockedReplayPropagatesToClaimedBlocker is the regression for Elon
// round-8 must-fix 2 (planned-registration path): a planned-id obligation must not
// be consumed once. C is registered as a planned input on running task A; before A
// completes, a NEWER different-head task B is claimed and holds the unique slot. A's
// completion replay of C is therefore blocked — and must be HANDED to B rather than
// discarded, so completing B finally covers C.
func TestReconcileBlockedReplayPropagatesToClaimedBlocker(t *testing.T) {
	if testHandler == nil {
		t.Skip("database not available")
	}
	ctx := context.Background()
	agentID, issueID, runtimeID := dupRaceFixture(t, "dup-race-propagate-claimed", 999319)
	seedDupRacePR(t, issueID, 999319)

	aTrigger := insertDupRaceComment(t, issueID, "A trigger", "11 minutes")
	loserCommentID := insertDupRaceComment(t, issueID, "comment C needing coverage", "8 minutes")
	bTrigger := insertDupRaceComment(t, issueID, "B trigger", "6 minutes")

	// A: running (outside the unique index), head A, with C already registered as a
	// planned-but-NOT-delivered input — the exact state the lost-race path leaves.
	var taskA string
	if err := testPool.QueryRow(ctx, `
		INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, trigger_comment_id, coalesced_comment_ids, delivered_comment_ids, status, priority, created_at, started_at, context)
		VALUES ($1, $2, $3, $4, ARRAY[$5::uuid], ARRAY[$4::uuid], 'running', 0, now() - interval '10 minutes', now() - interval '9 minutes', jsonb_build_object('head_sha', $6::text))
		RETURNING id
	`, agentID, runtimeID, issueID, aTrigger, loserCommentID, dupRaceHeadA).Scan(&taskA); err != nil {
		t.Fatalf("insert running task A: %v", err)
	}
	// B: a NEWER different-head task, already claimed (dispatched) — it holds the slot
	// and its own reconcile can never see the older C.
	var taskB string
	if err := testPool.QueryRow(ctx, `
		INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, trigger_comment_id, delivered_comment_ids, status, priority, created_at, dispatched_at, context)
		VALUES ($1, $2, $3, $4, ARRAY[$4::uuid], 'dispatched', 0, now() - interval '5 minutes', now() - interval '4 minutes', jsonb_build_object('head_sha', $5::text))
		RETURNING id
	`, agentID, runtimeID, issueID, bTrigger, dupRaceHeadA).Scan(&taskB); err != nil {
		t.Fatalf("insert dispatched blocker B: %v", err)
	}

	// A completes: its replay of C collides with B and must be handed to B.
	if w := completeTaskViaHandler(t, taskA, "done"); w.Code != 200 {
		t.Fatalf("complete A: %d: %s", w.Code, w.Body.String())
	}
	var plannedOnB, deliveredOnB bool
	if err := testPool.QueryRow(ctx, `
		SELECT $2::uuid = ANY(coalesced_comment_ids), $2::uuid = ANY(delivered_comment_ids)
		FROM agent_task_queue WHERE id = $1
	`, taskB, loserCommentID).Scan(&plannedOnB, &deliveredOnB); err != nil {
		t.Fatalf("read blocker B: %v", err)
	}
	if !plannedOnB {
		t.Fatal("blocked replay was discarded — C was not handed to the blocker, so it is lost")
	}
	if deliveredOnB {
		t.Fatal("C must not be faked as delivered on the already-claimed blocker")
	}

	// B completes: C is in its planned ids, so it is replayed and finally covered.
	if _, err := testPool.Exec(ctx, `UPDATE agent_task_queue SET status = 'running', started_at = now() WHERE id = $1`, taskB); err != nil {
		t.Fatalf("advance B to running: %v", err)
	}
	if w := completeTaskViaHandler(t, taskB, "done"); w.Code != 200 {
		t.Fatalf("complete B: %d: %s", w.Code, w.Body.String())
	}
	if !commentCoveredByAnyActiveTask(t, issueID, agentID, loserCommentID) {
		t.Fatal("after the blocker chain drained, no task covers C — the deferral was not durable")
	}
	// The hand-off never let an old-head run consume C: the follow-up that finally
	// covers it is stamped for the CURRENT head (TEN-356).
	if head := coveringTaskHead(t, issueID, agentID, loserCommentID); head != dupRaceHeadB {
		t.Fatalf("covering task head_sha = %q, want the current head %q", head, dupRaceHeadB)
	}
}

// TestReconcileBlockedReplayPropagatesToQueuedBlocker is the regression for Elon
// round-8 must-fix 1 (AlreadyPending path): C arrives while A is running, so A's
// reconcile finds it by timestamp — but a NEWER task B is QUEUED when A completes,
// blocking the replay. When B is at the CURRENT head the obligation is handed to it
// through the ATOMIC merge (never a bare append), so C ends up covered at that head.
func TestReconcileBlockedReplayPropagatesToQueuedBlocker(t *testing.T) {
	if testHandler == nil {
		t.Skip("database not available")
	}
	ctx := context.Background()
	agentID, issueID, runtimeID := dupRaceFixture(t, "dup-race-propagate-queued", 999320)
	seedDupRacePR(t, issueID, 999320)

	aTrigger := insertDupRaceComment(t, issueID, "A trigger", "11 minutes")
	// C lands while A is running (newer than A) and is NOT in A's arrays — the
	// AlreadyPending shape, covered by reconcile's timestamp window alone.
	loserCommentID := insertDupRaceComment(t, issueID, "comment C posted during the run", "5 minutes")
	bTrigger := insertDupRaceComment(t, issueID, "B trigger", "2 minutes")

	var taskA string
	if err := testPool.QueryRow(ctx, `
		INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, trigger_comment_id, delivered_comment_ids, status, priority, created_at, started_at, context)
		VALUES ($1, $2, $3, $4, ARRAY[$4::uuid], 'running', 0, now() - interval '10 minutes', now() - interval '9 minutes', jsonb_build_object('head_sha', $5::text))
		RETURNING id
	`, agentID, runtimeID, issueID, aTrigger, dupRaceHeadA).Scan(&taskA); err != nil {
		t.Fatalf("insert running task A: %v", err)
	}
	// B: NEWER than C, QUEUED, and at the CURRENT head — it occupies the unique slot
	// and its own reconcile can never see the older C.
	var taskB string
	if err := testPool.QueryRow(ctx, `
		INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, trigger_comment_id, status, priority, created_at, context)
		VALUES ($1, $2, $3, $4, 'queued', 0, now() - interval '1 minute', jsonb_build_object('head_sha', $5::text))
		RETURNING id
	`, agentID, runtimeID, issueID, bTrigger, dupRaceHeadB).Scan(&taskB); err != nil {
		t.Fatalf("insert queued blocker B: %v", err)
	}

	if w := completeTaskViaHandler(t, taskA, "done"); w.Code != 200 {
		t.Fatalf("complete A: %d: %s", w.Code, w.Body.String())
	}
	// The obligation must have been folded into the queued blocker.
	if !commentCoveredByAnyActiveTask(t, issueID, agentID, loserCommentID) {
		t.Fatal("blocked replay was discarded — C is covered by no active task, so it is lost")
	}
	// The fold went through the ATOMIC merge, so B still covers its own trigger too,
	// and the covering run is at the current head.
	if !commentCoveredByAnyActiveTask(t, issueID, agentID, bTrigger) {
		t.Fatal("the blocker's own trigger comment lost coverage during the hand-off")
	}
	if head := coveringTaskHead(t, issueID, agentID, loserCommentID); head != dupRaceHeadB {
		t.Fatalf("covering task head_sha = %q, want the current head %q", head, dupRaceHeadB)
	}
}

// TestReconcileBlockedReplayNeverFoldsIntoDifferentHeadQueuedBlocker is the
// regression for Elon round-9 must-fix 2: the hand-off must NOT merge across heads.
// A different-head QUEUED blocker would make the comment that run's trigger and mark
// it delivered at claim time, so an old-head run would consume a new-head request and
// no new-head follow-up would ever exist (TEN-356). The hand-off must decline instead
// — leaving the blocker untouched — rather than "cover" the comment incorrectly.
func TestReconcileBlockedReplayNeverFoldsIntoDifferentHeadQueuedBlocker(t *testing.T) {
	if testHandler == nil {
		t.Skip("database not available")
	}
	ctx := context.Background()
	agentID, issueID, runtimeID := dupRaceFixture(t, "dup-race-propagate-crosshead", 999321)
	seedDupRacePR(t, issueID, 999321) // current head = head B

	aTrigger := insertDupRaceComment(t, issueID, "A trigger", "11 minutes")
	loserCommentID := insertDupRaceComment(t, issueID, "comment C posted during the run", "5 minutes")
	bTrigger := insertDupRaceComment(t, issueID, "B trigger", "2 minutes")

	var taskA string
	if err := testPool.QueryRow(ctx, `
		INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, trigger_comment_id, delivered_comment_ids, status, priority, created_at, started_at, context)
		VALUES ($1, $2, $3, $4, ARRAY[$4::uuid], 'running', 0, now() - interval '10 minutes', now() - interval '9 minutes', jsonb_build_object('head_sha', $5::text))
		RETURNING id
	`, agentID, runtimeID, issueID, aTrigger, dupRaceHeadA).Scan(&taskA); err != nil {
		t.Fatalf("insert running task A: %v", err)
	}
	// B: QUEUED at the OLD head — folding C into it would hand a current-head
	// request to an old-head run.
	var taskB string
	if err := testPool.QueryRow(ctx, `
		INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, trigger_comment_id, status, priority, created_at, context)
		VALUES ($1, $2, $3, $4, 'queued', 0, now() - interval '1 minute', jsonb_build_object('head_sha', $5::text))
		RETURNING id
	`, agentID, runtimeID, issueID, bTrigger, dupRaceHeadA).Scan(&taskB); err != nil {
		t.Fatalf("insert queued cross-head blocker B: %v", err)
	}

	if w := completeTaskViaHandler(t, taskA, "done"); w.Code != 200 {
		t.Fatalf("complete A: %d: %s", w.Code, w.Body.String())
	}

	// C must NOT have been consumed by the old-head run, and B's own trigger and
	// head must be untouched.
	var folded bool
	var bHead string
	if err := testPool.QueryRow(ctx, `
		SELECT $2::uuid = trigger_comment_id OR $2::uuid = ANY(coalesced_comment_ids), COALESCE(context->>'head_sha','')
		FROM agent_task_queue WHERE id = $1
	`, taskB, loserCommentID).Scan(&folded, &bHead); err != nil {
		t.Fatalf("read cross-head blocker: %v", err)
	}
	if folded {
		t.Fatal("C was folded into a DIFFERENT-head queued run — an old-head run would consume a current-head request (TEN-356)")
	}
	if bHead != dupRaceHeadA {
		t.Fatalf("blocker head_sha = %q, want unchanged %q", bHead, dupRaceHeadA)
	}
}