forked from farhoodlabs/paperclip
6916e30f8e
## Thinking Path > - Paperclip orchestrates AI agents for zero-human companies > - Issue execution is guarded by run locks and bounded retry scheduling > - A failed run can schedule a retry, but the issue may be reassigned before that retry becomes due > - The old assignee's scheduled retry should not continue to hold or reclaim execution for the issue > - This pull request cancels stale scheduled retries when ownership changes and cancels live work when an issue is explicitly cancelled > - The benefit is cleaner issue handoff semantics and fewer stranded or incorrect execution locks ## What Changed - Cancel scheduled retry runs when their issue has been reassigned before the retry is promoted. - Clear stale issue execution locks and cancel the associated wakeup request when a stale retry is cancelled. - Avoid deferring a new assignee behind a previous assignee's scheduled retry. - Cancel an active run when an issue status is explicitly changed to `cancelled`, while leaving `done` transitions alone. - Added route and heartbeat regressions for reassignment and cancellation behavior. ## Verification - `pnpm exec vitest run --project @paperclipai/server server/src/__tests__/heartbeat-retry-scheduling.test.ts server/src/__tests__/issue-comment-reopen-routes.test.ts --pool=forks --poolOptions.forks.isolate=true` - `issue-comment-reopen-routes.test.ts`: 28 passed. - `heartbeat-retry-scheduling.test.ts`: skipped by the existing embedded Postgres host guard (`Postgres init script exited with code null`). - `pnpm --filter @paperclipai/server typecheck` ## Risks - Medium risk because this changes heartbeat retry lifecycle behavior. - The cancellation path is scoped to scheduled retries whose issue assignee no longer matches the retrying agent, and logs a lifecycle event for auditability. - No migrations. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - OpenAI Codex, GPT-5 coding agent, tool-enabled with shell/GitHub/Paperclip API access. Context window was not reported by the runtime. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] If this change affects the UI, I have included before/after screenshots - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
838 lines
25 KiB
TypeScript
838 lines
25 KiB
TypeScript
import { randomUUID } from "node:crypto";
|
|
import { eq, sql } from "drizzle-orm";
|
|
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
|
|
import {
|
|
agents,
|
|
agentRuntimeState,
|
|
agentWakeupRequests,
|
|
companies,
|
|
createDb,
|
|
environmentLeases,
|
|
heartbeatRunEvents,
|
|
heartbeatRuns,
|
|
issues,
|
|
} from "@paperclipai/db";
|
|
import {
|
|
getEmbeddedPostgresTestSupport,
|
|
startEmbeddedPostgresTestDatabase,
|
|
} from "./helpers/embedded-postgres.js";
|
|
import {
|
|
BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS,
|
|
heartbeatService,
|
|
} from "../services/heartbeat.ts";
|
|
|
|
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
|
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
|
|
|
if (!embeddedPostgresSupport.supported) {
|
|
console.warn(
|
|
`Skipping embedded Postgres heartbeat retry scheduling tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
|
|
);
|
|
}
|
|
|
|
describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => {
|
|
let db!: ReturnType<typeof createDb>;
|
|
let heartbeat!: ReturnType<typeof heartbeatService>;
|
|
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
|
|
|
beforeAll(async () => {
|
|
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-retry-scheduling-");
|
|
db = createDb(tempDb.connectionString);
|
|
heartbeat = heartbeatService(db);
|
|
}, 20_000);
|
|
|
|
afterEach(async () => {
|
|
await db.delete(heartbeatRunEvents);
|
|
await db.delete(environmentLeases);
|
|
await db.delete(issues);
|
|
await db.delete(heartbeatRuns);
|
|
await db.delete(agentWakeupRequests);
|
|
await db.delete(agentRuntimeState);
|
|
await db.delete(agents);
|
|
await db.delete(companies);
|
|
});
|
|
|
|
afterAll(async () => {
|
|
await tempDb?.cleanup();
|
|
});
|
|
|
|
async function seedRetryFixture(input: {
|
|
runId: string;
|
|
companyId: string;
|
|
agentId: string;
|
|
now: Date;
|
|
errorCode: string;
|
|
errorFamily?: "transient_upstream" | null;
|
|
retryNotBefore?: string | null;
|
|
scheduledRetryAttempt?: number;
|
|
resultJson?: Record<string, unknown> | null;
|
|
adapterType?: "codex_local" | "claude_local";
|
|
agentName?: string;
|
|
}) {
|
|
const adapterType = input.adapterType ?? "codex_local";
|
|
const agentName = input.agentName ?? (adapterType === "claude_local" ? "ClaudeCoder" : "CodexCoder");
|
|
await db.insert(companies).values({
|
|
id: input.companyId,
|
|
name: "Paperclip",
|
|
issuePrefix: `T${input.companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
|
|
requireBoardApprovalForNewAgents: false,
|
|
});
|
|
|
|
await db.insert(agents).values({
|
|
id: input.agentId,
|
|
companyId: input.companyId,
|
|
name: agentName,
|
|
role: "engineer",
|
|
status: "active",
|
|
adapterType,
|
|
adapterConfig: {},
|
|
runtimeConfig: {
|
|
heartbeat: {
|
|
wakeOnDemand: true,
|
|
maxConcurrentRuns: 1,
|
|
},
|
|
},
|
|
permissions: {},
|
|
});
|
|
|
|
await db.insert(heartbeatRuns).values({
|
|
id: input.runId,
|
|
companyId: input.companyId,
|
|
agentId: input.agentId,
|
|
invocationSource: "assignment",
|
|
status: "failed",
|
|
error: "upstream overload",
|
|
errorCode: input.errorCode,
|
|
finishedAt: input.now,
|
|
scheduledRetryAttempt: input.scheduledRetryAttempt ?? 0,
|
|
scheduledRetryReason: input.scheduledRetryAttempt ? "transient_failure" : null,
|
|
resultJson: input.resultJson ?? {
|
|
...(input.errorFamily ? { errorFamily: input.errorFamily } : {}),
|
|
...(input.retryNotBefore
|
|
? {
|
|
retryNotBefore: input.retryNotBefore,
|
|
transientRetryNotBefore: input.retryNotBefore,
|
|
}
|
|
: {}),
|
|
},
|
|
contextSnapshot: {
|
|
issueId: randomUUID(),
|
|
wakeReason: "issue_assigned",
|
|
},
|
|
updatedAt: input.now,
|
|
createdAt: input.now,
|
|
});
|
|
}
|
|
|
|
it("schedules a retry with durable metadata and only promotes it when due", async () => {
|
|
const companyId = randomUUID();
|
|
const agentId = randomUUID();
|
|
const sourceRunId = randomUUID();
|
|
const now = new Date("2026-04-20T12:00:00.000Z");
|
|
|
|
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: agentId,
|
|
companyId,
|
|
name: "CodexCoder",
|
|
role: "engineer",
|
|
status: "active",
|
|
adapterType: "codex_local",
|
|
adapterConfig: {},
|
|
runtimeConfig: {
|
|
heartbeat: {
|
|
wakeOnDemand: true,
|
|
maxConcurrentRuns: 1,
|
|
},
|
|
},
|
|
permissions: {},
|
|
});
|
|
|
|
await db.insert(heartbeatRuns).values({
|
|
id: sourceRunId,
|
|
companyId,
|
|
agentId,
|
|
invocationSource: "assignment",
|
|
status: "failed",
|
|
error: "upstream overload",
|
|
errorCode: "adapter_failed",
|
|
finishedAt: now,
|
|
contextSnapshot: {
|
|
issueId: randomUUID(),
|
|
wakeReason: "issue_assigned",
|
|
},
|
|
updatedAt: now,
|
|
createdAt: now,
|
|
});
|
|
|
|
const scheduled = await heartbeat.scheduleBoundedRetry(sourceRunId, {
|
|
now,
|
|
random: () => 0.5,
|
|
});
|
|
|
|
expect(scheduled.outcome).toBe("scheduled");
|
|
if (scheduled.outcome !== "scheduled") return;
|
|
|
|
const expectedDueAt = new Date(now.getTime() + BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS[0]);
|
|
expect(scheduled.attempt).toBe(1);
|
|
expect(scheduled.dueAt.toISOString()).toBe(expectedDueAt.toISOString());
|
|
|
|
const retryRun = await db
|
|
.select()
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, scheduled.run.id))
|
|
.then((rows) => rows[0] ?? null);
|
|
|
|
expect(retryRun).toMatchObject({
|
|
status: "scheduled_retry",
|
|
retryOfRunId: sourceRunId,
|
|
scheduledRetryAttempt: 1,
|
|
scheduledRetryReason: "transient_failure",
|
|
});
|
|
expect(retryRun?.scheduledRetryAt?.toISOString()).toBe(expectedDueAt.toISOString());
|
|
|
|
const earlyPromotion = await heartbeat.promoteDueScheduledRetries(new Date("2026-04-20T12:01:59.000Z"));
|
|
expect(earlyPromotion).toEqual({ promoted: 0, runIds: [] });
|
|
|
|
const stillScheduled = await db
|
|
.select({ status: heartbeatRuns.status })
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, scheduled.run.id))
|
|
.then((rows) => rows[0] ?? null);
|
|
expect(stillScheduled?.status).toBe("scheduled_retry");
|
|
|
|
const duePromotion = await heartbeat.promoteDueScheduledRetries(expectedDueAt);
|
|
expect(duePromotion).toEqual({ promoted: 1, runIds: [scheduled.run.id] });
|
|
|
|
const promotedRun = await db
|
|
.select({ status: heartbeatRuns.status })
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, scheduled.run.id))
|
|
.then((rows) => rows[0] ?? null);
|
|
expect(promotedRun?.status).toBe("queued");
|
|
});
|
|
|
|
it("does not defer a new assignee behind the previous assignee's scheduled retry", async () => {
|
|
const companyId = randomUUID();
|
|
const oldAgentId = randomUUID();
|
|
const newAgentId = randomUUID();
|
|
const issueId = randomUUID();
|
|
const sourceRunId = randomUUID();
|
|
const now = new Date("2026-04-20T13:00:00.000Z");
|
|
|
|
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: oldAgentId,
|
|
companyId,
|
|
name: "ClaudeCoder",
|
|
role: "engineer",
|
|
status: "active",
|
|
adapterType: "claude_local",
|
|
adapterConfig: {},
|
|
runtimeConfig: {
|
|
heartbeat: {
|
|
wakeOnDemand: true,
|
|
maxConcurrentRuns: 1,
|
|
},
|
|
},
|
|
permissions: {},
|
|
},
|
|
{
|
|
id: newAgentId,
|
|
companyId,
|
|
name: "CodexCoder",
|
|
role: "engineer",
|
|
status: "active",
|
|
adapterType: "codex_local",
|
|
adapterConfig: {},
|
|
runtimeConfig: {
|
|
heartbeat: {
|
|
wakeOnDemand: true,
|
|
maxConcurrentRuns: 1,
|
|
},
|
|
},
|
|
permissions: {},
|
|
},
|
|
]);
|
|
|
|
await db.insert(heartbeatRuns).values({
|
|
id: sourceRunId,
|
|
companyId,
|
|
agentId: oldAgentId,
|
|
invocationSource: "assignment",
|
|
triggerDetail: "system",
|
|
status: "failed",
|
|
error: "upstream overload",
|
|
errorCode: "adapter_failed",
|
|
finishedAt: now,
|
|
contextSnapshot: {
|
|
issueId,
|
|
wakeReason: "issue_assigned",
|
|
},
|
|
updatedAt: now,
|
|
createdAt: now,
|
|
});
|
|
|
|
await db.insert(issues).values({
|
|
id: issueId,
|
|
companyId,
|
|
title: "Retry reassignment",
|
|
status: "todo",
|
|
priority: "medium",
|
|
assigneeAgentId: oldAgentId,
|
|
executionRunId: sourceRunId,
|
|
executionAgentNameKey: "claudecoder",
|
|
executionLockedAt: now,
|
|
issueNumber: 1,
|
|
identifier: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}-1`,
|
|
});
|
|
|
|
const scheduled = await heartbeat.scheduleBoundedRetry(sourceRunId, {
|
|
now,
|
|
random: () => 0.5,
|
|
});
|
|
expect(scheduled.outcome).toBe("scheduled");
|
|
if (scheduled.outcome !== "scheduled") return;
|
|
|
|
await db.update(issues).set({
|
|
assigneeAgentId: newAgentId,
|
|
updatedAt: now,
|
|
}).where(eq(issues.id, issueId));
|
|
|
|
// Keep the new agent's queue from auto-claiming/executing during this unit test.
|
|
await db.insert(heartbeatRuns).values(
|
|
Array.from({ length: 5 }, () => ({
|
|
id: randomUUID(),
|
|
companyId,
|
|
agentId: newAgentId,
|
|
invocationSource: "automation",
|
|
triggerDetail: "system",
|
|
status: "running",
|
|
contextSnapshot: {
|
|
wakeReason: "test_busy_slot",
|
|
},
|
|
startedAt: now,
|
|
updatedAt: now,
|
|
createdAt: now,
|
|
})),
|
|
);
|
|
|
|
const newAssigneeRun = await heartbeat.wakeup(newAgentId, {
|
|
source: "assignment",
|
|
triggerDetail: "system",
|
|
reason: "issue_assigned",
|
|
payload: {
|
|
issueId,
|
|
mutation: "update",
|
|
},
|
|
contextSnapshot: {
|
|
issueId,
|
|
source: "issue.update",
|
|
},
|
|
requestedByActorType: "user",
|
|
requestedByActorId: "local-board",
|
|
});
|
|
|
|
expect(newAssigneeRun).not.toBeNull();
|
|
expect(newAssigneeRun?.agentId).toBe(newAgentId);
|
|
expect(newAssigneeRun?.status).toBe("queued");
|
|
|
|
const oldRetry = await db
|
|
.select({
|
|
status: heartbeatRuns.status,
|
|
errorCode: heartbeatRuns.errorCode,
|
|
})
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, scheduled.run.id))
|
|
.then((rows) => rows[0] ?? null);
|
|
expect(oldRetry).toEqual({
|
|
status: "cancelled",
|
|
errorCode: "issue_reassigned",
|
|
});
|
|
|
|
const deferredWakeups = await db
|
|
.select({ count: sql<number>`count(*)::int` })
|
|
.from(agentWakeupRequests)
|
|
.where(eq(agentWakeupRequests.status, "deferred_issue_execution"))
|
|
.then((rows) => rows[0]?.count ?? 0);
|
|
expect(deferredWakeups).toBe(0);
|
|
});
|
|
|
|
it("does not promote a scheduled retry after issue ownership changes", async () => {
|
|
const companyId = randomUUID();
|
|
const oldAgentId = randomUUID();
|
|
const newAgentId = randomUUID();
|
|
const issueId = randomUUID();
|
|
const sourceRunId = randomUUID();
|
|
const now = new Date("2026-04-20T14:00:00.000Z");
|
|
|
|
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: oldAgentId,
|
|
companyId,
|
|
name: "ClaudeCoder",
|
|
role: "engineer",
|
|
status: "active",
|
|
adapterType: "claude_local",
|
|
adapterConfig: {},
|
|
runtimeConfig: {
|
|
heartbeat: {
|
|
wakeOnDemand: true,
|
|
maxConcurrentRuns: 1,
|
|
},
|
|
},
|
|
permissions: {},
|
|
},
|
|
{
|
|
id: newAgentId,
|
|
companyId,
|
|
name: "CodexCoder",
|
|
role: "engineer",
|
|
status: "active",
|
|
adapterType: "codex_local",
|
|
adapterConfig: {},
|
|
runtimeConfig: {
|
|
heartbeat: {
|
|
wakeOnDemand: true,
|
|
maxConcurrentRuns: 1,
|
|
},
|
|
},
|
|
permissions: {},
|
|
},
|
|
]);
|
|
|
|
await db.insert(heartbeatRuns).values({
|
|
id: sourceRunId,
|
|
companyId,
|
|
agentId: oldAgentId,
|
|
invocationSource: "assignment",
|
|
triggerDetail: "system",
|
|
status: "failed",
|
|
error: "upstream overload",
|
|
errorCode: "adapter_failed",
|
|
finishedAt: now,
|
|
contextSnapshot: {
|
|
issueId,
|
|
wakeReason: "issue_assigned",
|
|
},
|
|
updatedAt: now,
|
|
createdAt: now,
|
|
});
|
|
|
|
await db.insert(issues).values({
|
|
id: issueId,
|
|
companyId,
|
|
title: "Retry promotion reassignment",
|
|
status: "todo",
|
|
priority: "medium",
|
|
assigneeAgentId: oldAgentId,
|
|
executionRunId: sourceRunId,
|
|
executionAgentNameKey: "claudecoder",
|
|
executionLockedAt: now,
|
|
issueNumber: 1,
|
|
identifier: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}-2`,
|
|
});
|
|
|
|
const scheduled = await heartbeat.scheduleBoundedRetry(sourceRunId, {
|
|
now,
|
|
random: () => 0.5,
|
|
});
|
|
expect(scheduled.outcome).toBe("scheduled");
|
|
if (scheduled.outcome !== "scheduled") return;
|
|
|
|
await db.update(issues).set({
|
|
assigneeAgentId: newAgentId,
|
|
updatedAt: now,
|
|
}).where(eq(issues.id, issueId));
|
|
|
|
const promotion = await heartbeat.promoteDueScheduledRetries(scheduled.dueAt);
|
|
expect(promotion).toEqual({ promoted: 0, runIds: [] });
|
|
|
|
const oldRetry = await db
|
|
.select({
|
|
status: heartbeatRuns.status,
|
|
errorCode: heartbeatRuns.errorCode,
|
|
})
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, scheduled.run.id))
|
|
.then((rows) => rows[0] ?? null);
|
|
expect(oldRetry).toEqual({
|
|
status: "cancelled",
|
|
errorCode: "issue_reassigned",
|
|
});
|
|
|
|
const issue = await db
|
|
.select({ executionRunId: issues.executionRunId })
|
|
.from(issues)
|
|
.where(eq(issues.id, issueId))
|
|
.then((rows) => rows[0] ?? null);
|
|
expect(issue?.executionRunId).toBeNull();
|
|
});
|
|
|
|
it("does not promote a scheduled retry after the issue is cancelled", async () => {
|
|
const companyId = randomUUID();
|
|
const agentId = randomUUID();
|
|
const issueId = randomUUID();
|
|
const sourceRunId = randomUUID();
|
|
const now = new Date("2026-04-20T15:00:00.000Z");
|
|
|
|
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: agentId,
|
|
companyId,
|
|
name: "CodexCoder",
|
|
role: "engineer",
|
|
status: "active",
|
|
adapterType: "codex_local",
|
|
adapterConfig: {},
|
|
runtimeConfig: {
|
|
heartbeat: {
|
|
wakeOnDemand: true,
|
|
maxConcurrentRuns: 1,
|
|
},
|
|
},
|
|
permissions: {},
|
|
});
|
|
|
|
await db.insert(heartbeatRuns).values({
|
|
id: sourceRunId,
|
|
companyId,
|
|
agentId,
|
|
invocationSource: "assignment",
|
|
triggerDetail: "system",
|
|
status: "failed",
|
|
error: "upstream overload",
|
|
errorCode: "adapter_failed",
|
|
finishedAt: now,
|
|
contextSnapshot: {
|
|
issueId,
|
|
wakeReason: "issue_assigned",
|
|
},
|
|
updatedAt: now,
|
|
createdAt: now,
|
|
});
|
|
|
|
await db.insert(issues).values({
|
|
id: issueId,
|
|
companyId,
|
|
title: "Retry promotion cancellation",
|
|
status: "todo",
|
|
priority: "medium",
|
|
assigneeAgentId: agentId,
|
|
executionRunId: sourceRunId,
|
|
executionAgentNameKey: "codexcoder",
|
|
executionLockedAt: now,
|
|
issueNumber: 1,
|
|
identifier: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}-3`,
|
|
});
|
|
|
|
const scheduled = await heartbeat.scheduleBoundedRetry(sourceRunId, {
|
|
now,
|
|
random: () => 0.5,
|
|
});
|
|
expect(scheduled.outcome).toBe("scheduled");
|
|
if (scheduled.outcome !== "scheduled") return;
|
|
|
|
await db.update(issues).set({
|
|
status: "cancelled",
|
|
updatedAt: now,
|
|
}).where(eq(issues.id, issueId));
|
|
|
|
const promotion = await heartbeat.promoteDueScheduledRetries(scheduled.dueAt);
|
|
expect(promotion).toEqual({ promoted: 0, runIds: [] });
|
|
|
|
const oldRetry = await db
|
|
.select({
|
|
status: heartbeatRuns.status,
|
|
errorCode: heartbeatRuns.errorCode,
|
|
})
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, scheduled.run.id))
|
|
.then((rows) => rows[0] ?? null);
|
|
expect(oldRetry).toEqual({
|
|
status: "cancelled",
|
|
errorCode: "issue_cancelled",
|
|
});
|
|
|
|
const issue = await db
|
|
.select({ executionRunId: issues.executionRunId })
|
|
.from(issues)
|
|
.where(eq(issues.id, issueId))
|
|
.then((rows) => rows[0] ?? null);
|
|
expect(issue?.executionRunId).toBeNull();
|
|
});
|
|
|
|
it("exhausts bounded retries after the hard cap", async () => {
|
|
const companyId = randomUUID();
|
|
const agentId = randomUUID();
|
|
const cappedRunId = randomUUID();
|
|
const now = new Date("2026-04-20T18:00:00.000Z");
|
|
|
|
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: agentId,
|
|
companyId,
|
|
name: "CodexCoder",
|
|
role: "engineer",
|
|
status: "active",
|
|
adapterType: "codex_local",
|
|
adapterConfig: {},
|
|
runtimeConfig: {
|
|
heartbeat: {
|
|
wakeOnDemand: true,
|
|
maxConcurrentRuns: 1,
|
|
},
|
|
},
|
|
permissions: {},
|
|
});
|
|
|
|
await db.insert(heartbeatRuns).values({
|
|
id: cappedRunId,
|
|
companyId,
|
|
agentId,
|
|
invocationSource: "automation",
|
|
status: "failed",
|
|
error: "still transient",
|
|
errorCode: "adapter_failed",
|
|
finishedAt: now,
|
|
scheduledRetryAttempt: BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS.length,
|
|
scheduledRetryReason: "transient_failure",
|
|
contextSnapshot: {
|
|
wakeReason: "transient_failure_retry",
|
|
},
|
|
updatedAt: now,
|
|
createdAt: now,
|
|
});
|
|
|
|
const exhausted = await heartbeat.scheduleBoundedRetry(cappedRunId, {
|
|
now,
|
|
random: () => 0.5,
|
|
});
|
|
|
|
expect(exhausted).toEqual({
|
|
outcome: "retry_exhausted",
|
|
attempt: BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS.length + 1,
|
|
maxAttempts: BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS.length,
|
|
});
|
|
|
|
const runCount = await db
|
|
.select({ count: sql<number>`count(*)::int` })
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.companyId, companyId))
|
|
.then((rows) => rows[0]?.count ?? 0);
|
|
expect(runCount).toBe(1);
|
|
|
|
const exhaustionEvent = await db
|
|
.select({
|
|
message: heartbeatRunEvents.message,
|
|
payload: heartbeatRunEvents.payload,
|
|
})
|
|
.from(heartbeatRunEvents)
|
|
.where(eq(heartbeatRunEvents.runId, cappedRunId))
|
|
.orderBy(sql`${heartbeatRunEvents.id} desc`)
|
|
.then((rows) => rows[0] ?? null);
|
|
|
|
expect(exhaustionEvent?.message).toContain("Bounded retry exhausted");
|
|
expect(exhaustionEvent?.payload).toMatchObject({
|
|
retryReason: "transient_failure",
|
|
scheduledRetryAttempt: BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS.length,
|
|
maxAttempts: BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS.length,
|
|
});
|
|
});
|
|
|
|
it("advances codex transient fallback stages across bounded retry attempts", async () => {
|
|
const fallbackModes = [
|
|
"same_session",
|
|
"safer_invocation",
|
|
"fresh_session",
|
|
"fresh_session_safer_invocation",
|
|
] as const;
|
|
|
|
for (const [index, expectedMode] of fallbackModes.entries()) {
|
|
const companyId = randomUUID();
|
|
const agentId = randomUUID();
|
|
const runId = randomUUID();
|
|
const now = new Date(`2026-04-20T1${index}:00:00.000Z`);
|
|
|
|
await seedRetryFixture({
|
|
runId,
|
|
companyId,
|
|
agentId,
|
|
now,
|
|
errorCode: "adapter_failed",
|
|
errorFamily: "transient_upstream",
|
|
scheduledRetryAttempt: index,
|
|
});
|
|
|
|
const scheduled = await heartbeat.scheduleBoundedRetry(runId, {
|
|
now,
|
|
random: () => 0.5,
|
|
});
|
|
|
|
expect(scheduled.outcome).toBe("scheduled");
|
|
if (scheduled.outcome !== "scheduled") continue;
|
|
|
|
const retryRun = await db
|
|
.select({
|
|
contextSnapshot: heartbeatRuns.contextSnapshot,
|
|
wakeupRequestId: heartbeatRuns.wakeupRequestId,
|
|
})
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, scheduled.run.id))
|
|
.then((rows) => rows[0] ?? null);
|
|
expect((retryRun?.contextSnapshot as Record<string, unknown> | null)?.codexTransientFallbackMode).toBe(expectedMode);
|
|
|
|
const wakeupRequest = await db
|
|
.select({ payload: agentWakeupRequests.payload })
|
|
.from(agentWakeupRequests)
|
|
.where(eq(agentWakeupRequests.id, retryRun?.wakeupRequestId ?? ""))
|
|
.then((rows) => rows[0] ?? null);
|
|
expect((wakeupRequest?.payload as Record<string, unknown> | null)?.codexTransientFallbackMode).toBe(expectedMode);
|
|
|
|
await db.delete(heartbeatRunEvents);
|
|
await db.delete(heartbeatRuns);
|
|
await db.delete(agentWakeupRequests);
|
|
await db.delete(agents);
|
|
await db.delete(companies);
|
|
}
|
|
});
|
|
|
|
it("honors codex retry-not-before timestamps when they exceed the default bounded backoff", async () => {
|
|
const companyId = randomUUID();
|
|
const agentId = randomUUID();
|
|
const runId = randomUUID();
|
|
const now = new Date(2026, 3, 22, 22, 29, 0);
|
|
const retryNotBefore = new Date(2026, 3, 22, 23, 31, 0);
|
|
|
|
await seedRetryFixture({
|
|
runId,
|
|
companyId,
|
|
agentId,
|
|
now,
|
|
errorCode: "adapter_failed",
|
|
errorFamily: "transient_upstream",
|
|
retryNotBefore: retryNotBefore.toISOString(),
|
|
});
|
|
|
|
const scheduled = await heartbeat.scheduleBoundedRetry(runId, {
|
|
now,
|
|
random: () => 0.5,
|
|
});
|
|
|
|
expect(scheduled.outcome).toBe("scheduled");
|
|
if (scheduled.outcome !== "scheduled") return;
|
|
expect(scheduled.dueAt.getTime()).toBe(retryNotBefore.getTime());
|
|
|
|
const retryRun = await db
|
|
.select({
|
|
contextSnapshot: heartbeatRuns.contextSnapshot,
|
|
scheduledRetryAt: heartbeatRuns.scheduledRetryAt,
|
|
wakeupRequestId: heartbeatRuns.wakeupRequestId,
|
|
})
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, scheduled.run.id))
|
|
.then((rows) => rows[0] ?? null);
|
|
|
|
expect(retryRun?.scheduledRetryAt?.getTime()).toBe(retryNotBefore.getTime());
|
|
expect((retryRun?.contextSnapshot as Record<string, unknown> | null)?.transientRetryNotBefore).toBe(
|
|
retryNotBefore.toISOString(),
|
|
);
|
|
|
|
const wakeupRequest = await db
|
|
.select({ payload: agentWakeupRequests.payload })
|
|
.from(agentWakeupRequests)
|
|
.where(eq(agentWakeupRequests.id, retryRun?.wakeupRequestId ?? ""))
|
|
.then((rows) => rows[0] ?? null);
|
|
|
|
expect((wakeupRequest?.payload as Record<string, unknown> | null)?.transientRetryNotBefore).toBe(
|
|
retryNotBefore.toISOString(),
|
|
);
|
|
});
|
|
|
|
it("schedules bounded retries for claude_transient_upstream and honors its retry-not-before hint", async () => {
|
|
const companyId = randomUUID();
|
|
const agentId = randomUUID();
|
|
const runId = randomUUID();
|
|
const now = new Date(2026, 3, 22, 10, 0, 0);
|
|
const retryNotBefore = new Date(2026, 3, 22, 16, 0, 0);
|
|
|
|
await seedRetryFixture({
|
|
runId,
|
|
companyId,
|
|
agentId,
|
|
now,
|
|
errorCode: "adapter_failed",
|
|
errorFamily: "transient_upstream",
|
|
adapterType: "claude_local",
|
|
retryNotBefore: retryNotBefore.toISOString(),
|
|
});
|
|
|
|
const scheduled = await heartbeat.scheduleBoundedRetry(runId, {
|
|
now,
|
|
random: () => 0.5,
|
|
});
|
|
|
|
expect(scheduled.outcome).toBe("scheduled");
|
|
if (scheduled.outcome !== "scheduled") return;
|
|
expect(scheduled.dueAt.getTime()).toBe(retryNotBefore.getTime());
|
|
|
|
const retryRun = await db
|
|
.select({
|
|
contextSnapshot: heartbeatRuns.contextSnapshot,
|
|
scheduledRetryAt: heartbeatRuns.scheduledRetryAt,
|
|
wakeupRequestId: heartbeatRuns.wakeupRequestId,
|
|
})
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, scheduled.run.id))
|
|
.then((rows) => rows[0] ?? null);
|
|
|
|
expect(retryRun?.scheduledRetryAt?.getTime()).toBe(retryNotBefore.getTime());
|
|
const contextSnapshot = (retryRun?.contextSnapshot as Record<string, unknown> | null) ?? {};
|
|
expect(contextSnapshot.transientRetryNotBefore).toBe(retryNotBefore.toISOString());
|
|
// Claude does not participate in the Codex fallback-mode ladder.
|
|
expect(contextSnapshot.codexTransientFallbackMode ?? null).toBeNull();
|
|
|
|
const wakeupRequest = await db
|
|
.select({ payload: agentWakeupRequests.payload })
|
|
.from(agentWakeupRequests)
|
|
.where(eq(agentWakeupRequests.id, retryRun?.wakeupRequestId ?? ""))
|
|
.then((rows) => rows[0] ?? null);
|
|
|
|
expect((wakeupRequest?.payload as Record<string, unknown> | null)?.transientRetryNotBefore).toBe(
|
|
retryNotBefore.toISOString(),
|
|
);
|
|
});
|
|
});
|