mirror of
https://github.com/multica-ai/multica.git
synced 2026-07-23 01:59:24 +02:00
* 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>
277 lines
11 KiB
Go
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)
|
|
}
|
|
} |