Files
multica/server/pkg/db/generated/chat.sql.go
Bohan Jiang 85a14cde37 fix(daemon): gate codex session pointer writes on rollout presence (MUL-5305) (#5960)
* fix(daemon): gate codex session pointer writes on rollout presence (MUL-5305)

Codex issue follow-ups on local_directory projects intermittently lost
their session: the server sent a prior session whose rollout was not in
the task CODEX_HOME, so the daemon dropped the resume and started a fresh
thread (gateCodexResumeToRolloutPresence), losing the conversation.

Root of the bad pointer: the daemon persists a Codex session id as the
resumable pointer at two points -- the mid-flight pin and the terminal
report -- before the rollout is guaranteed on disk. A task that exits
early (crash / runtime offline / timeout) leaves a pinned/reported
session id with no rollout; GetLastTaskSession (which accepts failed
rows) then hands it to the next follow-up, which drops it.

Enforce the invariant at write time: only record a Codex session as the
resumable pointer once its rollout is present in the per-issue store,
with a short bounded wait for flush. If it never lands, don't overwrite
the last good pointer -- a blanked session_id becomes NULL server-side,
so GetLastTaskSession falls back to the most recent session whose
rollout is real. Non-Codex providers are unaffected; crash recovery is
preserved because a present rollout still pins.

- codexSessionResumable: shared write-time presence check (bounded wait)
- runTask: gate the terminal session_id before reporting
- executeAndDrain: gate the mid-flight pin (thread codexHome through)
- tests: helper cases + behavioral pin test

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

* fix(daemon): address review — don't silently downgrade completed sessions (MUL-5305)

Follow-up to review feedback on #5960:

- Must-fix 1 (silent downgrade): limit the write-time session withholding
  to NON-completed terminal states. A missing rollout means no resumable
  conversation was persisted, so a withheld non-completed attempt loses
  nothing; a completed session is authoritative and, if its rollout is
  anomalously absent, is still recorded so the next run's resume gate
  discloses the loss (PriorSessionResumeUnavailable, MUL-4424) instead of
  silently falling back to an older session. Extracted
  resumableTerminalSessionID.
- Non-blocking risk: pin the mid-flight resume pointer with a per-status
  presence check instead of one fixed 2s window, and set sessionPinned
  only once the rollout is confirmed, so a rollout that lands shortly
  after the first status is still pinned this run.
- Must-fix 2 (regression coverage): pin skipped when rollout absent (no
  /session call); terminal helper (completed keeps / failed withholds);
  and a DB-backed GetLastTaskSession test proving the next claim falls
  back to the older recorded session when the latest was blanked.

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

* fix(daemon): disclose Codex session continuity gaps end-to-end (MUL-5305)

Addresses review feedback on #5960.

Must-fix 1 — a completed turn whose rollout is missing is exactly the
#5934 case (the reporter waits for each turn to finish), so it can no
longer be excluded from withholding. Withhold the session for ANY
terminal state, and pair the withhold with a persisted continuity-gap
signal so the next claim still discloses the loss even while resuming an
older good session:
  - new agent_task_queue.session_rollout_missing column (migration 224)
  - daemon sends session_rollout_missing on the terminal report; the
    handler clears the resume pointer (MarkTaskSessionRolloutMissing,
    overriding FailAgentTask's COALESCE) and flags the row
  - claim reads GetLatestTaskRolloutMissing and sets a new
    prior_session_resume_unavailable response field, which the daemon ORs
    into the brief's PriorSessionResumeUnavailable disclosure

Must-fix 2 — Codex reveals the session id on a single task_started
status, so a one-shot presence check missed a rollout that flushed later
and lost in-flight crash recovery. Pin via a background waiter bounded by
the run's context that pins the moment the rollout lands.

Tests: - completed + rollout missing -> next claim withholds the bad session
    AND flags the continuity gap (cross-layer DB test)
  - session pinned once its rollout appears after the status (mid-run)
  - pin skipped while the rollout is absent
Co-authored-by: multica-agent <github@multica.ai>

* fix(server): make continuity-gap write atomic + disclose on all claim paths (MUL-5305)

Addresses review round 3 of #5960.

Must-fix 1 — the previous handler-level marker ran AFTER the terminal
transaction committed, and FailTask creates + wakes the auto-retry inside
that same transaction, so a retry could claim the rollout-missing session
before the marker cleared it (and a marker failure was swallowed). Move
session_rollout_missing INTO the terminal write: CompleteAgentTask and
FailAgentTask now force session_id NULL (overriding Fail's COALESCE that
would keep a stale mid-flight pin) and set the flag in the SAME UPDATE, so
the withhold + gap flag commit atomically with the retry creation. The
flag is threaded through TaskService.CompleteTask/FailTask; the swallowed
best-effort MarkTaskSessionRolloutMissing query is removed.

Must-fix 2 — the daemon withholds for all Codex tasks, but only the issue
non-rerun claim consumed the disclosure. Now every fallback path sets
prior_session_resume_unavailable: the manual-rerun branch reads the source
task's session_rollout_missing, and the chat branch reads a new
GetLatestChatTaskRolloutMissing.

Tests (cross-layer DB):
- completed + rollout missing via the real CompleteAgentTask terminal
  write -> session withheld AND gap flagged
- failed + rollout missing forces session_id NULL over the COALESCE-
  preserved mid-flight pin in ONE statement

Deploy order: migration + server first, daemon second (new fields are
omitempty and ignored by an old peer).

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

* fix(handler): return 5xx on FailTask error + cover claim-response gap paths (MUL-5305)

Addresses review round 4 of #5960.

Must-fix 1 — the FailTask handler returned 400 on a service/DB error, but
the daemon's terminal callback treats 400 as permanent (postJSONWithRetry
/ isTransientError bails without retrying). Since the fail transaction is
now the sole persistence point for the withheld session + continuity-gap
flag + auto-retry, a rolled-back fail must be retried, so return 5xx (an
invalid request body still returns 400), mirroring CompleteTask.
Regression: client.FailTask retries on a transient 5xx and eventually
succeeds.

Must-fix 2 — add claim-response-level regressions that drive the two new
disclosure branches through buildClaimedTaskResponse:
  - chat: the latest terminal task on the session withheld -> the next
    chat claim sets prior_session_resume_unavailable
  - manual rerun: the source task withheld -> the rerun claim discloses
These handler DB tests run under CI's fully-migrated database (the local
workspace DB cannot set up the handler fixture).

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

---------

Co-authored-by: Bohan-J <bohan@devv.ai>
Co-authored-by: multica-agent <github@multica.ai>
2026-07-27 13:12:06 +08:00

1554 lines
53 KiB
Go

// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: chat.sql
package db
import (
"context"
"github.com/jackc/pgx/v5/pgtype"
)
const chatSessionHasUserMessage = `-- name: ChatSessionHasUserMessage :one
SELECT EXISTS (
SELECT 1 FROM chat_message
WHERE chat_session_id = $1 AND role = 'user'
) AS has_user_message
`
// Reports whether a session has any human (role='user') message yet. Used to
// scope the is_agent_intro self-introduction prompt to the very first,
// server-driven turn: an intro session starts with zero user messages, so the
// opening run gets the "introduce yourself" prompt. Once the creator replies,
// later turns in the same session must fall back to the normal reply prompt
// instead of repeating the introduction every turn (MUL-4259).
func (q *Queries) ChatSessionHasUserMessage(ctx context.Context, chatSessionID pgtype.UUID) (bool, error) {
row := q.db.QueryRow(ctx, chatSessionHasUserMessage, chatSessionID)
var has_user_message bool
err := row.Scan(&has_user_message)
return has_user_message, err
}
const clearChatSessionProjectByProject = `-- name: ClearChatSessionProjectByProject :exec
UPDATE chat_session
SET project_id = NULL
WHERE project_id = $1 AND workspace_id = $2
`
type ClearChatSessionProjectByProjectParams struct {
ProjectID pgtype.UUID `json:"project_id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
}
// Project references are intentionally soft (no database FK). Keep chat
// history while removing the context selection when a project is deleted.
// Do not touch updated_at: context cleanup is not chat activity.
func (q *Queries) ClearChatSessionProjectByProject(ctx context.Context, arg ClearChatSessionProjectByProjectParams) error {
_, err := q.db.Exec(ctx, clearChatSessionProjectByProject, arg.ProjectID, arg.WorkspaceID)
return err
}
const createChatDraftRestore = `-- name: CreateChatDraftRestore :one
INSERT INTO chat_draft_restore (id, chat_session_id, task_id, content, attachment_ids)
VALUES ($1, $2, $3, $4, $5)
RETURNING id, chat_session_id, task_id, content, attachment_ids, created_at
`
type CreateChatDraftRestoreParams struct {
ID pgtype.UUID `json:"id"`
ChatSessionID pgtype.UUID `json:"chat_session_id"`
TaskID pgtype.UUID `json:"task_id"`
Content string `json:"content"`
AttachmentIds []pgtype.UUID `json:"attachment_ids"`
}
// Persists the deferred-cancellation draft restore (#5219) in the same tx
// that deletes the triggering user message: the chat:cancel_finalized
// broadcast is best-effort, so an offline client recovers the draft from
// this row instead. id is the deleted message's id.
func (q *Queries) CreateChatDraftRestore(ctx context.Context, arg CreateChatDraftRestoreParams) (ChatDraftRestore, error) {
row := q.db.QueryRow(ctx, createChatDraftRestore,
arg.ID,
arg.ChatSessionID,
arg.TaskID,
arg.Content,
arg.AttachmentIds,
)
var i ChatDraftRestore
err := row.Scan(
&i.ID,
&i.ChatSessionID,
&i.TaskID,
&i.Content,
&i.AttachmentIds,
&i.CreatedAt,
)
return i, err
}
const createChatMessage = `-- name: CreateChatMessage :one
INSERT INTO chat_message (chat_session_id, role, content, task_id, failure_reason, elapsed_ms, message_kind)
VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7::text, 'message'))
RETURNING id, chat_session_id, role, content, task_id, created_at, failure_reason, elapsed_ms, message_kind
`
type CreateChatMessageParams struct {
ChatSessionID pgtype.UUID `json:"chat_session_id"`
Role string `json:"role"`
Content string `json:"content"`
TaskID pgtype.UUID `json:"task_id"`
FailureReason pgtype.Text `json:"failure_reason"`
ElapsedMs pgtype.Int8 `json:"elapsed_ms"`
MessageKind pgtype.Text `json:"message_kind"`
}
// message_kind defaults to 'message' via COALESCE so every existing caller
// (which omits it) keeps writing ordinary messages; the empty-reply path passes
// 'no_response' to mark a visible turn with no text output (MUL-4351).
func (q *Queries) CreateChatMessage(ctx context.Context, arg CreateChatMessageParams) (ChatMessage, error) {
row := q.db.QueryRow(ctx, createChatMessage,
arg.ChatSessionID,
arg.Role,
arg.Content,
arg.TaskID,
arg.FailureReason,
arg.ElapsedMs,
arg.MessageKind,
)
var i ChatMessage
err := row.Scan(
&i.ID,
&i.ChatSessionID,
&i.Role,
&i.Content,
&i.TaskID,
&i.CreatedAt,
&i.FailureReason,
&i.ElapsedMs,
&i.MessageKind,
)
return i, err
}
const createChatSession = `-- name: CreateChatSession :one
INSERT INTO chat_session (workspace_id, agent_id, creator_id, title, runtime_id, is_agent_intro, project_id)
VALUES ($1, $2, $3, $4, (SELECT runtime_id FROM agent WHERE id = $2), $5, $6)
RETURNING id, workspace_id, agent_id, creator_id, title, session_id, work_dir, status, created_at, updated_at, unread_since, runtime_id, last_read_at, is_agent_intro, pinned_at, project_id
`
type CreateChatSessionParams struct {
WorkspaceID pgtype.UUID `json:"workspace_id"`
AgentID pgtype.UUID `json:"agent_id"`
CreatorID pgtype.UUID `json:"creator_id"`
Title string `json:"title"`
IsAgentIntro bool `json:"is_agent_intro"`
ProjectID pgtype.UUID `json:"project_id"`
}
func (q *Queries) CreateChatSession(ctx context.Context, arg CreateChatSessionParams) (ChatSession, error) {
row := q.db.QueryRow(ctx, createChatSession,
arg.WorkspaceID,
arg.AgentID,
arg.CreatorID,
arg.Title,
arg.IsAgentIntro,
arg.ProjectID,
)
var i ChatSession
err := row.Scan(
&i.ID,
&i.WorkspaceID,
&i.AgentID,
&i.CreatorID,
&i.Title,
&i.SessionID,
&i.WorkDir,
&i.Status,
&i.CreatedAt,
&i.UpdatedAt,
&i.UnreadSince,
&i.RuntimeID,
&i.LastReadAt,
&i.IsAgentIntro,
&i.PinnedAt,
&i.ProjectID,
)
return i, err
}
const createChatTask = `-- 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, accountable_user_id, force_fresh_session, runtime_mcp_overlay,
runtime_connected_apps, originator_source, trigger_evidence_kind, trigger_evidence_ref_id
)
VALUES (
$1, $2, NULL, 'queued', $3, $4, $5,
$6,
$7,
COALESCE($8::boolean, FALSE),
$9,
$10,
$11,
$12,
$13
)
RETURNING id, agent_id, issue_id, status, priority, dispatched_at, started_at, completed_at, result, error, created_at, context, runtime_id, session_id, work_dir, trigger_comment_id, chat_session_id, autopilot_run_id, attempt, max_attempts, parent_task_id, failure_reason, trigger_summary, force_fresh_session, is_leader_task, wait_reason, initiator_user_id, handoff_note, prepare_lease_expires_at, squad_id, runtime_mcp_overlay, escalation_for_task_id, fire_at, originator_user_id, runtime_connected_apps, coalesced_comment_ids, delivered_comment_ids, chat_input_task_id, chat_finalize_deferred_at, originator_source, delegated_from_task_id, retry_of_task_id, rerun_of_task_id, rule_version_id, trigger_evidence_kind, trigger_evidence_ref_id, accountable_user_id, session_rollout_missing
`
type CreateChatTaskParams struct {
AgentID pgtype.UUID `json:"agent_id"`
RuntimeID pgtype.UUID `json:"runtime_id"`
Priority int32 `json:"priority"`
ChatSessionID pgtype.UUID `json:"chat_session_id"`
InitiatorUserID pgtype.UUID `json:"initiator_user_id"`
OriginatorUserID pgtype.UUID `json:"originator_user_id"`
AccountableUserID pgtype.UUID `json:"accountable_user_id"`
ForceFreshSession pgtype.Bool `json:"force_fresh_session"`
RuntimeMcpOverlay []byte `json:"runtime_mcp_overlay"`
RuntimeConnectedApps []byte `json:"runtime_connected_apps"`
OriginatorSource pgtype.Text `json:"originator_source"`
TriggerEvidenceKind pgtype.Text `json:"trigger_evidence_kind"`
TriggerEvidenceRefID pgtype.UUID `json:"trigger_evidence_ref_id"`
}
// The chat sender (initiator) is a direct_human originator and accountable;
// attribution provenance is stamped so this path is not a NULL-source enqueue
// bypass (MUL-4302 §2).
func (q *Queries) CreateChatTask(ctx context.Context, arg CreateChatTaskParams) (AgentTaskQueue, error) {
row := q.db.QueryRow(ctx, createChatTask,
arg.AgentID,
arg.RuntimeID,
arg.Priority,
arg.ChatSessionID,
arg.InitiatorUserID,
arg.OriginatorUserID,
arg.AccountableUserID,
arg.ForceFreshSession,
arg.RuntimeMcpOverlay,
arg.RuntimeConnectedApps,
arg.OriginatorSource,
arg.TriggerEvidenceKind,
arg.TriggerEvidenceRefID,
)
var i AgentTaskQueue
err := row.Scan(
&i.ID,
&i.AgentID,
&i.IssueID,
&i.Status,
&i.Priority,
&i.DispatchedAt,
&i.StartedAt,
&i.CompletedAt,
&i.Result,
&i.Error,
&i.CreatedAt,
&i.Context,
&i.RuntimeID,
&i.SessionID,
&i.WorkDir,
&i.TriggerCommentID,
&i.ChatSessionID,
&i.AutopilotRunID,
&i.Attempt,
&i.MaxAttempts,
&i.ParentTaskID,
&i.FailureReason,
&i.TriggerSummary,
&i.ForceFreshSession,
&i.IsLeaderTask,
&i.WaitReason,
&i.InitiatorUserID,
&i.HandoffNote,
&i.PrepareLeaseExpiresAt,
&i.SquadID,
&i.RuntimeMcpOverlay,
&i.EscalationForTaskID,
&i.FireAt,
&i.OriginatorUserID,
&i.RuntimeConnectedApps,
&i.CoalescedCommentIds,
&i.DeliveredCommentIds,
&i.ChatInputTaskID,
&i.ChatFinalizeDeferredAt,
&i.OriginatorSource,
&i.DelegatedFromTaskID,
&i.RetryOfTaskID,
&i.RerunOfTaskID,
&i.RuleVersionID,
&i.TriggerEvidenceKind,
&i.TriggerEvidenceRefID,
&i.AccountableUserID,
&i.SessionRolloutMissing,
)
return i, err
}
const deleteChatDraftRestore = `-- name: DeleteChatDraftRestore :execrows
DELETE FROM chat_draft_restore
WHERE id = $1 AND chat_session_id = $2
`
type DeleteChatDraftRestoreParams struct {
ID pgtype.UUID `json:"id"`
ChatSessionID pgtype.UUID `json:"chat_session_id"`
}
// Idempotent consume: deleting an already-consumed restore matches no row.
func (q *Queries) DeleteChatDraftRestore(ctx context.Context, arg DeleteChatDraftRestoreParams) (int64, error) {
result, err := q.db.Exec(ctx, deleteChatDraftRestore, arg.ID, arg.ChatSessionID)
if err != nil {
return 0, err
}
return result.RowsAffected(), nil
}
const deleteChatDraftRestoresByArchivedRuntimeAgents = `-- name: DeleteChatDraftRestoresByArchivedRuntimeAgents :exec
DELETE FROM chat_draft_restore
WHERE chat_session_id IN (
SELECT cs.id FROM chat_session cs
JOIN agent a ON a.id = cs.agent_id
WHERE a.runtime_id = $1 AND a.archived_at IS NOT NULL
)
`
// chat_session cascades from agent, so hard-deleting a runtime's archived agents
// silently drops their sessions — and, without an FK, would strand the pending
// restores (which still hold the user's prompt text) forever. Prune them in the
// same tx, BEFORE the agent rows go: the join below needs them. Mirrors
// DeleteChannelInstallationsByArchivedRuntimeAgents.
func (q *Queries) DeleteChatDraftRestoresByArchivedRuntimeAgents(ctx context.Context, runtimeID pgtype.UUID) error {
_, err := q.db.Exec(ctx, deleteChatDraftRestoresByArchivedRuntimeAgents, runtimeID)
return err
}
const deleteChatDraftRestoresBySession = `-- name: DeleteChatDraftRestoresBySession :exec
DELETE FROM chat_draft_restore
WHERE chat_session_id = $1
`
// chat_draft_restore carries no chat_session FK (MUL-3515), so DeleteChatSession
// prunes its pending restores in the same tx that deletes the session.
func (q *Queries) DeleteChatDraftRestoresBySession(ctx context.Context, chatSessionID pgtype.UUID) error {
_, err := q.db.Exec(ctx, deleteChatDraftRestoresBySession, chatSessionID)
return err
}
const deleteChatDraftRestoresBySystemRuntimeAgents = `-- name: DeleteChatDraftRestoresBySystemRuntimeAgents :exec
DELETE FROM chat_draft_restore
WHERE chat_session_id IN (
SELECT cs.id FROM chat_session cs
JOIN agent a ON a.id = cs.agent_id
WHERE a.runtime_id = $1 AND a.kind = 'system'
)
`
// Same cascade, for the system agents a runtime teardown also hard-deletes
// (DeleteSystemAgentsByRuntime). Split from the archived-agent prune because the
// runtime-profile teardown deletes only archived agents: pruning system-agent
// sessions there would destroy restores whose session survives.
func (q *Queries) DeleteChatDraftRestoresBySystemRuntimeAgents(ctx context.Context, runtimeID pgtype.UUID) error {
_, err := q.db.Exec(ctx, deleteChatDraftRestoresBySystemRuntimeAgents, runtimeID)
return err
}
const deleteChatSession = `-- name: DeleteChatSession :exec
DELETE FROM chat_session WHERE id = $1 AND workspace_id = $2
`
type DeleteChatSessionParams struct {
ID pgtype.UUID `json:"id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
}
// 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.
func (q *Queries) DeleteChatSession(ctx context.Context, arg DeleteChatSessionParams) error {
_, err := q.db.Exec(ctx, deleteChatSession, arg.ID, arg.WorkspaceID)
return err
}
const deleteUserChatMessageByTask = `-- name: DeleteUserChatMessageByTask :one
DELETE FROM chat_message
WHERE task_id = $1 AND role = 'user'
RETURNING id, chat_session_id, role, content, task_id, created_at, failure_reason, elapsed_ms, message_kind
`
func (q *Queries) DeleteUserChatMessageByTask(ctx context.Context, taskID pgtype.UUID) (ChatMessage, error) {
row := q.db.QueryRow(ctx, deleteUserChatMessageByTask, taskID)
var i ChatMessage
err := row.Scan(
&i.ID,
&i.ChatSessionID,
&i.Role,
&i.Content,
&i.TaskID,
&i.CreatedAt,
&i.FailureReason,
&i.ElapsedMs,
&i.MessageKind,
)
return i, err
}
const getChatMessage = `-- name: GetChatMessage :one
SELECT id, chat_session_id, role, content, task_id, created_at, failure_reason, elapsed_ms, message_kind FROM chat_message
WHERE id = $1
`
func (q *Queries) GetChatMessage(ctx context.Context, id pgtype.UUID) (ChatMessage, error) {
row := q.db.QueryRow(ctx, getChatMessage, id)
var i ChatMessage
err := row.Scan(
&i.ID,
&i.ChatSessionID,
&i.Role,
&i.Content,
&i.TaskID,
&i.CreatedAt,
&i.FailureReason,
&i.ElapsedMs,
&i.MessageKind,
)
return i, err
}
const getChatSession = `-- name: GetChatSession :one
SELECT id, workspace_id, agent_id, creator_id, title, session_id, work_dir, status, created_at, updated_at, unread_since, runtime_id, last_read_at, is_agent_intro, pinned_at, project_id FROM chat_session
WHERE id = $1
`
func (q *Queries) GetChatSession(ctx context.Context, id pgtype.UUID) (ChatSession, error) {
row := q.db.QueryRow(ctx, getChatSession, id)
var i ChatSession
err := row.Scan(
&i.ID,
&i.WorkspaceID,
&i.AgentID,
&i.CreatorID,
&i.Title,
&i.SessionID,
&i.WorkDir,
&i.Status,
&i.CreatedAt,
&i.UpdatedAt,
&i.UnreadSince,
&i.RuntimeID,
&i.LastReadAt,
&i.IsAgentIntro,
&i.PinnedAt,
&i.ProjectID,
)
return i, err
}
const getChatSessionInWorkspace = `-- name: GetChatSessionInWorkspace :one
SELECT id, workspace_id, agent_id, creator_id, title, session_id, work_dir, status, created_at, updated_at, unread_since, runtime_id, last_read_at, is_agent_intro, pinned_at, project_id FROM chat_session
WHERE id = $1 AND workspace_id = $2
`
type GetChatSessionInWorkspaceParams struct {
ID pgtype.UUID `json:"id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
}
func (q *Queries) GetChatSessionInWorkspace(ctx context.Context, arg GetChatSessionInWorkspaceParams) (ChatSession, error) {
row := q.db.QueryRow(ctx, getChatSessionInWorkspace, arg.ID, arg.WorkspaceID)
var i ChatSession
err := row.Scan(
&i.ID,
&i.WorkspaceID,
&i.AgentID,
&i.CreatorID,
&i.Title,
&i.SessionID,
&i.WorkDir,
&i.Status,
&i.CreatedAt,
&i.UpdatedAt,
&i.UnreadSince,
&i.RuntimeID,
&i.LastReadAt,
&i.IsAgentIntro,
&i.PinnedAt,
&i.ProjectID,
)
return i, err
}
const getLastChatTaskSession = `-- name: GetLastChatTaskSession :one
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', 'agent_error.context_overflow')
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
`
type GetLastChatTaskSessionRow struct {
SessionID pgtype.Text `json:"session_id"`
WorkDir pgtype.Text `json:"work_dir"`
RuntimeID pgtype.UUID `json:"runtime_id"`
}
// 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. Keep this list in sync with resumeUnsafeFailureReason
// and GetLastTaskSession.
func (q *Queries) GetLastChatTaskSession(ctx context.Context, chatSessionID pgtype.UUID) (GetLastChatTaskSessionRow, error) {
row := q.db.QueryRow(ctx, getLastChatTaskSession, chatSessionID)
var i GetLastChatTaskSessionRow
err := row.Scan(&i.SessionID, &i.WorkDir, &i.RuntimeID)
return i, err
}
const getMostRecentUserChatMessage = `-- name: GetMostRecentUserChatMessage :one
SELECT id, chat_session_id, role, content, task_id, created_at, failure_reason, elapsed_ms, message_kind FROM chat_message
WHERE chat_session_id = $1 AND role = 'user'
ORDER BY created_at DESC
LIMIT 1
`
// 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.
func (q *Queries) GetMostRecentUserChatMessage(ctx context.Context, chatSessionID pgtype.UUID) (ChatMessage, error) {
row := q.db.QueryRow(ctx, getMostRecentUserChatMessage, chatSessionID)
var i ChatMessage
err := row.Scan(
&i.ID,
&i.ChatSessionID,
&i.Role,
&i.Content,
&i.TaskID,
&i.CreatedAt,
&i.FailureReason,
&i.ElapsedMs,
&i.MessageKind,
)
return i, err
}
const getPendingChatTask = `-- name: GetPendingChatTask :one
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
`
type GetPendingChatTaskRow struct {
ID pgtype.UUID `json:"id"`
Status string `json:"status"`
CreatedAt pgtype.Timestamptz `json:"created_at"`
}
// 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".
func (q *Queries) GetPendingChatTask(ctx context.Context, chatSessionID pgtype.UUID) (GetPendingChatTaskRow, error) {
row := q.db.QueryRow(ctx, getPendingChatTask, chatSessionID)
var i GetPendingChatTaskRow
err := row.Scan(&i.ID, &i.Status, &i.CreatedAt)
return i, err
}
const hasPendingChatTasksByCreator = `-- name: HasPendingChatTasksByCreator :one
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 = $1
AND cs.creator_id = $2
AND cs.agent_id = ANY($3::uuid[])
) AS has_pending
`
type HasPendingChatTasksByCreatorParams struct {
WorkspaceID pgtype.UUID `json:"workspace_id"`
CreatorID pgtype.UUID `json:"creator_id"`
AgentIds []pgtype.UUID `json:"agent_ids"`
}
// 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.
func (q *Queries) HasPendingChatTasksByCreator(ctx context.Context, arg HasPendingChatTasksByCreatorParams) (bool, error) {
row := q.db.QueryRow(ctx, hasPendingChatTasksByCreator, arg.WorkspaceID, arg.CreatorID, arg.AgentIds)
var has_pending bool
err := row.Scan(&has_pending)
return has_pending, err
}
const linkChatMessageToTask = `-- name: LinkChatMessageToTask :exec
UPDATE chat_message
SET task_id = $2
WHERE id = $1 AND role = 'user'
`
type LinkChatMessageToTaskParams struct {
ID pgtype.UUID `json:"id"`
TaskID pgtype.UUID `json:"task_id"`
}
func (q *Queries) LinkChatMessageToTask(ctx context.Context, arg LinkChatMessageToTaskParams) error {
_, err := q.db.Exec(ctx, linkChatMessageToTask, arg.ID, arg.TaskID)
return err
}
const listAllChatSessionsByCreator = `-- name: ListAllChatSessionsByCreator :many
SELECT cs.id, cs.workspace_id, cs.agent_id, cs.creator_id, cs.title, cs.session_id, cs.work_dir, cs.status, cs.created_at, cs.updated_at, cs.unread_since, cs.runtime_id, cs.last_read_at, cs.is_agent_intro, cs.pinned_at, cs.project_id,
CASE WHEN cs.status = 'archived' THEN 0
ELSE (SELECT count(*) FROM chat_message m
WHERE m.chat_session_id = cs.id
AND m.role = 'assistant'
AND m.created_at > cs.last_read_at)
END::int AS unread_count,
COALESCE(lm.content, '') AS last_message_content,
COALESCE(lm.role, '') AS last_message_role,
lm.created_at AS last_message_at,
lm.failure_reason AS last_message_failure_reason,
COALESCE(lm.message_kind, '') AS last_message_kind
FROM chat_session cs
LEFT JOIN LATERAL (
SELECT content, role, created_at, failure_reason, message_kind
FROM chat_message m
WHERE m.chat_session_id = cs.id
ORDER BY m.created_at DESC
LIMIT 1
) lm ON true
WHERE cs.workspace_id = $1 AND cs.creator_id = $2
ORDER BY (cs.pinned_at IS NOT NULL) DESC, cs.pinned_at DESC, COALESCE(lm.created_at, cs.updated_at) DESC
`
type ListAllChatSessionsByCreatorParams struct {
WorkspaceID pgtype.UUID `json:"workspace_id"`
CreatorID pgtype.UUID `json:"creator_id"`
}
type ListAllChatSessionsByCreatorRow struct {
ID pgtype.UUID `json:"id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
AgentID pgtype.UUID `json:"agent_id"`
CreatorID pgtype.UUID `json:"creator_id"`
Title string `json:"title"`
SessionID pgtype.Text `json:"session_id"`
WorkDir pgtype.Text `json:"work_dir"`
Status string `json:"status"`
CreatedAt pgtype.Timestamptz `json:"created_at"`
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
UnreadSince pgtype.Timestamptz `json:"unread_since"`
RuntimeID pgtype.UUID `json:"runtime_id"`
LastReadAt pgtype.Timestamptz `json:"last_read_at"`
IsAgentIntro bool `json:"is_agent_intro"`
PinnedAt pgtype.Timestamptz `json:"pinned_at"`
ProjectID pgtype.UUID `json:"project_id"`
UnreadCount int32 `json:"unread_count"`
LastMessageContent string `json:"last_message_content"`
LastMessageRole string `json:"last_message_role"`
LastMessageAt pgtype.Timestamptz `json:"last_message_at"`
LastMessageFailureReason pgtype.Text `json:"last_message_failure_reason"`
LastMessageKind string `json:"last_message_kind"`
}
// Unlike ListChatSessionsByCreator this returns archived sessions too (for the
// "Archived" view), so unread must be forced to 0 for archived rows: archiving
// deliberately does NOT advance last_read_at (so unarchive can restore the true
// unread state), but an archived session is read-only and hidden from history,
// so any residual unread is uncleanable and must not light up any badge. Gating
// on status here is the single source of truth for all unread surfaces (FAB,
// sidebar Chat tab, chat-window header) — see MUL-4360.
func (q *Queries) ListAllChatSessionsByCreator(ctx context.Context, arg ListAllChatSessionsByCreatorParams) ([]ListAllChatSessionsByCreatorRow, error) {
rows, err := q.db.Query(ctx, listAllChatSessionsByCreator, arg.WorkspaceID, arg.CreatorID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ListAllChatSessionsByCreatorRow{}
for rows.Next() {
var i ListAllChatSessionsByCreatorRow
if err := rows.Scan(
&i.ID,
&i.WorkspaceID,
&i.AgentID,
&i.CreatorID,
&i.Title,
&i.SessionID,
&i.WorkDir,
&i.Status,
&i.CreatedAt,
&i.UpdatedAt,
&i.UnreadSince,
&i.RuntimeID,
&i.LastReadAt,
&i.IsAgentIntro,
&i.PinnedAt,
&i.ProjectID,
&i.UnreadCount,
&i.LastMessageContent,
&i.LastMessageRole,
&i.LastMessageAt,
&i.LastMessageFailureReason,
&i.LastMessageKind,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listChatDraftRestoresBySession = `-- name: ListChatDraftRestoresBySession :many
SELECT id, chat_session_id, task_id, content, attachment_ids, created_at FROM chat_draft_restore
WHERE chat_session_id = $1
ORDER BY created_at ASC
`
func (q *Queries) ListChatDraftRestoresBySession(ctx context.Context, chatSessionID pgtype.UUID) ([]ChatDraftRestore, error) {
rows, err := q.db.Query(ctx, listChatDraftRestoresBySession, chatSessionID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ChatDraftRestore{}
for rows.Next() {
var i ChatDraftRestore
if err := rows.Scan(
&i.ID,
&i.ChatSessionID,
&i.TaskID,
&i.Content,
&i.AttachmentIds,
&i.CreatedAt,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listChatInputMessages = `-- name: ListChatInputMessages :many
SELECT id, chat_session_id, role, content, task_id, created_at, failure_reason, elapsed_ms, message_kind FROM chat_message
WHERE task_id = $1 AND role = 'user'
ORDER BY created_at ASC, id ASC
`
// Loads the immutable user-message input batch owned by a direct-chat task.
// The caller passes the task's chat_input_task_id (itself for an original send,
// the root task for an auto-retry child), so a claim reads exactly the messages
// the user sent for this turn — and never absorbs a message that arrived after
// the batch was sealed, no matter what the assistant wrote or when. Only used
// for new task-owned direct-chat tasks; legacy/channel (chat_input_task_id
// NULL) tasks keep using ListChatMessages + trailingUserMessages.
func (q *Queries) ListChatInputMessages(ctx context.Context, taskID pgtype.UUID) ([]ChatMessage, error) {
rows, err := q.db.Query(ctx, listChatInputMessages, taskID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ChatMessage{}
for rows.Next() {
var i ChatMessage
if err := rows.Scan(
&i.ID,
&i.ChatSessionID,
&i.Role,
&i.Content,
&i.TaskID,
&i.CreatedAt,
&i.FailureReason,
&i.ElapsedMs,
&i.MessageKind,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listChatMessages = `-- name: ListChatMessages :many
SELECT id, chat_session_id, role, content, task_id, created_at, failure_reason, elapsed_ms, message_kind FROM chat_message
WHERE chat_session_id = $1
ORDER BY created_at ASC
`
func (q *Queries) ListChatMessages(ctx context.Context, chatSessionID pgtype.UUID) ([]ChatMessage, error) {
rows, err := q.db.Query(ctx, listChatMessages, chatSessionID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ChatMessage{}
for rows.Next() {
var i ChatMessage
if err := rows.Scan(
&i.ID,
&i.ChatSessionID,
&i.Role,
&i.Content,
&i.TaskID,
&i.CreatedAt,
&i.FailureReason,
&i.ElapsedMs,
&i.MessageKind,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listChatMessagesPage = `-- name: ListChatMessagesPage :many
SELECT id, chat_session_id, role, content, task_id, created_at, failure_reason, elapsed_ms, message_kind FROM chat_message
WHERE chat_session_id = $1
AND (
$3::timestamptz IS NULL
OR (created_at, id) < ($3::timestamptz, $4::uuid)
)
ORDER BY created_at DESC, id DESC
LIMIT $2
`
type ListChatMessagesPageParams struct {
ChatSessionID pgtype.UUID `json:"chat_session_id"`
Limit int32 `json:"limit"`
BeforeCreatedAt pgtype.Timestamptz `json:"before_created_at"`
BeforeID pgtype.UUID `json:"before_id"`
}
func (q *Queries) ListChatMessagesPage(ctx context.Context, arg ListChatMessagesPageParams) ([]ChatMessage, error) {
rows, err := q.db.Query(ctx, listChatMessagesPage,
arg.ChatSessionID,
arg.Limit,
arg.BeforeCreatedAt,
arg.BeforeID,
)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ChatMessage{}
for rows.Next() {
var i ChatMessage
if err := rows.Scan(
&i.ID,
&i.ChatSessionID,
&i.Role,
&i.Content,
&i.TaskID,
&i.CreatedAt,
&i.FailureReason,
&i.ElapsedMs,
&i.MessageKind,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listChatSessionsByCreator = `-- name: ListChatSessionsByCreator :many
SELECT cs.id, cs.workspace_id, cs.agent_id, cs.creator_id, cs.title, cs.session_id, cs.work_dir, cs.status, cs.created_at, cs.updated_at, cs.unread_since, cs.runtime_id, cs.last_read_at, cs.is_agent_intro, cs.pinned_at, cs.project_id,
(SELECT count(*) FROM chat_message m
WHERE m.chat_session_id = cs.id
AND m.role = 'assistant'
AND m.created_at > cs.last_read_at)::int AS unread_count,
COALESCE(lm.content, '') AS last_message_content,
COALESCE(lm.role, '') AS last_message_role,
lm.created_at AS last_message_at,
lm.failure_reason AS last_message_failure_reason,
COALESCE(lm.message_kind, '') AS last_message_kind
FROM chat_session cs
LEFT JOIN LATERAL (
SELECT content, role, created_at, failure_reason, message_kind
FROM chat_message m
WHERE m.chat_session_id = cs.id
ORDER BY m.created_at DESC
LIMIT 1
) lm ON true
WHERE cs.workspace_id = $1 AND cs.creator_id = $2 AND cs.status = 'active'
ORDER BY (cs.pinned_at IS NOT NULL) DESC, cs.pinned_at DESC, COALESCE(lm.created_at, cs.updated_at) DESC
`
type ListChatSessionsByCreatorParams struct {
WorkspaceID pgtype.UUID `json:"workspace_id"`
CreatorID pgtype.UUID `json:"creator_id"`
}
type ListChatSessionsByCreatorRow struct {
ID pgtype.UUID `json:"id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
AgentID pgtype.UUID `json:"agent_id"`
CreatorID pgtype.UUID `json:"creator_id"`
Title string `json:"title"`
SessionID pgtype.Text `json:"session_id"`
WorkDir pgtype.Text `json:"work_dir"`
Status string `json:"status"`
CreatedAt pgtype.Timestamptz `json:"created_at"`
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
UnreadSince pgtype.Timestamptz `json:"unread_since"`
RuntimeID pgtype.UUID `json:"runtime_id"`
LastReadAt pgtype.Timestamptz `json:"last_read_at"`
IsAgentIntro bool `json:"is_agent_intro"`
PinnedAt pgtype.Timestamptz `json:"pinned_at"`
ProjectID pgtype.UUID `json:"project_id"`
UnreadCount int32 `json:"unread_count"`
LastMessageContent string `json:"last_message_content"`
LastMessageRole string `json:"last_message_role"`
LastMessageAt pgtype.Timestamptz `json:"last_message_at"`
LastMessageFailureReason pgtype.Text `json:"last_message_failure_reason"`
LastMessageKind string `json:"last_message_kind"`
}
// IM-style list: each active session with its unread *count* (assistant
// messages after the read cursor), a preview of the latest message, and
// ordered by most-recent activity so a new reply bumps a session to the top.
func (q *Queries) ListChatSessionsByCreator(ctx context.Context, arg ListChatSessionsByCreatorParams) ([]ListChatSessionsByCreatorRow, error) {
rows, err := q.db.Query(ctx, listChatSessionsByCreator, arg.WorkspaceID, arg.CreatorID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ListChatSessionsByCreatorRow{}
for rows.Next() {
var i ListChatSessionsByCreatorRow
if err := rows.Scan(
&i.ID,
&i.WorkspaceID,
&i.AgentID,
&i.CreatorID,
&i.Title,
&i.SessionID,
&i.WorkDir,
&i.Status,
&i.CreatedAt,
&i.UpdatedAt,
&i.UnreadSince,
&i.RuntimeID,
&i.LastReadAt,
&i.IsAgentIntro,
&i.PinnedAt,
&i.ProjectID,
&i.UnreadCount,
&i.LastMessageContent,
&i.LastMessageRole,
&i.LastMessageAt,
&i.LastMessageFailureReason,
&i.LastMessageKind,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listPendingChatTasksByCreator = `-- name: ListPendingChatTasksByCreator :many
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
`
type ListPendingChatTasksByCreatorParams struct {
WorkspaceID pgtype.UUID `json:"workspace_id"`
CreatorID pgtype.UUID `json:"creator_id"`
}
type ListPendingChatTasksByCreatorRow struct {
TaskID pgtype.UUID `json:"task_id"`
Status string `json:"status"`
ChatSessionID pgtype.UUID `json:"chat_session_id"`
AgentID pgtype.UUID `json:"agent_id"`
}
// 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.
func (q *Queries) ListPendingChatTasksByCreator(ctx context.Context, arg ListPendingChatTasksByCreatorParams) ([]ListPendingChatTasksByCreatorRow, error) {
rows, err := q.db.Query(ctx, listPendingChatTasksByCreator, arg.WorkspaceID, arg.CreatorID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ListPendingChatTasksByCreatorRow{}
for rows.Next() {
var i ListPendingChatTasksByCreatorRow
if err := rows.Scan(
&i.TaskID,
&i.Status,
&i.ChatSessionID,
&i.AgentID,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const lockChatSessionForDelete = `-- name: LockChatSessionForDelete :one
SELECT id FROM chat_session
WHERE id = $1
FOR UPDATE
`
// 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.
func (q *Queries) LockChatSessionForDelete(ctx context.Context, id pgtype.UUID) (pgtype.UUID, error) {
row := q.db.QueryRow(ctx, lockChatSessionForDelete, id)
var id_2 pgtype.UUID
err := row.Scan(&id_2)
return id_2, err
}
const lockChatSessionForRuntimeBind = `-- name: LockChatSessionForRuntimeBind :one
SELECT id FROM chat_session
WHERE id = $1
FOR UPDATE
`
// Acquires an exclusive (FOR UPDATE) row lock on chat_session(id), serialising
// "which runtime does this session execute on" against "enqueue the next task".
//
// Both SendDirectChatMessage and the agent-builder runtime switch take this lock
// for their whole transaction. Without it the two are a read-then-write race: a
// send reads the carrier agent's runtime_id, the switch then passes its
// pending-task check and rebinds the carrier, and the send finally inserts a task
// still stamped with the pre-switch runtime — so the user is told the switch
// succeeded while their message runs on the old runtime (MUL-5163).
//
// The lock alone is not sufficient: the send path must also re-read the agent
// INSIDE the locked transaction, because a send blocked at INSERT would otherwise
// resume and write the runtime_id it read before blocking.
//
// Same row and same lock mode as LockChatSessionForDelete, and both take it as
// their first statement, so the delete path and this one cannot deadlock.
func (q *Queries) LockChatSessionForRuntimeBind(ctx context.Context, id pgtype.UUID) (pgtype.UUID, error) {
row := q.db.QueryRow(ctx, lockChatSessionForRuntimeBind, id)
var id_2 pgtype.UUID
err := row.Scan(&id_2)
return id_2, err
}
const lockChatSessionForTask = `-- name: LockChatSessionForTask :one
SELECT cs.id
FROM agent_task_queue t
JOIN chat_session cs ON cs.id = t.chat_session_id
WHERE t.id = $1
FOR UPDATE OF cs
`
// The chat_session row lock is the mutual-exclusion protocol between the
// draft-restore writer (FinalizeDeferredCancelledChat) and every deleter of a
// chat_session or one of its cascade parents. Without an FK, an INSERT into
// chat_draft_restore takes no lock on its session, so a prune-then-delete would
// otherwise miss a restore committed after its snapshot and strand it — with the
// user's prompt in it — forever (#5219).
//
// The contract, held by all five paths below:
//
// deleter: lock the sessions FOR UPDATE -> prune restores -> delete the parent
// finalizer: lock the session FOR UPDATE -> insert the restore
//
// Whoever locks first wins: a finalizer that got there first commits its row
// before the deleter's prune statement takes its snapshot, so the prune sweeps
// it; a deleter that got there first leaves no session for the finalizer to
// lock, so it never inserts.
//
// Lock order is chat_session -> agent_task_queue everywhere (the finalizer locks
// the session before claiming its task) so the deleters' cascade into
// agent_task_queue cannot deadlock against it.
//
// The single-session delete path needs no new query: it already holds
// LockChatSessionForDelete (same FOR UPDATE row lock) across its prune.
// The finalizer's half of the protocol. No rows means the session is already
// gone (its cascade NULLs agent_task_queue.chat_session_id), so there is nothing
// to lock and nothing to restore into.
func (q *Queries) LockChatSessionForTask(ctx context.Context, id pgtype.UUID) (pgtype.UUID, error) {
row := q.db.QueryRow(ctx, lockChatSessionForTask, id)
var id_2 pgtype.UUID
err := row.Scan(&id_2)
return id_2, err
}
const lockChatSessionsByArchivedRuntimeAgents = `-- name: LockChatSessionsByArchivedRuntimeAgents :many
SELECT cs.id FROM chat_session cs
JOIN agent a ON a.id = cs.agent_id
WHERE a.runtime_id = $1 AND a.archived_at IS NOT NULL
ORDER BY cs.id
FOR UPDATE OF cs
`
func (q *Queries) LockChatSessionsByArchivedRuntimeAgents(ctx context.Context, runtimeID pgtype.UUID) ([]pgtype.UUID, error) {
rows, err := q.db.Query(ctx, lockChatSessionsByArchivedRuntimeAgents, runtimeID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []pgtype.UUID{}
for rows.Next() {
var id pgtype.UUID
if err := rows.Scan(&id); err != nil {
return nil, err
}
items = append(items, id)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const lockChatSessionsBySystemRuntimeAgents = `-- name: LockChatSessionsBySystemRuntimeAgents :many
SELECT cs.id FROM chat_session cs
JOIN agent a ON a.id = cs.agent_id
WHERE a.runtime_id = $1 AND a.kind = 'system'
ORDER BY cs.id
FOR UPDATE OF cs
`
func (q *Queries) LockChatSessionsBySystemRuntimeAgents(ctx context.Context, runtimeID pgtype.UUID) ([]pgtype.UUID, error) {
rows, err := q.db.Query(ctx, lockChatSessionsBySystemRuntimeAgents, runtimeID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []pgtype.UUID{}
for rows.Next() {
var id pgtype.UUID
if err := rows.Scan(&id); err != nil {
return nil, err
}
items = append(items, id)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const lockChatSessionsByWorkspace = `-- name: LockChatSessionsByWorkspace :many
SELECT id FROM chat_session
WHERE workspace_id = $1
ORDER BY id
FOR UPDATE
`
// ORDER BY id: a stable lock order keeps two concurrent deleters from
// deadlocking against each other.
func (q *Queries) LockChatSessionsByWorkspace(ctx context.Context, workspaceID pgtype.UUID) ([]pgtype.UUID, error) {
rows, err := q.db.Query(ctx, lockChatSessionsByWorkspace, workspaceID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []pgtype.UUID{}
for rows.Next() {
var id pgtype.UUID
if err := rows.Scan(&id); err != nil {
return nil, err
}
items = append(items, id)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const markChatSessionRead = `-- name: MarkChatSessionRead :exec
UPDATE chat_session SET last_read_at = now()
WHERE id = $1
`
// Advances the read cursor to now, dropping the session's unread_count to 0.
func (q *Queries) MarkChatSessionRead(ctx context.Context, id pgtype.UUID) error {
_, err := q.db.Exec(ctx, markChatSessionRead, id)
return err
}
const setChatSessionArchived = `-- name: SetChatSessionArchived :one
UPDATE chat_session
SET status = CASE WHEN $2::bool THEN 'archived' ELSE 'active' END,
updated_at = now()
WHERE id = $1
RETURNING id, workspace_id, agent_id, creator_id, title, session_id, work_dir, status, created_at, updated_at, unread_since, runtime_id, last_read_at, is_agent_intro, pinned_at, project_id
`
type SetChatSessionArchivedParams struct {
ID pgtype.UUID `json:"id"`
Archived bool `json:"archived"`
}
// Archive/unarchive a chat session by flipping status between 'active' and
// 'archived'. Bumps updated_at so the row re-sorts on the receiving list. The
// send-message path refuses archived sessions (see SendChatMessage), so the
// conversation is effectively read-only until it is unarchived.
func (q *Queries) SetChatSessionArchived(ctx context.Context, arg SetChatSessionArchivedParams) (ChatSession, error) {
row := q.db.QueryRow(ctx, setChatSessionArchived, arg.ID, arg.Archived)
var i ChatSession
err := row.Scan(
&i.ID,
&i.WorkspaceID,
&i.AgentID,
&i.CreatorID,
&i.Title,
&i.SessionID,
&i.WorkDir,
&i.Status,
&i.CreatedAt,
&i.UpdatedAt,
&i.UnreadSince,
&i.RuntimeID,
&i.LastReadAt,
&i.IsAgentIntro,
&i.PinnedAt,
&i.ProjectID,
)
return i, err
}
const setChatSessionPinned = `-- name: SetChatSessionPinned :one
UPDATE chat_session
SET pinned_at = CASE WHEN $2::bool THEN COALESCE(pinned_at, now()) ELSE NULL END
WHERE id = $1
RETURNING id, workspace_id, agent_id, creator_id, title, session_id, work_dir, status, created_at, updated_at, unread_since, runtime_id, last_read_at, is_agent_intro, pinned_at, project_id
`
type SetChatSessionPinnedParams struct {
ID pgtype.UUID `json:"id"`
Pinned bool `json:"pinned"`
}
// Pin/unpin a chat. Deliberately does NOT touch updated_at: pinning is a
// list-ordering preference, not activity, so it must not bump the session's
// last-activity sort key (which would make an unpinned chat jump the list).
// pinned = true stamps pinned_at only when it was NULL, so re-pinning keeps
// the original pin order; pinned = false clears it.
func (q *Queries) SetChatSessionPinned(ctx context.Context, arg SetChatSessionPinnedParams) (ChatSession, error) {
row := q.db.QueryRow(ctx, setChatSessionPinned, arg.ID, arg.Pinned)
var i ChatSession
err := row.Scan(
&i.ID,
&i.WorkspaceID,
&i.AgentID,
&i.CreatorID,
&i.Title,
&i.SessionID,
&i.WorkDir,
&i.Status,
&i.CreatedAt,
&i.UpdatedAt,
&i.UnreadSince,
&i.RuntimeID,
&i.LastReadAt,
&i.IsAgentIntro,
&i.PinnedAt,
&i.ProjectID,
)
return i, err
}
const setChatTaskInputOwnerSelf = `-- name: SetChatTaskInputOwnerSelf :one
UPDATE agent_task_queue
SET chat_input_task_id = id
WHERE id = $1
RETURNING id, agent_id, issue_id, status, priority, dispatched_at, started_at, completed_at, result, error, created_at, context, runtime_id, session_id, work_dir, trigger_comment_id, chat_session_id, autopilot_run_id, attempt, max_attempts, parent_task_id, failure_reason, trigger_summary, force_fresh_session, is_leader_task, wait_reason, initiator_user_id, handoff_note, prepare_lease_expires_at, squad_id, runtime_mcp_overlay, escalation_for_task_id, fire_at, originator_user_id, runtime_connected_apps, coalesced_comment_ids, delivered_comment_ids, chat_input_task_id, chat_finalize_deferred_at, originator_source, delegated_from_task_id, retry_of_task_id, rerun_of_task_id, rule_version_id, trigger_evidence_kind, trigger_evidence_ref_id, accountable_user_id, session_rollout_missing
`
// Stamps a freshly-created direct-chat task as the owner of its own input batch
// (chat_input_task_id = id), so a later claim loads exactly the user messages
// tagged with this task id (ListChatInputMessages) rather than scanning trailing
// history. Runs in the same transaction as CreateChatTask + the user message
// insert on the direct-send path. Channel and legacy tasks skip this call and
// keep chat_input_task_id NULL, so a rolling deploy never replays their history.
func (q *Queries) SetChatTaskInputOwnerSelf(ctx context.Context, id pgtype.UUID) (AgentTaskQueue, error) {
row := q.db.QueryRow(ctx, setChatTaskInputOwnerSelf, id)
var i AgentTaskQueue
err := row.Scan(
&i.ID,
&i.AgentID,
&i.IssueID,
&i.Status,
&i.Priority,
&i.DispatchedAt,
&i.StartedAt,
&i.CompletedAt,
&i.Result,
&i.Error,
&i.CreatedAt,
&i.Context,
&i.RuntimeID,
&i.SessionID,
&i.WorkDir,
&i.TriggerCommentID,
&i.ChatSessionID,
&i.AutopilotRunID,
&i.Attempt,
&i.MaxAttempts,
&i.ParentTaskID,
&i.FailureReason,
&i.TriggerSummary,
&i.ForceFreshSession,
&i.IsLeaderTask,
&i.WaitReason,
&i.InitiatorUserID,
&i.HandoffNote,
&i.PrepareLeaseExpiresAt,
&i.SquadID,
&i.RuntimeMcpOverlay,
&i.EscalationForTaskID,
&i.FireAt,
&i.OriginatorUserID,
&i.RuntimeConnectedApps,
&i.CoalescedCommentIds,
&i.DeliveredCommentIds,
&i.ChatInputTaskID,
&i.ChatFinalizeDeferredAt,
&i.OriginatorSource,
&i.DelegatedFromTaskID,
&i.RetryOfTaskID,
&i.RerunOfTaskID,
&i.RuleVersionID,
&i.TriggerEvidenceKind,
&i.TriggerEvidenceRefID,
&i.AccountableUserID,
&i.SessionRolloutMissing,
)
return i, err
}
const touchChatSession = `-- name: TouchChatSession :exec
UPDATE chat_session SET updated_at = now()
WHERE id = $1
`
func (q *Queries) TouchChatSession(ctx context.Context, id pgtype.UUID) error {
_, err := q.db.Exec(ctx, touchChatSession, id)
return err
}
const updateChatSessionProject = `-- name: UpdateChatSessionProject :one
UPDATE chat_session
SET project_id = $1
WHERE id = $2 AND workspace_id = $3
RETURNING id, workspace_id, agent_id, creator_id, title, session_id, work_dir, status, created_at, updated_at, unread_since, runtime_id, last_read_at, is_agent_intro, pinned_at, project_id
`
type UpdateChatSessionProjectParams struct {
ProjectID pgtype.UUID `json:"project_id"`
ID pgtype.UUID `json:"id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
}
// Project context is user-editable session metadata. Do not touch updated_at:
// changing context is not conversation activity and must not reorder history.
func (q *Queries) UpdateChatSessionProject(ctx context.Context, arg UpdateChatSessionProjectParams) (ChatSession, error) {
row := q.db.QueryRow(ctx, updateChatSessionProject, arg.ProjectID, arg.ID, arg.WorkspaceID)
var i ChatSession
err := row.Scan(
&i.ID,
&i.WorkspaceID,
&i.AgentID,
&i.CreatorID,
&i.Title,
&i.SessionID,
&i.WorkDir,
&i.Status,
&i.CreatedAt,
&i.UpdatedAt,
&i.UnreadSince,
&i.RuntimeID,
&i.LastReadAt,
&i.IsAgentIntro,
&i.PinnedAt,
&i.ProjectID,
)
return i, err
}
const updateChatSessionSession = `-- name: UpdateChatSessionSession :exec
UPDATE chat_session
SET session_id = COALESCE($1, session_id),
work_dir = COALESCE($2, work_dir),
runtime_id = COALESCE($3, runtime_id),
updated_at = now()
WHERE id = $4
`
type UpdateChatSessionSessionParams struct {
SessionID pgtype.Text `json:"session_id"`
WorkDir pgtype.Text `json:"work_dir"`
RuntimeID pgtype.UUID `json:"runtime_id"`
ID pgtype.UUID `json:"id"`
}
// 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.
func (q *Queries) UpdateChatSessionSession(ctx context.Context, arg UpdateChatSessionSessionParams) error {
_, err := q.db.Exec(ctx, updateChatSessionSession,
arg.SessionID,
arg.WorkDir,
arg.RuntimeID,
arg.ID,
)
return err
}
const updateChatSessionTitle = `-- name: UpdateChatSessionTitle :one
UPDATE chat_session SET title = $2, updated_at = now()
WHERE id = $1
RETURNING id, workspace_id, agent_id, creator_id, title, session_id, work_dir, status, created_at, updated_at, unread_since, runtime_id, last_read_at, is_agent_intro, pinned_at, project_id
`
type UpdateChatSessionTitleParams struct {
ID pgtype.UUID `json:"id"`
Title string `json:"title"`
}
func (q *Queries) UpdateChatSessionTitle(ctx context.Context, arg UpdateChatSessionTitleParams) (ChatSession, error) {
row := q.db.QueryRow(ctx, updateChatSessionTitle, arg.ID, arg.Title)
var i ChatSession
err := row.Scan(
&i.ID,
&i.WorkspaceID,
&i.AgentID,
&i.CreatorID,
&i.Title,
&i.SessionID,
&i.WorkDir,
&i.Status,
&i.CreatedAt,
&i.UpdatedAt,
&i.UnreadSince,
&i.RuntimeID,
&i.LastReadAt,
&i.IsAgentIntro,
&i.PinnedAt,
&i.ProjectID,
)
return i, err
}
const updateChatSessionTitleIfCurrent = `-- name: UpdateChatSessionTitleIfCurrent :one
UPDATE chat_session SET title = $1, updated_at = now()
WHERE id = $2 AND title = $3
RETURNING id, workspace_id, agent_id, creator_id, title, session_id, work_dir, status, created_at, updated_at, unread_since, runtime_id, last_read_at, is_agent_intro, pinned_at, project_id
`
type UpdateChatSessionTitleIfCurrentParams struct {
NewTitle string `json:"new_title"`
ID pgtype.UUID `json:"id"`
ExpectedTitle string `json:"expected_title"`
}
// Compare-and-swap the title: only overwrite it when it still equals the
// value the caller observed (@expected_title). This is the idempotency /
// no-clobber guard behind LLM auto-titling (MUL-4295): the async generator
// captures the session's current (default/original) title before calling the
// model, and this write lands only if a manual rename or a competing writer
// has not changed the title in the meantime. A mismatch returns pgx.ErrNoRows
// (zero rows updated), which the caller treats as "someone renamed it — leave
// it alone", NOT as an error.
func (q *Queries) UpdateChatSessionTitleIfCurrent(ctx context.Context, arg UpdateChatSessionTitleIfCurrentParams) (ChatSession, error) {
row := q.db.QueryRow(ctx, updateChatSessionTitleIfCurrent, arg.NewTitle, arg.ID, arg.ExpectedTitle)
var i ChatSession
err := row.Scan(
&i.ID,
&i.WorkspaceID,
&i.AgentID,
&i.CreatorID,
&i.Title,
&i.SessionID,
&i.WorkDir,
&i.Status,
&i.CreatedAt,
&i.UpdatedAt,
&i.UnreadSince,
&i.RuntimeID,
&i.LastReadAt,
&i.IsAgentIntro,
&i.PinnedAt,
&i.ProjectID,
)
return i, err
}