Files
multica/server/internal/handler/chat_pending_tasks_test.go
LinYushen 566d51f1c0 perf(chat): fix pending-task index mismatch + cut aggregate request storm (MUL-4159) (#5018)
* perf(chat): fix pending-task index mismatch + cut aggregate request storm (MUL-4159)

ListPendingChatTasksByCreator was a top DB hotspot. Root cause: the partial
index idx_agent_task_queue_chat_pending (migration 040) only covers
status IN (queued, dispatched, running), but migration 109 added a fourth
in-flight status (waiting_local_directory) that both pending chat queries now
filter on. Postgres can only use a partial index when the query predicate is a
subset of the index predicate, so the 4-status query stopped using it and
degraded to a Seq Scan over the whole agent_task_queue.

Implements the reviewed P0-P3 plan in one PR:

P0 index fix (split single-statement CONCURRENTLY migrations per repo convention)
- 143: CREATE INDEX CONCURRENTLY idx_agent_task_queue_chat_pending_v2 covering
  all four in-flight statuses (same column list, so GetPendingChatTask still
  benefits).
- 144: DROP the superseded 3-status index, in its own migration.

P1 SQL + handler hot path
- ListPendingChatTasksByCreator now returns cs.agent_id and states
  chat_session_id IS NOT NULL so the planner can prove the partial-index subset.
- ListPendingChatTasks filters private-agent access against the already-loaded
  accessible-agent set using the returned agent_id, dropping the extra
  ListAllChatSessionsByCreator scan on the hot path.
- Regenerated sqlc.

P2 frontend request amplification
- FAB uses the new boolean has-any query gated on enabled:!isOpen, so the
  minimised button never holds the full aggregate.
- use-realtime-sync maintains the pending aggregate (list + has-any) in place
  from task lifecycle events (queued/dispatch/running/waiting_local_directory
  -> upsert; completed/failed/cancelled -> remove) instead of invalidating on
  every chat:message/chat:done, with a debounced fallback invalidate for
  reconnect / unknown payloads.

P3 boolean endpoint
- GET /api/chat/pending-tasks/has-any backed by HasPendingChatTasksByCreator
  (EXISTS). Permission filtering is baked in via agent_id = ANY($3); an empty
  accessible-agent set short-circuits to false. The detailed list stays for the
  ChatWindow history / stop-task flows.

Tests: new handler tests cover the private-agent gate on both endpoints
(hidden from a creator who lost access, visible to the agent owner) plus the
boolean status/terminal semantics.

EXPLAIN (ANALYZE, BUFFERS) on a 300k-row reproduction:
- before (3-status index): Parallel Seq Scan, ~300k rows filtered,
  shared hit=3012, 12.1 ms.
- after (v2 index): Index Scan on idx_agent_task_queue_chat_pending_v2,
  shared hit=131, 0.07 ms.

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

* fix(chat): stop optimistic cross-session pending aggregate writes from workspace-fanout task events (MUL-4159)

Review on PR #5018 flagged a real privilege-escalation bug in the P2 change:
use-realtime-sync optimistically upserted the cross-session pending aggregate
(pendingTasks / pendingTasksHasAny) from chat task:* events. Those events are a
workspace fanout delivered to every member (server still BroadcastToWorkspace,
see cmd/server/listeners.go), and the payload carries no creator / agent
visibility. So member B starting a chat task could flip member A's FAB to
has_pending=true, bypassing the server-side permission filter on
/api/chat/pending-tasks[/has-any].

Fix (option 1 from the review — the self-contained one): never optimistically
write the aggregate from task:* events. On every task lifecycle transition,
debounced-invalidate the aggregate so it is refetched through the
permission-filtering endpoint, which only returns the caller's own
creator-owned, accessible-agent tasks. The per-session pendingTask cache is
still written directly — it is keyed by chat_session_id and only rendered for
sessions the user can open (server-gated), so it is not a cross-user leak.
chat:message is still excluded from aggregate refresh, so the MUL-4159 request
storm stays fixed; task transitions are per-task and coalesced by the debounce.

- Removed upsertPendingAggregate / removePendingAggregate.
- Added exported refetchPendingChatAggregate(qc, wsId) — an invalidate, never a
  setQueryData — used by the debounced handler.
- Regression tests: refetchPendingChatAggregate leaves the cached
  has_pending/list untouched (no optimistic write) and only invalidates for an
  authoritative server-filtered refetch; no-ops without a workspace id.

Verified: @multica/core + @multica/views typecheck; full core vitest suite
(752 tests) green including the 2 new guard tests.

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

* chore(chat): address review nits on pending-tasks endpoints (MUL-4159)

- Restore the GetPendingChatTask godoc first line that was clipped when the
  has-any handler was inserted (nit#1).
- ListPendingChatTasks short-circuits to an empty list when the caller has no
  accessible agents, mirroring HasPendingChatTasks — skips the DB round-trip
  (nit#2).
- Add a cross-creator negative test: user A's in-flight task on a
  workspace-visible agent returns has_pending=false / empty list for user B,
  locking the cs.creator_id tenant gate that the agent-visibility filter does
  not cover (nit#3).

Verified: go build ./... and go test ./internal/handler -run PendingChatTasks
(7 tests) green against live Postgres.

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

---------

Co-authored-by: multica-agent <github@multica.ai>
2026-07-07 13:20:30 +08:00

277 lines
11 KiB
Go

package handler
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"github.com/multica-ai/multica/server/internal/middleware"
"github.com/multica-ai/multica/server/internal/util"
db "github.com/multica-ai/multica/server/pkg/db/generated"
)
// chatPendingCtxAs injects the workspace + member context that the chi
// workspace middleware would normally set, resolved for an arbitrary user.
// The pending-tasks handlers read the workspace id from ctxWorkspaceID and the
// caller's role from the member context, so a direct handler call needs both.
func chatPendingCtxAs(t *testing.T, req *http.Request, userID string) *http.Request {
t.Helper()
memberRow, err := testHandler.Queries.GetMemberByUserAndWorkspace(context.Background(), db.GetMemberByUserAndWorkspaceParams{
UserID: util.MustParseUUID(userID),
WorkspaceID: util.MustParseUUID(testWorkspaceID),
})
if err != nil {
t.Fatalf("load member row for %s: %v", userID, err)
}
return req.WithContext(middleware.SetMemberContext(req.Context(), testWorkspaceID, memberRow))
}
// insertChatSessionAs inserts a chat_session owned by an explicit creator +
// agent (createHandlerTestChatSession hardcodes testUserID as creator, which
// these permission tests need to vary).
func insertChatSessionAs(t *testing.T, agentID, creatorID string) string {
t.Helper()
var sessionID string
if err := testPool.QueryRow(context.Background(), `
INSERT INTO chat_session (workspace_id, agent_id, creator_id, title, status)
VALUES ($1, $2, $3, 'pending-tasks-test', 'active')
RETURNING id
`, testWorkspaceID, agentID, creatorID).Scan(&sessionID); err != nil {
t.Fatalf("insert chat session: %v", err)
}
t.Cleanup(func() {
testPool.Exec(context.Background(), `DELETE FROM chat_session WHERE id = $1`, sessionID)
})
return sessionID
}
// insertPendingChatTask seeds an in-flight agent_task_queue row bound to a chat
// session and returns its id.
func insertPendingChatTask(t *testing.T, agentID, sessionID, status string) string {
t.Helper()
var taskID string
if err := testPool.QueryRow(context.Background(), `
INSERT INTO agent_task_queue (agent_id, runtime_id, status, priority, chat_session_id)
VALUES ($1, $2, $3, 0, $4)
RETURNING id
`, agentID, handlerTestRuntimeID(t), status, sessionID).Scan(&taskID); err != nil {
t.Fatalf("insert pending chat task: %v", err)
}
t.Cleanup(func() {
testPool.Exec(context.Background(), `DELETE FROM agent_task_queue WHERE id = $1`, taskID)
})
return taskID
}
func decodePendingTasks(t *testing.T, w *httptest.ResponseRecorder) PendingChatTasksResponse {
t.Helper()
if w.Code != http.StatusOK {
t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
}
var resp PendingChatTasksResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("decode pending tasks: %v", err)
}
return resp
}
func decodeHasPending(t *testing.T, w *httptest.ResponseRecorder) bool {
t.Helper()
if w.Code != http.StatusOK {
t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
}
var resp HasPendingChatTasksResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("decode has-any: %v", err)
}
return resp.HasPending
}
func containsPendingTask(tasks []PendingChatTaskItem, taskID string) bool {
for _, it := range tasks {
if it.TaskID == taskID {
return true
}
}
return false
}
// TestListPendingChatTasks_HidesPrivateAgentFromLostAccessCreator verifies the
// P1 rewrite still enforces the private-agent gate now that filtering keys off
// the agent_id returned by the query (instead of a second session-list scan).
// A member who created a chat with a private agent they can no longer access
// must not see that task in the aggregate, while a task on a workspace-visible
// agent they still own is returned.
func TestListPendingChatTasks_HidesPrivateAgentFromLostAccessCreator(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
privateAgentID, _, memberID := privateAgentTestFixture(t)
publicAgentID := createHandlerTestAgent(t, "PendingPublicAgent", []byte("[]"))
// The plain member is the creator of BOTH sessions, so the creator_id
// filter admits both; only the private-agent gate should drop one.
privateSession := insertChatSessionAs(t, privateAgentID, memberID)
publicSession := insertChatSessionAs(t, publicAgentID, memberID)
privateTask := insertPendingChatTask(t, privateAgentID, privateSession, "running")
publicTask := insertPendingChatTask(t, publicAgentID, publicSession, "queued")
w := httptest.NewRecorder()
testHandler.ListPendingChatTasks(w, chatPendingCtxAs(t, newRequestAs(memberID, "GET", "/api/chat/pending-tasks", nil), memberID))
resp := decodePendingTasks(t, w)
if containsPendingTask(resp.Tasks, privateTask) {
t.Fatalf("private-agent task %s leaked to plain member: %+v", privateTask, resp.Tasks)
}
if !containsPendingTask(resp.Tasks, publicTask) {
t.Fatalf("public-agent task %s missing from plain member's pending list: %+v", publicTask, resp.Tasks)
}
}
// TestListPendingChatTasks_OwnerSeesPrivateAgentTask is the positive control:
// the agent owner keeps visibility of their own private-agent chat task.
func TestListPendingChatTasks_OwnerSeesPrivateAgentTask(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
privateAgentID, ownerID, _ := privateAgentTestFixture(t)
session := insertChatSessionAs(t, privateAgentID, ownerID)
task := insertPendingChatTask(t, privateAgentID, session, "running")
w := httptest.NewRecorder()
testHandler.ListPendingChatTasks(w, chatPendingCtxAs(t, newRequestAs(ownerID, "GET", "/api/chat/pending-tasks", nil), ownerID))
resp := decodePendingTasks(t, w)
if !containsPendingTask(resp.Tasks, task) {
t.Fatalf("agent owner did not see their own private-agent task %s: %+v", task, resp.Tasks)
}
}
// TestHasPendingChatTasks_FalseWhenOnlyInaccessiblePrivateAgent verifies the P3
// boolean endpoint preserves the same permission filtering as the list: a
// member whose only in-flight task is on a private agent they can't access
// gets has_pending=false (the FAB must not light up for a task they've lost
// access to).
func TestHasPendingChatTasks_FalseWhenOnlyInaccessiblePrivateAgent(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
privateAgentID, _, memberID := privateAgentTestFixture(t)
session := insertChatSessionAs(t, privateAgentID, memberID)
insertPendingChatTask(t, privateAgentID, session, "running")
w := httptest.NewRecorder()
testHandler.HasPendingChatTasks(w, chatPendingCtxAs(t, newRequestAs(memberID, "GET", "/api/chat/pending-tasks/has-any", nil), memberID))
if decodeHasPending(t, w) {
t.Fatalf("has-any returned true for a task on a private agent the member cannot access")
}
}
// TestHasPendingChatTasks_TrueWhenAccessiblePublicAgent verifies the boolean
// endpoint returns true once the member has an in-flight task on an agent they
// can access — and that an inaccessible private-agent task alongside it does
// not change the (already true) answer.
func TestHasPendingChatTasks_TrueWhenAccessiblePublicAgent(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
privateAgentID, _, memberID := privateAgentTestFixture(t)
publicAgentID := createHandlerTestAgent(t, "PendingPublicAgentHasAny", []byte("[]"))
privateSession := insertChatSessionAs(t, privateAgentID, memberID)
publicSession := insertChatSessionAs(t, publicAgentID, memberID)
insertPendingChatTask(t, privateAgentID, privateSession, "running")
insertPendingChatTask(t, publicAgentID, publicSession, "waiting_local_directory")
w := httptest.NewRecorder()
testHandler.HasPendingChatTasks(w, chatPendingCtxAs(t, newRequestAs(memberID, "GET", "/api/chat/pending-tasks/has-any", nil), memberID))
if !decodeHasPending(t, w) {
t.Fatalf("has-any returned false despite an in-flight task on an accessible public agent")
}
}
// TestHasPendingChatTasks_OwnerOfPrivateAgentSeesTask confirms the boolean
// endpoint's positive path for a private agent's owner.
func TestHasPendingChatTasks_OwnerOfPrivateAgentSeesTask(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
privateAgentID, ownerID, _ := privateAgentTestFixture(t)
session := insertChatSessionAs(t, privateAgentID, ownerID)
insertPendingChatTask(t, privateAgentID, session, "dispatched")
w := httptest.NewRecorder()
testHandler.HasPendingChatTasks(w, chatPendingCtxAs(t, newRequestAs(ownerID, "GET", "/api/chat/pending-tasks/has-any", nil), ownerID))
if !decodeHasPending(t, w) {
t.Fatalf("has-any returned false for the private agent's own owner")
}
}
// TestHasPendingChatTasks_IgnoresTerminalTasks guards the status predicate: a
// completed task is not "pending", so the endpoint must return false.
func TestHasPendingChatTasks_IgnoresTerminalTasks(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
// Use a freshly-created member (via the private-agent fixture) as the
// creator so no stray in-flight task from another test can bleed into the
// assertion — this user exists only for the duration of this test.
_, _, memberID := privateAgentTestFixture(t)
publicAgentID := createHandlerTestAgent(t, "PendingTerminalAgent", []byte("[]"))
session := insertChatSessionAs(t, publicAgentID, memberID)
insertPendingChatTask(t, publicAgentID, session, "completed")
w := httptest.NewRecorder()
testHandler.HasPendingChatTasks(w, chatPendingCtxAs(t, newRequestAs(memberID, "GET", "/api/chat/pending-tasks/has-any", nil), memberID))
if decodeHasPending(t, w) {
t.Fatalf("has-any returned true for a terminal (completed) task")
}
}
// TestHasPendingChatTasks_HidesOtherCreatorsTask locks the cs.creator_id gate:
// user A's in-flight task on a workspace-visible agent — one B can freely
// access — must still return has_pending=false for B, because B is not the
// creator. This is the tenant boundary that the agent-visibility filter does
// NOT cover (both users can see the agent), so it needs its own guard.
func TestHasPendingChatTasks_HidesOtherCreatorsTask(t *testing.T) {
if testHandler == nil {
t.Skip("database not available")
}
// ownerID = user A (task creator), memberID = user B (different member).
// Both are plain workspace members who can access a workspace-visible agent.
_, creatorA, otherB := privateAgentTestFixture(t)
publicAgentID := createHandlerTestAgent(t, "PendingCrossCreatorAgent", []byte("[]"))
session := insertChatSessionAs(t, publicAgentID, creatorA)
insertPendingChatTask(t, publicAgentID, session, "running")
// B is not the creator → the creator_id filter must drop A's task.
w := httptest.NewRecorder()
testHandler.HasPendingChatTasks(w, chatPendingCtxAs(t, newRequestAs(otherB, "GET", "/api/chat/pending-tasks/has-any", nil), otherB))
if decodeHasPending(t, w) {
t.Fatalf("has-any leaked user A's task to user B (cs.creator_id gate not enforced)")
}
// Sanity: A (the creator) does see their own task on the same agent.
w = httptest.NewRecorder()
testHandler.HasPendingChatTasks(w, chatPendingCtxAs(t, newRequestAs(creatorA, "GET", "/api/chat/pending-tasks/has-any", nil), creatorA))
if !decodeHasPending(t, w) {
t.Fatalf("has-any returned false for the task's own creator")
}
// The detailed list endpoint enforces the same creator gate.
w = httptest.NewRecorder()
testHandler.ListPendingChatTasks(w, chatPendingCtxAs(t, newRequestAs(otherB, "GET", "/api/chat/pending-tasks", nil), otherB))
if resp := decodePendingTasks(t, w); len(resp.Tasks) != 0 {
t.Fatalf("list leaked user A's task to user B: %+v", resp.Tasks)
}
}