klinsmaya/multica
1
0
Fork 0
mirror of https://gh-proxy.org/https://github.com/klinsmaya/multica.git synced 2026-11-08 01:16:45 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
zai-beta
multica/server/pkg/db/generated/wakeup.sql.go
T
klinsmayaandClaude Fable 5.1 8e04e8ba25 feat(server): open-connector connection service and task token lifecycle 
Add the open-connector business integration: user-owned OAuth and
API-key connections mirrored in user_connector_connection, a DB-backed
OAuth state table so callbacks survive restarts and replicas, and a
per-task runtime token minted at enqueue with allowedConnections pinned
to the agent owner's connections in the agent's service allowlist. The
token id is stored in agent_task_queue.runtime_connector_handles and
released when the task reaches a terminal state; a reaper entry point
releases stale handles and expired OAuth states.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
Claude-Session: https://claude.ai/code/session_01XqqpKySn2iWtXriK3Gsgak
2026-10-20 17:32:05 +00:00

2024 lines
72 KiB
Go
Raw Permalink Blame History

// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: wakeup.sql
package db
import (
"context"
"encoding/json"
"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, workflow_step, runtime_connector_handles
`
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,
&i.WorkflowStep,
&i.RuntimeConnectorHandles,
); 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, workflow_step, runtime_connector_handles
`
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,
&i.WorkflowStep,
&i.RuntimeConnectorHandles,
); 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 countActiveIssueRunsOfAgent = `-- name: CountActiveIssueRunsOfAgent :one
SELECT count(*) FROM agent_task_queue WHERE id=ANY($1::uuid[]) AND agent_id= $2 AND issue_id= $3
AND status IN ('dispatched','running','waiting_local_directory')
`
type CountActiveIssueRunsOfAgentParams struct {
Ids []pgtype.UUID `json:"ids"`
AgentID pgtype.UUID `json:"agent_id"`
IssueID pgtype.UUID `json:"issue_id"`
}
// How many of these runs are the agent's own runs on this issue that have
// not finished: the run that made a change still knows about it.
func (q *Queries) CountActiveIssueRunsOfAgent(ctx context.Context, arg CountActiveIssueRunsOfAgentParams) (int64, error) {
row := q.db.QueryRow(ctx, countActiveIssueRunsOfAgent, arg.Ids, arg.AgentID, arg.IssueID)
var count int64
err := row.Scan(&count)
return count, err
}
const countRecentWakeupTasks = `-- name: CountRecentWakeupTasks :one
SELECT count(*) FROM agent_task_queue WHERE context->>'wakeup_id'= $1::text AND issue_id= $2 AND created_at> $3
`
type CountRecentWakeupTasksParams struct {
WakeupID string `json:"wakeup_id"`
IssueID pgtype.UUID `json:"issue_id"`
Since pgtype.Timestamptz `json:"since"`
}
func (q *Queries) CountRecentWakeupTasks(ctx context.Context, arg CountRecentWakeupTasksParams) (int64, error) {
row := q.db.QueryRow(ctx, countRecentWakeupTasks, arg.WakeupID, arg.IssueID, arg.Since)
var count int64
err := row.Scan(&count)
return count, err
}
const countWakeupFires = `-- name: CountWakeupFires :exec
UPDATE issue_wakeup SET fire_count=fire_count+1 WHERE id= $1
`
func (q *Queries) CountWakeupFires(ctx context.Context, id pgtype.UUID) error {
_, err := q.db.Exec(ctx, countWakeupFires, id)
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,expires_at,expiry_seconds,on_timeout,condition,max_fires)
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24) 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, expires_at, expiry_seconds, on_timeout, timed_out_at, system_rule, customized_at, condition, condition_state, max_fires, fire_count, paused_reason
`
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"`
ExpiresAt pgtype.Timestamptz `json:"expires_at"`
ExpirySeconds pgtype.Int8 `json:"expiry_seconds"`
OnTimeout pgtype.Text `json:"on_timeout"`
Condition json.RawMessage `json:"condition"`
MaxFires pgtype.Int4 `json:"max_fires"`
}
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,
arg.ExpiresAt,
arg.ExpirySeconds,
arg.OnTimeout,
arg.Condition,
arg.MaxFires,
)
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,
&i.ExpiresAt,
&i.ExpirySeconds,
&i.OnTimeout,
&i.TimedOutAt,
&i.SystemRule,
&i.CustomizedAt,
&i.Condition,
&i.ConditionState,
&i.MaxFires,
&i.FireCount,
&i.PausedReason,
)
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, runtime_connector_handles,
originator_source, delegated_from_task_id, rule_version_id, rerun_of_task_id, trigger_evidence_kind, trigger_evidence_ref_id,
workflow_step, 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,
$24,
-- The workflow step this run works on (MUL-7420). A wakeup continues the
-- run that registered it, so it keeps that run's step: a status change it
-- makes after someone else moved the issue is refused like the run's own
-- would be. A wakeup a member registered starts from the issue's status.
COALESCE(
(SELECT s.workflow_step FROM agent_task_queue s WHERE s.id = $20 AND s.issue_id = $3),
(SELECT i.status FROM issue i WHERE i.id = $3)
),
COALESCE($25::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, workflow_step, runtime_connector_handles
`
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"`
RuntimeConnectorHandles []byte `json:"runtime_connector_handles"`
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.RuntimeConnectorHandles,
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,
&i.WorkflowStep,
&i.RuntimeConnectorHandles,
)
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 deleteIssueWakeup = `-- name: DeleteIssueWakeup :exec
DELETE FROM issue_wakeup WHERE id= $1 AND issue_id= $2
`
type DeleteIssueWakeupParams struct {
ID pgtype.UUID `json:"id"`
IssueID pgtype.UUID `json:"issue_id"`
}
func (q *Queries) DeleteIssueWakeup(ctx context.Context, arg DeleteIssueWakeupParams) error {
_, err := q.db.Exec(ctx, deleteIssueWakeup, arg.ID, arg.IssueID)
return err
}
const deleteIssueWakeupReceipts = `-- name: DeleteIssueWakeupReceipts :exec
DELETE FROM issue_wakeup_receipt WHERE wakeup_id= $1
`
func (q *Queries) DeleteIssueWakeupReceipts(ctx context.Context, id pgtype.UUID) error {
_, err := q.db.Exec(ctx, deleteIssueWakeupReceipts, id)
return err
}
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 AND system_rule IS NULL
`
// Closing wins over every subscription on the issue; reopening does not rearm.
// A system rule only rests while its issue is closed and resumes on reopen.
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, workflow_step, runtime_connector_handles 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,
&i.WorkflowStep,
&i.RuntimeConnectorHandles,
)
return i, err
}
const findWaitingIssueRun = `-- name: FindWaitingIssueRun :one
SELECT id FROM agent_task_queue WHERE issue_id= $1 AND agent_id= $2 AND status='queued'
AND originator_user_id= $3::uuid ORDER BY created_at,id LIMIT 1
`
type FindWaitingIssueRunParams struct {
IssueID pgtype.UUID `json:"issue_id"`
AgentID pgtype.UUID `json:"agent_id"`
OriginatorUserID pgtype.UUID `json:"originator_user_id"`
}
// A run of the agent on the issue that has not been claimed and runs as this
// person. A rule that fires meanwhile keeps its inputs for that run instead
// of queuing another.
func (q *Queries) FindWaitingIssueRun(ctx context.Context, arg FindWaitingIssueRunParams) (pgtype.UUID, error) {
row := q.db.QueryRow(ctx, findWaitingIssueRun, arg.IssueID, arg.AgentID, arg.OriginatorUserID)
var id pgtype.UUID
err := row.Scan(&id)
return id, 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, expires_at, expiry_seconds, on_timeout, timed_out_at, system_rule, customized_at, condition, condition_state, max_fires, fire_count, paused_reason 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,
&i.ExpiresAt,
&i.ExpirySeconds,
&i.OnTimeout,
&i.TimedOutAt,
&i.SystemRule,
&i.CustomizedAt,
&i.Condition,
&i.ConditionState,
&i.MaxFires,
&i.FireCount,
&i.PausedReason,
)
return i, err
}
const listDueWakeups = `-- name: ListDueWakeups :many
WITH candidates AS (
SELECT id FROM issue_wakeup WHERE enabled AND kind<>'event' AND next_fire_at<=now()
UNION
SELECT id FROM issue_wakeup WHERE enabled AND expires_at<=now()
UNION
SELECT id FROM issue_wakeup WHERE enabled AND condition IS NOT NULL AND next_fire_at<=now()
)
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, w.expires_at, w.expiry_seconds, w.on_timeout, w.timed_out_at, w.system_rule, w.customized_at, w.condition, w.condition_state, w.max_fires, w.fire_count, w.paused_reason FROM candidates c JOIN issue_wakeup w ON w.id=c.id
WHERE ($1::uuid[] IS NULL OR w.workspace_id = ANY($1::uuid[]))
AND NOT (w.id = ANY($2::uuid[]))
ORDER BY w.updated_at,w.id LIMIT $3
`
type ListDueWakeupsParams struct {
WorkspaceIds []pgtype.UUID `json:"workspace_ids"`
SkipIds []pgtype.UUID `json:"skip_ids"`
RowLimit int32 `json:"row_limit"`
}
// Scheduler lane 2: schedules, deadlines and condition checks that are due.
// Rules already visited in this pass are skipped.
func (q *Queries) ListDueWakeups(ctx context.Context, arg ListDueWakeupsParams) ([]IssueWakeup, error) {
rows, err := q.db.Query(ctx, listDueWakeups, arg.WorkspaceIds, arg.SkipIds, arg.RowLimit)
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,
&i.ExpiresAt,
&i.ExpirySeconds,
&i.OnTimeout,
&i.TimedOutAt,
&i.SystemRule,
&i.CustomizedAt,
&i.Condition,
&i.ConditionState,
&i.MaxFires,
&i.FireCount,
&i.PausedReason,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listInputWakeups = `-- name: ListInputWakeups :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_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, w.expires_at, w.expiry_seconds, w.on_timeout, w.timed_out_at, w.system_rule, w.customized_at, w.condition, w.condition_state, w.max_fires, w.fire_count, w.paused_reason FROM issue_wakeup w
WHERE w.id IN (SELECT r.wakeup_id FROM issue_wakeup_receipt r WHERE r.processed_at IS NULL)
AND ($1::uuid[] IS NULL OR w.workspace_id = ANY($1::uuid[]))
AND NOT (w.id = ANY($2::uuid[]))
ORDER BY w.updated_at,w.id LIMIT $3
`
type ListInputWakeupsParams struct {
WorkspaceIds []pgtype.UUID `json:"workspace_ids"`
SkipIds []pgtype.UUID `json:"skip_ids"`
RowLimit int32 `json:"row_limit"`
}
// Scheduler lane 1: rules with unprocessed inputs (subscribed events,
// condition hints, "wake now"). Deliberately not filtered by enabled: a once
// rule that stopped still delivers the inputs it captured before stopping.
// The subquery reads only pending rows (issue_wakeup_receipt_pending_idx).
// Rules already visited in this pass are skipped.
func (q *Queries) ListInputWakeups(ctx context.Context, arg ListInputWakeupsParams) ([]IssueWakeup, error) {
rows, err := q.db.Query(ctx, listInputWakeups, arg.WorkspaceIds, arg.SkipIds, arg.RowLimit)
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,
&i.ExpiresAt,
&i.ExpirySeconds,
&i.OnTimeout,
&i.TimedOutAt,
&i.SystemRule,
&i.CustomizedAt,
&i.Condition,
&i.ConditionState,
&i.MaxFires,
&i.FireCount,
&i.PausedReason,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
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,
w.expires_at,w.expiry_seconds,w.on_timeout,w.timed_out_at,
w.condition,w.max_fires,w.fire_count,w.paused_reason,
(w.source_task_id IS NOT NULL)::bool AS created_by_agent,creator.name AS created_by_name,
(CASE WHEN creator_agent.id IS NOT NULL THEN creator_agent.id END)::uuid AS source_agent_id,creator_agent.name AS source_agent_name
FROM issue_wakeup w JOIN agent a ON a.id=w.agent_id AND a.workspace_id=w.workspace_id
LEFT JOIN "user" creator ON creator.id=w.created_by
LEFT JOIN agent_task_queue creator_task ON creator_task.id=w.source_task_id
LEFT JOIN agent creator_agent ON creator_agent.id=creator_task.agent_id AND creator_agent.workspace_id=w.workspace_id AND creator_agent.id=ANY($1::uuid[])
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"`
ExpiresAt pgtype.Timestamptz `json:"expires_at"`
ExpirySeconds pgtype.Int8 `json:"expiry_seconds"`
OnTimeout pgtype.Text `json:"on_timeout"`
TimedOutAt pgtype.Timestamptz `json:"timed_out_at"`
Condition json.RawMessage `json:"condition"`
MaxFires pgtype.Int4 `json:"max_fires"`
FireCount int32 `json:"fire_count"`
PausedReason pgtype.Text `json:"paused_reason"`
CreatedByAgent bool `json:"created_by_agent"`
CreatedByName pgtype.Text `json:"created_by_name"`
SourceAgentID pgtype.UUID `json:"source_agent_id"`
SourceAgentName pgtype.Text `json:"source_agent_name"`
}
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,
&i.ExpiresAt,
&i.ExpirySeconds,
&i.OnTimeout,
&i.TimedOutAt,
&i.Condition,
&i.MaxFires,
&i.FireCount,
&i.PausedReason,
&i.CreatedByAgent,
&i.CreatedByName,
&i.SourceAgentID,
&i.SourceAgentName,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listPausedWakeupIssues = `-- name: ListPausedWakeupIssues :many
SELECT w.issue_id,w.id,w.agent_id,w.paused_reason
FROM issue_wakeup w JOIN issue i ON i.id=w.issue_id AND i.workspace_id=w.workspace_id
WHERE w.workspace_id= $1 AND NOT w.enabled AND w.paused_reason IN ('loop','rate','max_fires')
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'))
ORDER BY w.updated_at DESC LIMIT 200
`
type ListPausedWakeupIssuesRow struct {
IssueID pgtype.UUID `json:"issue_id"`
ID pgtype.UUID `json:"id"`
AgentID pgtype.UUID `json:"agent_id"`
PausedReason pgtype.Text `json:"paused_reason"`
}
// Rules the platform paused on open issues, for board cues and the
// workspace banner. Access follows shared issue visibility.
func (q *Queries) ListPausedWakeupIssues(ctx context.Context, workspaceID pgtype.UUID) ([]ListPausedWakeupIssuesRow, error) {
rows, err := q.db.Query(ctx, listPausedWakeupIssues, workspaceID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ListPausedWakeupIssuesRow{}
for rows.Next() {
var i ListPausedWakeupIssuesRow
if err := rows.Scan(
&i.IssueID,
&i.ID,
&i.AgentID,
&i.PausedReason,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listPendingWakeupInputGroups = `-- name: ListPendingWakeupInputGroups :many
WITH groups AS (
SELECT w.id AS wakeup_id, w.issue_id, (w.system_rule IS NOT NULL)::bool AS system_rule,
(r.task_id IS NOT NULL AND EXISTS (SELECT 1 FROM agent_task_queue t WHERE t.id=r.task_id
AND t.started_at IS NULL AND t.status NOT IN ('running','completed','failed','cancelled')))::bool AS reserved,
EXISTS (SELECT 1 FROM agent_task_queue t WHERE t.context->>'wakeup_id'=w.id::text AND t.status='dispatched')::bool AS own_claimed,
(w.system_rule IS NULL AND EXISTS (SELECT 1 FROM agent_task_queue t WHERE t.issue_id=w.issue_id AND t.status='queued'
AND t.agent_id=w.agent_id AND t.originator_user_id=w.created_by))::bool AS user_waiting,
min(r.created_at)::timestamptz AS oldest, count(*)::int AS inputs
FROM issue_wakeup_receipt r JOIN issue_wakeup w ON w.id=r.wakeup_id
WHERE r.processed_at IS NULL
AND ($5::uuid[] IS NULL OR w.workspace_id = ANY($5::uuid[]))
GROUP BY 1,2,3,4,5,6
)
SELECT g.wakeup_id, g.issue_id, g.system_rule, g.reserved, g.own_claimed, g.user_waiting, g.oldest, g.inputs FROM groups g
WHERE $1::timestamptz IS NULL
OR (g.oldest, g.wakeup_id, g.reserved) > ($1::timestamptz, $2::uuid, $3::bool)
ORDER BY g.oldest, g.wakeup_id, g.reserved LIMIT $4
`
type ListPendingWakeupInputGroupsParams struct {
AfterOldest pgtype.Timestamptz `json:"after_oldest"`
AfterWakeupID pgtype.UUID `json:"after_wakeup_id"`
AfterReserved bool `json:"after_reserved"`
PageSize int32 `json:"page_size"`
WorkspaceIds []pgtype.UUID `json:"workspace_ids"`
}
type ListPendingWakeupInputGroupsRow struct {
WakeupID pgtype.UUID `json:"wakeup_id"`
IssueID pgtype.UUID `json:"issue_id"`
SystemRule bool `json:"system_rule"`
Reserved bool `json:"reserved"`
OwnClaimed bool `json:"own_claimed"`
UserWaiting bool `json:"user_waiting"`
Oldest pgtype.Timestamptz `json:"oldest"`
Inputs int32 `json:"inputs"`
}
// Scheduler observability over unprocessed inputs, grouped per rule and by
// whether a run still reserves them. Dispatch deliberately keeps some inputs
// pending while a run holds them; these flags mirror its checks:
//
// reserved: a claim reserved them for a run that has not started and has
// not ended (takenReceipts). A run that ended without starting
// releases them, and one that started has them consumed on the
// next dispatch, so both are backlog.
// own_claimed: the rule's own run is claimed (FindPendingWakeupTask).
// user_waiting: a queued run of the rule's agent, run as its creator, will
// take them along (hasWaitingRun). Platform rules wait for the
// issue's current target instead; the caller resolves that.
//
// Oldest groups first, one page at a time: the caller pages with the last
// group's (oldest, wakeup_id, reserved) until a short page, so no group is
// left unread however many there are.
func (q *Queries) ListPendingWakeupInputGroups(ctx context.Context, arg ListPendingWakeupInputGroupsParams) ([]ListPendingWakeupInputGroupsRow, error) {
rows, err := q.db.Query(ctx, listPendingWakeupInputGroups,
arg.AfterOldest,
arg.AfterWakeupID,
arg.AfterReserved,
arg.PageSize,
arg.WorkspaceIds,
)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ListPendingWakeupInputGroupsRow{}
for rows.Next() {
var i ListPendingWakeupInputGroupsRow
if err := rows.Scan(
&i.WakeupID,
&i.IssueID,
&i.SystemRule,
&i.Reserved,
&i.OwnClaimed,
&i.UserWaiting,
&i.Oldest,
&i.Inputs,
); 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 listWaitingWakeups = `-- name: ListWaitingWakeups :many
SELECT w.id FROM issue_wakeup w
WHERE w.issue_id= $1 AND (w.agent_id= $2 OR w.system_rule IS NOT NULL) AND w.disabled_at IS NULL
AND EXISTS(SELECT 1 FROM issue_wakeup_receipt r WHERE r.wakeup_id=w.id AND r.revision=w.revision AND r.processed_at IS NULL
AND (w.condition IS NULL OR r.event_type IN ('condition.met','wakeup.timeout','wakeup.manual')))
ORDER BY w.created_at,w.id
`
type ListWaitingWakeupsParams struct {
IssueID pgtype.UUID `json:"issue_id"`
AgentID pgtype.UUID `json:"agent_id"`
}
// Rules on the issue with inputs a run of this agent can take along: the
// agent's own rules and the platform's. Read without locks; each one is
// locked and checked again before it joins.
func (q *Queries) ListWaitingWakeups(ctx context.Context, arg ListWaitingWakeupsParams) ([]pgtype.UUID, error) {
rows, err := q.db.Query(ctx, listWaitingWakeups, arg.IssueID, arg.AgentID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []pgtype.UUID{}
for rows.Next() {
var id pgtype.UUID
if err := rows.Scan(&id); err != nil {
return nil, err
}
items = append(items, id)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listWakeupChains = `-- name: ListWakeupChains :many
SELECT id,COALESCE(context->>'wakeup_id','')::text AS wakeup_id,COALESCE(context->'wakeup_chain','[]'::jsonb)::jsonb AS chain
FROM agent_task_queue WHERE id=ANY($1::uuid[])
`
type ListWakeupChainsRow struct {
ID pgtype.UUID `json:"id"`
WakeupID string `json:"wakeup_id"`
Chain []byte `json:"chain"`
}
// Rules whose runs produced the source events, including rules further up
// each run's own trigger chain.
func (q *Queries) ListWakeupChains(ctx context.Context, ids []pgtype.UUID) ([]ListWakeupChainsRow, error) {
rows, err := q.db.Query(ctx, listWakeupChains, ids)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ListWakeupChainsRow{}
for rows.Next() {
var i ListWakeupChainsRow
if err := rows.Scan(&i.ID, &i.WakeupID, &i.Chain); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listWakeupRuns = `-- name: ListWakeupRuns :many
SELECT t.id,t.status,t.created_at,t.started_at,t.completed_at,
COALESCE(t.context->'wakeup_checkin'->>'note','')::text AS checkin_note,
COALESCE((SELECT array_agg(DISTINCT f->>'event_type') FROM jsonb_array_elements(CASE WHEN jsonb_typeof(t.context->'wakeup_evidence'->'facts')='array' THEN t.context->'wakeup_evidence'->'facts' ELSE '[]'::jsonb END) f),'{}')::text[] AS triggers,
EXISTS(SELECT 1 FROM comment c WHERE c.issue_id=t.issue_id AND c.source_task_id=t.id)::bool AS commented
FROM agent_task_queue t WHERE t.context->>'wakeup_id'= $1::text AND t.issue_id= $2
ORDER BY t.created_at DESC,t.id DESC LIMIT 10
`
type ListWakeupRunsParams struct {
WakeupID string `json:"wakeup_id"`
IssueID pgtype.UUID `json:"issue_id"`
}
type ListWakeupRunsRow struct {
ID pgtype.UUID `json:"id"`
Status string `json:"status"`
CreatedAt pgtype.Timestamptz `json:"created_at"`
StartedAt pgtype.Timestamptz `json:"started_at"`
CompletedAt pgtype.Timestamptz `json:"completed_at"`
CheckinNote string `json:"checkin_note"`
Triggers []string `json:"triggers"`
Commented bool `json:"commented"`
}
// The rule's latest runs with what triggered each one, a check-in note when
// the run ended silently, and whether it left a comment.
func (q *Queries) ListWakeupRuns(ctx context.Context, arg ListWakeupRunsParams) ([]ListWakeupRunsRow, error) {
rows, err := q.db.Query(ctx, listWakeupRuns, arg.WakeupID, arg.IssueID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ListWakeupRunsRow{}
for rows.Next() {
var i ListWakeupRunsRow
if err := rows.Scan(
&i.ID,
&i.Status,
&i.CreatedAt,
&i.StartedAt,
&i.CompletedAt,
&i.CheckinNote,
&i.Triggers,
&i.Commented,
); 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,w.condition,
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,condition,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"`
Condition []byte `json:"condition"`
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.Condition,
&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, expires_at, expiry_seconds, on_timeout, timed_out_at, system_rule, customized_at, condition, condition_state, max_fires, fire_count, paused_reason 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,
&i.ExpiresAt,
&i.ExpirySeconds,
&i.OnTimeout,
&i.TimedOutAt,
&i.SystemRule,
&i.CustomizedAt,
&i.Condition,
&i.ConditionState,
&i.MaxFires,
&i.FireCount,
&i.PausedReason,
)
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, duplicate_of_issue_id 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,
&i.DuplicateOfIssueID,
)
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, workflow_step, runtime_connector_handles 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,
&i.WorkflowStep,
&i.RuntimeConnectorHandles,
)
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, expires_at, expiry_seconds, on_timeout, timed_out_at, system_rule, customized_at, condition, condition_state, max_fires, fire_count, paused_reason 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,
&i.ExpiresAt,
&i.ExpirySeconds,
&i.OnTimeout,
&i.TimedOutAt,
&i.SystemRule,
&i.CustomizedAt,
&i.Condition,
&i.ConditionState,
&i.MaxFires,
&i.FireCount,
&i.PausedReason,
)
return i, err
}
const markWakeupTimedOut = `-- name: MarkWakeupTimedOut :exec
UPDATE issue_wakeup SET enabled=false,next_fire_at=NULL,timed_out_at=clock_timestamp(),updated_at=clock_timestamp() WHERE id= $1
`
// The deadline ended the rule. disabled_at stays NULL: a timeout run created in
// the same transaction must remain claimable, and the rule reads as timed out,
// not as turned off by a person.
func (q *Queries) MarkWakeupTimedOut(ctx context.Context, id pgtype.UUID) error {
_, err := q.db.Exec(ctx, markWakeupTimedOut, id)
return 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 pauseIssueWakeup = `-- name: PauseIssueWakeup :exec
UPDATE issue_wakeup SET enabled=false,next_fire_at=NULL,paused_reason= $1,
disabled_at=CASE WHEN $2::bool THEN COALESCE(disabled_at,clock_timestamp()) ELSE disabled_at END,
updated_at=clock_timestamp() WHERE id= $3
`
type PauseIssueWakeupParams struct {
PausedReason pgtype.Text `json:"paused_reason"`
BlockRuns bool `json:"block_runs"`
ID pgtype.UUID `json:"id"`
}
// The platform stopped a rule. A detected loop or burst also sets disabled_at
// so queued runs of this rule can no longer be claimed; hitting the cap does
// not, because the run that reached it is legitimate.
func (q *Queries) PauseIssueWakeup(ctx context.Context, arg PauseIssueWakeupParams) error {
_, err := q.db.Exec(ctx, pauseIssueWakeup, arg.PausedReason, arg.BlockRuns, 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 releaseWakeupReceipts = `-- name: ReleaseWakeupReceipts :exec
UPDATE issue_wakeup_receipt SET task_id=NULL WHERE id=ANY($1::uuid[]) AND processed_at IS NULL
`
func (q *Queries) ReleaseWakeupReceipts(ctx context.Context, ids []pgtype.UUID) error {
_, err := q.db.Exec(ctx, releaseWakeupReceipts, ids)
return 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, workflow_step, runtime_connector_handles
`
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,
&i.WorkflowStep,
&i.RuntimeConnectorHandles,
)
return i, err
}
const reserveWakeupReceipts = `-- name: ReserveWakeupReceipts :exec
UPDATE issue_wakeup_receipt SET task_id= $1 WHERE id=ANY($2::uuid[]) AND processed_at IS NULL
`
type ReserveWakeupReceiptsParams struct {
TaskID pgtype.UUID `json:"task_id"`
Ids []pgtype.UUID `json:"ids"`
}
// A run being claimed takes these inputs along. They stay pending: they count
// as handled once that run starts, and go back to the rule if it never does.
func (q *Queries) ReserveWakeupReceipts(ctx context.Context, arg ReserveWakeupReceiptsParams) error {
_, err := q.db.Exec(ctx, reserveWakeupReceipts, arg.TaskID, arg.Ids)
return err
}
const setClaimedTaskContext = `-- name: SetClaimedTaskContext :one
UPDATE agent_task_queue SET context= $1::jsonb
WHERE id= $2 AND status='dispatched' AND dispatched_at= $3 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, workflow_step, runtime_connector_handles
`
type SetClaimedTaskContextParams struct {
Context []byte `json:"context"`
ID pgtype.UUID `json:"id"`
DispatchedAt pgtype.Timestamptz `json:"dispatched_at"`
}
// Only the claim being answered may change its task's context.
func (q *Queries) SetClaimedTaskContext(ctx context.Context, arg SetClaimedTaskContextParams) (AgentTaskQueue, error) {
row := q.db.QueryRow(ctx, setClaimedTaskContext, arg.Context, arg.ID, arg.DispatchedAt)
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
}
const setWakeupConditionState = `-- name: SetWakeupConditionState :exec
UPDATE issue_wakeup SET condition_state= $1,next_fire_at=$2,updated_at=clock_timestamp() WHERE id= $3
`
type SetWakeupConditionStateParams struct {
ConditionState string `json:"condition_state"`
NextFireAt pgtype.Timestamptz `json:"next_fire_at"`
ID pgtype.UUID `json:"id"`
}
func (q *Queries) SetWakeupConditionState(ctx context.Context, arg SetWakeupConditionStateParams) error {
_, err := q.db.Exec(ctx, setWakeupConditionState, arg.ConditionState, arg.NextFireAt, arg.ID)
return 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
}
const tryLockIssueWakeup = `-- name: TryLockIssueWakeup :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, expires_at, expiry_seconds, on_timeout, timed_out_at, system_rule, customized_at, condition, condition_state, max_fires, fire_count, paused_reason FROM issue_wakeup WHERE id= $1 FOR UPDATE SKIP LOCKED
`
// A rule another writer holds is skipped; it keeps its inputs.
func (q *Queries) TryLockIssueWakeup(ctx context.Context, id pgtype.UUID) (IssueWakeup, error) {
row := q.db.QueryRow(ctx, tryLockIssueWakeup, 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,
&i.ExpiresAt,
&i.ExpirySeconds,
&i.OnTimeout,
&i.TimedOutAt,
&i.SystemRule,
&i.CustomizedAt,
&i.Condition,
&i.ConditionState,
&i.MaxFires,
&i.FireCount,
&i.PausedReason,
)
return i, err
}
Reference in New Issue View Git Blame Copy Permalink