mirror of
https://github.com/multica-ai/multica.git
synced 2026-08-05 01:19:42 +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>
222 lines
8.9 KiB
SQL
222 lines
8.9 KiB
SQL
-- name: CreateChatSession :one
|
|
INSERT INTO chat_session (workspace_id, agent_id, creator_id, title, runtime_id)
|
|
VALUES ($1, $2, $3, $4, (SELECT runtime_id FROM agent WHERE id = $2))
|
|
RETURNING *;
|
|
|
|
-- name: GetChatSession :one
|
|
SELECT * FROM chat_session
|
|
WHERE id = $1;
|
|
|
|
-- name: GetChatSessionInWorkspace :one
|
|
SELECT * FROM chat_session
|
|
WHERE id = $1 AND workspace_id = $2;
|
|
|
|
-- name: ListChatSessionsByCreator :many
|
|
-- Returns active sessions with a boolean unread flag. Unread is strictly
|
|
-- per-session: either the user has uncleared assistant replies in this
|
|
-- session or they don't. Counting messages would be misleading.
|
|
SELECT cs.*,
|
|
(cs.unread_since IS NOT NULL)::bool AS has_unread
|
|
FROM chat_session cs
|
|
WHERE cs.workspace_id = $1 AND cs.creator_id = $2 AND cs.status = 'active'
|
|
ORDER BY cs.updated_at DESC;
|
|
|
|
-- name: ListAllChatSessionsByCreator :many
|
|
SELECT cs.*,
|
|
(cs.unread_since IS NOT NULL)::bool AS has_unread
|
|
FROM chat_session cs
|
|
WHERE cs.workspace_id = $1 AND cs.creator_id = $2
|
|
ORDER BY cs.updated_at DESC;
|
|
|
|
-- name: UpdateChatSessionTitle :one
|
|
UPDATE chat_session SET title = $2, updated_at = now()
|
|
WHERE id = $1
|
|
RETURNING *;
|
|
|
|
-- name: UpdateChatSessionSession :exec
|
|
-- Updates the resume pointer for a chat session. Empty/NULL inputs are
|
|
-- ignored via COALESCE so a task that completes without a session_id (e.g.
|
|
-- the agent crashed before establishing one) cannot wipe out a previously
|
|
-- recorded resume pointer. This makes the chat memory robust against
|
|
-- intermittent agent failures.
|
|
UPDATE chat_session
|
|
SET session_id = COALESCE(sqlc.narg('session_id'), session_id),
|
|
work_dir = COALESCE(sqlc.narg('work_dir'), work_dir),
|
|
runtime_id = COALESCE(sqlc.narg('runtime_id'), runtime_id),
|
|
updated_at = now()
|
|
WHERE id = sqlc.arg('id');
|
|
|
|
-- name: LockChatSessionForDelete :one
|
|
-- Acquires an exclusive (FOR UPDATE) row lock on chat_session(id). Used by
|
|
-- the delete path so that a concurrent SendChatMessage cannot enqueue a new
|
|
-- agent_task_queue row referencing this session between our cancel and
|
|
-- delete steps. The FK from agent_task_queue.chat_session_id takes a
|
|
-- KEY SHARE lock on the parent row during INSERT validation, which
|
|
-- conflicts with FOR UPDATE — concurrent inserts block here and then fail
|
|
-- their FK check after we commit the delete.
|
|
SELECT id FROM chat_session
|
|
WHERE id = $1
|
|
FOR UPDATE;
|
|
|
|
-- name: DeleteChatSession :exec
|
|
-- Hard delete. chat_message rows cascade via FK ON DELETE CASCADE; the
|
|
-- chat_session_id on agent_task_queue is set NULL by FK so completed/failed
|
|
-- task history survives the session being removed. Callers MUST run inside
|
|
-- the same transaction that holds LockChatSessionForDelete and that has
|
|
-- already cancelled any in-flight tasks (see CancelAgentTasksByChatSession)
|
|
-- so the daemon does not keep running work whose result has nowhere to
|
|
-- land. workspace_id in the WHERE clause is a SQL-layer tenant guard; see
|
|
-- DeleteIssue.
|
|
DELETE FROM chat_session WHERE id = $1 AND workspace_id = $2;
|
|
|
|
-- name: TouchChatSession :exec
|
|
UPDATE chat_session SET updated_at = now()
|
|
WHERE id = $1;
|
|
|
|
-- name: CreateChatMessage :one
|
|
INSERT INTO chat_message (chat_session_id, role, content, task_id, failure_reason, elapsed_ms)
|
|
VALUES ($1, $2, $3, sqlc.narg(task_id), sqlc.narg(failure_reason), sqlc.narg(elapsed_ms))
|
|
RETURNING *;
|
|
|
|
-- name: LinkChatMessageToTask :exec
|
|
UPDATE chat_message
|
|
SET task_id = $2
|
|
WHERE id = $1 AND role = 'user';
|
|
|
|
-- name: DeleteUserChatMessageByTask :one
|
|
DELETE FROM chat_message
|
|
WHERE task_id = $1 AND role = 'user'
|
|
RETURNING *;
|
|
|
|
-- name: ListChatMessages :many
|
|
SELECT * FROM chat_message
|
|
WHERE chat_session_id = $1
|
|
ORDER BY created_at ASC;
|
|
|
|
-- name: ListChatMessagesPage :many
|
|
SELECT * FROM chat_message
|
|
WHERE chat_session_id = $1
|
|
AND (
|
|
sqlc.narg('before_created_at')::timestamptz IS NULL
|
|
OR (created_at, id) < (sqlc.narg('before_created_at')::timestamptz, sqlc.narg('before_id')::uuid)
|
|
)
|
|
ORDER BY created_at DESC, id DESC
|
|
LIMIT $2;
|
|
|
|
-- name: GetChatMessage :one
|
|
SELECT * FROM chat_message
|
|
WHERE id = $1;
|
|
|
|
-- name: CreateChatTask :one
|
|
INSERT INTO agent_task_queue (
|
|
agent_id, runtime_id, issue_id, status, priority, chat_session_id,
|
|
initiator_user_id, originator_user_id, force_fresh_session, runtime_mcp_overlay,
|
|
runtime_connected_apps
|
|
)
|
|
VALUES (
|
|
$1, $2, NULL, 'queued', $3, $4, $5,
|
|
sqlc.narg(originator_user_id),
|
|
COALESCE(sqlc.narg('force_fresh_session')::boolean, FALSE),
|
|
sqlc.narg(runtime_mcp_overlay),
|
|
sqlc.narg(runtime_connected_apps)
|
|
)
|
|
RETURNING *;
|
|
|
|
-- name: GetLastChatTaskSession :one
|
|
-- Returns the most recent task in this chat session that managed to record a
|
|
-- session_id. Includes both completed and failed tasks: even a failed task
|
|
-- may have established a real agent session before failing, and we'd rather
|
|
-- resume there than start over and lose conversation memory. Used as a
|
|
-- fallback when chat_session.session_id is NULL. Resume-unsafe failures are
|
|
-- excluded because replaying those sessions deterministically reproduces the
|
|
-- same terminal state.
|
|
SELECT session_id, work_dir, runtime_id FROM agent_task_queue
|
|
WHERE chat_session_id = $1
|
|
AND (
|
|
status = 'completed'
|
|
OR (
|
|
status = 'failed'
|
|
AND COALESCE(failure_reason, '') NOT IN ('iteration_limit', 'agent_fallback_message', 'api_invalid_request', 'codex_semantic_inactivity')
|
|
AND NOT (COALESCE(error, '') ILIKE '%400%' AND COALESCE(error, '') ILIKE '%invalid_request_error%')
|
|
)
|
|
)
|
|
AND session_id IS NOT NULL
|
|
ORDER BY completed_at DESC
|
|
LIMIT 1;
|
|
|
|
-- name: GetPendingChatTask :one
|
|
-- Returns the most recent in-flight task for a chat session, if any.
|
|
-- Used by the frontend to recover pending state after refresh / reopen.
|
|
-- created_at is the anchor for the chat StatusPill timer (it computes
|
|
-- elapsed = now - task.created_at), so the pill survives refresh / reopen
|
|
-- without "resetting to 0s".
|
|
SELECT id, status, created_at FROM agent_task_queue
|
|
WHERE chat_session_id = $1 AND status IN ('queued', 'dispatched', 'running', 'waiting_local_directory')
|
|
ORDER BY created_at DESC
|
|
LIMIT 1;
|
|
|
|
-- name: ListPendingChatTasksByCreator :many
|
|
-- Aggregate view of all in-flight chat tasks owned by a given creator in a
|
|
-- workspace. Drives the FAB's "running" indicator when the chat window is
|
|
-- closed and no single session's query is active.
|
|
--
|
|
-- Returns cs.agent_id so the handler can filter tasks belonging to private
|
|
-- agents the caller has lost access to using the already-loaded `allowed`
|
|
-- set — no second ListAllChatSessionsByCreator scan on the hot path.
|
|
--
|
|
-- atq.chat_session_id IS NOT NULL is redundant given the JOIN, but stated
|
|
-- explicitly so the planner can prove the query predicate is a subset of the
|
|
-- idx_agent_task_queue_chat_pending_v2 partial-index predicate and use it.
|
|
SELECT atq.id AS task_id, atq.status, atq.chat_session_id, cs.agent_id
|
|
FROM agent_task_queue atq
|
|
JOIN chat_session cs ON cs.id = atq.chat_session_id
|
|
WHERE atq.chat_session_id IS NOT NULL
|
|
AND atq.status IN ('queued', 'dispatched', 'running', 'waiting_local_directory')
|
|
AND cs.workspace_id = $1
|
|
AND cs.creator_id = $2
|
|
ORDER BY atq.created_at DESC;
|
|
|
|
-- name: HasPendingChatTasksByCreator :one
|
|
-- Boolean fast-path for the FAB's "running" indicator. Returns a single
|
|
-- EXISTS row instead of the full task list, so the planner can stop at the
|
|
-- first matching in-flight task (LIMIT 1 semantics via EXISTS).
|
|
--
|
|
-- Permission filtering is baked into the query: agent_id = ANY($3) restricts
|
|
-- the result to the agents the caller may currently see, so a member who lost
|
|
-- access to a private agent never gets a true from a task they can no longer
|
|
-- reach. The handler must pass its resolved accessible-agent id set as $3;
|
|
-- an empty array yields false.
|
|
SELECT EXISTS (
|
|
SELECT 1
|
|
FROM agent_task_queue atq
|
|
JOIN chat_session cs ON cs.id = atq.chat_session_id
|
|
WHERE atq.chat_session_id IS NOT NULL
|
|
AND atq.status IN ('queued', 'dispatched', 'running', 'waiting_local_directory')
|
|
AND cs.workspace_id = sqlc.arg(workspace_id)
|
|
AND cs.creator_id = sqlc.arg(creator_id)
|
|
AND cs.agent_id = ANY(sqlc.arg(agent_ids)::uuid[])
|
|
) AS has_pending;
|
|
|
|
-- name: MarkChatSessionRead :exec
|
|
-- Clears unread_since, dropping the session's unread count to 0.
|
|
UPDATE chat_session SET unread_since = NULL
|
|
WHERE id = $1;
|
|
|
|
-- name: SetUnreadSinceIfNull :exec
|
|
-- Atomically stamps the first unread assistant message's arrival time.
|
|
-- No-op if the session is already in "has unread" state — keeps the earliest
|
|
-- unread boundary stable across multiple incoming replies.
|
|
UPDATE chat_session SET unread_since = now()
|
|
WHERE id = $1 AND unread_since IS NULL;
|
|
|
|
-- name: GetMostRecentUserChatMessage :one
|
|
-- Returns the most recent role='user' message in a session. Used by the
|
|
-- Lark `/issue` command parser: when the user types `/issue` with no
|
|
-- title, the spec falls back to "use the previous user message as the
|
|
-- title". Bot replies (role='assistant') are excluded — only human
|
|
-- input qualifies as a fallback title source.
|
|
SELECT * FROM chat_message
|
|
WHERE chat_session_id = $1 AND role = 'user'
|
|
ORDER BY created_at DESC
|
|
LIMIT 1;
|