Files
multica/server/pkg/db/generated/runtime.sql.go
Multica Eve 75695a2e40 fix(comments): guarantee at-least-once processing of user comments (MUL-4195) (#5068)
* fix(comments): guarantee at-least-once processing of user comments (MUL-4195)

Consecutive comments on an issue were silently dropped: a new comment that
arrived while the agent already had a queued/dispatched task was discarded by
the HasPendingTaskForIssueAndAgent dedup, losing the user's follow-up
instruction with no visible trace. Comments — unlike chat — are deliberate,
addressed, persisted input and must never vanish.

This makes comment handling at-least-once while keeping concurrency bounded to
one run per (issue, agent):

- Merge, don't drop (PR1): a comment landing while a not-yet-started task
  exists is folded into that task — the prior trigger becomes a coalesced
  comment and the new one becomes the trigger, so a single run still covers
  every deliberate comment. Falls back to a fresh enqueue if the pending task
  was claimed mid-flight, so nothing is lost in the race.
- Completion reconciliation (PR2): on task completion, a member comment newer
  than the run's started_at schedules exactly one follow-up via the normal
  trigger pipeline. Loop-safe: member-authored only, capped by the existing
  per-(issue,agent) dedup, and terminating.
- Visibility (PR3): coalesced_comment_ids is surfaced on the task API and in
  the run prompt so the covered comments are explicit.

Migration 145 adds agent_task_queue.coalesced_comment_ids UUID[].

Tests: merge-not-drop preserves all three of a rapid burst and repoints the
trigger to the newest; reconciliation query gates on member/since; e2e
CompleteTask enqueues a follow-up for a mid-run member comment and does not for
none.

Co-authored-by: multica-agent <github@multica.ai>

* fix(comments): address review — originator gate, agent-scoped reconcile, cross-thread coalesced prompt (MUL-4195)

Resolves GPT-Boy's Request-changes review on PR #5068.

Must-fix #1 — merge no longer inherits a stale originator/runtime context.
MergeCommentIntoPendingTask now only folds a comment into a pending task
whose originator_user_id IS NOT DISTINCT FROM the new comment's originator.
runtime_mcp_overlay / runtime_connected_apps are a pure function of
(originator, agent) and the agent is fixed, so a matching originator keeps
the stored overlay/attribution valid; a differing originator (e.g. user B
commenting on a task originated by user A) matches no row and the caller
enqueues a fresh follow-up with B's own context instead of reusing A's.
trigger_summary is refreshed to the new trigger comment.

Must-fix #2 — completion reconcile no longer re-wakes unrelated agents.
reconcileCommentsOnCompletion computes the latest member comment's triggers
and keeps ONLY the agent that just completed, instead of fanning the comment
out through the full pipeline. An @-mention of agent B during agent A's run
is triggered once at creation time and is no longer replayed (double-run)
when A completes.

Should-fix #3 — coalesced-comment prompt no longer assumes a single thread.
The claim response now carries each folded comment's thread id / author /
created_at / content (CoalescedCommentData); the prompt embeds them directly
so the agent addresses cross-thread folded comments without the wrong
"they are in the triggering thread" hint. Old servers that ship only ids
fall back to an issue-wide fetch, still without the same-thread assumption.

Tests: TestMergeCommentIntoPendingTask_OriginatorGate (query gate),
TestCompleteTask_DoesNotReTriggerOtherAgentMentionedDuringRun (reconcile
scoping), TestBuildCommentPromptCoalescedCrossThread / IDsOnlyFallback
(prompt). Existing MUL-4195 suites still pass.

Co-authored-by: multica-agent <github@multica.ai>

* fix(comments): close unique-index drop + dispatched-window race in comment coalescing (MUL-4195)

Second-round review follow-up on PR #5068.

Must-fix #1 — originator-mismatch no longer drops the comment.
The previous originator gate returned ErrNoRows on a different originator and
the caller fell through to a fresh enqueue, which collided with the
idx_one_pending_task_per_issue_agent unique index (one queued/dispatched task
per (issue, agent)) — silently dropping the second user's comment. Replaced
the gate with recompute-on-merge: MergeCommentIntoPendingTask now re-stamps
originator_user_id, runtime_mcp_overlay, runtime_connected_apps and
trigger_summary to the new comment's originator. A different member's comment
folds into the single coalescing run carrying the latest instruction's own
identity/overlay (no cross-user capability bleed, no drop, no collision).

Must-fix #2 — comment arriving in the claim→StartTask window is no longer lost.
Merge now targets only PRE-CLAIM states ('queued','deferred'); a
dispatched/running task is never a merge target, so a post-claim comment is
never falsely stamped into coalesced_comment_ids as "delivered". Completion
reconcile is re-anchored on dispatched_at (the moment the claim response is
built) instead of started_at, and sweeps ALL undelivered member comments since
that anchor — replaying each through the normal enqueue path so they coalesce
into one bounded, agent-scoped follow-up run. This covers the dispatch→start
window a started_at anchor missed.

Enqueue path: on a merge miss the caller no longer blindly fresh-enqueues
(which could collide with a dispatched sibling); it defers to the active
task's completion reconcile via HasActiveTaskForIssueAndAgent, and only
fresh-enqueues when no active task exists.

Tests: rewrote the query test to
TestMergeCommentIntoPendingTask_RecomputesOriginatorAndSkipsDispatched;
added TestConsecutiveCommentsDifferentOriginatorsFullEnqueuePath (full handler
enqueue path, two distinct originators) and
TestCompleteTask_ReconcilesDispatchedWindowComment (claim→start window). All
existing MUL-4195 handler/cmd-server/daemon/service suites still pass.

Co-authored-by: multica-agent <github@multica.ai>

* fix(comments): catch pre-dispatch merge-race comment in completion reconcile (MUL-4195)

Third-round review follow-up on PR #5068.

Race: a member comment is created while the task is still queued, but its
merge loses the race to the daemon claiming the task (queued→dispatched). The
merge then finds no pre-claim row (ErrNoRows), the enqueue path defers to
reconcile — but the comment's created_at is BEFORE dispatched_at, so the
dispatched_at-anchored reconcile skipped it and the comment vanished with no
task coverage.

Fix: anchor completion reconcile on the task's created_at (which always
precedes dispatch) instead of a dispatch/start timestamp, and exclude the
run's DELIVERED SET — trigger_comment_id ∪ coalesced_comment_ids. Because
merges only ever touch pre-claim rows, that set is exactly what the claim
response carried, so any member comment created since the task was made that
is NOT in it was genuinely undelivered and earns a bounded follow-up. This
catches the pre-dispatch merge-race comment and the dispatch→start comment,
while never re-firing a comment that was delivered as a pre-claim coalesced
entry.

Test: TestCompleteTask_ReconcilesPreDispatchMergeRaceComment reproduces the
race (comment created pre-dispatch, task dispatched before merge, plus a
delivered coalesced comment) and asserts exactly one follow-up, triggered by
the race comment, with the delivered coalesced comment excluded. Existing
reconcile fixtures updated to set a realistic created_at (the production
invariant that created_at is the earliest task timestamp).

Co-authored-by: multica-agent <github@multica.ai>

* fix(comments): merge only into the queued task, never a deferred fallback (MUL-4195)

Fourth-round review follow-up on PR #5068.

MergeCommentIntoPendingTask targeted status IN ('queued','deferred') ordered
by created_at DESC. When a (issue, agent) pair had both an older queued task
(the run about to be claimed) and a newer deferred assignee-fallback task, a
new comment merged into the deferred row instead of the queued one — so the
comment missed the imminent run and the deferred fallback could later promote
into a duplicate/conflicting run.

This merge is only ever reached when HasPendingTaskForIssueAndAgent matched a
queued/dispatched task (it never inspects deferred), so the coalescing target
must be the queued row. Restricted the merge target to status = 'queued'
(the unique index guarantees at most one). Deferred fallbacks keep their own
fire_at/promotion escalation lifecycle and are never a merge target.

Test: TestMergeCommentIntoPendingTask_TargetsQueuedNotDeferred seeds an older
queued task + a newer deferred fallback for the same (issue, agent), merges a
new comment, and asserts it lands on the queued task (trigger repointed, old
trigger coalesced) while the deferred fallback is left untouched.

Co-authored-by: multica-agent <github@multica.ai>

---------

Co-authored-by: Eve <eve@multica-ai.local>
Co-authored-by: multica-agent <github@multica.ai>
2026-07-09 12:48:57 +08:00

1286 lines
42 KiB
Go

// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: runtime.sql
package db
import (
"context"
"github.com/jackc/pgx/v5/pgtype"
)
const cancelAgentTasksByRuntimeOrAgent = `-- name: CancelAgentTasksByRuntimeOrAgent :many
UPDATE agent_task_queue
SET status = 'cancelled', completed_at = now()
WHERE (runtime_id = ANY($1::uuid[]) OR agent_id = ANY($2::uuid[]))
AND status IN ('queued', 'dispatched', 'running', 'waiting_local_directory')
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
`
type CancelAgentTasksByRuntimeOrAgentParams struct {
RuntimeIds []pgtype.UUID `json:"runtime_ids"`
AgentIds []pgtype.UUID `json:"agent_ids"`
}
// Cancels every active task that either lives on one of the given runtimes
// OR belongs to one of the given agents. Used by the member-revocation flow:
// the runtime-side covers tasks queued against the leaving member's runtimes;
// the agent-side covers tasks pinned to a different runtime that those agents
// left behind from a prior UpdateAgent (agent.runtime_id can change, but
// agent_task_queue.runtime_id does not get rewritten when it does, so a task
// queued on runtime A by agent X — later moved to runtime B — survives the
// runtime-only revoke and could still be claimed because ClaimAgentTask does
// not gate on agent.archived_at).
//
// We use 'cancelled' rather than 'failed' so the daemon's per-task status
// poller (watchTaskCancellation) interrupts the running agent gracefully.
// Returns the affected rows so the caller can broadcast task:cancelled and
// reconcile per-agent status.
func (q *Queries) CancelAgentTasksByRuntimeOrAgent(ctx context.Context, arg CancelAgentTasksByRuntimeOrAgentParams) ([]AgentTaskQueue, error) {
rows, err := q.db.Query(ctx, cancelAgentTasksByRuntimeOrAgent, arg.RuntimeIds, arg.AgentIds)
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,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const countActiveAgentsByRuntime = `-- name: CountActiveAgentsByRuntime :one
SELECT count(*) FROM agent WHERE runtime_id = $1 AND archived_at IS NULL
`
func (q *Queries) CountActiveAgentsByRuntime(ctx context.Context, runtimeID pgtype.UUID) (int64, error) {
row := q.db.QueryRow(ctx, countActiveAgentsByRuntime, runtimeID)
var count int64
err := row.Scan(&count)
return count, err
}
const countActiveSquadsWithArchivedLeadersByRuntime = `-- name: CountActiveSquadsWithArchivedLeadersByRuntime :one
SELECT count(*)
FROM squad
WHERE archived_at IS NULL
AND leader_id IN (
SELECT id FROM agent WHERE runtime_id = $1 AND archived_at IS NOT NULL
)
`
func (q *Queries) CountActiveSquadsWithArchivedLeadersByRuntime(ctx context.Context, runtimeID pgtype.UUID) (int64, error) {
row := q.db.QueryRow(ctx, countActiveSquadsWithArchivedLeadersByRuntime, runtimeID)
var count int64
err := row.Scan(&count)
return count, err
}
const deleteAgentRuntime = `-- name: DeleteAgentRuntime :exec
DELETE FROM agent_runtime WHERE id = $1
`
func (q *Queries) DeleteAgentRuntime(ctx context.Context, id pgtype.UUID) error {
_, err := q.db.Exec(ctx, deleteAgentRuntime, id)
return err
}
const deleteArchivedAgentsByRuntime = `-- name: DeleteArchivedAgentsByRuntime :exec
DELETE FROM agent WHERE runtime_id = $1 AND archived_at IS NOT NULL
`
func (q *Queries) DeleteArchivedAgentsByRuntime(ctx context.Context, runtimeID pgtype.UUID) error {
_, err := q.db.Exec(ctx, deleteArchivedAgentsByRuntime, runtimeID)
return err
}
const deleteSquadsByArchivedAgentsOnRuntime = `-- name: DeleteSquadsByArchivedAgentsOnRuntime :exec
DELETE FROM squad
WHERE leader_id IN (
SELECT id FROM agent WHERE runtime_id = $1 AND archived_at IS NOT NULL
)
AND archived_at IS NOT NULL
`
// Removes archived squads whose leader_id references an archived agent on the
// given runtime. Must run before DeleteArchivedAgentsByRuntime so the RESTRICT
// FK on squad.leader_id does not block the agent deletion. Active squads are
// handled separately by CountActiveSquadsWithArchivedLeadersByRuntime, which
// returns a 409 until the caller archives them or assigns a new leader.
func (q *Queries) DeleteSquadsByArchivedAgentsOnRuntime(ctx context.Context, runtimeID pgtype.UUID) error {
_, err := q.db.Exec(ctx, deleteSquadsByArchivedAgentsOnRuntime, runtimeID)
return err
}
const deleteStaleOfflineRuntimes = `-- name: DeleteStaleOfflineRuntimes :many
DELETE FROM agent_runtime
WHERE status = 'offline'
AND last_seen_at < now() - make_interval(secs => $1::double precision)
AND id NOT IN (SELECT DISTINCT runtime_id FROM agent)
RETURNING id, workspace_id
`
type DeleteStaleOfflineRuntimesRow struct {
ID pgtype.UUID `json:"id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
}
// Deletes runtimes that have been offline for longer than the TTL and have
// no agents bound (active or archived). The FK constraint on agent.runtime_id
// is ON DELETE RESTRICT, so we must exclude all agent references.
func (q *Queries) DeleteStaleOfflineRuntimes(ctx context.Context, staleSeconds float64) ([]DeleteStaleOfflineRuntimesRow, error) {
rows, err := q.db.Query(ctx, deleteStaleOfflineRuntimes, staleSeconds)
if err != nil {
return nil, err
}
defer rows.Close()
items := []DeleteStaleOfflineRuntimesRow{}
for rows.Next() {
var i DeleteStaleOfflineRuntimesRow
if err := rows.Scan(&i.ID, &i.WorkspaceID); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const failTasksForOfflineRuntimes = `-- name: FailTasksForOfflineRuntimes :many
UPDATE agent_task_queue
SET status = 'failed', completed_at = now(), error = 'runtime went offline',
failure_reason = 'runtime_offline',
wait_reason = NULL
WHERE status IN ('dispatched', 'running', 'waiting_local_directory')
AND runtime_id IN (
SELECT id FROM agent_runtime WHERE status = 'offline'
)
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
`
// Marks dispatched/running/waiting_local_directory tasks as failed when
// their runtime is offline. This cleans up orphaned tasks after a daemon
// crash or network partition.
func (q *Queries) FailTasksForOfflineRuntimes(ctx context.Context) ([]AgentTaskQueue, error) {
rows, err := q.db.Query(ctx, failTasksForOfflineRuntimes)
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,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const findLegacyRuntimesByDaemonID = `-- name: FindLegacyRuntimesByDaemonID :many
SELECT id, workspace_id, daemon_id, name, runtime_mode, provider, status, device_info, metadata, last_seen_at, created_at, updated_at, owner_id, legacy_daemon_id, visibility, profile_id, custom_name FROM agent_runtime
WHERE workspace_id = $1
AND provider = $2
AND LOWER(daemon_id) = LOWER($3)
`
type FindLegacyRuntimesByDaemonIDParams struct {
WorkspaceID pgtype.UUID `json:"workspace_id"`
Provider string `json:"provider"`
DaemonID string `json:"daemon_id"`
}
// Looks up runtime rows keyed on a prior (hostname-derived) daemon_id. Used
// at register-time to find rows owned by the same machine under its old
// identity so agents/tasks can be re-pointed at the new UUID-keyed row.
//
// Comparison is case-insensitive because os.Hostname() has been observed to
// return different casings on the same machine (e.g. `Jiayuans-MacBook-Pro`
// vs `jiayuans-macbook-pro`) across reboots/mDNS state changes. A case-
// sensitive `=` would strand the old row; LOWER() on both sides handles drift
// without forcing the daemon to enumerate cased permutations.
//
// Returns many rather than one because case drift may have already minted
// duplicate rows historically (e.g. `Foo.local` AND `foo.local` under the
// same workspace+provider). A single-row lookup would consolidate only one
// of them and leave the rest orphaned. Callers must merge every returned
// row into the new UUID-keyed runtime.
func (q *Queries) FindLegacyRuntimesByDaemonID(ctx context.Context, arg FindLegacyRuntimesByDaemonIDParams) ([]AgentRuntime, error) {
rows, err := q.db.Query(ctx, findLegacyRuntimesByDaemonID, arg.WorkspaceID, arg.Provider, arg.DaemonID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []AgentRuntime{}
for rows.Next() {
var i AgentRuntime
if err := rows.Scan(
&i.ID,
&i.WorkspaceID,
&i.DaemonID,
&i.Name,
&i.RuntimeMode,
&i.Provider,
&i.Status,
&i.DeviceInfo,
&i.Metadata,
&i.LastSeenAt,
&i.CreatedAt,
&i.UpdatedAt,
&i.OwnerID,
&i.LegacyDaemonID,
&i.Visibility,
&i.ProfileID,
&i.CustomName,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const forceOfflineRuntimesByIDs = `-- name: ForceOfflineRuntimesByIDs :many
UPDATE agent_runtime
SET status = 'offline', updated_at = now()
WHERE id = ANY($1::uuid[]) AND status = 'online'
RETURNING id, workspace_id, owner_id, daemon_id, provider
`
type ForceOfflineRuntimesByIDsRow struct {
ID pgtype.UUID `json:"id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
OwnerID pgtype.UUID `json:"owner_id"`
DaemonID pgtype.Text `json:"daemon_id"`
Provider string `json:"provider"`
}
// Unconditionally flips a known set of runtime IDs to offline. Distinct from
// MarkRuntimesOfflineByIDs (which keeps a stale-window predicate so the
// sweeper cannot demote a runtime that just heartbeated): this variant is
// used by intentional revocation paths — e.g. removing a workspace member —
// where the caller has already decided the runtime should be offline
// regardless of recent liveness.
func (q *Queries) ForceOfflineRuntimesByIDs(ctx context.Context, runtimeIds []pgtype.UUID) ([]ForceOfflineRuntimesByIDsRow, error) {
rows, err := q.db.Query(ctx, forceOfflineRuntimesByIDs, runtimeIds)
if err != nil {
return nil, err
}
defer rows.Close()
items := []ForceOfflineRuntimesByIDsRow{}
for rows.Next() {
var i ForceOfflineRuntimesByIDsRow
if err := rows.Scan(
&i.ID,
&i.WorkspaceID,
&i.OwnerID,
&i.DaemonID,
&i.Provider,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const getAgentRuntime = `-- name: GetAgentRuntime :one
SELECT id, workspace_id, daemon_id, name, runtime_mode, provider, status, device_info, metadata, last_seen_at, created_at, updated_at, owner_id, legacy_daemon_id, visibility, profile_id, custom_name FROM agent_runtime
WHERE id = $1
`
func (q *Queries) GetAgentRuntime(ctx context.Context, id pgtype.UUID) (AgentRuntime, error) {
row := q.db.QueryRow(ctx, getAgentRuntime, id)
var i AgentRuntime
err := row.Scan(
&i.ID,
&i.WorkspaceID,
&i.DaemonID,
&i.Name,
&i.RuntimeMode,
&i.Provider,
&i.Status,
&i.DeviceInfo,
&i.Metadata,
&i.LastSeenAt,
&i.CreatedAt,
&i.UpdatedAt,
&i.OwnerID,
&i.LegacyDaemonID,
&i.Visibility,
&i.ProfileID,
&i.CustomName,
)
return i, err
}
const getAgentRuntimeForWorkspace = `-- name: GetAgentRuntimeForWorkspace :one
SELECT id, workspace_id, daemon_id, name, runtime_mode, provider, status, device_info, metadata, last_seen_at, created_at, updated_at, owner_id, legacy_daemon_id, visibility, profile_id, custom_name FROM agent_runtime
WHERE id = $1 AND workspace_id = $2
`
type GetAgentRuntimeForWorkspaceParams struct {
ID pgtype.UUID `json:"id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
}
func (q *Queries) GetAgentRuntimeForWorkspace(ctx context.Context, arg GetAgentRuntimeForWorkspaceParams) (AgentRuntime, error) {
row := q.db.QueryRow(ctx, getAgentRuntimeForWorkspace, arg.ID, arg.WorkspaceID)
var i AgentRuntime
err := row.Scan(
&i.ID,
&i.WorkspaceID,
&i.DaemonID,
&i.Name,
&i.RuntimeMode,
&i.Provider,
&i.Status,
&i.DeviceInfo,
&i.Metadata,
&i.LastSeenAt,
&i.CreatedAt,
&i.UpdatedAt,
&i.OwnerID,
&i.LegacyDaemonID,
&i.Visibility,
&i.ProfileID,
&i.CustomName,
)
return i, err
}
const listAgentRuntimes = `-- name: ListAgentRuntimes :many
SELECT id, workspace_id, daemon_id, name, runtime_mode, provider, status, device_info, metadata, last_seen_at, created_at, updated_at, owner_id, legacy_daemon_id, visibility, profile_id, custom_name FROM agent_runtime
WHERE workspace_id = $1
ORDER BY created_at ASC
`
func (q *Queries) ListAgentRuntimes(ctx context.Context, workspaceID pgtype.UUID) ([]AgentRuntime, error) {
rows, err := q.db.Query(ctx, listAgentRuntimes, workspaceID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []AgentRuntime{}
for rows.Next() {
var i AgentRuntime
if err := rows.Scan(
&i.ID,
&i.WorkspaceID,
&i.DaemonID,
&i.Name,
&i.RuntimeMode,
&i.Provider,
&i.Status,
&i.DeviceInfo,
&i.Metadata,
&i.LastSeenAt,
&i.CreatedAt,
&i.UpdatedAt,
&i.OwnerID,
&i.LegacyDaemonID,
&i.Visibility,
&i.ProfileID,
&i.CustomName,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listAgentRuntimesByOwner = `-- name: ListAgentRuntimesByOwner :many
SELECT id, workspace_id, daemon_id, name, runtime_mode, provider, status, device_info, metadata, last_seen_at, created_at, updated_at, owner_id, legacy_daemon_id, visibility, profile_id, custom_name FROM agent_runtime
WHERE workspace_id = $1 AND owner_id = $2
ORDER BY created_at ASC
`
type ListAgentRuntimesByOwnerParams struct {
WorkspaceID pgtype.UUID `json:"workspace_id"`
OwnerID pgtype.UUID `json:"owner_id"`
}
func (q *Queries) ListAgentRuntimesByOwner(ctx context.Context, arg ListAgentRuntimesByOwnerParams) ([]AgentRuntime, error) {
rows, err := q.db.Query(ctx, listAgentRuntimesByOwner, arg.WorkspaceID, arg.OwnerID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []AgentRuntime{}
for rows.Next() {
var i AgentRuntime
if err := rows.Scan(
&i.ID,
&i.WorkspaceID,
&i.DaemonID,
&i.Name,
&i.RuntimeMode,
&i.Provider,
&i.Status,
&i.DeviceInfo,
&i.Metadata,
&i.LastSeenAt,
&i.CreatedAt,
&i.UpdatedAt,
&i.OwnerID,
&i.LegacyDaemonID,
&i.Visibility,
&i.ProfileID,
&i.CustomName,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listArchivedAgentIDsByRuntime = `-- name: ListArchivedAgentIDsByRuntime :many
SELECT id FROM agent WHERE runtime_id = $1 AND archived_at IS NOT NULL
`
// Companion to DeleteArchivedAgentsByRuntime: enumerates the archived agents
// about to be hard-deleted so the runtime teardown can pause autopilots that
// still point at them. Returns ids only — the caller only needs the set.
func (q *Queries) ListArchivedAgentIDsByRuntime(ctx context.Context, runtimeID pgtype.UUID) ([]pgtype.UUID, error) {
rows, err := q.db.Query(ctx, listArchivedAgentIDsByRuntime, runtimeID)
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 listDaemonCustomNames = `-- name: ListDaemonCustomNames :many
SELECT custom_name FROM agent_runtime
WHERE workspace_id = $1
AND daemon_id = $2
AND id <> $3
`
type ListDaemonCustomNamesParams struct {
WorkspaceID pgtype.UUID `json:"workspace_id"`
DaemonID pgtype.Text `json:"daemon_id"`
ExcludeID pgtype.UUID `json:"exclude_id"`
}
// Lists the custom_name of every OTHER runtime on (workspace_id, daemon_id)
// (MUL-4217). @exclude_id drops the just-registered row. The caller derives
// the machine-level name in Go — the same "all runtimes share one non-null
// name" rule the frontend applies in sharedCustomName — so a freshly-added
// runtime on an already-named machine can inherit that name and keep the
// machine's display name stable. A daemon hosts only a handful of runtimes
// (one per provider), so this is a tiny read.
func (q *Queries) ListDaemonCustomNames(ctx context.Context, arg ListDaemonCustomNamesParams) ([]pgtype.Text, error) {
rows, err := q.db.Query(ctx, listDaemonCustomNames, arg.WorkspaceID, arg.DaemonID, arg.ExcludeID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []pgtype.Text{}
for rows.Next() {
var custom_name pgtype.Text
if err := rows.Scan(&custom_name); err != nil {
return nil, err
}
items = append(items, custom_name)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const lockAgentRuntime = `-- name: LockAgentRuntime :one
SELECT id, workspace_id, daemon_id, name, runtime_mode, provider, status, device_info, metadata, last_seen_at, created_at, updated_at, owner_id, legacy_daemon_id, visibility, profile_id, custom_name FROM agent_runtime
WHERE id = $1
FOR UPDATE
`
// Acquires a row-level exclusive lock on the runtime row. Used at the
// top of the cascade-delete transaction so that:
// 1. PostgreSQL's FK validation on agent.runtime_id (FK ... ON DELETE
// RESTRICT) needs FOR KEY SHARE on the parent runtime row, which
// conflicts with FOR UPDATE — so any concurrent INSERT or UPDATE
// that would point a new/moved agent at this runtime blocks until
// our transaction finishes; and
// 2. concurrent UPDATE/DELETE of the runtime row itself (e.g. another
// delete attempt) waits for us to commit.
//
// Combined with ListActiveAgentsByRuntimeForUpdate (which row-locks the
// existing active set) this closes the plan-compare → archive race that
// was possible at read-committed isolation between the snapshot and the
// bulk archive.
func (q *Queries) LockAgentRuntime(ctx context.Context, id pgtype.UUID) (AgentRuntime, error) {
row := q.db.QueryRow(ctx, lockAgentRuntime, id)
var i AgentRuntime
err := row.Scan(
&i.ID,
&i.WorkspaceID,
&i.DaemonID,
&i.Name,
&i.RuntimeMode,
&i.Provider,
&i.Status,
&i.DeviceInfo,
&i.Metadata,
&i.LastSeenAt,
&i.CreatedAt,
&i.UpdatedAt,
&i.OwnerID,
&i.LegacyDaemonID,
&i.Visibility,
&i.ProfileID,
&i.CustomName,
)
return i, err
}
const markAgentRuntimeOnline = `-- name: MarkAgentRuntimeOnline :one
UPDATE agent_runtime
SET status = 'online', last_seen_at = now(), updated_at = now()
WHERE id = $1
RETURNING id, workspace_id, daemon_id, name, runtime_mode, provider, status, device_info, metadata, last_seen_at, created_at, updated_at, owner_id, legacy_daemon_id, visibility, profile_id, custom_name
`
// Used on the offline→online transition (and on first heartbeat after
// registration). Writes status, last_seen_at, and updated_at because the
// status flip is a real state change and we want updated_at to reflect it.
func (q *Queries) MarkAgentRuntimeOnline(ctx context.Context, id pgtype.UUID) (AgentRuntime, error) {
row := q.db.QueryRow(ctx, markAgentRuntimeOnline, id)
var i AgentRuntime
err := row.Scan(
&i.ID,
&i.WorkspaceID,
&i.DaemonID,
&i.Name,
&i.RuntimeMode,
&i.Provider,
&i.Status,
&i.DeviceInfo,
&i.Metadata,
&i.LastSeenAt,
&i.CreatedAt,
&i.UpdatedAt,
&i.OwnerID,
&i.LegacyDaemonID,
&i.Visibility,
&i.ProfileID,
&i.CustomName,
)
return i, err
}
const markRuntimesOfflineByIDs = `-- name: MarkRuntimesOfflineByIDs :many
UPDATE agent_runtime
SET status = 'offline', updated_at = now()
WHERE status = 'online'
AND id = ANY($1::uuid[])
AND last_seen_at < now() - make_interval(secs => $2::double precision)
RETURNING id, workspace_id, owner_id, daemon_id, provider
`
type MarkRuntimesOfflineByIDsParams struct {
Ids []pgtype.UUID `json:"ids"`
StaleSeconds float64 `json:"stale_seconds"`
}
type MarkRuntimesOfflineByIDsRow struct {
ID pgtype.UUID `json:"id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
OwnerID pgtype.UUID `json:"owner_id"`
DaemonID pgtype.Text `json:"daemon_id"`
Provider string `json:"provider"`
}
// Flips a known set of runtime IDs from online to offline. Paired with
// SelectStaleOnlineRuntimes in the sweeper so the candidate selection and
// the actual write are decoupled (the LivenessStore filter sits between).
//
// Re-checks the stale predicate inside the UPDATE so a concurrent heartbeat
// between the SELECT (candidate gather), the LivenessStore filter, and this
// UPDATE cannot demote a runtime that just refreshed last_seen_at. The
// legacy MarkStaleRuntimesOffline UPDATE had this property implicitly
// because the predicate and the write lived in one statement; here we
// carry it forward explicitly so the SELECT/filter/UPDATE pipeline retains
// the same race-freedom.
func (q *Queries) MarkRuntimesOfflineByIDs(ctx context.Context, arg MarkRuntimesOfflineByIDsParams) ([]MarkRuntimesOfflineByIDsRow, error) {
rows, err := q.db.Query(ctx, markRuntimesOfflineByIDs, arg.Ids, arg.StaleSeconds)
if err != nil {
return nil, err
}
defer rows.Close()
items := []MarkRuntimesOfflineByIDsRow{}
for rows.Next() {
var i MarkRuntimesOfflineByIDsRow
if err := rows.Scan(
&i.ID,
&i.WorkspaceID,
&i.OwnerID,
&i.DaemonID,
&i.Provider,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const pauseAutopilotsByAgentAssignees = `-- name: PauseAutopilotsByAgentAssignees :exec
UPDATE autopilot
SET status = 'paused', updated_at = now()
WHERE status = 'active'
AND assignee_type = 'agent'
AND assignee_id = ANY($1::uuid[])
`
// Pauses every active autopilot whose agent assignee is in the supplied list.
// Called before hard-deleting archived agents on runtime teardown so the rows
// do not become dangling (autopilot.assignee_id no longer has an agent FK
// since migration 096). Status='paused' makes the breakage visible in the UI
// — operators can re-point the autopilot at a live agent or delete it —
// rather than silently piling skipped runs.
func (q *Queries) PauseAutopilotsByAgentAssignees(ctx context.Context, assigneeIds []pgtype.UUID) error {
_, err := q.db.Exec(ctx, pauseAutopilotsByAgentAssignees, assigneeIds)
return err
}
const reassignAgentsToRuntime = `-- name: ReassignAgentsToRuntime :execrows
UPDATE agent
SET runtime_id = $1
WHERE runtime_id = $2
`
type ReassignAgentsToRuntimeParams struct {
NewRuntimeID pgtype.UUID `json:"new_runtime_id"`
OldRuntimeID pgtype.UUID `json:"old_runtime_id"`
}
// Re-points every agent referencing old_runtime_id at new_runtime_id.
func (q *Queries) ReassignAgentsToRuntime(ctx context.Context, arg ReassignAgentsToRuntimeParams) (int64, error) {
result, err := q.db.Exec(ctx, reassignAgentsToRuntime, arg.NewRuntimeID, arg.OldRuntimeID)
if err != nil {
return 0, err
}
return result.RowsAffected(), nil
}
const reassignTasksToRuntime = `-- name: ReassignTasksToRuntime :execrows
UPDATE agent_task_queue
SET runtime_id = $1
WHERE runtime_id = $2
`
type ReassignTasksToRuntimeParams struct {
NewRuntimeID pgtype.UUID `json:"new_runtime_id"`
OldRuntimeID pgtype.UUID `json:"old_runtime_id"`
}
// Re-points every queued/running/completed task referencing old_runtime_id.
// Required before deleting the old runtime row because agent_task_queue has
// an ON DELETE CASCADE FK that would otherwise drop historical tasks.
func (q *Queries) ReassignTasksToRuntime(ctx context.Context, arg ReassignTasksToRuntimeParams) (int64, error) {
result, err := q.db.Exec(ctx, reassignTasksToRuntime, arg.NewRuntimeID, arg.OldRuntimeID)
if err != nil {
return 0, err
}
return result.RowsAffected(), nil
}
const recordRuntimeLegacyDaemonID = `-- name: RecordRuntimeLegacyDaemonID :exec
UPDATE agent_runtime
SET legacy_daemon_id = COALESCE(legacy_daemon_id, $2)
WHERE id = $1
`
type RecordRuntimeLegacyDaemonIDParams struct {
ID pgtype.UUID `json:"id"`
LegacyDaemonID pgtype.Text `json:"legacy_daemon_id"`
}
// Remembers the most recent hostname-derived daemon_id that was merged into
// this row. Useful for debugging when tracing back why a given runtime row
// subsumed an old one, and only overwrites NULL so the earliest merge is
// preserved.
func (q *Queries) RecordRuntimeLegacyDaemonID(ctx context.Context, arg RecordRuntimeLegacyDaemonIDParams) error {
_, err := q.db.Exec(ctx, recordRuntimeLegacyDaemonID, arg.ID, arg.LegacyDaemonID)
return err
}
const selectStaleOnlineRuntimes = `-- name: SelectStaleOnlineRuntimes :many
SELECT id, workspace_id, owner_id, daemon_id, provider FROM agent_runtime
WHERE status = 'online'
AND last_seen_at < now() - make_interval(secs => $1::double precision)
`
type SelectStaleOnlineRuntimesRow struct {
ID pgtype.UUID `json:"id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
OwnerID pgtype.UUID `json:"owner_id"`
DaemonID pgtype.Text `json:"daemon_id"`
Provider string `json:"provider"`
}
// Lists online runtimes whose last_seen_at exceeds the stale window. The
// sweeper uses this as a candidate set, then optionally filters via the
// LivenessStore before flipping rows to offline (a fresh Redis liveness
// record means the DB row is just lagging, not actually dead).
func (q *Queries) SelectStaleOnlineRuntimes(ctx context.Context, staleSeconds float64) ([]SelectStaleOnlineRuntimesRow, error) {
rows, err := q.db.Query(ctx, selectStaleOnlineRuntimes, staleSeconds)
if err != nil {
return nil, err
}
defer rows.Close()
items := []SelectStaleOnlineRuntimesRow{}
for rows.Next() {
var i SelectStaleOnlineRuntimesRow
if err := rows.Scan(
&i.ID,
&i.WorkspaceID,
&i.OwnerID,
&i.DaemonID,
&i.Provider,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const setAgentRuntimeOffline = `-- name: SetAgentRuntimeOffline :exec
UPDATE agent_runtime
SET status = 'offline', updated_at = now()
WHERE id = $1
`
func (q *Queries) SetAgentRuntimeOffline(ctx context.Context, id pgtype.UUID) error {
_, err := q.db.Exec(ctx, setAgentRuntimeOffline, id)
return err
}
const touchAgentRuntimeLastSeen = `-- name: TouchAgentRuntimeLastSeen :execrows
UPDATE agent_runtime
SET last_seen_at = now()
WHERE id = $1 AND status = 'online'
`
// Bumps last_seen_at on an already-online runtime. Deliberately does NOT
// touch status or updated_at: status is unchanged on the hot heartbeat path,
// and avoiding updated_at keeps the row HOT-eligible (no index columns
// change) and avoids invalidating any downstream consumer that watches
// updated_at.
//
// The status='online' predicate is load-bearing: callers read rt.Status from
// a prior SELECT and may race with the sweeper, which can flip the row to
// offline between that SELECT and this UPDATE. Without the predicate this
// query would silently leave a freshly-heartbeated runtime stuck in offline.
// Returning affected rows lets callers detect that race and fall back to
// MarkAgentRuntimeOnline to flip the row back online.
func (q *Queries) TouchAgentRuntimeLastSeen(ctx context.Context, id pgtype.UUID) (int64, error) {
result, err := q.db.Exec(ctx, touchAgentRuntimeLastSeen, id)
if err != nil {
return 0, err
}
return result.RowsAffected(), nil
}
const touchAgentRuntimesLastSeenBatch = `-- name: TouchAgentRuntimesLastSeenBatch :execrows
UPDATE agent_runtime
SET last_seen_at = now()
WHERE id = ANY($1::uuid[]) AND status = 'online'
`
// Bulk variant of TouchAgentRuntimeLastSeen used by the BatchedHeartbeatScheduler:
// coalesces N per-runtime "bump last_seen_at" requests into a single UPDATE so a
// fleet beating every 15s costs ~1 DB transaction per batch tick instead of N.
//
// Same load-bearing predicate as the single-id form: status='online' avoids
// silently un-deleting a sweeper-flipped offline row, and we deliberately do
// NOT touch updated_at so the rows stay HOT-eligible. Affected-rows < len(ids)
// means some IDs raced to offline between Schedule and flush; their next beat
// will fall through the recordHeartbeat sync path and call MarkAgentRuntimeOnline.
func (q *Queries) TouchAgentRuntimesLastSeenBatch(ctx context.Context, ids []pgtype.UUID) (int64, error) {
result, err := q.db.Exec(ctx, touchAgentRuntimesLastSeenBatch, ids)
if err != nil {
return 0, err
}
return result.RowsAffected(), nil
}
const updateAgentRuntimeCustomName = `-- name: UpdateAgentRuntimeCustomName :one
UPDATE agent_runtime
SET custom_name = $1, updated_at = now()
WHERE id = $2
RETURNING id, workspace_id, daemon_id, name, runtime_mode, provider, status, device_info, metadata, last_seen_at, created_at, updated_at, owner_id, legacy_daemon_id, visibility, profile_id, custom_name
`
type UpdateAgentRuntimeCustomNameParams struct {
CustomName pgtype.Text `json:"custom_name"`
ID pgtype.UUID `json:"id"`
}
// Sets or clears a runtime's user-facing custom name (MUL-4217). custom_name
// overrides the daemon-proposed `name` for display; passing NULL reverts to
// the default. Kept separate from the registration upserts above (which do
// name = EXCLUDED.name on every heartbeat) so a custom name is never
// clobbered by the daemon. Gated at the handler to owner / workspace admin.
func (q *Queries) UpdateAgentRuntimeCustomName(ctx context.Context, arg UpdateAgentRuntimeCustomNameParams) (AgentRuntime, error) {
row := q.db.QueryRow(ctx, updateAgentRuntimeCustomName, arg.CustomName, arg.ID)
var i AgentRuntime
err := row.Scan(
&i.ID,
&i.WorkspaceID,
&i.DaemonID,
&i.Name,
&i.RuntimeMode,
&i.Provider,
&i.Status,
&i.DeviceInfo,
&i.Metadata,
&i.LastSeenAt,
&i.CreatedAt,
&i.UpdatedAt,
&i.OwnerID,
&i.LegacyDaemonID,
&i.Visibility,
&i.ProfileID,
&i.CustomName,
)
return i, err
}
const updateAgentRuntimeCustomNameByDaemon = `-- name: UpdateAgentRuntimeCustomNameByDaemon :many
UPDATE agent_runtime
SET custom_name = $1, updated_at = now()
WHERE workspace_id = $2
AND daemon_id = $3
AND ($4::uuid IS NULL OR owner_id = $4)
RETURNING id, workspace_id, daemon_id, name, runtime_mode, provider, status, device_info, metadata, last_seen_at, created_at, updated_at, owner_id, legacy_daemon_id, visibility, profile_id, custom_name
`
type UpdateAgentRuntimeCustomNameByDaemonParams struct {
CustomName pgtype.Text `json:"custom_name"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
DaemonID pgtype.Text `json:"daemon_id"`
OwnerID pgtype.UUID `json:"owner_id"`
}
// Machine-level rename (MUL-4217): applies one custom name to every runtime
// sharing a daemon_id in the workspace, since a single machine hosts one
// runtime per provider. @owner_id is NULL for workspace owners/admins (rename
// the whole machine) or the actor's user id otherwise (only their own
// runtimes on that machine), so a member cannot relabel someone else's
// runtime that happens to share the host.
func (q *Queries) UpdateAgentRuntimeCustomNameByDaemon(ctx context.Context, arg UpdateAgentRuntimeCustomNameByDaemonParams) ([]AgentRuntime, error) {
rows, err := q.db.Query(ctx, updateAgentRuntimeCustomNameByDaemon,
arg.CustomName,
arg.WorkspaceID,
arg.DaemonID,
arg.OwnerID,
)
if err != nil {
return nil, err
}
defer rows.Close()
items := []AgentRuntime{}
for rows.Next() {
var i AgentRuntime
if err := rows.Scan(
&i.ID,
&i.WorkspaceID,
&i.DaemonID,
&i.Name,
&i.RuntimeMode,
&i.Provider,
&i.Status,
&i.DeviceInfo,
&i.Metadata,
&i.LastSeenAt,
&i.CreatedAt,
&i.UpdatedAt,
&i.OwnerID,
&i.LegacyDaemonID,
&i.Visibility,
&i.ProfileID,
&i.CustomName,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const updateAgentRuntimeVisibility = `-- name: UpdateAgentRuntimeVisibility :one
UPDATE agent_runtime
SET visibility = $1, updated_at = now()
WHERE id = $2
RETURNING id, workspace_id, daemon_id, name, runtime_mode, provider, status, device_info, metadata, last_seen_at, created_at, updated_at, owner_id, legacy_daemon_id, visibility, profile_id, custom_name
`
type UpdateAgentRuntimeVisibilityParams struct {
Visibility string `json:"visibility"`
ID pgtype.UUID `json:"id"`
}
// Toggles a runtime between 'private' (only owner can bind agents) and
// 'public' (any workspace member can). Default for new rows is 'private'
// (see migration 083). Gated at the handler layer to owner / workspace
// admin only.
func (q *Queries) UpdateAgentRuntimeVisibility(ctx context.Context, arg UpdateAgentRuntimeVisibilityParams) (AgentRuntime, error) {
row := q.db.QueryRow(ctx, updateAgentRuntimeVisibility, arg.Visibility, arg.ID)
var i AgentRuntime
err := row.Scan(
&i.ID,
&i.WorkspaceID,
&i.DaemonID,
&i.Name,
&i.RuntimeMode,
&i.Provider,
&i.Status,
&i.DeviceInfo,
&i.Metadata,
&i.LastSeenAt,
&i.CreatedAt,
&i.UpdatedAt,
&i.OwnerID,
&i.LegacyDaemonID,
&i.Visibility,
&i.ProfileID,
&i.CustomName,
)
return i, err
}
const upsertAgentRuntime = `-- name: UpsertAgentRuntime :one
INSERT INTO agent_runtime (
workspace_id,
daemon_id,
name,
runtime_mode,
provider,
status,
device_info,
metadata,
owner_id,
last_seen_at
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, now())
ON CONFLICT (workspace_id, daemon_id, provider) WHERE profile_id IS NULL
DO UPDATE SET
name = EXCLUDED.name,
runtime_mode = EXCLUDED.runtime_mode,
status = EXCLUDED.status,
device_info = EXCLUDED.device_info,
metadata = EXCLUDED.metadata,
owner_id = COALESCE(EXCLUDED.owner_id, agent_runtime.owner_id),
last_seen_at = now(),
updated_at = now()
RETURNING id, workspace_id, daemon_id, name, runtime_mode, provider, status, device_info, metadata, last_seen_at, created_at, updated_at, owner_id, legacy_daemon_id, visibility, profile_id, custom_name, (xmax = 0) AS inserted
`
type UpsertAgentRuntimeParams struct {
WorkspaceID pgtype.UUID `json:"workspace_id"`
DaemonID pgtype.Text `json:"daemon_id"`
Name string `json:"name"`
RuntimeMode string `json:"runtime_mode"`
Provider string `json:"provider"`
Status string `json:"status"`
DeviceInfo string `json:"device_info"`
Metadata []byte `json:"metadata"`
OwnerID pgtype.UUID `json:"owner_id"`
}
type UpsertAgentRuntimeRow struct {
ID pgtype.UUID `json:"id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
DaemonID pgtype.Text `json:"daemon_id"`
Name string `json:"name"`
RuntimeMode string `json:"runtime_mode"`
Provider string `json:"provider"`
Status string `json:"status"`
DeviceInfo string `json:"device_info"`
Metadata []byte `json:"metadata"`
LastSeenAt pgtype.Timestamptz `json:"last_seen_at"`
CreatedAt pgtype.Timestamptz `json:"created_at"`
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
OwnerID pgtype.UUID `json:"owner_id"`
LegacyDaemonID pgtype.Text `json:"legacy_daemon_id"`
Visibility string `json:"visibility"`
ProfileID pgtype.UUID `json:"profile_id"`
CustomName pgtype.Text `json:"custom_name"`
Inserted bool `json:"inserted"`
}
// (xmax = 0) AS inserted distinguishes a fresh insert (true) from an upsert
// that updated an existing row (false). Analytics reads this to fire
// runtime_registered/runtime_ready only on first-time registration.
// Built-in runtimes carry no profile_id. The arbiter is the partial unique
// index from migration 121 (WHERE profile_id IS NULL); the predicate must be
// spelled out so Postgres selects that partial index, not the custom-runtime
// one on (workspace_id, daemon_id, profile_id).
func (q *Queries) UpsertAgentRuntime(ctx context.Context, arg UpsertAgentRuntimeParams) (UpsertAgentRuntimeRow, error) {
row := q.db.QueryRow(ctx, upsertAgentRuntime,
arg.WorkspaceID,
arg.DaemonID,
arg.Name,
arg.RuntimeMode,
arg.Provider,
arg.Status,
arg.DeviceInfo,
arg.Metadata,
arg.OwnerID,
)
var i UpsertAgentRuntimeRow
err := row.Scan(
&i.ID,
&i.WorkspaceID,
&i.DaemonID,
&i.Name,
&i.RuntimeMode,
&i.Provider,
&i.Status,
&i.DeviceInfo,
&i.Metadata,
&i.LastSeenAt,
&i.CreatedAt,
&i.UpdatedAt,
&i.OwnerID,
&i.LegacyDaemonID,
&i.Visibility,
&i.ProfileID,
&i.CustomName,
&i.Inserted,
)
return i, err
}
const upsertAgentRuntimeWithProfile = `-- name: UpsertAgentRuntimeWithProfile :one
INSERT INTO agent_runtime (
workspace_id,
daemon_id,
name,
runtime_mode,
provider,
status,
device_info,
metadata,
owner_id,
profile_id,
last_seen_at
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, now())
ON CONFLICT (workspace_id, daemon_id, profile_id) WHERE profile_id IS NOT NULL
DO UPDATE SET
name = EXCLUDED.name,
runtime_mode = EXCLUDED.runtime_mode,
provider = EXCLUDED.provider,
status = EXCLUDED.status,
device_info = EXCLUDED.device_info,
metadata = EXCLUDED.metadata,
owner_id = COALESCE(EXCLUDED.owner_id, agent_runtime.owner_id),
last_seen_at = now(),
updated_at = now()
RETURNING id, workspace_id, daemon_id, name, runtime_mode, provider, status, device_info, metadata, last_seen_at, created_at, updated_at, owner_id, legacy_daemon_id, visibility, profile_id, custom_name, (xmax = 0) AS inserted
`
type UpsertAgentRuntimeWithProfileParams struct {
WorkspaceID pgtype.UUID `json:"workspace_id"`
DaemonID pgtype.Text `json:"daemon_id"`
Name string `json:"name"`
RuntimeMode string `json:"runtime_mode"`
Provider string `json:"provider"`
Status string `json:"status"`
DeviceInfo string `json:"device_info"`
Metadata []byte `json:"metadata"`
OwnerID pgtype.UUID `json:"owner_id"`
ProfileID pgtype.UUID `json:"profile_id"`
}
type UpsertAgentRuntimeWithProfileRow struct {
ID pgtype.UUID `json:"id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
DaemonID pgtype.Text `json:"daemon_id"`
Name string `json:"name"`
RuntimeMode string `json:"runtime_mode"`
Provider string `json:"provider"`
Status string `json:"status"`
DeviceInfo string `json:"device_info"`
Metadata []byte `json:"metadata"`
LastSeenAt pgtype.Timestamptz `json:"last_seen_at"`
CreatedAt pgtype.Timestamptz `json:"created_at"`
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
OwnerID pgtype.UUID `json:"owner_id"`
LegacyDaemonID pgtype.Text `json:"legacy_daemon_id"`
Visibility string `json:"visibility"`
ProfileID pgtype.UUID `json:"profile_id"`
CustomName pgtype.Text `json:"custom_name"`
Inserted bool `json:"inserted"`
}
// Custom-runtime registration: a daemon resolved a workspace runtime_profile's
// command_name on PATH and is registering an instance of it. The arbiter is the
// partial unique index from migration 120 (WHERE profile_id IS NOT NULL), so a
// single daemon can host the built-in provider AND any number of custom
// profiles of the same protocol family. provider stays the protocol family so
// task routing (agent.New(provider)) is unchanged; profile_id is the stable
// identity. (xmax = 0) AS inserted mirrors UpsertAgentRuntime.
func (q *Queries) UpsertAgentRuntimeWithProfile(ctx context.Context, arg UpsertAgentRuntimeWithProfileParams) (UpsertAgentRuntimeWithProfileRow, error) {
row := q.db.QueryRow(ctx, upsertAgentRuntimeWithProfile,
arg.WorkspaceID,
arg.DaemonID,
arg.Name,
arg.RuntimeMode,
arg.Provider,
arg.Status,
arg.DeviceInfo,
arg.Metadata,
arg.OwnerID,
arg.ProfileID,
)
var i UpsertAgentRuntimeWithProfileRow
err := row.Scan(
&i.ID,
&i.WorkspaceID,
&i.DaemonID,
&i.Name,
&i.RuntimeMode,
&i.Provider,
&i.Status,
&i.DeviceInfo,
&i.Metadata,
&i.LastSeenAt,
&i.CreatedAt,
&i.UpdatedAt,
&i.OwnerID,
&i.LegacyDaemonID,
&i.Visibility,
&i.ProfileID,
&i.CustomName,
&i.Inserted,
)
return i, err
}