multica/server/internal/handler/comment_steer_test.go

package handler

import (
	"context"
	"fmt"
	"net/http"
	"testing"

	"github.com/multica-ai/multica/server/internal/events"
	"github.com/multica-ai/multica/server/internal/testutil"
	"github.com/multica-ai/multica/server/pkg/protocol"
)

func agentMention(name, id string) string {
	return fmt.Sprintf("[@%s](mention://agent/%s)", name, id)
}

func postSteeredComment(t *testing.T, issueID, parentID, content string, steerTaskIDs ...string) *testutil.Response {
	t.Helper()
	body := map[string]any{"content": content, "steer_task_ids": steerTaskIDs}
	if parentID != "" {
		body["parent_id"] = parentID
	}
	return testutil.Call(t, testHandler.CreateComment, withURLParam(
		newRequest(http.MethodPost, "/api/issues/"+issueID+"/comments", body), "id", issueID))
}

// runningSteerableTask starts another agent's running, steer-capable turn on
// the fixture's issue.
func runningSteerableTask(t *testing.T, f supplementFixture, name string) (agentID, taskID string) {
	t.Helper()
	runtimeID := dbfx.Runtime(t, "steer-"+name, testutil.Cols{"provider": "claude"})
	agentID = dbfx.Agent(t, name, runtimeID)
	taskID = dbfx.Task(t, agentID, testutil.Cols{
		"issue_id": f.issueID, "runtime_id": runtimeID, "status": "running", "started_at": testutil.Raw("now()"),
	})
	dbfx.Exec(t, `INSERT INTO task_supplement_capability (task_id, workspace_id, issue_id, capability) VALUES ($1, $2, $3, $4)`,
		taskID, testWorkspaceID, f.issueID, protocol.DaemonCapabilityTaskSupplementV1)
	dbfx.Cleanup(t, `DELETE FROM task_supplement_capability WHERE task_id = $1`, taskID)
	dbfx.Cleanup(t, `DELETE FROM task_supplement WHERE task_id = $1`, taskID)
	return agentID, taskID
}

func completeTaskViaDaemon(t *testing.T, taskID string) {
	t.Helper()
	req := newDaemonTokenRequest(http.MethodPost, "/api/daemon/tasks/"+taskID+"/complete",
		map[string]any{"output": "done"}, testWorkspaceID, "legit-daemon")
	testutil.Call(t, testHandler.CompleteTask, withURLParam(req, "taskId", taskID)).Want(http.StatusOK)
}

func outcomeFor(outcomes []CommentTriggerOutcome, agentID string) (CommentTriggerOutcome, bool) {
	for _, outcome := range outcomes {
		if outcome.TargetType == "agent" && outcome.TargetID == agentID {
			return outcome, true
		}
	}
	return CommentTriggerOutcome{}, false
}

func TestCreateCommentSteersTheThreadAgentsRunningTurn(t *testing.T) {
	for _, provider := range []string{"codex", "cursor"} {
		t.Run(provider, func(t *testing.T) {
			f := newSupplementFixture(t, provider, "running", true)
			dbfx.Exec(t, `UPDATE comment SET content = $2 WHERE id = $1`, f.triggerID,
				agentMention("Supplement", f.agentID)+" original objective")
			dbfx.Exec(t, `UPDATE agent_task_queue SET delivered_comment_ids = ARRAY[trigger_comment_id] WHERE id = $1`, f.taskID)
			dbfx.Cleanup(t, `DELETE FROM agent_task_queue WHERE issue_id = $1`, f.issueID)

			var created CommentResponse
			postSteeredComment(t, f.issueID, f.triggerID, "Only fix web.", f.taskID).Want(http.StatusCreated).JSON(&created)
			if len(created.Supplements) != 1 {
				t.Fatalf("receipts = %+v, want one for the running turn", created.Supplements)
			}
			receipt := created.Supplements[0]
			if receipt.TaskID != f.taskID || receipt.AgentID != f.agentID || receipt.Status != "pending" {
				t.Fatalf("receipt = %+v, want pending for task %s / agent %s", receipt, f.taskID, f.agentID)
			}
			if created.SupplementTaskID != f.taskID || created.SupplementStatus != "pending" {
				t.Fatalf("legacy receipt fields = %s/%s, want the first receipt", created.SupplementTaskID, created.SupplementStatus)
			}
			if n := dbfx.Count(t, `SELECT count(*) FROM agent_task_queue WHERE issue_id = $1`, f.issueID); n != 1 {
				t.Fatalf("steering created %d runs, want the running turn only", n)
			}

			completeTaskViaDaemon(t, f.taskID)
			if n := dbfx.Count(t, `SELECT count(*) FROM agent_task_queue WHERE issue_id = $1`, f.issueID); n != 1 {
				t.Fatalf("completion replayed a steered comment into %d runs", n)
			}
		})
	}
}

func TestCreateCommentSteerKeepsTheNormalTriggerWhenTheTurnCannotTakeInput(t *testing.T) {
	for _, tc := range []struct {
		name       string
		status     string
		negotiated bool
	}{
		{name: "turn without the capability", status: "running", negotiated: false},
		{name: "turn not started", status: "dispatched", negotiated: true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			f := newSupplementFixture(t, "codex", tc.status, tc.negotiated)
			dbfx.Exec(t, `UPDATE comment SET content = $2 WHERE id = $1`, f.triggerID,
				agentMention("Supplement", f.agentID)+" original objective")
			dbfx.Exec(t, `UPDATE agent_task_queue SET delivered_comment_ids = ARRAY[trigger_comment_id] WHERE id = $1`, f.taskID)
			dbfx.Cleanup(t, `DELETE FROM agent_task_queue WHERE issue_id = $1`, f.issueID)

			var created CommentResponse
			postSteeredComment(t, f.issueID, f.triggerID, "Only fix web.", f.taskID).Want(http.StatusCreated).JSON(&created)
			if len(created.Supplements) != 0 {
				t.Fatalf("bound a turn that cannot take input: %+v", created.Supplements)
			}
			if n := dbfx.Count(t, `SELECT count(*) FROM task_supplement WHERE comment_id = $1`, created.ID); n != 0 {
				t.Fatalf("stored %d receipts", n)
			}
			if tc.status != "running" {
				return
			}
			// The comment keeps its deferred trigger and becomes the follow-up.
			completeTaskViaDaemon(t, f.taskID)
			var trigger string
			dbfx.QueryRow(t, `SELECT trigger_comment_id FROM agent_task_queue WHERE issue_id = $1 AND id <> $2`,
				f.issueID, f.taskID).Scan(&trigger)
			if trigger != created.ID {
				t.Fatalf("follow-up trigger = %s, want the unsteered comment %s", trigger, created.ID)
			}
		})
	}
}

func TestCreateCommentSteersOnlyRunningRecipients(t *testing.T) {
	f := newSupplementFixture(t, "codex", "running", true)
	dbfx.Cleanup(t, `DELETE FROM agent_task_queue WHERE issue_id = $1`, f.issueID)
	idleAgentID := dbfx.Agent(t, "Idle recipient", f.runtimeID)
	content := agentMention("Supplement", f.agentID) + " only web; " + agentMention("Idle recipient", idleAgentID) + " check desktop"

	var created CommentResponse
	postSteeredComment(t, f.issueID, "", content, f.taskID).Want(http.StatusCreated).JSON(&created)
	if len(created.Supplements) != 1 || created.Supplements[0].AgentID != f.agentID {
		t.Fatalf("receipts = %+v, want the running agent only", created.Supplements)
	}
	if outcome, ok := outcomeFor(created.TriggerOutcomes, f.agentID); !ok || outcome.Status != DispatchSteered || outcome.ReasonCode != ReasonSteered {
		t.Fatalf("running recipient outcome = %+v (%v), want steered", outcome, ok)
	}
	if outcome, ok := outcomeFor(created.TriggerOutcomes, idleAgentID); !ok || outcome.Status == DispatchSteered || outcome.Status == DispatchBlocked {
		t.Fatalf("idle recipient outcome = %+v (%v), want a normal trigger", outcome, ok)
	}
	if n := dbfx.Count(t, `SELECT count(*) FROM agent_task_queue WHERE agent_id = $1 AND trigger_comment_id = $2`, idleAgentID, created.ID); n != 1 {
		t.Fatalf("idle recipient has %d runs for the comment, want 1", n)
	}
	if n := dbfx.Count(t, `SELECT count(*) FROM agent_task_queue WHERE agent_id = $1`, f.agentID); n != 1 {
		t.Fatalf("steered recipient has %d runs, want its running turn only", n)
	}
}

func TestCreateCommentSteersSeveralRunningTurns(t *testing.T) {
	f := newSupplementFixture(t, "codex", "running", true)
	otherAgentID, otherTaskID := runningSteerableTask(t, f, "Second runner")
	content := agentMention("Supplement", f.agentID) + " and " + agentMention("Second runner", otherAgentID) + " stop touching desktop"

	var created CommentResponse
	postSteeredComment(t, f.issueID, "", content, f.taskID, otherTaskID).Want(http.StatusCreated).JSON(&created)
	got := map[string]string{}
	for _, receipt := range created.Supplements {
		got[receipt.AgentID] = receipt.TaskID
	}
	if len(created.Supplements) != 2 || got[f.agentID] != f.taskID || got[otherAgentID] != otherTaskID {
		t.Fatalf("receipts = %+v, want one per running turn", created.Supplements)
	}

	var timeline []TimelineEntry
	testutil.Call(t, testHandler.ListTimeline, withURLParam(
		newRequest(http.MethodGet, "/api/issues/"+f.issueID+"/timeline", nil), "id", f.issueID)).Want(http.StatusOK).JSON(&timeline)
	for _, entry := range timeline {
		if entry.ID == created.ID && len(entry.Supplements) != 2 {
			t.Fatalf("timeline receipts = %+v, want both runs", entry.Supplements)
		}
	}
}

func TestCompletionReplaysASteeredCommentOnlyForAgentsItDidNotSteer(t *testing.T) {
	f := newSupplementFixture(t, "codex", "running", true)
	dbfx.Exec(t, `UPDATE agent_task_queue SET delivered_comment_ids = ARRAY[trigger_comment_id] WHERE id = $1`, f.taskID)
	dbfx.Cleanup(t, `DELETE FROM agent_task_queue WHERE issue_id = $1`, f.issueID)
	otherAgentID, otherTaskID := runningSteerableTask(t, f, "Waits its turn")
	content := agentMention("Supplement", f.agentID) + " only web; " + agentMention("Waits its turn", otherAgentID) + " next round"

	var created CommentResponse
	postSteeredComment(t, f.issueID, "", content, f.taskID).Want(http.StatusCreated).JSON(&created)
	if len(created.Supplements) != 1 || created.Supplements[0].AgentID != f.agentID {
		t.Fatalf("receipts = %+v, want the steered agent only", created.Supplements)
	}

	completeTaskViaDaemon(t, otherTaskID)
	if n := dbfx.Count(t, `SELECT count(*) FROM agent_task_queue WHERE agent_id = $1 AND trigger_comment_id = $2`, otherAgentID, created.ID); n != 1 {
		t.Fatalf("unsteered recipient got %d follow-ups, want 1", n)
	}
	completeTaskViaDaemon(t, f.taskID)
	if n := dbfx.Count(t, `SELECT count(*) FROM agent_task_queue WHERE agent_id = $1`, f.agentID); n != 1 {
		t.Fatalf("steered recipient has %d runs, want no follow-up", n)
	}
}

func TestCreateCommentSteerNeverRetargetsAnotherTurn(t *testing.T) {
	f := newSupplementFixture(t, "codex", "running", true)
	dbfx.Exec(t, `UPDATE comment SET content = $2 WHERE id = $1`, f.triggerID, agentMention("A", f.agentID)+" original thread")
	dbfx.Cleanup(t, `DELETE FROM agent_task_queue WHERE issue_id = $1`, f.issueID)
	// The chosen turn ends before the send arrives, and the same agent has
	// meanwhile started a turn for another thread.
	dbfx.Exec(t, `UPDATE agent_task_queue SET status = 'completed', completed_at = now() WHERE id = $1`, f.taskID)
	otherRoot := dbfx.Comment(t, f.issueID, agentMention("A", f.agentID)+" unrelated thread")
	successor := dbfx.Task(t, f.agentID, testutil.Cols{
		"issue_id": f.issueID, "runtime_id": f.runtimeID, "trigger_comment_id": otherRoot,
		"status": "running", "started_at": testutil.Raw("now()"),
	})
	dbfx.Exec(t, `INSERT INTO task_supplement_capability (task_id, workspace_id, issue_id, capability) VALUES ($1, $2, $3, $4)`,
		successor, testWorkspaceID, f.issueID, protocol.DaemonCapabilityTaskSupplementV1)
	dbfx.Cleanup(t, `DELETE FROM task_supplement_capability WHERE task_id = $1`, successor)
	dbfx.Cleanup(t, `DELETE FROM task_supplement WHERE task_id = $1`, successor)

	var sent CommentResponse
	postSteeredComment(t, f.issueID, f.triggerID, "Only change the original task", f.taskID).Want(http.StatusCreated).JSON(&sent)
	if len(sent.Supplements) != 0 {
		t.Fatalf("the chosen turn ended, but the send steered another turn: %+v", sent.Supplements)
	}
	if n := dbfx.Count(t, `SELECT count(*) FROM task_supplement WHERE task_id = $1`, successor); n != 0 {
		t.Fatalf("successor turn received %d steered messages", n)
	}
}

func TestCompletionKeepsAnAgentTheAuthorChoseNotToStart(t *testing.T) {
	f := newSupplementFixture(t, "codex", "running", true)
	dbfx.Cleanup(t, `DELETE FROM agent_task_queue WHERE issue_id = $1`, f.issueID)
	otherAgent, otherTask := runningSteerableTask(t, f, "Other recipient")
	content := agentMention("A", f.agentID) + " and " + agentMention("B", otherAgent) + " only do this once"
	dbfx.Exec(t, `UPDATE comment SET content = $2 WHERE id = $1`, f.triggerID, content)
	dbfx.Exec(t, `UPDATE agent_task_queue SET trigger_comment_id = $2, comment_thread_id = $2, delivered_comment_ids = ARRAY[$2::uuid] WHERE id = $1`, otherTask, f.triggerID)
	// B is working in this thread; the reply goes to A only ("Won't start
	// this time" for B, or "Send as a new run" for a receipt A missed).
	body := map[string]any{"content": content, "parent_id": f.triggerID, "suppress_agent_ids": []string{otherAgent}}
	testutil.Call(t, testHandler.CreateComment, withURLParam(
		newRequest(http.MethodPost, "/api/issues/"+f.issueID+"/comments", body), "id", f.issueID)).Want(http.StatusCreated)
	if n := dbfx.Count(t, `SELECT count(*) FROM agent_task_queue WHERE agent_id = $1`, otherAgent); n != 1 {
		t.Fatalf("the reply started another B run: %d", n)
	}
	completeTaskViaDaemon(t, otherTask)
	if n := dbfx.Count(t, `SELECT count(*) FROM agent_task_queue WHERE agent_id = $1`, otherAgent); n != 1 {
		t.Fatalf("B's completion replayed the reply into %d runs, want 1", n)
	}
}

func TestCreateCommentSteerIgnoresATurnOfAnotherIssue(t *testing.T) {
	f := newSupplementFixture(t, "codex", "running", true)
	dbfx.Cleanup(t, `DELETE FROM agent_task_queue WHERE issue_id = $1`, f.issueID)
	// The same agent is also running on another issue.
	otherIssueID := dbfx.Issue(t, "Another issue")
	elsewhere := dbfx.Task(t, f.agentID, testutil.Cols{
		"issue_id": otherIssueID, "runtime_id": f.runtimeID, "status": "running", "started_at": testutil.Raw("now()"),
	})
	dbfx.Exec(t, `INSERT INTO task_supplement_capability (task_id, workspace_id, issue_id, capability) VALUES ($1, $2, $3, $4)`,
		elsewhere, testWorkspaceID, otherIssueID, protocol.DaemonCapabilityTaskSupplementV1)
	dbfx.Cleanup(t, `DELETE FROM task_supplement_capability WHERE task_id = $1`, elsewhere)
	content := agentMention("Supplement", f.agentID) + " only web"

	var created CommentResponse
	postSteeredComment(t, f.issueID, "", content, elsewhere).Want(http.StatusCreated).JSON(&created)
	if len(created.Supplements) != 0 {
		t.Fatalf("receipts = %+v, want none for a turn of another issue", created.Supplements)
	}
	if n := dbfx.Count(t, `SELECT count(*) FROM task_supplement WHERE task_id = $1`, elsewhere); n != 0 {
		t.Fatalf("another issue's turn received %d messages", n)
	}
	if outcome, ok := outcomeFor(created.TriggerOutcomes, f.agentID); !ok || outcome.Status == DispatchSteered {
		t.Fatalf("recipient outcome = %+v (%v), want its normal trigger", outcome, ok)
	}
}

func TestUpdateCommentKeepsTheLatestEditsSuppression(t *testing.T) {
	f := newSupplementFixture(t, "codex", "running", true)
	dbfx.Cleanup(t, `DELETE FROM agent_task_queue WHERE issue_id = $1`, f.issueID)
	commentID := dbfx.Comment(t, f.issueID, "initial body")
	h := *testHandler
	h.Bus = events.New()
	laterContent := agentMention("A", f.agentID) + " latest edit"
	nested := false
	h.Bus.Subscribe(protocol.EventCommentUpdated, func(events.Event) {
		if nested {
			return
		}
		nested = true
		// A second editor saves right after the first edit's text commits.
		later := withURLParam(newRequest(http.MethodPut, "/api/comments/"+commentID, map[string]any{
			"content": laterContent, "expected_revision": 2, "suppress_agent_ids": []string{},
		}), "commentId", commentID)
		testutil.Call(t, h.UpdateComment, later).Want(http.StatusOK)
	})
	first := withURLParam(newRequest(http.MethodPut, "/api/comments/"+commentID, map[string]any{
		"content": agentMention("A", f.agentID) + " first edit", "expected_revision": 1, "suppress_agent_ids": []string{f.agentID},
	}), "commentId", commentID)
	testutil.Call(t, h.UpdateComment, first).Want(http.StatusOK)
	saved, err := testHandler.Queries.GetComment(context.Background(), parseUUID(commentID))
	if err != nil {
		t.Fatal(err)
	}
	if saved.Content != laterContent || saved.Revision != 3 {
		t.Fatalf("latest body/revision = %q/%d", saved.Content, saved.Revision)
	}
	if len(saved.SuppressedAgentIds) != 0 {
		t.Fatalf("the latest edit starts A, but an earlier edit's choice stayed: %v", saved.SuppressedAgentIds)
	}
}