mirror of
https://gh-proxy.org/https://github.com/klinsmaya/multica.git
synced 2026-11-08 01:16:45 +08:00
Claude
2ce16d8842
Resolves the UpdateAgent conflict in handler/agent.go by keeping both the connector_service_allowlist write and main's runtime-binding transaction, renumbers the open-connector migrations to 590-596 because main took 583/584, and regenerates sqlc. Co-Authored-By: Claude Fable 5.1 <[email protected]> Claude-Session: https://claude.ai/code/session_01XqqpKySn2iWtXriK3Gsgak
833 lines
28 KiB
Go
833 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', 'cursor')
|
|
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', 'cursor')
|
|
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', 'cursor')
|
|
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, t.workflow_step, t.runtime_connector_handles
|
|
`
|
|
|
|
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,
|
|
&i.WorkflowStep,
|
|
&i.RuntimeConnectorHandles,
|
|
)
|
|
return i, err
|
|
}
|