mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-08 12:33:05 +01:00
fix(native): deliver messages saved while a native run finalizes (#19183)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A user can reply on a task while the agent's run is still finishing > - A remote native run first commits its result and its status decision (for example `done`). Then its sandbox environment settles (harness backup, warm retention). > - A message that arrives after the final turn took its input is saved as a deferred receipt behind that run > - The saved-message recovery path holds the receipt while the source run still owns its environment lease or its workspace copy-back owner > - Nothing notified that path when the lease or the owner settled, so the messages waited for an unrelated company event > - This pull request notifies the existing deferred-receipt work when a run's environment lease settles or its workspace finalization owner is released > - The benefit is that messages sent while a run finalizes are delivered once, soon after the run settles, and a task closed in that window is reopened for them ## Linked Issues or Issue Description **What happened?** A user posted three plain comments on a task while the agent's remote native run was finalizing. The final turn had already called its finish tool. The comments were saved in one deferred receipt. The run then committed `done` and its status became `succeeded`. No follow-up run started. The task stayed `done` with three unanswered requests. About 20 minutes later, an unrelated event in the same company woke the deferred-receipt worker, and the messages were delivered. **Expected behavior** A message that arrives after the last turn took its input, but before the run releases, gets one follow-up turn as soon as the run settles. If finalization marked the task `done` in that window, the task is reopened for the message. Messages that the finished turn already consumed are not replayed. **Steps to reproduce** 1. Start a native run with a remote (sandbox) environment. 2. After the agent calls its finish tool, post a user comment on the task. 3. Let finalization commit `done` and project the run as `succeeded`. 4. Observe that the comment's deferred receipt stays `deferred_issue_execution` after the environment lease settles. **Root cause** A remote native run is finished by native finalization, which projects the terminal status. The in-process run completion (`releaseIssueExecutionAndPromote`) does not run for it. Its saved messages depend on the stranded-queue sweep in `resumeDeferredExecutionReceipts`, which calls the release with `resumeStrandedQueue`. That release correctly waits while `hasPendingRunCleanup` is true (an unsettled environment lease or a workspace finalization owner). Finalization notifies the deferred-receipt worker when it projects the run, but the cleanup is still pending at that time. The worker then drops the entry. Lease settlement and owner release did not notify the worker again. Refs #19122 (saves a redirect sent while a stopped run settles; this change reuses the same deferred-receipt delivery path for normally completing runs). ## What Changed - `environmentService.releaseLease` notifies deferred-execution work for the company when a run's lease leaves `active` or `pending_cleanup` (released, retained, or failed). - `environmentService.releaseLeasesForRun` notifies deferred-execution work in the same way. - `withNativeWorkspaceFinalizationOwnership` notifies deferred-execution work when it releases its owner. - Delivery uses the existing path: `resumeDeferredExecutionReceipts` → stranded-queue release → deferred wake promotion (with the existing reopen rule for user messages). No new delivery mechanism. - New regression in `deferred-execution-receipt-work.test.ts` with the real deferred-receipt worker, the real release adapter, and the real lease service. - `doc/execution-semantics.md`: one additive paragraph for messages sent while a run finalizes. ## Verification - Staging evidence (read-only): the task's saved queue held the three comments as one `deferred_issue_execution` receipt after the run succeeded. The run had no `run succeeded` lifecycle event (it was finished by native finalization, not run completion). The receipt was promoted about 20 minutes later with `reopenedFrom: done` by the same recovery path. - Red before the fix: the new test failed with `expected null to be truthy` (the receipt stayed deferred with no run after the lease settled). - Green after the fix: `cd server && npx vitest run src/__tests__/deferred-execution-receipt-work.test.ts -t "finishing native run"`. - Safety cases in the new test: - The receipt stays deferred while the environment lease is active (cleanup holds still apply). - After settlement, exactly one run carries both messages in order, and the done task is reopened once. - A second settlement signal does not deliver the messages again. - Related suites: `deferred-execution-receipt-work`, `environment-service`, `environment-cleanup-work`, `native-environment-settlement-work`, `idle-sleep-safety`. - Not done: a live staging reproduction at the fixed revision. Deployment is out of scope. ## Risks - The deferred-receipt worker reloads the company's deferred receipts after each run lease settlement. Each reload is bounded and runs only when deferred receipts exist. The notification is transactional, so a rolled-back release does not wake the worker. - Low risk: no schema change, no API change, and no change to admission or reopen rules. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, in Claude Code with tool use (shell, file edits, browser for read-only staging diagnostics). ## 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 - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [ ] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
fec0b57659
commit
512fbea0bb
4 files changed
+69
-1
No files matched your search
@@ -802,6 +802,8 @@ Normal issue-thread feedback can still wake the current assignee according to th
|
||||
|
||||
Restart recovery revisits accepted deferred messages through the latest issue run, including historical mention queues for another agent. Ordinary messages for a former owner retain their assignment handoff. Native sources require their matching committed finalization, settled environment leases, and released finalizer and copyback ownership before recovery clears old issue claims. A successful retained warm release can satisfy the lease check. Released local bookkeeping without a provider allocation and a matching SSH bookkeeping release need no provider cleanup receipt. Plugin releases record their successful acknowledgement. Pending or failed cleanup and durable owners still hold work. A successful fenced cleanup completion clears its in-flight claim in the same database write, so a restart before retry bookkeeping cannot strand saved input. The normal release checks preserve cleanup holds, pauses, and operator Stops. Recovery checks for a competing turn under the issue lock before it promotes saved input.
|
||||
|
||||
A user can send messages after a run's final turn took its input but before the run settles. A native run can finish its result and status decision before its environment cleanup completes. The messages remain saved while that cleanup holds the source. Releasing or retaining the run's environment lease, or releasing its workspace finalization owner, reconsiders the saved messages through this recovery path. They are delivered once. A task that finalization marked `done` is reopened for them. The release and admission gates above still apply.
|
||||
|
||||
Repeated eligibility notifications must not move an unchecked, ready deferred receipt behind receipts already reconsidered. Paginated reconstruction preserves pending ready receipts' dispatch priority while refreshing their current state. Receipts whose wait deadline has passed join the ready queue on reconstruction. Future wait grace periods, claimed-work deadlines, infrastructure retry backoffs, and current admission checks remain effective.
|
||||
|
||||
A paused recipient does not fail an accepted deferred request. Release and periodic recovery preserve its pending receipt and may process an eligible request behind it. Resuming the agent allows normal admission to reconsider the saved input. Missing, foreign, terminated, and unapproved recipients retain their rejection rules. Explicit task holds and operator Stops still prevent unauthorized execution. This does not replay historical failed requests.
|
||||
|
||||
@@ -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 { agents, agentWakeupRequests, authUsers, chatActions, 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, 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";
|
||||
@@ -24,6 +24,7 @@ import { registerDeferredReceiptWork } from "../services/heartbeat/deferred-rece
|
||||
import { heartbeatService } from "../services/heartbeat.js";
|
||||
import { requestTaskRetry } from "../services/task-retry.js";
|
||||
import { issueService } from "../services/issues.js";
|
||||
import { environmentService } from "../services/environments.js";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
|
||||
vi.setConfig({ testTimeout: 90_000 });
|
||||
@@ -1014,4 +1015,61 @@ describe("deferred receipt durable dispatch", () => {
|
||||
});
|
||||
await quiet(); expect(dispatchNative).not.toHaveBeenCalled();
|
||||
});
|
||||
it("delivers messages saved behind a finishing native run once, when its environment cleanup settles", async () => {
|
||||
// Remote native finalization commits Done and projects the run terminal
|
||||
// without the in-process release. The provider environment then settles
|
||||
// (harness backup, warm retention). Messages the user sent after the final
|
||||
// turn took its input wait on that cleanup and must then run exactly once.
|
||||
const f = await fixture();
|
||||
await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issue.id));
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded", errorCode: null, resultJson: {}, runtimeMode: "native",
|
||||
nativeIssueId: f.issue.id, finishedAt: new Date() }).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" });
|
||||
const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: f.source.id,
|
||||
issueId: f.issue.id, provider: "daytona", providerLeaseId: "sandbox-finalizing", status: "active",
|
||||
leasePolicy: "reuse_by_environment" }).returning();
|
||||
const comments = [];
|
||||
for (const body of ["Also mention saved searches.", "Also mention it works on mobile."]) {
|
||||
comments.push((await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issue.id, authorType: "user",
|
||||
authorUserId: f.actorId, body }).returning())[0]);
|
||||
}
|
||||
const commentIds = comments.map(comment => comment.id);
|
||||
const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,
|
||||
source: "automation", triggerDetail: "system", reason: "issue_execution_deferred", status: "deferred_issue_execution",
|
||||
requestedByActorType: "user", requestedByActorId: f.actorId,
|
||||
payload: { issueId: f.issue.id, commentId: commentIds[1], mutation: "comment", _paperclipWakeContext: {
|
||||
issueId: f.issue.id, taskId: f.issue.id, commentId: commentIds[1], wakeCommentId: commentIds[1],
|
||||
wakeCommentIds: commentIds, wakeReason: "issue_commented", source: "issue.comment" } } }).returning();
|
||||
const readWake = async () => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)))[0];
|
||||
const issueRuns = () => db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${f.issue.id}`);
|
||||
|
||||
// The receipt is checked while the source environment is still active.
|
||||
await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
|
||||
await waitForReal(async () => {
|
||||
await Promise.allSettled([...passes]);
|
||||
expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0);
|
||||
}, { timeout: 10_000 });
|
||||
expect(await readWake()).toMatchObject({ status: "deferred_issue_execution", runId: null });
|
||||
expect(await issueRuns()).toHaveLength(1);
|
||||
|
||||
await environmentService(db).releaseLease(lease.id, "retained", { cleanupStatus: "success" });
|
||||
await waitForReal(async () => {
|
||||
await Promise.allSettled([...passes]);
|
||||
expect((await readWake()).runId).toBeTruthy();
|
||||
}, { timeout: 10_000 });
|
||||
// Exactly one turn carries the saved messages, in order, under the
|
||||
// original user's authority, and the completed task is reopened for it.
|
||||
const commentRuns = async () => (await issueRuns()).filter(run => run.contextSnapshot?.wakeReason === "issue_commented");
|
||||
const [successor, ...duplicates] = await commentRuns();
|
||||
expect(duplicates).toEqual([]);
|
||||
expect(successor.contextSnapshot?.wakeCommentIds).toEqual(commentIds);
|
||||
expect((await readWake()).runId).toBe(successor.id);
|
||||
expect(await db.select().from(activityLog).where(and(eq(activityLog.entityId, f.issue.id),
|
||||
sql`${activityLog.details}->>'reopened' = 'true'`))).toHaveLength(1);
|
||||
await finish(f.issue.id);
|
||||
// A later settlement signal cannot deliver the same messages again.
|
||||
await environmentService(db).releaseLease(lease.id, "retained", { cleanupStatus: "success" });
|
||||
await quiet();
|
||||
expect(await commentRuns()).toHaveLength(1);
|
||||
});
|
||||
});
|
||||
@@ -4,6 +4,7 @@ import { notifyNativeEnvironmentSettlement } from "./native-environment-settleme
|
||||
import { notifyRetainedNativeSessionCleanup } from "./retained-native-session-cleanup-notifications.js";
|
||||
import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js";
|
||||
import { notifyEnvironmentCleanup } from "./environment-cleanup-notifications.js";
|
||||
import { notifyDeferredExecutionWork } from "./deferred-execution-notifications.js";
|
||||
import { withoutCleanupVerification } from "./cleanup-verification-proof.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
|
||||
@@ -1729,6 +1730,9 @@ export function environmentService(db: Db) {
|
||||
await notifyRetainedNativeSessionCleanup(tx, {
|
||||
companyId: previous.companyId, runId: previous.heartbeatRunId,
|
||||
});
|
||||
// Messages saved behind a finished run wait for its cleanup to
|
||||
// settle. Reconsider them through normal deferred admission.
|
||||
await notifyDeferredExecutionWork(tx, { companyId: previous.companyId });
|
||||
}
|
||||
if (previous.creationOperationId && previous.status === "pending_cleanup" && options?.cleanupStatus === "success" && status !== "pending_cleanup") {
|
||||
await tx.update(sandboxCreationOperations).set({ state: "resolved", updatedAt: now })
|
||||
@@ -1868,6 +1872,7 @@ export function environmentService(db: Db) {
|
||||
if (run) {
|
||||
await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: heartbeatRunId });
|
||||
await notifyRetainedNativeSessionCleanup(tx, { companyId: run.companyId, runId: heartbeatRunId });
|
||||
await notifyDeferredExecutionWork(tx, { companyId: run.companyId });
|
||||
}
|
||||
return tx.update(environmentLeases)
|
||||
.set({
|
||||
|
||||
@@ -2,6 +2,7 @@ import { notifyNativeResultCommit } from "../native-result-commit-notifications.
|
||||
import { notifyNativeWorkspaceFinalization, releaseNativeWorkspaceFinalization } from "../native-workspace-finalization-notifications.js";
|
||||
import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js";
|
||||
import { notifyQueuedRunDispatch } from "../run-dispatch-notifications.js";
|
||||
import { notifyDeferredExecutionWork } from "../deferred-execution-notifications.js";
|
||||
import { beginRunPhase } from "../execution-preparation-timing.js";
|
||||
import { logActivity } from "../activity-log.js";
|
||||
import { issueRecoveryActionService } from "../issue-recovery-actions.js";
|
||||
@@ -304,6 +305,8 @@ export async function withNativeWorkspaceFinalizationOwnership<T>(
|
||||
await notifyNativeWorkspaceFinalization(tx, { companyId: input.companyId, runId: input.runId });
|
||||
await notifyNativeStatusReassessment(tx, { companyId: input.companyId, runId: input.runId });
|
||||
await notifyQueuedRunDispatch(tx, { companyId: input.companyId });
|
||||
// Saved messages also wait on this owner after the run finished.
|
||||
await notifyDeferredExecutionWork(tx, { companyId: input.companyId });
|
||||
await tx.update(heartbeatRuns).set({
|
||||
runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) - ${OWNER_KEY}`,
|
||||
}).where(owned);
|
||||
|
||||
Reference in new issue
Block a user