Jannik/PaperClipAI
1
0 Fork 0
mirror of https://github.com/paperclipai/paperclip.git synced 2026-10-23 09:07:46 +02:00
Code Issues Packages Projects Releases Wiki Activity

Revert "PAP-2731 Implement sky explicit-wait UI surfaces"

Browse Source 
This reverts commit 08a9e0caf0.
This commit is contained in:
Dotta committed 2026-04-29 16:02:11 -05:00
1 parent 0a345e9648
commit caf37a355d
43 files changed
+181 -4777
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/__tests__/agent-permissions-routes.test.ts
-127
View File
@@ -66,10 +66,6 @@ const mockHeartbeatService = vi.hoisted(() => ({
cancelRun: vi.fn(),
}));
const mockRecoveryService = vi.hoisted(() => ({
recordWatchdogDecision: vi.fn(),
}));
const mockIssueApprovalService = vi.hoisted(() => ({
linkManyForApproval: vi.fn(),
}));
@@ -146,10 +142,6 @@ function registerModuleMocks() {
heartbeatService: () => mockHeartbeatService,
}));
vi.doMock("../services/recovery/service.js", () => ({
recoveryService: () => mockRecoveryService,
}));
vi.doMock("../services/issue-approvals.js", () => ({
issueApprovalService: () => mockIssueApprovalService,
}));
@@ -279,7 +271,6 @@ describe.sequential("agent permission routes", () => {
vi.doUnmock("../services/budgets.js");
vi.doUnmock("../services/company-skills.js");
vi.doUnmock("../services/heartbeat.js");
vi.doUnmock("../services/recovery/service.js");
vi.doUnmock("../services/index.js");
vi.doUnmock("../services/instance-settings.js");
vi.doUnmock("../services/issue-approvals.js");
@@ -315,7 +306,6 @@ describe.sequential("agent permission routes", () => {
mockHeartbeatService.resetRuntimeSession.mockReset();
mockHeartbeatService.getRun.mockReset();
mockHeartbeatService.cancelRun.mockReset();
mockRecoveryService.recordWatchdogDecision.mockReset();
mockIssueApprovalService.linkManyForApproval.mockReset();
mockIssueService.list.mockReset();
mockSecretService.normalizeAdapterConfigForPersistence.mockReset();
@@ -384,13 +374,6 @@ describe.sequential("agent permission routes", () => {
censorUsernameInLogs: false,
});
mockLogActivity.mockResolvedValue(undefined);
mockRecoveryService.recordWatchdogDecision.mockResolvedValue({
id: "watchdog-decision-1",
runId: "run-1",
evaluationIssueId: "evaluation-1",
decision: "cancel_run",
reason: "The stale run is stuck",
});
});
it("redacts agent detail for authenticated company members without agent admin permission", async () => {
@@ -1211,114 +1194,4 @@ describe.sequential("agent permission routes", () => {
expect(res.status).toBe(403);
expect(mockHeartbeatService.cancelRun).not.toHaveBeenCalled();
});
it("keeps generic heartbeat cancellation board-only", async () => {
const app = await createApp({
type: "agent",
agentId,
companyId,
source: "agent_key",
runId: "actor-run-1",
});
const res = await requestApp(app, (baseUrl) => request(baseUrl)
.post("/api/heartbeat-runs/run-1/cancel")
.send({}));
expect(res.status).toBe(403);
expect(mockHeartbeatService.cancelRun).not.toHaveBeenCalled();
});
it("lets an agent use the watchdog recovery cancel route without board access", async () => {
mockHeartbeatService.getRun.mockResolvedValue({
id: "run-1",
companyId,
agentId,
status: "running",
});
mockHeartbeatService.cancelRun.mockResolvedValue({
id: "run-1",
companyId,
agentId,
status: "cancelled",
});
const app = await createApp({
type: "agent",
agentId,
companyId,
source: "agent_key",
runId: "actor-run-1",
});
const res = await requestApp(app, (baseUrl) => request(baseUrl)
.post("/api/heartbeat-runs/run-1/watchdog-decisions")
.send({
decision: "cancel_run",
evaluationIssueId: "evaluation-1",
reason: "The stale run is stuck",
}));
expect(res.status).toBe(200);
expect(mockRecoveryService.recordWatchdogDecision).toHaveBeenCalledWith({
runId: "run-1",
actor: expect.objectContaining({
type: "agent",
agentId,
runId: "actor-run-1",
}),
decision: "cancel_run",
evaluationIssueId: "evaluation-1",
reason: "The stale run is stuck",
snoozedUntil: null,
createdByRunId: "actor-run-1",
});
expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("run-1", "The stale run is stuck");
expect(mockLogActivity).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({
actorType: "agent",
actorId: agentId,
agentId,
runId: "run-1",
action: "heartbeat.watchdog_cancelled",
entityType: "heartbeat_run",
entityId: "run-1",
details: expect.objectContaining({
decision: "cancel_run",
evaluationIssueId: "evaluation-1",
reason: "The stale run is stuck",
createdByRunId: "actor-run-1",
targetRunId: "run-1",
previousStatus: "running",
resultStatus: "cancelled",
}),
}));
});
it("requires an evaluation issue on watchdog recovery cancellation", async () => {
mockHeartbeatService.getRun.mockResolvedValue({
id: "run-1",
companyId,
agentId,
status: "running",
});
const app = await createApp({
type: "agent",
agentId,
companyId,
source: "agent_key",
runId: "actor-run-1",
});
const res = await requestApp(app, (baseUrl) => request(baseUrl)
.post("/api/heartbeat-runs/run-1/watchdog-decisions")
.send({
decision: "cancel_run",
reason: "The stale run is stuck",
}));
expect(res.status).toBe(400);
expect(mockRecoveryService.recordWatchdogDecision).not.toHaveBeenCalled();
expect(mockHeartbeatService.cancelRun).not.toHaveBeenCalled();
});
});
server/src/__tests__/heartbeat-active-run-output-watchdog.test.ts
-138
View File
@@ -2,7 +2,6 @@ import { randomUUID } from "node:crypto";
import { and, eq, sql } from "drizzle-orm";
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
import {
activityLog,
agents,
companies,
createDb,
@@ -354,87 +353,6 @@ describeEmbeddedPostgres("active-run output watchdog", () => {
});
});
it("lets the assigned recovery owner record cancel_run and uses the normal cancellation path", async () => {
const now = new Date("2026-04-22T20:00:00.000Z");
const { companyId, managerId, runId, issueId } = await seedRunningRun({
now,
ageMs: ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS + 60_000,
});
const heartbeat = heartbeatService(db);
const recovery = recoveryService(db, { enqueueWakeup: vi.fn() });
const scan = await heartbeat.scanSilentActiveRuns({ now, companyId });
const evaluationIssueId = scan.evaluationIssueIds[0];
expect(evaluationIssueId).toBeTruthy();
const managerRunId = randomUUID();
await db.insert(heartbeatRuns).values({
id: managerRunId,
companyId,
agentId: managerId,
status: "succeeded",
invocationSource: "assignment",
triggerDetail: "system",
startedAt: now,
finishedAt: now,
processStartedAt: now,
lastOutputAt: now,
lastOutputSeq: 1,
lastOutputStream: "stdout",
contextSnapshot: {},
logBytes: 0,
});
const decision = await recovery.recordWatchdogDecision({
runId,
actor: { type: "agent", agentId: managerId, runId: managerRunId },
decision: "cancel_run",
evaluationIssueId,
reason: "Run is silent and evidence shows it is stuck",
});
expect(decision).toMatchObject({
runId,
evaluationIssueId,
decision: "cancel_run",
reason: "Run is silent and evidence shows it is stuck",
createdByAgentId: managerId,
createdByRunId: managerRunId,
});
const cancelled = await heartbeat.cancelRun(runId, decision.reason ?? undefined);
expect(cancelled).toMatchObject({
id: runId,
status: "cancelled",
error: "Run is silent and evidence shows it is stuck",
errorCode: "cancelled",
});
const [sourceIssue] = await db.select().from(issues).where(eq(issues.id, issueId));
expect(sourceIssue?.executionRunId).toBeNull();
expect(sourceIssue?.executionLockedAt).toBeNull();
const [audit] = await db
.select()
.from(activityLog)
.where(and(eq(activityLog.companyId, companyId), eq(activityLog.action, "heartbeat.watchdog_cancel_requested")));
expect(audit).toMatchObject({
actorType: "agent",
actorId: managerId,
agentId: managerId,
runId,
entityType: "heartbeat_run",
entityId: runId,
});
expect(audit?.details).toMatchObject({
decision: "cancel_run",
evaluationIssueId,
targetRunId: runId,
reason: "Run is silent and evidence shows it is stuck",
createdByRunId: managerRunId,
});
});
it("re-arms continue decisions after the default quiet window", async () => {
const now = new Date("2026-04-22T20:00:00.000Z");
const { companyId, managerId, runId } = await seedRunningRun({
@@ -552,7 +470,6 @@ describeEmbeddedPostgres("active-run output watchdog", () => {
snoozedUntil: new Date(now.getTime() + 60 * 60 * 1000),
},
{ decision: "continue" as const, evaluationIssueId: otherEvaluationIssueId },
{ decision: "cancel_run" as const, evaluationIssueId: otherEvaluationIssueId },
];
for (const attempt of attempts) {
@@ -578,61 +495,6 @@ describeEmbeddedPostgres("active-run output watchdog", () => {
).rejects.toMatchObject({ status: 403 });
});
it("rejects cancel_run for terminal runs before writing a decision", async () => {
const now = new Date("2026-04-22T20:00:00.000Z");
const { companyId, managerId, runId } = await seedRunningRun({
now,
ageMs: ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS + 60_000,
});
const heartbeat = heartbeatService(db);
const recovery = recoveryService(db, { enqueueWakeup: vi.fn() });
const scan = await heartbeat.scanSilentActiveRuns({ now, companyId });
const evaluationIssueId = scan.evaluationIssueIds[0];
expect(evaluationIssueId).toBeTruthy();
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: now }).where(eq(heartbeatRuns.id, runId));
await expect(
recovery.recordWatchdogDecision({
runId,
actor: { type: "agent", agentId: managerId },
decision: "cancel_run",
evaluationIssueId,
reason: "Too late to cancel",
}),
).rejects.toMatchObject({ status: 409 });
const rows = await db
.select()
.from(heartbeatRunWatchdogDecisions)
.where(and(eq(heartbeatRunWatchdogDecisions.runId, runId), eq(heartbeatRunWatchdogDecisions.decision, "cancel_run")));
expect(rows).toHaveLength(0);
});
it("requires an evaluation issue for cancel_run decisions", async () => {
const now = new Date("2026-04-22T20:00:00.000Z");
const { managerId, runId } = await seedRunningRun({
now,
ageMs: ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS + 60_000,
});
const recovery = recoveryService(db, { enqueueWakeup: vi.fn() });
await expect(
recovery.recordWatchdogDecision({
runId,
actor: { type: "agent", agentId: managerId },
decision: "cancel_run",
reason: "No evaluation issue binding",
}),
).rejects.toMatchObject({ status: 400 });
const rows = await db
.select()
.from(heartbeatRunWatchdogDecisions)
.where(and(eq(heartbeatRunWatchdogDecisions.runId, runId), eq(heartbeatRunWatchdogDecisions.decision, "cancel_run")));
expect(rows).toHaveLength(0);
});
it("validates createdByRunId before storing watchdog decisions", async () => {
const now = new Date("2026-04-22T20:00:00.000Z");
const { companyId, managerId, runId } = await seedRunningRun({
server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts
+3 -27
View File
@@ -117,7 +117,7 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => {
});
}
async function seedBlockedChain(opts: { outsideLookback?: boolean; freshUpdateOutsideLookback?: boolean } = {}) {
async function seedBlockedChain(opts: { outsideLookback?: boolean } = {}) {
const companyId = randomUUID();
const managerId = randomUUID();
const coderId = randomUUID();
@@ -161,9 +161,6 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => {
const issueTimestamp = opts.outsideLookback === true
? new Date(Date.now() - 25 * 60 * 60 * 1000)
: new Date(Date.now() - 60 * 60 * 1000);
const updatedAt = opts.freshUpdateOutsideLookback === true
? new Date(Date.now() - 60 * 60 * 1000)
: issueTimestamp;
await db.insert(issues).values([
{
id: blockedIssueId,
@@ -175,7 +172,7 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => {
issueNumber: 1,
identifier: `${issuePrefix}-1`,
createdAt: issueTimestamp,
updatedAt,
updatedAt: issueTimestamp,
},
{
id: blockerIssueId,
@@ -186,7 +183,7 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => {
issueNumber: 2,
identifier: `${issuePrefix}-2`,
createdAt: issueTimestamp,
updatedAt,
updatedAt: issueTimestamp,
},
]);
@@ -239,27 +236,6 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => {
expect(escalations).toHaveLength(0);
});
it("does not create recovery issues for old legacy findings that were only updated recently", async () => {
await enableAutoRecovery();
const { companyId } = await seedBlockedChain({
outsideLookback: true,
freshUpdateOutsideLookback: true,
});
const heartbeat = heartbeatService(db);
const result = await heartbeat.reconcileIssueGraphLiveness();
expect(result.findings).toBe(1);
expect(result.escalationsCreated).toBe(0);
expect(result.skippedOutsideLookback).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("suppresses liveness escalation when the source issue is under an active pause hold", async () => {
await enableAutoRecovery();
const { companyId, blockedIssueId } = await seedBlockedChain();
server/src/__tests__/heartbeat-retry-scheduling.test.ts
-99
View File
@@ -62,105 +62,6 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => {
await tempDb?.cleanup();
});
it("cancels queued issue runs and deferred wakeups when an issue is cancelled", async () => {
const companyId = randomUUID();
const agentId = randomUUID();
const issueId = randomUUID();
const queuedWakeupId = randomUUID();
const deferredWakeupId = randomUUID();
const queuedRunId = randomUUID();
await db.insert(companies).values({
id: companyId,
name: "Paperclip",
issuePrefix: "TQUEUE",
requireBoardApprovalForNewAgents: false,
});
await db.insert(agents).values({
id: agentId,
companyId,
name: "CodexCoder",
role: "engineer",
status: "active",
adapterType: "codex_local",
adapterConfig: {},
runtimeConfig: {
heartbeat: {
wakeOnDemand: true,
maxConcurrentRuns: 1,
},
},
permissions: {},
});
await db.insert(issues).values({
id: issueId,
companyId,
title: "Cancel queued work",
status: "cancelled",
priority: "medium",
assigneeAgentId: agentId,
cancelledAt: new Date(),
});
await db.insert(agentWakeupRequests).values([
{
id: queuedWakeupId,
companyId,
agentId,
source: "automation",
triggerDetail: "system",
reason: "issue_commented",
payload: { issueId },
status: "queued",
},
{
id: deferredWakeupId,
companyId,
agentId,
source: "automation",
triggerDetail: "system",
reason: "issue_execution_deferred",
payload: { issueId },
status: "deferred_issue_execution",
},
]);
await db.insert(heartbeatRuns).values({
id: queuedRunId,
companyId,
agentId,
invocationSource: "automation",
triggerDetail: "system",
status: "queued",
wakeupRequestId: queuedWakeupId,
contextSnapshot: { issueId, wakeCommentId: randomUUID() },
});
await db
.update(agentWakeupRequests)
.set({ runId: queuedRunId })
.where(eq(agentWakeupRequests.id, queuedWakeupId));
const result = await heartbeat.cancelIssueRuns({
companyId,
issueId,
reason: "Cancelled because the issue was marked cancelled",
});
expect(result.runs.map((run) => run.id)).toEqual([queuedRunId]);
expect(result.wakeups.map((wakeup) => wakeup.id)).toEqual([deferredWakeupId]);
const [queuedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, queuedRunId));
const wakeups = await db
.select()
.from(agentWakeupRequests)
.where(eq(agentWakeupRequests.companyId, companyId));
expect(queuedRun?.status).toBe("cancelled");
expect(queuedRun?.errorCode).toBe("cancelled");
expect(Object.fromEntries(wakeups.map((wakeup) => [wakeup.id, wakeup.status]))).toEqual({
[queuedWakeupId]: "cancelled",
[deferredWakeupId]: "cancelled",
});
});
async function seedRetryFixture(input: {
runId: string;
companyId: string;
server/src/__tests__/issue-blocker-attention.test.ts
-54
View File
@@ -82,9 +82,6 @@ describeEmbeddedPostgres("issue blocker attention", () => {
status: string;
parentId?: string | null;
assigneeAgentId?: string | null;
originKind?: string | null;
originId?: string | null;
originFingerprint?: string | null;
}) {
const id = input.id ?? randomUUID();
await db.insert(issues).values({
@@ -96,9 +93,6 @@ describeEmbeddedPostgres("issue blocker attention", () => {
priority: "medium",
parentId: input.parentId ?? null,
assigneeAgentId: input.assigneeAgentId ?? null,
originKind: input.originKind ?? "manual",
originId: input.originId ?? null,
originFingerprint: input.originFingerprint ?? "default",
});
return id;
}
@@ -418,54 +412,6 @@ describeEmbeddedPostgres("issue blocker attention", () => {
});
});
it("treats open liveness escalation blockers as explicit waiting paths", async () => {
const { companyId, agentId } = await createCompany("PBL");
const parentId = await insertIssue({ companyId, identifier: "PBL-1", title: "Parent", status: "blocked" });
const cancelledLeafId = await insertIssue({
companyId,
identifier: "PBL-2",
title: "Cancelled blocker",
status: "cancelled",
assigneeAgentId: agentId,
});
const incidentKey = [
"harness_liveness",
companyId,
parentId,
"blocked_by_cancelled_issue",
cancelledLeafId,
].join(":");
const escalationId = await insertIssue({
companyId,
identifier: "PBL-3",
title: "Liveness escalation",
status: "todo",
assigneeAgentId: agentId,
originKind: "harness_liveness_escalation",
originId: incidentKey,
originFingerprint: [
"harness_liveness_leaf",
companyId,
"blocked_by_cancelled_issue",
cancelledLeafId,
].join(":"),
});
await block({ companyId, blockerIssueId: cancelledLeafId, blockedIssueId: parentId });
await block({ companyId, blockerIssueId: escalationId, blockedIssueId: parentId });
const parent = (await svc.list(companyId, { status: "blocked,todo" })).find((issue) => issue.id === parentId);
expect(parent?.blockerAttention).toMatchObject({
state: "covered",
reason: "explicit_waiting",
unresolvedBlockerCount: 2,
coveredBlockerCount: 2,
attentionBlockerCount: 0,
nextActionHint: "needs_human_review",
});
expect(parent?.blockerAttention?.nextActionOwner).toMatchObject({ type: "agent", agentId });
});
async function recordedRun(input: {
companyId: string;
agentId: string;
server/src/__tests__/issue-comment-reopen-routes.test.ts
+13 -18
View File
@@ -24,7 +24,6 @@ const mockHeartbeatService = vi.hoisted(() => ({
getRun: vi.fn(async () => null),
getActiveRunForAgent: vi.fn(async () => null),
cancelRun: vi.fn(async () => null),
cancelIssueRuns: vi.fn(async () => ({ runs: [], wakeups: [] })),
}));
const mockAgentService = vi.hoisted(() => ({
@@ -111,7 +110,6 @@ vi.mock("../services/routines.js", () => ({
vi.mock("../services/index.js", () => ({
accessService: () => mockAccessService,
agentService: () => mockAgentService,
companyService: () => ({}),
documentService: () => ({}),
executionWorkspaceService: () => ({}),
feedbackService: () => mockFeedbackService,
@@ -223,7 +221,6 @@ describe.sequential("issue comment reopen routes", () => {
mockHeartbeatService.getRun.mockReset();
mockHeartbeatService.getActiveRunForAgent.mockReset();
mockHeartbeatService.cancelRun.mockReset();
mockHeartbeatService.cancelIssueRuns.mockReset();
mockAgentService.getById.mockReset();
mockAgentService.list.mockReset();
mockAgentService.resolveByReference.mockReset();
@@ -245,7 +242,6 @@ describe.sequential("issue comment reopen routes", () => {
mockHeartbeatService.getRun.mockResolvedValue(null);
mockHeartbeatService.getActiveRunForAgent.mockResolvedValue(null);
mockHeartbeatService.cancelRun.mockResolvedValue(null);
mockHeartbeatService.cancelIssueRuns.mockResolvedValue({ runs: [], wakeups: [] });
mockLogActivity.mockResolvedValue(undefined);
mockFeedbackService.listIssueVotesForUser.mockResolvedValue([]);
mockFeedbackService.saveIssueVote.mockResolvedValue({
@@ -971,14 +967,17 @@ describe.sequential("issue comment reopen routes", () => {
...issue,
...patch,
}));
mockHeartbeatService.cancelIssueRuns.mockResolvedValue({
runs: [{
id: "run-1",
companyId: "company-1",
agentId: "22222222-2222-4222-8222-222222222222",
status: "cancelled",
}],
wakeups: [],
mockHeartbeatService.getRun.mockResolvedValue({
id: "run-1",
companyId: "company-1",
agentId: "22222222-2222-4222-8222-222222222222",
status: "running",
});
mockHeartbeatService.cancelRun.mockResolvedValue({
id: "run-1",
companyId: "company-1",
agentId: "22222222-2222-4222-8222-222222222222",
status: "cancelled",
});
const res = await request(await installActor(createApp()))
@@ -986,11 +985,8 @@ describe.sequential("issue comment reopen routes", () => {
.send({ status: "cancelled" });
expect(res.status).toBe(200);
expect(mockHeartbeatService.cancelIssueRuns).toHaveBeenCalledWith({
companyId: "company-1",
issueId: "11111111-1111-4111-8111-111111111111",
reason: "Cancelled because the issue was marked cancelled",
});
expect(mockHeartbeatService.getRun).toHaveBeenCalledWith("run-1");
expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("run-1");
expect(mockLogActivity).toHaveBeenCalledWith(
expect.anything(),
expect.objectContaining({
@@ -1026,7 +1022,6 @@ describe.sequential("issue comment reopen routes", () => {
expect(res.status).toBe(200);
expect(mockHeartbeatService.cancelRun).not.toHaveBeenCalled();
expect(mockHeartbeatService.cancelIssueRuns).not.toHaveBeenCalled();
});
it("writes decision ids into executionState and inserts the decision inside the transaction", async () => {
server/src/__tests__/issue-execution-disposition-resolver.test.ts
-359
View File
@@ -1,359 +0,0 @@
import { randomUUID } from "node:crypto";
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
import {
agents,
agentWakeupRequests,
approvals,
companies,
createDb,
heartbeatRuns,
issueApprovals,
issueRelations,
issueThreadInteractions,
issues,
} from "@paperclipai/db";
import {
getEmbeddedPostgresTestSupport,
startEmbeddedPostgresTestDatabase,
} from "./helpers/embedded-postgres.js";
import {
backfillBlockedBySummaries,
listIssueExecutionDispositionsMap,
} from "../services/issue-execution-disposition-resolver.js";
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
if (!embeddedPostgresSupport.supported) {
console.warn(
`Skipping embedded Postgres execution-disposition resolver tests: ${embeddedPostgresSupport.reason ?? "unsupported"}`,
);
}
describeEmbeddedPostgres("issue execution disposition resolver", () => {
let db!: ReturnType<typeof createDb>;
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
beforeAll(async () => {
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-execution-disposition-");
db = createDb(tempDb.connectionString);
}, 20_000);
afterEach(async () => {
await db.delete(heartbeatRuns);
await db.delete(agentWakeupRequests);
await db.delete(issueThreadInteractions);
await db.delete(issueApprovals);
await db.delete(approvals);
await db.delete(issueRelations);
await db.delete(issues);
await db.delete(agents);
await db.delete(companies);
});
afterAll(async () => {
await tempDb?.cleanup();
});
async function seed(prefix = "DSP") {
const companyId = randomUUID();
const agentId = randomUUID();
await db.insert(companies).values({
id: companyId,
name: `Company ${prefix}`,
issuePrefix: prefix,
requireBoardApprovalForNewAgents: false,
});
await db.insert(agents).values({
id: agentId,
companyId,
name: `${prefix} Agent`,
role: "engineer",
status: "idle",
});
return { companyId, agentId };
}
async function insertIssue(input: {
companyId: string;
identifier: string;
title: string;
status: string;
assigneeAgentId?: string | null;
assigneeUserId?: string | null;
originKind?: string | null;
originId?: string | null;
originFingerprint?: string | null;
}) {
const id = randomUUID();
await db.insert(issues).values({
id,
companyId: input.companyId,
identifier: input.identifier,
title: input.title,
status: input.status,
priority: "medium",
assigneeAgentId: input.assigneeAgentId ?? null,
assigneeUserId: input.assigneeUserId ?? null,
originKind: input.originKind ?? "manual",
originId: input.originId ?? null,
originFingerprint: input.originFingerprint ?? "default",
});
return id;
}
function inputFor(input: {
id: string;
companyId: string;
status: string;
assigneeAgentId?: string | null;
assigneeUserId?: string | null;
originKind?: string | null;
}) {
return {
id: input.id,
companyId: input.companyId,
status: input.status,
assigneeAgentId: input.assigneeAgentId ?? null,
assigneeUserId: input.assigneeUserId ?? null,
originKind: input.originKind ?? "manual",
executionRunId: null as string | null,
executionState: null as Record<string, unknown> | null,
};
}
it("classifies an idle agent-owned todo as dispatchable", async () => {
const { companyId, agentId } = await seed();
const issueId = await insertIssue({
companyId,
identifier: "DSP-1",
title: "Ready",
status: "todo",
assigneeAgentId: agentId,
});
const map = await listIssueExecutionDispositionsMap(db, companyId, [
inputFor({ id: issueId, companyId, status: "todo", assigneeAgentId: agentId }),
]);
const disposition = map.get(issueId);
expect(disposition?.kind).toBe("dispatchable");
});
it("classifies an in_progress issue with an active heartbeat run as live", async () => {
const { companyId, agentId } = await seed();
const issueId = await insertIssue({
companyId,
identifier: "DSP-2",
title: "Running",
status: "in_progress",
assigneeAgentId: agentId,
});
const runId = randomUUID();
await db.insert(heartbeatRuns).values({
id: runId,
companyId,
agentId,
status: "running",
contextSnapshot: { issueId },
});
const map = await listIssueExecutionDispositionsMap(db, companyId, [
inputFor({ id: issueId, companyId, status: "in_progress", assigneeAgentId: agentId }),
]);
expect(map.get(issueId)?.kind).toBe("live");
});
it("classifies a pending interaction as waiting on interaction", async () => {
const { companyId, agentId } = await seed();
const issueId = await insertIssue({
companyId,
identifier: "DSP-3",
title: "Waiting on confirmation",
status: "in_review",
assigneeAgentId: agentId,
});
await db.insert(issueThreadInteractions).values({
id: randomUUID(),
companyId,
issueId,
kind: "request_confirmation",
status: "pending",
payload: { target: { type: "issue_document", key: "plan", revisionId: "r1" } },
createdByAgentId: agentId,
});
const map = await listIssueExecutionDispositionsMap(db, companyId, [
inputFor({ id: issueId, companyId, status: "in_review", assigneeAgentId: agentId }),
]);
const disposition = map.get(issueId);
expect(disposition?.kind).toBe("waiting");
if (disposition?.kind === "waiting") {
expect(disposition.path).toBe("interaction");
}
});
it("flags an in_review issue without any action path as invalid", async () => {
const { companyId, agentId } = await seed();
const issueId = await insertIssue({
companyId,
identifier: "DSP-4",
title: "Stalled review",
status: "in_review",
assigneeAgentId: agentId,
});
const map = await listIssueExecutionDispositionsMap(db, companyId, [
inputFor({ id: issueId, companyId, status: "in_review", assigneeAgentId: agentId }),
]);
const disposition = map.get(issueId);
expect(disposition?.kind).toBe("invalid");
if (disposition?.kind === "invalid") {
expect(disposition.reason).toBe("in_review_without_action_path");
}
});
it("classifies a blocked issue with a healthy chain leaf as waiting on blocker chain", async () => {
const { companyId, agentId } = await seed();
const blockerId = await insertIssue({
companyId,
identifier: "DSP-5b",
title: "Blocker",
status: "in_progress",
assigneeAgentId: agentId,
});
const blockedId = await insertIssue({
companyId,
identifier: "DSP-5",
title: "Blocked",
status: "blocked",
assigneeAgentId: agentId,
});
await db.insert(issueRelations).values({
id: randomUUID(),
companyId,
issueId: blockerId,
relatedIssueId: blockedId,
type: "blocks",
});
const runId = randomUUID();
await db.insert(heartbeatRuns).values({
id: runId,
companyId,
agentId,
status: "running",
contextSnapshot: { issueId: blockerId },
});
const inputs = [
inputFor({ id: blockedId, companyId, status: "blocked", assigneeAgentId: agentId }),
];
await backfillBlockedBySummaries(db, companyId, inputs);
const map = await listIssueExecutionDispositionsMap(db, companyId, inputs);
const disposition = map.get(blockedId);
expect(disposition?.kind).toBe("waiting");
if (disposition?.kind === "waiting") {
expect(disposition.path).toBe("blocker_chain");
}
});
it("flags a blocked issue with a cancelled blocker as invalid", async () => {
const { companyId, agentId } = await seed();
const blockerId = await insertIssue({
companyId,
identifier: "DSP-6b",
title: "Cancelled blocker",
status: "cancelled",
assigneeAgentId: agentId,
});
const blockedId = await insertIssue({
companyId,
identifier: "DSP-6",
title: "Blocked by cancelled",
status: "blocked",
assigneeAgentId: agentId,
});
await db.insert(issueRelations).values({
id: randomUUID(),
companyId,
issueId: blockerId,
relatedIssueId: blockedId,
type: "blocks",
});
const inputs = [
inputFor({ id: blockedId, companyId, status: "blocked", assigneeAgentId: agentId }),
];
await backfillBlockedBySummaries(db, companyId, inputs);
const map = await listIssueExecutionDispositionsMap(db, companyId, inputs);
const disposition = map.get(blockedId);
// The cancelled blocker is filtered out by ne(status, "done") + companyId checks; backfill
// excludes it because it's not "open". So the issue ends up blocked-without-an-action-path,
// which is also invalid. Either invalid form is acceptable here; assert the kind only.
expect(disposition?.kind).toBe("invalid");
});
it("treats an open liveness escalation as a waiting path for a cancelled blocker incident", async () => {
const { companyId, agentId } = await seed();
const blockerId = await insertIssue({
companyId,
identifier: "DSP-8b",
title: "Cancelled blocker",
status: "cancelled",
assigneeAgentId: agentId,
});
const blockedId = await insertIssue({
companyId,
identifier: "DSP-8",
title: "Blocked by cancelled with escalation",
status: "blocked",
assigneeAgentId: agentId,
});
const incidentKey = [
"harness_liveness",
companyId,
blockedId,
"blocked_by_cancelled_issue",
blockerId,
].join(":");
const escalationId = await insertIssue({
companyId,
identifier: "DSP-8r",
title: "Liveness escalation",
status: "todo",
assigneeAgentId: agentId,
originKind: "harness_liveness_escalation",
originId: incidentKey,
originFingerprint: [
"harness_liveness_leaf",
companyId,
"blocked_by_cancelled_issue",
blockerId,
].join(":"),
});
await db.insert(issueRelations).values([
{
id: randomUUID(),
companyId,
issueId: blockerId,
relatedIssueId: blockedId,
type: "blocks",
},
{
id: randomUUID(),
companyId,
issueId: escalationId,
relatedIssueId: blockedId,
type: "blocks",
},
]);
const inputs = [
inputFor({ id: blockedId, companyId, status: "blocked", assigneeAgentId: agentId }),
];
await backfillBlockedBySummaries(db, companyId, inputs);
const map = await listIssueExecutionDispositionsMap(db, companyId, inputs);
const disposition = map.get(blockedId);
expect(disposition?.kind).toBe("waiting");
if (disposition?.kind === "waiting") {
expect(disposition.path).toBe("review_artifact");
}
});
});
server/src/__tests__/issue-execution-disposition.test.ts
-525
View File
@@ -1,525 +0,0 @@
import { describe, expect, it } from "vitest";
import type { IssueStatus } from "@paperclipai/shared";
import type {
IssueExecutionDisposition,
IssueExecutionInvalidReason,
IssueExecutionWaitingPath,
} from "@paperclipai/shared";
import {
classifyIssueExecutionDisposition,
type IssueExecutionStateVector,
} from "../services/issue-execution-disposition.ts";
const agentId = "agent-1";
const humanId = "user-1";
function state(
input: Partial<IssueExecutionStateVector> & {
issue?: Partial<IssueExecutionStateVector["issue"]>;
},
): IssueExecutionStateVector {
const issue = {
id: "issue-1",
status: "todo" as IssueStatus,
assigneeAgentId: agentId,
assigneeUserId: null,
...input.issue,
};
return {
agent: issue.assigneeAgentId
? { id: issue.assigneeAgentId, status: "idle" }
: null,
...input,
issue,
};
}
function classify(input: IssueExecutionStateVector) {
const disposition = classifyIssueExecutionDisposition(input);
expect(disposition.kind).toBeTruthy();
return disposition;
}
function expectDisposition(
input: IssueExecutionStateVector,
expected: Partial<IssueExecutionDisposition> & { kind: IssueExecutionDisposition["kind"] },
) {
expect(classify(input)).toMatchObject(expected);
}
describe("issue execution disposition classifier", () => {
it.each([
["row 1 terminal done", state({ issue: { status: "done" } }), { kind: "terminal" }],
[
"row 1 terminal cancelled",
state({ issue: { status: "cancelled" } }),
{ kind: "terminal" },
],
[
"row 2 dual owner",
state({ issue: { assigneeAgentId: agentId, assigneeUserId: humanId } }),
{ kind: "invalid", reason: "dual_assignee" },
],
[
"row 3 backlog rests",
state({ issue: { status: "backlog" } }),
{ kind: "resting" },
],
[
"row 4 unassigned todo rests",
state({ issue: { assigneeAgentId: null, assigneeUserId: null }, agent: null }),
{ kind: "resting" },
],
[
"row 5 queued todo is live",
state({ execution: { queuedWake: true } }),
{ kind: "live", path: "queued_wake" },
],
["row 6 assigned todo dispatches", state({}), { kind: "dispatchable", wakeTarget: agentId }],
[
"row 7 assigned todo waits on interaction",
state({ waits: { pendingInteraction: "live" } }),
{ kind: "waiting", path: "interaction" },
],
[
"row 8 failed todo is dispatch recoverable",
state({ latestRun: { latestRunStatus: "failed", recoveryAttemptRemaining: true } }),
{ kind: "recoverable_by_control_plane", recovery: "dispatch" },
],
[
"row 9 paused assignee escalates",
state({ agent: { id: agentId, status: "paused" } }),
{ kind: "human_escalation_required", owner: "manager" },
],
[
"row 10 pause hold waits",
state({ issue: { status: "in_progress" }, waits: { pauseHold: true } }),
{ kind: "waiting", path: "pause_hold" },
],
[
"row 11 active in-progress run is live",
state({ issue: { status: "in_progress" }, execution: { activeRun: true } }),
{ kind: "live", path: "active_run" },
],
[
"row 12 queued in-progress continuation is live",
state({ issue: { status: "in_progress" }, execution: { scheduledRetry: true } }),
{ kind: "live", path: "scheduled_retry" },
],
[
"row 13 in-progress waits on approval",
state({ issue: { status: "in_progress" }, waits: { pendingApproval: "live" } }),
{ kind: "waiting", path: "approval" },
],
[
"row 14 failed in-progress run is continuation recoverable",
state({
issue: { status: "in_progress" },
latestRun: { latestRunStatus: "timed_out", recoveryAttemptRemaining: true },
}),
{ kind: "recoverable_by_control_plane", recovery: "continuation" },
],
[
"row 15 successful runnable progress is continuable",
state({
issue: { status: "in_progress" },
latestRun: {
latestRunStatus: "succeeded",
livenessState: "advanced",
nextAction: "runnable",
continuationAttempt: 0,
maxContinuationAttempts: 2,
},
}),
{ kind: "agent_continuable", continuationAttempt: 0, maxAttempts: 2 },
],
[
"row 15b low-progress terminal run is continuable",
state({
issue: { status: "in_progress" },
latestRun: {
latestRunStatus: "succeeded",
livenessState: "plan_only",
nextAction: "none",
continuationAttempt: 0,
maxContinuationAttempts: 2,
},
}),
{ kind: "agent_continuable", continuationAttempt: 0, maxAttempts: 2 },
],
[
"row 16 ambiguous successful progress escalates",
state({
issue: { status: "in_progress" },
latestRun: {
latestRunStatus: "succeeded",
livenessState: "advanced",
nextAction: "manager_review",
continuationAttempt: 0,
maxContinuationAttempts: 2,
},
}),
{ kind: "human_escalation_required", owner: "manager" },
],
[
"row 17 human review assignee waits",
state({
issue: { status: "in_review", assigneeAgentId: null, assigneeUserId: humanId },
agent: null,
}),
{ kind: "waiting", path: "human_owner" },
],
[
"row 18 agent review participant waits",
state({ issue: { status: "in_review" }, waits: { participant: "valid" } }),
{ kind: "waiting", path: "participant" },
],
[
"row 19 bare agent review is invalid",
state({ issue: { status: "in_review" } }),
{ kind: "invalid", reason: "in_review_without_action_path" },
],
[
"row 20 healthy blocker chain waits",
state({
issue: { status: "blocked" },
blockers: [
{
issue: {
id: "blocker-1",
status: "todo",
assigneeAgentId: agentId,
assigneeUserId: null,
},
disposition: { kind: "dispatchable", wakeTarget: agentId },
},
],
}),
{ kind: "waiting", path: "blocker_chain" },
],
[
"row 21 cancelled blocker leaf is invalid",
state({
issue: { status: "blocked" },
blockers: [
{
issue: {
id: "blocker-1",
status: "cancelled",
assigneeAgentId: agentId,
assigneeUserId: null,
},
},
],
}),
{ kind: "invalid", reason: "blocked_by_cancelled_issue" },
],
[
"row 22 blocked external owner waits",
state({ issue: { status: "blocked" }, waits: { externalOwnerAction: true } }),
{ kind: "waiting", path: "external_owner_action" },
],
[
"row 23 blocked without path is invalid",
state({ issue: { status: "blocked" } }),
{ kind: "invalid", reason: "blocked_without_action_path" },
],
[
"row 24 human-owned non-terminal waits",
state({
issue: { status: "todo", assigneeAgentId: null, assigneeUserId: humanId },
agent: null,
}),
{ kind: "waiting", path: "human_owner" },
],
[
"row 25 open productivity review waits",
state({ issue: { status: "in_progress" }, waits: { openProductivityReviewIssue: true } }),
{ kind: "waiting", path: "review_artifact" },
],
[
"row 26 recovery issue failure escalates to recovery owner",
state({
issue: { status: "in_progress", originKind: "stranded_issue_recovery" },
latestRun: { latestRunStatus: "failed", recoveryAttemptRemaining: true },
}),
{ kind: "human_escalation_required", owner: "recovery_owner" },
],
[
"row 27 todo recovery issue failure escalates in place",
state({
issue: { status: "todo", originKind: "stranded_issue_recovery" },
latestRun: { latestRunStatus: "failed", recoveryAttemptRemaining: true },
}),
{ kind: "human_escalation_required", owner: "recovery_owner" },
],
] satisfies Array<[string, IssueExecutionStateVector, Partial<IssueExecutionDisposition> & { kind: IssueExecutionDisposition["kind"] }]>)(
"%s",
(_name, input, expected) => {
expectDisposition(input, expected);
},
);
it.each([
[
"invalid-review-leaf-pi-autoresearch",
state({ issue: { status: "in_review" } }),
"invalid",
"in_review_without_action_path",
undefined,
],
[
"invalid-review-leaf-object-detection",
state({ issue: { status: "in_review" } }),
"invalid",
"in_review_without_action_path",
undefined,
],
[
"healthy-review-participant",
state({ issue: { status: "in_review" }, waits: { participant: "valid" } }),
"waiting",
undefined,
"participant",
],
[
"healthy-review-confirmation",
state({ issue: { status: "in_progress" }, waits: { pendingInteraction: "live" } }),
"waiting",
undefined,
"interaction",
],
[
"healthy-linked-approval",
state({ waits: { pendingApproval: "live" } }),
"waiting",
undefined,
"approval",
],
[
"productive-terminal-run-runnable-next-action",
state({
issue: { status: "in_progress" },
latestRun: {
latestRunStatus: "succeeded",
livenessState: "advanced",
nextAction: "runnable",
continuationAttempt: 1,
maxContinuationAttempts: 2,
},
}),
"agent_continuable",
undefined,
undefined,
],
[
"productive-terminal-run-no-next-action",
state({
issue: { status: "in_progress" },
latestRun: {
latestRunStatus: "succeeded",
livenessState: "advanced",
nextAction: "unknown",
continuationAttempt: 1,
maxContinuationAttempts: 2,
},
}),
"human_escalation_required",
undefined,
undefined,
],
[
"true-failed-continuation-recovery",
state({
issue: { status: "in_progress" },
latestRun: { latestRunStatus: "cancelled", recoveryAttemptRemaining: true },
}),
"recoverable_by_control_plane",
undefined,
undefined,
],
[
"explicit-productivity-review-hold",
state({ issue: { status: "in_progress" }, waits: { openProductivityReviewIssue: true } }),
"waiting",
undefined,
"review_artifact",
],
[
"productivity-threshold-without-review",
state({ issue: { status: "in_progress" }, waits: { productivityReviewNeeded: true } }),
"human_escalation_required",
undefined,
undefined,
],
[
"healthy-long-active-run",
state({ issue: { status: "in_progress" }, execution: { activeRun: true } }),
"live",
undefined,
undefined,
],
[
"recovery-of-recovery-failure",
state({
issue: { status: "in_progress", originKind: "stranded_issue_recovery" },
latestRun: { latestRunStatus: "failed", recoveryAttemptRemaining: true },
}),
"human_escalation_required",
undefined,
undefined,
],
[
"normal-source-recovery-dedupe",
state({
issue: { status: "in_progress" },
latestRun: { latestRunStatus: "failed", recoveryAttemptRemaining: false },
}),
"human_escalation_required",
undefined,
undefined,
],
[
"blocked-chain-healthy-leaf",
state({
issue: { status: "blocked" },
blockers: [
{
issue: {
id: "blocker-1",
status: "in_progress",
assigneeAgentId: agentId,
assigneeUserId: null,
},
disposition: { kind: "live", path: "active_run" },
},
],
}),
"waiting",
undefined,
"blocker_chain",
],
[
"blocked-chain-cancelled-leaf",
state({
issue: { status: "blocked" },
blockers: [
{
issue: {
id: "blocker-1",
status: "cancelled",
assigneeAgentId: agentId,
assigneeUserId: null,
},
},
],
}),
"invalid",
"blocked_by_cancelled_issue",
undefined,
],
[
"budget-hard-stop",
state({ issue: { status: "in_progress" }, gates: { budgetBlocked: true } }),
"human_escalation_required",
undefined,
undefined,
],
[
"pause-held-subtree",
state({ issue: { status: "in_progress" }, waits: { pauseHold: true } }),
"waiting",
undefined,
"pause_hold",
],
[
"dual-assignee-write",
state({ issue: { assigneeAgentId: agentId, assigneeUserId: humanId } }),
"invalid",
"dual_assignee",
undefined,
],
] satisfies Array<
[
string,
IssueExecutionStateVector,
IssueExecutionDisposition["kind"],
IssueExecutionInvalidReason | undefined,
IssueExecutionWaitingPath | undefined,
]
>)("%s fixture", (_name, input, expectedKind, expectedReason, expectedPath) => {
const disposition = classify(input);
expect(disposition.kind).toBe(expectedKind);
if (expectedReason) expect(disposition).toMatchObject({ reason: expectedReason });
if (expectedPath) expect(disposition).toMatchObject({ path: expectedPath });
});
it("does not treat stale waits as durable coverage", () => {
expectDisposition(
state({ issue: { status: "in_review" }, waits: { pendingInteraction: "stale" } }),
{ kind: "invalid", reason: "in_review_without_action_path" },
);
expectDisposition(
state({
issue: { status: "in_progress" },
waits: { pendingApproval: "stale" },
latestRun: {
latestRunStatus: "succeeded",
livenessState: "advanced",
nextAction: "runnable",
continuationAttempt: 0,
maxContinuationAttempts: 2,
},
}),
{ kind: "agent_continuable", continuationAttempt: 0, maxAttempts: 2 },
);
});
it("rejects malformed review participants as invalid wait paths", () => {
expectDisposition(
state({ issue: { status: "in_review" }, waits: { participant: "invalid" } }),
{ kind: "invalid", reason: "invalid_review_participant" },
);
});
it("surfaces invalid and unassigned blocker leaves instead of covering parents", () => {
expectDisposition(
state({
issue: { status: "blocked" },
blockers: [
{
issue: {
id: "blocker-1",
status: "in_review",
assigneeAgentId: agentId,
assigneeUserId: null,
},
disposition: {
kind: "invalid",
reason: "in_review_without_action_path",
suggestedCorrection: "Add an explicit wait path.",
},
},
],
}),
{ kind: "invalid", reason: "blocked_by_invalid_issue" },
);
expectDisposition(
state({
issue: { status: "blocked" },
blockers: [
{
issue: {
id: "blocker-1",
status: "todo",
assigneeAgentId: null,
assigneeUserId: null,
},
},
],
}),
{ kind: "invalid", reason: "blocked_by_unassigned_issue" },
);
});
});
server/src/__tests__/issue-transition-guard.test.ts
-166
View File
@@ -1,166 +0,0 @@
import { randomUUID } from "node:crypto";
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
import {
agents,
companies,
createDb,
issueRelations,
issues,
} from "@paperclipai/db";
import {
getEmbeddedPostgresTestSupport,
startEmbeddedPostgresTestDatabase,
} from "./helpers/embedded-postgres.js";
import { issueService } from "../services/issues.js";
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
if (!embeddedPostgresSupport.supported) {
console.warn(
`Skipping embedded Postgres issue transition guard tests on this host: ${
embeddedPostgresSupport.reason ?? "unsupported environment"
}`,
);
}
describeEmbeddedPostgres("issue transition guard", () => {
let db!: ReturnType<typeof createDb>;
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
beforeAll(async () => {
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-transition-guard-");
db = createDb(tempDb.connectionString);
}, 20_000);
afterEach(async () => {
await db.delete(issueRelations);
await db.delete(issues);
await db.delete(agents);
await db.delete(companies);
});
afterAll(async () => {
await tempDb?.cleanup();
});
async function seedCompanyAndAgents() {
const companyId = randomUUID();
const coderAgentId = randomUUID();
const reviewerAgentId = randomUUID();
await db.insert(companies).values({
id: companyId,
name: "Paperclip",
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
requireBoardApprovalForNewAgents: false,
});
await db.insert(agents).values([
{
id: coderAgentId,
companyId,
name: "CodexCoder",
role: "engineer",
status: "active",
adapterType: "codex_local",
adapterConfig: {},
runtimeConfig: {},
permissions: {},
},
{
id: reviewerAgentId,
companyId,
name: "Reviewer",
role: "qa",
status: "active",
adapterType: "codex_local",
adapterConfig: {},
runtimeConfig: {},
permissions: {},
},
]);
return { companyId, coderAgentId, reviewerAgentId };
}
it("rejects bare agent-owned in_review creates", async () => {
const { companyId, coderAgentId } = await seedCompanyAndAgents();
const svc = issueService(db);
await expect(
svc.create(companyId, {
title: "Bare review",
status: "in_review",
assigneeAgentId: coderAgentId,
}),
).rejects.toMatchObject({
status: 422,
message: "Invalid issue transition: in_review_without_action_path",
});
});
it("rejects blocked updates without blockers or a wait path", async () => {
const { companyId, coderAgentId } = await seedCompanyAndAgents();
const svc = issueService(db);
const issue = await svc.create(companyId, {
title: "Needs a real blocker",
status: "todo",
assigneeAgentId: coderAgentId,
});
await expect(svc.update(issue.id, { status: "blocked" })).rejects.toMatchObject({
status: 422,
message: "Invalid issue transition: blocked_without_action_path",
});
});
it("accepts blocked writes with a healthy first-class blocker chain", async () => {
const { companyId, coderAgentId } = await seedCompanyAndAgents();
const svc = issueService(db);
const blocker = await svc.create(companyId, {
title: "Do this first",
status: "todo",
assigneeAgentId: coderAgentId,
});
const issue = await svc.create(companyId, {
title: "Wait on blocker",
status: "blocked",
assigneeAgentId: coderAgentId,
blockedByIssueIds: [blocker.id],
});
expect(issue.status).toBe("blocked");
});
it("accepts workflow-normalized review state with a typed participant", async () => {
const { companyId, coderAgentId, reviewerAgentId } = await seedCompanyAndAgents();
const svc = issueService(db);
const issue = await svc.create(companyId, {
title: "Review me",
status: "in_progress",
assigneeAgentId: coderAgentId,
});
const updated = await svc.update(issue.id, {
status: "in_review",
assigneeAgentId: reviewerAgentId,
executionState: {
status: "pending",
currentStageId: randomUUID(),
currentStageIndex: 0,
currentStageType: "review",
currentParticipant: { type: "agent", agentId: reviewerAgentId, userId: null },
returnAssignee: { type: "agent", agentId: coderAgentId, userId: null },
reviewRequest: null,
completedStageIds: [],
lastDecisionId: null,
lastDecisionOutcome: null,
},
});
expect(updated).toMatchObject({
id: issue.id,
status: "in_review",
assigneeAgentId: reviewerAgentId,
});
});
});
server/src/__tests__/recovery-classifiers.test.ts
-21
View File
@@ -1,6 +1,5 @@
import { describe, expect, it } from "vitest";
import { classifyIssueGraphLiveness as classifyIssueGraphLivenessCompat } from "../services/issue-liveness.ts";
import { classifyIssueExecutionDisposition } from "../services/issue-execution-disposition.ts";
import { decideRunLivenessContinuation as decideRunLivenessContinuationCompat } from "../services/run-continuations.ts";
import {
RECOVERY_KEY_PREFIXES,
@@ -152,24 +151,4 @@ describe("recovery classifier boundary", () => {
expect(isStrandedIssueRecoveryOriginKind("manual")).toBe(false);
expect(isStrandedIssueRecoveryOriginKind(null)).toBe(false);
});
it("aligns invalid review handoff parity with the canonical disposition classifier", () => {
const disposition = classifyIssueExecutionDisposition({
issue: {
id: issueId,
status: "in_review",
assigneeAgentId: agentId,
assigneeUserId: null,
},
agent: {
id: agentId,
status: "idle",
},
});
expect(disposition).toMatchObject({
kind: "invalid",
reason: "in_review_without_action_path",
});
});
});
server/src/routes/agents.ts
+4 -48
View File
@@ -85,9 +85,6 @@ import { recoveryService } from "../services/recovery/service.js";
const RUN_LOG_DEFAULT_LIMIT_BYTES = 256_000;
const RUN_LOG_MAX_LIMIT_BYTES = 1024 * 1024;
const LIVE_RUNS_DEFAULT_LIMIT = 50;
const WATCHDOG_DECISIONS = ["snooze", "continue", "dismissed_false_positive", "cancel_run"] as const;
type WatchdogDecision = (typeof WATCHDOG_DECISIONS)[number];
function readRunLogLimitBytes(value: unknown) {
const parsed = Number(value ?? RUN_LOG_DEFAULT_LIMIT_BYTES);
@@ -2575,7 +2572,7 @@ export function agentRoutes(
assertCompanyAccess(req, companyId);
const minCount = readLiveRunsQueryInt(req.query.minCount, 50);
const limit = readLiveRunsQueryInt(req.query.limit, LIVE_RUNS_DEFAULT_LIMIT, LIVE_RUNS_DEFAULT_LIMIT);
const limit = readLiveRunsQueryInt(req.query.limit, 50);
const columns = {
id: heartbeatRuns.id,
@@ -2698,20 +2695,12 @@ export function agentRoutes(
}
assertCompanyAccess(req, existing.companyId);
const decision = typeof req.body?.decision === "string" ? req.body.decision : "";
if (!WATCHDOG_DECISIONS.includes(decision as WatchdogDecision)) {
if (!["snooze", "continue", "dismissed_false_positive"].includes(decision)) {
res.status(400).json({ error: "Unsupported watchdog decision" });
return;
}
const evaluationIssueId = typeof req.body?.evaluationIssueId === "string" ? req.body.evaluationIssueId : null;
const reason = typeof req.body?.reason === "string" ? req.body.reason.slice(0, 4000) : null;
if (decision === "cancel_run" && !reason?.trim()) {
res.status(400).json({ error: "Cancellation reason is required" });
return;
}
if (decision === "cancel_run" && !evaluationIssueId) {
res.status(400).json({ error: "evaluationIssueId is required for cancellation" });
return;
}
const snoozedUntil = decision === "snooze"
? new Date(String(req.body?.snoozedUntil ?? ""))
: null;
@@ -2723,47 +2712,14 @@ export function agentRoutes(
const row = await recovery.recordWatchdogDecision({
runId: existing.id,
actor: req.actor,
decision: decision as WatchdogDecision,
decision: decision as "snooze" | "continue" | "dismissed_false_positive",
evaluationIssueId,
reason,
snoozedUntil,
createdByRunId: req.actor.runId ?? null,
});
if (decision !== "cancel_run") {
res.json(row);
return;
}
const cancelledRun = await heartbeat.cancelRun(existing.id, reason?.trim());
if (cancelledRun) {
await logActivity(db, {
companyId: cancelledRun.companyId,
actorType: req.actor.type === "agent" ? "agent" : "user",
actorId: req.actor.type === "agent"
? req.actor.agentId ?? "agent"
: req.actor.type === "board"
? req.actor.userId ?? "board"
: "unknown",
agentId: req.actor.type === "agent" ? req.actor.agentId ?? null : null,
runId: cancelledRun.id,
action: "heartbeat.watchdog_cancelled",
entityType: "heartbeat_run",
entityId: cancelledRun.id,
details: {
source: "routes.heartbeat_runs.watchdog_decisions",
decision,
evaluationIssueId,
reason: reason?.trim() ?? null,
createdByRunId: req.actor.runId ?? null,
targetRunId: cancelledRun.id,
previousStatus: existing.status,
resultStatus: cancelledRun.status,
},
});
}
res.json({ decision: row, run: cancelledRun });
res.json(row);
});
router.get("/heartbeat-runs/:runId/events", async (req, res) => {
server/src/routes/issues.ts
+16 -74
View File
@@ -1113,21 +1113,6 @@ export function issueRoutes(
documentsSvc.getIssueDocumentByKey(issue.id, ISSUE_CONTINUATION_SUMMARY_DOCUMENT_KEY),
currentExecutionWorkspacePromise,
]);
const executionDisposition = await svc
.listExecutionDispositions(issue.companyId, [
{
id: issue.id,
companyId: issue.companyId,
status: issue.status,
assigneeAgentId: issue.assigneeAgentId,
assigneeUserId: issue.assigneeUserId,
originKind: issue.originKind ?? null,
executionRunId: issue.executionRunId ?? null,
executionState: issue.executionState ?? null,
blockedBy: relations.blockedBy,
},
])
.then((map) => map.get(issue.id) ?? null);
res.json({
issue: {
@@ -1137,7 +1122,6 @@ export function issueRoutes(
description: issue.description,
status: issue.status,
...(blockerAttention ? { blockerAttention } : {}),
...(executionDisposition ? { executionDisposition } : {}),
productivityReview,
priority: issue.priority,
projectId: issue.projectId,
@@ -1236,27 +1220,11 @@ export function issueRoutes(
? await executionWorkspacesSvc.getById(issue.executionWorkspaceId)
: null;
const workProducts = await workProductsSvc.listForIssue(issue.id);
const executionDisposition = await svc
.listExecutionDispositions(issue.companyId, [
{
id: issue.id,
companyId: issue.companyId,
status: issue.status,
assigneeAgentId: issue.assigneeAgentId,
assigneeUserId: issue.assigneeUserId,
originKind: issue.originKind ?? null,
executionRunId: issue.executionRunId ?? null,
executionState: issue.executionState ?? null,
blockedBy: relations.blockedBy,
},
])
.then((map) => map.get(issue.id) ?? null);
res.json({
...issue,
goalId: goal?.id ?? issue.goalId,
ancestors,
...(blockerAttention ? { blockerAttention } : {}),
...(executionDisposition ? { executionDisposition } : {}),
productivityReview,
blockedBy: relations.blockedBy,
blocks: relations.blocks,
@@ -2115,6 +2083,10 @@ export function issueRoutes(
}
}
const runToCancelForCancelledStatus = shouldCancelActiveRunForCancelledStatus
? await resolveActiveIssueRun(existing)
: null;
if (hiddenAtRaw !== undefined) {
updateFields.hiddenAt = hiddenAtRaw ? new Date(hiddenAtRaw) : null;
}
@@ -2297,50 +2269,26 @@ export function issueRoutes(
return;
}
const cancelledStatusRunIds: string[] = [];
const cancelledStatusWakeupIds: string[] = [];
if (shouldCancelActiveRunForCancelledStatus) {
let cancelledStatusRunId: string | null = null;
if (runToCancelForCancelledStatus) {
try {
const cancelled = await heartbeat.cancelIssueRuns({
companyId: issue.companyId,
issueId: issue.id,
reason: "Cancelled because the issue was marked cancelled",
});
for (const cancelledRun of cancelled.runs) {
cancelledStatusRunIds.push(cancelledRun.id);
const cancelled = await heartbeat.cancelRun(runToCancelForCancelledStatus.id);
if (cancelled) {
cancelledStatusRunId = cancelled.id;
await logActivity(db, {
companyId: cancelledRun.companyId,
companyId: cancelled.companyId,
actorType: actor.actorType,
actorId: actor.actorId,
agentId: actor.agentId,
runId: actor.runId,
action: "heartbeat.cancelled",
entityType: "heartbeat_run",
entityId: cancelledRun.id,
details: { agentId: cancelledRun.agentId, source: "issue_status_cancelled", issueId: existing.id },
});
}
for (const cancelledWakeup of cancelled.wakeups) {
cancelledStatusWakeupIds.push(cancelledWakeup.id);
await logActivity(db, {
companyId: issue.companyId,
actorType: actor.actorType,
actorId: actor.actorId,
agentId: actor.agentId,
runId: actor.runId,
action: "issue.wakeup_cancelled",
entityType: "agent_wakeup_request",
entityId: cancelledWakeup.id,
details: {
agentId: cancelledWakeup.agentId,
previousReason: cancelledWakeup.reason,
source: "issue_status_cancelled",
issueId: existing.id,
},
entityId: cancelled.id,
details: { agentId: cancelled.agentId, source: "issue_status_cancelled", issueId: existing.id },
});
}
} catch (err) {
logger.warn({ err, issueId: existing.id }, "failed to cancel runs for cancelled issue");
logger.warn({ err, issueId: existing.id, runId: runToCancelForCancelledStatus.id }, "failed to cancel run for cancelled issue");
await logActivity(db, {
companyId: existing.companyId,
actorType: actor.actorType,
@@ -2348,8 +2296,8 @@ export function issueRoutes(
agentId: actor.agentId,
runId: actor.runId,
action: "heartbeat.cancel_failed",
entityType: "issue",
entityId: existing.id,
entityType: "heartbeat_run",
entityId: runToCancelForCancelledStatus.id,
details: { source: "issue_status_cancelled", issueId: existing.id },
});
}
@@ -2421,13 +2369,7 @@ export function issueRoutes(
...(resumeRequested === true ? { resumeIntent: true, followUpRequested: true } : {}),
...(reopened ? { reopened: true, reopenedFrom: reopenFromStatus } : {}),
...(interruptedRunId ? { interruptedRunId } : {}),
...(cancelledStatusRunIds.length > 0
? {
cancelledStatusRunId: cancelledStatusRunIds[0],
cancelledStatusRunIds,
}
: {}),
...(cancelledStatusWakeupIds.length > 0 ? { cancelledStatusWakeupIds } : {}),
...(cancelledStatusRunId ? { cancelledStatusRunId } : {}),
_previous: hasFieldChanges ? previous : undefined,
...summarizeIssueReferenceActivityDetails(
updateReferenceDiff
server/src/services/heartbeat.ts
+3 -89
View File
@@ -7304,11 +7304,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
return wakeupIds.length;
}
async function cancelRunInternal(
runId: string,
reason = "Cancelled by control plane",
opts?: { startNextQueuedRun?: boolean },
) {
async function cancelRunInternal(runId: string, reason = "Cancelled by control plane") {
const run = await getRun(runId);
if (!run) throw notFound("Heartbeat run not found");
if (!CANCELLABLE_HEARTBEAT_RUN_STATUSES.includes(run.status as (typeof CANCELLABLE_HEARTBEAT_RUN_STATUSES)[number])) return run;
@@ -7358,90 +7354,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
runningProcesses.delete(run.id);
await finalizeAgentStatus(run.agentId, "cancelled");
if (opts?.startNextQueuedRun !== false) {
await startNextQueuedRunForAgent(run.agentId);
}
await startNextQueuedRunForAgent(run.agentId);
return cancelled;
}
async function cancelIssueRunsInternal(input: {
companyId: string;
issueId: string;
reason?: string;
}) {
const reason = input.reason ?? "Cancelled because the issue was cancelled";
const now = new Date();
const cancelledWakeups = await db
.update(agentWakeupRequests)
.set({
status: "cancelled",
finishedAt: now,
error: reason,
updatedAt: now,
})
.where(
and(
eq(agentWakeupRequests.companyId, input.companyId),
inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]),
isNull(agentWakeupRequests.runId),
sql`${agentWakeupRequests.payload} ->> 'issueId' = ${input.issueId}`,
),
)
.returning({
id: agentWakeupRequests.id,
agentId: agentWakeupRequests.agentId,
reason: agentWakeupRequests.reason,
status: agentWakeupRequests.status,
payload: agentWakeupRequests.payload,
});
const issue = await db
.select({
executionRunId: issues.executionRunId,
})
.from(issues)
.where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId)))
.then((rows) => rows[0] ?? null);
const contextIssueId = sql<string | null>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`;
const runs = await db
.select()
.from(heartbeatRuns)
.where(
and(
eq(heartbeatRuns.companyId, input.companyId),
inArray(heartbeatRuns.status, [...CANCELLABLE_HEARTBEAT_RUN_STATUSES]),
issue?.executionRunId
? or(eq(heartbeatRuns.id, issue.executionRunId), sql`${contextIssueId} = ${input.issueId}`)
: sql`${contextIssueId} = ${input.issueId}`,
),
)
.orderBy(
sql`case when ${heartbeatRuns.status} = 'running' then 0 when ${heartbeatRuns.status} = 'queued' then 1 else 2 end`,
asc(heartbeatRuns.createdAt),
);
const cancelledRuns: Array<typeof heartbeatRuns.$inferSelect> = [];
const affectedAgentIds = new Set<string>();
for (const run of runs) {
const cancelled = await cancelRunInternal(run.id, reason, { startNextQueuedRun: false });
affectedAgentIds.add(run.agentId);
if (cancelled?.status === "cancelled") {
cancelledRuns.push(cancelled);
}
}
for (const agentId of affectedAgentIds) {
await startNextQueuedRunForAgent(agentId);
}
return {
runs: cancelledRuns,
wakeups: cancelledWakeups,
};
}
async function cancelActiveForAgentInternal(agentId: string, reason = "Cancelled due to agent pause") {
const agent = await getAgent(agentId);
const runs = await db
@@ -7816,9 +7732,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
return { checked, enqueued, skipped };
},
cancelRun: (runId: string, reason?: string) => cancelRunInternal(runId, reason),
cancelIssueRuns: (input: { companyId: string; issueId: string; reason?: string }) => cancelIssueRunsInternal(input),
cancelRun: (runId: string) => cancelRunInternal(runId),
cancelActiveForAgent: (agentId: string) => cancelActiveForAgentInternal(agentId),
server/src/services/issue-execution-disposition-resolver.ts
-521
View File
@@ -1,521 +0,0 @@
import { and, desc, eq, inArray, isNull, ne, notInArray, sql } from "drizzle-orm";
import {
agentWakeupRequests,
agents,
approvals,
heartbeatRuns,
issueApprovals,
issueRelations,
issueThreadInteractions,
issues,
} from "@paperclipai/db";
import type {
IssueExecutionDisposition,
IssueOriginKind,
IssueRelationIssueSummary,
IssueStatus,
} from "@paperclipai/shared";
import {
isLiveExplicitApprovalWaitingPath,
isLiveExplicitInteractionWaitingPath,
} from "./recovery/explicit-waiting-paths.js";
import { parseIssueGraphLivenessIncidentKey } from "./recovery/origins.js";
import {
classifyIssueExecutionDisposition,
type IssueExecutionParticipantState,
type IssueExecutionRunLivenessState,
type IssueExecutionRunStatus,
} from "./issue-execution-disposition.js";
const ACTIVE_RUN_STATUSES = ["queued", "running"] as const;
const ACTIVE_WAKE_STATUSES = ["queued", "deferred_issue_execution"] as const;
const PENDING_INTERACTION_STATUSES = ["pending"] as const;
const WAITING_INTERACTION_KINDS = ["request_confirmation", "ask_user_questions", "suggest_tasks"] as const;
const PENDING_APPROVAL_STATUSES = ["pending", "revision_requested"] as const;
const STRANDED_RECOVERY_ORIGIN_KIND = "stranded_issue_recovery" as const;
const HARNESS_LIVENESS_ESCALATION_ORIGIN_KIND = "harness_liveness_escalation" as const;
const PRODUCTIVITY_REVIEW_ORIGIN_KIND = "issue_productivity_review" as const;
const NON_TERMINAL_RECOVERY_STATUSES = ["done", "cancelled"] as const;
const INVOKABLE_AGENT_STATUSES: ReadonlySet<string> = new Set(["active", "idle", "running", "error"]);
const DEFAULT_MAX_CONTINUATION_ATTEMPTS = 2;
const QUERY_CHUNK_SIZE = 500;
export interface IssueExecutionDispositionInputNode {
id: string;
companyId: string;
status: IssueStatus | string;
assigneeAgentId: string | null;
assigneeUserId: string | null;
originKind: IssueOriginKind | string | null;
executionRunId: string | null;
executionState?: Record<string, unknown> | null;
blockedBy?: IssueRelationIssueSummary[];
}
interface ExplicitInteractionRow {
companyId: string;
issueId: string;
status: string;
createdByUserId: string | null;
createdAt: Date;
}
interface ExplicitApprovalRow {
companyId: string;
issueId: string;
status: string;
requestedByUserId: string | null;
linkedByUserId: string | null;
createdAt: Date;
updatedAt: Date;
linkedAt: Date;
}
interface LatestRunRow {
issueId: string;
status: string;
livenessState: string | null;
continuationAttempt: number | null;
contextSnapshot: unknown;
}
function chunk<T>(items: T[], size: number): T[][] {
if (items.length === 0) return [];
const out: T[][] = [];
for (let i = 0; i < items.length; i += size) {
out.push(items.slice(i, i + size));
}
return out;
}
function readPrincipalAgentId(principal: unknown): string | null {
if (!principal || typeof principal !== "object") return null;
const value = principal as Record<string, unknown>;
return value.type === "agent" && typeof value.agentId === "string" && value.agentId.length > 0
? value.agentId
: null;
}
function principalIsResolvableUser(principal: unknown): boolean {
if (!principal || typeof principal !== "object") return false;
const value = principal as Record<string, unknown>;
return value.type === "user" && typeof value.userId === "string" && value.userId.length > 0;
}
function participantStateForExecutionState(
executionState: Record<string, unknown> | null | undefined,
agentStatusById: ReadonlyMap<string, string>,
): IssueExecutionParticipantState {
if (!executionState) return "none";
const status = (executionState.status as string | undefined) ?? null;
if (!status || status === "satisfied") return "none";
const principal = executionState.currentParticipant ?? null;
if (!principal) return "none";
const agentId = readPrincipalAgentId(principal);
if (agentId) {
const agentStatus = agentStatusById.get(agentId);
if (agentStatus && INVOKABLE_AGENT_STATUSES.has(agentStatus)) return "valid";
return "invalid";
}
if (principalIsResolvableUser(principal)) return "valid";
return "invalid";
}
function readNextActionFlag(contextSnapshot: unknown): boolean {
if (!contextSnapshot || typeof contextSnapshot !== "object") return false;
const next = (contextSnapshot as Record<string, unknown>).nextAction;
return typeof next === "string" && next.trim().length > 0;
}
export async function listIssueExecutionDispositionsMap(
dbOrTx: any,
companyId: string,
issueRows: IssueExecutionDispositionInputNode[],
): Promise<Map<string, IssueExecutionDisposition>> {
const result = new Map<string, IssueExecutionDisposition>();
const sameCompanyRows = issueRows.filter((row) => row.companyId === companyId);
if (sameCompanyRows.length === 0) return result;
const issueIds = sameCompanyRows.map((row) => row.id);
// The list select drops executionState for performance; rehydrate it from the DB
// for the issues that actually need participant evidence (in_review).
const reviewIdsMissingState = sameCompanyRows
.filter((row) => row.status === "in_review" && row.executionState === undefined)
.map((row) => row.id);
const executionStateById = new Map<string, Record<string, unknown> | null>();
for (const part of chunk(reviewIdsMissingState, QUERY_CHUNK_SIZE)) {
const rows: Array<{ id: string; executionState: Record<string, unknown> | null }> = await dbOrTx
.select({ id: issues.id, executionState: issues.executionState })
.from(issues)
.where(and(eq(issues.companyId, companyId), inArray(issues.id, part)));
for (const r of rows) executionStateById.set(r.id, r.executionState ?? null);
}
const resolvedExecutionStateFor = (row: IssueExecutionDispositionInputNode) =>
row.executionState !== undefined ? row.executionState : executionStateById.get(row.id) ?? null;
const agentIds = new Set<string>();
for (const row of sameCompanyRows) {
if (row.assigneeAgentId) agentIds.add(row.assigneeAgentId);
const state = resolvedExecutionStateFor(row);
const participant = state?.currentParticipant ?? null;
const participantAgentId = readPrincipalAgentId(participant);
if (participantAgentId) agentIds.add(participantAgentId);
}
const activeRunIssueIds = new Set<string>();
const queuedWakeIssueIds = new Set<string>();
const explicitInteractionLive = new Set<string>();
const explicitApprovalLive = new Set<string>();
const openRecoveryIssueOrigins = new Set<string>();
const openProductivityReviewOrigins = new Set<string>();
const latestRunByIssueId = new Map<string, LatestRunRow>();
// Active runs: any heartbeatRun in queued/running referencing the issue via contextSnapshot.
for (const part of chunk(issueIds, QUERY_CHUNK_SIZE)) {
const rows: Array<{ issueId: string | null }> = await dbOrTx
.select({
issueId: sql<string | null>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`,
})
.from(heartbeatRuns)
.where(
and(
eq(heartbeatRuns.companyId, companyId),
inArray(heartbeatRuns.status, [...ACTIVE_RUN_STATUSES]),
inArray(sql<string>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`, part),
),
);
for (const row of rows) {
if (row.issueId) activeRunIssueIds.add(row.issueId);
}
}
// Active runs by executionRunId — covers cases where the run claimed the issue but the
// contextSnapshot lookup misses (legacy snapshots).
const executionRunIds = sameCompanyRows
.map((row) => row.executionRunId)
.filter((value): value is string => Boolean(value));
for (const part of chunk(executionRunIds, QUERY_CHUNK_SIZE)) {
const rows: Array<{ id: string }> = await dbOrTx
.select({ id: heartbeatRuns.id })
.from(heartbeatRuns)
.where(
and(
eq(heartbeatRuns.companyId, companyId),
inArray(heartbeatRuns.status, [...ACTIVE_RUN_STATUSES]),
inArray(heartbeatRuns.id, part),
),
);
if (rows.length === 0) continue;
const liveExecutionRunIds = new Set(rows.map((row) => row.id));
for (const row of sameCompanyRows) {
if (row.executionRunId && liveExecutionRunIds.has(row.executionRunId)) {
activeRunIssueIds.add(row.id);
}
}
}
// Queued wakes: agent_wakeup_requests with issueId payload, no runId yet.
for (const part of chunk(issueIds, QUERY_CHUNK_SIZE)) {
const rows: Array<{ issueId: string | null }> = await dbOrTx
.select({
issueId: sql<string | null>`${agentWakeupRequests.payload} ->> 'issueId'`,
})
.from(agentWakeupRequests)
.where(
and(
eq(agentWakeupRequests.companyId, companyId),
inArray(agentWakeupRequests.status, [...ACTIVE_WAKE_STATUSES]),
sql`${agentWakeupRequests.runId} is null`,
inArray(sql<string>`${agentWakeupRequests.payload} ->> 'issueId'`, part),
),
);
for (const row of rows) {
if (row.issueId) queuedWakeIssueIds.add(row.issueId);
}
}
// Pending interactions.
const interactionsByIssueId = new Map<string, ExplicitInteractionRow[]>();
for (const part of chunk(issueIds, QUERY_CHUNK_SIZE)) {
const rows: ExplicitInteractionRow[] = await dbOrTx
.select({
companyId: issueThreadInteractions.companyId,
issueId: issueThreadInteractions.issueId,
status: issueThreadInteractions.status,
createdByUserId: issueThreadInteractions.createdByUserId,
createdAt: issueThreadInteractions.createdAt,
})
.from(issueThreadInteractions)
.where(
and(
eq(issueThreadInteractions.companyId, companyId),
inArray(issueThreadInteractions.status, [...PENDING_INTERACTION_STATUSES]),
inArray(issueThreadInteractions.kind, [...WAITING_INTERACTION_KINDS]),
inArray(issueThreadInteractions.issueId, part),
),
);
for (const row of rows) {
const list = interactionsByIssueId.get(row.issueId) ?? [];
list.push(row);
interactionsByIssueId.set(row.issueId, list);
}
}
// Pending approvals.
const approvalsByIssueId = new Map<string, ExplicitApprovalRow[]>();
for (const part of chunk(issueIds, QUERY_CHUNK_SIZE)) {
const rows: ExplicitApprovalRow[] = await dbOrTx
.select({
companyId: issueApprovals.companyId,
issueId: issueApprovals.issueId,
status: approvals.status,
requestedByUserId: approvals.requestedByUserId,
linkedByUserId: issueApprovals.linkedByUserId,
createdAt: approvals.createdAt,
updatedAt: approvals.updatedAt,
linkedAt: issueApprovals.createdAt,
})
.from(issueApprovals)
.innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id))
.where(
and(
eq(issueApprovals.companyId, companyId),
inArray(approvals.status, [...PENDING_APPROVAL_STATUSES]),
inArray(issueApprovals.issueId, part),
),
);
for (const row of rows) {
const list = approvalsByIssueId.get(row.issueId) ?? [];
list.push(row);
approvalsByIssueId.set(row.issueId, list);
}
}
const now = new Date();
for (const row of sameCompanyRows) {
const issueShape = {
companyId: row.companyId,
id: row.id,
assigneeUserId: row.assigneeUserId,
executionState: resolvedExecutionStateFor(row),
};
const interactions = interactionsByIssueId.get(row.id) ?? [];
if (interactions.some((entry) => isLiveExplicitInteractionWaitingPath(issueShape, entry, now))) {
explicitInteractionLive.add(row.id);
}
const approvalsForIssue = approvalsByIssueId.get(row.id) ?? [];
if (approvalsForIssue.some((entry) => isLiveExplicitApprovalWaitingPath(issueShape, entry, now))) {
explicitApprovalLive.add(row.id);
}
}
// Open recovery / harness escalation / productivity review issues whose origin points at our inputs.
for (const part of chunk(issueIds, QUERY_CHUNK_SIZE)) {
const rows: Array<{ originId: string | null }> = await dbOrTx
.select({ originId: issues.originId })
.from(issues)
.where(
and(
eq(issues.companyId, companyId),
eq(issues.originKind, STRANDED_RECOVERY_ORIGIN_KIND),
isNull(issues.hiddenAt),
inArray(issues.originId, part),
notInArray(issues.status, [...NON_TERMINAL_RECOVERY_STATUSES]),
),
);
for (const r of rows) {
if (r.originId) openRecoveryIssueOrigins.add(r.originId);
}
}
const livenessRecoveryRows: Array<{ originId: string | null }> = await dbOrTx
.select({ originId: issues.originId })
.from(issues)
.where(
and(
eq(issues.companyId, companyId),
eq(issues.originKind, HARNESS_LIVENESS_ESCALATION_ORIGIN_KIND),
isNull(issues.hiddenAt),
notInArray(issues.status, [...NON_TERMINAL_RECOVERY_STATUSES]),
),
);
for (const r of livenessRecoveryRows) {
const parsed = parseIssueGraphLivenessIncidentKey(r.originId);
if (!parsed || parsed.companyId !== companyId) continue;
openRecoveryIssueOrigins.add(parsed.issueId);
openRecoveryIssueOrigins.add(parsed.leafIssueId);
}
for (const part of chunk(issueIds, QUERY_CHUNK_SIZE)) {
const rows: Array<{ originId: string | null }> = await dbOrTx
.select({ originId: issues.originId })
.from(issues)
.where(
and(
eq(issues.companyId, companyId),
eq(issues.originKind, PRODUCTIVITY_REVIEW_ORIGIN_KIND),
isNull(issues.hiddenAt),
inArray(issues.originId, part),
notInArray(issues.status, [...NON_TERMINAL_RECOVERY_STATUSES]),
),
);
for (const r of rows) {
if (r.originId) openProductivityReviewOrigins.add(r.originId);
}
}
// Latest issue-linked run, used to populate run evidence.
const recoveryCandidateIds = sameCompanyRows
.filter((row) =>
row.assigneeAgentId &&
!activeRunIssueIds.has(row.id) &&
!queuedWakeIssueIds.has(row.id) &&
(row.status === "todo" || row.status === "in_progress"),
)
.map((row) => row.id);
for (const part of chunk(recoveryCandidateIds, QUERY_CHUNK_SIZE)) {
const rows: LatestRunRow[] = await dbOrTx
.select({
issueId: sql<string>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`,
status: heartbeatRuns.status,
livenessState: heartbeatRuns.livenessState,
continuationAttempt: heartbeatRuns.continuationAttempt,
contextSnapshot: heartbeatRuns.contextSnapshot,
})
.from(heartbeatRuns)
.where(
and(
eq(heartbeatRuns.companyId, companyId),
inArray(sql<string>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`, part),
),
)
.orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id));
for (const row of rows) {
if (!row.issueId) continue;
if (latestRunByIssueId.has(row.issueId)) continue;
latestRunByIssueId.set(row.issueId, row);
}
}
// Agents.
const agentStatusById = new Map<string, string>();
if (agentIds.size > 0) {
const rows: Array<{ id: string; status: string }> = await dbOrTx
.select({ id: agents.id, status: agents.status })
.from(agents)
.where(and(eq(agents.companyId, companyId), inArray(agents.id, [...agentIds])));
for (const r of rows) agentStatusById.set(r.id, r.status);
}
for (const row of sameCompanyRows) {
const agentStatus = row.assigneeAgentId ? agentStatusById.get(row.assigneeAgentId) ?? null : null;
const latestRun = latestRunByIssueId.get(row.id) ?? null;
const productivityReviewNeeded = openProductivityReviewOrigins.has(row.id);
const openRecoveryIssue = openRecoveryIssueOrigins.has(row.id);
const blockers = (row.blockedBy ?? []).map((blocker) => ({
issue: {
id: blocker.id,
status: blocker.status as IssueStatus,
assigneeAgentId: blocker.assigneeAgentId ?? null,
assigneeUserId: blocker.assigneeUserId ?? null,
},
}));
const disposition = classifyIssueExecutionDisposition({
issue: {
id: row.id,
status: row.status as IssueStatus,
assigneeAgentId: row.assigneeAgentId,
assigneeUserId: row.assigneeUserId,
originKind: row.originKind ?? null,
},
agent: row.assigneeAgentId
? { id: row.assigneeAgentId, status: agentStatus ?? "terminated" }
: null,
execution: {
activeRun: activeRunIssueIds.has(row.id),
queuedWake: queuedWakeIssueIds.has(row.id),
},
waits: {
participant: participantStateForExecutionState(resolvedExecutionStateFor(row), agentStatusById),
pendingInteraction: explicitInteractionLive.has(row.id) ? "live" : "none",
pendingApproval: explicitApprovalLive.has(row.id) ? "live" : "none",
openRecoveryIssue,
openProductivityReviewIssue: productivityReviewNeeded,
productivityReviewNeeded,
},
latestRun: latestRun
? {
latestRunStatus: latestRun.status as IssueExecutionRunStatus,
livenessState: (latestRun.livenessState as IssueExecutionRunLivenessState | null) ?? null,
nextAction: readNextActionFlag(latestRun.contextSnapshot) ? "runnable" : "none",
continuationAttempt: latestRun.continuationAttempt,
maxContinuationAttempts: DEFAULT_MAX_CONTINUATION_ATTEMPTS,
}
: null,
blockers,
});
result.set(row.id, disposition);
}
return result;
}
// Helper: load blockedBy relation summaries for issues that don't already have them on the input.
export async function backfillBlockedBySummaries(
dbOrTx: any,
companyId: string,
rows: IssueExecutionDispositionInputNode[],
): Promise<void> {
const needsBackfill = rows.filter((row) => row.status === "blocked" && !row.blockedBy);
if (needsBackfill.length === 0) return;
const ids = needsBackfill.map((row) => row.id);
const blockerRows: Array<{
issueId: string;
blockerIssueId: string;
identifier: string | null;
title: string;
status: string;
priority: string | null;
assigneeAgentId: string | null;
assigneeUserId: string | null;
}> = await dbOrTx
.select({
issueId: issueRelations.relatedIssueId,
blockerIssueId: issues.id,
identifier: issues.identifier,
title: issues.title,
status: issues.status,
priority: issues.priority,
assigneeAgentId: issues.assigneeAgentId,
assigneeUserId: issues.assigneeUserId,
})
.from(issueRelations)
.innerJoin(issues, eq(issueRelations.issueId, issues.id))
.where(
and(
eq(issueRelations.companyId, companyId),
eq(issueRelations.type, "blocks"),
inArray(issueRelations.relatedIssueId, ids),
eq(issues.companyId, companyId),
ne(issues.status, "done"),
),
);
const blockersByIssueId = new Map<string, IssueRelationIssueSummary[]>();
for (const row of blockerRows) {
const list = blockersByIssueId.get(row.issueId) ?? [];
list.push({
id: row.blockerIssueId,
identifier: row.identifier,
title: row.title,
status: row.status as IssueStatus,
priority: (row.priority ?? "medium") as IssueRelationIssueSummary["priority"],
assigneeAgentId: row.assigneeAgentId,
assigneeUserId: row.assigneeUserId,
});
blockersByIssueId.set(row.issueId, list);
}
for (const row of needsBackfill) {
row.blockedBy = blockersByIssueId.get(row.id) ?? [];
}
}
server/src/services/issue-execution-disposition.ts
-337
View File
@@ -1,337 +0,0 @@
import type {
AgentStatus,
IssueOriginKind,
IssueStatus,
RunLivenessState,
} from "@paperclipai/shared";
import type {
IssueExecutionDisposition,
IssueExecutionInvalidReason,
IssueExecutionLivePath,
IssueExecutionWaitingPath,
} from "@paperclipai/shared";
export type IssueExecutionWaitSignalState = "none" | "live" | "stale";
export type IssueExecutionParticipantState = "none" | "valid" | "invalid";
export type IssueExecutionRunStatus =
| "queued"
| "running"
| "succeeded"
| "failed"
| "timed_out"
| "cancelled";
export type IssueExecutionRunLivenessState = RunLivenessState | "unknown";
export type IssueExecutionNextActionState =
| "runnable"
| "manager_review"
| "approval_required"
| "unknown"
| "none";
export interface IssueExecutionDispositionIssueInput {
id: string;
status: IssueStatus;
assigneeAgentId?: string | null;
assigneeUserId?: string | null;
originKind?: IssueOriginKind | string | null;
}
export interface IssueExecutionDispositionAgentInput {
id: string;
status: AgentStatus | string;
}
export interface IssueExecutionPathVector {
activeRun?: boolean;
queuedWake?: boolean;
scheduledRetry?: boolean;
deferredExecution?: boolean;
}
export interface IssueExecutionWaitVector {
participant?: IssueExecutionParticipantState;
pendingInteraction?: IssueExecutionWaitSignalState;
pendingApproval?: IssueExecutionWaitSignalState;
pauseHold?: boolean;
openRecoveryIssue?: boolean;
openWatchdogIssue?: boolean;
openProductivityReviewIssue?: boolean;
productivityReviewNeeded?: boolean;
externalOwnerAction?: boolean;
}
export interface IssueExecutionRunEvidence {
latestRunStatus?: IssueExecutionRunStatus | null;
livenessState?: IssueExecutionRunLivenessState | null;
nextAction?: IssueExecutionNextActionState | null;
continuationAttempt?: number | null;
maxContinuationAttempts?: number | null;
recoveryAttemptRemaining?: boolean;
}
export interface IssueExecutionBlockerVector {
issue: Pick<IssueExecutionDispositionIssueInput, "id" | "status" | "assigneeAgentId" | "assigneeUserId">;
disposition?: IssueExecutionDisposition;
stateVector?: IssueExecutionStateVector;
}
export interface IssueExecutionGateVector {
budgetBlocked?: boolean;
}
export interface IssueExecutionStateVector {
issue: IssueExecutionDispositionIssueInput;
agent?: IssueExecutionDispositionAgentInput | null;
execution?: IssueExecutionPathVector;
waits?: IssueExecutionWaitVector;
latestRun?: IssueExecutionRunEvidence | null;
blockers?: IssueExecutionBlockerVector[];
gates?: IssueExecutionGateVector;
}
const INVOKABLE_AGENT_STATUSES = new Set(["active", "idle", "running", "error"]);
const RECOVERY_ORIGIN_KINDS = new Set(["stranded_issue_recovery", "harness_liveness_escalation"]);
function invalid(
reason: IssueExecutionInvalidReason,
suggestedCorrection: string,
): IssueExecutionDisposition {
return { kind: "invalid", reason, suggestedCorrection };
}
function hasAgentOwner(issue: IssueExecutionDispositionIssueInput) {
return Boolean(issue.assigneeAgentId);
}
function hasHumanOwner(issue: IssueExecutionDispositionIssueInput) {
return Boolean(issue.assigneeUserId);
}
function agentIsInvokable(
issue: IssueExecutionDispositionIssueInput,
agent: IssueExecutionDispositionAgentInput | null | undefined,
budgetBlocked: boolean,
) {
if (!issue.assigneeAgentId || !agent || agent.id !== issue.assigneeAgentId) return false;
if (budgetBlocked) return false;
return INVOKABLE_AGENT_STATUSES.has(agent.status);
}
function livePath(execution: IssueExecutionPathVector | undefined): IssueExecutionLivePath | null {
if (execution?.activeRun) return "active_run";
if (execution?.queuedWake) return "queued_wake";
if (execution?.scheduledRetry) return "scheduled_retry";
if (execution?.deferredExecution) return "deferred_execution";
return null;
}
function waitingPath(
issue: IssueExecutionDispositionIssueInput,
waits: IssueExecutionWaitVector | undefined,
): IssueExecutionWaitingPath | null {
if (waits?.pauseHold) return "pause_hold";
if (hasHumanOwner(issue)) return "human_owner";
if (waits?.participant === "valid") return "participant";
if (waits?.pendingInteraction === "live") return "interaction";
if (waits?.pendingApproval === "live") return "approval";
if (waits?.openRecoveryIssue || waits?.openWatchdogIssue || waits?.openProductivityReviewIssue) {
return "review_artifact";
}
if (waits?.externalOwnerAction) return "external_owner_action";
return null;
}
function runFailed(run: IssueExecutionRunEvidence | null | undefined) {
return (
run?.latestRunStatus === "failed" ||
run?.latestRunStatus === "timed_out" ||
run?.latestRunStatus === "cancelled"
);
}
function hasRunnableContinuation(run: IssueExecutionRunEvidence | null | undefined) {
const attempt = run?.continuationAttempt ?? 0;
const maxAttempts = run?.maxContinuationAttempts ?? 2;
const livenessState = run?.livenessState ?? null;
const actionableLowProgressState = livenessState === "plan_only" || livenessState === "empty_response";
const runnableFollowupState =
(livenessState === "advanced" || livenessState === "needs_followup") &&
run?.nextAction === "runnable";
return (
run?.latestRunStatus === "succeeded" &&
(actionableLowProgressState || runnableFollowupState) &&
attempt < maxAttempts
);
}
function hasExhaustedOrAmbiguousContinuation(run: IssueExecutionRunEvidence | null | undefined) {
if (run?.latestRunStatus !== "succeeded") return false;
if (run.livenessState === "completed") return true;
if (run.livenessState === "blocked" || run.livenessState === "failed") return true;
if (run.livenessState === "plan_only" || run.livenessState === "empty_response") {
return (run.continuationAttempt ?? 0) >= (run.maxContinuationAttempts ?? 2);
}
if (run.livenessState !== "advanced" && run.livenessState !== "needs_followup") return true;
if (run.nextAction !== "runnable") return true;
return (run.continuationAttempt ?? 0) >= (run.maxContinuationAttempts ?? 2);
}
function isRecoveryIssue(issue: IssueExecutionDispositionIssueInput) {
return Boolean(issue.originKind && RECOVERY_ORIGIN_KINDS.has(issue.originKind));
}
function classifyBlocker(
blocker: IssueExecutionBlockerVector,
seen: Set<string>,
): IssueExecutionDisposition {
if (blocker.disposition) return blocker.disposition;
if (blocker.stateVector) return classifyIssueExecutionDisposition(blocker.stateVector, seen);
return classifyIssueExecutionDisposition({ issue: blocker.issue as IssueExecutionDispositionIssueInput }, seen);
}
function blockerInvalidReason(blocker: IssueExecutionBlockerVector): IssueExecutionInvalidReason {
if (blocker.issue.status === "cancelled") return "blocked_by_cancelled_issue";
if (!blocker.issue.assigneeAgentId && !blocker.issue.assigneeUserId) return "blocked_by_unassigned_issue";
return "blocked_by_resting_issue";
}
function classifyBlockedIssue(
input: IssueExecutionStateVector,
seen: Set<string>,
): IssueExecutionDisposition {
if (isRecoveryIssue(input.issue) && (hasAgentOwner(input.issue) || hasHumanOwner(input.issue))) {
return { kind: "waiting", path: "external_owner_action" };
}
const path = waitingPath(input.issue, input.waits);
if (path) return { kind: "waiting", path };
const blockers = input.blockers ?? [];
if (blockers.length > 0) {
for (const blocker of blockers) {
if (blocker.issue.status === "cancelled") {
return invalid(
"blocked_by_cancelled_issue",
"Remove or replace the cancelled blocker before treating the dependency chain as covered.",
);
}
const disposition = classifyBlocker(blocker, seen);
if (disposition.kind === "invalid") {
return invalid(
"blocked_by_invalid_issue",
"Repair the invalid unresolved blocker or create an explicit recovery/escalation path.",
);
}
if (disposition.kind === "resting" || disposition.kind === "terminal") {
return invalid(
blockerInvalidReason(blocker),
"Give the blocker a lawful next action, remove the blocker, or escalate to a human owner.",
);
}
}
return { kind: "waiting", path: "blocker_chain" };
}
return invalid(
"blocked_without_action_path",
"Blocked issues require first-class blockers, a live wait primitive, or a structured external owner/action.",
);
}
export function classifyIssueExecutionDisposition(
input: IssueExecutionStateVector,
seen = new Set<string>(),
): IssueExecutionDisposition {
const { issue } = input;
const status = issue.status;
const budgetBlocked = Boolean(input.gates?.budgetBlocked);
const agentInvokable = agentIsInvokable(issue, input.agent, budgetBlocked);
const path = livePath(input.execution);
const wait = waitingPath(issue, input.waits);
if (status === "done" || status === "cancelled") return { kind: "terminal" };
if (seen.has(issue.id)) {
return invalid(
"blocked_by_invalid_issue",
"Break the recursive blocker chain or create an explicit recovery issue.",
);
}
seen.add(issue.id);
if (hasAgentOwner(issue) && hasHumanOwner(issue)) {
return invalid("dual_assignee", "Keep exactly one issue assignee: agent or human user, not both.");
}
if (status === "backlog") return { kind: "resting" };
if (wait) return { kind: "waiting", path: wait };
if (input.waits?.participant === "invalid") {
return invalid(
"invalid_review_participant",
"Set a typed execution participant with a valid agentId/userId or replace it with another explicit wait path.",
);
}
if (status === "blocked") return classifyBlockedIssue(input, seen);
if (hasHumanOwner(issue)) return { kind: "waiting", path: "human_owner" };
if (path) return { kind: "live", path };
if (!hasAgentOwner(issue)) return { kind: "resting" };
if (status === "todo") {
if (isRecoveryIssue(issue) && runFailed(input.latestRun)) {
return { kind: "human_escalation_required", owner: "recovery_owner" };
}
if (!agentInvokable) {
return { kind: "human_escalation_required", owner: budgetBlocked ? "board" : "manager" };
}
if (runFailed(input.latestRun) && input.latestRun?.recoveryAttemptRemaining !== false) {
return { kind: "recoverable_by_control_plane", recovery: "dispatch" };
}
if (runFailed(input.latestRun)) {
return { kind: "human_escalation_required", owner: "manager" };
}
if (input.waits?.productivityReviewNeeded) {
return { kind: "human_escalation_required", owner: "manager" };
}
return { kind: "dispatchable", wakeTarget: issue.assigneeAgentId! };
}
if (status === "in_progress") {
if (!agentInvokable) {
return { kind: "human_escalation_required", owner: budgetBlocked ? "board" : "manager" };
}
if (isRecoveryIssue(issue) && runFailed(input.latestRun)) {
return { kind: "human_escalation_required", owner: "recovery_owner" };
}
if (runFailed(input.latestRun) && input.latestRun?.recoveryAttemptRemaining !== false) {
return { kind: "recoverable_by_control_plane", recovery: "continuation" };
}
if (runFailed(input.latestRun)) {
return { kind: "human_escalation_required", owner: "manager" };
}
if (input.waits?.productivityReviewNeeded) {
return { kind: "human_escalation_required", owner: "manager" };
}
if (hasRunnableContinuation(input.latestRun)) {
return {
kind: "agent_continuable",
continuationAttempt: input.latestRun?.continuationAttempt ?? 0,
maxAttempts: input.latestRun?.maxContinuationAttempts ?? 2,
};
}
if (hasExhaustedOrAmbiguousContinuation(input.latestRun)) {
return { kind: "human_escalation_required", owner: "manager" };
}
return { kind: "dispatchable", wakeTarget: issue.assigneeAgentId! };
}
if (status === "in_review") {
return invalid(
"in_review_without_action_path",
"Agent-owned review requires a typed participant, pending interaction, approval, active/queued execution path, human owner, or explicit recovery issue.",
);
}
return { kind: "human_escalation_required", owner: "manager" };
}
server/src/services/issue-transition-guard.ts
-277
View File
@@ -1,277 +0,0 @@
import { and, desc, eq, inArray, isNull, notInArray, sql } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import {
agentWakeupRequests,
agents,
approvals,
heartbeatRuns,
issueApprovals,
issueRelations,
issueThreadInteractions,
issues,
} from "@paperclipai/db";
import type { IssueExecutionDisposition } from "@paperclipai/shared";
import { unprocessable } from "../errors.js";
import {
classifyIssueExecutionDisposition,
type IssueExecutionParticipantState,
type IssueExecutionStateVector,
} from "./issue-execution-disposition.js";
import { parseIssueExecutionState } from "./issue-execution-policy.js";
import {
isLiveExplicitApprovalWaitingPath,
isLiveExplicitInteractionWaitingPath,
} from "./recovery/explicit-waiting-paths.js";
const ACTIVE_RUN_STATUSES = ["queued", "running"] as const;
const SCHEDULED_RETRY_RUN_STATUSES = ["scheduled_retry"] as const;
const ACTIVE_WAKE_STATUSES = ["queued", "deferred_issue_execution"] as const;
const PENDING_INTERACTION_STATUSES = ["pending"] as const;
const WAITING_INTERACTION_KINDS = ["request_confirmation", "ask_user_questions"] as const;
const PENDING_APPROVAL_STATUSES = ["pending", "revision_requested"] as const;
const OPEN_RECOVERY_ORIGIN_KINDS = ["harness_liveness_escalation", "stranded_issue_recovery"] as const;
const OPEN_RECOVERY_TERMINAL_STATUSES = ["done", "cancelled"] as const;
type DbReader = Pick<Db, "select">;
type IssueRow = typeof issues.$inferSelect;
export type IssueTransitionGuardSource = "create" | "update";
function isValidParticipant(participant: unknown) {
if (!participant || typeof participant !== "object") return false;
const value = participant as Record<string, unknown>;
if (value.type === "agent") return typeof value.agentId === "string" && value.agentId.length > 0;
if (value.type === "user") return typeof value.userId === "string" && value.userId.length > 0;
return false;
}
function isGuardedStatus(status: string | null | undefined) {
return status === "in_progress" || status === "in_review" || status === "blocked";
}
function invalidMessage(disposition: Extract<IssueExecutionDisposition, { kind: "invalid" }>) {
return `Invalid issue transition: ${disposition.reason}`;
}
async function buildIssueExecutionStateVector(
dbOrTx: DbReader,
issue: IssueRow,
): Promise<IssueExecutionStateVector> {
const [agent] = issue.assigneeAgentId
? await dbOrTx
.select({
id: agents.id,
status: agents.status,
})
.from(agents)
.where(and(eq(agents.companyId, issue.companyId), eq(agents.id, issue.assigneeAgentId)))
: [];
const [executionRun] = issue.executionRunId
? await dbOrTx
.select({ status: heartbeatRuns.status })
.from(heartbeatRuns)
.where(and(eq(heartbeatRuns.companyId, issue.companyId), eq(heartbeatRuns.id, issue.executionRunId)))
: [];
const [queuedWake] = await dbOrTx
.select({
issueId: sql<string | null>`${agentWakeupRequests.payload} ->> 'issueId'`,
status: agentWakeupRequests.status,
})
.from(agentWakeupRequests)
.where(
and(
eq(agentWakeupRequests.companyId, issue.companyId),
inArray(agentWakeupRequests.status, [...ACTIVE_WAKE_STATUSES]),
sql`${agentWakeupRequests.runId} is null`,
eq(sql<string>`${agentWakeupRequests.payload} ->> 'issueId'`, issue.id),
),
)
.orderBy(desc(agentWakeupRequests.createdAt), desc(agentWakeupRequests.id))
.limit(1);
const executionState = parseIssueExecutionState(issue.executionState);
let participant: IssueExecutionParticipantState = "none";
if (executionState?.status === "pending") {
participant = isValidParticipant(executionState.currentParticipant) ? "valid" : "invalid";
}
let pendingInteraction: "none" | "live" = "none";
const interactionRows: Array<{
issueId: string;
companyId: string;
status: string;
kind: string;
createdByUserId: string | null;
createdAt: Date;
}> = await dbOrTx
.select({
issueId: issueThreadInteractions.issueId,
companyId: issueThreadInteractions.companyId,
status: issueThreadInteractions.status,
kind: issueThreadInteractions.kind,
createdByUserId: issueThreadInteractions.createdByUserId,
createdAt: issueThreadInteractions.createdAt,
})
.from(issueThreadInteractions)
.where(
and(
eq(issueThreadInteractions.companyId, issue.companyId),
eq(issueThreadInteractions.issueId, issue.id),
inArray(issueThreadInteractions.status, [...PENDING_INTERACTION_STATUSES]),
inArray(issueThreadInteractions.kind, [...WAITING_INTERACTION_KINDS]),
),
);
for (const row of interactionRows) {
if (isLiveExplicitInteractionWaitingPath(issue, row)) {
pendingInteraction = "live";
break;
}
}
let pendingApproval: "none" | "live" = "none";
const approvalRows: Array<{
issueId: string;
companyId: string;
status: string;
requestedByUserId: string | null;
linkedByUserId: string | null;
createdAt: Date;
updatedAt: Date;
linkedAt: Date;
}> = await dbOrTx
.select({
issueId: issueApprovals.issueId,
companyId: issueApprovals.companyId,
status: approvals.status,
requestedByUserId: approvals.requestedByUserId,
linkedByUserId: issueApprovals.linkedByUserId,
createdAt: approvals.createdAt,
updatedAt: approvals.updatedAt,
linkedAt: issueApprovals.createdAt,
})
.from(issueApprovals)
.innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id))
.where(
and(
eq(issueApprovals.companyId, issue.companyId),
eq(issueApprovals.issueId, issue.id),
inArray(approvals.status, [...PENDING_APPROVAL_STATUSES]),
),
);
for (const row of approvalRows) {
if (isLiveExplicitApprovalWaitingPath(issue, row)) {
pendingApproval = "live";
break;
}
}
const recoveryRows: Array<{ id: string }> = await dbOrTx
.select({ id: issues.id })
.from(issues)
.where(
and(
eq(issues.companyId, issue.companyId),
inArray(issues.originKind, [...OPEN_RECOVERY_ORIGIN_KINDS]),
eq(issues.originId, issue.id),
isNull(issues.hiddenAt),
notInArray(issues.status, [...OPEN_RECOVERY_TERMINAL_STATUSES]),
),
)
.limit(1);
const blockerRows: Array<{
id: string;
status: string;
assigneeAgentId: string | null;
assigneeUserId: string | null;
originKind: string | null;
}> = await dbOrTx
.select({
id: issues.id,
status: issues.status,
assigneeAgentId: issues.assigneeAgentId,
assigneeUserId: issues.assigneeUserId,
originKind: issues.originKind,
})
.from(issueRelations)
.innerJoin(issues, eq(issueRelations.issueId, issues.id))
.where(
and(
eq(issueRelations.companyId, issue.companyId),
eq(issueRelations.relatedIssueId, issue.id),
eq(issueRelations.type, "blocks"),
eq(issues.companyId, issue.companyId),
),
);
return {
issue: {
id: issue.id,
status: issue.status as IssueExecutionStateVector["issue"]["status"],
assigneeAgentId: issue.assigneeAgentId,
assigneeUserId: issue.assigneeUserId,
originKind: issue.originKind,
},
agent: agent
? {
id: agent.id,
status: agent.status,
}
: null,
execution: {
activeRun: executionRun ? (ACTIVE_RUN_STATUSES as readonly string[]).includes(executionRun.status) : false,
scheduledRetry: executionRun
? (SCHEDULED_RETRY_RUN_STATUSES as readonly string[]).includes(executionRun.status)
: false,
queuedWake: queuedWake?.status === "queued",
deferredExecution: queuedWake?.status === "deferred_issue_execution",
},
waits: {
participant,
pendingInteraction,
pendingApproval,
openRecoveryIssue: recoveryRows.length > 0 ||
blockerRows.some((blocker) =>
(OPEN_RECOVERY_ORIGIN_KINDS as readonly string[]).includes(blocker.originKind ?? "")
),
},
blockers: blockerRows.map((blocker) => ({
issue: {
id: blocker.id,
status: blocker.status as IssueExecutionStateVector["issue"]["status"],
assigneeAgentId: blocker.assigneeAgentId,
assigneeUserId: blocker.assigneeUserId,
},
})),
};
}
export async function assertIssueTransitionAllowed(input: {
db: DbReader;
issue: IssueRow;
source: IssueTransitionGuardSource;
statusTouched?: boolean;
assigneeTouched?: boolean;
blockerTouched?: boolean;
executionStateTouched?: boolean;
}) {
const shouldGuard =
isGuardedStatus(input.issue.status) &&
(input.source === "create" ||
input.statusTouched ||
input.assigneeTouched ||
input.blockerTouched ||
input.executionStateTouched);
if (!shouldGuard) return;
const stateVector = await buildIssueExecutionStateVector(input.db, input.issue);
const disposition = classifyIssueExecutionDisposition(stateVector);
if (disposition.kind !== "invalid") return;
throw unprocessable(invalidMessage(disposition), {
disposition,
suggestedCorrection: disposition.suggestedCorrection,
});
}
server/src/services/issues.ts
+23 -120
View File
@@ -33,7 +33,6 @@ import type {
IssueBlockerAttentionState,
IssueBlockerNextActionHint,
IssueBlockerNextActionOwner,
IssueStatus,
IssueProductivityReview,
IssueProductivityReviewTrigger,
IssueRelationIssueSummary,
@@ -60,17 +59,6 @@ import {
isLiveExplicitApprovalWaitingPath,
isLiveExplicitInteractionWaitingPath,
} from "./recovery/explicit-waiting-paths.js";
import { parseIssueGraphLivenessIncidentKey } from "./recovery/origins.js";
import {
classifyIssueExecutionDisposition,
type IssueExecutionRunLivenessState,
} from "./issue-execution-disposition.js";
import {
backfillBlockedBySummaries,
listIssueExecutionDispositionsMap,
type IssueExecutionDispositionInputNode,
} from "./issue-execution-disposition-resolver.js";
import { assertIssueTransitionAllowed } from "./issue-transition-guard.js";
const ALL_ISSUE_STATUSES = ["backlog", "todo", "in_progress", "in_review", "blocked", "done", "cancelled"];
const MAX_ISSUE_COMMENT_PAGE_LIMIT = 500;
@@ -896,37 +884,6 @@ function classifyLeafRecovery(input: {
// the existing "no owner" surface continues to apply.
if (owner.type === "none") return null;
const disposition = classifyIssueExecutionDisposition({
issue: {
id: node.id,
status: node.status as IssueStatus,
assigneeAgentId: node.assigneeAgentId,
assigneeUserId: node.assigneeUserId,
},
agent: node.assigneeAgentId
? {
id: node.assigneeAgentId,
status: agentInvokable ? "idle" : "paused",
}
: null,
execution: {
activeRun: hasActivePath,
},
waits: {
productivityReviewNeeded: hasProductivityReview,
},
latestRun: {
latestRunStatus: "succeeded",
livenessState: latestRun.livenessState as IssueExecutionRunLivenessState,
nextAction: "runnable",
continuationAttempt: latestRun.continuationAttempt,
maxContinuationAttempts,
},
});
if (disposition.kind !== "agent_continuable" && disposition.kind !== "human_escalation_required") {
return null;
}
let reason: LeafRecoveryClassification["reason"];
if (hasProductivityReview) {
reason = "continuation_suppressed";
@@ -1428,36 +1385,29 @@ async function listIssueBlockerAttentionMap(
})));
}
}
const recoveryRows: Array<{
id: string;
originId: string | null;
assigneeAgentId: string | null;
assigneeUserId: string | null;
}> = await dbOrTx
.select({
id: issues.id,
originId: issues.originId,
assigneeAgentId: issues.assigneeAgentId,
assigneeUserId: issues.assigneeUserId,
})
.from(issues)
.where(
and(
eq(issues.companyId, companyId),
eq(issues.originKind, BLOCKER_ATTENTION_OPEN_RECOVERY_ORIGIN_KIND),
isNull(issues.hiddenAt),
notInArray(issues.status, BLOCKER_ATTENTION_OPEN_RECOVERY_TERMINAL_STATUSES),
),
);
for (const row of recoveryRows) {
const parsed = parseIssueGraphLivenessIncidentKey(row.originId);
if (!parsed || parsed.companyId !== companyId) continue;
const waitingPath = explicitWaitingPathForOwner(ownerForExplicitWait(row));
setExplicitWaitingPath(row.id, waitingPath);
setExplicitWaitingPath(parsed.issueId, waitingPath);
setExplicitWaitingPath(parsed.leafIssueId, waitingPath);
const recoveryRows: Array<{
originId: string | null;
assigneeAgentId: string | null;
assigneeUserId: string | null;
}> = await dbOrTx
.select({
originId: issues.originId,
assigneeAgentId: issues.assigneeAgentId,
assigneeUserId: issues.assigneeUserId,
})
.from(issues)
.where(
and(
eq(issues.companyId, companyId),
eq(issues.originKind, BLOCKER_ATTENTION_OPEN_RECOVERY_ORIGIN_KIND),
isNull(issues.hiddenAt),
inArray(issues.originId, chunk),
notInArray(issues.status, BLOCKER_ATTENTION_OPEN_RECOVERY_TERMINAL_STATUSES),
),
);
for (const row of recoveryRows) {
setExplicitWaitingPath(row.originId, explicitWaitingPathForOwner(ownerForExplicitWait(row)));
}
}
}
@@ -2716,23 +2666,6 @@ export function issueService(db: Db) {
listIssueBlockerAttentionMap(db, companyId, withRuns),
listIssueProductivityReviewMap(db, companyId, issueIds),
]);
const dispositionInputs: IssueExecutionDispositionInputNode[] = withRuns.map((row) => ({
id: row.id,
companyId: row.companyId,
status: row.status,
assigneeAgentId: row.assigneeAgentId,
assigneeUserId: row.assigneeUserId,
originKind: row.originKind ?? null,
executionRunId: row.executionRunId ?? null,
// executionState is intentionally omitted from the list select; the resolver rehydrates it.
blockedBy: includeBlockedBy ? blockedByMap.get(row.id) ?? [] : undefined,
}));
await backfillBlockedBySummaries(db, companyId, dispositionInputs);
const dispositionByIssueId = await listIssueExecutionDispositionsMap(
db,
companyId,
dispositionInputs,
);
if (!contextUserId) {
return withRuns.map((row) => {
@@ -2747,7 +2680,6 @@ export function issueService(db: Db) {
...(includeBlockedBy ? { blockedBy: blockedByMap.get(row.id) ?? [] } : {}),
lastActivityAt,
...(blockerAttentionByIssueId.has(row.id) ? { blockerAttention: blockerAttentionByIssueId.get(row.id) } : {}),
...(dispositionByIssueId.has(row.id) ? { executionDisposition: dispositionByIssueId.get(row.id) } : {}),
...(productivityReviewByIssueId.has(row.id)
? { productivityReview: productivityReviewByIssueId.get(row.id) }
: {}),
@@ -2769,7 +2701,6 @@ export function issueService(db: Db) {
...(includeBlockedBy ? { blockedBy: blockedByMap.get(row.id) ?? [] } : {}),
lastActivityAt,
...(blockerAttentionByIssueId.has(row.id) ? { blockerAttention: blockerAttentionByIssueId.get(row.id) } : {}),
...(dispositionByIssueId.has(row.id) ? { executionDisposition: dispositionByIssueId.get(row.id) } : {}),
...(productivityReviewByIssueId.has(row.id)
? { productivityReview: productivityReviewByIssueId.get(row.id) }
: {}),
@@ -2924,16 +2855,6 @@ export function issueService(db: Db) {
return listIssueBlockerAttentionMap(dbOrTx, companyId, issueRows);
},
listExecutionDispositions: async (
companyId: string,
issueRows: IssueExecutionDispositionInputNode[],
dbOrTx: any = db,
) => {
const inputs = issueRows.map((row) => ({ ...row }));
await backfillBlockedBySummaries(dbOrTx, companyId, inputs);
return listIssueExecutionDispositionsMap(dbOrTx, companyId, inputs);
},
listProductivityReviews: async (
companyId: string,
sourceIssueIds: string[],
@@ -3309,15 +3230,6 @@ export function issueService(db: Db) {
tx,
);
}
await assertIssueTransitionAllowed({
db: tx,
issue,
source: "create",
statusTouched: true,
assigneeTouched: issueData.assigneeAgentId !== undefined || issueData.assigneeUserId !== undefined,
blockerTouched: blockedByIssueIds !== undefined,
executionStateTouched: issueData.executionState !== undefined,
});
const [enriched] = await withIssueLabels(tx, [issue]);
return enriched;
});
@@ -3471,15 +3383,6 @@ export function issueService(db: Db) {
tx,
);
}
await assertIssueTransitionAllowed({
db: tx,
issue: updated,
source: "update",
statusTouched: issueData.status !== undefined,
assigneeTouched: issueData.assigneeAgentId !== undefined || issueData.assigneeUserId !== undefined,
blockerTouched: blockedByIssueIds !== undefined,
executionStateTouched: issueData.executionState !== undefined,
});
const [enriched] = await withIssueLabels(tx, [updated]);
return enriched;
};
server/src/services/recovery/issue-graph-liveness.ts
+13 -93
View File
@@ -1,14 +1,8 @@
import { buildIssueGraphLivenessIncidentKey } from "./origins.js";
import type { IssueStatus } from "@paperclipai/shared";
import {
isLiveExplicitApprovalWaitingPath,
isLiveExplicitInteractionWaitingPath,
} from "./explicit-waiting-paths.js";
import {
classifyIssueExecutionDisposition,
type IssueExecutionDispositionAgentInput,
type IssueExecutionParticipantState,
} from "../issue-execution-disposition.js";
export type IssueLivenessSeverity = "warning" | "critical";
@@ -148,26 +142,6 @@ function hasActiveExecutionPath(
);
}
function hasActiveRunPath(
companyId: string,
issueId: string,
activeRuns: IssueLivenessExecutionPathInput[],
) {
return activeRuns.some(
(entry) => entry.companyId === companyId && entry.issueId === issueId,
);
}
function hasQueuedWakePath(
companyId: string,
issueId: string,
queuedWakeRequests: IssueLivenessExecutionPathInput[],
) {
return queuedWakeRequests.some(
(entry) => entry.companyId === companyId && entry.issueId === issueId,
);
}
function hasWaitingPath(
issue: IssueLivenessIssueInput,
waitingPaths: IssueLivenessWaitingPathInput[],
@@ -195,23 +169,6 @@ function principalIsResolvableUser(principal: unknown): boolean {
return value.type === "user" && typeof value.userId === "string" && value.userId.length > 0;
}
function participantStateForIssue(
issue: IssueLivenessIssueInput,
agentsById: Map<string, IssueLivenessAgentInput>,
): IssueExecutionParticipantState {
if (!issue.executionState) return "none";
const participant = issue.executionState.currentParticipant;
const participantAgentId = readPrincipalAgentId(participant);
if (participantAgentId) {
const participantAgent = agentsById.get(participantAgentId);
return participantAgent && participantAgent.companyId === issue.companyId && isInvokableAgent(participantAgent)
? "valid"
: "invalid";
}
if (principalIsResolvableUser(participant)) return "valid";
return "invalid";
}
function addOwnerCandidate(
candidates: IssueLivenessOwnerCandidate[],
seen: Set<string>,
@@ -416,42 +373,13 @@ export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): Issu
openRecoveryIssues.some((entry) => entry.companyId === issue.companyId && entry.issueId === issue.id);
}
function canonicalDisposition(issue: IssueLivenessIssueInput) {
const agent: IssueExecutionDispositionAgentInput | null = issue.assigneeAgentId
? agentsById.get(issue.assigneeAgentId) ?? null
: null;
return classifyIssueExecutionDisposition({
issue: {
id: issue.id,
status: issue.status as IssueStatus,
assigneeAgentId: issue.assigneeAgentId,
assigneeUserId: issue.assigneeUserId,
},
agent,
execution: {
activeRun: hasActiveRunPath(issue.companyId, issue.id, activeRuns),
queuedWake: hasQueuedWakePath(issue.companyId, issue.id, queuedWakeRequests),
},
waits: {
participant: participantStateForIssue(issue, agentsById),
pendingInteraction: hasWaitingPath(issue, pendingInteractions, "interaction", now) ? "live" : "none",
pendingApproval: hasWaitingPath(issue, pendingApprovals, "approval", now) ? "live" : "none",
openRecoveryIssue: openRecoveryIssues.some((entry) =>
entry.companyId === issue.companyId && entry.issueId === issue.id
),
},
});
}
function reviewFinding(
source: IssueLivenessIssueInput,
reviewIssue: IssueLivenessIssueInput,
dependencyPath: IssueLivenessIssueInput[],
): IssueLivenessFinding | null {
if (reviewIssue.status !== "in_review") return null;
const disposition = canonicalDisposition(reviewIssue);
if (disposition.kind !== "invalid") return null;
if (hasExplicitWaitingPath(reviewIssue)) return null;
const ownerCandidates = ownerCandidatesForRecoveryIssue(reviewIssue, input.agents, agentsById, {
includeStalledAssignee: true,
@@ -459,8 +387,9 @@ export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): Issu
const participant = reviewIssue.executionState?.currentParticipant;
const participantAgentId = readPrincipalAgentId(participant);
if (participantAgentId && disposition.reason === "invalid_review_participant") {
if (participantAgentId) {
const participantAgent = agentsById.get(participantAgentId);
if (isInvokableAgent(participantAgent) && participantAgent?.companyId === reviewIssue.companyId) return null;
return finding({
issue: source,
@@ -478,7 +407,9 @@ export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): Issu
});
}
if (disposition.reason === "invalid_review_participant") {
if (principalIsResolvableUser(participant)) return null;
if (reviewIssue.executionState) {
return finding({
issue: source,
state: "invalid_review_participant",
@@ -492,7 +423,7 @@ export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): Issu
});
}
if (disposition.reason !== "in_review_without_action_path") return null;
if (!reviewIssue.assigneeAgentId || reviewIssue.assigneeUserId) return null;
return finding({
issue: source,
@@ -516,7 +447,6 @@ export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): Issu
const ownerCandidates = ownerCandidatesForRecoveryIssue(blocker, input.agents, agentsById, {
includeStalledAssignee: true,
});
const disposition = canonicalDisposition(blocker);
if (blocker.status === "cancelled") {
return finding({
@@ -533,21 +463,13 @@ export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): Issu
});
}
if (
disposition.kind === "waiting" ||
disposition.kind === "live" ||
disposition.kind === "dispatchable" ||
disposition.kind === "agent_continuable" ||
disposition.kind === "recoverable_by_control_plane"
) {
return null;
}
if (hasExplicitWaitingPath(blocker)) return null;
if (blocker.status === "in_review") {
return reviewFinding(source, blocker, dependencyPath);
}
if (disposition.kind === "resting" && !blocker.assigneeAgentId && !blocker.assigneeUserId) {
if (!blocker.assigneeAgentId && !blocker.assigneeUserId) {
return finding({
issue: source,
state: "blocked_by_unassigned_issue",
@@ -562,12 +484,10 @@ export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): Issu
});
}
const blockerAgent = blocker.assigneeAgentId ? agentsById.get(blocker.assigneeAgentId) : null;
if (
disposition.kind === "human_escalation_required" &&
blocker.assigneeAgentId &&
(!blockerAgent || blockerAgent.companyId !== source.companyId || BLOCKING_AGENT_STATUSES.has(blockerAgent.status))
) {
if (!blocker.assigneeAgentId) return null;
const blockerAgent = agentsById.get(blocker.assigneeAgentId);
if (!blockerAgent || blockerAgent.companyId !== source.companyId || BLOCKING_AGENT_STATUSES.has(blockerAgent.status)) {
return finding({
issue: source,
state: "blocked_by_uninvokable_assignee",
server/src/services/recovery/run-liveness-continuations.ts
+17 -52
View File
@@ -1,18 +1,18 @@
import { and, eq, inArray } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import { agentWakeupRequests, agents, heartbeatRuns, issues } from "@paperclipai/db";
import type { IssueStatus, RunLivenessState } from "@paperclipai/shared";
import type { RunLivenessState } from "@paperclipai/shared";
import { RECOVERY_REASON_KINDS } from "./origins.js";
import {
classifyIssueExecutionDisposition,
type IssueExecutionRunLivenessState,
type IssueExecutionNextActionState,
} from "../issue-execution-disposition.js";
export const RUN_LIVENESS_CONTINUATION_REASON = RECOVERY_REASON_KINDS.runLivenessContinuation;
export const DEFAULT_MAX_LIVENESS_CONTINUATION_ATTEMPTS = 2;
const ACTIONABLE_LIVENESS_STATES = new Set<RunLivenessState>(["plan_only", "empty_response"]);
const PROGRESS_WITH_FOLLOWUP_LIVENESS_STATES = new Set<RunLivenessState>(["advanced"]);
const CONTINUATION_ACTIVE_ISSUE_STATUSES = new Set(["todo", "in_progress"]);
// A prior adapter error should not permanently suppress bounded liveness
// continuations; the max-attempt/idempotency guards prevent unbounded retries.
const CONTINUATION_AGENT_STATUSES = new Set(["active", "idle", "running", "error"]);
const IDEMPOTENT_WAKE_STATUSES = ["queued", "deferred_issue_execution", "completed"];
type HeartbeatRunRow = Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "agentId" | "continuationAttempt">;
@@ -61,14 +61,6 @@ export function buildRunLivenessContinuationIdempotencyKey(input: {
].join(":");
}
function nextActionStateForContinuation(nextAction: string | null): IssueExecutionNextActionState {
return nextAction?.trim() ? "runnable" : "none";
}
function livenessStateForDisposition(livenessState: RunLivenessState): IssueExecutionRunLivenessState {
return livenessState;
}
export async function findExistingRunLivenessContinuationWake(
db: Db,
input: {
@@ -117,6 +109,13 @@ export function decideRunLivenessContinuation(input: {
return { kind: "skip", reason: "liveness state is not actionable for continuation" };
}
const hasExplicitRunnableFollowup = Boolean(nextAction?.trim());
const livenessCanContinue =
ACTIONABLE_LIVENESS_STATES.has(livenessState) ||
PROGRESS_WITH_FOLLOWUP_LIVENESS_STATES.has(livenessState) && hasExplicitRunnableFollowup;
if (!livenessCanContinue) {
return { kind: "skip", reason: "liveness state is not actionable for continuation" };
}
if (!issue) return { kind: "skip", reason: "issue not found" };
if (!agent) return { kind: "skip", reason: "agent not found" };
if (issue.companyId !== run.companyId || agent.companyId !== run.companyId) {
@@ -131,48 +130,14 @@ export function decideRunLivenessContinuation(input: {
if (issue.executionState) {
return { kind: "skip", reason: "issue is blocked by execution policy state" };
}
const currentAttempt = readContinuationAttempt(run.continuationAttempt);
const disposition = classifyIssueExecutionDisposition({
issue: {
id: issue.id,
status: issue.status as IssueStatus,
assigneeAgentId: issue.assigneeAgentId,
assigneeUserId: null,
},
agent,
latestRun: {
latestRunStatus: "succeeded",
livenessState: livenessStateForDisposition(livenessState),
nextAction: nextActionStateForContinuation(nextAction),
continuationAttempt: currentAttempt,
maxContinuationAttempts: maxAttempts,
},
gates: { budgetBlocked },
});
if (!CONTINUATION_AGENT_STATUSES.has(agent.status)) {
return { kind: "skip", reason: `agent status ${agent.status} is not invokable` };
}
if (budgetBlocked) {
return { kind: "skip", reason: "budget hard stop blocks continuation" };
}
if (disposition.kind === "human_escalation_required" && disposition.owner === "manager" && currentAttempt >= maxAttempts) {
return {
kind: "exhausted",
attempt: currentAttempt,
maxAttempts,
comment: [
"Bounded liveness continuation exhausted",
"",
`- Last liveness state: \`${livenessState}\``,
`- Attempts used: ${currentAttempt}/${maxAttempts}`,
`- Reason: ${livenessReason ?? "Run ended without concrete progress"}`,
"- Next action: a human or manager should inspect the run and either clarify the task, mark it blocked, or assign a concrete follow-up.",
].join("\n"),
};
}
if (disposition.kind !== "agent_continuable") {
return { kind: "skip", reason: `canonical disposition ${disposition.kind} is not continuable` };
}
const currentAttempt = readContinuationAttempt(run.continuationAttempt);
if (currentAttempt >= maxAttempts) {
return {
kind: "exhausted",
server/src/services/recovery/service.ts
+27 -101
View File
@@ -6,7 +6,6 @@ import {
MIN_ISSUE_GRAPH_LIVENESS_AUTO_RECOVERY_LOOKBACK_HOURS,
type IssueGraphLivenessAutoRecoveryPreview,
type IssueGraphLivenessAutoRecoveryPreviewItem,
type IssueStatus,
type RunLivenessState,
} from "@paperclipai/shared";
import {
@@ -24,7 +23,7 @@ import {
} from "@paperclipai/db";
import { parseObject, asBoolean, asNumber } from "../../adapters/utils.js";
import { runningProcesses } from "../../adapters/index.js";
import { badRequest, conflict, forbidden, notFound } from "../../errors.js";
import { forbidden, notFound } from "../../errors.js";
import { logger } from "../../middleware/logger.js";
import { redactCurrentUserText } from "../../log-redaction.js";
import { redactSensitiveText } from "../../redaction.js";
@@ -57,14 +56,8 @@ import {
findExistingRunLivenessContinuationWake,
readContinuationAttempt,
} from "./run-liveness-continuations.js";
import {
classifyIssueExecutionDisposition,
type IssueExecutionRunLivenessState,
type IssueExecutionRunStatus,
} from "../issue-execution-disposition.js";
const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const;
const CANCELLABLE_ACTIVE_RUN_RECOVERY_STATUSES = ["queued", "running", "scheduled_retry"] as const;
const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = ["failed", "cancelled", "timed_out"] as const;
export const ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS = 60 * 60 * 1000;
export const ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS = 4 * 60 * 60 * 1000;
@@ -113,8 +106,6 @@ type WatchdogDecisionActor =
| { type: "agent"; agentId?: string | null; runId?: string | null }
| { type: "none" };
type WatchdogDecision = "snooze" | "continue" | "dismissed_false_positive" | "cancel_run";
export type RunOutputSilenceSummary = {
lastOutputAt: Date | null;
lastOutputSeq: number;
@@ -488,47 +479,14 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
issue: typeof issues.$inferSelect,
latestRun: LatestIssueRun,
): Promise<PostRunIssueDisposition> {
const [activePath, explicitWait, agent, budgetBlocked] = await Promise.all([
hasActiveExecutionPath(issue.companyId, issue.id),
hasExplicitWaitingPath(issue),
issue.assigneeAgentId ? getAgent(issue.assigneeAgentId) : Promise.resolve(null),
issue.assigneeAgentId ? isInvocationBudgetBlocked(issue, issue.assigneeAgentId) : Promise.resolve(false),
]);
const activeRunFromLatest = Boolean(
latestRun &&
EXECUTION_PATH_HEARTBEAT_RUN_STATUSES.includes(
latestRun.status as (typeof EXECUTION_PATH_HEARTBEAT_RUN_STATUSES)[number],
),
);
const disposition = classifyIssueExecutionDisposition({
issue: {
id: issue.id,
status: issue.status as IssueStatus,
assigneeAgentId: issue.assigneeAgentId,
assigneeUserId: issue.assigneeUserId,
originKind: issue.originKind,
},
agent,
execution: {
activeRun: activePath || activeRunFromLatest,
},
waits: {
openRecoveryIssue: explicitWait,
},
latestRun: latestRun
? {
latestRunStatus: latestRun.status as IssueExecutionRunStatus,
livenessState: latestRun.livenessState as IssueExecutionRunLivenessState | null,
nextAction: latestRun.nextAction ? "runnable" : "none",
continuationAttempt: latestRun.continuationAttempt,
}
: null,
gates: { budgetBlocked },
});
if (disposition.kind === "terminal") return "terminal";
if (disposition.kind === "live") return "explicitly_live";
if (disposition.kind === "waiting") return "explicitly_waiting";
if (issue.status === "done" || issue.status === "cancelled") return "terminal";
if (await hasActiveExecutionPath(issue.companyId, issue.id)) return "explicitly_live";
if (await hasExplicitWaitingPath(issue)) return "explicitly_waiting";
if (latestRun && EXECUTION_PATH_HEARTBEAT_RUN_STATUSES.includes(
latestRun.status as (typeof EXECUTION_PATH_HEARTBEAT_RUN_STATUSES)[number],
)) {
return "explicitly_live";
}
return "invalid";
}
@@ -1345,7 +1303,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
async function recordWatchdogDecision(input: {
runId: string;
actor: WatchdogDecisionActor;
decision: WatchdogDecision;
decision: "snooze" | "continue" | "dismissed_false_positive";
evaluationIssueId?: string | null;
reason?: string | null;
snoozedUntil?: Date | null;
@@ -1359,15 +1317,6 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
.limit(1);
if (!run) throw notFound("Heartbeat run not found");
const normalizedReason = typeof input.reason === "string" ? input.reason.trim() : "";
if (input.decision === "cancel_run") {
if (!normalizedReason) throw badRequest("Cancellation reason is required");
if (!input.evaluationIssueId) throw badRequest("evaluationIssueId is required for cancellation");
if (!CANCELLABLE_ACTIVE_RUN_RECOVERY_STATUSES.includes(run.status as (typeof CANCELLABLE_ACTIVE_RUN_RECOVERY_STATUSES)[number])) {
throw conflict("Heartbeat run is not cancellable", { status: run.status });
}
}
let evaluationIssue: {
id: string;
assigneeAgentId: string | null;
@@ -1454,7 +1403,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
evaluationIssueId: input.evaluationIssueId ?? null,
decision: input.decision,
snoozedUntil: effectiveSnoozedUntil,
reason: input.decision === "cancel_run" ? normalizedReason : input.reason ?? null,
reason: input.reason ?? null,
createdByAgentId: input.actor.type === "agent" ? input.actor.agentId ?? null : null,
createdByUserId: input.actor.type === "board" ? input.actor.userId ?? null : null,
createdByRunId,
@@ -1471,21 +1420,15 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
: "unknown",
agentId: input.actor.type === "agent" ? input.actor.agentId ?? null : null,
runId: run.id,
action: input.decision === "snooze"
? "heartbeat.watchdog_snoozed"
: input.decision === "cancel_run"
? "heartbeat.watchdog_cancel_requested"
: "heartbeat.watchdog_decision_recorded",
action: input.decision === "snooze" ? "heartbeat.watchdog_snoozed" : "heartbeat.watchdog_decision_recorded",
entityType: "heartbeat_run",
entityId: run.id,
details: {
source: "recovery.record_watchdog_decision",
decision: input.decision,
evaluationIssueId: input.evaluationIssueId ?? null,
targetRunId: run.id,
snoozedUntil: effectiveSnoozedUntil?.toISOString() ?? null,
reason: input.decision === "cancel_run" ? normalizedReason : input.reason ?? null,
createdByRunId,
reason: input.reason ?? null,
},
});
@@ -2414,36 +2357,31 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
return `${companyId}:${issueId}`;
}
type LivenessDependencyTimestamps = {
createdAt: Date;
updatedAt: Date;
};
async function loadLivenessDependencyTimestampsByIssue(findings: IssueLivenessFinding[]) {
async function loadLivenessDependencyUpdatedAtByIssue(findings: IssueLivenessFinding[]) {
const issueIds = [
...new Set(
findings.flatMap((finding) => finding.dependencyPath.map((entry) => entry.issueId)),
),
];
if (issueIds.length === 0) return new Map<string, LivenessDependencyTimestamps>();
if (issueIds.length === 0) return new Map<string, Date>();
const rows = await db
.select({ id: issues.id, companyId: issues.companyId, createdAt: issues.createdAt, updatedAt: issues.updatedAt })
.select({ id: issues.id, companyId: issues.companyId, updatedAt: issues.updatedAt })
.from(issues)
.where(inArray(issues.id, issueIds));
return new Map(rows.map((row) => [
livenessDependencyIssueKey(row.companyId, row.id),
{ createdAt: row.createdAt, updatedAt: row.updatedAt },
row.updatedAt,
]));
}
function latestDependencyUpdatedAtForLivenessFinding(
finding: IssueLivenessFinding,
timestampsByIssueKey: Map<string, LivenessDependencyTimestamps>,
updatedAtByIssueKey: Map<string, Date>,
) {
const dependencyIssueIds = [...new Set(finding.dependencyPath.map((entry) => entry.issueId))];
if (dependencyIssueIds.length === 0) return null;
const timestamps = dependencyIssueIds.map((issueId) =>
timestampsByIssueKey.get(livenessDependencyIssueKey(finding.companyId, issueId))?.updatedAt ?? null
updatedAtByIssueKey.get(livenessDependencyIssueKey(finding.companyId, issueId)) ?? null
);
if (timestamps.some((timestamp) => !timestamp)) return null;
const [firstTimestamp, ...remainingTimestamps] = timestamps as Date[];
@@ -2452,22 +2390,13 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
firstTimestamp!);
}
function recoveryIssueCreatedAtForLivenessFinding(
finding: IssueLivenessFinding,
timestampsByIssueKey: Map<string, LivenessDependencyTimestamps>,
) {
return timestampsByIssueKey.get(
livenessDependencyIssueKey(finding.companyId, finding.recoveryIssueId),
)?.createdAt ?? null;
}
function isLivenessFindingInsideAutoRecoveryLookback(
finding: IssueLivenessFinding,
cutoff: Date,
timestampsByIssueKey: Map<string, LivenessDependencyTimestamps>,
updatedAtByIssueKey: Map<string, Date>,
) {
const recoveryIssueCreatedAt = recoveryIssueCreatedAtForLivenessFinding(finding, timestampsByIssueKey);
return Boolean(recoveryIssueCreatedAt && recoveryIssueCreatedAt >= cutoff);
const latestUpdatedAt = latestDependencyUpdatedAtForLivenessFinding(finding, updatedAtByIssueKey);
return Boolean(latestUpdatedAt && latestUpdatedAt >= cutoff);
}
async function buildIssueGraphLivenessAutoRecoveryPreview(
@@ -2477,7 +2406,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
const lookbackHours = normalizeIssueGraphLivenessAutoRecoveryLookbackHours(opts?.lookbackHours);
const cutoff = new Date(now.getTime() - lookbackHours * 60 * 60 * 1000);
const findings = await collectIssueGraphLivenessFindings();
const timestampsByIssueKey = await loadLivenessDependencyTimestampsByIssue(findings);
const updatedAtByIssueKey = await loadLivenessDependencyUpdatedAtByIssue(findings);
const issueIds = [...new Set(findings.map((finding) => finding.recoveryIssueId))];
const recoveryRows = issueIds.length > 0
? await db
@@ -2492,12 +2421,9 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
for (const finding of findings) {
const latestDependencyUpdatedAt = latestDependencyUpdatedAtForLivenessFinding(
finding,
timestampsByIssueKey,
updatedAtByIssueKey,
);
if (
!latestDependencyUpdatedAt ||
!isLivenessFindingInsideAutoRecoveryLookback(finding, cutoff, timestampsByIssueKey)
) {
if (!latestDependencyUpdatedAt || latestDependencyUpdatedAt < cutoff) {
skippedOutsideLookback += 1;
continue;
}
@@ -2807,7 +2733,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
const now = new Date();
const cutoff = new Date(now.getTime() - lookbackHours * 60 * 60 * 1000);
const obsoleteRecoveryCleanup = await retireObsoleteLivenessRecoveryIssues(findings);
const timestampsByIssueKey = await loadLivenessDependencyTimestampsByIssue(findings);
const updatedAtByIssueKey = await loadLivenessDependencyUpdatedAtByIssue(findings);
const result = {
findings: findings.length,
autoRecoveryEnabled,
@@ -2832,7 +2758,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
}
for (const finding of findings) {
if (!isLivenessFindingInsideAutoRecoveryLookback(finding, cutoff, timestampsByIssueKey)) {
if (!isLivenessFindingInsideAutoRecoveryLookback(finding, cutoff, updatedAtByIssueKey)) {
result.skippedOutsideLookback += 1;
result.skipped += 1;
continue;