mirror of
https://gh-proxy.org/https://github.com/klinsmaya/multica.git
synced 2026-11-09 18:06:45 +08:00
Bohan Jiangandmultica-agent
bd70632999
Migration 518/520 (#8467) registered stop_issue_wakeups, an AFTER UPDATE OF status trigger on issue that disabled issue_wakeup rows and cancelled queued/deferred wakeup runs in agent_task_queue. That is a database-level cascading update, which the repository's migration rules forbid. Drop the trigger and function (migration 533) and apply the same close rule in the transaction that writes the status: single and batch issue updates and the merged-PR auto-close. Cancelled runs are now broadcast after commit. Co-authored-by: multica-agent <[email protected]>
1261 lines
44 KiB
Go
1261 lines
44 KiB
Go
// Code generated by sqlc. DO NOT EDIT.
|
|
// versions:
|
|
// sqlc v1.31.1
|
|
// source: wakeup.sql
|
|
|
|
package db
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
)
|
|
|
|
const advanceIssueWakeup = `-- name: AdvanceIssueWakeup :exec
|
|
UPDATE issue_wakeup SET enabled= $1,next_fire_at=$2,last_task_id=COALESCE($3,last_task_id),last_error=$4,updated_at=clock_timestamp() WHERE id= $5
|
|
`
|
|
|
|
type AdvanceIssueWakeupParams struct {
|
|
Enabled bool `json:"enabled"`
|
|
NextFireAt pgtype.Timestamptz `json:"next_fire_at"`
|
|
LastTaskID pgtype.UUID `json:"last_task_id"`
|
|
LastError pgtype.Text `json:"last_error"`
|
|
ID pgtype.UUID `json:"id"`
|
|
}
|
|
|
|
func (q *Queries) AdvanceIssueWakeup(ctx context.Context, arg AdvanceIssueWakeupParams) error {
|
|
_, err := q.db.Exec(ctx, advanceIssueWakeup,
|
|
arg.Enabled,
|
|
arg.NextFireAt,
|
|
arg.LastTaskID,
|
|
arg.LastError,
|
|
arg.ID,
|
|
)
|
|
return err
|
|
}
|
|
|
|
const cancelUnstartedIssueWakeupTasks = `-- name: CancelUnstartedIssueWakeupTasks :many
|
|
UPDATE agent_task_queue SET status='cancelled',completed_at=now(),error='Issue closed; wakeup disabled'
|
|
WHERE issue_id= $1 AND context->>'wakeup_id' IS NOT NULL AND status IN ('queued','deferred') AND started_at IS NULL RETURNING id, agent_id, issue_id, status, priority, dispatched_at, started_at, completed_at, result, error, created_at, context, runtime_id, session_id, work_dir, trigger_comment_id, chat_session_id, autopilot_run_id, attempt, max_attempts, parent_task_id, failure_reason, trigger_summary, force_fresh_session, is_leader_task, wait_reason, initiator_user_id, handoff_note, prepare_lease_expires_at, squad_id, runtime_mcp_overlay, escalation_for_task_id, fire_at, originator_user_id, runtime_connected_apps, coalesced_comment_ids, delivered_comment_ids, chat_input_task_id, chat_finalize_deferred_at, originator_source, delegated_from_task_id, retry_of_task_id, rerun_of_task_id, rule_version_id, trigger_evidence_kind, trigger_evidence_ref_id, accountable_user_id, session_rollout_missing, retired_session_id, quick_actions_disabled, regenerate_quick_actions_for, branch_name, durable_work_dir, channel_context_revision, comment_thread_id, cancelled_by_type, cancelled_by_id, cancelled_by_name, issue_snapshot
|
|
`
|
|
|
|
func (q *Queries) CancelUnstartedIssueWakeupTasks(ctx context.Context, issueID pgtype.UUID) ([]AgentTaskQueue, error) {
|
|
rows, err := q.db.Query(ctx, cancelUnstartedIssueWakeupTasks, issueID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []AgentTaskQueue{}
|
|
for rows.Next() {
|
|
var i AgentTaskQueue
|
|
if err := rows.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,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const cancelUnstartedWakeupTasks = `-- name: CancelUnstartedWakeupTasks :many
|
|
UPDATE agent_task_queue SET status='cancelled',completed_at=now(),error='Wakeup disabled or updated'
|
|
WHERE context->>'wakeup_id'= $1::text AND status IN ('queued','deferred') AND started_at IS NULL RETURNING id, agent_id, issue_id, status, priority, dispatched_at, started_at, completed_at, result, error, created_at, context, runtime_id, session_id, work_dir, trigger_comment_id, chat_session_id, autopilot_run_id, attempt, max_attempts, parent_task_id, failure_reason, trigger_summary, force_fresh_session, is_leader_task, wait_reason, initiator_user_id, handoff_note, prepare_lease_expires_at, squad_id, runtime_mcp_overlay, escalation_for_task_id, fire_at, originator_user_id, runtime_connected_apps, coalesced_comment_ids, delivered_comment_ids, chat_input_task_id, chat_finalize_deferred_at, originator_source, delegated_from_task_id, retry_of_task_id, rerun_of_task_id, rule_version_id, trigger_evidence_kind, trigger_evidence_ref_id, accountable_user_id, session_rollout_missing, retired_session_id, quick_actions_disabled, regenerate_quick_actions_for, branch_name, durable_work_dir, channel_context_revision, comment_thread_id, cancelled_by_type, cancelled_by_id, cancelled_by_name, issue_snapshot
|
|
`
|
|
|
|
func (q *Queries) CancelUnstartedWakeupTasks(ctx context.Context, wakeupID string) ([]AgentTaskQueue, error) {
|
|
rows, err := q.db.Query(ctx, cancelUnstartedWakeupTasks, wakeupID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []AgentTaskQueue{}
|
|
for rows.Next() {
|
|
var i AgentTaskQueue
|
|
if err := rows.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,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const consumeWakeupReceipts = `-- name: ConsumeWakeupReceipts :exec
|
|
UPDATE issue_wakeup_receipt SET task_id=$1,processed_at=now() WHERE id=ANY($2::uuid[])
|
|
`
|
|
|
|
type ConsumeWakeupReceiptsParams struct {
|
|
TaskID pgtype.UUID `json:"task_id"`
|
|
Ids []pgtype.UUID `json:"ids"`
|
|
}
|
|
|
|
func (q *Queries) ConsumeWakeupReceipts(ctx context.Context, arg ConsumeWakeupReceiptsParams) error {
|
|
_, err := q.db.Exec(ctx, consumeWakeupReceipts, arg.TaskID, arg.Ids)
|
|
return err
|
|
}
|
|
|
|
const createIssueWakeup = `-- name: CreateIssueWakeup :one
|
|
INSERT INTO issue_wakeup(id,workspace_id,issue_id,agent_id,created_by,source_task_id,parent_comment_id,instruction,kind,mode,event_types,filter_agent_id,filter_task_id,filter_actor_type,filter_actor_id,interval_seconds,cron_expression,timezone,next_fire_at)
|
|
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19) RETURNING id, workspace_id, issue_id, agent_id, created_by, source_task_id, parent_comment_id, instruction, kind, mode, event_types, filter_agent_id, filter_task_id, interval_seconds, cron_expression, timezone, next_fire_at, enabled, disabled_at, revision, last_task_id, last_error, created_at, updated_at, filter_actor_type, filter_actor_id
|
|
`
|
|
|
|
type CreateIssueWakeupParams struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
IssueID pgtype.UUID `json:"issue_id"`
|
|
AgentID pgtype.UUID `json:"agent_id"`
|
|
CreatedBy pgtype.UUID `json:"created_by"`
|
|
SourceTaskID pgtype.UUID `json:"source_task_id"`
|
|
ParentCommentID pgtype.UUID `json:"parent_comment_id"`
|
|
Instruction string `json:"instruction"`
|
|
Kind string `json:"kind"`
|
|
Mode string `json:"mode"`
|
|
EventTypes []string `json:"event_types"`
|
|
FilterAgentID pgtype.UUID `json:"filter_agent_id"`
|
|
FilterTaskID pgtype.UUID `json:"filter_task_id"`
|
|
FilterActorType pgtype.Text `json:"filter_actor_type"`
|
|
FilterActorID pgtype.UUID `json:"filter_actor_id"`
|
|
IntervalSeconds pgtype.Int8 `json:"interval_seconds"`
|
|
CronExpression pgtype.Text `json:"cron_expression"`
|
|
Timezone string `json:"timezone"`
|
|
NextFireAt pgtype.Timestamptz `json:"next_fire_at"`
|
|
}
|
|
|
|
func (q *Queries) CreateIssueWakeup(ctx context.Context, arg CreateIssueWakeupParams) (IssueWakeup, error) {
|
|
row := q.db.QueryRow(ctx, createIssueWakeup,
|
|
arg.ID,
|
|
arg.WorkspaceID,
|
|
arg.IssueID,
|
|
arg.AgentID,
|
|
arg.CreatedBy,
|
|
arg.SourceTaskID,
|
|
arg.ParentCommentID,
|
|
arg.Instruction,
|
|
arg.Kind,
|
|
arg.Mode,
|
|
arg.EventTypes,
|
|
arg.FilterAgentID,
|
|
arg.FilterTaskID,
|
|
arg.FilterActorType,
|
|
arg.FilterActorID,
|
|
arg.IntervalSeconds,
|
|
arg.CronExpression,
|
|
arg.Timezone,
|
|
arg.NextFireAt,
|
|
)
|
|
var i IssueWakeup
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.IssueID,
|
|
&i.AgentID,
|
|
&i.CreatedBy,
|
|
&i.SourceTaskID,
|
|
&i.ParentCommentID,
|
|
&i.Instruction,
|
|
&i.Kind,
|
|
&i.Mode,
|
|
&i.EventTypes,
|
|
&i.FilterAgentID,
|
|
&i.FilterTaskID,
|
|
&i.IntervalSeconds,
|
|
&i.CronExpression,
|
|
&i.Timezone,
|
|
&i.NextFireAt,
|
|
&i.Enabled,
|
|
&i.DisabledAt,
|
|
&i.Revision,
|
|
&i.LastTaskID,
|
|
&i.LastError,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
&i.FilterActorType,
|
|
&i.FilterActorID,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const createWakeupTask = `-- name: CreateWakeupTask :one
|
|
INSERT INTO agent_task_queue (
|
|
agent_id, runtime_id, issue_id, status, priority, trigger_comment_id,
|
|
coalesced_comment_ids, trigger_summary, force_fresh_session, is_leader_task, handoff_note,
|
|
squad_id, context, originator_user_id, accountable_user_id, runtime_mcp_overlay, runtime_connected_apps,
|
|
originator_source, delegated_from_task_id, rule_version_id, rerun_of_task_id, trigger_evidence_kind, trigger_evidence_ref_id,
|
|
id
|
|
)
|
|
SELECT
|
|
$1, $2, $3, 'queued', $4, $5,
|
|
COALESCE($6::uuid[], '{}'),
|
|
$7,
|
|
COALESCE($8::boolean, FALSE),
|
|
COALESCE($9::boolean, FALSE),
|
|
$10,
|
|
$11,
|
|
CASE
|
|
WHEN COALESCE($12::text, '') <> ''
|
|
THEN jsonb_build_object('head_sha', $12::text)
|
|
ELSE '{}'::jsonb
|
|
END || $13::jsonb,
|
|
$14,
|
|
$15,
|
|
$16,
|
|
$17,
|
|
$18,
|
|
$19,
|
|
$20,
|
|
$21,
|
|
$22,
|
|
$23,
|
|
COALESCE($24::uuid, gen_random_uuid())
|
|
WHERE lock_task_owner_rows($1, $3, $2)
|
|
RETURNING id, agent_id, issue_id, status, priority, dispatched_at, started_at, completed_at, result, error, created_at, context, runtime_id, session_id, work_dir, trigger_comment_id, chat_session_id, autopilot_run_id, attempt, max_attempts, parent_task_id, failure_reason, trigger_summary, force_fresh_session, is_leader_task, wait_reason, initiator_user_id, handoff_note, prepare_lease_expires_at, squad_id, runtime_mcp_overlay, escalation_for_task_id, fire_at, originator_user_id, runtime_connected_apps, coalesced_comment_ids, delivered_comment_ids, chat_input_task_id, chat_finalize_deferred_at, originator_source, delegated_from_task_id, retry_of_task_id, rerun_of_task_id, rule_version_id, trigger_evidence_kind, trigger_evidence_ref_id, accountable_user_id, session_rollout_missing, retired_session_id, quick_actions_disabled, regenerate_quick_actions_for, branch_name, durable_work_dir, channel_context_revision, comment_thread_id, cancelled_by_type, cancelled_by_id, cancelled_by_name, issue_snapshot
|
|
`
|
|
|
|
type CreateWakeupTaskParams struct {
|
|
AgentID pgtype.UUID `json:"agent_id"`
|
|
RuntimeID pgtype.UUID `json:"runtime_id"`
|
|
IssueID pgtype.UUID `json:"issue_id"`
|
|
Priority int32 `json:"priority"`
|
|
TriggerCommentID pgtype.UUID `json:"trigger_comment_id"`
|
|
CoalescedCommentIds []pgtype.UUID `json:"coalesced_comment_ids"`
|
|
TriggerSummary pgtype.Text `json:"trigger_summary"`
|
|
ForceFreshSession pgtype.Bool `json:"force_fresh_session"`
|
|
IsLeaderTask pgtype.Bool `json:"is_leader_task"`
|
|
HandoffNote pgtype.Text `json:"handoff_note"`
|
|
SquadID pgtype.UUID `json:"squad_id"`
|
|
HeadSha pgtype.Text `json:"head_sha"`
|
|
WakeupContext []byte `json:"wakeup_context"`
|
|
OriginatorUserID pgtype.UUID `json:"originator_user_id"`
|
|
AccountableUserID pgtype.UUID `json:"accountable_user_id"`
|
|
RuntimeMcpOverlay []byte `json:"runtime_mcp_overlay"`
|
|
RuntimeConnectedApps []byte `json:"runtime_connected_apps"`
|
|
OriginatorSource pgtype.Text `json:"originator_source"`
|
|
DelegatedFromTaskID pgtype.UUID `json:"delegated_from_task_id"`
|
|
RuleVersionID pgtype.UUID `json:"rule_version_id"`
|
|
RerunOfTaskID pgtype.UUID `json:"rerun_of_task_id"`
|
|
TriggerEvidenceKind pgtype.Text `json:"trigger_evidence_kind"`
|
|
TriggerEvidenceRefID pgtype.UUID `json:"trigger_evidence_ref_id"`
|
|
ID pgtype.UUID `json:"id"`
|
|
}
|
|
|
|
// Fenced against workspace teardown: lock_task_owner_rows (migration 284)
|
|
// locks the owners' workspace rows in the writer's own transaction and returns
|
|
// false once they are gone, so this statement writes no row instead of stranding
|
|
// a task in a workspace that has just been deleted (MUL-5999).
|
|
// head_sha stamps the commit under review into the task's context JSONB so the
|
|
// reviewer-loop dedup (HasPendingTaskForIssueAndAgent) can tell a pending run
|
|
// against an OLD head apart from a fresh request against a NEW head (TEN-356).
|
|
// Empty/absent head_sha leaves context NULL, preserving pre-TEN-356 behavior for
|
|
// issues with no linked PR. Issue-linked tasks never hit quick-create context
|
|
// parsing (parseQuickCreateContext short-circuits on IssueID.Valid), so this
|
|
// key rides harmlessly alongside.
|
|
// id is minted by the application as a UUIDv7 (pkg/dbid) so consecutive
|
|
// enqueues cluster in a narrow contiguous primary-key range instead of
|
|
// scattering across the B-tree. On a table with existing v4 ids, that range
|
|
// is not necessarily the tree's right edge.
|
|
// COALESCE keeps the column's gen_random_uuid() default reachable, so a caller
|
|
// that passes no id still inserts — it just gets a random v4, exactly as before.
|
|
// The same pattern is used by every INSERT listed in pkg/dbid's write table.
|
|
func (q *Queries) CreateWakeupTask(ctx context.Context, arg CreateWakeupTaskParams) (AgentTaskQueue, error) {
|
|
row := q.db.QueryRow(ctx, createWakeupTask,
|
|
arg.AgentID,
|
|
arg.RuntimeID,
|
|
arg.IssueID,
|
|
arg.Priority,
|
|
arg.TriggerCommentID,
|
|
arg.CoalescedCommentIds,
|
|
arg.TriggerSummary,
|
|
arg.ForceFreshSession,
|
|
arg.IsLeaderTask,
|
|
arg.HandoffNote,
|
|
arg.SquadID,
|
|
arg.HeadSha,
|
|
arg.WakeupContext,
|
|
arg.OriginatorUserID,
|
|
arg.AccountableUserID,
|
|
arg.RuntimeMcpOverlay,
|
|
arg.RuntimeConnectedApps,
|
|
arg.OriginatorSource,
|
|
arg.DelegatedFromTaskID,
|
|
arg.RuleVersionID,
|
|
arg.RerunOfTaskID,
|
|
arg.TriggerEvidenceKind,
|
|
arg.TriggerEvidenceRefID,
|
|
arg.ID,
|
|
)
|
|
var i AgentTaskQueue
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.AgentID,
|
|
&i.IssueID,
|
|
&i.Status,
|
|
&i.Priority,
|
|
&i.DispatchedAt,
|
|
&i.StartedAt,
|
|
&i.CompletedAt,
|
|
&i.Result,
|
|
&i.Error,
|
|
&i.CreatedAt,
|
|
&i.Context,
|
|
&i.RuntimeID,
|
|
&i.SessionID,
|
|
&i.WorkDir,
|
|
&i.TriggerCommentID,
|
|
&i.ChatSessionID,
|
|
&i.AutopilotRunID,
|
|
&i.Attempt,
|
|
&i.MaxAttempts,
|
|
&i.ParentTaskID,
|
|
&i.FailureReason,
|
|
&i.TriggerSummary,
|
|
&i.ForceFreshSession,
|
|
&i.IsLeaderTask,
|
|
&i.WaitReason,
|
|
&i.InitiatorUserID,
|
|
&i.HandoffNote,
|
|
&i.PrepareLeaseExpiresAt,
|
|
&i.SquadID,
|
|
&i.RuntimeMcpOverlay,
|
|
&i.EscalationForTaskID,
|
|
&i.FireAt,
|
|
&i.OriginatorUserID,
|
|
&i.RuntimeConnectedApps,
|
|
&i.CoalescedCommentIds,
|
|
&i.DeliveredCommentIds,
|
|
&i.ChatInputTaskID,
|
|
&i.ChatFinalizeDeferredAt,
|
|
&i.OriginatorSource,
|
|
&i.DelegatedFromTaskID,
|
|
&i.RetryOfTaskID,
|
|
&i.RerunOfTaskID,
|
|
&i.RuleVersionID,
|
|
&i.TriggerEvidenceKind,
|
|
&i.TriggerEvidenceRefID,
|
|
&i.AccountableUserID,
|
|
&i.SessionRolloutMissing,
|
|
&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
|
|
}
|
|
|
|
const deleteExpiredWakeupReceipts = `-- name: DeleteExpiredWakeupReceipts :execrows
|
|
WITH batch AS MATERIALIZED (
|
|
SELECT expired.id FROM issue_wakeup_receipt expired WHERE expired.processed_at < $1
|
|
ORDER BY expired.processed_at,expired.id LIMIT 1000 FOR UPDATE SKIP LOCKED
|
|
)
|
|
DELETE FROM issue_wakeup_receipt r USING batch WHERE r.id=batch.id
|
|
`
|
|
|
|
// Pending inputs are never expired. Bound work and avoid waiting on dispatch.
|
|
// MATERIALIZED evaluates the batch once. As an IN subquery, a nested-loop plan
|
|
// rescans it per outer row, skips the rows this DELETE already removed, and
|
|
// slides the LIMIT window until every expired row is gone.
|
|
func (q *Queries) DeleteExpiredWakeupReceipts(ctx context.Context, cutoff pgtype.Timestamptz) (int64, error) {
|
|
result, err := q.db.Exec(ctx, deleteExpiredWakeupReceipts, cutoff)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected(), nil
|
|
}
|
|
|
|
const disableIssueWakeups = `-- name: DisableIssueWakeups :exec
|
|
UPDATE issue_wakeup SET enabled=false,disabled_at=clock_timestamp(),updated_at=clock_timestamp()
|
|
WHERE issue_id= $1 AND disabled_at IS NULL
|
|
`
|
|
|
|
// Closing wins over every subscription on the issue; reopening does not rearm.
|
|
func (q *Queries) DisableIssueWakeups(ctx context.Context, issueID pgtype.UUID) error {
|
|
_, err := q.db.Exec(ctx, disableIssueWakeups, issueID)
|
|
return err
|
|
}
|
|
|
|
const discardWakeupReceipts = `-- name: DiscardWakeupReceipts :exec
|
|
UPDATE issue_wakeup_receipt SET processed_at=now() WHERE wakeup_id= $1 AND processed_at IS NULL
|
|
`
|
|
|
|
func (q *Queries) DiscardWakeupReceipts(ctx context.Context, id pgtype.UUID) error {
|
|
_, err := q.db.Exec(ctx, discardWakeupReceipts, id)
|
|
return err
|
|
}
|
|
|
|
const findPendingWakeupTask = `-- name: FindPendingWakeupTask :one
|
|
SELECT id, agent_id, issue_id, status, priority, dispatched_at, started_at, completed_at, result, error, created_at, context, runtime_id, session_id, work_dir, trigger_comment_id, chat_session_id, autopilot_run_id, attempt, max_attempts, parent_task_id, failure_reason, trigger_summary, force_fresh_session, is_leader_task, wait_reason, initiator_user_id, handoff_note, prepare_lease_expires_at, squad_id, runtime_mcp_overlay, escalation_for_task_id, fire_at, originator_user_id, runtime_connected_apps, coalesced_comment_ids, delivered_comment_ids, chat_input_task_id, chat_finalize_deferred_at, originator_source, delegated_from_task_id, retry_of_task_id, rerun_of_task_id, rule_version_id, trigger_evidence_kind, trigger_evidence_ref_id, accountable_user_id, session_rollout_missing, retired_session_id, quick_actions_disabled, regenerate_quick_actions_for, branch_name, durable_work_dir, channel_context_revision, comment_thread_id, cancelled_by_type, cancelled_by_id, cancelled_by_name, issue_snapshot FROM agent_task_queue WHERE context->>'wakeup_id'= $1::text AND status IN ('queued','dispatched') ORDER BY created_at LIMIT 1 FOR UPDATE
|
|
`
|
|
|
|
func (q *Queries) FindPendingWakeupTask(ctx context.Context, wakeupID string) (AgentTaskQueue, error) {
|
|
row := q.db.QueryRow(ctx, findPendingWakeupTask, wakeupID)
|
|
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
|
|
}
|
|
|
|
const getIssueWakeup = `-- name: GetIssueWakeup :one
|
|
SELECT id, workspace_id, issue_id, agent_id, created_by, source_task_id, parent_comment_id, instruction, kind, mode, event_types, filter_agent_id, filter_task_id, interval_seconds, cron_expression, timezone, next_fire_at, enabled, disabled_at, revision, last_task_id, last_error, created_at, updated_at, filter_actor_type, filter_actor_id FROM issue_wakeup WHERE id= $1 AND workspace_id= $2
|
|
`
|
|
|
|
type GetIssueWakeupParams struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
}
|
|
|
|
func (q *Queries) GetIssueWakeup(ctx context.Context, arg GetIssueWakeupParams) (IssueWakeup, error) {
|
|
row := q.db.QueryRow(ctx, getIssueWakeup, arg.ID, arg.WorkspaceID)
|
|
var i IssueWakeup
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.IssueID,
|
|
&i.AgentID,
|
|
&i.CreatedBy,
|
|
&i.SourceTaskID,
|
|
&i.ParentCommentID,
|
|
&i.Instruction,
|
|
&i.Kind,
|
|
&i.Mode,
|
|
&i.EventTypes,
|
|
&i.FilterAgentID,
|
|
&i.FilterTaskID,
|
|
&i.IntervalSeconds,
|
|
&i.CronExpression,
|
|
&i.Timezone,
|
|
&i.NextFireAt,
|
|
&i.Enabled,
|
|
&i.DisabledAt,
|
|
&i.Revision,
|
|
&i.LastTaskID,
|
|
&i.LastError,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
&i.FilterActorType,
|
|
&i.FilterActorID,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const listIssueWakeups = `-- name: ListIssueWakeups :many
|
|
SELECT w.id,w.workspace_id,w.issue_id,w.agent_id,w.created_by,w.source_task_id,w.parent_comment_id,w.instruction,
|
|
w.kind,w.mode,w.event_types,w.filter_actor_type,
|
|
(CASE WHEN actor_agent.id IS NOT NULL OR actor_member.user_id IS NOT NULL THEN w.filter_actor_id END)::uuid AS filter_actor_id,
|
|
COALESCE(actor_agent.name,actor_user.name,'')::text AS filter_actor_name,
|
|
(CASE WHEN source.id IS NOT NULL THEN w.filter_agent_id END)::uuid AS filter_agent_id,
|
|
(CASE WHEN EXISTS(SELECT 1 FROM agent_task_queue ft JOIN agent fa ON fa.id=ft.agent_id AND fa.workspace_id=w.workspace_id
|
|
WHERE ft.id=w.filter_task_id AND ft.issue_id=w.issue_id AND fa.id=ANY($1::uuid[])) THEN w.filter_task_id END)::uuid AS filter_task_id,
|
|
w.interval_seconds,w.cron_expression,w.timezone,w.next_fire_at,w.enabled,w.disabled_at,w.revision,
|
|
w.last_task_id,w.last_error,w.created_at,w.updated_at,a.name AS agent_name,source.name AS filter_agent_name,t.status AS last_task_status
|
|
FROM issue_wakeup w JOIN agent a ON a.id=w.agent_id AND a.workspace_id=w.workspace_id
|
|
LEFT JOIN agent actor_agent ON w.filter_actor_type='agent' AND actor_agent.id=w.filter_actor_id AND actor_agent.workspace_id=w.workspace_id AND actor_agent.id=ANY($1::uuid[])
|
|
LEFT JOIN member actor_member ON w.filter_actor_type='member' AND actor_member.user_id=w.filter_actor_id AND actor_member.workspace_id=w.workspace_id
|
|
LEFT JOIN "user" actor_user ON actor_user.id=actor_member.user_id
|
|
LEFT JOIN agent source ON source.id=w.filter_agent_id AND source.workspace_id=w.workspace_id AND source.id=ANY($1::uuid[])
|
|
LEFT JOIN agent_task_queue t ON t.id=w.last_task_id AND t.issue_id=w.issue_id AND t.agent_id=w.agent_id
|
|
WHERE w.workspace_id= $2 AND w.issue_id= $3 ORDER BY w.created_at,w.id
|
|
`
|
|
|
|
type ListIssueWakeupsParams struct {
|
|
AgentIds []pgtype.UUID `json:"agent_ids"`
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
IssueID pgtype.UUID `json:"issue_id"`
|
|
}
|
|
|
|
type ListIssueWakeupsRow struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
IssueID pgtype.UUID `json:"issue_id"`
|
|
AgentID pgtype.UUID `json:"agent_id"`
|
|
CreatedBy pgtype.UUID `json:"created_by"`
|
|
SourceTaskID pgtype.UUID `json:"source_task_id"`
|
|
ParentCommentID pgtype.UUID `json:"parent_comment_id"`
|
|
Instruction string `json:"instruction"`
|
|
Kind string `json:"kind"`
|
|
Mode string `json:"mode"`
|
|
EventTypes []string `json:"event_types"`
|
|
FilterActorType pgtype.Text `json:"filter_actor_type"`
|
|
FilterActorID pgtype.UUID `json:"filter_actor_id"`
|
|
FilterActorName string `json:"filter_actor_name"`
|
|
FilterAgentID pgtype.UUID `json:"filter_agent_id"`
|
|
FilterTaskID pgtype.UUID `json:"filter_task_id"`
|
|
IntervalSeconds pgtype.Int8 `json:"interval_seconds"`
|
|
CronExpression pgtype.Text `json:"cron_expression"`
|
|
Timezone string `json:"timezone"`
|
|
NextFireAt pgtype.Timestamptz `json:"next_fire_at"`
|
|
Enabled bool `json:"enabled"`
|
|
DisabledAt pgtype.Timestamptz `json:"disabled_at"`
|
|
Revision int64 `json:"revision"`
|
|
LastTaskID pgtype.UUID `json:"last_task_id"`
|
|
LastError pgtype.Text `json:"last_error"`
|
|
CreatedAt pgtype.Timestamptz `json:"created_at"`
|
|
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
|
|
AgentName string `json:"agent_name"`
|
|
FilterAgentName pgtype.Text `json:"filter_agent_name"`
|
|
LastTaskStatus pgtype.Text `json:"last_task_status"`
|
|
}
|
|
|
|
func (q *Queries) ListIssueWakeups(ctx context.Context, arg ListIssueWakeupsParams) ([]ListIssueWakeupsRow, error) {
|
|
rows, err := q.db.Query(ctx, listIssueWakeups, arg.AgentIds, arg.WorkspaceID, arg.IssueID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListIssueWakeupsRow{}
|
|
for rows.Next() {
|
|
var i ListIssueWakeupsRow
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.IssueID,
|
|
&i.AgentID,
|
|
&i.CreatedBy,
|
|
&i.SourceTaskID,
|
|
&i.ParentCommentID,
|
|
&i.Instruction,
|
|
&i.Kind,
|
|
&i.Mode,
|
|
&i.EventTypes,
|
|
&i.FilterActorType,
|
|
&i.FilterActorID,
|
|
&i.FilterActorName,
|
|
&i.FilterAgentID,
|
|
&i.FilterTaskID,
|
|
&i.IntervalSeconds,
|
|
&i.CronExpression,
|
|
&i.Timezone,
|
|
&i.NextFireAt,
|
|
&i.Enabled,
|
|
&i.DisabledAt,
|
|
&i.Revision,
|
|
&i.LastTaskID,
|
|
&i.LastError,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
&i.AgentName,
|
|
&i.FilterAgentName,
|
|
&i.LastTaskStatus,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listPendingWakeupReceipts = `-- name: ListPendingWakeupReceipts :many
|
|
SELECT id, wakeup_id, revision, event_key, event_type, payload, task_id, processed_at, created_at, coalesce_key FROM issue_wakeup_receipt WHERE wakeup_id= $1 AND revision= $2 AND processed_at IS NULL ORDER BY created_at,id LIMIT 100 FOR UPDATE
|
|
`
|
|
|
|
type ListPendingWakeupReceiptsParams struct {
|
|
WakeupID pgtype.UUID `json:"wakeup_id"`
|
|
Revision int64 `json:"revision"`
|
|
}
|
|
|
|
func (q *Queries) ListPendingWakeupReceipts(ctx context.Context, arg ListPendingWakeupReceiptsParams) ([]IssueWakeupReceipt, error) {
|
|
rows, err := q.db.Query(ctx, listPendingWakeupReceipts, arg.WakeupID, arg.Revision)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []IssueWakeupReceipt{}
|
|
for rows.Next() {
|
|
var i IssueWakeupReceipt
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.WakeupID,
|
|
&i.Revision,
|
|
&i.EventKey,
|
|
&i.EventType,
|
|
&i.Payload,
|
|
&i.TaskID,
|
|
&i.ProcessedAt,
|
|
&i.CreatedAt,
|
|
&i.CoalesceKey,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listReadyWakeups = `-- name: ListReadyWakeups :many
|
|
WITH candidates AS (
|
|
SELECT id FROM issue_wakeup WHERE enabled AND kind<>'event' AND next_fire_at<=now()
|
|
UNION
|
|
SELECT wakeup_id FROM issue_wakeup_receipt WHERE processed_at IS NULL
|
|
)
|
|
SELECT w.id, w.workspace_id, w.issue_id, w.agent_id, w.created_by, w.source_task_id, w.parent_comment_id, w.instruction, w.kind, w.mode, w.event_types, w.filter_agent_id, w.filter_task_id, w.interval_seconds, w.cron_expression, w.timezone, w.next_fire_at, w.enabled, w.disabled_at, w.revision, w.last_task_id, w.last_error, w.created_at, w.updated_at, w.filter_actor_type, w.filter_actor_id FROM candidates c JOIN issue_wakeup w ON w.id=c.id
|
|
ORDER BY w.updated_at,w.id LIMIT 100
|
|
`
|
|
|
|
func (q *Queries) ListReadyWakeups(ctx context.Context) ([]IssueWakeup, error) {
|
|
rows, err := q.db.Query(ctx, listReadyWakeups)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []IssueWakeup{}
|
|
for rows.Next() {
|
|
var i IssueWakeup
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.IssueID,
|
|
&i.AgentID,
|
|
&i.CreatedBy,
|
|
&i.SourceTaskID,
|
|
&i.ParentCommentID,
|
|
&i.Instruction,
|
|
&i.Kind,
|
|
&i.Mode,
|
|
&i.EventTypes,
|
|
&i.FilterAgentID,
|
|
&i.FilterTaskID,
|
|
&i.IntervalSeconds,
|
|
&i.CronExpression,
|
|
&i.Timezone,
|
|
&i.NextFireAt,
|
|
&i.Enabled,
|
|
&i.DisabledAt,
|
|
&i.Revision,
|
|
&i.LastTaskID,
|
|
&i.LastError,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
&i.FilterActorType,
|
|
&i.FilterActorID,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listWorkspaceWakeupSummaryRows = `-- name: ListWorkspaceWakeupSummaryRows :many
|
|
WITH ranked AS (
|
|
SELECT w.issue_id,w.id,w.agent_id,a.name AS agent_name,w.kind,w.mode,w.event_types,w.filter_actor_type,
|
|
(CASE WHEN actor_agent.id IS NOT NULL OR actor_member.user_id IS NOT NULL THEN w.filter_actor_id END)::uuid AS filter_actor_id,
|
|
COALESCE(actor_agent.name,actor_user.name,'')::text AS filter_actor_name,
|
|
(CASE WHEN EXISTS(SELECT 1 FROM agent_task_queue ft JOIN agent fa ON fa.id=ft.agent_id AND fa.workspace_id=w.workspace_id
|
|
WHERE ft.id=w.filter_task_id AND ft.issue_id=w.issue_id AND fa.id=ANY($1::uuid[])) THEN w.filter_task_id END)::uuid AS filter_task_id,
|
|
source.name AS filter_agent_name,w.interval_seconds,w.cron_expression,w.timezone,w.next_fire_at,
|
|
count(*) OVER(PARTITION BY w.issue_id) AS active_count,
|
|
count(*) FILTER(WHERE w.kind='event') OVER(PARTITION BY w.issue_id) AS event_count,
|
|
row_number() OVER(PARTITION BY w.issue_id ORDER BY w.next_fire_at NULLS LAST,w.created_at,w.id) AS rank
|
|
FROM issue_wakeup w
|
|
JOIN issue i ON i.id=w.issue_id AND i.workspace_id=w.workspace_id
|
|
JOIN agent a ON a.id=w.agent_id AND a.workspace_id=w.workspace_id
|
|
LEFT JOIN agent actor_agent ON w.filter_actor_type='agent' AND actor_agent.id=w.filter_actor_id AND actor_agent.workspace_id=w.workspace_id AND actor_agent.id=ANY($1::uuid[])
|
|
LEFT JOIN member actor_member ON w.filter_actor_type='member' AND actor_member.user_id=w.filter_actor_id AND actor_member.workspace_id=w.workspace_id
|
|
LEFT JOIN "user" actor_user ON actor_user.id=actor_member.user_id
|
|
LEFT JOIN agent source ON source.id=w.filter_agent_id AND source.workspace_id=w.workspace_id AND source.id=ANY($1::uuid[])
|
|
WHERE w.workspace_id= $2 AND w.enabled
|
|
AND i.status NOT IN ('done','cancelled')
|
|
AND NOT EXISTS(SELECT 1 FROM issue_status s WHERE s.workspace_id=i.workspace_id AND s.key=i.status AND s.category IN ('done','closed'))
|
|
)
|
|
SELECT issue_id,id,agent_id,agent_name,kind,mode,event_types,filter_actor_type,filter_actor_id,filter_actor_name,filter_task_id,filter_agent_name,interval_seconds,cron_expression,timezone,next_fire_at,active_count,event_count
|
|
FROM ranked WHERE rank<=3 ORDER BY issue_id,rank
|
|
`
|
|
|
|
type ListWorkspaceWakeupSummaryRowsParams struct {
|
|
AgentIds []pgtype.UUID `json:"agent_ids"`
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
}
|
|
|
|
type ListWorkspaceWakeupSummaryRowsRow struct {
|
|
IssueID pgtype.UUID `json:"issue_id"`
|
|
ID pgtype.UUID `json:"id"`
|
|
AgentID pgtype.UUID `json:"agent_id"`
|
|
AgentName string `json:"agent_name"`
|
|
Kind string `json:"kind"`
|
|
Mode string `json:"mode"`
|
|
EventTypes []string `json:"event_types"`
|
|
FilterActorType pgtype.Text `json:"filter_actor_type"`
|
|
FilterActorID pgtype.UUID `json:"filter_actor_id"`
|
|
FilterActorName string `json:"filter_actor_name"`
|
|
FilterTaskID pgtype.UUID `json:"filter_task_id"`
|
|
FilterAgentName pgtype.Text `json:"filter_agent_name"`
|
|
IntervalSeconds pgtype.Int8 `json:"interval_seconds"`
|
|
CronExpression pgtype.Text `json:"cron_expression"`
|
|
Timezone string `json:"timezone"`
|
|
NextFireAt pgtype.Timestamptz `json:"next_fire_at"`
|
|
ActiveCount int64 `json:"active_count"`
|
|
EventCount int64 `json:"event_count"`
|
|
}
|
|
|
|
// No prompts/history; at most three previews per issue plus exact counts.
|
|
func (q *Queries) ListWorkspaceWakeupSummaryRows(ctx context.Context, arg ListWorkspaceWakeupSummaryRowsParams) ([]ListWorkspaceWakeupSummaryRowsRow, error) {
|
|
rows, err := q.db.Query(ctx, listWorkspaceWakeupSummaryRows, arg.AgentIds, arg.WorkspaceID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ListWorkspaceWakeupSummaryRowsRow{}
|
|
for rows.Next() {
|
|
var i ListWorkspaceWakeupSummaryRowsRow
|
|
if err := rows.Scan(
|
|
&i.IssueID,
|
|
&i.ID,
|
|
&i.AgentID,
|
|
&i.AgentName,
|
|
&i.Kind,
|
|
&i.Mode,
|
|
&i.EventTypes,
|
|
&i.FilterActorType,
|
|
&i.FilterActorID,
|
|
&i.FilterActorName,
|
|
&i.FilterTaskID,
|
|
&i.FilterAgentName,
|
|
&i.IntervalSeconds,
|
|
&i.CronExpression,
|
|
&i.Timezone,
|
|
&i.NextFireAt,
|
|
&i.ActiveCount,
|
|
&i.EventCount,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const lockIssueWakeup = `-- name: LockIssueWakeup :one
|
|
SELECT id, workspace_id, issue_id, agent_id, created_by, source_task_id, parent_comment_id, instruction, kind, mode, event_types, filter_agent_id, filter_task_id, interval_seconds, cron_expression, timezone, next_fire_at, enabled, disabled_at, revision, last_task_id, last_error, created_at, updated_at, filter_actor_type, filter_actor_id FROM issue_wakeup WHERE id= $1 FOR UPDATE
|
|
`
|
|
|
|
func (q *Queries) LockIssueWakeup(ctx context.Context, id pgtype.UUID) (IssueWakeup, error) {
|
|
row := q.db.QueryRow(ctx, lockIssueWakeup, id)
|
|
var i IssueWakeup
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.IssueID,
|
|
&i.AgentID,
|
|
&i.CreatedBy,
|
|
&i.SourceTaskID,
|
|
&i.ParentCommentID,
|
|
&i.Instruction,
|
|
&i.Kind,
|
|
&i.Mode,
|
|
&i.EventTypes,
|
|
&i.FilterAgentID,
|
|
&i.FilterTaskID,
|
|
&i.IntervalSeconds,
|
|
&i.CronExpression,
|
|
&i.Timezone,
|
|
&i.NextFireAt,
|
|
&i.Enabled,
|
|
&i.DisabledAt,
|
|
&i.Revision,
|
|
&i.LastTaskID,
|
|
&i.LastError,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
&i.FilterActorType,
|
|
&i.FilterActorID,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const lockWakeupIssue = `-- name: LockWakeupIssue :one
|
|
SELECT id, workspace_id, title, description, status, priority, assignee_type, assignee_id, creator_type, creator_id, parent_issue_id, acceptance_criteria, context_refs, position, due_date, created_at, updated_at, number, project_id, origin_type, origin_id, first_executed_at, start_date, metadata, stage, properties, revision, last_activity_at, triage_state FROM issue WHERE id= $1 FOR NO KEY UPDATE
|
|
`
|
|
|
|
func (q *Queries) LockWakeupIssue(ctx context.Context, id pgtype.UUID) (Issue, error) {
|
|
row := q.db.QueryRow(ctx, lockWakeupIssue, id)
|
|
var i Issue
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.Title,
|
|
&i.Description,
|
|
&i.Status,
|
|
&i.Priority,
|
|
&i.AssigneeType,
|
|
&i.AssigneeID,
|
|
&i.CreatorType,
|
|
&i.CreatorID,
|
|
&i.ParentIssueID,
|
|
&i.AcceptanceCriteria,
|
|
&i.ContextRefs,
|
|
&i.Position,
|
|
&i.DueDate,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
&i.Number,
|
|
&i.ProjectID,
|
|
&i.OriginType,
|
|
&i.OriginID,
|
|
&i.FirstExecutedAt,
|
|
&i.StartDate,
|
|
&i.Metadata,
|
|
&i.Stage,
|
|
&i.Properties,
|
|
&i.Revision,
|
|
&i.LastActivityAt,
|
|
&i.TriageState,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const lockWakeupSourceTask = `-- name: LockWakeupSourceTask :one
|
|
SELECT id, agent_id, issue_id, status, priority, dispatched_at, started_at, completed_at, result, error, created_at, context, runtime_id, session_id, work_dir, trigger_comment_id, chat_session_id, autopilot_run_id, attempt, max_attempts, parent_task_id, failure_reason, trigger_summary, force_fresh_session, is_leader_task, wait_reason, initiator_user_id, handoff_note, prepare_lease_expires_at, squad_id, runtime_mcp_overlay, escalation_for_task_id, fire_at, originator_user_id, runtime_connected_apps, coalesced_comment_ids, delivered_comment_ids, chat_input_task_id, chat_finalize_deferred_at, originator_source, delegated_from_task_id, retry_of_task_id, rerun_of_task_id, rule_version_id, trigger_evidence_kind, trigger_evidence_ref_id, accountable_user_id, session_rollout_missing, retired_session_id, quick_actions_disabled, regenerate_quick_actions_for, branch_name, durable_work_dir, channel_context_revision, comment_thread_id, cancelled_by_type, cancelled_by_id, cancelled_by_name, issue_snapshot FROM agent_task_queue WHERE id= $1 AND issue_id= $2 FOR UPDATE NOWAIT
|
|
`
|
|
|
|
type LockWakeupSourceTaskParams struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
IssueID pgtype.UUID `json:"issue_id"`
|
|
}
|
|
|
|
func (q *Queries) LockWakeupSourceTask(ctx context.Context, arg LockWakeupSourceTaskParams) (AgentTaskQueue, error) {
|
|
row := q.db.QueryRow(ctx, lockWakeupSourceTask, arg.ID, arg.IssueID)
|
|
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
|
|
}
|
|
|
|
const locklessWakeup = `-- name: LocklessWakeup :one
|
|
SELECT id, workspace_id, issue_id, agent_id, created_by, source_task_id, parent_comment_id, instruction, kind, mode, event_types, filter_agent_id, filter_task_id, interval_seconds, cron_expression, timezone, next_fire_at, enabled, disabled_at, revision, last_task_id, last_error, created_at, updated_at, filter_actor_type, filter_actor_id FROM issue_wakeup WHERE id= $1
|
|
`
|
|
|
|
func (q *Queries) LocklessWakeup(ctx context.Context, id pgtype.UUID) (IssueWakeup, error) {
|
|
row := q.db.QueryRow(ctx, locklessWakeup, id)
|
|
var i IssueWakeup
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.IssueID,
|
|
&i.AgentID,
|
|
&i.CreatedBy,
|
|
&i.SourceTaskID,
|
|
&i.ParentCommentID,
|
|
&i.Instruction,
|
|
&i.Kind,
|
|
&i.Mode,
|
|
&i.EventTypes,
|
|
&i.FilterAgentID,
|
|
&i.FilterTaskID,
|
|
&i.IntervalSeconds,
|
|
&i.CronExpression,
|
|
&i.Timezone,
|
|
&i.NextFireAt,
|
|
&i.Enabled,
|
|
&i.DisabledAt,
|
|
&i.Revision,
|
|
&i.LastTaskID,
|
|
&i.LastError,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
&i.FilterActorType,
|
|
&i.FilterActorID,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const noteWakeupFailure = `-- name: NoteWakeupFailure :exec
|
|
UPDATE issue_wakeup SET last_error=$1,updated_at=clock_timestamp() WHERE id= $2
|
|
`
|
|
|
|
type NoteWakeupFailureParams struct {
|
|
LastError pgtype.Text `json:"last_error"`
|
|
ID pgtype.UUID `json:"id"`
|
|
}
|
|
|
|
func (q *Queries) NoteWakeupFailure(ctx context.Context, arg NoteWakeupFailureParams) error {
|
|
_, err := q.db.Exec(ctx, noteWakeupFailure, arg.LastError, arg.ID)
|
|
return err
|
|
}
|
|
|
|
const recordWakeupReceipt = `-- name: RecordWakeupReceipt :one
|
|
INSERT INTO issue_wakeup_receipt(id,wakeup_id,revision,event_key,event_type,payload)
|
|
VALUES($1,$2,$3,$4,$5,$6)
|
|
ON CONFLICT(wakeup_id,revision,event_key) DO UPDATE SET event_key=EXCLUDED.event_key RETURNING id, wakeup_id, revision, event_key, event_type, payload, task_id, processed_at, created_at, coalesce_key
|
|
`
|
|
|
|
type RecordWakeupReceiptParams struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
WakeupID pgtype.UUID `json:"wakeup_id"`
|
|
Revision int64 `json:"revision"`
|
|
EventKey string `json:"event_key"`
|
|
EventType string `json:"event_type"`
|
|
Payload []byte `json:"payload"`
|
|
}
|
|
|
|
func (q *Queries) RecordWakeupReceipt(ctx context.Context, arg RecordWakeupReceiptParams) (IssueWakeupReceipt, error) {
|
|
row := q.db.QueryRow(ctx, recordWakeupReceipt,
|
|
arg.ID,
|
|
arg.WakeupID,
|
|
arg.Revision,
|
|
arg.EventKey,
|
|
arg.EventType,
|
|
arg.Payload,
|
|
)
|
|
var i IssueWakeupReceipt
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.WakeupID,
|
|
&i.Revision,
|
|
&i.EventKey,
|
|
&i.EventType,
|
|
&i.Payload,
|
|
&i.TaskID,
|
|
&i.ProcessedAt,
|
|
&i.CreatedAt,
|
|
&i.CoalesceKey,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const replaceWakeupEvidence = `-- name: ReplaceWakeupEvidence :one
|
|
UPDATE agent_task_queue SET handoff_note=$1, context=COALESCE(context,'{}'::jsonb) || jsonb_build_object('wakeup_evidence', $2::jsonb) WHERE id= $3 AND status='queued' RETURNING id, agent_id, issue_id, status, priority, dispatched_at, started_at, completed_at, result, error, created_at, context, runtime_id, session_id, work_dir, trigger_comment_id, chat_session_id, autopilot_run_id, attempt, max_attempts, parent_task_id, failure_reason, trigger_summary, force_fresh_session, is_leader_task, wait_reason, initiator_user_id, handoff_note, prepare_lease_expires_at, squad_id, runtime_mcp_overlay, escalation_for_task_id, fire_at, originator_user_id, runtime_connected_apps, coalesced_comment_ids, delivered_comment_ids, chat_input_task_id, chat_finalize_deferred_at, originator_source, delegated_from_task_id, retry_of_task_id, rerun_of_task_id, rule_version_id, trigger_evidence_kind, trigger_evidence_ref_id, accountable_user_id, session_rollout_missing, retired_session_id, quick_actions_disabled, regenerate_quick_actions_for, branch_name, durable_work_dir, channel_context_revision, comment_thread_id, cancelled_by_type, cancelled_by_id, cancelled_by_name, issue_snapshot
|
|
`
|
|
|
|
type ReplaceWakeupEvidenceParams struct {
|
|
HandoffNote pgtype.Text `json:"handoff_note"`
|
|
WakeupEvidence []byte `json:"wakeup_evidence"`
|
|
ID pgtype.UUID `json:"id"`
|
|
}
|
|
|
|
func (q *Queries) ReplaceWakeupEvidence(ctx context.Context, arg ReplaceWakeupEvidenceParams) (AgentTaskQueue, error) {
|
|
row := q.db.QueryRow(ctx, replaceWakeupEvidence, arg.HandoffNote, arg.WakeupEvidence, arg.ID)
|
|
var i AgentTaskQueue
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.AgentID,
|
|
&i.IssueID,
|
|
&i.Status,
|
|
&i.Priority,
|
|
&i.DispatchedAt,
|
|
&i.StartedAt,
|
|
&i.CompletedAt,
|
|
&i.Result,
|
|
&i.Error,
|
|
&i.CreatedAt,
|
|
&i.Context,
|
|
&i.RuntimeID,
|
|
&i.SessionID,
|
|
&i.WorkDir,
|
|
&i.TriggerCommentID,
|
|
&i.ChatSessionID,
|
|
&i.AutopilotRunID,
|
|
&i.Attempt,
|
|
&i.MaxAttempts,
|
|
&i.ParentTaskID,
|
|
&i.FailureReason,
|
|
&i.TriggerSummary,
|
|
&i.ForceFreshSession,
|
|
&i.IsLeaderTask,
|
|
&i.WaitReason,
|
|
&i.InitiatorUserID,
|
|
&i.HandoffNote,
|
|
&i.PrepareLeaseExpiresAt,
|
|
&i.SquadID,
|
|
&i.RuntimeMcpOverlay,
|
|
&i.EscalationForTaskID,
|
|
&i.FireAt,
|
|
&i.OriginatorUserID,
|
|
&i.RuntimeConnectedApps,
|
|
&i.CoalescedCommentIds,
|
|
&i.DeliveredCommentIds,
|
|
&i.ChatInputTaskID,
|
|
&i.ChatFinalizeDeferredAt,
|
|
&i.OriginatorSource,
|
|
&i.DelegatedFromTaskID,
|
|
&i.RetryOfTaskID,
|
|
&i.RerunOfTaskID,
|
|
&i.RuleVersionID,
|
|
&i.TriggerEvidenceKind,
|
|
&i.TriggerEvidenceRefID,
|
|
&i.AccountableUserID,
|
|
&i.SessionRolloutMissing,
|
|
&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
|
|
}
|
|
|
|
const touchWakeupDispatch = `-- name: TouchWakeupDispatch :exec
|
|
UPDATE issue_wakeup SET updated_at=clock_timestamp() WHERE id= $1
|
|
`
|
|
|
|
// Move blocked configurations to the back of the scan so one noisy issue
|
|
// cannot monopolize the bounded batch.
|
|
func (q *Queries) TouchWakeupDispatch(ctx context.Context, id pgtype.UUID) error {
|
|
_, err := q.db.Exec(ctx, touchWakeupDispatch, id)
|
|
return err
|
|
}
|