From 81be1d8a95629a3fc0987f9d59b22dd9cc289540 Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 22 Apr 2026 13:57:20 -0500 Subject: [PATCH] Gate liveness recovery issue creation --- .../0065_liveness_recovery_dedupe.sql | 13 + packages/db/src/schema/issues.ts | 16 + packages/db/src/test-embedded-postgres.ts | 146 ++++--- packages/shared/src/types/instance.ts | 1 + packages/shared/src/validators/instance.ts | 1 + server/src/__tests__/README.md | 10 + server/src/__tests__/feedback-service.test.ts | 85 +--- .../heartbeat-comment-wake-batching.test.ts | 88 +--- ...eartbeat-issue-liveness-escalation.test.ts | 308 +++++++++++++- .../instance-settings-routes.test.ts | 27 +- server/src/__tests__/issue-liveness.test.ts | 52 +++ server/src/services/heartbeat.ts | 391 ++++++++++++++++-- server/src/services/instance-settings.ts | 2 + server/src/services/issue-liveness.ts | 154 +++++-- ui/src/pages/InstanceExperimentalSettings.tsx | 24 ++ 15 files changed, 1022 insertions(+), 296 deletions(-) create mode 100644 packages/db/src/migrations/0065_liveness_recovery_dedupe.sql create mode 100644 server/src/__tests__/README.md diff --git a/packages/db/src/migrations/0065_liveness_recovery_dedupe.sql b/packages/db/src/migrations/0065_liveness_recovery_dedupe.sql new file mode 100644 index 00000000000..bcd4cbc8174 --- /dev/null +++ b/packages/db/src/migrations/0065_liveness_recovery_dedupe.sql @@ -0,0 +1,13 @@ +CREATE UNIQUE INDEX IF NOT EXISTS "issues_active_liveness_recovery_incident_uq" + ON "issues" USING btree ("company_id","origin_kind","origin_id") + WHERE "origin_kind" = 'harness_liveness_escalation' + AND "origin_id" IS NOT NULL + AND "hidden_at" IS NULL + AND "status" NOT IN ('done', 'cancelled'); +--> statement-breakpoint +CREATE UNIQUE INDEX IF NOT EXISTS "issues_active_liveness_recovery_leaf_uq" + ON "issues" USING btree ("company_id","origin_kind","origin_fingerprint") + WHERE "origin_kind" = 'harness_liveness_escalation' + AND "origin_fingerprint" <> 'default' + AND "hidden_at" IS NULL + AND "status" NOT IN ('done', 'cancelled'); diff --git a/packages/db/src/schema/issues.ts b/packages/db/src/schema/issues.ts index d4dae91a83c..35ae509515f 100644 --- a/packages/db/src/schema/issues.ts +++ b/packages/db/src/schema/issues.ts @@ -91,5 +91,21 @@ export const issues = pgTable( and ${table.executionRunId} is not null and ${table.status} in ('backlog', 'todo', 'in_progress', 'in_review', 'blocked')`, ), + activeLivenessRecoveryIncidentIdx: uniqueIndex("issues_active_liveness_recovery_incident_uq") + .on(table.companyId, table.originKind, table.originId) + .where( + sql`${table.originKind} = 'harness_liveness_escalation' + and ${table.originId} is not null + and ${table.hiddenAt} is null + and ${table.status} not in ('done', 'cancelled')`, + ), + activeLivenessRecoveryLeafIdx: uniqueIndex("issues_active_liveness_recovery_leaf_uq") + .on(table.companyId, table.originKind, table.originFingerprint) + .where( + sql`${table.originKind} = 'harness_liveness_escalation' + and ${table.originFingerprint} <> 'default' + and ${table.hiddenAt} is null + and ${table.status} not in ('done', 'cancelled')`, + ), }), ); diff --git a/packages/db/src/test-embedded-postgres.ts b/packages/db/src/test-embedded-postgres.ts index 04fa642d9ec..11ab966920a 100644 --- a/packages/db/src/test-embedded-postgres.ts +++ b/packages/db/src/test-embedded-postgres.ts @@ -33,77 +33,56 @@ export type EmbeddedPostgresTestDatabase = { let embeddedPostgresSupportPromise: Promise | null = null; +const DEFAULT_PAPERCLIP_EMBEDDED_POSTGRES_PORT = 54329; + +function getReservedTestPorts(): Set { + const configuredPorts = [ + DEFAULT_PAPERCLIP_EMBEDDED_POSTGRES_PORT, + Number.parseInt(process.env.PAPERCLIP_EMBEDDED_POSTGRES_PORT ?? "", 10), + ...String(process.env.PAPERCLIP_TEST_POSTGRES_RESERVED_PORTS ?? "") + .split(",") + .map((value) => Number.parseInt(value.trim(), 10)), + ]; + return new Set(configuredPorts.filter((port) => Number.isInteger(port) && port > 0 && port <= 65535)); +} + async function getEmbeddedPostgresCtor(): Promise { const mod = await import("embedded-postgres"); return mod.default as EmbeddedPostgresCtor; } async function getAvailablePort(): Promise { - return await new Promise((resolve, reject) => { - const server = net.createServer(); - server.unref(); - server.on("error", reject); - server.listen(0, "127.0.0.1", () => { - const address = server.address(); - if (!address || typeof address === "string") { - server.close(() => reject(new Error("Failed to allocate test port"))); - return; - } - const { port } = address; - server.close((error) => { - if (error) reject(error); - else resolve(port); + const reservedPorts = getReservedTestPorts(); + for (let attempt = 0; attempt < 20; attempt += 1) { + const port = await new Promise((resolve, reject) => { + const server = net.createServer(); + server.unref(); + server.on("error", reject); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + if (!address || typeof address === "string") { + server.close(() => reject(new Error("Failed to allocate test port"))); + return; + } + const { port } = address; + server.close((error) => { + if (error) reject(error); + else resolve(port); + }); }); }); - }); -} -function formatEmbeddedPostgresError(error: unknown): string { - if (error instanceof Error && error.message.length > 0) return error.message; - if (typeof error === "string" && error.length > 0) return error; - return "embedded Postgres startup failed"; -} - -async function probeEmbeddedPostgresSupport(): Promise { - const dataDir = fs.mkdtempSync(path.join(os.tmpdir(), "paperclip-embedded-postgres-probe-")); - const port = await getAvailablePort(); - const EmbeddedPostgres = await getEmbeddedPostgresCtor(); - const instance = new EmbeddedPostgres({ - databaseDir: dataDir, - user: "paperclip", - password: "paperclip", - port, - persistent: true, - initdbFlags: ["--encoding=UTF8", "--locale=C", "--lc-messages=C"], - onLog: () => {}, - onError: () => {}, - }); - - try { - await instance.initialise(); - await instance.start(); - return { supported: true }; - } catch (error) { - return { - supported: false, - reason: formatEmbeddedPostgresError(error), - }; - } finally { - await instance.stop().catch(() => {}); - fs.rmSync(dataDir, { recursive: true, force: true }); + if (!reservedPorts.has(port)) return port; } + + throw new Error( + `Failed to allocate embedded Postgres test port outside reserved Paperclip ports: ${[ + ...reservedPorts, + ].join(", ")}`, + ); } -export async function getEmbeddedPostgresTestSupport(): Promise { - if (!embeddedPostgresSupportPromise) { - embeddedPostgresSupportPromise = probeEmbeddedPostgresSupport(); - } - return await embeddedPostgresSupportPromise; -} - -export async function startEmbeddedPostgresTestDatabase( - tempDirPrefix: string, -): Promise { +async function createEmbeddedPostgresTestInstance(tempDirPrefix: string) { const dataDir = fs.mkdtempSync(path.join(os.tmpdir(), tempDirPrefix)); const port = await getAvailablePort(); const EmbeddedPostgres = await getEmbeddedPostgresCtor(); @@ -118,6 +97,51 @@ export async function startEmbeddedPostgresTestDatabase( onError: () => {}, }); + return { dataDir, port, instance }; +} + +function cleanupEmbeddedPostgresTestDirs(dataDir: string) { + fs.rmSync(dataDir, { recursive: true, force: true }); +} + +function formatEmbeddedPostgresError(error: unknown): string { + if (error instanceof Error && error.message.length > 0) return error.message; + if (typeof error === "string" && error.length > 0) return error; + return "embedded Postgres startup failed"; +} + +async function probeEmbeddedPostgresSupport(): Promise { + const { dataDir, instance } = await createEmbeddedPostgresTestInstance( + "paperclip-embedded-postgres-probe-", + ); + + try { + await instance.initialise(); + await instance.start(); + return { supported: true }; + } catch (error) { + return { + supported: false, + reason: formatEmbeddedPostgresError(error), + }; + } finally { + await instance.stop().catch(() => {}); + cleanupEmbeddedPostgresTestDirs(dataDir); + } +} + +export async function getEmbeddedPostgresTestSupport(): Promise { + if (!embeddedPostgresSupportPromise) { + embeddedPostgresSupportPromise = probeEmbeddedPostgresSupport(); + } + return await embeddedPostgresSupportPromise; +} + +export async function startEmbeddedPostgresTestDatabase( + tempDirPrefix: string, +): Promise { + const { dataDir, port, instance } = await createEmbeddedPostgresTestInstance(tempDirPrefix); + try { await instance.initialise(); await instance.start(); @@ -131,12 +155,12 @@ export async function startEmbeddedPostgresTestDatabase( connectionString, cleanup: async () => { await instance.stop().catch(() => {}); - fs.rmSync(dataDir, { recursive: true, force: true }); + cleanupEmbeddedPostgresTestDirs(dataDir); }, }; } catch (error) { await instance.stop().catch(() => {}); - fs.rmSync(dataDir, { recursive: true, force: true }); + cleanupEmbeddedPostgresTestDirs(dataDir); throw new Error( `Failed to start embedded PostgreSQL test database: ${formatEmbeddedPostgresError(error)}`, ); diff --git a/packages/shared/src/types/instance.ts b/packages/shared/src/types/instance.ts index 41686664949..73e690ef96d 100644 --- a/packages/shared/src/types/instance.ts +++ b/packages/shared/src/types/instance.ts @@ -27,6 +27,7 @@ export interface InstanceExperimentalSettings { enableEnvironments: boolean; enableIsolatedWorkspaces: boolean; autoRestartDevServerWhenIdle: boolean; + enableIssueGraphLivenessAutoRecovery: boolean; } export interface InstanceSettings { diff --git a/packages/shared/src/validators/instance.ts b/packages/shared/src/validators/instance.ts index f7638a88118..94d78226cee 100644 --- a/packages/shared/src/validators/instance.ts +++ b/packages/shared/src/validators/instance.ts @@ -36,6 +36,7 @@ export const instanceExperimentalSettingsSchema = z.object({ enableEnvironments: z.boolean().default(false), enableIsolatedWorkspaces: z.boolean().default(false), autoRestartDevServerWhenIdle: z.boolean().default(false), + enableIssueGraphLivenessAutoRecovery: z.boolean().default(false), }).strict(); export const patchInstanceExperimentalSettingsSchema = instanceExperimentalSettingsSchema.partial(); diff --git a/server/src/__tests__/README.md b/server/src/__tests__/README.md new file mode 100644 index 00000000000..86b8c2e2ee6 --- /dev/null +++ b/server/src/__tests__/README.md @@ -0,0 +1,10 @@ +# Server Tests + +Server tests that need a real PostgreSQL process must use +`./helpers/embedded-postgres.ts` instead of constructing `embedded-postgres` +directly. + +The shared helper creates a throwaway data directory and a reserved-safe +loopback port for each test database. This protects the live Paperclip +control-plane Postgres from server vitest runs; see PAP-2033 for the incident +that introduced this guard. diff --git a/server/src/__tests__/feedback-service.test.ts b/server/src/__tests__/feedback-service.test.ts index 7a3a36d8ba0..8a979d84c52 100644 --- a/server/src/__tests__/feedback-service.test.ts +++ b/server/src/__tests__/feedback-service.test.ts @@ -1,6 +1,5 @@ import { randomUUID } from "node:crypto"; import fs from "node:fs"; -import net from "node:net"; import os from "node:os"; import path from "node:path"; import { eq } from "drizzle-orm"; @@ -8,14 +7,12 @@ import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest import { writePaperclipSkillSyncPreference } from "@paperclipai/adapter-utils/server-utils"; import { agents, - applyPendingMigrations, companies, companySkills, costEvents, createDb, documents, documentRevisions, - ensurePostgresDatabase, feedbackExports, feedbackVotes, heartbeatRuns, @@ -25,72 +22,7 @@ import { issues, } from "@paperclipai/db"; import { feedbackService } from "../services/feedback.ts"; - -type EmbeddedPostgresInstance = { - initialise(): Promise; - start(): Promise; - stop(): Promise; -}; - -type EmbeddedPostgresCtor = new (opts: { - databaseDir: string; - user: string; - password: string; - port: number; - persistent: boolean; - initdbFlags?: string[]; - onLog?: (message: unknown) => void; - onError?: (message: unknown) => void; -}) => EmbeddedPostgresInstance; - -async function getEmbeddedPostgresCtor(): Promise { - const mod = await import("embedded-postgres"); - return mod.default as EmbeddedPostgresCtor; -} - -async function getAvailablePort(): Promise { - return await new Promise((resolve, reject) => { - const server = net.createServer(); - server.unref(); - server.on("error", reject); - server.listen(0, "127.0.0.1", () => { - const address = server.address(); - if (!address || typeof address === "string") { - server.close(() => reject(new Error("Failed to allocate test port"))); - return; - } - const { port } = address; - server.close((error) => { - if (error) reject(error); - else resolve(port); - }); - }); - }); -} - -async function startTempDatabase() { - const dataDir = fs.mkdtempSync(path.join(os.tmpdir(), "paperclip-feedback-service-")); - const port = await getAvailablePort(); - const EmbeddedPostgres = await getEmbeddedPostgresCtor(); - const instance = new EmbeddedPostgres({ - databaseDir: dataDir, - user: "paperclip", - password: "paperclip", - port, - persistent: true, - initdbFlags: ["--encoding=UTF8", "--locale=C", "--lc-messages=C"], - onLog: () => {}, - onError: () => {}, - }); - await instance.initialise(); - await instance.start(); - - const adminConnectionString = `postgres://paperclip:paperclip@127.0.0.1:${port}/postgres`; - await ensurePostgresDatabase(adminConnectionString, "paperclip"); - const connectionString = `postgres://paperclip:paperclip@127.0.0.1:${port}/paperclip`; - await applyPendingMigrations(connectionString); - return { connectionString, dataDir, instance }; -} +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.ts"; async function closeDbClient(db: ReturnType | undefined) { await db?.$client?.end?.({ timeout: 0 }); @@ -99,17 +31,15 @@ async function closeDbClient(db: ReturnType | undefined) { describe("feedbackService.saveIssueVote", () => { let db!: ReturnType; let svc!: ReturnType; - let instance: EmbeddedPostgresInstance | null = null; - let dataDir = ""; + let tempDb: Awaited> | null = null; let tempDirs: string[] = []; beforeAll(async () => { - const started = await startTempDatabase(); + const started = await startEmbeddedPostgresTestDatabase("paperclip-feedback-service-"); db = createDb(started.connectionString); svc = feedbackService(db); - instance = started.instance; - dataDir = started.dataDir; - }, 20_000); + tempDb = started; + }, 120_000); afterEach(async () => { await db.delete(feedbackExports); @@ -134,10 +64,7 @@ describe("feedbackService.saveIssueVote", () => { afterAll(async () => { await closeDbClient(db); - await instance?.stop(); - if (dataDir) { - fs.rmSync(dataDir, { recursive: true, force: true }); - } + await tempDb?.cleanup(); }); async function seedIssueWithAgentComment() { diff --git a/server/src/__tests__/heartbeat-comment-wake-batching.test.ts b/server/src/__tests__/heartbeat-comment-wake-batching.test.ts index 4c3887d3d3e..57ce9a054c8 100644 --- a/server/src/__tests__/heartbeat-comment-wake-batching.test.ts +++ b/server/src/__tests__/heartbeat-comment-wake-batching.test.ts @@ -1,8 +1,4 @@ import { randomUUID } from "node:crypto"; -import fs from "node:fs"; -import net from "node:net"; -import os from "node:os"; -import path from "node:path"; import { createServer } from "node:http"; import { and, asc, eq } from "drizzle-orm"; import { WebSocketServer } from "ws"; @@ -10,81 +6,14 @@ import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { agents, agentWakeupRequests, - applyPendingMigrations, companies, createDb, - ensurePostgresDatabase, heartbeatRuns, issueComments, issues, } from "@paperclipai/db"; import { heartbeatService } from "../services/heartbeat.ts"; - -type EmbeddedPostgresInstance = { - initialise(): Promise; - start(): Promise; - stop(): Promise; -}; - -type EmbeddedPostgresCtor = new (opts: { - databaseDir: string; - user: string; - password: string; - port: number; - persistent: boolean; - initdbFlags?: string[]; - onLog?: (message: unknown) => void; - onError?: (message: unknown) => void; -}) => EmbeddedPostgresInstance; - -async function getEmbeddedPostgresCtor(): Promise { - const mod = await import("embedded-postgres"); - return mod.default as EmbeddedPostgresCtor; -} - -async function getAvailablePort(): Promise { - return await new Promise((resolve, reject) => { - const server = net.createServer(); - server.unref(); - server.on("error", reject); - server.listen(0, "127.0.0.1", () => { - const address = server.address(); - if (!address || typeof address === "string") { - server.close(() => reject(new Error("Failed to allocate test port"))); - return; - } - const { port } = address; - server.close((error) => { - if (error) reject(error); - else resolve(port); - }); - }); - }); -} - -async function startTempDatabase() { - const dataDir = fs.mkdtempSync(path.join(os.tmpdir(), "paperclip-heartbeat-comment-wake-")); - const port = await getAvailablePort(); - const EmbeddedPostgres = await getEmbeddedPostgresCtor(); - const instance = new EmbeddedPostgres({ - databaseDir: dataDir, - user: "paperclip", - password: "paperclip", - port, - persistent: true, - initdbFlags: ["--encoding=UTF8", "--locale=C", "--lc-messages=C"], - onLog: () => {}, - onError: () => {}, - }); - await instance.initialise(); - await instance.start(); - - const adminConnectionString = `postgres://paperclip:paperclip@127.0.0.1:${port}/postgres`; - await ensurePostgresDatabase(adminConnectionString, "paperclip"); - const connectionString = `postgres://paperclip:paperclip@127.0.0.1:${port}/paperclip`; - await applyPendingMigrations(connectionString); - return { connectionString, instance, dataDir }; -} +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.ts"; async function waitFor(condition: () => boolean | Promise, timeoutMs = 10_000, intervalMs = 50) { const startedAt = Date.now(); @@ -218,22 +147,17 @@ async function createControlledGatewayServer() { describe("heartbeat comment wake batching", () => { let db!: ReturnType; - let instance: EmbeddedPostgresInstance | null = null; - let dataDir = ""; + let tempDb: Awaited> | null = null; beforeAll(async () => { - const started = await startTempDatabase(); + const started = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-comment-wake-"); db = createDb(started.connectionString); - instance = started.instance; - dataDir = started.dataDir; - }, 45_000); + tempDb = started; + }, 120_000); afterAll(async () => { await closeDbClient(db); - await instance?.stop(); - if (dataDir) { - fs.rmSync(dataDir, { recursive: true, force: true }); - } + await tempDb?.cleanup(); }); it("defers approval-approved wakes for a running issue so the assignee resumes after the run", async () => { diff --git a/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts b/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts index 5e3f80abf45..e6c9c807d02 100644 --- a/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts +++ b/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts @@ -4,13 +4,15 @@ import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest import { activityLog, agents, - agentWakeupRequests, companies, createDb, + executionWorkspaces, heartbeatRuns, issueComments, issueRelations, issues, + projects, + projectWorkspaces, } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, @@ -55,6 +57,7 @@ vi.mock("../adapters/index.ts", async () => { }); import { heartbeatService } from "../services/heartbeat.ts"; +import { instanceSettingsService } from "../services/instance-settings.ts"; import { runningProcesses } from "../adapters/index.ts"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); @@ -94,13 +97,23 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { } await new Promise((resolve) => setTimeout(resolve, 50)); await db.execute(sql.raw(`TRUNCATE TABLE "companies" CASCADE`)); + await instanceSettingsService(db).updateExperimental({ + enableIssueGraphLivenessAutoRecovery: false, + enableIsolatedWorkspaces: false, + }); }); afterAll(async () => { await tempDb?.cleanup(); }); - async function seedBlockedChain() { + async function enableAutoRecovery() { + await instanceSettingsService(db).updateExperimental({ + enableIssueGraphLivenessAutoRecovery: true, + }); + } + + async function seedBlockedChain(opts: { stale?: boolean } = {}) { const companyId = randomUUID(); const managerId = randomUUID(); const coderId = randomUUID(); @@ -124,7 +137,7 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { status: "idle", adapterType: "codex_local", adapterConfig: {}, - runtimeConfig: {}, + runtimeConfig: { heartbeat: { wakeOnDemand: false } }, permissions: {}, }, { @@ -136,11 +149,14 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { reportsTo: managerId, adapterType: "codex_local", adapterConfig: {}, - runtimeConfig: {}, + runtimeConfig: { heartbeat: { wakeOnDemand: false } }, permissions: {}, }, ]); + const issueTimestamp = opts.stale === false + ? new Date() + : new Date(Date.now() - 25 * 60 * 60 * 1000); await db.insert(issues).values([ { id: blockedIssueId, @@ -151,6 +167,8 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { assigneeAgentId: coderId, issueNumber: 1, identifier: `${issuePrefix}-1`, + createdAt: issueTimestamp, + updatedAt: issueTimestamp, }, { id: blockerIssueId, @@ -160,6 +178,8 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { priority: "medium", issueNumber: 2, identifier: `${issuePrefix}-2`, + createdAt: issueTimestamp, + updatedAt: issueTimestamp, }, ]); @@ -173,7 +193,64 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { return { companyId, managerId, blockedIssueId, blockerIssueId }; } - it("creates one manager escalation, preserves blockers, and wakes the assignee", async () => { + it("keeps liveness findings advisory when auto recovery is disabled", async () => { + const { companyId } = await seedBlockedChain(); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileIssueGraphLiveness(); + + expect(result.findings).toBe(1); + expect(result.autoRecoveryEnabled).toBe(false); + expect(result.escalationsCreated).toBe(0); + expect(result.skippedAutoRecoveryDisabled).toBe(1); + + const escalations = await db + .select() + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.originKind, "harness_liveness_escalation"))); + expect(escalations).toHaveLength(0); + }); + + it("does not create recovery issues until the dependency path is stale for 24 hours", async () => { + await enableAutoRecovery(); + const { companyId } = await seedBlockedChain({ stale: false }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileIssueGraphLiveness(); + + expect(result.findings).toBe(1); + expect(result.escalationsCreated).toBe(0); + expect(result.skippedAutoRecoveryTooYoung).toBe(1); + + const escalations = await db + .select() + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.originKind, "harness_liveness_escalation"))); + expect(escalations).toHaveLength(0); + }); + + it("treats an active executionRunId on the leaf blocker as a live execution path", async () => { + await enableAutoRecovery(); + const { companyId, managerId, blockedIssueId, blockerIssueId } = await seedBlockedChain(); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId: managerId, + status: "running", + contextSnapshot: { issueId: blockedIssueId }, + }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, blockerIssueId)); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileIssueGraphLiveness(); + + expect(result.findings).toBe(0); + expect(result.escalationsCreated).toBe(0); + }); + + it("creates one manager escalation, preserves blockers, and records owner selection", async () => { + await enableAutoRecovery(); const { companyId, managerId, blockedIssueId, blockerIssueId } = await seedBlockedChain(); const heartbeat = heartbeatService(db); @@ -182,7 +259,6 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { expect(first.escalationsCreated).toBe(1); expect(second.escalationsCreated).toBe(0); - expect(second.existingEscalations).toBe(1); const escalations = await db .select() @@ -195,9 +271,15 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { ); expect(escalations).toHaveLength(1); expect(escalations[0]).toMatchObject({ - parentId: blockedIssueId, + parentId: blockerIssueId, assigneeAgentId: managerId, status: expect.stringMatching(/^(todo|in_progress|done)$/), + originFingerprint: [ + "harness_liveness_leaf", + companyId, + "blocked_by_unassigned_issue", + blockerIssueId, + ].join(":"), }); const blockers = await db @@ -213,15 +295,217 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { expect(comments[0]?.body).toContain("harness-level liveness incident"); expect(comments[0]?.body).toContain(escalations[0]?.identifier ?? escalations[0]!.id); - const wakes = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, managerId)); - expect(wakes.some((wake) => wake.reason === "issue_assigned")).toBe(true); - const events = await db.select().from(activityLog).where(eq(activityLog.companyId, companyId)); - expect(events.some((event) => event.action === "issue.harness_liveness_escalation_created")).toBe(true); + const createdEvent = events.find((event) => event.action === "issue.harness_liveness_escalation_created"); + expect(createdEvent).toBeTruthy(); + expect(createdEvent?.details).toMatchObject({ + recoveryIssueId: blockerIssueId, + ownerSelection: { + selectedAgentId: managerId, + selectedReason: "root_agent", + selectedSourceIssueId: blockerIssueId, + }, + workspaceSelection: { + reuseRecoveryExecutionWorkspace: false, + inheritedExecutionWorkspaceFromIssueId: null, + projectWorkspaceSourceIssueId: blockerIssueId, + }, + }); expect(events.some((event) => event.action === "issue.blockers.updated")).toBe(true); }); + it("parents recovery under the leaf blocker without inheriting dependent or blocker execution state for manager-owned recovery", async () => { + await enableAutoRecovery(); + await instanceSettingsService(db).updateExperimental({ enableIsolatedWorkspaces: true }); + + const companyId = randomUUID(); + const managerId = randomUUID(); + const blockedIssueId = randomUUID(); + const blockerIssueId = randomUUID(); + const dependentProjectId = randomUUID(); + const blockerProjectId = randomUUID(); + const dependentProjectWorkspaceId = randomUUID(); + const blockerProjectWorkspaceId = randomUUID(); + const dependentExecutionWorkspaceId = randomUUID(); + const blockerExecutionWorkspaceId = randomUUID(); + const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + const issueTimestamp = new Date(Date.now() - 25 * 60 * 60 * 1000); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: managerId, + companyId, + name: "Root Operator", + role: "operator", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: { heartbeat: { wakeOnDemand: false } }, + permissions: {}, + }); + await db.insert(projects).values([ + { + id: dependentProjectId, + companyId, + name: "Dependent workspace project", + status: "in_progress", + }, + { + id: blockerProjectId, + companyId, + name: "Blocker workspace project", + status: "in_progress", + }, + ]); + await db.insert(projectWorkspaces).values([ + { + id: dependentProjectWorkspaceId, + companyId, + projectId: dependentProjectId, + name: "Dependent primary", + }, + { + id: blockerProjectWorkspaceId, + companyId, + projectId: blockerProjectId, + name: "Blocker primary", + }, + ]); + await db.insert(executionWorkspaces).values([ + { + id: dependentExecutionWorkspaceId, + companyId, + projectId: dependentProjectId, + projectWorkspaceId: dependentProjectWorkspaceId, + mode: "operator_branch", + strategyType: "git_worktree", + name: "Dependent branch", + status: "active", + providerType: "git_worktree", + }, + { + id: blockerExecutionWorkspaceId, + companyId, + projectId: blockerProjectId, + projectWorkspaceId: blockerProjectWorkspaceId, + mode: "operator_branch", + strategyType: "git_worktree", + name: "Blocker branch", + status: "active", + providerType: "git_worktree", + }, + ]); + await db.insert(issues).values([ + { + id: blockedIssueId, + companyId, + projectId: dependentProjectId, + projectWorkspaceId: dependentProjectWorkspaceId, + executionWorkspaceId: dependentExecutionWorkspaceId, + executionWorkspacePreference: "reuse_existing", + executionWorkspaceSettings: { mode: "operator_branch" }, + title: "Blocked dependent", + status: "blocked", + priority: "medium", + issueNumber: 1, + identifier: `${issuePrefix}-1`, + createdAt: issueTimestamp, + updatedAt: issueTimestamp, + }, + { + id: blockerIssueId, + companyId, + projectId: blockerProjectId, + projectWorkspaceId: blockerProjectWorkspaceId, + executionWorkspaceId: blockerExecutionWorkspaceId, + executionWorkspacePreference: "reuse_existing", + executionWorkspaceSettings: { mode: "operator_branch" }, + title: "Unassigned leaf blocker", + status: "todo", + priority: "medium", + issueNumber: 2, + identifier: `${issuePrefix}-2`, + createdAt: issueTimestamp, + updatedAt: issueTimestamp, + }, + ]); + await db.insert(issueRelations).values({ + companyId, + issueId: blockerIssueId, + relatedIssueId: blockedIssueId, + type: "blocks", + }); + + const result = await heartbeatService(db).reconcileIssueGraphLiveness(); + + expect(result.escalationsCreated).toBe(1); + const escalations = await db + .select() + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.originKind, "harness_liveness_escalation"))); + expect(escalations).toHaveLength(1); + expect(escalations[0]).toMatchObject({ + parentId: blockerIssueId, + projectId: blockerProjectId, + projectWorkspaceId: blockerProjectWorkspaceId, + executionWorkspaceId: null, + executionWorkspacePreference: null, + assigneeAgentId: managerId, + }); + }); + + it("reuses one open recovery issue for multiple dependents with the same leaf blocker", async () => { + await enableAutoRecovery(); + const { companyId, blockedIssueId, blockerIssueId } = await seedBlockedChain(); + const secondBlockedIssueId = randomUUID(); + const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + const issueTimestamp = new Date(Date.now() - 25 * 60 * 60 * 1000); + await db.insert(issues).values({ + id: secondBlockedIssueId, + companyId, + title: "Second blocked parent", + status: "blocked", + priority: "medium", + issueNumber: 3, + identifier: `${issuePrefix}-3`, + createdAt: issueTimestamp, + updatedAt: issueTimestamp, + }); + await db.insert(issueRelations).values({ + companyId, + issueId: blockerIssueId, + relatedIssueId: secondBlockedIssueId, + type: "blocks", + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileIssueGraphLiveness(); + + expect(result.findings).toBe(2); + expect(result.escalationsCreated).toBe(1); + expect(result.existingEscalations).toBe(1); + const escalations = await db + .select() + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.originKind, "harness_liveness_escalation"))); + expect(escalations).toHaveLength(1); + + const blockers = await db + .select({ blockedIssueId: issueRelations.relatedIssueId }) + .from(issueRelations) + .where(and(eq(issueRelations.companyId, companyId), eq(issueRelations.issueId, escalations[0]!.id))); + expect(blockers.map((row) => row.blockedIssueId).sort()).toEqual( + [blockedIssueId, secondBlockedIssueId].sort(), + ); + }); + it("creates a fresh escalation when the previous matching escalation is terminal", async () => { + await enableAutoRecovery(); const { companyId, managerId, blockedIssueId, blockerIssueId } = await seedBlockedChain(); const heartbeat = heartbeatService(db); const incidentKey = [ @@ -265,7 +549,7 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { expect(openEscalations).toHaveLength(2); const freshEscalation = openEscalations.find((issue) => issue.status !== "done"); expect(freshEscalation).toMatchObject({ - parentId: blockedIssueId, + parentId: blockerIssueId, assigneeAgentId: managerId, status: expect.stringMatching(/^(todo|in_progress|done)$/), }); diff --git a/server/src/__tests__/instance-settings-routes.test.ts b/server/src/__tests__/instance-settings-routes.test.ts index 72d408820ed..cad212c6492 100644 --- a/server/src/__tests__/instance-settings-routes.test.ts +++ b/server/src/__tests__/instance-settings-routes.test.ts @@ -58,6 +58,7 @@ describe("instance settings routes", () => { enableEnvironments: false, enableIsolatedWorkspaces: false, autoRestartDevServerWhenIdle: false, + enableIssueGraphLivenessAutoRecovery: false, }); mockInstanceSettingsService.updateGeneral.mockResolvedValue({ id: "instance-settings-1", @@ -73,6 +74,7 @@ describe("instance settings routes", () => { enableEnvironments: true, enableIsolatedWorkspaces: true, autoRestartDevServerWhenIdle: false, + enableIssueGraphLivenessAutoRecovery: false, }, }); mockInstanceSettingsService.listCompanyIds.mockResolvedValue(["company-1", "company-2"]); @@ -92,6 +94,7 @@ describe("instance settings routes", () => { enableEnvironments: false, enableIsolatedWorkspaces: false, autoRestartDevServerWhenIdle: false, + enableIssueGraphLivenessAutoRecovery: false, }); const patchRes = await request(app) @@ -103,7 +106,7 @@ describe("instance settings routes", () => { enableIsolatedWorkspaces: true, }); expect(mockLogActivity).toHaveBeenCalledTimes(2); - }); + }, 10_000); it("allows local board users to update guarded dev-server auto-restart", async () => { const app = await createApp({ @@ -118,8 +121,28 @@ describe("instance settings routes", () => { .send({ autoRestartDevServerWhenIdle: true }) .expect(200); + expect( + mockInstanceSettingsService.updateExperimental.mock.calls.some( + ([patch]) => patch?.autoRestartDevServerWhenIdle === true, + ), + ).toBe(true); + }); + + it("allows local board users to update issue graph liveness auto-recovery", async () => { + const app = await createApp({ + type: "board", + userId: "local-board", + source: "local_implicit", + isInstanceAdmin: true, + }); + + await request(app) + .patch("/api/instance/settings/experimental") + .send({ enableIssueGraphLivenessAutoRecovery: true }) + .expect(200); + expect(mockInstanceSettingsService.updateExperimental).toHaveBeenCalledWith({ - autoRestartDevServerWhenIdle: true, + enableIssueGraphLivenessAutoRecovery: true, }); }); diff --git a/server/src/__tests__/issue-liveness.test.ts b/server/src/__tests__/issue-liveness.test.ts index c675254514b..bda50dfc761 100644 --- a/server/src/__tests__/issue-liveness.test.ts +++ b/server/src/__tests__/issue-liveness.test.ts @@ -67,6 +67,7 @@ describe("issue graph liveness classifier", () => { issueId: blockedId, identifier: "PAP-1703", state: "blocked_by_unassigned_issue", + recoveryIssueId: blockerId, recommendedOwnerAgentId: managerId, dependencyPath: [ expect.objectContaining({ issueId: blockedId }), @@ -76,6 +77,57 @@ describe("issue graph liveness classifier", () => { }); }); + it("does not use free-form executive role or name matching for recovery ownership", () => { + const rootAgentId = "root-agent"; + const spoofedExecutiveId = "spoofed-executive"; + + const findings = classifyIssueGraphLiveness({ + issues: [ + issue({ + assigneeAgentId: null, + createdByAgentId: null, + }), + issue({ + id: blockerId, + identifier: "PAP-1704", + title: "Missing unblock work", + status: "todo", + assigneeAgentId: null, + createdByAgentId: null, + }), + ], + relations: blocks, + agents: [ + agent({ + id: spoofedExecutiveId, + name: "Chief Executive Recovery", + role: "cto", + title: "CEO", + reportsTo: rootAgentId, + }), + agent({ + id: rootAgentId, + name: "Root Operator", + role: "operator", + title: null, + reportsTo: null, + }), + ], + }); + + expect(findings).toHaveLength(1); + expect(findings[0]?.recommendedOwnerAgentId).toBe(rootAgentId); + expect(findings[0]?.recommendedOwnerCandidates[0]).toMatchObject({ + agentId: rootAgentId, + reason: "root_agent", + sourceIssueId: blockerId, + }); + expect(findings[0]?.recommendedOwnerCandidateAgentIds).toEqual([ + rootAgentId, + spoofedExecutiveId, + ]); + }); + it("does not flag a live blocked chain with an active assignee and wake path", () => { const findings = classifyIssueGraphLiveness({ issues: [ diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 8dbc3b74b82..dd62888d775 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -157,6 +157,7 @@ const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_r const CANCELLABLE_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const; const HEARTBEAT_RUN_TERMINAL_STATUSES = ["succeeded", "failed", "cancelled", "timed_out"] as const; const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = ["failed", "cancelled", "timed_out"] as const; +const ISSUE_GRAPH_LIVENESS_AUTO_RECOVERY_MIN_STALE_MS = 24 * 60 * 60 * 1000; export const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS = [ 2 * 60 * 1000, 10 * 60 * 1000, @@ -4434,7 +4435,7 @@ export function heartbeatService(db: Db) { } async function collectIssueGraphLivenessFindings() { - const [issueRows, relationRows, agentRows, activeRunRows, wakeRows] = await Promise.all([ + const [issueRows, relationRows, agentRows, activeRunRows, activeIssueRunRows, wakeRows] = await Promise.all([ db .select({ id: issues.id, @@ -4452,7 +4453,12 @@ export function heartbeatService(db: Db) { executionState: issues.executionState, }) .from(issues) - .where(isNull(issues.hiddenAt)), + .where( + and( + isNull(issues.hiddenAt), + notInArray(issues.originKind, ["harness_liveness_escalation"]), + ), + ), db .select({ companyId: issueRelations.companyId, @@ -4481,6 +4487,22 @@ export function heartbeatService(db: Db) { }) .from(heartbeatRuns) .where(inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES])), + db + .select({ + companyId: issues.companyId, + agentId: heartbeatRuns.agentId, + status: heartbeatRuns.status, + issueId: issues.id, + }) + .from(issues) + .innerJoin(heartbeatRuns, eq(issues.executionRunId, heartbeatRuns.id)) + .where( + and( + isNull(issues.hiddenAt), + notInArray(issues.originKind, ["harness_liveness_escalation"]), + inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]), + ), + ), db .select({ companyId: agentWakeupRequests.companyId, @@ -4501,7 +4523,12 @@ export function heartbeatService(db: Db) { agentId: row.agentId, status: row.status, issueId: issueIdFromRunContext(row.contextSnapshot), - })), + })).concat(activeIssueRunRows.map((row) => ({ + companyId: row.companyId, + agentId: row.agentId, + status: row.status, + issueId: row.issueId, + }))), queuedWakeRequests: wakeRows.map((row) => ({ companyId: row.companyId, agentId: row.agentId, @@ -4528,6 +4555,67 @@ export function heartbeatService(db: Db) { .then((rows) => rows[0] ?? null); } + function parseLivenessIncidentKey(incidentKey: string | null | undefined) { + if (!incidentKey) return null; + const parts = incidentKey.split(":"); + if (parts.length !== 5 || parts[0] !== "harness_liveness") return null; + const [, companyId, issueId, state, leafIssueId] = parts; + if (!companyId || !issueId || !state || !leafIssueId) return null; + return { companyId, issueId, state, leafIssueId }; + } + + function livenessRecoveryLeafIssueId(finding: IssueLivenessFinding) { + return finding.recoveryIssueId; + } + + function livenessRecoveryLeafFingerprint(finding: IssueLivenessFinding) { + return [ + "harness_liveness_leaf", + finding.companyId, + finding.state, + livenessRecoveryLeafIssueId(finding), + ].join(":"); + } + + function livenessRecoveryLeafKey(companyId: string, state: string, leafIssueId: string) { + return ["harness_liveness_leaf", companyId, state, leafIssueId].join(":"); + } + + async function findOpenLivenessRecoveryIssueForLeaf(finding: IssueLivenessFinding) { + const byFingerprint = await db + .select() + .from(issues) + .where( + and( + eq(issues.companyId, finding.companyId), + eq(issues.originKind, "harness_liveness_escalation"), + eq(issues.originFingerprint, livenessRecoveryLeafFingerprint(finding)), + isNull(issues.hiddenAt), + notInArray(issues.status, ["done", "cancelled"]), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null); + if (byFingerprint) return byFingerprint; + + const leafIssueId = livenessRecoveryLeafIssueId(finding); + const openRecoveries = await db + .select() + .from(issues) + .where( + and( + eq(issues.companyId, finding.companyId), + eq(issues.originKind, "harness_liveness_escalation"), + isNull(issues.hiddenAt), + notInArray(issues.status, ["done", "cancelled"]), + ), + ); + return openRecoveries.find((row) => { + const parsed = parseLivenessIncidentKey(row.originId); + return parsed?.state === finding.state && parsed.leafIssueId === leafIssueId; + }) ?? null; + } + async function existingBlockerIssueIds(companyId: string, issueId: string) { return db .select({ blockerIssueId: issueRelations.issueId }) @@ -4542,6 +4630,139 @@ export function heartbeatService(db: Db) { .then((rows) => rows.map((row) => row.blockerIssueId)); } + async function removeRecoveryBlockerFromSource(recovery: typeof issues.$inferSelect) { + const parsed = parseLivenessIncidentKey(recovery.originId); + if (!parsed) return false; + const sourceIssue = await db + .select() + .from(issues) + .where(and(eq(issues.companyId, recovery.companyId), eq(issues.id, parsed.issueId))) + .then((rows) => rows[0] ?? null); + if (!sourceIssue) return false; + + const blockerIds = await existingBlockerIssueIds(sourceIssue.companyId, sourceIssue.id); + if (!blockerIds.includes(recovery.id)) return false; + await issuesSvc.update(sourceIssue.id, { + blockedByIssueIds: blockerIds.filter((blockerId) => blockerId !== recovery.id), + }); + return true; + } + + async function hasActiveRunForIssueId(companyId: string, issueId: string) { + const [contextRun, issueRun] = await Promise.all([ + db + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, companyId), + inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]), + sql`(${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId} + OR ${heartbeatRuns.contextSnapshot}->>'taskId' = ${issueId})`, + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null), + db + .select({ id: heartbeatRuns.id }) + .from(issues) + .innerJoin(heartbeatRuns, eq(issues.executionRunId, heartbeatRuns.id)) + .where( + and( + eq(issues.companyId, companyId), + eq(issues.id, issueId), + inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null), + ]); + return Boolean(contextRun || issueRun); + } + + async function retireObsoleteLivenessRecoveryIssues(findings: IssueLivenessFinding[]) { + const currentIncidentKeys = new Set(findings.map((finding) => finding.incidentKey)); + const currentLeafKeys = new Set( + findings.map((finding) => + livenessRecoveryLeafKey( + finding.companyId, + finding.state, + livenessRecoveryLeafIssueId(finding), + ), + ), + ); + const openRecoveries = await db + .select() + .from(issues) + .where( + and( + eq(issues.originKind, "harness_liveness_escalation"), + isNull(issues.hiddenAt), + notInArray(issues.status, ["done", "cancelled"]), + ), + ); + const result = { + retired: 0, + activeSkipped: 0, + blockerRelationsRemoved: 0, + retiredIssueIds: [] as string[], + }; + + for (const recovery of openRecoveries) { + if (recovery.originId && currentIncidentKeys.has(recovery.originId)) continue; + const parsed = parseLivenessIncidentKey(recovery.originId); + if (!parsed) continue; + if ( + currentLeafKeys.has( + livenessRecoveryLeafKey(parsed.companyId, parsed.state, parsed.leafIssueId), + ) + ) { + continue; + } + if (await removeRecoveryBlockerFromSource(recovery)) { + result.blockerRelationsRemoved += 1; + } + if (await hasActiveRunForIssueId(recovery.companyId, recovery.id)) { + result.activeSkipped += 1; + continue; + } + await issuesSvc.update(recovery.id, { status: "cancelled" }); + result.retired += 1; + result.retiredIssueIds.push(recovery.id); + } + + return result; + } + + async function isLivenessFindingOldEnoughForAutoRecovery(finding: IssueLivenessFinding, now = new Date()) { + const issueIds = [...new Set(finding.dependencyPath.map((entry) => entry.issueId))]; + if (issueIds.length === 0) return false; + const rows = await db + .select({ id: issues.id, updatedAt: issues.updatedAt }) + .from(issues) + .where(and(eq(issues.companyId, finding.companyId), inArray(issues.id, issueIds))); + if (rows.length !== issueIds.length) return false; + const latestUpdatedAt = rows.reduce((latest, row) => + row.updatedAt > latest ? row.updatedAt : latest, + rows[0]!.updatedAt); + return now.getTime() - latestUpdatedAt.getTime() >= ISSUE_GRAPH_LIVENESS_AUTO_RECOVERY_MIN_STALE_MS; + } + + function isUniqueLivenessRecoveryConflict(error: unknown) { + if (!error || typeof error !== "object") return false; + const maybe = error as { code?: string; constraint?: string; message?: string }; + return maybe.code === "23505" && + ( + maybe.constraint === "issues_active_liveness_recovery_incident_uq" || + maybe.constraint === "issues_active_liveness_recovery_leaf_uq" || + typeof maybe.message === "string" && + ( + maybe.message.includes("issues_active_liveness_recovery_incident_uq") || + maybe.message.includes("issues_active_liveness_recovery_leaf_uq") + ) + ); + } + function formatDependencyPath(finding: IssueLivenessFinding) { return finding.dependencyPath .map((entry) => entry.identifier ?? entry.issueId) @@ -4581,22 +4802,55 @@ export function heartbeatService(db: Db) { finding: IssueLivenessFinding, issue: typeof issues.$inferSelect, ) { - const candidates = [ - finding.recommendedOwnerAgentId, - ...finding.recommendedOwnerCandidateAgentIds, - ].filter((candidate): candidate is string => Boolean(candidate)); + const detailedCandidates = finding.recommendedOwnerCandidates.length > 0 + ? finding.recommendedOwnerCandidates + : finding.recommendedOwnerCandidateAgentIds.map((agentId) => ({ + agentId, + reason: "ordered_invokable_fallback" as const, + sourceIssueId: finding.recoveryIssueId, + })); + const seenCandidates = new Set(); + const candidates = detailedCandidates.filter((candidate) => { + if (seenCandidates.has(candidate.agentId)) return false; + seenCandidates.add(candidate.agentId); + return true; + }); + const budgetBlockedCandidateAgentIds: string[] = []; - for (const candidate of [...new Set(candidates)]) { - const budgetBlock = await budgets.getInvocationBlock(issue.companyId, candidate, { + for (const candidate of candidates) { + const budgetBlock = await budgets.getInvocationBlock(issue.companyId, candidate.agentId, { issueId: issue.id, projectId: issue.projectId, }); - if (!budgetBlock) return candidate; + if (!budgetBlock) { + return { + agentId: candidate.agentId, + reason: candidate.reason, + sourceIssueId: candidate.sourceIssueId, + candidateAgentIds: candidates.map((entry) => entry.agentId), + candidateReasons: candidates.map((entry) => ({ + agentId: entry.agentId, + reason: entry.reason, + sourceIssueId: entry.sourceIssueId, + })), + budgetBlockedCandidateAgentIds, + }; + } + budgetBlockedCandidateAgentIds.push(candidate.agentId); } return null; } + function shouldReuseRecoveryExecutionWorkspace(input: { + finding: IssueLivenessFinding; + recoveryIssue: typeof issues.$inferSelect; + ownerAgentId: string; + }) { + if (input.finding.recoveryIssueId === input.finding.issueId) return false; + return input.recoveryIssue.assigneeAgentId === input.ownerAgentId; + } + async function ensureIssueBlockedByEscalation(input: { issue: typeof issues.$inferSelect; escalationIssueId: string; @@ -4649,7 +4903,16 @@ export function heartbeatService(db: Db) { .then((rows) => rows[0] ?? null); if (!issue || issue.companyId !== input.finding.companyId) return { kind: "skipped" as const }; - const existing = await findOpenLivenessEscalation(issue.companyId, input.finding.incidentKey); + const recoveryIssue = await db + .select() + .from(issues) + .where(and(eq(issues.id, input.finding.recoveryIssueId), eq(issues.companyId, issue.companyId))) + .then((rows) => rows[0] ?? null); + if (!recoveryIssue) return { kind: "skipped" as const }; + + const existing = + await findOpenLivenessEscalation(issue.companyId, input.finding.incidentKey) ?? + await findOpenLivenessRecoveryIssueForLeaf(input.finding); if (existing) { await ensureIssueBlockedByEscalation({ issue, @@ -4660,24 +4923,52 @@ export function heartbeatService(db: Db) { return { kind: "existing" as const, escalationIssueId: existing.id }; } - const ownerAgentId = await resolveEscalationOwnerAgentId(input.finding, issue); - if (!ownerAgentId) return { kind: "skipped" as const }; - - const escalation = await issuesSvc.create(issue.companyId, { - title: `Unblock liveness incident for ${issue.identifier ?? issue.title}`, - description: buildLivenessEscalationDescription(input.finding), - status: "todo", - priority: "high", - parentId: issue.id, - projectId: issue.projectId, - goalId: issue.goalId, - assigneeAgentId: ownerAgentId, - originKind: "harness_liveness_escalation", - originId: input.finding.incidentKey, - billingCode: issue.billingCode, - inheritExecutionWorkspaceFromIssueId: issue.id, + const ownerSelection = await resolveEscalationOwnerAgentId(input.finding, recoveryIssue); + if (!ownerSelection) return { kind: "skipped" as const }; + const reuseRecoveryExecutionWorkspace = shouldReuseRecoveryExecutionWorkspace({ + finding: input.finding, + recoveryIssue, + ownerAgentId: ownerSelection.agentId, }); + let escalation: Awaited>; + try { + escalation = await issuesSvc.create(issue.companyId, { + title: `Unblock liveness incident for ${recoveryIssue.identifier ?? recoveryIssue.title}`, + description: buildLivenessEscalationDescription(input.finding), + status: "todo", + priority: "high", + parentId: recoveryIssue.id, + projectId: recoveryIssue.projectId, + goalId: recoveryIssue.goalId, + assigneeAgentId: ownerSelection.agentId, + originKind: "harness_liveness_escalation", + originId: input.finding.incidentKey, + originFingerprint: livenessRecoveryLeafFingerprint(input.finding), + billingCode: recoveryIssue.billingCode, + ...(reuseRecoveryExecutionWorkspace + ? { inheritExecutionWorkspaceFromIssueId: recoveryIssue.id } + : { + executionWorkspaceId: null, + executionWorkspacePreference: null, + executionWorkspaceSettings: null, + }), + }); + } catch (error) { + if (!isUniqueLivenessRecoveryConflict(error)) throw error; + const raced = + await findOpenLivenessEscalation(issue.companyId, input.finding.incidentKey) ?? + await findOpenLivenessRecoveryIssueForLeaf(input.finding); + if (!raced) throw error; + await ensureIssueBlockedByEscalation({ + issue, + escalationIssueId: raced.id, + finding: input.finding, + runId: input.runId ?? null, + }); + return { kind: "existing" as const, escalationIssueId: raced.id }; + } + await ensureIssueBlockedByEscalation({ issue, escalationIssueId: escalation.id, @@ -4695,7 +4986,7 @@ export function heartbeatService(db: Db) { companyId: issue.companyId, actorType: "system", actorId: "system", - agentId: ownerAgentId, + agentId: ownerSelection.agentId, runId: input.runId ?? null, action: "issue.harness_liveness_escalation_created", entityType: "issue", @@ -4706,19 +4997,35 @@ export function heartbeatService(db: Db) { findingState: input.finding.state, sourceIssueId: issue.id, sourceIdentifier: issue.identifier, + recoveryIssueId: recoveryIssue.id, + recoveryIdentifier: recoveryIssue.identifier, escalationIssueId: escalation.id, escalationIdentifier: escalation.identifier, dependencyPath: input.finding.dependencyPath, + ownerSelection: { + selectedAgentId: ownerSelection.agentId, + selectedReason: ownerSelection.reason, + selectedSourceIssueId: ownerSelection.sourceIssueId, + candidateAgentIds: ownerSelection.candidateAgentIds, + candidateReasons: ownerSelection.candidateReasons, + budgetBlockedCandidateAgentIds: ownerSelection.budgetBlockedCandidateAgentIds, + }, + workspaceSelection: { + reuseRecoveryExecutionWorkspace, + inheritedExecutionWorkspaceFromIssueId: reuseRecoveryExecutionWorkspace ? recoveryIssue.id : null, + projectWorkspaceSourceIssueId: recoveryIssue.id, + }, }, }); - const wake = await enqueueWakeup(ownerAgentId, { + const wake = await enqueueWakeup(ownerSelection.agentId, { source: "assignment", triggerDetail: "system", reason: "issue_assigned", payload: { issueId: escalation.id, sourceIssueId: issue.id, + recoveryIssueId: recoveryIssue.id, incidentKey: input.finding.incidentKey, }, requestedByActorType: "system", @@ -4729,6 +5036,7 @@ export function heartbeatService(db: Db) { wakeReason: "issue_assigned", source: "harness_liveness_escalation", sourceIssueId: issue.id, + recoveryIssueId: recoveryIssue.id, incidentKey: input.finding.incidentKey, }, }); @@ -4737,8 +5045,10 @@ export function heartbeatService(db: Db) { incidentKey: input.finding.incidentKey, findingState: input.finding.state, sourceIssueId: issue.id, + recoveryIssueId: recoveryIssue.id, escalationIssueId: escalation.id, - ownerAgentId, + ownerAgentId: ownerSelection.agentId, + ownerSelectionReason: ownerSelection.reason, wakeupRunId: wake?.id ?? null, }, "created issue graph liveness escalation"); @@ -4747,16 +5057,37 @@ export function heartbeatService(db: Db) { async function reconcileIssueGraphLiveness(opts?: { runId?: string | null }) { const findings = await collectIssueGraphLivenessFindings(); + const experimentalSettings = await instanceSettings.getExperimental(); + const autoRecoveryEnabled = experimentalSettings.enableIssueGraphLivenessAutoRecovery === true; + const obsoleteRecoveryCleanup = await retireObsoleteLivenessRecoveryIssues(findings); const result = { findings: findings.length, + autoRecoveryEnabled, escalationsCreated: 0, existingEscalations: 0, skipped: 0, + skippedAutoRecoveryDisabled: 0, + skippedAutoRecoveryTooYoung: 0, + obsoleteRecoveriesRetired: obsoleteRecoveryCleanup.retired, + obsoleteRecoveriesActiveSkipped: obsoleteRecoveryCleanup.activeSkipped, + obsoleteRecoveryBlockerRelationsRemoved: obsoleteRecoveryCleanup.blockerRelationsRemoved, issueIds: [] as string[], escalationIssueIds: [] as string[], + retiredRecoveryIssueIds: obsoleteRecoveryCleanup.retiredIssueIds, }; + if (!autoRecoveryEnabled) { + result.skippedAutoRecoveryDisabled = findings.length; + return result; + } + + const now = new Date(); for (const finding of findings) { + if (!await isLivenessFindingOldEnoughForAutoRecovery(finding, now)) { + result.skippedAutoRecoveryTooYoung += 1; + result.skipped += 1; + continue; + } const escalation = await createIssueGraphLivenessEscalation({ finding, runId: opts?.runId ?? null, diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 4a6abb9e9e0..7ada2ef7b8c 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -41,12 +41,14 @@ function normalizeExperimentalSettings(raw: unknown): InstanceExperimentalSettin enableEnvironments: parsed.data.enableEnvironments ?? false, enableIsolatedWorkspaces: parsed.data.enableIsolatedWorkspaces ?? false, autoRestartDevServerWhenIdle: parsed.data.autoRestartDevServerWhenIdle ?? false, + enableIssueGraphLivenessAutoRecovery: parsed.data.enableIssueGraphLivenessAutoRecovery ?? false, }; } return { enableEnvironments: false, enableIsolatedWorkspaces: false, autoRestartDevServerWhenIdle: false, + enableIssueGraphLivenessAutoRecovery: false, }; } diff --git a/server/src/services/issue-liveness.ts b/server/src/services/issue-liveness.ts index c7cc58f6cf9..09892dd622a 100644 --- a/server/src/services/issue-liveness.ts +++ b/server/src/services/issue-liveness.ts @@ -52,6 +52,19 @@ export interface IssueLivenessDependencyPathEntry { status: string; } +export type IssueLivenessOwnerCandidateReason = + | "stalled_blocker_assignee" + | "assignee_reporting_chain" + | "creator_reporting_chain" + | "root_agent" + | "ordered_invokable_fallback"; + +export interface IssueLivenessOwnerCandidate { + agentId: string; + reason: IssueLivenessOwnerCandidateReason; + sourceIssueId: string; +} + export interface IssueLivenessFinding { issueId: string; companyId: string; @@ -60,8 +73,10 @@ export interface IssueLivenessFinding { severity: IssueLivenessSeverity; reason: string; dependencyPath: IssueLivenessDependencyPathEntry[]; + recoveryIssueId: string; recommendedOwnerAgentId: string | null; recommendedOwnerCandidateAgentIds: string[]; + recommendedOwnerCandidates: IssueLivenessOwnerCandidate[]; recommendedAction: string; incidentKey: string; } @@ -119,48 +134,111 @@ function principalIsResolvableUser(principal: unknown): boolean { return value.type === "user" && typeof value.userId === "string" && value.userId.length > 0; } -function agentChainCandidates( +function addOwnerCandidate( + candidates: IssueLivenessOwnerCandidate[], + seen: Set, + agentsById: Map, + companyId: string, + agentId: string | null | undefined, + reason: IssueLivenessOwnerCandidateReason, + sourceIssueId: string, +) { + if (!agentId || seen.has(agentId)) return; + const agent = agentsById.get(agentId); + if (!agent || agent.companyId !== companyId || !isInvokableAgent(agent)) return; + seen.add(agentId); + candidates.push({ agentId, reason, sourceIssueId }); +} + +function addAgentChainCandidates( + candidates: IssueLivenessOwnerCandidate[], + seen: Set, startAgentId: string | null | undefined, agentsById: Map, companyId: string, + reason: IssueLivenessOwnerCandidateReason, + sourceIssueId: string, ) { - const candidates: string[] = []; - const seen = new Set(); + const chainSeen = new Set(); let current = startAgentId ? agentsById.get(startAgentId) : null; while (current?.reportsTo) { - if (seen.has(current.reportsTo)) break; - seen.add(current.reportsTo); + if (chainSeen.has(current.reportsTo)) break; + chainSeen.add(current.reportsTo); const manager = agentsById.get(current.reportsTo); if (!manager || manager.companyId !== companyId) break; - if (isInvokableAgent(manager)) candidates.push(manager.id); + addOwnerCandidate(candidates, seen, agentsById, companyId, manager.id, reason, sourceIssueId); current = manager; } - - return candidates; } -function fallbackExecutiveCandidates(agents: IssueLivenessAgentInput[], companyId: string) { - const active = agents.filter((agent) => agent.companyId === companyId && isInvokableAgent(agent)); - const executive = active.filter((agent) => { - const haystack = `${agent.role} ${agent.title ?? ""} ${agent.name}`.toLowerCase(); - return /\b(cto|chief technology|ceo|chief executive)\b/.test(haystack); - }); - const roots = active.filter((agent) => !agent.reportsTo); - return [...executive, ...roots, ...active].map((agent) => agent.id); +function orderedInvokableAgents(agents: IssueLivenessAgentInput[], companyId: string) { + return agents + .filter((agent) => agent.companyId === companyId && isInvokableAgent(agent)) + .sort((left, right) => left.id.localeCompare(right.id)); } -function ownerCandidatesForIssue( +function ownerCandidatesForRecoveryIssue( issue: IssueLivenessIssueInput, agents: IssueLivenessAgentInput[], agentsById: Map, + options: { + includeStalledAssignee?: boolean; + } = {}, ) { - const candidates = [ - ...agentChainCandidates(issue.assigneeAgentId, agentsById, issue.companyId), - ...agentChainCandidates(issue.createdByAgentId, agentsById, issue.companyId), - ...fallbackExecutiveCandidates(agents, issue.companyId), - ]; - return [...new Set(candidates)]; + const candidates: IssueLivenessOwnerCandidate[] = []; + const seen = new Set(); + + if (options.includeStalledAssignee && issue.status !== "cancelled" && issue.status !== "done") { + addOwnerCandidate( + candidates, + seen, + agentsById, + issue.companyId, + issue.assigneeAgentId, + "stalled_blocker_assignee", + issue.id, + ); + } + + addAgentChainCandidates( + candidates, + seen, + issue.assigneeAgentId, + agentsById, + issue.companyId, + "assignee_reporting_chain", + issue.id, + ); + addAgentChainCandidates( + candidates, + seen, + issue.createdByAgentId, + agentsById, + issue.companyId, + "creator_reporting_chain", + issue.id, + ); + + const invokableAgents = orderedInvokableAgents(agents, issue.companyId); + for (const agent of invokableAgents) { + if (!agent.reportsTo) { + addOwnerCandidate(candidates, seen, agentsById, issue.companyId, agent.id, "root_agent", issue.id); + } + } + for (const agent of invokableAgents) { + addOwnerCandidate( + candidates, + seen, + agentsById, + issue.companyId, + agent.id, + "ordered_invokable_fallback", + issue.id, + ); + } + + return candidates; } function incidentKey(input: { @@ -185,7 +263,9 @@ function finding(input: { severity?: IssueLivenessSeverity; reason: string; dependencyPath: IssueLivenessIssueInput[]; + recoveryIssue: IssueLivenessIssueInput; recommendedOwnerCandidateAgentIds: string[]; + recommendedOwnerCandidates: IssueLivenessOwnerCandidate[]; recommendedAction: string; blockerIssueId?: string | null; participantAgentId?: string | null; @@ -198,8 +278,10 @@ function finding(input: { severity: input.severity ?? "critical", reason: input.reason, dependencyPath: input.dependencyPath.map(pathEntry), + recoveryIssueId: input.recoveryIssue.id, recommendedOwnerAgentId: input.recommendedOwnerCandidateAgentIds[0] ?? null, recommendedOwnerCandidateAgentIds: input.recommendedOwnerCandidateAgentIds, + recommendedOwnerCandidates: input.recommendedOwnerCandidates, recommendedAction: input.recommendedAction, incidentKey: incidentKey({ companyId: input.issue.companyId, @@ -226,14 +308,15 @@ export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): Issu } for (const issue of input.issues) { - const ownerCandidates = ownerCandidatesForIssue(issue, input.agents, agentsById); - if (issue.status === "blocked") { const relations = blockersByBlockedIssueId.get(issue.id) ?? []; for (const relation of relations) { if (relation.companyId !== issue.companyId) continue; const blocker = issuesById.get(relation.blockerIssueId); if (!blocker || blocker.companyId !== issue.companyId || blocker.status === "done") continue; + const ownerCandidates = ownerCandidatesForRecoveryIssue(blocker, input.agents, agentsById, { + includeStalledAssignee: true, + }); if (blocker.status === "cancelled") { findings.push(finding({ @@ -241,7 +324,9 @@ export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): Issu state: "blocked_by_cancelled_issue", reason: `${issueLabel(issue)} is still blocked by cancelled issue ${issueLabel(blocker)}.`, dependencyPath: [issue, blocker], - recommendedOwnerCandidateAgentIds: ownerCandidates, + recoveryIssue: blocker, + recommendedOwnerCandidateAgentIds: ownerCandidates.map((candidate) => candidate.agentId), + recommendedOwnerCandidates: ownerCandidates, recommendedAction: `Inspect ${issueLabel(blocker)} and either remove it from ${issueLabel(issue)}'s blockers or replace it with an actionable unblock issue.`, blockerIssueId: blocker.id, @@ -256,7 +341,9 @@ export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): Issu state: "blocked_by_unassigned_issue", reason: `${issueLabel(issue)} is blocked by unassigned issue ${issueLabel(blocker)} with no user owner.`, dependencyPath: [issue, blocker], - recommendedOwnerCandidateAgentIds: ownerCandidates, + recoveryIssue: blocker, + recommendedOwnerCandidateAgentIds: ownerCandidates.map((candidate) => candidate.agentId), + recommendedOwnerCandidates: ownerCandidates, recommendedAction: `Assign ${issueLabel(blocker)} to an owner who can complete it, or remove it from ${issueLabel(issue)}'s blockers if it is no longer required.`, blockerIssueId: blocker.id, @@ -276,7 +363,9 @@ export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): Issu ? `${issueLabel(issue)} is blocked by ${issueLabel(blocker)}, but its assignee is ${blockerAgent.status}.` : `${issueLabel(issue)} is blocked by ${issueLabel(blocker)}, but its assignee no longer exists.`, dependencyPath: [issue, blocker], - recommendedOwnerCandidateAgentIds: ownerCandidates, + recoveryIssue: blocker, + recommendedOwnerCandidateAgentIds: ownerCandidates.map((candidate) => candidate.agentId), + recommendedOwnerCandidates: ownerCandidates, recommendedAction: `Review ${issueLabel(blocker)} and assign it to an active owner or replace the blocker with an actionable issue.`, blockerIssueId: blocker.id, @@ -286,6 +375,7 @@ export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): Issu } if (issue.status !== "in_review" || !issue.executionState) continue; + const ownerCandidates = ownerCandidatesForRecoveryIssue(issue, input.agents, agentsById); const participant = issue.executionState.currentParticipant; const participantAgentId = readPrincipalAgentId(participant); if (participantAgentId) { @@ -298,7 +388,9 @@ export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): Issu ? `${issueLabel(issue)} is in review, but current participant agent is ${participantAgent.status}.` : `${issueLabel(issue)} is in review, but current participant agent cannot be resolved.`, dependencyPath: [issue], - recommendedOwnerCandidateAgentIds: ownerCandidates, + recoveryIssue: issue, + recommendedOwnerCandidateAgentIds: ownerCandidates.map((candidate) => candidate.agentId), + recommendedOwnerCandidates: ownerCandidates, recommendedAction: `Repair ${issueLabel(issue)}'s review participant or return the issue to an active assignee with a clear change request.`, participantAgentId, @@ -313,7 +405,9 @@ export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): Issu state: "invalid_review_participant", reason: `${issueLabel(issue)} is in review, but its current participant cannot be resolved.`, dependencyPath: [issue], - recommendedOwnerCandidateAgentIds: ownerCandidates, + recoveryIssue: issue, + recommendedOwnerCandidateAgentIds: ownerCandidates.map((candidate) => candidate.agentId), + recommendedOwnerCandidates: ownerCandidates, recommendedAction: `Repair ${issueLabel(issue)}'s review participant or return the issue to an active assignee with a clear change request.`, })); diff --git a/ui/src/pages/InstanceExperimentalSettings.tsx b/ui/src/pages/InstanceExperimentalSettings.tsx index f2e2a9b0d15..a45997d9f0e 100644 --- a/ui/src/pages/InstanceExperimentalSettings.tsx +++ b/ui/src/pages/InstanceExperimentalSettings.tsx @@ -56,6 +56,8 @@ export function InstanceExperimentalSettings() { const enableEnvironments = experimentalQuery.data?.enableEnvironments === true; const enableIsolatedWorkspaces = experimentalQuery.data?.enableIsolatedWorkspaces === true; const autoRestartDevServerWhenIdle = experimentalQuery.data?.autoRestartDevServerWhenIdle === true; + const enableIssueGraphLivenessAutoRecovery = + experimentalQuery.data?.enableIssueGraphLivenessAutoRecovery === true; return (
@@ -128,6 +130,28 @@ export function InstanceExperimentalSettings() { />
+ +
+
+
+

Auto-Create Issue Recovery Tasks

+

+ Let the heartbeat scheduler create recovery issues for issue dependency chains that have been stalled for + at least 24 hours. +

+
+ + toggleMutation.mutate({ + enableIssueGraphLivenessAutoRecovery: !enableIssueGraphLivenessAutoRecovery, + }) + } + disabled={toggleMutation.isPending} + aria-label="Toggle issue graph liveness auto-recovery" + /> +
+
); }