mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-07 06:16:06 -10:00
fix(heartbeat): recover deferred dependency completion wakes (#19457)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Tasks wait for other tasks through saved dependency edges and wake requests. > - A dependency can finish while its assignee answers a question about the blocked task. > - Paperclip saves the completion wake behind that active turn. > - Recovery did not find that saved wake unless it also contained a comment or an interaction response. > - This pull request restores that wake through the existing release and dispatch checks. > - The task can then continue after the active turn and its cleanup finish. ## Linked Issues or Issue Description **What happened?** A task stayed blocked after its dependency finished. The completion wake was saved while an audit turn was active. The wake stayed deferred after that turn ended. It had no comment or interaction response, so both recovery queries and the input guard excluded it. **Expected behavior** Recover the saved dependency wake when the prior turn and its cleanup finish. Start one dependency continuation under the current execution rules. **Steps to reproduce** 1. Run a question response on a dependency-blocked task. 2. Finish the dependency while the response is active. 3. Save the completion wake behind that response. 4. Finish the response and its environment cleanup, or restart the workers. 5. Observe that the saved wake has no run. **Paperclip version or commit** Reproduced against `70d23612dd`. Rechecked against `2f35eb1080` with the final regression fixture. **Deployment mode** Source checkout on Linux x64, Node 25.9.0, pnpm 9.15.4, and isolated embedded PostgreSQL. Related work: #16673 added recovery for saved messages. #17288 and #16966 address assignment wakes after cleanup. This change covers dependency completion receipts with no message. ## What Changed - Include saved `issue_blockers_resolved` receipts in worker reconstruction and stranded queue selection. - Let these receipts pass the input guard without a comment or interaction response. - Use the existing release transaction and dispatch checks. Add no timer, migration, or direct task status change. - Add process-backed regression cases for restart, cleanup release, cancellation, agent pause, pending cleanup, and unresolved blockers. - Update the dependency wake recovery document. - Share the dependency-wake payload predicate and SQL fragment across reconstruction and stranded receipt admission, so these eligibility checks have one owner. ## Verification - The final regression fails on the base code: the receipt stays `deferred_issue_execution` and has no successor. The same test passes with the fix. - Six focused cases pass. They use real PostgreSQL and a real process adapter. They check saved output, one dependency continuation, clean task locks, repeat recovery, and execution gates. They do not call a model provider. - Three nearby suites pass: 125 tests across queue wiring, dependency scheduling, and dependency recovery. - Final-head server layered typechecking passes with `PAPERCLIP_TYPECHECK_LOW_MEMORY=1`. Whitespace and module boundary checks pass. Two parallel checker attempts reached the sandbox memory limit; the completed low-memory check covers all source and test layers. - The full deferred-receipt suite passes: 45 tests. Combined with the nearby suites, 170 local tests pass. These results cover head `8f9d3aa6ac19a6d2532bd869eb8adda7cf122a5a`. Required CI remains in progress. - Local workspace typechecking and build stop at the Runner Rust step because Cargo is absent. The full local test suite was not run. Required CI must supply broad coverage. - The hosted incident has not been recovered or deployed from this branch. - Review follow-up at `182385e9bc433adac06440997de7e4cfbf889a72`: normal merge of master `0d03610d2f73b7cd799547e43929bb2303319024`; no conflict or history rewrite. All 54 receipt/helper cases pass on macOS arm64, Node 26.4.0, real embedded PostgreSQL and process adapters. The six dependency continuation and safety cases also pass separately. Module boundaries and whitespace checks pass. No assertion, timeout, execution gate or retry policy changed for this extraction. - Current shared, SDK and adapter-utils source builds pass. The first broad local server typecheck read stale borrowed adapter declarations; rebuilding them exposed nominal type duplication between the borrowed workspace and this checkout. It is not counted as a server typecheck pass. All 30 SDK and 169 adapter runtime JavaScript outputs match the previously tested artifacts. Final revision `182385e9bc433adac06440997de7e4cfbf889a72` now passes all 55 applicable CI checks (three skipped, one neutral review), including clean-workspace typecheck, build and broad tests. [CI result](https://github.com/paperclipai/paperclip/actions/runs/42262173960). Greptile and Paperclip review both give 5/5 with no open findings. ## Risks - Recovery now considers a saved dependency receipt with no message. Current ownership, source finalization, cleanup, pause, cancellation, and dependency checks still control execution. - The scan remains bounded and uses existing transaction notifications and restart reconstruction. The integration checks verify that it becomes quiet after settlement. ## Model Used - OpenAI Codex, based on GPT-6. The exact provider model ID and context size are not exposed in this session. Used reasoning, tool calls, TypeScript edits, and test 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 <[email protected]>
This commit is contained in:
1 parent
e3fb4bc73a
commit
5d25117ffa
5 files changed
+119
-5
No files matched your search
@@ -13,6 +13,13 @@ unattributed operation in the same shared workspace does not remove its wait.
|
||||
Normal readiness, duplicate-wake, interaction, pause, and execution gates still
|
||||
apply. A board transition into blocked keeps the existing recovery behavior.
|
||||
|
||||
A dependency completion can arrive while the assignee answers a question about
|
||||
the blocked task. Its saved wake waits behind that turn. Deferred-receipt recovery
|
||||
reconstructs these dependency wakes after the turn and its cleanup finish, even
|
||||
when they contain no comment or interaction response. It uses the existing
|
||||
release and dispatch checks; task cancellation, agent pause, and pending cleanup
|
||||
still prevent execution.
|
||||
|
||||
Cancelling a run or pausing its agent suppresses immediate stranded-work recovery.
|
||||
A queued user message remains a separate continuation request. The release path
|
||||
still preserves independently authorized deferred input; stopping the old run
|
||||
|
||||
@@ -15,7 +15,7 @@ import { issueTreeControlService } from "../services/issue-tree-control.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { activityLog, agents, agentWakeupRequests, authUsers, chatActions, environmentLeases, nativeRunFinalizations, chatEndpoints, toolApplications, toolConnections, companies, companyMemberships, costEvents, createDb, projects, reconcileDatabaseWork, heartbeatRuns, issueComments, issues } from "@paperclipai/db";
|
||||
import { activityLog, agents, agentWakeupRequests, authUsers, chatActions, environmentLeases, nativeRunFinalizations, chatEndpoints, toolApplications, toolConnections, companies, companyMemberships, costEvents, createDb, projects, reconcileDatabaseWork, heartbeatRuns, issueComments, issueRelations, issues } from "@paperclipai/db";
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
@@ -158,6 +158,96 @@ describe("deferred receipt durable dispatch", () => {
|
||||
source: "on_demand", status: "completed", reason: "task_retry", payload: { issueId: f.issue.id } });
|
||||
await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; await quiet();
|
||||
});
|
||||
async function strandedDependency() {
|
||||
const f = await fixture();
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: f.issue.id,
|
||||
status: "succeeded", errorCode: null, contextSnapshot: { issueId: f.issue.id,
|
||||
wakeReason: "issue_commented", dependencyBlockedInteraction: true } }).where(eq(heartbeatRuns.id, f.source.id));
|
||||
await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issue.id,
|
||||
runId: f.source.id, phase: "committed" });
|
||||
await db.update(issues).set({ status: "blocked", executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issue.id));
|
||||
const [blocker] = await db.insert(issues).values({ companyId: f.companyId, title: "Completed review",
|
||||
status: "done", completedAt: new Date() }).returning();
|
||||
await db.insert(issueRelations).values({ companyId: f.companyId, issueId: blocker.id,
|
||||
relatedIssueId: f.issue.id, type: "blocks" });
|
||||
const [intent] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,
|
||||
source: "automation", reason: "issue_blockers_resolved", status: "coalesced",
|
||||
requestedByActorType: "system", requestedByActorId: "native-status-committer",
|
||||
payload: { issueId: f.issue.id, resolvedBlockerIssueId: blocker.id },
|
||||
}).returning();
|
||||
const [receipt] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,
|
||||
source: "automation", reason: "issue_execution_deferred", status: "deferred_issue_execution",
|
||||
requestedByActorType: "system", requestedByActorId: `native-status-wake-dispatch:${intent.id}`,
|
||||
payload: { issueId: f.issue.id, resolvedBlockerIssueId: blocker.id, _paperclipWakeContext: {
|
||||
issueId: f.issue.id, wakeReason: "issue_blockers_resolved", statusDecisionSource: "native_status_decision",
|
||||
nativeStatusWakeIntentId: intent.id } },
|
||||
}).returning();
|
||||
return { ...f, receipt, blocker };
|
||||
}
|
||||
|
||||
it.each(["restart", "cleanup"] as const)("recovers a dependency wake after native audit %s into one continuation", async event => {
|
||||
const f = await strandedDependency();
|
||||
const { receipt } = f;
|
||||
const [lease] = event === "cleanup" ? await db.insert(environmentLeases).values({ companyId: f.companyId,
|
||||
heartbeatRunId: f.source.id, issueId: f.issue.id, provider: "daytona", providerLeaseId: "dependency-finalizing",
|
||||
status: "active", leasePolicy: "reuse_by_environment" }).returning() : [];
|
||||
await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
|
||||
await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready;
|
||||
if (lease) {
|
||||
await waitForReal(async () => {
|
||||
await Promise.allSettled([...passes]);
|
||||
expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0);
|
||||
}, { timeout: 10_000 });
|
||||
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.id)))[0])
|
||||
.toMatchObject({ status: "deferred_issue_execution", runId: null });
|
||||
await environmentService(db).releaseLease(lease.id, "retained", { cleanupStatus: "success" });
|
||||
}
|
||||
await waitForReal(async () => {
|
||||
await Promise.allSettled([...passes]); await heartbeat.drainActiveRunExecutions();
|
||||
const [saved] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.id));
|
||||
const pending = saved.runId ? (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, saved.runId)))[0] : null;
|
||||
expect(saved.status, JSON.stringify({ status: pending?.status, error: pending?.error, errors })).toBe("completed");
|
||||
expect(saved.runId).not.toBeNull();
|
||||
}, { timeout: 15_000 });
|
||||
await finish(f.issue.id);
|
||||
const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId));
|
||||
const dependencyRuns = runs.filter(run => run.contextSnapshot?.wakeReason === "issue_blockers_resolved");
|
||||
expect(dependencyRuns).toHaveLength(1);
|
||||
const [resumed] = dependencyRuns;
|
||||
expect(resumed).toMatchObject({ status: "succeeded", stdoutExcerpt: expect.stringContaining("receipt-proof"),
|
||||
contextSnapshot: { wakeReason: "issue_blockers_resolved" } });
|
||||
expect(resumed.contextSnapshot?.dependencyBlockedInteraction).not.toBe(true);
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
const after = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId));
|
||||
expect(after.filter(run => run.contextSnapshot?.wakeReason === "issue_blockers_resolved")).toHaveLength(1);
|
||||
expect(after.find(run => run.id === f.source.id)?.status).toBe("succeeded");
|
||||
expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0])
|
||||
.toMatchObject({ status: "done", executionRunId: null, checkoutRunId: null });
|
||||
await quiet();
|
||||
});
|
||||
it.each(["paused", "cleanup", "cancelled"] as const)("keeps the %s gate when recovering a dependency wake", async gate => {
|
||||
const f = await strandedDependency();
|
||||
if (gate === "paused") await db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agent.id));
|
||||
if (gate === "cancelled") await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, f.issue.id));
|
||||
if (gate === "cleanup") await db.insert(environmentLeases).values({ companyId: f.companyId,
|
||||
heartbeatRunId: f.source.id, issueId: f.issue.id, provider: "daytona", providerLeaseId: "dependency-cleanup",
|
||||
status: "active", leasePolicy: "reuse_by_environment" });
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
const [saved] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.receipt.id));
|
||||
expect(saved).toMatchObject({ status: gate === "cancelled" ? "cancelled" : "deferred_issue_execution", runId: null });
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1);
|
||||
});
|
||||
it("checks unresolved blockers before a recovered dependency wake can execute", async () => {
|
||||
const f = await strandedDependency();
|
||||
await db.update(issues).set({ status: "blocked", completedAt: null }).where(eq(issues.id, f.blocker.id));
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.receipt.id));
|
||||
expect(receipt).toMatchObject({ status: "skipped", runId: expect.any(String) });
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, receipt.runId!));
|
||||
expect(run).toMatchObject({ status: "cancelled", errorCode: "issue_dependencies_blocked", stdoutExcerpt: null });
|
||||
expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0].status).toBe("blocked");
|
||||
});
|
||||
it.each([false, true])("reconsiders every saved receipt during repeated company eligibility changes (expired wait: %s)", async expiredWait => {
|
||||
const f = await fixture();
|
||||
const ids = Array.from({ length: 178 }, (_, index) => `${(index + 1).toString(16).padStart(8, "0")}-0000-4000-8000-000000000000`);
|
||||
|
||||
@@ -5,6 +5,7 @@ import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator
|
||||
import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
|
||||
import type { DeferredExecutionScope, DeferredReceiptRecoveryOptions } from "../deferred-execution-notifications.js";
|
||||
import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
|
||||
import { issueBlockersResolvedWakePayloadSql } from "../issue-dependency-wakeups.js";
|
||||
|
||||
type Entry = { id: string; companyId: string; dueAt: number; immediate: boolean };
|
||||
type NativeIntentOptions = DeferredExecutionScope & {
|
||||
@@ -74,6 +75,7 @@ export function registerDeferredReceiptWork(
|
||||
sql`jsonb_typeof(${agentWakeupRequests.payload}#>'{_paperclipWakeContext,wakeCommentIds}') = 'array'
|
||||
and ${agentWakeupRequests.payload}#>'{_paperclipWakeContext,wakeCommentIds}' <> '[]'::jsonb`,
|
||||
sql`${agentWakeupRequests.payload}->>'mutation' = 'interaction'`,
|
||||
issueBlockersResolvedWakePayloadSql,
|
||||
sql`exists (select 1 from heartbeat_runs r where r.company_id = ${agentWakeupRequests.companyId}
|
||||
and r.runtime_mode = 'legacy' and r.status = 'cancelled'
|
||||
and r.result_json->>'queuedCommentInterruptQueueId' = ${agentWakeupRequests.id}::text)`,
|
||||
|
||||
@@ -144,7 +144,11 @@ import {
|
||||
isThrottleCandidateIssueRewake,
|
||||
} from "../issue-rewake-throttle.js";
|
||||
import { logActivity } from "../activity-log.js";
|
||||
import { ISSUE_BLOCKERS_RESOLVED_WAKE_REASON } from "../issue-dependency-wakeups.js";
|
||||
import {
|
||||
ISSUE_BLOCKERS_RESOLVED_WAKE_REASON,
|
||||
isIssueBlockersResolvedWakePayload,
|
||||
issueBlockersResolvedWakePayloadSql,
|
||||
} from "../issue-dependency-wakeups.js";
|
||||
import { resumeIfPauseAlreadyReleased } from "../issue-tree-resume.js";
|
||||
import {
|
||||
ISSUE_PAUSED_ERROR_CODE,
|
||||
@@ -1757,7 +1761,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
or(and(
|
||||
sql`jsonb_typeof(${agentWakeupRequests.payload} #> '{_paperclipWakeContext,wakeCommentIds}') = 'array'`,
|
||||
sql`${agentWakeupRequests.payload} #> '{_paperclipWakeContext,wakeCommentIds}' <> '[]'::jsonb`,
|
||||
), sql`${agentWakeupRequests.payload}->>'mutation' = 'interaction'`),
|
||||
), sql`${agentWakeupRequests.payload}->>'mutation' = 'interaction'`,
|
||||
issueBlockersResolvedWakePayloadSql),
|
||||
sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is null`,
|
||||
sql`NOT EXISTS (SELECT 1 FROM heartbeat_runs live_issue_run
|
||||
WHERE live_issue_run.company_id = ${issues.companyId}
|
||||
@@ -1768,7 +1773,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
.orderBy(asc(agentWakeupRequests.updatedAt)).limit(50);
|
||||
const releasedNativeSources = new Set<string>();
|
||||
for (const { wake } of strandedQueues) {
|
||||
if (!queuedCommentIdsFromWakePayload(wake.payload).length &&
|
||||
const dependencyWake = isIssueBlockersResolvedWakePayload(wake.payload);
|
||||
if (!dependencyWake && !queuedCommentIdsFromWakePayload(wake.payload).length &&
|
||||
!await readQueuedInteractionResponse(db, wake.companyId, String(wake.payload?.issueId), wake.payload)) continue;
|
||||
const [latest] = await db.select({
|
||||
id: heartbeatRuns.id,
|
||||
|
||||
@@ -1,10 +1,19 @@
|
||||
import { and, eq, inArray } from "drizzle-orm";
|
||||
import { and, eq, inArray, sql } from "drizzle-orm";
|
||||
import { createHash } from "node:crypto";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { agentWakeupRequests } from "@paperclipai/db";
|
||||
|
||||
export const ISSUE_BLOCKERS_RESOLVED_WAKE_REASON = "issue_blockers_resolved";
|
||||
|
||||
export const issueBlockersResolvedWakePayloadSql =
|
||||
sql`${agentWakeupRequests.payload}#>>'{_paperclipWakeContext,wakeReason}' = ${ISSUE_BLOCKERS_RESOLVED_WAKE_REASON}`;
|
||||
|
||||
export function isIssueBlockersResolvedWakePayload(payload: Record<string, unknown> | null | undefined): boolean {
|
||||
const context = payload?._paperclipWakeContext;
|
||||
return context !== null && typeof context === "object" && !Array.isArray(context)
|
||||
&& (context as Record<string, unknown>).wakeReason === ISSUE_BLOCKERS_RESOLVED_WAKE_REASON;
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether moving an issue into `blocked` should immediately wake its assignee
|
||||
* because every recorded blocker is already resolved ("blockers restored").
|
||||
|
||||
Reference in new issue
Block a user