Files
multica/server/pkg/db/generated/supplement.sql.go
T
e2f4a22030 feat: steer a running agent from the normal composer, per recipient (MUL-7631) (#8760)
* feat(server): steer running turns from ordinary comments (MUL-7631)

A comment can now name recipients whose running turn should receive it
(`steer_agent_ids` on POST /issues/:id/comments). Each named recipient
with a running, steer-capable turn gets the comment bound to that turn
instead of a follow-up run; any other recipient keeps its normal
queued / coalesced / deferred trigger, so losing a race never drops input.

- One comment can steer several turns: receipts are keyed by
  (comment_id, task_id) and returned as `supplements` on comments and
  timeline entries. The single supplement_* fields mirror the first
  receipt for older clients.
- Completion reconciliation skips a steered comment only for the agent
  it steered, so the same comment still reaches recipients that were
  addressed without steering.
- Delivery claims match the (comment, run) pair.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* feat(issues): choose per recipient how a message reaches a running agent (MUL-7631)

Steering moves from a separate "Add message" box on the run row into the
normal reply and comment composers. Recipients keep the existing trigger
rules; each recipient's current run on the issue decides what the message
can do:

- running (and steer-capable): add to the current run (default in that
  run's own thread), start after this run, stop and start over, or skip
- queued: include when it starts, or skip
- idle: start on send, or skip

Receipts follow the message ("Waiting for Lambda to read it" → "Read by
Lambda · after step N"), a failed delivery offers retry or sending it as
a new run, and the run's step list marks where it read the message.
A draft whose recipient finished meanwhile stays in place and says it
will start a new run; a message with files is handled after the run.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* fix(server): steer only the chosen turn, and make a retried steer idempotent (MUL-7631)

- Steering names the exact running turns the author chose
  (`steer_task_ids`). A chosen turn that ended before the send arrives is
  never swapped for a later turn of the same agent; that recipient keeps
  its normal trigger.
- A steering send carries `client_request_id`. A retry after a lost
  response returns the original comment (200) instead of creating a second
  comment and a second delivery; a concurrent twin binding the same request
  is treated as already steered.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* fix(issues): guard steering sends against stale previews and broad resends (MUL-7631)

- "Stop and start over" never stops a run while the trigger preview is
  still catching up with an edited @mention: the stop is dropped unless the
  preview answers the submitted text.
- Steering sends the chosen turns' task ids and one request id per draft
  text, reused on retries of the same text.
- "Send as a new run" on a failed receipt reaches only that receipt's
  agent: every other current recipient is suppressed.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* fix(issues): keep the reply box placeholder neutral (MUL-7631)

The empty reply box cannot know who a reply goes to: the author may
@ someone else or only talk to a teammate. The recipient chip already
says what Send will do once there is text.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* test(issues): give steering composer tests room for the preview debounce (MUL-7631)

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* feat(issues): choose the default reply to a running agent (MUL-7631)

Preferences > Comments gains "When replying to a running agent":
"Add to current run" (default) or "Start after this run". It only
changes what a reply does without a per-message choice; the recipient
chip still offers both, and replies that never steered by default
(another thread, an idle or queued agent) are unaffected. Saved on this
device next to the sticky comment bar.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* fix(server): save one comment per send and honor "don't start" on replay (MUL-7631)

Idempotency now lives on the comment, not the delivery receipt. A send's
client_request_id is stored on the comment behind a partial unique index
(migrations 550/551), so a retry, or a concurrent twin, returns the saved
comment (200) whether the send steered a running turn or fell back to a
normal trigger. Before, a twin could save a second comment, and a send whose
turn had ended kept no receipt to find.

The agents an author chose not to start are stored on the comment too, and
completion replay skips them. "Send as a new run" for one missed recipient,
and "Won't start this time", no longer wake the other agent once its current
run completes. An edit re-records the choice.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* fix(issues): repeat the original steering request on retry (MUL-7631)

Until a send is settled, retrying the same text sends its first steering
request unchanged, even after the chosen turn ended. The server then returns
the comment the first attempt saved instead of posting it again as new input.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* fix(migrations): fail fast on locks in 549 and 550 (MUL-7631)

Both take an ACCESS EXCLUSIVE lock on a table the product touches
continuously: comment for every read and write, task_supplement for every
daemon claim. Bound lock acquisition and execution, as 483 does for issue,
so a long transaction makes the migration fail and retry on the next run
instead of queueing all traffic behind it.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* fix(server): steer at most one turn per comment during rollout (MUL-7631)

Steering is live, and servers from before per-run receipts claim every
receipt of a comment at once. While they still serve daemons during a
rollout, a comment bound to two turns could have the second copy marked in
flight and never delivered. Until every server claims per (comment, run),
a comment steers the first chosen turn only; later recipients keep their
normal trigger. A follow-up lifts the limit after this ships.

Also covers a steer_task_ids entry that names another issue's turn.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* fix(server): finish a saved send on retry; keep an edit's choices with its text (MUL-7631)

A send's comment was saved before its agents were started or steered, and
deduplication treated the saved comment as the whole send. A dropped
connection cancelled the request between the two, and the retry only
returned the comment, so no agent ever received it.

- Attempts of one logical send now run one at a time under a transaction
  advisory lock, held on its own connection: a twin waits for the attempt in
  progress (up to 30s, then 409) instead of racing it, and a process that
  dies releases it.
- After the comment is saved, dispatch runs on a non-cancelable context, and
  comment.client_request_dispatched_at (migration 552) records that it
  finished. A retry that finds a saved but undispatched comment finishes the
  dispatch with the comment's recorded choices.

An edit's "don't start" choices were written by a separate statement after
the text, so interleaved edits could leave an older edit's choice. They are
now part of UpdateComment, under the same revision check as the text.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* fix(issues): offer one steered recipient per message during rollout (MUL-7631)

The server steers at most one turn per comment until #8800, but the
composer showed "Add to current run" for every running recipient and the
server quietly queued the rest. Now only one recipient steers: the one the
author picked, otherwise the first that would by default. The others show
"Start after this run", and picking "Add to current run" on one of them
moves the steer there.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* fix(server): claim a send before dispatching it; drop the request lock (MUL-7631)

The previous recovery marked a send dispatched only after reaching its
agents, so a failure after dispatch made a retry reach them all again. It
also held a pooled connection per send for an advisory lock while the send
needed more from the same pool, which could wait on itself until timeout.

A send's comment now holds a dispatch claim instead
(comment.client_request_dispatched_at, one conditional update): only the
attempt whose claim takes effect publishes the comment and reaches its
agents. A failed claim writes nothing else and returns 500 for the client
to retry; a retry that finds an unclaimed comment claims and dispatches
it; a claimed one is only returned. Concurrent attempts no longer wait on
each other, and post-save dispatch keeps a 30s bound of its own.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* feat(server): let one comment steer several running turns (MUL-7631)

Lifts the rollout limit from #8760. Every server now claims receipts per
(comment, run), so a comment can be bound to each running turn its author
chose, and each recipient receives it in its own run.

Merge only after #8760 is deployed everywhere.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* feat(issues): steer every running recipient a message addresses (MUL-7631)

The client side of lifting the rollout limit: each running recipient can
take the message in its current turn again, instead of only one per
message.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* refactor(issues): drop send-retry idempotency and two display extras (MUL-7631)

A steering send now behaves like any other comment when its response is
lost: a retry posts it again. The machinery that tried to make that retry
return the saved comment grew with every edge it met, and ordinary comments
never had it:

- Remove comment.client_request_id, its unique index, and the dispatch
  claim (migrations 551/552, and 550 reduced to suppressed_agent_ids).
- Remove the replay/resume path in CreateComment, the post-save
  non-cancelable dispatch context, and the client's sticky request id.
  Steering sends only `steer_task_ids`; each binding uses the comment id
  as its receipt request id.

Two display extras go too, since the UI already shows the same thing:

- The "run ended while you were writing" notice: the recipient chip
  already switches from "Add to current run" to its new action.
- "after step N" on a read receipt: the run trace already places the
  message at the step where it was read.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* fix(issues): space agent names in Chinese copy; neutral recipients header (MUL-7631)

- The steer badge joined names with Intl.ListFormat, which gives
  "Lambda和Orion" in Chinese. Latin names are now spaced from the Chinese
  joiner ("Lambda 和 Orion"), per the zh copy conventions; the enumeration
  comma stays unspaced.
- The multi-recipient popover was still titled "Will start when sent" and
  its tooltip said "choose who runs", though each row now picks how that
  agent handles the message. Title it "Recipients" and reword the tooltip.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

* fix(issues): translate the recipient source labels in Chinese (MUL-7631)

"assignee" and "@mention" were left in English in zh-Hans; the recipient
menu header shows them. Use 负责人 (per the zh conventions) and @提及.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>

---------

Co-authored-by: Claude Opus 5.5 <[email protected]>
Co-authored-by: multica-agent <[email protected]>
2026-09-25 04:42:26 +08:00

831 lines
28 KiB
Go

// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: supplement.sql
package db
import (
"context"
"github.com/jackc/pgx/v5/pgtype"
)
const ackTaskSupplementDelivered = `-- name: AckTaskSupplementDelivered :one
WITH task AS MATERIALIZED (
SELECT id FROM agent_task_queue WHERE id = $2 FOR UPDATE
)
UPDATE task_supplement s
SET status = 'delivered',
delivered_at = COALESCE(delivered_at, now()),
failure_reason = NULL,
updated_at = now()
FROM task t
WHERE s.task_id = t.id
AND s.comment_id = $1
AND (s.status IN ('delivering', 'delivered')
OR (s.status = 'failed' AND s.failure_reason = 'turn_ended' AND s.attempt_count > 0))
RETURNING s.task_id, s.workspace_id, s.issue_id, s.comment_id, s.author_id, s.client_request_id, s.status, s.failure_reason, s.attempt_count, s.created_at, s.updated_at, s.delivered_at
`
type AckTaskSupplementDeliveredParams struct {
CommentID pgtype.UUID `json:"comment_id"`
TaskID pgtype.UUID `json:"task_id"`
}
// The provider can accept input before completion while its HTTP acknowledgement
// arrives afterward. Serialize with settlement and preserve that successful write.
func (q *Queries) AckTaskSupplementDelivered(ctx context.Context, arg AckTaskSupplementDeliveredParams) (TaskSupplement, error) {
row := q.db.QueryRow(ctx, ackTaskSupplementDelivered, arg.CommentID, arg.TaskID)
var i TaskSupplement
err := row.Scan(
&i.TaskID,
&i.WorkspaceID,
&i.IssueID,
&i.CommentID,
&i.AuthorID,
&i.ClientRequestID,
&i.Status,
&i.FailureReason,
&i.AttemptCount,
&i.CreatedAt,
&i.UpdatedAt,
&i.DeliveredAt,
)
return i, err
}
const ackTaskSupplementFailed = `-- name: AckTaskSupplementFailed :one
UPDATE task_supplement
SET status = 'failed',
failure_reason = $1,
updated_at = now()
WHERE task_id = $2
AND comment_id = $3
AND status = 'delivering'
RETURNING task_id, workspace_id, issue_id, comment_id, author_id, client_request_id, status, failure_reason, attempt_count, created_at, updated_at, delivered_at
`
type AckTaskSupplementFailedParams struct {
FailureReason pgtype.Text `json:"failure_reason"`
TaskID pgtype.UUID `json:"task_id"`
CommentID pgtype.UUID `json:"comment_id"`
}
func (q *Queries) AckTaskSupplementFailed(ctx context.Context, arg AckTaskSupplementFailedParams) (TaskSupplement, error) {
row := q.db.QueryRow(ctx, ackTaskSupplementFailed, arg.FailureReason, arg.TaskID, arg.CommentID)
var i TaskSupplement
err := row.Scan(
&i.TaskID,
&i.WorkspaceID,
&i.IssueID,
&i.CommentID,
&i.AuthorID,
&i.ClientRequestID,
&i.Status,
&i.FailureReason,
&i.AttemptCount,
&i.CreatedAt,
&i.UpdatedAt,
&i.DeliveredAt,
)
return i, err
}
const bindCommentTaskSupplement = `-- name: BindCommentTaskSupplement :one
WITH locked_task AS MATERIALIZED (
SELECT t.id, t.issue_id, t.runtime_id
FROM agent_task_queue t
JOIN agent_runtime r ON r.id = t.runtime_id
JOIN task_supplement_capability cap ON cap.task_id = t.id
WHERE t.id = $1
AND t.issue_id = $2
AND t.agent_id = $3
AND r.workspace_id = $4
AND t.status = 'running'
AND cap.capability = 'task-supplement-v1'
AND r.provider IN ('codex', 'claude')
FOR UPDATE OF t
), inserted AS (
INSERT INTO task_supplement (
task_id, workspace_id, issue_id, comment_id, author_id,
client_request_id, status
)
SELECT t.id, $4, t.issue_id, $5, $6,
$7, 'pending'
FROM locked_task t
RETURNING task_id, workspace_id, issue_id, comment_id, author_id, client_request_id, status, failure_reason, attempt_count, created_at, updated_at, delivered_at
)
SELECT inserted.task_id, inserted.workspace_id, inserted.issue_id, inserted.comment_id, inserted.author_id, inserted.client_request_id, inserted.status, inserted.failure_reason, inserted.attempt_count, inserted.created_at, inserted.updated_at, inserted.delivered_at, locked_task.runtime_id
FROM inserted
JOIN locked_task ON locked_task.id = inserted.task_id
`
type BindCommentTaskSupplementParams struct {
TaskID pgtype.UUID `json:"task_id"`
IssueID pgtype.UUID `json:"issue_id"`
AgentID pgtype.UUID `json:"agent_id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
CommentID pgtype.UUID `json:"comment_id"`
AuthorID pgtype.UUID `json:"author_id"`
ClientRequestID pgtype.UUID `json:"client_request_id"`
}
type BindCommentTaskSupplementRow struct {
TaskID pgtype.UUID `json:"task_id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
IssueID pgtype.UUID `json:"issue_id"`
CommentID pgtype.UUID `json:"comment_id"`
AuthorID pgtype.UUID `json:"author_id"`
ClientRequestID pgtype.UUID `json:"client_request_id"`
Status string `json:"status"`
FailureReason pgtype.Text `json:"failure_reason"`
AttemptCount int32 `json:"attempt_count"`
CreatedAt pgtype.Timestamptz `json:"created_at"`
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
DeliveredAt pgtype.Timestamptz `json:"delivered_at"`
RuntimeID pgtype.UUID `json:"runtime_id"`
}
// Binds an ordinary member comment, already created through the normal comment
// path, to the exact running turn its author chose. Locking the task
// serializes against terminal transitions: when that turn ended first nothing
// is bound — never a later turn of the same agent — and the caller keeps the
// comment's normal trigger instead.
func (q *Queries) BindCommentTaskSupplement(ctx context.Context, arg BindCommentTaskSupplementParams) (BindCommentTaskSupplementRow, error) {
row := q.db.QueryRow(ctx, bindCommentTaskSupplement,
arg.TaskID,
arg.IssueID,
arg.AgentID,
arg.WorkspaceID,
arg.CommentID,
arg.AuthorID,
arg.ClientRequestID,
)
var i BindCommentTaskSupplementRow
err := row.Scan(
&i.TaskID,
&i.WorkspaceID,
&i.IssueID,
&i.CommentID,
&i.AuthorID,
&i.ClientRequestID,
&i.Status,
&i.FailureReason,
&i.AttemptCount,
&i.CreatedAt,
&i.UpdatedAt,
&i.DeliveredAt,
&i.RuntimeID,
)
return i, err
}
const claimNextTaskSupplement = `-- name: ClaimNextTaskSupplement :one
WITH next AS MATERIALIZED (
SELECT s.comment_id, s.task_id
FROM task_supplement s
JOIN agent_task_queue t ON t.id = s.task_id
JOIN task_supplement_capability cap ON cap.task_id = t.id
WHERE s.task_id = $1
AND s.status = 'pending'
AND t.status = 'running'
AND cap.capability = 'task-supplement-v1'
ORDER BY s.created_at, s.comment_id
FOR UPDATE OF s SKIP LOCKED
LIMIT 1
), claimed AS (
UPDATE task_supplement s
SET status = 'delivering',
attempt_count = attempt_count + 1,
failure_reason = NULL,
updated_at = now()
FROM next
WHERE s.comment_id = next.comment_id
AND s.task_id = next.task_id
RETURNING s.task_id, s.workspace_id, s.issue_id, s.comment_id, s.author_id, s.client_request_id, s.status, s.failure_reason, s.attempt_count, s.created_at, s.updated_at, s.delivered_at
)
SELECT claimed.comment_id, claimed.attempt_count, c.content,
COALESCE(NULLIF(btrim(u.name), ''), 'a user')::text AS author_name
FROM claimed
JOIN comment c ON c.id = claimed.comment_id
LEFT JOIN "user" u ON u.id = claimed.author_id
`
type ClaimNextTaskSupplementRow struct {
CommentID pgtype.UUID `json:"comment_id"`
AttemptCount int32 `json:"attempt_count"`
Content string `json:"content"`
AuthorName string `json:"author_name"`
}
func (q *Queries) ClaimNextTaskSupplement(ctx context.Context, taskID pgtype.UUID) (ClaimNextTaskSupplementRow, error) {
row := q.db.QueryRow(ctx, claimNextTaskSupplement, taskID)
var i ClaimNextTaskSupplementRow
err := row.Scan(
&i.CommentID,
&i.AttemptCount,
&i.Content,
&i.AuthorName,
)
return i, err
}
const commentHasTaskSupplement = `-- name: CommentHasTaskSupplement :one
SELECT EXISTS (
SELECT 1 FROM task_supplement
WHERE comment_id = $1 AND workspace_id = $2
) AS bound
`
type CommentHasTaskSupplementParams struct {
CommentID pgtype.UUID `json:"comment_id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
}
func (q *Queries) CommentHasTaskSupplement(ctx context.Context, arg CommentHasTaskSupplementParams) (bool, error) {
row := q.db.QueryRow(ctx, commentHasTaskSupplement, arg.CommentID, arg.WorkspaceID)
var bound bool
err := row.Scan(&bound)
return bound, err
}
const createTaskSupplement = `-- name: CreateTaskSupplement :one
WITH locked_task AS MATERIALIZED (
SELECT t.id, t.issue_id, t.agent_id, t.trigger_comment_id
FROM agent_task_queue t
JOIN agent_runtime r ON r.id = t.runtime_id
JOIN task_supplement_capability cap ON cap.task_id = t.id
WHERE t.id = $1
AND t.issue_id = $2
AND r.workspace_id = $3
AND t.status = 'running'
AND cap.capability = 'task-supplement-v1'
AND r.provider IN ('codex', 'claude', 'grok')
FOR UPDATE OF t
), touched_issue AS (
UPDATE issue i SET
updated_at = now(),
revision = revision + 1,
last_activity_at = GREATEST(COALESCE(last_activity_at, updated_at), now())
FROM locked_task t
WHERE i.id = t.issue_id AND i.workspace_id = $3
RETURNING i.id, i.workspace_id, i.revision
), inserted_comment AS (
INSERT INTO comment (
issue_id, workspace_id, author_type, author_id, content, type, parent_id
)
SELECT i.id, i.workspace_id, 'member', $4, $5, 'comment', t.trigger_comment_id
FROM touched_issue i
JOIN locked_task t ON t.issue_id = i.id
RETURNING id, issue_id, author_type, author_id, content, type, created_at, updated_at, parent_id, workspace_id, resolved_at, resolved_by_type, resolved_by_id, source_task_id, quick_action_id, via_plugin_id, revision, recovery_settled_at, deleted_at, suppressed_agent_ids
), inserted_supplement AS (
INSERT INTO task_supplement (
task_id, workspace_id, issue_id, comment_id, author_id,
client_request_id, status
)
SELECT t.id, i.workspace_id, i.id, c.id, $4,
$6,
'pending'
FROM locked_task t
JOIN touched_issue i ON i.id = t.issue_id
JOIN inserted_comment c ON c.issue_id = i.id
RETURNING task_id, workspace_id, issue_id, comment_id, author_id, client_request_id, status, failure_reason, attempt_count, created_at, updated_at, delivered_at
)
SELECT c.id, c.issue_id, c.author_type, c.author_id, c.content, c.type, c.created_at, c.updated_at, c.parent_id, c.workspace_id, c.resolved_at, c.resolved_by_type, c.resolved_by_id, c.source_task_id, c.quick_action_id, c.via_plugin_id, c.revision, c.recovery_settled_at, c.deleted_at, c.suppressed_agent_ids, i.revision AS issue_revision,
s.task_id AS supplement_task_id, s.status AS supplement_status,
s.failure_reason AS supplement_failure_reason,
s.delivered_at AS supplement_delivered_at
FROM inserted_comment c
JOIN touched_issue i ON i.id = c.issue_id
JOIN inserted_supplement s ON s.comment_id = c.id
`
type CreateTaskSupplementParams struct {
TaskID pgtype.UUID `json:"task_id"`
IssueID pgtype.UUID `json:"issue_id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
AuthorID pgtype.UUID `json:"author_id"`
Content string `json:"content"`
ClientRequestID pgtype.UUID `json:"client_request_id"`
}
type CreateTaskSupplementRow struct {
ID pgtype.UUID `json:"id"`
IssueID pgtype.UUID `json:"issue_id"`
AuthorType string `json:"author_type"`
AuthorID pgtype.UUID `json:"author_id"`
Content string `json:"content"`
Type string `json:"type"`
CreatedAt pgtype.Timestamptz `json:"created_at"`
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
ParentID pgtype.UUID `json:"parent_id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
ResolvedAt pgtype.Timestamptz `json:"resolved_at"`
ResolvedByType pgtype.Text `json:"resolved_by_type"`
ResolvedByID pgtype.UUID `json:"resolved_by_id"`
SourceTaskID pgtype.UUID `json:"source_task_id"`
QuickActionID pgtype.UUID `json:"quick_action_id"`
ViaPluginID pgtype.UUID `json:"via_plugin_id"`
Revision int64 `json:"revision"`
RecoverySettledAt pgtype.Timestamptz `json:"recovery_settled_at"`
DeletedAt pgtype.Timestamptz `json:"deleted_at"`
SuppressedAgentIds []pgtype.UUID `json:"suppressed_agent_ids"`
IssueRevision int64 `json:"issue_revision"`
SupplementTaskID pgtype.UUID `json:"supplement_task_id"`
SupplementStatus string `json:"supplement_status"`
SupplementFailureReason pgtype.Text `json:"supplement_failure_reason"`
SupplementDeliveredAt pgtype.Timestamptz `json:"supplement_delivered_at"`
}
// Locking the exact task serializes against terminal transitions.
// Comment creation, explicit task binding and run coverage then
// commit as one statement: a terminal-race loser creates nothing.
func (q *Queries) CreateTaskSupplement(ctx context.Context, arg CreateTaskSupplementParams) (CreateTaskSupplementRow, error) {
row := q.db.QueryRow(ctx, createTaskSupplement,
arg.TaskID,
arg.IssueID,
arg.WorkspaceID,
arg.AuthorID,
arg.Content,
arg.ClientRequestID,
)
var i CreateTaskSupplementRow
err := row.Scan(
&i.ID,
&i.IssueID,
&i.AuthorType,
&i.AuthorID,
&i.Content,
&i.Type,
&i.CreatedAt,
&i.UpdatedAt,
&i.ParentID,
&i.WorkspaceID,
&i.ResolvedAt,
&i.ResolvedByType,
&i.ResolvedByID,
&i.SourceTaskID,
&i.QuickActionID,
&i.ViaPluginID,
&i.Revision,
&i.RecoverySettledAt,
&i.DeletedAt,
&i.SuppressedAgentIds,
&i.IssueRevision,
&i.SupplementTaskID,
&i.SupplementStatus,
&i.SupplementFailureReason,
&i.SupplementDeliveredAt,
)
return i, err
}
const deleteTaskSupplementByComment = `-- name: DeleteTaskSupplementByComment :exec
DELETE FROM task_supplement
WHERE comment_id = $1 AND workspace_id = $2
`
type DeleteTaskSupplementByCommentParams struct {
CommentID pgtype.UUID `json:"comment_id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
}
func (q *Queries) DeleteTaskSupplementByComment(ctx context.Context, arg DeleteTaskSupplementByCommentParams) error {
_, err := q.db.Exec(ctx, deleteTaskSupplementByComment, arg.CommentID, arg.WorkspaceID)
return err
}
const getTaskSupplementByRequest = `-- name: GetTaskSupplementByRequest :one
SELECT s.task_id, s.workspace_id, s.issue_id, s.comment_id, s.author_id, s.client_request_id, s.status, s.failure_reason, s.attempt_count, s.created_at, s.updated_at, s.delivered_at
FROM task_supplement s
WHERE s.task_id = $1
AND s.workspace_id = $2
AND s.author_id = $3
AND s.client_request_id = $4
`
type GetTaskSupplementByRequestParams struct {
TaskID pgtype.UUID `json:"task_id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
AuthorID pgtype.UUID `json:"author_id"`
ClientRequestID pgtype.UUID `json:"client_request_id"`
}
func (q *Queries) GetTaskSupplementByRequest(ctx context.Context, arg GetTaskSupplementByRequestParams) (TaskSupplement, error) {
row := q.db.QueryRow(ctx, getTaskSupplementByRequest,
arg.TaskID,
arg.WorkspaceID,
arg.AuthorID,
arg.ClientRequestID,
)
var i TaskSupplement
err := row.Scan(
&i.TaskID,
&i.WorkspaceID,
&i.IssueID,
&i.CommentID,
&i.AuthorID,
&i.ClientRequestID,
&i.Status,
&i.FailureReason,
&i.AttemptCount,
&i.CreatedAt,
&i.UpdatedAt,
&i.DeliveredAt,
)
return i, err
}
const getTaskSupplementCapability = `-- name: GetTaskSupplementCapability :one
SELECT task_id, workspace_id, issue_id, capability, created_at FROM task_supplement_capability WHERE task_id = $1
`
func (q *Queries) GetTaskSupplementCapability(ctx context.Context, taskID pgtype.UUID) (TaskSupplementCapability, error) {
row := q.db.QueryRow(ctx, getTaskSupplementCapability, taskID)
var i TaskSupplementCapability
err := row.Scan(
&i.TaskID,
&i.WorkspaceID,
&i.IssueID,
&i.Capability,
&i.CreatedAt,
)
return i, err
}
const getTaskSupplementForRun = `-- name: GetTaskSupplementForRun :one
SELECT task_id, workspace_id, issue_id, comment_id, author_id, client_request_id, status, failure_reason, attempt_count, created_at, updated_at, delivered_at FROM task_supplement
WHERE comment_id = $1 AND task_id = $2 AND workspace_id = $3
`
type GetTaskSupplementForRunParams struct {
CommentID pgtype.UUID `json:"comment_id"`
TaskID pgtype.UUID `json:"task_id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
}
func (q *Queries) GetTaskSupplementForRun(ctx context.Context, arg GetTaskSupplementForRunParams) (TaskSupplement, error) {
row := q.db.QueryRow(ctx, getTaskSupplementForRun, arg.CommentID, arg.TaskID, arg.WorkspaceID)
var i TaskSupplement
err := row.Scan(
&i.TaskID,
&i.WorkspaceID,
&i.IssueID,
&i.CommentID,
&i.AuthorID,
&i.ClientRequestID,
&i.Status,
&i.FailureReason,
&i.AttemptCount,
&i.CreatedAt,
&i.UpdatedAt,
&i.DeliveredAt,
)
return i, err
}
const listTaskSupplementMetadata = `-- name: ListTaskSupplementMetadata :many
SELECT cap.task_id, cap.capability,
COALESCE(array_agg(s.comment_id ORDER BY s.created_at, s.comment_id)
FILTER (WHERE s.comment_id IS NOT NULL), '{}'::uuid[])::uuid[] AS comment_ids
FROM task_supplement_capability cap
LEFT JOIN task_supplement s ON s.task_id = cap.task_id
WHERE cap.workspace_id = $1
AND cap.task_id = ANY($2::uuid[])
GROUP BY cap.task_id, cap.capability
`
type ListTaskSupplementMetadataParams struct {
WorkspaceID pgtype.UUID `json:"workspace_id"`
TaskIds []pgtype.UUID `json:"task_ids"`
}
type ListTaskSupplementMetadataRow struct {
TaskID pgtype.UUID `json:"task_id"`
Capability string `json:"capability"`
CommentIds []pgtype.UUID `json:"comment_ids"`
}
func (q *Queries) ListTaskSupplementMetadata(ctx context.Context, arg ListTaskSupplementMetadataParams) ([]ListTaskSupplementMetadataRow, error) {
rows, err := q.db.Query(ctx, listTaskSupplementMetadata, arg.WorkspaceID, arg.TaskIds)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ListTaskSupplementMetadataRow{}
for rows.Next() {
var i ListTaskSupplementMetadataRow
if err := rows.Scan(&i.TaskID, &i.Capability, &i.CommentIds); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listTaskSupplementsByCommentIDs = `-- name: ListTaskSupplementsByCommentIDs :many
SELECT s.task_id, s.workspace_id, s.issue_id, s.comment_id, s.author_id, s.client_request_id, s.status, s.failure_reason, s.attempt_count, s.created_at, s.updated_at, s.delivered_at, t.agent_id
FROM task_supplement s
LEFT JOIN agent_task_queue t ON t.id = s.task_id
WHERE s.workspace_id = $1
AND s.comment_id = ANY($2::uuid[])
ORDER BY s.created_at, s.task_id
`
type ListTaskSupplementsByCommentIDsParams struct {
WorkspaceID pgtype.UUID `json:"workspace_id"`
CommentIds []pgtype.UUID `json:"comment_ids"`
}
type ListTaskSupplementsByCommentIDsRow struct {
TaskID pgtype.UUID `json:"task_id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
IssueID pgtype.UUID `json:"issue_id"`
CommentID pgtype.UUID `json:"comment_id"`
AuthorID pgtype.UUID `json:"author_id"`
ClientRequestID pgtype.UUID `json:"client_request_id"`
Status string `json:"status"`
FailureReason pgtype.Text `json:"failure_reason"`
AttemptCount int32 `json:"attempt_count"`
CreatedAt pgtype.Timestamptz `json:"created_at"`
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
DeliveredAt pgtype.Timestamptz `json:"delivered_at"`
AgentID pgtype.UUID `json:"agent_id"`
}
// One comment may steer several runs; receipts carry the run's agent so a
// client can say whose turn received it.
func (q *Queries) ListTaskSupplementsByCommentIDs(ctx context.Context, arg ListTaskSupplementsByCommentIDsParams) ([]ListTaskSupplementsByCommentIDsRow, error) {
rows, err := q.db.Query(ctx, listTaskSupplementsByCommentIDs, arg.WorkspaceID, arg.CommentIds)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ListTaskSupplementsByCommentIDsRow{}
for rows.Next() {
var i ListTaskSupplementsByCommentIDsRow
if err := rows.Scan(
&i.TaskID,
&i.WorkspaceID,
&i.IssueID,
&i.CommentID,
&i.AuthorID,
&i.ClientRequestID,
&i.Status,
&i.FailureReason,
&i.AttemptCount,
&i.CreatedAt,
&i.UpdatedAt,
&i.DeliveredAt,
&i.AgentID,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const lockTaskSupplementsByComment = `-- name: LockTaskSupplementsByComment :many
SELECT task_id, workspace_id, issue_id, comment_id, author_id, client_request_id, status, failure_reason, attempt_count, created_at, updated_at, delivered_at FROM task_supplement
WHERE comment_id = $1 AND workspace_id = $2
FOR UPDATE
`
type LockTaskSupplementsByCommentParams struct {
CommentID pgtype.UUID `json:"comment_id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
}
func (q *Queries) LockTaskSupplementsByComment(ctx context.Context, arg LockTaskSupplementsByCommentParams) ([]TaskSupplement, error) {
rows, err := q.db.Query(ctx, lockTaskSupplementsByComment, arg.CommentID, arg.WorkspaceID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []TaskSupplement{}
for rows.Next() {
var i TaskSupplement
if err := rows.Scan(
&i.TaskID,
&i.WorkspaceID,
&i.IssueID,
&i.CommentID,
&i.AuthorID,
&i.ClientRequestID,
&i.Status,
&i.FailureReason,
&i.AttemptCount,
&i.CreatedAt,
&i.UpdatedAt,
&i.DeliveredAt,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const retryTaskSupplement = `-- name: RetryTaskSupplement :one
WITH comment AS MATERIALIZED (
SELECT id FROM comment
WHERE id = $1 AND workspace_id = $2 AND deleted_at IS NULL
FOR UPDATE
), active_task AS MATERIALIZED (
SELECT agent_task_queue.id FROM agent_task_queue
WHERE agent_task_queue.id = $3
AND agent_task_queue.issue_id = $4
AND agent_task_queue.status = 'running'
AND EXISTS (SELECT 1 FROM comment)
AND EXISTS (
SELECT 1 FROM task_supplement_capability cap
WHERE cap.task_id = agent_task_queue.id
AND cap.capability = 'task-supplement-v1'
)
FOR UPDATE OF agent_task_queue
)
UPDATE task_supplement s
SET status = 'pending',
failure_reason = NULL,
updated_at = now()
FROM active_task t
WHERE s.task_id = t.id
AND s.comment_id = $1
AND s.workspace_id = $2
AND s.status = 'failed'
RETURNING s.task_id, s.workspace_id, s.issue_id, s.comment_id, s.author_id, s.client_request_id, s.status, s.failure_reason, s.attempt_count, s.created_at, s.updated_at, s.delivered_at
`
type RetryTaskSupplementParams struct {
CommentID pgtype.UUID `json:"comment_id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
TaskID pgtype.UUID `json:"task_id"`
IssueID pgtype.UUID `json:"issue_id"`
}
func (q *Queries) RetryTaskSupplement(ctx context.Context, arg RetryTaskSupplementParams) (TaskSupplement, error) {
row := q.db.QueryRow(ctx, retryTaskSupplement,
arg.CommentID,
arg.WorkspaceID,
arg.TaskID,
arg.IssueID,
)
var i TaskSupplement
err := row.Scan(
&i.TaskID,
&i.WorkspaceID,
&i.IssueID,
&i.CommentID,
&i.AuthorID,
&i.ClientRequestID,
&i.Status,
&i.FailureReason,
&i.AttemptCount,
&i.CreatedAt,
&i.UpdatedAt,
&i.DeliveredAt,
)
return i, err
}
const settleTerminalTaskSupplements = `-- name: SettleTerminalTaskSupplements :execrows
UPDATE task_supplement AS supplement
SET status = 'failed',
failure_reason = 'turn_ended',
updated_at = now()
FROM agent_task_queue AS task
WHERE supplement.task_id = task.id
AND task.id = ANY($1::uuid[])
AND task.status IN ('completed', 'failed', 'cancelled')
AND supplement.status IN ('pending', 'delivering')
`
// Application terminal transitions call this in the same transaction as the
// agent_task_queue update. Re-checking the persisted task status makes an
// accidental early call a no-op while preserving the task-row -> supplement
// lock order used by creation and delivery acknowledgement.
func (q *Queries) SettleTerminalTaskSupplements(ctx context.Context, taskIds []pgtype.UUID) (int64, error) {
result, err := q.db.Exec(ctx, settleTerminalTaskSupplements, taskIds)
if err != nil {
return 0, err
}
return result.RowsAffected(), nil
}
const startAgentTaskWithSupplement = `-- name: StartAgentTaskWithSupplement :one
WITH candidate AS MATERIALIZED (
SELECT t.id, t.issue_id, r.workspace_id, r.provider
FROM agent_task_queue t
JOIN agent_runtime r ON r.id = t.runtime_id
WHERE t.id = $1
AND t.status IN ('dispatched', 'waiting_local_directory')
FOR UPDATE OF t
), capability AS (
INSERT INTO task_supplement_capability (task_id, workspace_id, issue_id, capability)
SELECT id, workspace_id, issue_id, 'task-supplement-v1'
FROM candidate
WHERE $2::boolean
AND provider IN ('codex', 'claude', 'grok')
AND issue_id IS NOT NULL
ON CONFLICT DO NOTHING
RETURNING task_id
)
UPDATE agent_task_queue t
SET status = 'running',
started_at = now(),
wait_reason = NULL,
prepare_lease_expires_at = NULL
FROM candidate
WHERE t.id = candidate.id
-- Reference the data-modifying CTE explicitly: capability persistence and
-- the returned running row are one indivisible statement.
AND (SELECT count(*) FROM capability) >= 0
RETURNING t.id, t.agent_id, t.issue_id, t.status, t.priority, t.dispatched_at, t.started_at, t.completed_at, t.result, t.error, t.created_at, t.context, t.runtime_id, t.session_id, t.work_dir, t.trigger_comment_id, t.chat_session_id, t.autopilot_run_id, t.attempt, t.max_attempts, t.parent_task_id, t.failure_reason, t.trigger_summary, t.force_fresh_session, t.is_leader_task, t.wait_reason, t.initiator_user_id, t.handoff_note, t.prepare_lease_expires_at, t.squad_id, t.runtime_mcp_overlay, t.escalation_for_task_id, t.fire_at, t.originator_user_id, t.runtime_connected_apps, t.coalesced_comment_ids, t.delivered_comment_ids, t.chat_input_task_id, t.chat_finalize_deferred_at, t.originator_source, t.delegated_from_task_id, t.retry_of_task_id, t.rerun_of_task_id, t.rule_version_id, t.trigger_evidence_kind, t.trigger_evidence_ref_id, t.accountable_user_id, t.session_rollout_missing, t.retired_session_id, t.quick_actions_disabled, t.regenerate_quick_actions_for, t.branch_name, t.durable_work_dir, t.channel_context_revision, t.comment_thread_id, t.cancelled_by_type, t.cancelled_by_id, t.cancelled_by_name, t.issue_snapshot
`
type StartAgentTaskWithSupplementParams struct {
TaskID pgtype.UUID `json:"task_id"`
EnableTaskSupplement bool `json:"enable_task_supplement"`
}
// Starting the task and recording the exact daemon/server capability handshake
// are one state transition. A missing row is the fail-closed value for old
// daemons, old servers, unsupported providers and application rollback.
func (q *Queries) StartAgentTaskWithSupplement(ctx context.Context, arg StartAgentTaskWithSupplementParams) (AgentTaskQueue, error) {
row := q.db.QueryRow(ctx, startAgentTaskWithSupplement, arg.TaskID, arg.EnableTaskSupplement)
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,
&i.RetiredSessionID,
&i.QuickActionsDisabled,
&i.RegenerateQuickActionsFor,
&i.BranchName,
&i.DurableWorkDir,
&i.ChannelContextRevision,
&i.CommentThreadID,
&i.CancelledByType,
&i.CancelledByID,
&i.CancelledByName,
&i.IssueSnapshot,
)
return i, err
}