mirror of
https://github.com/multica-ai/multica.git
synced 2026-07-16 22:59:04 +02:00
* feat(daemon-claim): machine-level batch task claim endpoint (MUL-4257) Collapse the per-runtime /tasks/claim poll fan-out into a single machine-level batch claim to cut /api/daemon claim request volume. Server: - agent.sql: = ANY(runtime_ids) batch variants of the claim queries (ListQueuedClaimCandidatesByRuntimes, PromoteDueDeferredTasksForRuntimes, ReclaimStaleDispatchedTasksForRuntimes); runtime.sql: GetAgentRuntimes(= ANY) so a whole machine's runtimes are resolved/promoted/reclaimed/listed in a constant number of queries instead of N. - service.ClaimTasksForRuntimes: claim up to max_tasks across a runtime set, preserving per-(issue,agent) serialization, the concurrency cap, the empty-claim cache short-circuit, and every dispatch side effect. Batch promote replays the per-row side effects (task:queued + empty-cache Bump). - handler.ClaimTasksByRuntime (canonical POST /api/daemon/tasks/claim, with a transitional /claim alias): validates daemon_id (required; must match the mdt_ token) and rejects runtimes bound to a different daemon (group-ownership check mirroring the WS path); resolves+authorizes each runtime_id; claims; and finalizes each task through the SAME FinalizeTaskClaim as the per-runtime endpoint (atomic token + delivered_comment_ids receipt), requeueing the exact claim and omitting it on failure. buildClaimedTaskResponse is extracted from the per-runtime handler and returns the delivered-comment ids plus a structured *claimBuildFailure so both paths share identical payload building and failure semantics (workspace-isolation, chat-input load/empty). - max_tasks: negative -> 400, zero -> empty (never coerce to 1), positive capped at 32. runtime_ids parsed with non-panicking util.ParseUUID. Daemon: - Client.ClaimTasks posts daemon_id + runtime set + free-slot count to the canonical path under a short request-scoped timeout, bounding the head-of-line coupling the per-runtime pollers avoid (MUL-1744). Tests: service batch drain / max_tasks cap / deferred-promote receipt / finalize-failure rollback+requeue; handler routing + token, cross-workspace skip, cross-daemon skip, daemon_id required, owner-missing cancel, max_tasks=0/negative, invalid-uuid skip, comment delivery receipt, stale-reclaim replacement receipt; client posts/parses (daemon_id + canonical path). Follow-up: cut the daemon pollLoop over to a single batched poller (flips the MUL-1744 isolation contract; needs its concurrency tests redesigned). Co-authored-by: multica-agent <github@multica.ai> * feat(daemon-ws): generic WS request/response transport for daemon RPC (MUL-4257) Add a generic daemon->server request/response layer over the existing WS control connection, the transport for WS-first claim (HTTP fallback): - protocol: daemon:rpc_request / daemon:rpc_response envelopes with a correlation request_id + method + body, and an rpc-v1 capability gate. - daemonws.Hub: SetRPCHandler + goroutine-dispatched handleRPCFrame (bounded by a per-connection in-flight cap) that echoes the request_id; missing handler / saturation return non-2xx so the daemon falls back to HTTP. Read limit raised to 64KB for rpc requests carrying a runtime set. - hub tests: round-trip, handler-error->non-2xx, no-handler->503. Co-authored-by: multica-agent <github@multica.ai> * feat(daemon-ws): WS-first task claim over the generic RPC transport (MUL-4257) Bind claim to the WS request/response layer, with HTTP fallback: - server: handler.DaemonRPCHandler adapts a daemon:rpc_request (method tasks.claim) to the existing HTTP ClaimTasksByRuntime via a synthetic in-process request carrying the WS connection's identity (daemon_id + workspace + capabilities), so all auth / payload-building / finalization is reused unchanged. Wired via daemonHub.SetRPCHandler. ClientIdentity now captures X-Client-Capabilities so capability gating matches the HTTP path. - daemon: wsRPCClient correlates responses by request_id over the shared WS connection; attached to the live connection's write channel (guarded so a Call racing teardown never sends on a closed channel) and detached on disconnect. rpc_response frames are routed in the read loop. Daemon.ClaimTasksWSFirst issues tasks.claim over WS and falls back to the HTTP claim endpoint on any transport failure (no conn / buffer full / timeout) — wired into the poller at the poller cutover. - tests: handler tasks.claim RPC end-to-end (claims + dispatches) + unknown method 404; daemon wsRPCClient round-trip / timeout / unavailable / server-error / detach-fails-pending (all under -race). Co-authored-by: multica-agent <github@multica.ai> * feat(daemon): cut claim poller over to machine-level ClaimTasksWSFirst (MUL-4257) Replace the per-runtime HTTP poll loop with a single batch poller: each cycle acquires all free execution slots (slot-before-claim) and issues ONE ClaimTasksWSFirst across every runtime the daemon hosts (WS-first, HTTP fallback), dispatching each returned task to its runtime. Wakeups (targeted / catch-up / runtime-set change) collapse to one nudge. Removes runRuntimePoller + runtimePollOffset. The WS handshake now advertises the same capabilities as HTTP (+ rpc-v1) so WS-built claim payloads keep skill-ref / coalesced-comment gating. Trades per-runtime isolation (MUL-1744) for one request, bounded by the short per-request WS timeout / client timeout. Tests: batch poller claims across runtimes + skips-at-capacity + pollLoop shutdown drain (replacing the per-runtime poller tests); heartbeat isolation + runtime-set watcher kept. Co-authored-by: multica-agent <github@multica.ai> * fix(daemon-ws): WS RPC disconnect-race panic + batch stale-comment-plan repair (MUL-4257) Two PR #5193 review blockers: 1) WS RPC send-on-closed-channel race, both ends: - server: give each connection a cancelable ctx (cancelled on readPump teardown) and run the RPC handler under it, so a slow claim stops on disconnect; guard c.send with sendMu/sendClosed (trySend) so a late RPC response goroutine never writes to the closed channel. Heartbeat ack routed through the same guard. - daemon: wsRPCClient.deliver now sends under the mutex, serialized with attach(nil)'s close+delete, so a delivered response can't hit a channel the detach path just closed. - regressions (-race): daemon deliver-vs-detach; server disconnect-during-handler-response. 2) batch claim now runs the stale-comment-plan repair: extracted the per-runtime handler's repair (trigger deleted, only coalesced survive -> cancel + replay survivors) into shared repairStaleCommentPlanIfNeeded, called by both claim paths. Prevents the batch path (now the default poller) from finalizing+dispatching a task with no comment input and silently dropping the surviving user comment. Regression: batch omits the stale task, cancels it, and rebuilds the survivor into a new trigger plan. Co-authored-by: multica-agent <github@multica.ai> * fix(daemon-ws): server-side RPC deadline + legacy claim fallback (MUL-4257) Two review blockers: 1) WS RPC timeout/fallback (GPT-Boy): the daemon's WS wait didn't cancel server-side claim, so a slow WS claim could commit after the daemon fell back to HTTP, leaking dispatched tasks and breaking the free-slot bound. Fix: RPC envelope carries TimeoutMs; the server bounds the handler ctx by it (so ClaimTasksByRuntime's tx is cancelled/rolled back at the deadline), and the daemon waits budget + grace so a claim that committed before the deadline still reports back. A committed-then-unreported claim degrades to the same stale-reclaim safety net as HTTP, never a double effective claim. Regression: server-side TimeoutMs cancels the handler. 2) Backward compat (Terra-Boy): a new daemon against a server without the batch route (/api/daemon/tasks/claim 404) couldn't claim. Fix: ClaimTasksWSFirst falls back to the legacy per-runtime ClaimTask loop on a batch 404 and caches 'batch unsupported' (reset on WS reconnect to re-probe after a server upgrade). Regression: server exposing only the legacy route. Co-authored-by: multica-agent <github@multica.ai> * fix(daemon-ws): no double-claim on WS teardown/detach (MUL-4257) Sol-Boy review blocker: on reconnect, teardown failed the pending RPC (→ HTTP fallback) but then flushed the queued tasks.claim frame to the still-alive socket, so the server committed the WS claim on top of the HTTP one — double claim, WS batch orphaned to stale reclaim, breaking the free-slot bound. - Teardown now closes the connection FIRST, so runWSWriter discards the queued RPC frame (write error path) instead of delivering it. - A detach while a claim's frame is already in flight now returns a distinct errWSRPCUncertain; ClaimTasksWSFirst does NOT HTTP-fall-back on uncertain (the WS claim may have committed) — it skips the cycle and lets reclaim / the next poll recover. Genuine 'not sent' / timeout still fall back (safe: the server-side deadline guarantees no uncommitted claim by budget+grace). - Regression: detach during an in-flight WS claim asserts zero HTTP claims (at most one path claims); plus the existing detach/deliver-race and server-timeout tests. Co-authored-by: multica-agent <github@multica.ai> * fix(daemon-ws): cancelable RPC frames close the backpressure double-claim (MUL-4257) Sol-Boy review blocker: the client's response budget starts at enqueue, but the socket write is async (10s write deadline). A backpressured writer could hold a tasks.claim in the local queue past the client timeout — the daemon HTTP-fell-back, then the writer woke and delivered the stale WS frame, so the server committed it too: same free slots claimed twice. No detach occurs, so the prior errWSRPCUncertain fix did not cover it. - WS frames are now cancelable (wsOutbound{sent,canceled} under a mutex). The writer calls beginWrite() before WriteMessage and skips cancelled frames. - On give-up (timeout / detach / ctx), Call cancels the queued frame: if it was still pending the cancel wins and the frame is guaranteed never delivered (errWSRPCUnavailable → safe HTTP fallback); if the writer already began sending it the cancel loses and the outcome is errWSRPCUncertain (no fallback). The decision is atomic, so at most one transport claims. Tests: wsOutbound cancel-before-write vs write-before-cancel; Call timeout cancels an unsent frame (writer then drops it) vs uncertain when already sent; plus the updated detach and existing timeout/race tests. Co-authored-by: multica-agent <github@multica.ai> * fix(batch-claim): return partial success instead of dropping committed claims (MUL-4257) Sol-Boy review blocker: ClaimTasksForRuntimes reclaims (step 2) and claims per agent (step 6) in independent transactions, but a step-4 candidate-SELECT error or a mid-loop ClaimTask error did 'return nil, err' — discarding tasks already committed as dispatched. The handler 500s; the daemon sees a definite (non- uncertain) 500 and HTTP-falls-back, claiming a SECOND batch into the same free slots while the first batch waits for stale reclaim — the double-claim this PR removes. - Both error paths now prefer partial success: if any task has already committed (claimed non-empty), return it (nil error) so the handler finalizes and returns 200; the errored candidates stay queued for the next poll. The remaining error is logged. Only a genuinely empty result still returns the error (safe: no committed claim to lose, HTTP fallback just re-fails). Regression (internal/service, DB-backed, fault-injected): - PartialSuccessOnSecondAgentClaimFailure: fail the 2nd ClaimTask's Begin → the first agent's committed task is returned, not dropped. - PartialSuccessOnCandidateQueryFailureAfterReclaim: a stale dispatched task is reclaimed, then the candidate SELECT fails → the reclaimed task is returned. Co-authored-by: multica-agent <github@multica.ai> --------- Co-authored-by: Eve <eve@multica-ai.local> Co-authored-by: multica-agent <github@multica.ai>
388 lines
17 KiB
SQL
388 lines
17 KiB
SQL
-- name: ListAgentRuntimes :many
|
|
SELECT * FROM agent_runtime
|
|
WHERE workspace_id = $1
|
|
ORDER BY created_at ASC;
|
|
|
|
-- name: GetAgentRuntime :one
|
|
SELECT * FROM agent_runtime
|
|
WHERE id = $1;
|
|
|
|
-- name: GetAgentRuntimes :many
|
|
-- Batch variant of GetAgentRuntime (MUL-4257): loads every runtime in the
|
|
-- input set in one round trip so the machine-level batch claim handler can
|
|
-- resolve+authorize all of a daemon's runtimes without one point query per
|
|
-- runtime. Rows are returned only for ids that exist; the caller matches them
|
|
-- back by id and skips any that are missing.
|
|
SELECT * FROM agent_runtime
|
|
WHERE id = ANY(@ids::uuid[]);
|
|
|
|
-- name: LockAgentRuntime :one
|
|
-- 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.
|
|
SELECT * FROM agent_runtime
|
|
WHERE id = $1
|
|
FOR UPDATE;
|
|
|
|
-- name: GetAgentRuntimeForWorkspace :one
|
|
SELECT * FROM agent_runtime
|
|
WHERE id = $1 AND workspace_id = $2;
|
|
|
|
-- name: UpsertAgentRuntime :one
|
|
-- (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.
|
|
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())
|
|
-- 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).
|
|
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 *, (xmax = 0) AS inserted;
|
|
|
|
-- name: UpsertAgentRuntimeWithProfile :one
|
|
-- 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.
|
|
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 *, (xmax = 0) AS inserted;
|
|
|
|
-- name: UpdateAgentRuntimeVisibility :one
|
|
-- 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.
|
|
UPDATE agent_runtime
|
|
SET visibility = @visibility, updated_at = now()
|
|
WHERE id = @id
|
|
RETURNING *;
|
|
|
|
-- name: UpdateAgentRuntimeCustomName :one
|
|
-- 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.
|
|
UPDATE agent_runtime
|
|
SET custom_name = @custom_name, updated_at = now()
|
|
WHERE id = @id
|
|
RETURNING *;
|
|
|
|
-- name: UpdateAgentRuntimeCustomNameByDaemon :many
|
|
-- 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.
|
|
UPDATE agent_runtime
|
|
SET custom_name = @custom_name, updated_at = now()
|
|
WHERE workspace_id = @workspace_id
|
|
AND daemon_id = @daemon_id
|
|
AND (@owner_id::uuid IS NULL OR owner_id = @owner_id)
|
|
RETURNING *;
|
|
|
|
-- name: ListDaemonCustomNames :many
|
|
-- 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.
|
|
SELECT custom_name FROM agent_runtime
|
|
WHERE workspace_id = @workspace_id
|
|
AND daemon_id = @daemon_id
|
|
AND id <> @exclude_id;
|
|
|
|
|
|
-- name: TouchAgentRuntimeLastSeen :execrows
|
|
-- 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.
|
|
UPDATE agent_runtime
|
|
SET last_seen_at = now()
|
|
WHERE id = $1 AND status = 'online';
|
|
|
|
-- name: TouchAgentRuntimesLastSeenBatch :execrows
|
|
-- 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.
|
|
UPDATE agent_runtime
|
|
SET last_seen_at = now()
|
|
WHERE id = ANY(@ids::uuid[]) AND status = 'online';
|
|
|
|
-- name: MarkAgentRuntimeOnline :one
|
|
-- 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.
|
|
UPDATE agent_runtime
|
|
SET status = 'online', last_seen_at = now(), updated_at = now()
|
|
WHERE id = $1
|
|
RETURNING *;
|
|
|
|
-- name: SetAgentRuntimeOffline :exec
|
|
UPDATE agent_runtime
|
|
SET status = 'offline', updated_at = now()
|
|
WHERE id = $1;
|
|
|
|
-- name: SelectStaleOnlineRuntimes :many
|
|
-- 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).
|
|
SELECT id, workspace_id, owner_id, daemon_id, provider FROM agent_runtime
|
|
WHERE status = 'online'
|
|
AND last_seen_at < now() - make_interval(secs => @stale_seconds::double precision);
|
|
|
|
-- name: MarkRuntimesOfflineByIDs :many
|
|
-- 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.
|
|
UPDATE agent_runtime
|
|
SET status = 'offline', updated_at = now()
|
|
WHERE status = 'online'
|
|
AND id = ANY(@ids::uuid[])
|
|
AND last_seen_at < now() - make_interval(secs => @stale_seconds::double precision)
|
|
RETURNING id, workspace_id, owner_id, daemon_id, provider;
|
|
|
|
-- name: FailTasksForOfflineRuntimes :many
|
|
-- 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.
|
|
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 *;
|
|
|
|
-- name: ListAgentRuntimesByOwner :many
|
|
SELECT * FROM agent_runtime
|
|
WHERE workspace_id = $1 AND owner_id = $2
|
|
ORDER BY created_at ASC;
|
|
|
|
-- name: ForceOfflineRuntimesByIDs :many
|
|
-- 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.
|
|
UPDATE agent_runtime
|
|
SET status = 'offline', updated_at = now()
|
|
WHERE id = ANY(@runtime_ids::uuid[]) AND status = 'online'
|
|
RETURNING id, workspace_id, owner_id, daemon_id, provider;
|
|
|
|
-- name: CancelAgentTasksByRuntimeOrAgent :many
|
|
-- 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.
|
|
UPDATE agent_task_queue
|
|
SET status = 'cancelled', completed_at = now()
|
|
WHERE (runtime_id = ANY(@runtime_ids::uuid[]) OR agent_id = ANY(@agent_ids::uuid[]))
|
|
AND status IN ('queued', 'dispatched', 'running', 'waiting_local_directory')
|
|
RETURNING *;
|
|
|
|
-- name: DeleteAgentRuntime :exec
|
|
DELETE FROM agent_runtime WHERE id = $1;
|
|
|
|
-- name: DeleteSystemAgentsByRuntime :exec
|
|
-- System agents are invisible execution infrastructure (for example the Agent
|
|
-- Builder). Remove them before deleting their runtime so the RESTRICT runtime
|
|
-- FK cannot block an otherwise dependency-free delete.
|
|
DELETE FROM agent WHERE runtime_id = $1 AND kind = 'system';
|
|
|
|
-- name: CountActiveAgentsByRuntime :one
|
|
SELECT count(*) FROM agent WHERE runtime_id = $1 AND archived_at IS NULL;
|
|
|
|
-- 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
|
|
);
|
|
|
|
-- name: DeleteArchivedAgentsByRuntime :exec
|
|
DELETE FROM agent WHERE runtime_id = $1 AND archived_at IS NOT NULL;
|
|
|
|
-- name: PauseAutopilotsByAgentAssignees :exec
|
|
-- 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.
|
|
UPDATE autopilot
|
|
SET status = 'paused', updated_at = now()
|
|
WHERE status = 'active'
|
|
AND assignee_type = 'agent'
|
|
AND assignee_id = ANY(@assignee_ids::uuid[]);
|
|
|
|
-- name: ListArchivedAgentIDsByRuntime :many
|
|
-- 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.
|
|
SELECT id FROM agent WHERE runtime_id = $1 AND archived_at IS NOT NULL;
|
|
|
|
-- name: DeleteSquadsByArchivedAgentsOnRuntime :exec
|
|
-- 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.
|
|
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;
|
|
|
|
-- name: FindLegacyRuntimesByDaemonID :many
|
|
-- 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.
|
|
SELECT * FROM agent_runtime
|
|
WHERE workspace_id = @workspace_id
|
|
AND provider = @provider
|
|
AND LOWER(daemon_id) = LOWER(@daemon_id);
|
|
|
|
-- name: ReassignAgentsToRuntime :execrows
|
|
-- Re-points every agent referencing old_runtime_id at new_runtime_id.
|
|
UPDATE agent
|
|
SET runtime_id = @new_runtime_id
|
|
WHERE runtime_id = @old_runtime_id;
|
|
|
|
-- name: ReassignTasksToRuntime :execrows
|
|
-- 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.
|
|
UPDATE agent_task_queue
|
|
SET runtime_id = @new_runtime_id
|
|
WHERE runtime_id = @old_runtime_id;
|
|
|
|
-- name: RecordRuntimeLegacyDaemonID :exec
|
|
-- 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.
|
|
UPDATE agent_runtime
|
|
SET legacy_daemon_id = COALESCE(legacy_daemon_id, $2)
|
|
WHERE id = $1;
|
|
|
|
-- name: DeleteStaleOfflineRuntimes :many
|
|
-- 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.
|
|
DELETE FROM agent_runtime
|
|
WHERE status = 'offline'
|
|
AND last_seen_at < now() - make_interval(secs => @stale_seconds::double precision)
|
|
AND id NOT IN (SELECT DISTINCT runtime_id FROM agent)
|
|
RETURNING id, workspace_id;
|