Files
Bohan Jiangandmultica-agent 1df4e0a8bf MUL-7477 fix(telegram): one reply per turn across replicas, restarts and retries (#8545)
* feat(db): record who owns a channel reply and what the provider accepted

A task's reply reaches a chat through three code paths — the streamed
placeholder, the final answer, the failure notice — and across replicas
through three processes: the daemon's transcript report and its completion
callback are independent HTTP requests, and the event bus is in-process. There
is nowhere for them to agree on whether a reply has already been sent.

channel_reply_delivery is that place: one row per task, carrying the phase the
reply is in, whether a send is out, and which message id it produced.

No foreign keys, and each index builds CONCURRENTLY in its own migration.
Workspace teardown deletes the rows alongside channel_outbound_message.

Co-authored-by: multica-agent <[email protected]>

* fix(telegram): deliver one reply per turn, across replicas and restarts

Users see every reply twice (#8049, #7750). #8117 closed the window where a
chat:done landed while the placeholder's sendMessage was still in flight, but
the reply's owner was an in-process map, so the duplicate simply came through
another door — most often the one #8117 could not reach at all: on the cloud
deployment the text report and the completion callback land on different
replicas, and the replica that cannot see the placeholder posts its own copy
of the same answer.

Every path now claims channel_reply_delivery before touching Telegram:

- The placeholder send is claimed in the database, so exactly one process
  opens a message and every other path edits it — including a process that
  never saw the stream.
- A send whose response was lost is recorded as unknown rather than absent.
  sendMessage takes no caller-supplied idempotency key, so re-sending cannot
  be deduplicated by the provider: delivery stops and the row says why.
- Terminal delivery claims from any phase but settled, so a second completion
  event is dropped while an interrupted delivery resumes from the chunk it
  reached. Chunk progress is written after each part, not after the last.
- Edit failures are classified instead of falling back to a fresh send:
  markup is retried as plain text, an ambiguous failure is retried on the same
  message under a budget, a permanently rejected edit settles the reply rather
  than spinning (which held the session's queue and blocked later answers),
  and only a target Telegram confirms is gone justifies a new message.
- The failure notice goes through the same ownership and the same
  classification, so it replaces the placeholder instead of appearing beside it.
- Cancelled and empty completions close the reply, so a text frame still in
  flight cannot reopen it.
- An automatic retry inherits the placeholder its previous attempt left behind
  instead of answering beside it. Scoped to a parked row, so two ordinary
  turns that happen to say the same thing still get one reply each.

The regression tests are the audit probes written against this bug: 12 of the
14 failed on main, and the two that passed are controls that must keep
passing — a normal streamed reply, and two separate turns with identical text.

Co-authored-by: multica-agent <[email protected]>

* refactor(db): key reply delivery by user turn and guard it with a lease

Review of the first version found the state model, not the implementation, to
be the problem. Three invariants were missing and are now in the schema:

- A placeholder is not progress. Recording the streamed message used to bump
  chunks_sent, which terminal delivery then read as "the first part of the
  final answer is already out" — so the answer was never written. The columns
  are now independent.
- One turn, one owner. owner_token / owner_expires_at are a lease: a path
  proves it holds the turn before calling the provider and proves it still
  holds it when recording the outcome. An atomic UPDATE was never the same
  thing as a single delivery, and two processes could both claim a reply.
- Ownership follows the real retry lineage. The row is keyed by the root of
  the agent_task_queue retry_of_task_id chain instead of "the most recently
  parked reply on this binding", which let an ordinary new request adopt an
  unrelated one and left the previous attempt able to reopen its own.

An expired lease frees a turn whose owner died, but never resets an
outstanding send to "not sent": there is no idempotency key to make that safe.

Co-authored-by: multica-agent <[email protected]>

* fix(telegram): one reply per turn across replicas, restarts and retries

Follow-up to the review of the first attempt, which shipped a broken success
path: its in-memory stand-in for the ownership table disagreed with the SQL
about one column, and the whole suite passed while the final answer never
reached Telegram. The tests no longer model that state machine — every
ownership query in this package now runs for real against Postgres.

Beyond the truncation, the review found seven ways a reply could still arrive
twice, be overwritten, or block the session behind it. All are covered here:

- Terminal delivery is exclusive. Two processes could both take a reply and
  both send it; taking the turn is now a lease, re-proven before every call.
- Chunk sends get the same published-before, recorded-after treatment the
  placeholder had, so an interrupted multi-part answer resumes after the last
  part instead of repeating it, and a lost response ends delivery rather than
  risking a second copy.
- A send left in flight by a process that died no longer blocks its session
  forever: the lease expires, the successor inherits it as unknown, and the
  turn settles instead of waiting on an owner that will never return.
- A streaming edit re-proves the turn immediately before it edits. Guarding
  only the first send left a stalled frame able to overwrite a final answer
  another replica had already published.
- Cancelled and empty completions create the closing record when the turn
  never reached Telegram at all, so a first frame arriving afterwards cannot
  open a placeholder nothing will finish.
- The failure notice honours an in-flight or unknown send instead of posting a
  second bubble about a turn the user may already have been answered about.
- Delivery without ownership is gone. It was written up as an accepted risk;
  it was a default duplicate path. A reply that cannot be owned is reported as
  a failure, on a budget that does not hold up the session.

Both suites — the delivery contract and the review's boundary cases — run by
default, and against real SQL, because that is what the fake hid.

Co-authored-by: multica-agent <[email protected]>

* refactor(db): order reply attempts and index the paths that delete them

Keying reply ownership by user turn stopped a retry from opening a second
answer, but left every attempt of that turn equal: a late frame from an
attempt the chain had already moved past could take the turn back and rewrite
what the user was reading with content from a superseded run. attempt_depth
records how far down the retry chain the holder sits, and the claim only ever
moves forward.

The table is also deleted by installation and by binding — workspace teardown,
a re-pointed installation, a retired chat-session binding — and had an index
on neither, so each of those was a full scan that grows with every turn. Both
are indexed now, and the three teardown paths that clean channel_outbound_
message clean this table alongside it, which the no-foreign-key rule requires.

Co-authored-by: multica-agent <[email protected]>

* fix(telegram): close the remaining duplicate and lost-notice paths

Review of the lease design found five more ways a reply could go wrong, each
fixed with a test that fails without it:

- A 5xx was treated as proof that nothing was posted. Telegram can accept a
  sendMessage and fail on the way back; the reply then went out twice. Only a
  4xx is evidence of a message that never landed.
- A failed retry-lineage lookup fell back to "this task is its own turn",
  which opens a second reply beside the one the previous attempt is holding.
  It is reported now, and delivery waits or stops rather than guessing.
- An attempt the retry chain had moved past could re-acquire its turn and
  overwrite the visible answer with its own stale tail.
- A reply settled by another replica left this process holding the stream and
  its chat-scheduler reference with nothing left to release them. Enough of
  those and the scheduler refuses new chats, so streaming stops for chats with
  nothing wrong with them. The next frame releases them, and an idle sweep
  covers turns where no next frame ever comes.
- The failure notice was posted inline and simply dropped whenever another
  path held the turn at that instant — likely, since that is exactly when a
  text frame is mid-request on another replica — leaving the placeholder as
  the last thing the user saw. It is terminal work now, so it waits for the
  turn like the answer does.

The notice moving to the queue also takes the last Telegram call off the
synchronous event bus.

Co-authored-by: multica-agent <[email protected]>

* fix(db): upgrade installed databases to attempt_depth, and guard the close path

attempt_depth was added by editing migration 500, which every database that
already installed this feature has recorded. Those databases never saw the
change, so every query touching the column failed on exactly the deployments
where the table already existed — and rebuilding from scratch passed, which is
why it went unnoticed. It moves to its own forward migration, written so a
database created by 500 and one upgraded through 504 end up identical, with a
test that runs the upgrade against the old table shape rather than a fresh one.

The close path also skipped the ordering the claim enforces. A cancellation or
empty completion from an attempt the retry chain had moved past could settle
the whole turn, and the retry's answer was then dropped as already delivered —
the user gets nothing. It now follows the same one-way rule, and a close that
belongs to a superseded attempt is a no-op rather than something to wait for.

Co-authored-by: multica-agent <[email protected]>

* fix(telegram): bound the notice to one call per step, and sweep on evidence

Two paths still escaped the ownership model:

The failure notice looped inside a single step — up to two edit attempts, and
a 429 wait taken inline. A step can then outlast the lease by minutes, so the
call after the wait lands on a turn another replica has taken over and may
already have finished. It is now shaped like the answer's delivery: at most
one Telegram call, hand back a retryAt, and let the next step re-prove the
turn first.

The stale-stream sweep treated thirty quiet minutes as abandoned. A run's
silence budget is hours — a task can sit in a tool call or a test run far
longer than any interval worth picking — so the sweep was free to reclaim a
live reply, after which the next frame would rebuild the stream with an empty
accumulator and edit the existing message down to just that fragment. Quiet
now only makes a stream a candidate; what releases it is the delivery row
saying the turn is settled or that a later attempt holds it.

Co-authored-by: multica-agent <[email protected]>

* chore(db): renumber the reply-delivery migrations above main's

main landed 500 and 501 while this branch was in review, so the numeric
prefixes collided on the merge ref and TestMigrationNumericPrefixesAreUnique
failed in CI. These move to 502-506.

Renumbering is safe for a database that already applied the old numbers: every
file here is idempotent — CREATE TABLE IF NOT EXISTS, CREATE INDEX
CONCURRENTLY IF NOT EXISTS, ADD COLUMN IF NOT EXISTS — so re-running them
under the new versions is a no-op that only records the new rows. Verified
against the local database, which had the old numbering applied.

Co-authored-by: multica-agent <[email protected]>

* fix(telegram): keep a provider call inside the lease that authorises it

The lease is only meaningful if a call made under it cannot outlive it. The
shared Bot API client allows 65s — which getUpdates needs for long polling —
against a 30s lease, and the turn was re-proven before waiting for the chat's
scheduler lock rather than after. A slow call could therefore still be talking
to Telegram after another process had taken the turn over and settled it, and
its result would then be fenced off: the reply is lost rather than retried.

Delivery calls now get their own budget, sized so one call plus recording its
outcome fits inside the lease, and every path re-proves the turn immediately
before the call rather than merely at the top of the step. A test asserts the
budget relationship so changing one constant cannot quietly reopen the gap.

Reaching the bad outcome needed a duplicate terminal event, which the task
completion path does not currently produce, so this was assessed as hardening
rather than a live defect. It is cheap enough to make hold by construction.

Co-authored-by: multica-agent <[email protected]>

---------

Co-authored-by: multica-agent <[email protected]>
2026-09-20 18:12:39 +08:00

490 lines
16 KiB
Go

// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: workspace.sql
package db
import (
"context"
"github.com/jackc/pgx/v5/pgtype"
)
const createWorkspace = `-- name: CreateWorkspace :one
INSERT INTO workspace (name, slug, description, context, issue_prefix)
VALUES ($1, $2, $3, $4, $5)
RETURNING id, name, slug, description, settings, created_at, updated_at, context, repos, issue_prefix, issue_counter, avatar_url, attribution_fail_closed
`
type CreateWorkspaceParams struct {
Name string `json:"name"`
Slug string `json:"slug"`
Description pgtype.Text `json:"description"`
Context pgtype.Text `json:"context"`
IssuePrefix string `json:"issue_prefix"`
}
func (q *Queries) CreateWorkspace(ctx context.Context, arg CreateWorkspaceParams) (Workspace, error) {
row := q.db.QueryRow(ctx, createWorkspace,
arg.Name,
arg.Slug,
arg.Description,
arg.Context,
arg.IssuePrefix,
)
var i Workspace
err := row.Scan(
&i.ID,
&i.Name,
&i.Slug,
&i.Description,
&i.Settings,
&i.CreatedAt,
&i.UpdatedAt,
&i.Context,
&i.Repos,
&i.IssuePrefix,
&i.IssueCounter,
&i.AvatarUrl,
&i.AttributionFailClosed,
)
return i, err
}
const deleteWorkspace = `-- name: DeleteWorkspace :exec
WITH ws_installations AS (
SELECT id FROM channel_installation WHERE workspace_id = $1
),
ws_sessions AS (
SELECT id FROM chat_session WHERE workspace_id = $1
),
ws_agents AS (
SELECT id FROM agent WHERE workspace_id = $1
),
ws_skills AS (
SELECT id FROM skill WHERE workspace_id = $1
),
cleared_agent_label_assignments AS (
DELETE FROM agent_to_label WHERE agent_id IN (SELECT id FROM ws_agents)
),
cleared_skill_label_assignments AS (
DELETE FROM skill_to_label WHERE skill_id IN (SELECT id FROM ws_skills)
),
cleared_channel_task_deliveries AS (
DELETE FROM channel_task_delivery WHERE installation_id IN (SELECT id FROM ws_installations)
),
cleared_channel_outbound_messages AS (
DELETE FROM channel_outbound_message WHERE installation_id IN (SELECT id FROM ws_installations)
),
cleared_channel_reply_deliveries AS (
DELETE FROM channel_reply_delivery WHERE installation_id IN (SELECT id FROM ws_installations)
),
cleared_chat_sessions AS (
DELETE FROM channel_chat_session_binding WHERE installation_id IN (SELECT id FROM ws_installations)
RETURNING chat_session_id
),
cleared_chat_contexts AS (
DELETE FROM channel_chat_context_generation
WHERE chat_session_id IN (SELECT chat_session_id FROM cleared_chat_sessions)
OR chat_session_id IN (SELECT id FROM ws_sessions)
),
cleared_outbound_cards AS (
-- channel_outbound_card_message is keyed by chat_session_id (no FK); its own
-- chat_session rows cascade away with the workspace, so reach the cards through
-- the just-removed chat-session bindings, which still carry the id.
DELETE FROM channel_outbound_card_message
WHERE chat_session_id IN (SELECT chat_session_id FROM cleared_chat_sessions)
),
cleared_draft_restores AS (
-- chat_draft_restore is keyed by chat_session_id with no FK (MUL-3515) and has
-- no reaper, while its chat_session rows cascade away with the workspace. Reach
-- them directly through chat_session (unlike the cards above, this is not
-- limited to channel-bound sessions) or every pending restore — each holding a
-- user's prompt text — would outlive the workspace permanently (#5219).
--
-- This sweep only sees restores committed before the statement's snapshot, so
-- the caller must already hold LockChatSessionsByWorkspace: that lock is what
-- keeps FinalizeDeferredCancelledChat from inserting one behind it.
DELETE FROM chat_draft_restore
WHERE chat_session_id IN (SELECT id FROM chat_session WHERE workspace_id = $1)
),
cleared_inbound_dedup AS (
DELETE FROM channel_inbound_message_dedup WHERE installation_id IN (SELECT id FROM ws_installations)
),
cleared_dingtalk_group_presence AS (
DELETE FROM dingtalk_group_presence WHERE installation_id IN (SELECT id FROM ws_installations)
),
cleared_dingtalk_bot_identity AS (
DELETE FROM dingtalk_bot_identity WHERE installation_id IN (SELECT id FROM ws_installations)
),
cleared_dingtalk_group_routes AS (
DELETE FROM dingtalk_group_route WHERE workspace_id = $1
),
cleared_audit AS (
-- Purge, don't detach: the workspace is gone and channel_inbound_audit has no
-- workspace_id and no reaper, so a detached (NULL) row would be permanently
-- unattributable. (Reclaim, where the workspace survives, still detaches.)
DELETE FROM channel_inbound_audit WHERE installation_id IN (SELECT id FROM ws_installations)
),
cleared_user_bindings AS (
DELETE FROM channel_user_binding WHERE workspace_id = $1
),
cleared_binding_tokens AS (
DELETE FROM channel_binding_token WHERE workspace_id = $1
),
cleared_installations AS (
DELETE FROM channel_installation WHERE workspace_id = $1
),
cleared_issue_properties AS (
DELETE FROM issue_property WHERE workspace_id = $1
),
cleared_quick_actions AS (
DELETE FROM quick_action WHERE workspace_id = $1
),
ws_mcp_servers AS (
SELECT id FROM workspace_mcp_server WHERE workspace_id = $1
),
cleared_agent_mcp_bindings AS (
-- agent_mcp_server carries no FK in either direction, so sweep it from
-- both sides: the workspace's own servers, and any binding held by an
-- agent that is about to be removed with the workspace.
DELETE FROM agent_mcp_server
WHERE server_id IN (SELECT id FROM ws_mcp_servers)
OR agent_id IN (SELECT id FROM ws_agents)
),
cleared_workspace_mcp_servers AS (
DELETE FROM workspace_mcp_server WHERE workspace_id = $1
),
deleted_pending_check_suites AS (
DELETE FROM github_pending_check_suite WHERE workspace_id = $1
),
ws_github_prs AS (
SELECT id FROM github_pull_request WHERE workspace_id = $1
),
cleared_github_pr_check_runs AS (
-- github_pull_request_check_run intentionally has no FK. Remove its rows
-- before the workspace delete cascades away the parent PR mirrors.
DELETE FROM github_pull_request_check_run
WHERE pr_id IN (SELECT id FROM ws_github_prs)
),
ws_vcs_prs AS (
SELECT id FROM vcs_pull_request WHERE workspace_id = $1
),
ws_vcs_connections AS (
SELECT id FROM vcs_connection WHERE workspace_id = $1
),
cleared_vcs_pr_links AS (
DELETE FROM issue_vcs_pull_request
WHERE pull_request_id IN (SELECT id FROM ws_vcs_prs)
),
cleared_vcs_commit_statuses AS (
DELETE FROM vcs_commit_status
WHERE connection_id IN (SELECT id FROM ws_vcs_connections)
),
cleared_vcs_prs AS (
DELETE FROM vcs_pull_request WHERE workspace_id = $1
),
cleared_vcs_connections AS (
DELETE FROM vcs_connection WHERE workspace_id = $1
),
cleared_client_usage_workspace AS (
UPDATE client_usage_daily SET workspace_id = NULL WHERE workspace_id = $1
)
DELETE FROM workspace WHERE workspace.id = $1
`
// The channel_* tables (MUL-3515 §4), resource-label junctions, custom issue
// property definitions, and quick actions carry NO FK to workspace, so — unlike the CASCADE-backed
// tables the DELETE below sweeps — they are not cleaned up implicitly. Remove
// their workspace-owned rows here so they commit or roll back atomically with
// the workspace row.
// VCS tables (migration 213) carry no FK to workspace, so they are not cascaded
// away by the DELETE below. Sweep the workspace's connections, mirrored PRs,
// their issue links, and CI statuses here. issue_vcs_pull_request has no
// workspace_id, so reach it through the workspace's PRs; vcs_commit_status has
// none either, so reach it through the workspace's connections.
func (q *Queries) DeleteWorkspace(ctx context.Context, id pgtype.UUID) error {
_, err := q.db.Exec(ctx, deleteWorkspace, id)
return err
}
const getDaemonWorkspace = `-- name: GetDaemonWorkspace :one
SELECT id, name
FROM workspace
WHERE id = $1
`
type GetDaemonWorkspaceRow struct {
ID pgtype.UUID `json:"id"`
Name string `json:"name"`
}
// Workspace-scoped daemon tokens do not carry a user ID. This narrow lookup
// lets them use the same endpoint without widening their token scope.
func (q *Queries) GetDaemonWorkspace(ctx context.Context, id pgtype.UUID) (GetDaemonWorkspaceRow, error) {
row := q.db.QueryRow(ctx, getDaemonWorkspace, id)
var i GetDaemonWorkspaceRow
err := row.Scan(&i.ID, &i.Name)
return i, err
}
const getWorkspace = `-- name: GetWorkspace :one
SELECT id, name, slug, description, settings, created_at, updated_at, context, repos, issue_prefix, issue_counter, avatar_url, attribution_fail_closed FROM workspace
WHERE id = $1
`
func (q *Queries) GetWorkspace(ctx context.Context, id pgtype.UUID) (Workspace, error) {
row := q.db.QueryRow(ctx, getWorkspace, id)
var i Workspace
err := row.Scan(
&i.ID,
&i.Name,
&i.Slug,
&i.Description,
&i.Settings,
&i.CreatedAt,
&i.UpdatedAt,
&i.Context,
&i.Repos,
&i.IssuePrefix,
&i.IssueCounter,
&i.AvatarUrl,
&i.AttributionFailClosed,
)
return i, err
}
const getWorkspaceAttributionFailClosed = `-- name: GetWorkspaceAttributionFailClosed :one
SELECT attribution_fail_closed FROM workspace
WHERE id = $1
`
// Lean read of the fail-closed attribution policy for the enqueue hot path
// (MUL-4302 §3.5), avoiding a full workspace-row fetch.
func (q *Queries) GetWorkspaceAttributionFailClosed(ctx context.Context, id pgtype.UUID) (bool, error) {
row := q.db.QueryRow(ctx, getWorkspaceAttributionFailClosed, id)
var attribution_fail_closed bool
err := row.Scan(&attribution_fail_closed)
return attribution_fail_closed, err
}
const getWorkspaceBySlug = `-- name: GetWorkspaceBySlug :one
SELECT id, name, slug, description, settings, created_at, updated_at, context, repos, issue_prefix, issue_counter, avatar_url, attribution_fail_closed FROM workspace
WHERE slug = $1
`
func (q *Queries) GetWorkspaceBySlug(ctx context.Context, slug string) (Workspace, error) {
row := q.db.QueryRow(ctx, getWorkspaceBySlug, slug)
var i Workspace
err := row.Scan(
&i.ID,
&i.Name,
&i.Slug,
&i.Description,
&i.Settings,
&i.CreatedAt,
&i.UpdatedAt,
&i.Context,
&i.Repos,
&i.IssuePrefix,
&i.IssueCounter,
&i.AvatarUrl,
&i.AttributionFailClosed,
)
return i, err
}
const incrementIssueCounter = `-- name: IncrementIssueCounter :one
UPDATE workspace SET issue_counter = issue_counter + 1
WHERE id = $1
RETURNING issue_counter
`
func (q *Queries) IncrementIssueCounter(ctx context.Context, id pgtype.UUID) (int32, error) {
row := q.db.QueryRow(ctx, incrementIssueCounter, id)
var issue_counter int32
err := row.Scan(&issue_counter)
return issue_counter, err
}
const listDaemonWorkspaces = `-- name: ListDaemonWorkspaces :many
SELECT w.id, w.name
FROM member m
JOIN workspace w ON w.id = m.workspace_id
WHERE m.user_id = $1
ORDER BY w.id ASC
`
type ListDaemonWorkspacesRow struct {
ID pgtype.UUID `json:"id"`
Name string `json:"name"`
}
// Daemons only need the membership set and display name to discover which
// workspaces should have local runtimes. Keep this projection intentionally
// narrow so the periodic consistency check never reads UI-only JSON/text
// columns such as settings, repos, or context.
func (q *Queries) ListDaemonWorkspaces(ctx context.Context, userID pgtype.UUID) ([]ListDaemonWorkspacesRow, error) {
rows, err := q.db.Query(ctx, listDaemonWorkspaces, userID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ListDaemonWorkspacesRow{}
for rows.Next() {
var i ListDaemonWorkspacesRow
if err := rows.Scan(&i.ID, &i.Name); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listWorkspaces = `-- name: ListWorkspaces :many
SELECT w.id, w.name, w.slug, w.description, w.settings,
w.created_at, w.updated_at, w.context, w.repos,
w.issue_prefix, w.issue_counter, w.avatar_url, w.attribution_fail_closed
FROM member m
JOIN workspace w ON w.id = m.workspace_id
WHERE m.user_id = $1
ORDER BY w.created_at ASC
`
func (q *Queries) ListWorkspaces(ctx context.Context, userID pgtype.UUID) ([]Workspace, error) {
rows, err := q.db.Query(ctx, listWorkspaces, userID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []Workspace{}
for rows.Next() {
var i Workspace
if err := rows.Scan(
&i.ID,
&i.Name,
&i.Slug,
&i.Description,
&i.Settings,
&i.CreatedAt,
&i.UpdatedAt,
&i.Context,
&i.Repos,
&i.IssuePrefix,
&i.IssueCounter,
&i.AvatarUrl,
&i.AttributionFailClosed,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const lockWorkspaceForChatSessionCreate = `-- name: LockWorkspaceForChatSessionCreate :one
SELECT id FROM workspace WHERE id = $1 FOR KEY SHARE
`
// The creator half of the workspace delete/create protocol (#5219). Every
// production path that inserts a chat_session takes this FOR KEY SHARE lock on the
// parent workspace row, inside its transaction, before CreateChatSession. It
// conflicts with DeleteWorkspace's FOR UPDATE (so a create is blocked while a
// delete is in progress, and vice versa) but not with other creators (FOR KEY
// SHARE locks share), so concurrent session creation stays unserialized. This
// makes the mutual exclusion explicit rather than leaning on the workspace FK's
// implicit FOR KEY SHARE, which would vanish if that FK is dropped.
func (q *Queries) LockWorkspaceForChatSessionCreate(ctx context.Context, id pgtype.UUID) (pgtype.UUID, error) {
row := q.db.QueryRow(ctx, lockWorkspaceForChatSessionCreate, id)
var id_2 pgtype.UUID
err := row.Scan(&id_2)
return id_2, err
}
const lockWorkspaceForDelete = `-- name: LockWorkspaceForDelete :one
SELECT id FROM workspace WHERE id = $1 FOR UPDATE
`
// Taken first by DeleteWorkspace, before it enumerates the workspace's chat
// sessions. LockChatSessionsByWorkspace only covers sessions that exist when it
// runs; a CreateChatSession committing during the delete window would add one
// the lock set never saw, and a finalizer could then insert a restore for it
// after the sweep's snapshot — orphaning the prompt (#5219).
//
// The delete window is held closed against new sessions by an EXPLICIT protocol,
// not the chat_session.workspace_id FK: every session creator takes
// LockWorkspaceForChatSessionCreate (FOR KEY SHARE) on this row first, and this
// FOR UPDATE conflicts with it. Keeping the bar in the app layer means it does
// not silently break if that FK is ever dropped (the codebase is moving FK
// relationships into the application layer, MUL-3515). Lock order is
// workspace -> chat_session -> agent_task_queue; the finalizer never touches
// workspace, so this cannot deadlock against it.
func (q *Queries) LockWorkspaceForDelete(ctx context.Context, id pgtype.UUID) (pgtype.UUID, error) {
row := q.db.QueryRow(ctx, lockWorkspaceForDelete, id)
var id_2 pgtype.UUID
err := row.Scan(&id_2)
return id_2, err
}
const updateWorkspace = `-- name: UpdateWorkspace :one
UPDATE workspace SET
name = COALESCE($2, name),
description = COALESCE($3, description),
context = COALESCE($4, context),
settings = COALESCE($5, settings),
repos = COALESCE($6, repos),
issue_prefix = COALESCE($7, issue_prefix),
avatar_url = COALESCE($8, avatar_url),
updated_at = now()
WHERE id = $1
RETURNING id, name, slug, description, settings, created_at, updated_at, context, repos, issue_prefix, issue_counter, avatar_url, attribution_fail_closed
`
type UpdateWorkspaceParams struct {
ID pgtype.UUID `json:"id"`
Name pgtype.Text `json:"name"`
Description pgtype.Text `json:"description"`
Context pgtype.Text `json:"context"`
Settings []byte `json:"settings"`
Repos []byte `json:"repos"`
IssuePrefix pgtype.Text `json:"issue_prefix"`
AvatarUrl pgtype.Text `json:"avatar_url"`
}
func (q *Queries) UpdateWorkspace(ctx context.Context, arg UpdateWorkspaceParams) (Workspace, error) {
row := q.db.QueryRow(ctx, updateWorkspace,
arg.ID,
arg.Name,
arg.Description,
arg.Context,
arg.Settings,
arg.Repos,
arg.IssuePrefix,
arg.AvatarUrl,
)
var i Workspace
err := row.Scan(
&i.ID,
&i.Name,
&i.Slug,
&i.Description,
&i.Settings,
&i.CreatedAt,
&i.UpdatedAt,
&i.Context,
&i.Repos,
&i.IssuePrefix,
&i.IssueCounter,
&i.AvatarUrl,
&i.AttributionFailClosed,
)
return i, err
}