refactor(server): extract heartbeat retry scheduling (#15617)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Heartbeat orchestration starts runs and recovers from supported
failures.
> - The heartbeat service still contains more than 22,000 lines after
earlier extractions.
> - Retry scheduling has a clear boundary, but it shares lifecycle
callbacks with the service.
> - This pull request moves retry scheduling into one module with
explicit dependencies.
> - It preserves the scheduling rules, database writes, and lock order.
> - The benefit is a smaller service and focused tests for the retry
boundary.

## Linked Issues or Issue Description

**Current behavior**

`server/src/services/heartbeat.ts` owns retry schedules, contention
deferrals, due promotion, and retry-now requests. These operations sit
among execution and recovery code.

**Proposed behavior**

Move these operations into `server/src/services/heartbeat/retries.ts`.
Keep the public heartbeat API and existing helper exports. The service
supplies its lifecycle callbacks and worktree cutoff.

**Reason and benefit**

Extract one complete area in a separate PR. This removes 1,641 lines
from the main service without changing retry policy. Future retry
changes can use a smaller module and direct tests.

**Breaking changes**

None. Existing imports, error-class identity, retry metadata, and API
outcomes remain available.

**Additional context**

Refs #15601, the preceding run-state extraction. Related open retry
changes: #14930, #13975, #15157, #12587, and #13276. This PR moves the
existing implementation. Those policy changes remain separate.

## What Changed

- Add `createHeartbeatRetries(db, dependencies)` for bounded retries,
workspace and connection deferrals, shared-workspace holder checks, due
promotion, and retry-now requests.
- Keep status transitions, event publication, issue-lock release,
plan-resume reporting, and worktree settings in the heartbeat service.
- Preserve all 140 existing public exports. Re-export the same helpers,
constants, and workspace-busy error class.
- Add 14 tests for construction, callback order, cancellation races,
concurrent scheduling, committed promotion, and retry-now scope.
- Document the module boundary in `doc/DEVELOPING.md`.

## Verification

- Existing focused suite before the move: 84 tests passed.
- Expanded focused suite after the move: 243 tests passed across 11
files. Command:
`pnpm exec vitest run server/src/services/heartbeat/retries.test.ts
server/src/__tests__/heartbeat-retry-scheduling.test.ts
server/src/__tests__/heartbeat-workspace-busy.test.ts
server/src/__tests__/heartbeat-model-fallback-warning.test.ts
server/src/__tests__/heartbeat-running-followup.test.ts
server/src/__tests__/heartbeat-issue-rewake-throttle.test.ts
server/src/__tests__/issue-scheduled-retry-routes.test.ts
server/src/modules/run-dispatch`
- After rebasing onto the latest master, 61 retry and plugin-idle tests
passed across four files.
- `pnpm -r typecheck` passed again on the rebased branch.
- `pnpm build` passed again on the rebased branch.
- `pnpm test:run` was started locally. The long serial run was stopped
after full CI passed; it is not claimed as a complete local pass.
- Full CI passed on commit `69e76f3eec75b0703570f1fce37ebfe9c63fa3ac`:
54 successful checks, two intentional Storybook skips, zero failures.
This includes general tests, serialized server tests, browser tests,
runner checks, typecheck, build, and canary verification.
- The setup-timeout review fix passed all 14 module tests and the server
typecheck.
- Greptile completed a fresh review of commit
`69e76f3eec75b0703570f1fce37ebfe9c63fa3ac`: 5/5, no open findings.
- A structural comparison confirmed 22 unchanged moved functions and one
unchanged three-line string reader, 16 unchanged declarations, one
unchanged API body, and all 140 preserved exports. The remaining service
body is unchanged after accounting for the new binding and removed
declarations.

## Risks

- A missing lifecycle callback or an eager call during construction
could change retry behavior. The dependency contract and direct tests
cover these boundaries.
- Concurrent scheduling must keep the issue-first lock order and reuse
one successor. The original transaction bodies are unchanged. PostgreSQL
tests cover concurrent calls from separate factory instances.
- Promotion must publish after commit and preserve worktree cutoffs.
Direct tests cover both paths.
- No schema, route, adapter, or retry-policy changes are included.

## Model Used

OpenAI GPT-6 via Codex. The exact serving model ID and context window
were not exposed in this session. Used reasoning, repository inspection,
shell tools, and code execution.

## 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 searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-10-08 15:38:48 -05:00
1 parent ed6abbf158
commit 2a5f65c9ae
4 files changed
+2207 -1700

No files matched your search

+10 -1
View File
@@ -1241,7 +1241,7 @@ issue and wake context, responsible-user resolution, routine environment snapsho
skill mentions, adapter environment configuration, and MCP/tool access setup.
`createHeartbeatRunPreparation(db)` binds the context loaders to a service's database
without doing database work during construction. `heartbeat.ts` keeps queueing,
dispatch, retries, cancellation, and execution order, and re-exports the existing
dispatch, cancellation, and execution order, and re-exports the existing
public helpers and configuration-incomplete error class. Keep preparation policy
changes in this module and its tests.
@@ -1254,6 +1254,15 @@ during construction. The encoding-check cache belongs to each factory instance.
and status transitions, and re-exports the existing public helpers. Keep session
policy changes separate from run orchestration changes.
Retry scheduling is in `server/src/services/heartbeat/retries.ts`. It owns bounded
retry schedules, connection and workspace contention deferrals, shared-workspace
holder checks, due retry promotion, and retry-now requests. `createHeartbeatRetries`
binds these operations to the service database and explicit lifecycle callbacks
without doing work during construction. `heartbeat.ts` supplies status writes,
run events, issue-lock release, plan-resume reporting, and worktree cutoffs. It
re-exports the existing retry helpers and workspace-busy error class. Keep retry
policy changes separate from this extraction.
## Wake Context Delivery
Built-in adapters deliver wake context through the run prompt, including structured
File diff suppressed because it is too large. Load diff
@@ -0,0 +1,286 @@
import { randomUUID } from "node:crypto";
import type { Db } from "@paperclipai/db";
import { agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRunEvents, heartbeatRuns, issues } from "@paperclipai/db";
import { and, eq } from "drizzle-orm";
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
import { HttpError } from "../../errors.js";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
import { createRunDispatch, type PostCommitEffect } from "../../modules/run-dispatch/index.js";
import { evaluateAgentInvokabilityFromDb } from "../agent-invokability.js";
import * as legacy from "../heartbeat.js";
import { createHeartbeatRunState } from "./run-state.js";
import * as extracted from "./retries.js";
import type { HeartbeatRetryDependencies } from "./retries.js";
type Run = typeof heartbeatRuns.$inferSelect;
type Agent = typeof agents.$inferSelect;
function callbacks() {
return {
getRun: vi.fn<HeartbeatRetryDependencies["getRun"]>(async () => null),
getAgent: vi.fn<HeartbeatRetryDependencies["getAgent"]>(async () => undefined),
resolveSessionBeforeForWakeup: vi.fn(async () => null),
resolveResponsibleUserIdForRunContext: vi.fn(async () => "retry-user"),
getAgentInvokability: vi.fn<HeartbeatRetryDependencies["getAgentInvokability"]>(async () => ({ invokable: true })),
appendRunEvent: vi.fn<HeartbeatRetryDependencies["appendRunEvent"]>(async () => undefined),
escalatePlanApprovalResumeFailureNeedsAttention: vi.fn(async () => null),
recordPlanApprovalResumeFailureRetry: vi.fn(async () => null),
setRunStatusIfRunning: vi.fn<HeartbeatRetryDependencies["setRunStatusIfRunning"]>(async () => ({ updated: false, run: null })),
setWakeupStatus: vi.fn(async () => undefined),
releaseIssueExecutionAndPromote: vi.fn(async () => undefined),
finalizeAgentStatus: vi.fn(async () => undefined),
getWorktreeExecutionCutoff: vi.fn(async (): Promise<Date | null> => null),
applyRunDispatchPostCommitEffects: vi.fn<(effects: PostCommitEffect[]) => void>(),
runDispatch: {
evaluateScheduledRetryGate: vi.fn<HeartbeatRetryDependencies["runDispatch"]["evaluateScheduledRetryGate"]>(async () => ({ allowed: true })),
promoteDueScheduledRetries: vi.fn<HeartbeatRetryDependencies["runDispatch"]["promoteDueScheduledRetries"]>(async () => ({ promoted: 0, runIds: [], postCommitEffects: [] })),
promoteScheduledRetry: vi.fn<HeartbeatRetryDependencies["runDispatch"]["promoteScheduledRetry"]>(async () => ({ outcome: "not_promoted" })),
},
} satisfies HeartbeatRetryDependencies;
}
// These tests must never query: only the injected lifecycle callbacks may run.
function guardedDatabase() {
const access = vi.fn(() => { throw new Error("Unexpected retry database access"); });
return { db: new Proxy({}, { get: access }) as Db, access };
}
function unitRun(): Run {
const run: Partial<Run> = {
id: "run", companyId: "company", agentId: "agent", status: "running",
scopeKind: "issue", issueId: "issue",
wakeupRequestId: "wake", scheduledRetryAttempt: 0,
contextSnapshot: { issueId: "issue", timerClaimWasFirstHeartbeat: true },
};
return run as Run;
}
function deferral() {
return new extracted.WorkspaceBusyDeferral({
holder: { runId: "holder", agentId: "other-agent", issueId: "other-issue", issueIdentifier: "T-2" },
projectWorkspaceId: "workspace", deferralAttempt: 0, wasIssueAssignee: true,
});
}
describe("heartbeat retry module callbacks", () => {
it("preserves legacy helper and error-class identity", () => {
expect(legacy.computeBoundedTransientHeartbeatRetrySchedule).toBe(extracted.computeBoundedTransientHeartbeatRetrySchedule);
expect(legacy.computeWorkspaceBusyRetryDelayMs).toBe(extracted.computeWorkspaceBusyRetryDelayMs);
expect(legacy.BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS).toBe(extracted.BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS);
expect(legacy.WorkspaceBusyDeferral).toBe(extracted.WorkspaceBusyDeferral);
expect(deferral()).toBeInstanceOf(legacy.WorkspaceBusyDeferral);
});
it("constructs without database access or lifecycle effects", () => {
const database = guardedDatabase();
const deps = callbacks();
extracted.createHeartbeatRetries(database.db, deps);
expect(database.access).not.toHaveBeenCalled();
for (const callback of Object.values(deps)) {
if (typeof callback === "function") expect(callback).not.toHaveBeenCalled();
}
for (const callback of Object.values(deps.runDispatch)) expect(callback).not.toHaveBeenCalled();
});
it.each(["missing_run", "missing_agent"] as const)("keeps %s lookup behavior and requests the full source result", async (outcome) => {
const database = guardedDatabase();
const deps = callbacks();
if (outcome === "missing_agent") deps.getRun.mockResolvedValue(unitRun());
expect(await extracted.createHeartbeatRetries(database.db, deps).scheduleBoundedRetry("run")).toEqual({ outcome });
expect(deps.getRun).toHaveBeenCalledWith("run", { unsafeFullResultJson: true });
expect(deps.getAgent).toHaveBeenCalledTimes(outcome === "missing_agent" ? 1 : 0);
expect(database.access).not.toHaveBeenCalled();
});
it("records exhaustion before escalating a plan-resume failure", async () => {
const database = guardedDatabase();
const deps = callbacks();
const run = unitRun();
const retry = await extracted.createHeartbeatRetries(database.db, deps).scheduleBoundedRetryForRun(
run, { adapterType: "codex_local" } as Agent,
{ retryReason: legacy.INTERACTION_CONTINUATION_INFRA_RETRY_REASON, maxAttempts: 0 },
);
expect(retry).toMatchObject({ outcome: "retry_exhausted", attempt: 1, maxAttempts: 0 });
expect(deps.appendRunEvent).toHaveBeenCalledWith(run, expect.objectContaining({
retryExhaustion: { retryReason: legacy.INTERACTION_CONTINUATION_INFRA_RETRY_REASON, scheduledRetryAttempt: 0, maxAttempts: 0 },
}));
expect(deps.escalatePlanApprovalResumeFailureNeedsAttention).toHaveBeenCalledAfter(deps.appendRunEvent);
expect(database.access).not.toHaveBeenCalled();
});
it("passes the service's worktree cutoff and publishes only returned effects", async () => {
const database = guardedDatabase();
const deps = callbacks();
const now = new Date("2026-10-08T12:00:00Z");
const cutoff = new Date("2026-10-08T11:00:00Z");
const effect: PostCommitEffect = {
kind: "run_queued", runId: "retry", companyId: "company", agentId: "agent",
invocationSource: "automation", triggerDetail: "system", wakeupRequestId: "wake",
};
deps.getWorktreeExecutionCutoff.mockResolvedValue(cutoff);
deps.runDispatch.promoteDueScheduledRetries.mockResolvedValue({ promoted: 1, runIds: ["retry"], postCommitEffects: [effect] });
expect(await extracted.createHeartbeatRetries(database.db, deps).promoteDueScheduledRetries(now)).toEqual({ promoted: 1, runIds: ["retry"] });
expect(deps.runDispatch.promoteDueScheduledRetries).toHaveBeenCalledWith({ now, cutoff });
expect(deps.applyRunDispatchPostCommitEffects).toHaveBeenCalledWith([effect]);
expect(deps.applyRunDispatchPostCommitEffects).toHaveBeenCalledAfter(deps.runDispatch.promoteDueScheduledRetries);
expect(database.access).not.toHaveBeenCalled();
});
it("does not publish effects when promotion rejects", async () => {
const deps = callbacks();
deps.runDispatch.promoteDueScheduledRetries.mockRejectedValue(new Error("promotion failed"));
await expect(extracted.createHeartbeatRetries(guardedDatabase().db, deps).promoteDueScheduledRetries()).rejects.toThrow("promotion failed");
expect(deps.applyRunDispatchPostCommitEffects).not.toHaveBeenCalled();
});
it.each(["workspace", "connection"] as const)("leaves a %s cancellation race winner alone", async (kind) => {
const database = guardedDatabase();
const deps = callbacks();
const retries = extracted.createHeartbeatRetries(database.db, deps);
if (kind === "workspace") await retries.finalizeWorkspaceBusyDeferral(unitRun(), deferral());
else await retries.finalizeAiConnectionBusyDeferral(unitRun(), new HttpError(409, "busy"), true);
expect(deps.setRunStatusIfRunning).toHaveBeenCalledTimes(1);
expect(deps.setWakeupStatus).not.toHaveBeenCalled();
expect(deps.getAgent).not.toHaveBeenCalled();
expect(deps.releaseIssueExecutionAndPromote).not.toHaveBeenCalled();
expect(deps.finalizeAgentStatus).not.toHaveBeenCalled();
expect(database.access).not.toHaveBeenCalled();
});
it.each(["workspace", "connection"] as const)("settles the agent after a %s deferral cannot keep its issue lock", async (kind) => {
const database = guardedDatabase();
const deps = callbacks();
const run = { ...unitRun(), status: "cancelled" };
deps.setRunStatusIfRunning.mockResolvedValue({ updated: true, run });
deps.releaseIssueExecutionAndPromote.mockRejectedValue(new Error("release failed"));
const retries = extracted.createHeartbeatRetries(database.db, deps);
if (kind === "workspace") await retries.finalizeWorkspaceBusyDeferral(run, deferral());
else await expect(retries.finalizeAiConnectionBusyDeferral(run, new HttpError(409, "busy"), true)).rejects.toThrow("release failed");
expect(deps.setWakeupStatus).toHaveBeenCalledWith("wake", "cancelled", expect.any(Object));
expect(deps.releaseIssueExecutionAndPromote).toHaveBeenCalledWith(run);
expect(deps.finalizeAgentStatus).toHaveBeenCalledWith("agent", "cancelled", null, { wasFirstHeartbeat: true });
expect(deps.finalizeAgentStatus).toHaveBeenCalledAfter(deps.releaseIssueExecutionAndPromote);
expect(database.access).not.toHaveBeenCalled();
});
});
const support = await getEmbeddedPostgresTestSupport();
const describePostgres = support.supported ? describe : describe.skip;
if (!support.supported) console.warn(`Skipping retry PostgreSQL tests: ${support.reason ?? "unsupported host"}`);
describePostgres("heartbeat retry module database wiring", () => {
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
let db: Db;
const now = new Date("2026-10-08T12:00:00Z");
beforeAll(async () => {
database = await startEmbeddedPostgresTestDatabase("heartbeat-retries-");
db = createDb(database.connectionString);
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
afterAll(async () => { await database?.cleanup(); });
afterEach(async () => {
await db.delete(issues);
await db.delete(heartbeatRunEvents);
await db.delete(heartbeatRuns);
await db.delete(agentWakeupRequests);
await db.delete(agents);
await db.delete(companies);
});
function boundRetries() {
const state = createHeartbeatRunState(db);
const deps = {
...callbacks(), getRun: state.getRun, getAgent: state.getAgent,
resolveSessionBeforeForWakeup: state.resolveSessionBeforeForWakeup,
getAgentInvokability: (agent: Agent | null | undefined) => evaluateAgentInvokabilityFromDb(db, agent),
runDispatch: createRunDispatch(db),
};
return { retries: extracted.createHeartbeatRetries(db, deps), deps };
}
async function fixture() {
const companyId = randomUUID();
await db.insert(companies).values({ id: companyId, name: "Retries", issuePrefix: `R${companyId.slice(0, 8)}` });
const [agent] = await db.insert(agents).values({ companyId, name: "Retry agent", adapterType: "codex_local", status: "active" }).returning();
const [issue] = await db.insert(issues).values({ companyId, title: "Retry task", status: "in_progress", assigneeAgentId: agent.id }).returning();
const [run] = await db.insert(heartbeatRuns).values({
companyId, agentId: agent.id, scopeKind: "issue", issueId: issue.id,
status: "failed", errorCode: "codex_transient_upstream", finishedAt: now,
resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } },
contextSnapshot: { issueId: issue.id },
}).returning();
await db.update(issues).set({ executionRunId: run.id, checkoutRunId: run.id }).where(eq(issues.id, issue.id));
return { companyId, agent, issue, run };
}
it("reuses one successor across concurrent factories and keeps writes committed before callbacks", async () => {
const { run, agent, issue, companyId } = await fixture();
const first = boundRetries(), second = boundRetries();
for (const binding of [first, second]) {
binding.deps.appendRunEvent.mockImplementation(async (_source, event) => {
const retryId = event.payload?.retryRunId;
expect(typeof retryId).toBe("string");
expect(await binding.deps.getRun(retryId as string)).toMatchObject({ retryOfRunId: run.id });
});
}
const results = await Promise.all([
first.retries.scheduleBoundedRetryForRun(run, agent, { now, random: () => 0 }),
second.retries.scheduleBoundedRetryForRun(run, agent, { now, random: () => 0 }),
]);
expect(results.every((result) => result.outcome === "scheduled")).toBe(true);
const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.retryOfRunId, run.id)));
expect(successors).toHaveLength(1);
const successor = successors[0];
expect(successor).toMatchObject({
status: "scheduled_retry", issueId: issue.id, responsibleUserId: "retry-user",
scheduledRetryAttempt: 1, scheduledRetryReason: "transient_failure",
});
expect(successor.contextSnapshot).toMatchObject({ retryOfRunId: run.id, codexTransientFallbackMode: "same_session" });
expect(first.deps.resolveResponsibleUserIdForRunContext).toHaveBeenCalledWith(run, expect.objectContaining({ retryOfRunId: run.id }));
const [locked] = await db.select().from(issues).where(eq(issues.id, issue.id));
expect(locked).toMatchObject({ executionRunId: successor.id, checkoutRunId: null });
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.runId, successor.id))).toHaveLength(1);
expect(await boundRetries().retries.scheduleBoundedRetry(run.id, { now })).toMatchObject({ outcome: "scheduled", run: { id: successor.id }, reusedExisting: true });
expect(first.deps.appendRunEvent).toHaveBeenCalledTimes(1);
expect(second.deps.appendRunEvent).toHaveBeenCalledTimes(1);
});
it("promotes only due retries and publishes after commit", async () => {
const { run } = await fixture();
const { retries, deps } = boundRetries();
const scheduled = await retries.scheduleBoundedRetry(run.id, { now, delayMs: 60_000 });
expect(scheduled.outcome).toBe("scheduled");
if (scheduled.outcome !== "scheduled") throw new Error("Expected retry");
expect(await retries.promoteDueScheduledRetries(now)).toEqual({ promoted: 0, runIds: [] });
deps.applyRunDispatchPostCommitEffects.mockClear();
const committedStatuses: string[] = [];
const reads: Promise<void>[] = [];
deps.applyRunDispatchPostCommitEffects.mockImplementation(() => {
// Read outside the dispatch transaction after the callback is invoked.
// The assertion below waits for that read without making publication async.
reads.push(deps.getRun(scheduled.run.id).then((row) => { committedStatuses.push(row!.status); }));
});
expect(await retries.promoteDueScheduledRetries(scheduled.dueAt)).toEqual({ promoted: 1, runIds: [scheduled.run.id] });
await Promise.all(reads);
expect(committedStatuses).toEqual(["queued"]);
expect(deps.applyRunDispatchPostCommitEffects).toHaveBeenCalledWith([expect.objectContaining({ kind: "run_queued", runId: scheduled.run.id })]);
});
it("binds retry-now to the requested issue and persists its actor before promotion", async () => {
const selected = await fixture(), other = await fixture();
const { retries, deps } = boundRetries();
const scheduled = await retries.scheduleBoundedRetry(selected.run.id, { now, delayMs: 60_000 });
const foreign = await retries.scheduleBoundedRetry(other.run.id, { now, delayMs: 60_000 });
if (scheduled.outcome !== "scheduled" || foreign.outcome !== "scheduled") throw new Error("Expected retries");
expect(await retries.retryScheduledRetryNow({ issueId: selected.issue.id, actor: { actorType: "user", actorId: "retry-user" }, now })).toMatchObject({
outcome: "promoted", scheduledRetry: { runId: scheduled.run.id, status: "queued" },
});
expect(await deps.getRun(scheduled.run.id)).toMatchObject({ contextSnapshot: {
retryNowRequestedByActorType: "user", retryNowRequestedByActorId: "retry-user", retryNowRequestedAt: now.toISOString(),
} });
expect(await deps.getRun(foreign.run.id)).toMatchObject({ status: "scheduled_retry", scheduledRetryAt: foreign.dueAt });
expect(await retries.retryScheduledRetryNow({ issueId: selected.issue.id, now })).toMatchObject({ outcome: "already_promoted" });
});
});
File diff suppressed because it is too large. Load diff