From 8f02a7ef977b6ea55c626a599f0bd53792ba728b Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 6 Jun 2026 14:36:27 +0000 Subject: [PATCH] PAP-10462 block invalid agent heartbeat dispatch --- .../src/__tests__/agent-invokability.test.ts | 68 +++++++ .../heartbeat-archived-company-guard.test.ts | 174 ++++++++++++++++ server/src/routes/agents.ts | 29 ++- server/src/services/agent-invokability.ts | 182 +++++++++++++++++ server/src/services/heartbeat.ts | 192 ++++++++++++++++-- server/src/services/recovery/service.ts | 17 +- 6 files changed, 633 insertions(+), 29 deletions(-) create mode 100644 server/src/__tests__/agent-invokability.test.ts create mode 100644 server/src/services/agent-invokability.ts diff --git a/server/src/__tests__/agent-invokability.test.ts b/server/src/__tests__/agent-invokability.test.ts new file mode 100644 index 00000000000..a2a91024d69 --- /dev/null +++ b/server/src/__tests__/agent-invokability.test.ts @@ -0,0 +1,68 @@ +import { describe, expect, it } from "vitest"; +import { + evaluateAgentInvokability, + listInvalidOrgChainDescendantIds, + type AgentOrgRow, +} from "../services/agent-invokability.ts"; + +function agent(partial: Partial & Pick): AgentOrgRow { + return { + companyId: "company-1", + reportsTo: null, + status: "active", + ...partial, + }; +} + +describe("agent invokability", () => { + it("blocks active descendants under a terminated manager as invalid-org-chain", () => { + const rows = [ + agent({ id: "ceo", status: "terminated" }), + agent({ id: "cto", reportsTo: "ceo" }), + agent({ id: "coder", reportsTo: "cto" }), + ]; + + const result = evaluateAgentInvokability(rows[2], rows); + + expect(result).toMatchObject({ + invokable: false, + reason: "manager_terminated", + invalidOrgChain: true, + details: { + managerId: "ceo", + reportingChainAgentIds: ["cto", "ceo"], + }, + }); + }); + + it("reports missing managers and cycles as invalid-org-chain", () => { + const missingManager = [agent({ id: "coder", reportsTo: "missing" })]; + expect(evaluateAgentInvokability(missingManager[0], missingManager)).toMatchObject({ + invokable: false, + reason: "manager_missing", + invalidOrgChain: true, + }); + + const cycle = [ + agent({ id: "a", reportsTo: "b" }), + agent({ id: "b", reportsTo: "a" }), + ]; + expect(evaluateAgentInvokability(cycle[0], cycle)).toMatchObject({ + invokable: false, + reason: "reporting_cycle", + invalidOrgChain: true, + }); + }); + + it("lists non-terminated descendants made invalid by a terminated root", () => { + const rows = [ + agent({ id: "ceo", status: "terminated" }), + agent({ id: "cto", reportsTo: "ceo" }), + agent({ id: "coder", reportsTo: "cto" }), + agent({ id: "old-coder", reportsTo: "cto", status: "terminated" }), + agent({ id: "other-root" }), + ]; + + expect(listInvalidOrgChainDescendantIds("ceo", rows).sort()).toEqual(["coder", "cto"]); + }); +}); diff --git a/server/src/__tests__/heartbeat-archived-company-guard.test.ts b/server/src/__tests__/heartbeat-archived-company-guard.test.ts index b0eb6b54fc4..085f8bfd21b 100644 --- a/server/src/__tests__/heartbeat-archived-company-guard.test.ts +++ b/server/src/__tests__/heartbeat-archived-company-guard.test.ts @@ -5,6 +5,7 @@ import { agentWakeupRequests, companies, createDb, + heartbeatRunEvents, heartbeatRuns, issues, } from "@paperclipai/db"; @@ -33,6 +34,7 @@ describeEmbeddedPostgres("heartbeat archived-company guard", () => { }, 20_000); afterEach(async () => { + await db.delete(heartbeatRunEvents); await db.delete(heartbeatRuns); await db.delete(agentWakeupRequests); await db.delete(issues); @@ -77,6 +79,60 @@ describeEmbeddedPostgres("heartbeat archived-company guard", () => { return { companyId, agentId }; } + async function insertInvalidOrgChainAgent() { + const companyId = randomUUID(); + const managerId = randomUUID(); + const childId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Invalid Org Co", + status: "active", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values([ + { + id: managerId, + companyId, + name: "Terminated Manager", + role: "cto", + status: "terminated", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: { + heartbeat: { + enabled: true, + intervalSec: 60, + wakeOnDemand: true, + }, + }, + permissions: {}, + }, + { + id: childId, + companyId, + name: "Invalid Chain Child", + role: "engineer", + reportsTo: managerId, + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: { + heartbeat: { + enabled: true, + intervalSec: 60, + wakeOnDemand: true, + }, + }, + permissions: {}, + }, + ]); + + return { companyId, managerId, childId }; + } + it("does not iterate archived-company agents in tickTimers", async () => { const { agentId } = await insertArchivedAgent(); @@ -205,4 +261,122 @@ describeEmbeddedPostgres("heartbeat archived-company guard", () => { .then((rows) => rows.filter((row) => row.agentId === agentId).length); expect(runCount).toBe(0); }); + + it("rejects explicit user invokes for invalid-org-chain agents", async () => { + const { childId } = await insertInvalidOrgChainAgent(); + + const heartbeat = heartbeatService(db); + + await expect(heartbeat.wakeup(childId, { + source: "on_demand", + triggerDetail: "manual", + requestedByActorType: "user", + requestedByActorId: "board-user", + })).rejects.toMatchObject({ + status: 409, + details: { + reason: "manager_terminated", + invalidOrgChain: true, + }, + }); + + const runCount = await db + .select() + .from(heartbeatRuns) + .then((rows) => rows.filter((row) => row.agentId === childId).length); + expect(runCount).toBe(0); + }); + + it("cancels existing queued runs for invalid-org-chain agents instead of starting them", async () => { + const { companyId, childId } = await insertInvalidOrgChainAgent(); + const wakeupRequestId = randomUUID(); + const runId = randomUUID(); + + await db.insert(agentWakeupRequests).values({ + id: wakeupRequestId, + companyId, + agentId: childId, + source: "assignment", + status: "queued", + }); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId: childId, + invocationSource: "assignment", + status: "queued", + wakeupRequestId, + }); + + const heartbeat = heartbeatService(db); + await heartbeat.resumeQueuedRuns(); + + const run = await db + .select({ + status: heartbeatRuns.status, + error: heartbeatRuns.error, + }) + .from(heartbeatRuns) + .then((rows) => rows.find((row) => row.status === "cancelled") ?? null); + expect(run).toMatchObject({ + status: "cancelled", + error: "Cancelled because the agent is not invokable: manager_terminated", + }); + + const wakeup = await db + .select({ + status: agentWakeupRequests.status, + error: agentWakeupRequests.error, + }) + .from(agentWakeupRequests) + .then((rows) => rows[0] ?? null); + expect(wakeup).toMatchObject({ + status: "cancelled", + error: "Cancelled because the agent is not invokable: manager_terminated", + }); + }); + + it("suppresses due scheduled retries for invalid-org-chain agents", async () => { + const { companyId, childId } = await insertInvalidOrgChainAgent(); + const wakeupRequestId = randomUUID(); + const runId = randomUUID(); + const now = new Date("2026-06-04T00:10:00Z"); + + await db.insert(agentWakeupRequests).values({ + id: wakeupRequestId, + companyId, + agentId: childId, + source: "automation", + status: "queued", + }); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId: childId, + invocationSource: "automation", + status: "scheduled_retry", + wakeupRequestId, + scheduledRetryAt: new Date("2026-06-04T00:00:00Z"), + scheduledRetryReason: "transient_failure", + scheduledRetryAttempt: 1, + }); + + const heartbeat = heartbeatService(db); + const promoted = await heartbeat.promoteDueScheduledRetries(now); + + expect(promoted).toEqual({ promoted: 0, runIds: [] }); + const run = await db + .select({ + status: heartbeatRuns.status, + errorCode: heartbeatRuns.errorCode, + error: heartbeatRuns.error, + }) + .from(heartbeatRuns) + .then((rows) => rows[0] ?? null); + expect(run).toMatchObject({ + status: "cancelled", + errorCode: "agent_not_invokable", + error: "Scheduled retry suppressed because the agent is not invokable", + }); + }); }); diff --git a/server/src/routes/agents.ts b/server/src/routes/agents.ts index c63a86eb573..386a8dde8ef 100644 --- a/server/src/routes/agents.ts +++ b/server/src/routes/agents.ts @@ -102,6 +102,7 @@ import { assertEnvironmentSelectionForCompany } from "./environment-selection.js import { recoveryService } from "../services/recovery/service.js"; import { resolveCoreTrustPreset } from "../services/trust-preset-resolver.js"; import { readObject } from "../lib/objects.js"; +import { listInvalidOrgChainDescendantIds } from "../services/agent-invokability.js"; const RUN_LOG_DEFAULT_LIMIT_BYTES = 256_000; const RUN_LOG_MAX_LIMIT_BYTES = 1024 * 1024; @@ -2918,7 +2919,21 @@ export function agentRoutes( return; } - await heartbeat.cancelActiveForAgent(id); + const companyAgentRows = await db + .select({ + id: agentsTable.id, + companyId: agentsTable.companyId, + name: agentsTable.name, + reportsTo: agentsTable.reportsTo, + status: agentsTable.status, + }) + .from(agentsTable) + .where(eq(agentsTable.companyId, agent.companyId)); + const invalidOrgChainDescendantIds = listInvalidOrgChainDescendantIds(id, companyAgentRows); + const cancellation = await heartbeat.cancelInvocationsForAgents( + [id, ...invalidOrgChainDescendantIds], + "Cancelled because the agent was terminated or became invalid-org-chain under a terminated manager", + ); await logActivity(db, { companyId: agent.companyId, @@ -2927,6 +2942,18 @@ export function agentRoutes( action: "agent.terminated", entityType: "agent", entityId: agent.id, + details: { + invalidOrgChain: { + descendantCount: invalidOrgChainDescendantIds.length, + descendantIds: invalidOrgChainDescendantIds, + state: invalidOrgChainDescendantIds.length > 0 ? "descendants_invalid_under_terminated_manager" : "none", + }, + cancellation: { + agentIds: cancellation.agentIds, + runsCancelled: cancellation.runsCancelled, + wakeupsCancelled: cancellation.wakeupsCancelled, + }, + }, }); res.json(agent); diff --git a/server/src/services/agent-invokability.ts b/server/src/services/agent-invokability.ts new file mode 100644 index 00000000000..237b05a089b --- /dev/null +++ b/server/src/services/agent-invokability.ts @@ -0,0 +1,182 @@ +import type { Db } from "@paperclipai/db"; +import { agents } from "@paperclipai/db"; +import { eq } from "drizzle-orm"; + +type AgentStatus = (typeof agents.$inferSelect)["status"]; + +export type AgentOrgRow = Pick< + typeof agents.$inferSelect, + "id" | "companyId" | "reportsTo" | "status" +>; + +export type AgentInvokabilityBlockReason = + | "missing" + | "paused" + | "terminated" + | "pending_approval" + | "manager_missing" + | "manager_company_mismatch" + | "manager_terminated" + | "reporting_cycle" + | "reporting_chain_too_deep"; + +export type AgentInvokability = + | { invokable: true } + | { + invokable: false; + reason: AgentInvokabilityBlockReason; + message: string; + details: Record; + invalidOrgChain: boolean; + }; + +const DIRECT_NON_INVOKABLE_STATUSES = new Set([ + "paused", + "terminated", + "pending_approval", +]); + +function blocked( + reason: AgentInvokabilityBlockReason, + message: string, + details: Record, + invalidOrgChain = false, +): AgentInvokability { + return { invokable: false, reason, message, details, invalidOrgChain }; +} + +function statusBlockReason(status: AgentStatus): AgentInvokabilityBlockReason | null { + if (status === "paused") return "paused"; + if (status === "terminated") return "terminated"; + if (status === "pending_approval") return "pending_approval"; + return null; +} + +export function evaluateAgentInvokability( + agent: AgentOrgRow | null | undefined, + companyAgents: AgentOrgRow[], +): AgentInvokability { + if (!agent) { + return blocked("missing", "Agent no longer exists", {}, false); + } + + const directStatusReason = statusBlockReason(agent.status); + if (directStatusReason) { + return blocked( + directStatusReason, + "Agent is not invokable in its current state", + { agentId: agent.id, agentStatus: agent.status }, + false, + ); + } + + const byId = new Map(companyAgents.map((row) => [row.id, row])); + const visited = new Set([agent.id]); + const reportingChainAgentIds: string[] = []; + let managerId = agent.reportsTo; + + while (managerId) { + if (visited.has(managerId)) { + return blocked( + "reporting_cycle", + "Agent is not invokable because its reporting chain is invalid", + { agentId: agent.id, managerId, reportingChainAgentIds }, + true, + ); + } + visited.add(managerId); + + if (reportingChainAgentIds.length >= 100) { + return blocked( + "reporting_chain_too_deep", + "Agent is not invokable because its reporting chain is invalid", + { agentId: agent.id, reportingChainAgentIds }, + true, + ); + } + + const manager = byId.get(managerId); + if (!manager) { + return blocked( + "manager_missing", + "Agent is not invokable because its reporting chain is invalid", + { agentId: agent.id, managerId, reportingChainAgentIds }, + true, + ); + } + + reportingChainAgentIds.push(manager.id); + if (manager.companyId !== agent.companyId) { + return blocked( + "manager_company_mismatch", + "Agent is not invokable because its reporting chain is invalid", + { agentId: agent.id, managerId: manager.id, managerCompanyId: manager.companyId }, + true, + ); + } + if (manager.status === "terminated") { + return blocked( + "manager_terminated", + "Agent is not invokable because its reporting chain is invalid", + { + agentId: agent.id, + managerId: manager.id, + managerStatus: manager.status, + reportingChainAgentIds, + }, + true, + ); + } + + managerId = manager.reportsTo; + } + + return { invokable: true }; +} + +export async function evaluateAgentInvokabilityFromDb( + db: Db, + agent: AgentOrgRow | null | undefined, +): Promise { + if (!agent) return evaluateAgentInvokability(agent, []); + const companyAgents = await db + .select({ + id: agents.id, + companyId: agents.companyId, + reportsTo: agents.reportsTo, + status: agents.status, + }) + .from(agents) + .where(eq(agents.companyId, agent.companyId)); + return evaluateAgentInvokability(agent, companyAgents); +} + +export function listInvalidOrgChainDescendantIds( + terminatedAgentId: string, + companyAgents: AgentOrgRow[], +): string[] { + const byManager = new Map(); + for (const row of companyAgents) { + const siblings = byManager.get(row.reportsTo ?? null) ?? []; + siblings.push(row); + byManager.set(row.reportsTo ?? null, siblings); + } + + const invalidDescendantIds: string[] = []; + const stack = [...(byManager.get(terminatedAgentId) ?? [])]; + const seen = new Set([terminatedAgentId]); + while (stack.length > 0) { + const current = stack.pop(); + if (!current || seen.has(current.id)) continue; + seen.add(current.id); + if (current.status !== "terminated") { + invalidDescendantIds.push(current.id); + } + stack.push(...(byManager.get(current.id) ?? [])); + } + return invalidDescendantIds; +} + +export function shouldCancelRunsForNonInvokableAgent(result: AgentInvokability) { + return !result.invokable && (result.reason === "terminated" || result.invalidOrgChain); +} diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 076f53c5eb7..90252246511 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -155,6 +155,11 @@ import { import { recoveryService } from "./recovery/service.js"; import { productivityReviewService } from "./productivity-review.js"; import { withAgentStartLock } from "./agent-start-lock.js"; +import { + evaluateAgentInvokability, + evaluateAgentInvokabilityFromDb, + shouldCancelRunsForNonInvokableAgent, +} from "./agent-invokability.js"; import { redactQuarantinedBodyForHigherTrust, sanitizeQuarantinedCommentForHigherTrust, @@ -3061,6 +3066,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) .then((rows) => rows[0] ?? null); } + async function getAgentInvokability(agent: typeof agents.$inferSelect | null | undefined) { + return evaluateAgentInvokabilityFromDb(db, agent); + } + async function getRun(runId: string, opts?: { unsafeFullResultJson?: boolean }) { const safeForLegacyEncoding = !opts?.unsafeFullResultJson && await hasUnsafeTextProjectionDatabase(); return db @@ -5146,6 +5155,22 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) agent: typeof agents.$inferSelect, issueId: string, ) { + const invokability = await getAgentInvokability(agent); + if (!invokability.invokable) { + await appendRunEvent(run, await nextRunEventSeq(run.id), { + eventType: "lifecycle", + stream: "system", + level: "warn", + message: "Missing-comment retry suppressed because the agent is not invokable", + payload: { + reason: invokability.reason, + invalidOrgChain: invokability.invalidOrgChain, + ...invokability.details, + }, + }); + return null; + } + const contextSnapshot = parseObject(run.contextSnapshot); const taskKey = deriveTaskKeyWithHeartbeatFallback(contextSnapshot, null); const sessionBefore = await resolveSessionBeforeForWakeup(agent, taskKey); @@ -5366,6 +5391,23 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) agent: typeof agents.$inferSelect, now: Date, ) { + const invokability = await getAgentInvokability(agent); + if (!invokability.invokable) { + await appendRunEvent(run, await nextRunEventSeq(run.id), { + eventType: "lifecycle", + stream: "system", + level: "warn", + message: "Process-loss retry suppressed because the agent is not invokable", + payload: { + reason: invokability.reason, + invalidOrgChain: invokability.invalidOrgChain, + ...invokability.details, + }, + }); + await releaseIssueExecutionAndPromote(run); + return null; + } + const contextSnapshot = parseObject(run.contextSnapshot); const issueId = readNonEmptyString(contextSnapshot.issueId); const taskKey = deriveTaskKeyWithHeartbeatFallback(contextSnapshot, null); @@ -5516,15 +5558,16 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }; } - if (agent.status === "paused" || agent.status === "terminated" || agent.status === "pending_approval") { + const agentInvokability = await getAgentInvokability(agent); + if (!agentInvokability.invokable) { return { allowed: false, reason: "Scheduled retry suppressed because the agent is not invokable", errorCode: "agent_not_invokable", issueId, details: { - agentId: agent.id, - agentStatus: agent.status, + ...agentInvokability.details, + invalidOrgChain: agentInvokability.invalidOrgChain, }, }; } @@ -5896,6 +5939,33 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) maxAttempts, }; } + + const invokability = await getAgentInvokability(agent); + if (!invokability.invokable) { + const contextSnapshot = parseObject(run.contextSnapshot); + const issueId = readNonEmptyString(contextSnapshot.issueId); + await appendRunEvent(run, await nextRunEventSeq(run.id), { + eventType: "lifecycle", + stream: "system", + level: "warn", + message: "Scheduled retry suppressed because the agent is not invokable", + payload: { + retryReason, + scheduledRetryAttempt: nextAttempt, + maxAttempts, + reason: invokability.reason, + invalidOrgChain: invokability.invalidOrgChain, + ...invokability.details, + }, + }); + return { + outcome: "not_scheduled" as const, + reason: "Scheduled retry suppressed because the agent is not invokable", + errorCode: "agent_not_invokable" as const, + issueId, + }; + } + const schedule = transientRetryNotBefore && transientRetryNotBefore.getTime() > baseSchedule.dueAt.getTime() ? { @@ -6521,8 +6591,9 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) await cancelRunInternal(run.id, "Cancelled because the agent no longer exists"); return null; } - if (agent.status === "paused" || agent.status === "terminated" || agent.status === "pending_approval") { - await cancelRunInternal(run.id, "Cancelled because the agent is not invokable"); + const invokability = await getAgentInvokability(agent); + if (!invokability.invokable) { + await cancelRunInternal(run.id, `Cancelled because the agent is not invokable: ${invokability.reason}`); return null; } @@ -7433,7 +7504,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return withAgentStartLock(agentId, async () => { const agent = await getAgent(agentId); if (!agent) return []; - if (agent.status === "paused" || agent.status === "terminated" || agent.status === "pending_approval") { + const invokability = await getAgentInvokability(agent); + if (!invokability.invokable) { + if (shouldCancelRunsForNonInvokableAgent(invokability)) { + await cancelActiveForAgentInternal(agentId, `Cancelled because the agent is not invokable: ${invokability.reason}`); + } return []; } const policy = parseHeartbeatPolicy(agent); @@ -9278,13 +9353,24 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) .where(eq(agents.id, deferred.agentId)) .then((rows) => rows[0] ?? null); - if ( - !deferredAgent || - deferredAgent.companyId !== issue.companyId || - deferredAgent.status === "paused" || - deferredAgent.status === "terminated" || - deferredAgent.status === "pending_approval" - ) { + const companyAgents = deferredAgent + ? await tx + .select({ + id: agents.id, + companyId: agents.companyId, + name: agents.name, + reportsTo: agents.reportsTo, + status: agents.status, + }) + .from(agents) + .where(eq(agents.companyId, issue.companyId)) + : []; + const deferredInvokability = + deferredAgent?.companyId === issue.companyId + ? evaluateAgentInvokability(deferredAgent, companyAgents) + : evaluateAgentInvokability(null, companyAgents); + + if (!deferredAgent || deferredAgent.companyId !== issue.companyId || !deferredInvokability.invokable) { await tx .update(agentWakeupRequests) .set({ @@ -9767,12 +9853,19 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }); } - if ( - agent.status === "paused" || - agent.status === "terminated" || - agent.status === "pending_approval" - ) { - throw conflict("Agent is not invokable in its current state", { status: agent.status }); + const invokability = await getAgentInvokability(agent); + if (!invokability.invokable) { + if (opts.requestedByActorType !== "user") { + await writeSkippedRequest("agent.not_invokable", { + error: invokability.message, + }); + } + throw conflict(invokability.message, { + status: agent.status, + reason: invokability.reason, + invalidOrgChain: invokability.invalidOrgChain, + ...invokability.details, + }); } const policy = parseHeartbeatPolicy(agent); @@ -10576,6 +10669,52 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return runs.length; } + async function cancelPendingWakeupsForAgentsInternal(agentIds: string[], reason: string) { + const uniqueAgentIds = [...new Set(agentIds)].filter((agentId) => agentId.length > 0); + if (uniqueAgentIds.length === 0) return 0; + + const now = new Date(); + const wakeupIds = await db + .select({ id: agentWakeupRequests.id }) + .from(agentWakeupRequests) + .where( + and( + inArray(agentWakeupRequests.agentId, uniqueAgentIds), + inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]), + sql`${agentWakeupRequests.runId} is null`, + ), + ) + .then((rows) => rows.map((row) => row.id)); + + if (wakeupIds.length === 0) return 0; + + await db + .update(agentWakeupRequests) + .set({ + status: "cancelled", + finishedAt: now, + error: reason, + updatedAt: now, + }) + .where(inArray(agentWakeupRequests.id, wakeupIds)); + + return wakeupIds.length; + } + + async function cancelInvocationsForAgentsInternal(agentIds: string[], reason: string) { + const uniqueAgentIds = [...new Set(agentIds)].filter((agentId) => agentId.length > 0); + let runsCancelled = 0; + for (const agentId of uniqueAgentIds) { + runsCancelled += await cancelActiveForAgentInternal(agentId, reason); + } + const wakeupsCancelled = await cancelPendingWakeupsForAgentsInternal(uniqueAgentIds, reason); + return { + agentIds: uniqueAgentIds, + runsCancelled, + wakeupsCancelled, + }; + } + async function cancelBudgetScopeWork(scope: BudgetEnforcementScope) { if (scope.scopeType === "agent") { await cancelActiveForAgentInternal(scope.scopeId, "Cancelled due to budget pause"); @@ -10882,7 +11021,17 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) let skipped = 0; for (const agent of allAgents) { - if (agent.status === "paused" || agent.status === "terminated" || agent.status === "pending_approval") continue; + const companyAgents = allAgents + .filter((candidate) => candidate.companyId === agent.companyId) + .map((candidate) => ({ + id: candidate.id, + companyId: candidate.companyId, + name: candidate.name, + reportsTo: candidate.reportsTo, + status: candidate.status, + })); + const invokability = evaluateAgentInvokability(agent, companyAgents); + if (!invokability.invokable) continue; const policy = parseHeartbeatPolicy(agent); if (!policy.enabled || policy.intervalSec <= 0) continue; @@ -10920,6 +11069,9 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) cancelActiveForAgent: (agentId: string, reason?: string) => cancelActiveForAgentInternal(agentId, reason), + cancelInvocationsForAgents: (agentIds: string[], reason: string) => + cancelInvocationsForAgentsInternal(agentIds, reason), + cancelBudgetScopeWork, getRunIssueSummary: async (runId: string) => { diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 468d76c9c8f..c9f4b4fdf90 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -36,6 +36,7 @@ import { instanceSettingsService } from "../instance-settings.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { issueTreeControlService } from "../issue-tree-control.js"; import { issueService } from "../issues.js"; +import { evaluateAgentInvokabilityFromDb } from "../agent-invokability.js"; import { getRunLogStore } from "../run-log-store.js"; import { DEFAULT_MAX_SUCCESSFUL_RUN_HANDOFF_ATTEMPTS, @@ -327,10 +328,6 @@ function unwrapDatabaseConflictError(error: unknown) { }; } -function isAgentInvokable(agent: typeof agents.$inferSelect | null | undefined) { - return Boolean(agent && !["paused", "terminated", "pending_approval"].includes(agent.status)); -} - function isStrandedIssueRecoveryIssue(issue: Pick) { return isStrandedIssueRecoveryOriginKind(issue.originKind); } @@ -466,6 +463,10 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) return db.select().from(agents).where(eq(agents.id, agentId)).then((rows) => rows[0] ?? null); } + async function isAgentInvokable(agent: typeof agents.$inferSelect | null | undefined) { + return (await evaluateAgentInvokabilityFromDb(db, agent)).invokable; + } + async function getLatestIssueRun(companyId: string, issueId: string): Promise { return db .select({ @@ -697,7 +698,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) continue; } const creatorAgent = await getAgent(creatorAgentId); - if (!creatorAgent || creatorAgent.companyId !== candidate.companyId || !isAgentInvokable(creatorAgent)) { + if (!creatorAgent || creatorAgent.companyId !== candidate.companyId || !(await isAgentInvokable(creatorAgent))) { skipped += 1; continue; } @@ -1249,7 +1250,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) issueId: input.sourceIssue?.id ?? null, projectId: input.sourceIssue?.projectId ?? null, }); - if (isAgentInvokable(candidate) && !budgetBlock) return candidate.id; + if ((await isAgentInvokable(candidate)) && !budgetBlock) return candidate.id; } return null; @@ -1872,7 +1873,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) issueId: issue.id, projectId: issue.projectId, }); - if (isAgentInvokable(candidate) && !budgetBlock) return candidate.id; + if ((await isAgentInvokable(candidate)) && !budgetBlock) return candidate.id; } return null; @@ -2486,7 +2487,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) } const agent = await getAgent(agentId); - if (!agent || agent.companyId !== issue.companyId || !isAgentInvokable(agent)) { + if (!agent || agent.companyId !== issue.companyId || !(await isAgentInvokable(agent))) { result.skipped += 1; continue; }