From ecc70df2f927f9547acc490f6da4bb52fe17b429 Mon Sep 17 00:00:00 2001 From: Martin Montes Date: Thu, 5 Nov 2026 14:40:51 +0000 Subject: [PATCH 1/2] feat(agents): move queued tasks up/down in the agent task queue Adds POST /api/tasks/{taskId}/move so a member can hand-reorder one agent's queued tasks one step at a time in claim order (priority DESC, created_at ASC, id ASC). The move runs under the agent's claim lock, reuses the CancelTaskByUser tenancy and privacy gates, and broadcasts the existing task:queued event so clients re-sort live. Co-authored-by: multica-agent --- server/cmd/server/router.go | 1 + server/internal/handler/task_lifecycle.go | 147 +++++++++ server/internal/handler/task_move_test.go | 365 ++++++++++++++++++++++ server/pkg/db/generated/agent.sql.go | 95 ++++++ server/pkg/db/queries/agent.sql | 69 ++++ 5 files changed, 677 insertions(+) create mode 100644 server/internal/handler/task_move_test.go diff --git a/server/cmd/server/router.go b/server/cmd/server/router.go index f92744e8048..9cb8b002e8f 100644 --- a/server/cmd/server/router.go +++ b/server/cmd/server/router.go @@ -2495,6 +2495,7 @@ func NewRouterWithOptions(pool *pgxpool.Pool, hub *realtime.Hub, bus *events.Bus // Tasks (user-facing, with ownership check) r.Post("/api/tasks/{taskId}/cancel", h.CancelTaskByUser) r.Post("/api/tasks/{taskId}/rerun", h.RerunTaskByUser) + r.Post("/api/tasks/{taskId}/move", h.MoveTaskByUser) // Workspace-wide agent task snapshot for presence derivation: // every active task + each agent's most recent terminal task. diff --git a/server/internal/handler/task_lifecycle.go b/server/internal/handler/task_lifecycle.go index 7b837b8eb8f..4ef1a97529b 100644 --- a/server/internal/handler/task_lifecycle.go +++ b/server/internal/handler/task_lifecycle.go @@ -313,3 +313,150 @@ func (h *Handler) RetrySourceContextQuickCreate(w http.ResponseWriter, r *http.R h.hydrateTaskAttributions(r.Context(), []*TaskAttribution{resp.Attribution}) writeJSON(w, http.StatusAccepted, resp) } + +// MoveTaskResponse reports the outcome of a manual queue reorder. Moved is +// false when the task was already at the edge of its agent's queue, or could +// not gain a position within the coarse 0–4 priority scale; Priority is then +// unchanged. +type MoveTaskResponse struct { + TaskID string `json:"task_id"` + Priority int32 `json:"priority"` + Moved bool `json:"moved"` +} + +// MoveTaskByUser moves one queued task one step up or down in its agent's +// claim order (priority DESC, created_at ASC, id ASC — the order +// ClaimAgentTask uses), letting a member hand-reorder the queue. +// +// Tenancy and privacy mirror CancelTaskByUser exactly: the task is loaded +// through GetAgentTaskInWorkspace (the single tenant guard regardless of which +// optional source FK is set), a chat task is private to the member who opened +// the conversation, and every other task passes canAccessPrivateAgent so the +// id-only endpoint is never more permissive than the Activity tab that +// exposes it. The move runs under the agent's claim lock (the same row lock +// ClaimTask takes before selecting work), so the reorder cannot race a claim +// in flight; the broadcast on commit lets every client re-sort its queue from +// the existing task:queued event. +func (h *Handler) MoveTaskByUser(w http.ResponseWriter, r *http.Request) { + userID, ok := requireUserID(w, r) + if !ok { + return + } + workspaceID := ctxWorkspaceID(r.Context()) + wsUUID, ok := parseUUIDOrBadRequest(w, workspaceID, "workspace id") + if !ok { + return + } + taskID, ok := parseUUIDOrBadRequest(w, chi.URLParam(r, "taskId"), "task id") + if !ok { + return + } + var req struct { + Direction string `json:"direction"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil && err != io.EOF { + writeError(w, http.StatusBadRequest, "invalid request body") + return + } + if req.Direction != "up" && req.Direction != "down" { + writeError(w, http.StatusBadRequest, "direction must be up or down") + return + } + + task, err := h.Queries.GetAgentTaskInWorkspace(r.Context(), db.GetAgentTaskInWorkspaceParams{ + ID: taskID, + WorkspaceID: wsUUID, + }) + if err != nil { + writeError(w, http.StatusNotFound, "task not found") + return + } + + if task.ChatSessionID.Valid { + // Chat privacy: only the member who opened the conversation may + // reorder its tasks, even though the workspace is shared. + cs, err := h.Queries.GetChatSessionInWorkspace(r.Context(), db.GetChatSessionInWorkspaceParams{ + ID: task.ChatSessionID, + WorkspaceID: wsUUID, + }) + if err != nil { + writeError(w, http.StatusNotFound, "task not found") + return + } + if uuidToString(cs.CreatorID) != userID { + writeError(w, http.StatusForbidden, "not your task") + return + } + } else { + // Issue / autopilot / quick_create tasks all surface on the agent + // Activity tab, which gates private agents. Mirror that gate here. + agent, err := h.Queries.GetAgentInWorkspace(r.Context(), db.GetAgentInWorkspaceParams{ + ID: task.AgentID, + WorkspaceID: wsUUID, + }) + if err != nil { + writeError(w, http.StatusNotFound, "task not found") + return + } + actorType, actorID := h.resolveActor(r, userID, workspaceID) + if !h.canAccessPrivateAgent(r.Context(), agent, actorType, actorID, workspaceID) { + writeError(w, http.StatusForbidden, "you do not have access to this agent") + return + } + } + + tx, err := h.TxStarter.Begin(r.Context()) + if err != nil { + writeError(w, http.StatusInternalServerError, "failed to start move transaction") + return + } + defer tx.Rollback(r.Context()) + qtx := h.Queries.WithTx(tx) + + // ClaimTask takes this same row lock before selecting work. Holding it + // here makes "reorder" one server-authoritative decision instead of a + // priority write racing an in-flight claim. + if _, err := qtx.GetAgentForClaimUpdate(r.Context(), task.AgentID); err != nil { + writeError(w, http.StatusInternalServerError, "failed to lock agent") + return + } + + var ( + movedID pgtype.UUID + movedPrio int32 + moveErr error + ) + if req.Direction == "up" { + var row db.MoveQueuedTaskUpRow + row, moveErr = qtx.MoveQueuedTaskUp(r.Context(), taskID) + movedID, movedPrio = row.ID, row.Priority + } else { + var row db.MoveQueuedTaskDownRow + row, moveErr = qtx.MoveQueuedTaskDown(r.Context(), taskID) + movedID, movedPrio = row.ID, row.Priority + } + if errors.Is(moveErr, pgx.ErrNoRows) { + writeError(w, http.StatusConflict, "task is no longer queued") + return + } + if moveErr != nil { + writeError(w, http.StatusInternalServerError, "failed to move queued task") + return + } + queuedTask, err := qtx.GetAgentTask(r.Context(), taskID) + if err != nil { + writeError(w, http.StatusInternalServerError, "failed to load moved task") + return + } + if err := tx.Commit(r.Context()); err != nil { + writeError(w, http.StatusInternalServerError, "failed to commit queued task move") + return + } + h.TaskService.BroadcastTaskQueued(r.Context(), queuedTask) + + writeJSON(w, http.StatusOK, MoveTaskResponse{ + TaskID: uuidToString(movedID), + Priority: movedPrio, + Moved: movedPrio != task.Priority, + }) +} diff --git a/server/internal/handler/task_move_test.go b/server/internal/handler/task_move_test.go new file mode 100644 index 00000000000..bb596bee3b3 --- /dev/null +++ b/server/internal/handler/task_move_test.go @@ -0,0 +1,365 @@ +package handler + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/multica-ai/multica/server/internal/testutil" +) + +// MoveTaskByUser (POST /api/tasks/{taskId}/move) reorders one queued task one +// step up/down in its agent's claim order (priority DESC, created_at ASC, +// id ASC). These tests pin the reorder semantics on the coarse 0–4 priority +// scale (take the neighbor's level across levels, step one level on a +// same-level tie, no-op at the queue edge) and the tenancy/privacy gates +// copied from CancelTaskByUser. + +// moveTaskByUserRequest builds a /move request the way the router delivers +// one: member actor, JSON body, taskId URL param, test workspace context. +func moveTaskByUserRequest(t *testing.T, userID, taskID, direction string) *http.Request { + t.Helper() + req := newRequestAs(userID, http.MethodPost, "/api/tasks/"+taskID+"/move", map[string]string{ + "direction": direction, + }) + req = withURLParam(req, "taskId", taskID) + return withChatTestWorkspaceCtx(t, req) +} + +func moveTaskPriority(t *testing.T, taskID string) int { + t.Helper() + var priority int + dbfx.QueryRow(t, + `SELECT priority FROM agent_task_queue WHERE id = $1`, taskID, + ).Scan(&priority) + return priority +} + +func decodeMoveTaskResponse(t *testing.T, w *httptest.ResponseRecorder) MoveTaskResponse { + t.Helper() + var response MoveTaskResponse + if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil { + t.Fatalf("decode move task response: %v", err) + } + return response +} + +// seedMoveQueueTask queues a task for agentID at an explicit priority and +// created_at so the claim order (priority DESC, created_at ASC, id ASC) is +// fully deterministic — gen_random_uuid() ids make the id tie-break +// arbitrary, and now() timestamps can collide within one test. +func seedMoveQueueTask(t *testing.T, agentID string, priority int, createdOffset string) string { + t.Helper() + return dbfx.Task(t, agentID, testutil.Cols{ + "runtime_id": handlerTestRuntimeID(t), + "priority": priority, + "created_at": testutil.Raw("now() - interval '" + createdOffset + "'"), + }) +} + +func TestMoveTaskByUser_UpStepsTowardFront(t *testing.T) { + if testHandler == nil { + t.Skip("database not available") + } + + agentID := createHandlerTestAgent(t, "MoveUpAgent", []byte("[]")) + // Claim order: A (p2) → B (p1) → C (p0). + seedMoveQueueTask(t, agentID, 2, "3 seconds") + taskB := seedMoveQueueTask(t, agentID, 1, "2 seconds") + taskC := seedMoveQueueTask(t, agentID, 0, "1 second") + + // C takes its neighbor B's level: 0 → 1. + w := httptest.NewRecorder() + testHandler.MoveTaskByUser(w, moveTaskByUserRequest(t, testUserID, taskC, "up")) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + response := decodeMoveTaskResponse(t, w) + if response.TaskID != taskC || !response.Moved { + t.Fatalf("response = %+v, want task %s moved", response, taskC) + } + if response.Priority != 1 || moveTaskPriority(t, taskC) != 1 { + t.Fatalf("task C priority = %d (db %d), want 1", response.Priority, moveTaskPriority(t, taskC)) + } + + // B takes its neighbor A's level: 1 → 2. + w = httptest.NewRecorder() + testHandler.MoveTaskByUser(w, moveTaskByUserRequest(t, testUserID, taskB, "up")) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + if got := decodeMoveTaskResponse(t, w); !got.Moved || got.Priority != 2 { + t.Fatalf("response = %+v, want moved to priority 2", got) + } + + // A is at the front: no-op, moved=false, priority unchanged. + var frontID string + dbfx.QueryRow(t, `SELECT id FROM agent_task_queue WHERE agent_id = $1 AND priority = 2 ORDER BY created_at ASC LIMIT 1`, agentID).Scan(&frontID) + w = httptest.NewRecorder() + testHandler.MoveTaskByUser(w, moveTaskByUserRequest(t, testUserID, frontID, "up")) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + if got := decodeMoveTaskResponse(t, w); got.Moved || got.Priority != 2 { + t.Fatalf("front-of-queue move = %+v, want moved=false at priority 2", got) + } +} + +func TestMoveTaskByUser_UpSameLevelStepsOneLevel(t *testing.T) { + if testHandler == nil { + t.Skip("database not available") + } + + agentID := createHandlerTestAgent(t, "MoveUpSameLevelAgent", []byte("[]")) + seedMoveQueueTask(t, agentID, 1, "2 seconds") + taskB := seedMoveQueueTask(t, agentID, 1, "1 second") + + // Same-level peer ahead: bump one level, 1 → 2. + w := httptest.NewRecorder() + testHandler.MoveTaskByUser(w, moveTaskByUserRequest(t, testUserID, taskB, "up")) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + if got := decodeMoveTaskResponse(t, w); !got.Moved || got.Priority != 2 { + t.Fatalf("response = %+v, want moved to priority 2", got) + } + if got := moveTaskPriority(t, taskB); got != 2 { + t.Fatalf("db priority = %d, want 2", got) + } +} + +func TestMoveTaskByUser_UpClampedAtTop(t *testing.T) { + if testHandler == nil { + t.Skip("database not available") + } + + agentID := createHandlerTestAgent(t, "MoveUpClampAgent", []byte("[]")) + seedMoveQueueTask(t, agentID, 4, "2 seconds") + taskB := seedMoveQueueTask(t, agentID, 4, "1 second") + + // Already at the top level with a same-level peer ahead: nothing to give + // it — no-op, moved=false. + w := httptest.NewRecorder() + testHandler.MoveTaskByUser(w, moveTaskByUserRequest(t, testUserID, taskB, "up")) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + if got := decodeMoveTaskResponse(t, w); got.Moved || got.Priority != 4 { + t.Fatalf("clamped move = %+v, want moved=false at priority 4", got) + } + if got := moveTaskPriority(t, taskB); got != 4 { + t.Fatalf("db priority = %d, want 4", got) + } +} + +func TestMoveTaskByUser_DownStepsTowardBack(t *testing.T) { + if testHandler == nil { + t.Skip("database not available") + } + + agentID := createHandlerTestAgent(t, "MoveDownAgent", []byte("[]")) + // Claim order: A (p2) → B (p1) → C (p0). + seedMoveQueueTask(t, agentID, 2, "3 seconds") + taskB := seedMoveQueueTask(t, agentID, 1, "2 seconds") + seedMoveQueueTask(t, agentID, 0, "1 second") + + // B takes its neighbor C's level: 1 → 0, falling behind C (older). + w := httptest.NewRecorder() + testHandler.MoveTaskByUser(w, moveTaskByUserRequest(t, testUserID, taskB, "down")) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + if got := decodeMoveTaskResponse(t, w); !got.Moved || got.Priority != 0 { + t.Fatalf("response = %+v, want moved to priority 0", got) + } + if got := moveTaskPriority(t, taskB); got != 0 { + t.Fatalf("db priority = %d, want 0", got) + } + + // C is at the back: no-op, moved=false, priority unchanged. + var backID string + dbfx.QueryRow(t, `SELECT id FROM agent_task_queue WHERE agent_id = $1 AND priority = 0 ORDER BY created_at DESC LIMIT 1`, agentID).Scan(&backID) + w = httptest.NewRecorder() + testHandler.MoveTaskByUser(w, moveTaskByUserRequest(t, testUserID, backID, "down")) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + if got := decodeMoveTaskResponse(t, w); got.Moved || got.Priority != 0 { + t.Fatalf("back-of-queue move = %+v, want moved=false at priority 0", got) + } +} + +func TestMoveTaskByUser_DownSameLevelStepsOneLevel(t *testing.T) { + if testHandler == nil { + t.Skip("database not available") + } + + agentID := createHandlerTestAgent(t, "MoveDownSameLevelAgent", []byte("[]")) + taskA := seedMoveQueueTask(t, agentID, 1, "2 seconds") + seedMoveQueueTask(t, agentID, 1, "1 second") + + // Same-level peer behind: drop one level, 1 → 0. + w := httptest.NewRecorder() + testHandler.MoveTaskByUser(w, moveTaskByUserRequest(t, testUserID, taskA, "down")) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + if got := decodeMoveTaskResponse(t, w); !got.Moved || got.Priority != 0 { + t.Fatalf("response = %+v, want moved to priority 0", got) + } +} + +func TestMoveTaskByUser_DownClampedAtBottom(t *testing.T) { + if testHandler == nil { + t.Skip("database not available") + } + + agentID := createHandlerTestAgent(t, "MoveDownClampAgent", []byte("[]")) + taskA := seedMoveQueueTask(t, agentID, 0, "2 seconds") + seedMoveQueueTask(t, agentID, 0, "1 second") + + // Already at the bottom level with a same-level peer behind: no-op. + w := httptest.NewRecorder() + testHandler.MoveTaskByUser(w, moveTaskByUserRequest(t, testUserID, taskA, "down")) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + if got := decodeMoveTaskResponse(t, w); got.Moved || got.Priority != 0 { + t.Fatalf("clamped move = %+v, want moved=false at priority 0", got) + } +} + +func TestMoveTaskByUser_NonQueuedTaskConflicts(t *testing.T) { + if testHandler == nil { + t.Skip("database not available") + } + + agentID := createHandlerTestAgent(t, "MoveNonQueuedAgent", []byte("[]")) + runningID := dbfx.Task(t, agentID, testutil.Cols{ + "runtime_id": handlerTestRuntimeID(t), + "status": "running", + }) + + w := httptest.NewRecorder() + testHandler.MoveTaskByUser(w, moveTaskByUserRequest(t, testUserID, runningID, "up")) + if w.Code != http.StatusConflict { + t.Fatalf("expected 409, got %d: %s", w.Code, w.Body.String()) + } +} + +func TestMoveTaskByUser_UnknownTaskNotFound(t *testing.T) { + if testHandler == nil { + t.Skip("database not available") + } + + w := httptest.NewRecorder() + testHandler.MoveTaskByUser(w, moveTaskByUserRequest(t, testUserID, "00000000-0000-0000-0000-00000000dead", "up")) + if w.Code != http.StatusNotFound { + t.Fatalf("expected 404, got %d: %s", w.Code, w.Body.String()) + } +} + +func TestMoveTaskByUser_ForeignWorkspaceTaskNotFound(t *testing.T) { + if testHandler == nil { + t.Skip("database not available") + } + + agentID := createForeignWorkspaceAgent(t) + taskID := createAutopilotRunOnlyTask(t, agentID) + + w := httptest.NewRecorder() + testHandler.MoveTaskByUser(w, moveTaskByUserRequest(t, testUserID, taskID, "up")) + if w.Code != http.StatusNotFound { + t.Fatalf("expected 404, got %d: %s", w.Code, w.Body.String()) + } +} + +func TestMoveTaskByUser_BadDirectionRejected(t *testing.T) { + if testHandler == nil { + t.Skip("database not available") + } + + agentID := createHandlerTestAgent(t, "MoveBadDirectionAgent", []byte("[]")) + taskID := seedMoveQueueTask(t, agentID, 0, "1 second") + + w := httptest.NewRecorder() + testHandler.MoveTaskByUser(w, moveTaskByUserRequest(t, testUserID, taskID, "left")) + if w.Code != http.StatusBadRequest { + t.Fatalf("expected 400, got %d: %s", w.Code, w.Body.String()) + } + + w = httptest.NewRecorder() + testHandler.MoveTaskByUser(w, moveTaskByUserRequest(t, testUserID, taskID, "")) + if w.Code != http.StatusBadRequest { + t.Fatalf("expected 400 for missing direction, got %d: %s", w.Code, w.Body.String()) + } + if got := moveTaskPriority(t, taskID); got != 0 { + t.Fatalf("rejected move changed priority to %d", got) + } +} + +func TestMoveTaskByUser_PrivateAgentGate(t *testing.T) { + if testHandler == nil { + t.Skip("database not available") + } + + agentID, ownerID, memberID := privateAgentTestFixture(t) + seedMoveQueueTask(t, agentID, 0, "1 second") + + // A plain member without access to the private agent is refused. + w := httptest.NewRecorder() + testHandler.MoveTaskByUser(w, moveTaskByUserRequest(t, memberID, firstTaskForAgent(t, agentID), "up")) + if w.Code != http.StatusForbidden { + t.Fatalf("plain member: expected 403, got %d: %s", w.Code, w.Body.String()) + } + + // The agent owner is admitted. + w = httptest.NewRecorder() + testHandler.MoveTaskByUser(w, moveTaskByUserRequest(t, ownerID, firstTaskForAgent(t, agentID), "up")) + if w.Code != http.StatusOK { + t.Fatalf("agent owner: expected 200, got %d: %s", w.Code, w.Body.String()) + } +} + +func TestMoveTaskByUser_ChatTaskCreatorOnly(t *testing.T) { + if testHandler == nil { + t.Skip("database not available") + } + + agentID := createHandlerTestAgent(t, "MoveChatAgent", []byte("[]")) + sessionID := createHandlerTestChatSession(t, agentID) + seedMoveQueueTask(t, agentID, 0, "2 seconds") + taskB := insertPendingChatTask(t, agentID, sessionID, "queued") + + // The session creator (testUserID) may reorder their own chat task. + w := httptest.NewRecorder() + testHandler.MoveTaskByUser(w, moveTaskByUserRequest(t, testUserID, taskB, "up")) + if w.Code != http.StatusOK { + t.Fatalf("creator: expected 200, got %d: %s", w.Code, w.Body.String()) + } + if got := decodeMoveTaskResponse(t, w); !got.Moved || got.Priority != 1 { + t.Fatalf("creator response = %+v, want moved to priority 1", got) + } + + // A plain workspace member is refused. + memberID := createWorkspaceMemberUser(t, "Move Chat Member", "move-chat-member@multica.test") + w = httptest.NewRecorder() + testHandler.MoveTaskByUser(w, moveTaskByUserRequest(t, memberID, taskB, "up")) + if w.Code != http.StatusForbidden { + t.Fatalf("non-creator: expected 403, got %d: %s", w.Code, w.Body.String()) + } + if got := moveTaskPriority(t, taskB); got != 1 { + t.Fatalf("refused move changed priority to %d", got) + } +} + +// firstTaskForAgent returns the oldest queued task id for agentID. +func firstTaskForAgent(t *testing.T, agentID string) string { + t.Helper() + var taskID string + dbfx.QueryRow(t, + `SELECT id FROM agent_task_queue WHERE agent_id = $1 ORDER BY created_at ASC LIMIT 1`, agentID, + ).Scan(&taskID) + return taskID +} diff --git a/server/pkg/db/generated/agent.sql.go b/server/pkg/db/generated/agent.sql.go index 24c6d38e385..d1677913691 100644 --- a/server/pkg/db/generated/agent.sql.go +++ b/server/pkg/db/generated/agent.sql.go @@ -7368,6 +7368,101 @@ func (q *Queries) MergeDelegatedFailureCommentIntoPendingTask(ctx context.Contex return i, err } +const moveQueuedTaskDown = `-- name: MoveQueuedTaskDown :one +WITH target AS ( + SELECT atq.id, atq.agent_id, atq.priority, atq.created_at + FROM agent_task_queue atq + WHERE atq.id = $1 AND atq.status = 'queued' +), +neighbor AS ( + SELECT p.priority + FROM agent_task_queue p + JOIN target t ON p.agent_id = t.agent_id + WHERE p.status = 'queued' + AND p.id <> t.id + AND (-p.priority, p.created_at, p.id) > (-t.priority, t.created_at, t.id) + ORDER BY (-p.priority, p.created_at, p.id) ASC + LIMIT 1 +) +UPDATE agent_task_queue atq +SET priority = CASE + WHEN neighbor.priority IS NULL THEN atq.priority + WHEN neighbor.priority < atq.priority THEN neighbor.priority + ELSE GREATEST(atq.priority - 1, 0) + END +FROM target +LEFT JOIN neighbor ON TRUE +WHERE atq.id = target.id +RETURNING atq.id, atq.priority +` + +type MoveQueuedTaskDownRow struct { + ID pgtype.UUID `json:"id"` + Priority int32 `json:"priority"` +} + +// Move one queued task one step toward the back of its agent's claim order, +// mirroring MoveQueuedTaskUp: take the neighbor's level when it is lower, +// otherwise drop the task's own level by one (clamped at 0). No neighbor +// means the task is already last — unchanged, moved=false. +func (q *Queries) MoveQueuedTaskDown(ctx context.Context, id pgtype.UUID) (MoveQueuedTaskDownRow, error) { + row := q.db.QueryRow(ctx, moveQueuedTaskDown, id) + var i MoveQueuedTaskDownRow + err := row.Scan(&i.ID, &i.Priority) + return i, err +} + +const moveQueuedTaskUp = `-- name: MoveQueuedTaskUp :one +WITH target AS ( + SELECT atq.id, atq.agent_id, atq.priority, atq.created_at + FROM agent_task_queue atq + WHERE atq.id = $1 AND atq.status = 'queued' +), +neighbor AS ( + SELECT p.priority + FROM agent_task_queue p + JOIN target t ON p.agent_id = t.agent_id + WHERE p.status = 'queued' + AND p.id <> t.id + AND (-p.priority, p.created_at, p.id) < (-t.priority, t.created_at, t.id) + ORDER BY (-p.priority, p.created_at, p.id) DESC + LIMIT 1 +) +UPDATE agent_task_queue atq +SET priority = CASE + WHEN neighbor.priority IS NULL THEN atq.priority + WHEN neighbor.priority > atq.priority THEN neighbor.priority + ELSE LEAST(atq.priority + 1, 4) + END +FROM target +LEFT JOIN neighbor ON TRUE +WHERE atq.id = target.id +RETURNING atq.id, atq.priority +` + +type MoveQueuedTaskUpRow struct { + ID pgtype.UUID `json:"id"` + Priority int32 `json:"priority"` +} + +// Move one queued task one step toward the front of its agent's claim order — +// priority DESC, created_at ASC, id ASC, the same order ClaimAgentTask uses. +// The caller holds the agent's claim lock (GetAgentForClaimUpdate), which +// serializes this against claims and other moves, so a plain +// read-compute-update is race-free. +// +// If a queued peer sorts ahead of the task, the task takes that neighbor's +// level when it is higher, or bumps its own level by one (clamped at 4) to +// clear a same-level peer. The 0–4 priority scale is coarse, so a bump can +// clear several lower levels at once. If nothing sorts ahead, the task is +// already first: it comes back unchanged and the caller reports moved=false. +func (q *Queries) MoveQueuedTaskUp(ctx context.Context, id pgtype.UUID) (MoveQueuedTaskUpRow, error) { + row := q.db.QueryRow(ctx, moveQueuedTaskUp, id) + var i MoveQueuedTaskUpRow + err := row.Scan(&i.ID, &i.Priority) + return i, err +} + const nextDeferredTaskFireAtForRuntimes = `-- name: NextDeferredTaskFireAtForRuntimes :one SELECT MIN(fire_at)::timestamptz FROM agent_task_queue t diff --git a/server/pkg/db/queries/agent.sql b/server/pkg/db/queries/agent.sql index decbb1e6cc7..a5000de3978 100644 --- a/server/pkg/db/queries/agent.sql +++ b/server/pkg/db/queries/agent.sql @@ -820,6 +820,75 @@ WHERE id = ( ) RETURNING *; +-- name: MoveQueuedTaskUp :one +-- Move one queued task one step toward the front of its agent's claim order — +-- priority DESC, created_at ASC, id ASC, the same order ClaimAgentTask uses. +-- The caller holds the agent's claim lock (GetAgentForClaimUpdate), which +-- serializes this against claims and other moves, so a plain +-- read-compute-update is race-free. +-- +-- If a queued peer sorts ahead of the task, the task takes that neighbor's +-- level when it is higher, or bumps its own level by one (clamped at 4) to +-- clear a same-level peer. The 0–4 priority scale is coarse, so a bump can +-- clear several lower levels at once. If nothing sorts ahead, the task is +-- already first: it comes back unchanged and the caller reports moved=false. +WITH target AS ( + SELECT atq.id, atq.agent_id, atq.priority, atq.created_at + FROM agent_task_queue atq + WHERE atq.id = $1 AND atq.status = 'queued' +), +neighbor AS ( + SELECT p.priority + FROM agent_task_queue p + JOIN target t ON p.agent_id = t.agent_id + WHERE p.status = 'queued' + AND p.id <> t.id + AND (-p.priority, p.created_at, p.id) < (-t.priority, t.created_at, t.id) + ORDER BY (-p.priority, p.created_at, p.id) DESC + LIMIT 1 +) +UPDATE agent_task_queue atq +SET priority = CASE + WHEN neighbor.priority IS NULL THEN atq.priority + WHEN neighbor.priority > atq.priority THEN neighbor.priority + ELSE LEAST(atq.priority + 1, 4) + END +FROM target +LEFT JOIN neighbor ON TRUE +WHERE atq.id = target.id +RETURNING atq.id, atq.priority; + +-- name: MoveQueuedTaskDown :one +-- Move one queued task one step toward the back of its agent's claim order, +-- mirroring MoveQueuedTaskUp: take the neighbor's level when it is lower, +-- otherwise drop the task's own level by one (clamped at 0). No neighbor +-- means the task is already last — unchanged, moved=false. +WITH target AS ( + SELECT atq.id, atq.agent_id, atq.priority, atq.created_at + FROM agent_task_queue atq + WHERE atq.id = $1 AND atq.status = 'queued' +), +neighbor AS ( + SELECT p.priority + FROM agent_task_queue p + JOIN target t ON p.agent_id = t.agent_id + WHERE p.status = 'queued' + AND p.id <> t.id + AND (-p.priority, p.created_at, p.id) > (-t.priority, t.created_at, t.id) + ORDER BY (-p.priority, p.created_at, p.id) ASC + LIMIT 1 +) +UPDATE agent_task_queue atq +SET priority = CASE + WHEN neighbor.priority IS NULL THEN atq.priority + WHEN neighbor.priority < atq.priority THEN neighbor.priority + ELSE GREATEST(atq.priority - 1, 0) + END +FROM target +LEFT JOIN neighbor ON TRUE +WHERE atq.id = target.id +RETURNING atq.id, atq.priority; + -- name: SetTaskDeliveredCommentIDs :one -- Replace (rather than append to) the delivery receipt for this claim. A stale -- dispatched task may be reclaimed by a daemon with different capabilities, From 38c20903d21af2a4de306e5a486bfeabfb421a92 Mon Sep 17 00:00:00 2001 From: Martin Montes Date: Thu, 5 Nov 2026 16:32:26 +0000 Subject: [PATCH 2/2] feat(agents): reorder queued runs from the activity tab Adds hover chevron actions to queued rows in the agent activity tab that call the new /api/tasks/{taskId}/move endpoint. Queued rows now sort in claim order (priority DESC, created_at ASC) so the row order matches what the daemon claims next; reorders propagate through the existing task:queued WS event. Co-authored-by: multica-agent --- packages/core/api/client.ts | 19 +++++ packages/core/api/schemas.ts | 13 +++ packages/core/types/agent.ts | 9 ++ packages/core/types/index.ts | 1 + .../agents/components/tabs/activity-tab.tsx | 85 +++++++++++++++++-- packages/views/locales/en/agents.json | 7 ++ packages/views/locales/es/agents.json | 7 ++ packages/views/locales/fr/agents.json | 7 ++ packages/views/locales/ja/agents.json | 7 ++ packages/views/locales/ko/agents.json | 7 ++ packages/views/locales/zh-Hans/agents.json | 7 ++ packages/views/locales/zh-Hant/agents.json | 7 ++ 12 files changed, 171 insertions(+), 5 deletions(-) diff --git a/packages/core/api/client.ts b/packages/core/api/client.ts index 5f8aa547531..890a71935d4 100644 --- a/packages/core/api/client.ts +++ b/packages/core/api/client.ts @@ -105,6 +105,7 @@ import type { ChatDraftRestoresResponse, ChatPendingTask, PrioritizeQueuedChatTaskResponse, + MoveTaskResponse, PendingChatTasksResponse, HasPendingChatTasksResponse, SendChatMessageResponse, @@ -276,6 +277,7 @@ import { ChatSessionListSchema, ChatSessionSchema, PrioritizeQueuedChatTaskResponseSchema, + MoveTaskResponseSchema, SendChatMessageResponseSchema, StartMikaOnboardingResponseSchema, ChildIssuesResponseSchema, @@ -304,6 +306,7 @@ import { EMPTY_CHAT_SESSION, EMPTY_CHAT_SESSION_LIST, EMPTY_PRIORITIZE_QUEUED_CHAT_TASK_RESPONSE, + EMPTY_MOVE_TASK_RESPONSE, EMPTY_CLOUD_RUNTIME_NODE, EMPTY_CLOUD_RUNTIME_NODE_LIST, EMPTY_AGENT_BUILDER_SESSION, @@ -4133,6 +4136,22 @@ export class ApiClient { }); } + /** + * Move one queued task a step up or down in its agent's claim order. + * moved=false means the task was already at the edge of the queue (or + * could not gain a position within the 0–4 priority scale); the queue is + * then unchanged. + */ + async moveQueuedTask(taskId: string, direction: "up" | "down"): Promise { + const raw = await this.fetch(`/api/tasks/${taskId}/move`, { + method: "POST", + body: JSON.stringify({ direction }), + }); + return parseWithFallback(raw, MoveTaskResponseSchema, EMPTY_MOVE_TASK_RESPONSE, { + endpoint: "POST /api/tasks/{taskId}/move", + }); + } + async listAttachments(issueId: string): Promise { return this.fetch(`/api/issues/${issueId}/attachments`); } diff --git a/packages/core/api/schemas.ts b/packages/core/api/schemas.ts index 40cf1283a0a..9ef63a14256 100644 --- a/packages/core/api/schemas.ts +++ b/packages/core/api/schemas.ts @@ -23,6 +23,7 @@ import type { ChatPendingTask, ChatSession, PrioritizeQueuedChatTaskResponse, + MoveTaskResponse, SendChatMessageResponse, StartMikaOnboardingResponse, Comment, @@ -2326,6 +2327,18 @@ export const PrioritizeQueuedChatTaskResponseSchema: export const EMPTY_PRIORITIZE_QUEUED_CHAT_TASK_RESPONSE: PrioritizeQueuedChatTaskResponse = { task_id: "" }; +export const MoveTaskResponseSchema: z.ZodType = z.object({ + task_id: z.string(), + priority: z.number(), + moved: z.boolean(), +}).loose(); + +export const EMPTY_MOVE_TASK_RESPONSE: MoveTaskResponse = { + task_id: "", + priority: 0, + moved: false, +}; + export const EMPTY_CHAT_DRAFT_RESTORES: ChatDraftRestoresResponse = { restores: [], }; diff --git a/packages/core/types/agent.ts b/packages/core/types/agent.ts index 0565b051adc..f4b488505d1 100644 --- a/packages/core/types/agent.ts +++ b/packages/core/types/agent.ts @@ -1390,3 +1390,12 @@ export interface RuntimeLocalSkillImportResult { skill?: Skill; conflict?: RuntimeLocalSkillImportConflict; } + +export interface MoveTaskResponse { + task_id: string; + priority: number; + /** False when the task was already at the edge of its agent's queue, or + * could not gain a position within the 0–4 priority scale; priority is + * then unchanged. */ + moved: boolean; +} diff --git a/packages/core/types/index.ts b/packages/core/types/index.ts index adadfe9ce53..2d2bdd964ea 100644 --- a/packages/core/types/index.ts +++ b/packages/core/types/index.ts @@ -111,6 +111,7 @@ export type { RuntimeLocalSkillImportResult, IssueUsageSummary, MikaBootstrapResponse, + MoveTaskResponse, } from "./agent"; export { RUNTIME_PROFILE_PROTOCOL_FAMILIES, RUNTIME_PROFILE_RUNTIME_TYPES } from "./agent"; export type { Workspace, WorkspaceRepo, WorkspaceMcpServer, Member, MemberRole, User, MemberWithUser, Invitation, ShareLink, ShareLinkInfo } from "./workspace"; diff --git a/packages/views/agents/components/tabs/activity-tab.tsx b/packages/views/agents/components/tabs/activity-tab.tsx index dfb50327e4a..ade151cc07a 100644 --- a/packages/views/agents/components/tabs/activity-tab.tsx +++ b/packages/views/agents/components/tabs/activity-tab.tsx @@ -4,6 +4,8 @@ import type { ReactNode } from "react"; import { forwardRef, memo, useCallback, useMemo, useState } from "react"; import { ArrowUpRight, + ChevronDown, + ChevronUp, CircleHelp, Hash, MessageSquare, @@ -145,11 +147,20 @@ export function AgentWorkSections({ t.status === "dispatched" || t.status === "waiting_local_directory"), ) - .sort( - (a, b) => - (statusRank[a.status] ?? 99) - (statusRank[b.status] ?? 99) || - new Date(a.created_at).getTime() - new Date(b.created_at).getTime(), - ); + .sort((a, b) => { + const rank = (statusRank[a.status] ?? 99) - (statusRank[b.status] ?? 99); + if (rank !== 0) return rank; + // Queued rows follow the server's claim order (priority DESC, + // created_at ASC) so the reorder buttons move a row toward or away + // from what the daemon will claim next. + if (a.status === "queued" && b.status === "queued") { + return ( + b.priority - a.priority || + new Date(a.created_at).getTime() - new Date(b.created_at).getTime() + ); + } + return new Date(a.created_at).getTime() - new Date(b.created_at).getTime(); + }); }, [snapshot, agent.id]); // Most recent terminal tasks. Includes cancelled — users searching @@ -627,6 +638,7 @@ function LiveTaskRow({ const timeAgo = useTimeAgo(); const paths = useWorkspacePaths(); const [cancelling, setCancelling] = useState(false); + const [moving, setMoving] = useState(false); const { cfg, hasIssue, issue, title, SourceIcon, sourceLabel, sourceFallback, statusLabel } = useTaskRowModel(task); const Icon = cfg.icon; @@ -638,6 +650,33 @@ function LiveTaskRow({ task.status === "queued" || task.status === "dispatched" || task.status === "running"; + // Only queued rows are reorderable — dispatched/running work is already + // claimed and its position is no longer the queue's to change. + const showMove = task.status === "queued"; + + const handleMove = async (direction: "up" | "down") => { + if (moving) return; + setMoving(true); + try { + const result = await api.moveQueuedTask(task.id, direction); + if (!result.moved) { + toast.info( + t(($) => + direction === "up" + ? $.tab_body.activity.move_front_toast + : $.tab_body.activity.move_back_toast, + ), + ); + } + // No manual invalidate needed — the task:queued WS event flows + // through useRealtimeSync's `task:` prefix path which already + // invalidates the snapshot. + } catch (e) { + toast.error(e instanceof Error ? e.message : t(($) => $.tab_body.activity.move_failed_toast)); + } finally { + setMoving(false); + } + }; const handleCancel = async () => { if (cancelling) return; @@ -687,6 +726,42 @@ function LiveTaskRow({ {activeTaskTimeText(task, t, timeAgo)}
+ {showMove && ( + <> + + void handleMove("up")} + disabled={moving} + aria-label={t(($) => $.tab_body.activity.move_up_aria)} + /> + } + className={`flex items-center justify-center rounded-xs p-1 text-muted-foreground transition-colors hover:bg-accent/50 hover:text-foreground disabled:cursor-not-allowed disabled:opacity-50 ${TOUCH_ACTION_CLASS}`} + > + + {t(($) => $.tab_body.activity.move_up_tooltip)} + + + void handleMove("down")} + disabled={moving} + aria-label={t(($) => $.tab_body.activity.move_down_aria)} + /> + } + className={`flex items-center justify-center rounded-xs p-1 text-muted-foreground transition-colors hover:bg-accent/50 hover:text-foreground disabled:cursor-not-allowed disabled:opacity-50 ${TOUCH_ACTION_CLASS}`} + > + + {t(($) => $.tab_body.activity.move_down_tooltip)} + + + )} {showTranscript && (