mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-01 15:35:12 +01:00
refactor(heartbeat): extract run lifecycle handling (#15672)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The heartbeat service admits, executes, and settles agent runs. > - Earlier extractions separated workspaces, preparation, state, retries, recovery, and queue admission. > - Run status, progress, liveness, and completion handling still live in the main service. > - These operations share status predicates, event writes, and completion policies. > - This pull request moves them into one lifecycle module and adds boundary tests. > - Reviewers can inspect lifecycle policy separately from adapter execution. ## Linked Issues or Issue Description **What existing behavior does this improve?** The structure and testability of heartbeat run lifecycle handling. **Subsystem affected** server/ — orchestration services. **Current behavior** Run status writes, progress, events, liveness, completion handoffs, issue-comment finalization, and runtime settlement live inside `heartbeat.ts`. **Proposed behavior** Keep these operations in `server/src/services/heartbeat/run-lifecycle.ts`. Bind the database and reporting, recovery, and wakeup callbacks with `createHeartbeatLifecycle`. **Reason and benefit** This removes 2,682 lines from `heartbeat.ts`, leaving 11,558 lines. The lifecycle module has 2,982 lines and keeps related policy together in one file. **Breaking changes** None. Existing function bodies, service methods, public exports, status conditions, and side effects keep their behavior. **Additional context** Continues #15626 and the earlier merged heartbeat extractions. A search of open and closed issues and PRs found no duplicate lifecycle extraction. This does not add a roadmap feature. ## What Changed - Move status transitions, run events and progress, liveness classification, completion handoffs, plan-resume failure reporting, issue-comment finalization, and runtime/cost settlement into `heartbeat/run-lifecycle.ts`. - Supply reporting, recovery, and admission callbacks through an explicit dependency interface. Preserve construction order with forwarding callbacks. - Keep adapter execution, cancellation, and terminal telemetry reporting in the main service. - Add 16 tests for inert construction, continuation gates, late progress, status races, native ownership, usage metadata, revoked wakes, event sequencing, and redaction. - Update the existing publication-order source assertion to read the extracted event writer. Keep its persist-before-publish check. - Document the lifecycle module boundary in `doc/DEVELOPING.md`. ## Verification - Before extraction: 87 tests passed across seven existing suites. - After extraction: 103 tests passed across eight files, including the 16 new module tests and real PostgreSQL coverage. ```sh pnpm exec vitest run server/src/services/heartbeat/run-lifecycle.test.ts server/src/__tests__/heartbeat-run-status-payload.test.ts server/src/__tests__/heartbeat-run-event-sequencing.test.ts server/src/__tests__/heartbeat-run-terminalize-before-release.test.ts server/src/__tests__/heartbeat-run-lease-release-terminalization.test.ts server/src/__tests__/heartbeat-cost-accounting.test.ts server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts server/src/__tests__/run-liveness.test.ts ``` - Additional recovery, cancellation, plan-resume, summary, and handoff coverage: **595 tests passed across 14 files**. The module suite appears in both runs. With the publication-order suite below, combined focused coverage is **694 distinct tests across 22 files**, with no skips. ```sh pnpm exec vitest run server/src/services/heartbeat/run-lifecycle.test.ts server/src/services/heartbeat/recovery.test.ts server/src/services/heartbeat/retries.test.ts server/src/services/heartbeat/queue.test.ts server/src/services/recovery/successful-run-handoff.test.ts server/src/services/recovery/review-path-recovery.test.ts server/src/services/heartbeat-run-runtime-status.test.ts server/src/__tests__/heartbeat-process-recovery.test.ts server/src/__tests__/heartbeat-native-runner-cancellation.test.ts server/src/__tests__/heartbeat-native-cleanup-admission.test.ts server/src/__tests__/heartbeat-accepted-plan-workspace-refresh.test.ts server/src/__tests__/heartbeat-runtime-state.test.ts server/src/__tests__/heartbeat-run-summary.test.ts server/src/__tests__/heartbeat-context-summary.test.ts ``` - The publication-order and lifecycle suites passed together: **28 tests across two files**. The first CI run exposed an assertion that still read the old event-writer source path. That assertion now reads the lifecycle module. ```sh pnpm exec vitest run server/src/services/chat-publication-reconciliation.test.ts server/src/services/heartbeat/run-lifecycle.test.ts ``` - After resolving the import conflict with the Pi Runner integration and rebasing onto `57e977be7`: **44 tests passed across four files**, including the updated source assertion, lifecycle module, run-status payloads, and accepted-plan workspace refresh. ```sh pnpm exec vitest run server/src/services/chat-publication-reconciliation.test.ts server/src/services/heartbeat/run-lifecycle.test.ts server/src/__tests__/heartbeat-run-status-payload.test.ts server/src/__tests__/heartbeat-accepted-plan-workspace-refresh.test.ts ``` - Full `pnpm -r typecheck` and `pnpm build` passed on final head `bca87df089c541450932f6b678ca776942d267b2`. Full local `pnpm test:run` was restarted on this head and stopped after the entire CI suite passed. It did not finish and is not counted as a full local pass. - The earlier full local run had one transient public MCP Cloud bootstrap failure. All **82 public MCP tests passed in isolation**. The first local run was stopped before the source-assertion fix and rebase. - All **54 latest-head checks passed**; two optional Storybook checks were skipped. One general-test shard passed all 1,314 tests but failed during runner temporary-directory cleanup with `EACCES`. One rerun passed. [CI run](https://github.com/paperclipai/paperclip/actions/runs/37940997283). - Greptile completed on final head `bca87df089c541450932f6b678ca776942d267b2` with **5/5 and no actionable findings**. There are no review threads or merge conflicts. - Structural comparison confirms 63 module function bodies and ten declarations match the originals. Remaining service function bodies and all 140 public exports are preserved. ## Risks Moving closures can break callback construction order or status settlement. The factory receives explicit callbacks and starts no work during construction. Compare-and-set predicates, native ownership holds, revoked-wake guards, event sequencing, redaction, and completion policies keep their original bodies. This PR has no schema or API changes. ## 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:
1 parent
57e977be72
commit
0b80ea17ef
5 files changed
+3317
-2808
No files matched your search
@@ -220,18 +220,18 @@ describe("createCoalescedAsyncTrigger", () => {
|
||||
|
||||
describe("chat publication commit signals", () => {
|
||||
it("emits the accepted signals only after their durable source writes", () => {
|
||||
const heartbeatSource = readFileSync(
|
||||
new URL("./heartbeat.ts", import.meta.url),
|
||||
const lifecycleSource = readFileSync(
|
||||
new URL("./heartbeat/run-lifecycle.ts", import.meta.url),
|
||||
"utf8",
|
||||
);
|
||||
const appendRunEventStart = heartbeatSource.indexOf(
|
||||
const appendRunEventStart = lifecycleSource.indexOf(
|
||||
"async function appendRunEvent(",
|
||||
);
|
||||
const persistedEvent = heartbeatSource.indexOf(
|
||||
const persistedEvent = lifecycleSource.indexOf(
|
||||
"await appendHeartbeatRunEvent",
|
||||
appendRunEventStart,
|
||||
);
|
||||
const emittedEvent = heartbeatSource.indexOf(
|
||||
const emittedEvent = lifecycleSource.indexOf(
|
||||
"publishLiveEvent({",
|
||||
persistedEvent,
|
||||
);
|
||||
@@ -239,6 +239,10 @@ describe("chat publication commit signals", () => {
|
||||
expect(persistedEvent).toBeGreaterThan(appendRunEventStart);
|
||||
expect(emittedEvent).toBeGreaterThan(persistedEvent);
|
||||
|
||||
const heartbeatSource = readFileSync(
|
||||
new URL("./heartbeat.ts", import.meta.url),
|
||||
"utf8",
|
||||
);
|
||||
const presentationMarker = heartbeatSource.indexOf(
|
||||
'eventType: "run.presentation.resolved"',
|
||||
);
|
||||
|
||||
+118
-2800
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,196 @@
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
|
||||
import type { CurrentUserRedactionOptions } from "../../log-redaction.js";
|
||||
import { publishLiveEvent } from "../live-events.js";
|
||||
import { getHeartbeatRunRuntimeStatus } from "../heartbeat-run-runtime-status.js";
|
||||
import { createHeartbeatRunState } from "./run-state.js";
|
||||
import { createHeartbeatRunPreparation } from "./run-preparation.js";
|
||||
import { createHeartbeatLifecycle, type HeartbeatLifecycleDependencies } from "./run-lifecycle.js";
|
||||
|
||||
vi.mock("../live-events.js", () => ({ publishLiveEvent: vi.fn() }));
|
||||
|
||||
type Run = typeof heartbeatRuns.$inferSelect;
|
||||
function callbacks(db: Db) {
|
||||
const state = createHeartbeatRunState(db);
|
||||
return {
|
||||
...state,
|
||||
...createHeartbeatRunPreparation(db),
|
||||
getRun: vi.fn(state.getRun),
|
||||
issuesSvc: {
|
||||
addComment: vi.fn<HeartbeatLifecycleDependencies["issuesSvc"]["addComment"]>(),
|
||||
listReviewAttention: vi.fn<HeartbeatLifecycleDependencies["issuesSvc"]["listReviewAttention"]>(),
|
||||
},
|
||||
recovery: { escalateStrandedAssignedIssue: vi.fn<HeartbeatLifecycleDependencies["recovery"]["escalateStrandedAssignedIssue"]>() },
|
||||
companySkills: { completeTestRunForIssue: vi.fn<HeartbeatLifecycleDependencies["companySkills"]["completeTestRunForIssue"]>() },
|
||||
budgets: { getInvocationBlock: vi.fn(async () => null) },
|
||||
treeControlSvc: { getActivePauseHoldGate: vi.fn(async () => null) },
|
||||
enqueueWakeup: vi.fn<HeartbeatLifecycleDependencies["enqueueWakeup"]>(),
|
||||
getCurrentUserRedactionOptions: vi.fn(async (): Promise<CurrentUserRedactionOptions> => ({})),
|
||||
getAgentInvokability: vi.fn<HeartbeatLifecycleDependencies["getAgentInvokability"]>(async () => ({ invokable: true })),
|
||||
emitTerminalAgentTaskRun: vi.fn<HeartbeatLifecycleDependencies["emitTerminalAgentTaskRun"]>(),
|
||||
budgetHooks: {},
|
||||
} satisfies HeartbeatLifecycleDependencies;
|
||||
}
|
||||
function guardedDatabase() {
|
||||
const access = vi.fn(() => { throw new Error("Unexpected lifecycle database access"); });
|
||||
return { db: new Proxy({}, { get: access }) as Db, access };
|
||||
}
|
||||
|
||||
describe("heartbeat lifecycle module boundary", () => {
|
||||
it("constructs without database work or reporting effects", () => {
|
||||
const database = guardedDatabase();
|
||||
const deps = callbacks(database.db);
|
||||
createHeartbeatLifecycle(database.db, deps);
|
||||
expect(database.access).not.toHaveBeenCalled();
|
||||
for (const callback of Object.values(deps)) if (vi.isMockFunction(callback)) expect(callback).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("ignores absent wake requests without accessing the database", async () => {
|
||||
const database = guardedDatabase();
|
||||
const lifecycle = createHeartbeatLifecycle(database.db, callbacks(database.db));
|
||||
await lifecycle.setWakeupStatus(null, "completed");
|
||||
await lifecycle.setWakeupStatus(undefined, "completed");
|
||||
expect(database.access).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ livenessState: "empty_response", contextSnapshot: { goalControlRequestId: "goal" } },
|
||||
{ livenessState: "plan_only", contextSnapshot: { resumeSessionGoalHeartbeat: true } },
|
||||
{ livenessState: "productive", contextSnapshot: { issueId: "issue" } },
|
||||
{ livenessState: "empty_response", contextSnapshot: {} },
|
||||
])("keeps ineligible continuations out of wake admission (%j)", async (input) => {
|
||||
const database = guardedDatabase();
|
||||
const deps = callbacks(database.db);
|
||||
await createHeartbeatLifecycle(database.db, deps).handleRunLivenessContinuation(input as unknown as Run);
|
||||
expect(database.access).not.toHaveBeenCalled();
|
||||
expect(deps.enqueueWakeup).not.toHaveBeenCalled();
|
||||
expect(deps.budgets.getInvocationBlock).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("preserves retained native ownership before lease-release teardown reads or writes", async () => {
|
||||
const database = guardedDatabase();
|
||||
const deps = callbacks(database.db);
|
||||
const run = { runtimeMode: "native", status: "running", nativePhase: "terminal_failure", errorCode: "native_execution_ownership_unverified" } as Run;
|
||||
expect(await createHeartbeatLifecycle(database.db, deps).terminalizeRunOnLeaseRelease(run)).toBe(run);
|
||||
expect(database.access).not.toHaveBeenCalled();
|
||||
expect(deps.emitTerminalAgentTaskRun).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("rejects late progress after re-reading a terminal run through the service callback", async () => {
|
||||
const database = guardedDatabase();
|
||||
const deps = callbacks(database.db);
|
||||
const run = { id: "late-progress", companyId: "company", agentId: "agent", status: "running", contextSnapshot: {} } as Run;
|
||||
deps.getRun.mockResolvedValue({ ...run, status: "succeeded" });
|
||||
expect(await createHeartbeatLifecycle(database.db, deps).recordCurrentHeartbeatRunRuntimeProgress(run, { phase: "finalize", message: "late update" }, null)).toBeNull();
|
||||
expect(deps.getRun).toHaveBeenCalledWith(run.id);
|
||||
expect(getHeartbeatRunRuntimeStatus(run.id)).toBeNull();
|
||||
expect(database.access).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
if (!support.supported) console.warn(`Skipping lifecycle module database tests: ${support.reason}`);
|
||||
describe.skipIf(!support.supported)("heartbeat lifecycle database wiring", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: Db;
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-lifecycle-module");
|
||||
db = createDb(database.connectionString);
|
||||
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
|
||||
afterAll(async () => {
|
||||
await db?.$client.end();
|
||||
await database?.cleanup();
|
||||
});
|
||||
beforeEach(async () => {
|
||||
vi.mocked(publishLiveEvent).mockClear();
|
||||
await db.execute(sql`truncate table companies restart identity cascade`);
|
||||
});
|
||||
|
||||
async function fixture(patch: Partial<typeof heartbeatRuns.$inferInsert> = {}) {
|
||||
const [company] = await db.insert(companies).values({ name: "Lifecycle module", issuePrefix: "LM" }).returning();
|
||||
const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Lifecycle agent" }).returning();
|
||||
const [wake] = await db.insert(agentWakeupRequests).values({ companyId: company.id, agentId: agent.id, source: "assignment", status: "claimed" }).returning();
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, wakeupRequestId: wake.id, status: "running", runtimeMode: "native", ...patch }).returning();
|
||||
return { company, agent, wake, run };
|
||||
}
|
||||
|
||||
it("keeps one terminal outcome and reports only the winner of competing writes", async () => {
|
||||
const { run } = await fixture();
|
||||
const deps = callbacks(db);
|
||||
const lifecycle = createHeartbeatLifecycle(db, deps);
|
||||
const results = await Promise.all([
|
||||
lifecycle.setRunStatusIfRunning(run.id, "succeeded", { finishedAt: new Date() }),
|
||||
lifecycle.setRunStatusIfRunning(run.id, "failed", { errorCode: "late_failure", finishedAt: new Date() }),
|
||||
]);
|
||||
const winner = results.find(result => result.updated)!;
|
||||
expect(results.filter(result => result.updated)).toHaveLength(1);
|
||||
const [persisted] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
|
||||
expect(persisted.status).toBe(winner.run?.status);
|
||||
expect(results.every(result => result.run?.status === persisted.status)).toBe(true);
|
||||
expect(deps.emitTerminalAgentTaskRun).toHaveBeenCalledTimes(1);
|
||||
expect(deps.emitTerminalAgentTaskRun).toHaveBeenCalledWith(winner.run, "running", undefined);
|
||||
expect(vi.mocked(publishLiveEvent).mock.calls.filter(([event]) => event.type === "heartbeat.run.status")).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("rechecks native ownership atomically before a terminal write", async () => {
|
||||
const { run } = await fixture({ errorCode: "native_execution_ownership_unverified", nativePhase: "terminal_failure" });
|
||||
const deps = callbacks(db);
|
||||
const result = await createHeartbeatLifecycle(db, deps).setRunStatusIfRunning(run.id, "failed");
|
||||
expect(result).toMatchObject({ updated: false, run: { status: "running", errorCode: run.errorCode } });
|
||||
expect(deps.emitTerminalAgentTaskRun).not.toHaveBeenCalled();
|
||||
expect(publishLiveEvent).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each(["setRunStatus", "setRunStatusIfRunning"] as const)("preserves accounting metadata while %s enriches usage", async (operation) => {
|
||||
const { run } = await fixture({ usageJson: { accountingReceiptReady: false, usageSource: "per_run", inputTokens: 5 } });
|
||||
await createHeartbeatLifecycle(db, callbacks(db))[operation](run.id, "succeeded", { usageJson: { inputTokens: 8, outputTokens: 3 } });
|
||||
const [persisted] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
|
||||
expect(persisted.usageJson).toEqual({ accountingReceiptReady: false, usageSource: "per_run", inputTokens: 8, outputTokens: 3 });
|
||||
});
|
||||
|
||||
it.each(["claimed", "cancelled"] as const)("settles a %s wake without reviving a revoked request", async (status) => {
|
||||
const { wake } = await fixture();
|
||||
await db.update(agentWakeupRequests).set({ status }).where(eq(agentWakeupRequests.id, wake.id));
|
||||
await createHeartbeatLifecycle(db, callbacks(db)).setWakeupStatus(wake.id, "completed", { reason: "settled" });
|
||||
const [persisted] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id));
|
||||
expect(persisted.status).toBe(status === "cancelled" ? "cancelled" : "completed");
|
||||
expect(persisted.reason).toBe(status === "cancelled" ? wake.reason : "settled");
|
||||
});
|
||||
|
||||
it("publishes the persisted sequence for concurrent event writes", async () => {
|
||||
const { run } = await fixture();
|
||||
const lifecycle = createHeartbeatLifecycle(db, callbacks(db));
|
||||
await Promise.all(["first", "second"].map(message => lifecycle.appendRunEvent(run, { eventType: "lifecycle", message })));
|
||||
const rows = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, run.id)).orderBy(heartbeatRunEvents.seq);
|
||||
expect(rows.map(row => row.seq)).toEqual([1, 2]);
|
||||
const published = vi.mocked(publishLiveEvent).mock.calls.map(([event]) => event).filter(event => event.type === "heartbeat.run.event");
|
||||
expect(published).toHaveLength(2);
|
||||
for (const row of rows) expect(published.find(event => event.payload?.seq === row.seq)?.payload).toMatchObject({ runId: run.id, message: row.message });
|
||||
});
|
||||
|
||||
it("reads current redaction settings for each event and persists the published sanitized payload", async () => {
|
||||
const { run } = await fixture();
|
||||
const deps = callbacks(db);
|
||||
deps.getCurrentUserRedactionOptions.mockResolvedValueOnce({ enabled: true, userNames: ["alice"], homeDirs: ["/Users/alice"] });
|
||||
deps.getCurrentUserRedactionOptions.mockResolvedValueOnce({ enabled: true, userNames: ["bob"], homeDirs: ["/Users/bob"] });
|
||||
const lifecycle = createHeartbeatLifecycle(db, deps);
|
||||
for (const name of ["alice", "bob"]) await lifecycle.appendRunEvent(run, {
|
||||
eventType: "lifecycle", message: `Opening /Users/${name}/project`, payload: { path: `/Users/${name}/project`, token: "test-credential-material" },
|
||||
});
|
||||
const rows = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, run.id)).orderBy(heartbeatRunEvents.seq);
|
||||
expect(deps.getCurrentUserRedactionOptions).toHaveBeenCalledTimes(2);
|
||||
expect(rows).toHaveLength(2);
|
||||
const published = vi.mocked(publishLiveEvent).mock.calls.map(([event]) => event).filter(event => event.type === "heartbeat.run.event");
|
||||
expect(published).toHaveLength(2);
|
||||
for (const row of rows) {
|
||||
const text = JSON.stringify({ message: row.message, payload: row.payload });
|
||||
expect(text).not.toContain("alice");
|
||||
expect(text).not.toContain("bob");
|
||||
expect(text).not.toContain("test-credential-material");
|
||||
expect(published.find(event => event.payload?.seq === row.seq)?.payload).toMatchObject({ message: row.message, payload: row.payload });
|
||||
}
|
||||
});
|
||||
});
|
||||
File diff suppressed because it is too large.
Load diff
Reference in new issue
Block a user