Gate liveness recovery issue creation

This commit is contained in:
Dotta committed 2026-04-24 13:39:09 -05:00
1 parent 8f1cd0474f
commit 51d2933cc1
15 files changed
+1022 -296

No files matched your search

@@ -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');
+16
View File
@@ -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')`,
),
}),
);
+85 -61
View File
@@ -33,77 +33,56 @@ export type EmbeddedPostgresTestDatabase = {
let embeddedPostgresSupportPromise: Promise<EmbeddedPostgresTestSupport> | null = null;
const DEFAULT_PAPERCLIP_EMBEDDED_POSTGRES_PORT = 54329;
function getReservedTestPorts(): Set<number> {
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<EmbeddedPostgresCtor> {
const mod = await import("embedded-postgres");
return mod.default as EmbeddedPostgresCtor;
}
async function getAvailablePort(): Promise<number> {
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<number>((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<EmbeddedPostgresTestSupport> {
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<EmbeddedPostgresTestSupport> {
if (!embeddedPostgresSupportPromise) {
embeddedPostgresSupportPromise = probeEmbeddedPostgresSupport();
}
return await embeddedPostgresSupportPromise;
}
export async function startEmbeddedPostgresTestDatabase(
tempDirPrefix: string,
): Promise<EmbeddedPostgresTestDatabase> {
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<EmbeddedPostgresTestSupport> {
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<EmbeddedPostgresTestSupport> {
if (!embeddedPostgresSupportPromise) {
embeddedPostgresSupportPromise = probeEmbeddedPostgresSupport();
}
return await embeddedPostgresSupportPromise;
}
export async function startEmbeddedPostgresTestDatabase(
tempDirPrefix: string,
): Promise<EmbeddedPostgresTestDatabase> {
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)}`,
);
+1
View File
@@ -27,6 +27,7 @@ export interface InstanceExperimentalSettings {
enableEnvironments: boolean;
enableIsolatedWorkspaces: boolean;
autoRestartDevServerWhenIdle: boolean;
enableIssueGraphLivenessAutoRecovery: boolean;
}
export interface InstanceSettings {
@@ -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();
+10
View File
@@ -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.
+6 -79
View File
@@ -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<void>;
start(): Promise<void>;
stop(): Promise<void>;
};
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<EmbeddedPostgresCtor> {
const mod = await import("embedded-postgres");
return mod.default as EmbeddedPostgresCtor;
}
async function getAvailablePort(): Promise<number> {
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<typeof createDb> | undefined) {
await db?.$client?.end?.({ timeout: 0 });
@@ -99,17 +31,15 @@ async function closeDbClient(db: ReturnType<typeof createDb> | undefined) {
describe("feedbackService.saveIssueVote", () => {
let db!: ReturnType<typeof createDb>;
let svc!: ReturnType<typeof feedbackService>;
let instance: EmbeddedPostgresInstance | null = null;
let dataDir = "";
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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() {
@@ -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<void>;
start(): Promise<void>;
stop(): Promise<void>;
};
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<EmbeddedPostgresCtor> {
const mod = await import("embedded-postgres");
return mod.default as EmbeddedPostgresCtor;
}
async function getAvailablePort(): Promise<number> {
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<boolean>, timeoutMs = 10_000, intervalMs = 50) {
const startedAt = Date.now();
@@ -218,22 +147,17 @@ async function createControlledGatewayServer() {
describe("heartbeat comment wake batching", () => {
let db!: ReturnType<typeof createDb>;
let instance: EmbeddedPostgresInstance | null = null;
let dataDir = "";
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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 () => {
@@ -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)$/),
});
@@ -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,
});
});
@@ -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: [
+361 -30
View File
@@ -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,
@@ -4508,7 +4509,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,
@@ -4526,7 +4527,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,
@@ -4555,6 +4561,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,
@@ -4575,7 +4597,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,
@@ -4602,6 +4629,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 })
@@ -4616,6 +4704,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)
@@ -4655,22 +4876,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<string>();
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;
@@ -4723,7 +4977,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,
@@ -4734,24 +4997,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<ReturnType<typeof issuesSvc.create>>;
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,
@@ -4769,7 +5060,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",
@@ -4780,19 +5071,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",
@@ -4803,6 +5110,7 @@ export function heartbeatService(db: Db) {
wakeReason: "issue_assigned",
source: "harness_liveness_escalation",
sourceIssueId: issue.id,
recoveryIssueId: recoveryIssue.id,
incidentKey: input.finding.incidentKey,
},
});
@@ -4811,8 +5119,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");
@@ -4821,16 +5131,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,
+2
View File
@@ -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,
};
}
+124 -30
View File
@@ -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<string>,
agentsById: Map<string, IssueLivenessAgentInput>,
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<string>,
startAgentId: string | null | undefined,
agentsById: Map<string, IssueLivenessAgentInput>,
companyId: string,
reason: IssueLivenessOwnerCandidateReason,
sourceIssueId: string,
) {
const candidates: string[] = [];
const seen = new Set<string>();
const chainSeen = new Set<string>();
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<string, IssueLivenessAgentInput>,
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<string>();
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.`,
}));
@@ -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 (
<div className="max-w-4xl space-y-6">
@@ -128,6 +130,28 @@ export function InstanceExperimentalSettings() {
/>
</div>
</section>
<section className="rounded-xl border border-border bg-card p-5">
<div className="flex items-start justify-between gap-4">
<div className="space-y-1.5">
<h2 className="text-sm font-semibold">Auto-Create Issue Recovery Tasks</h2>
<p className="max-w-2xl text-sm text-muted-foreground">
Let the heartbeat scheduler create recovery issues for issue dependency chains that have been stalled for
at least 24 hours.
</p>
</div>
<ToggleSwitch
checked={enableIssueGraphLivenessAutoRecovery}
onCheckedChange={() =>
toggleMutation.mutate({
enableIssueGraphLivenessAutoRecovery: !enableIssueGraphLivenessAutoRecovery,
})
}
disabled={toggleMutation.isPending}
aria-label="Toggle issue graph liveness auto-recovery"
/>
</div>
</section>
</div>
);
}