mirror of
https://gh-proxy.org/https://github.com/klinsmaya/multica.git
synced 2026-11-09 09:56:47 +08:00
b7cdd76ad8
* feat(server): add the duplicate-mark column and keep it correct (MUL-7349) Expand step for marking an issue as a duplicate of another. A duplicate is an ordinary cancelled issue that records which issue it duplicates; there is no duplicate status. This release adds the column and the code that maintains it, and nothing writes a pointer yet. - A mark survives only a write that leaves an already cancelled issue cancelled. Every other write drops it, so reopening removes a mark and cancelling again does not bring one back. Both status-writing queries carry the rule, which covers the UI, batch updates and background writers. - Deleting an issue clears the pointers of its duplicates in the same transaction, the way deleting a parent detaches its children. Shipping this before marking is what makes marking safe to roll back: the release a rollback lands on already clears the pointer. Tests seed a mark the way the next release will write it. Co-authored-by: multica-agent <[email protected]> * fix(server): compute issue updates from the locked row (MUL-7349) UpdateIssue built its next_* values in a CTE that read the row before the UPDATE took its row lock. A write that waited there behind another transaction then resumed with the values it had read before the wait and wrote them back, so an ordinary priority or status edit could restore a duplicate mark the other transaction had just cleared — by reopening the issue or by deleting its original — and did_change/did_activity were decided against the pre-wait row as well. Lock the row in the CTE instead, so every computed value comes from the row the statement is about to write. Tests hold the clearing transaction open and wait for the update to block on the row lock, so the interleaving is deterministic; both fail without this change. Co-authored-by: multica-agent <[email protected]> --------- Co-authored-by: J <[email protected]> Co-authored-by: multica-agent <[email protected]>
1262 lines
44 KiB
Go
1262 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, 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 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
|
|
}
|