From a9cf4ee731327833c01545b2435d20b59d9450f9 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 8 Nov 2026 10:08:00 -0600 Subject: [PATCH] fix(outbox): deliver a user comment whose wake was lost on a task its agent then closed (#20066) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A user's comment on a task with a live run must reach the agent: it is steered into the run or delivered in a follow-up run. Per-comment exactly-once delivery is plan invariant 4 > - A comment commits before its wake is written. A control-plane restart in between loses the wake, and the reconciler's lost-wake rule (chaos s9401/t69) saves the missing outbox entry > - That rule only looked at `todo` and `in_progress` tasks. In the restart chaos Daytona wave s9501, task t06 got a comment 10 ms before a K1 kill. The live run then finished, its agent closed the task, and the comment was never read > - This pull request extends the rule to a task its agent closed, and reopens that task for the user's message, as the release drain already does > - The benefit is that a message sent while the agent works is never silently dropped by a restart ## Linked Issues or Issue Description Refs #19831 (plan, section E: work-intent outbox, invariant 4), and #19992 (the s9401/t69 and t72 rules this extends). Found by the restart chaos campaign, wave w-20261108T142906-s9501, task t06: - The comment was saved at 14:40:17.048. - The K1 kill came at .058. - No wake request ever carried the comment, and no run event shows it steered in. - The run that started before the comment finished at 14:44:22 and closed the task. ## What Changed - `recovery/ownership-invariant-report.ts` (`checkUserCommentsWithoutWake`): also reports a comment on a `done` task its agent closed (`taskClosedByAgentSql`), when no wake carries it, no run started after it, and no run of the task logged it as a steered `userMessage` item. - `work-intent-outbox.ts` (`saveUndeliveredUserComments`): a new `reopenAgentClosedDone` option. In the same transaction that locks the issue, re-checks the agent close and saves the receipt, it reopens the task to `todo` through the issues service. It also logs `issue.updated` (reopened from done, source `undelivered_comment_backstop`). - `native-finalization-reconciler.ts` (`applyOwnershipReportBackstops`): the lost-wake backstop allows `done` and passes `reopenAgentClosedDone`. The receipt keeps its `wake_lost` / `reconciler_backstop` wait, so the existing release path delivers it once. ## Verification - `heartbeat-held-comment-wake.test.ts`, new test "delivers once a comment whose wake a restart lost mid-run on a task the run then closed". It fails on `master` and passes here. The violation is found, the backstop saves and reopens, the comment is delivered exactly once, and a second pass and backstop deliver nothing more. All 25 tests in the file pass. - `ownership-reconciliation-report.test.ts`, new test "finds a user comment that a restart left without a wake on a task its agent then closed". It fails on `master` and passes here: only the lost comment is reported, not one steered into the run and not one on a task a person closed. All 26 pass. - The server package typecheck passes. ## Risks - **Reopening a done task.** This only happens when the latest recorded close was not made by a person, the comment is a direct human comment, and no run received it or had it steered in. That's the same rule the t72 stale-continuation path and the release drain already apply. - **The steer check** joins the task's runs to their `item.started` events by comment ID. It runs only for candidate comments within the lookback window. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code execution in Claude Code. ## 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 - [x] 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 --- .../heartbeat-held-comment-wake.test.ts | 25 +++++++++++++++ .../ownership-reconciliation-report.test.ts | 31 +++++++++++++++++++ .../native-finalization-reconciler.ts | 5 +-- .../recovery/ownership-invariant-report.ts | 13 +++++++- server/src/services/work-intent-outbox.ts | 17 +++++++++- 5 files changed, 87 insertions(+), 4 deletions(-) diff --git a/server/src/__tests__/heartbeat-held-comment-wake.test.ts b/server/src/__tests__/heartbeat-held-comment-wake.test.ts index 24856d54ef2..c11ac218272 100644 --- a/server/src/__tests__/heartbeat-held-comment-wake.test.ts +++ b/server/src/__tests__/heartbeat-held-comment-wake.test.ts @@ -432,6 +432,31 @@ describePostgres("saved intents while another run's accounting is pending", () = expect(deliveries(f.commentId)).toHaveLength(1); }, 180_000); + it("delivers once a comment whose wake a restart lost mid-run on a task the run then closed", async () => { + const f = await seedHeldComment(); + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + await db.update(heartbeatRuns).set({ runtimeMode: "native", startedAt: new Date(Date.now() - 180_000), + createdAt: new Date(Date.now() - 180_000), finishedAt: new Date(Date.now() - 30_000) }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId)); + await db.update(issueComments).set({ createdAt: new Date(Date.now() - 120_000), updatedAt: new Date(Date.now() - 120_000) }) + .where(eq(issueComments.id, f.commentId)); + await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId)); + await closeByAgent(f); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(report.violations.filter(row => row.reason === "user_comment_without_wake").map(row => row.recordId)) + .toEqual([f.commentId]); + await applyOwnershipReportBackstops(db, report); + await db.update(agentWakeupRequests).set({ updatedAt: new Date(Date.now() - 60_000) }) + .where(eq(agentWakeupRequests.companyId, f.companyId)); + await settle(f); await settle(f); + expect(deliveries(f.commentId)).toHaveLength(1); + const again = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(again.violations.filter(row => row.reason === "user_comment_without_wake")).toEqual([]); + await applyOwnershipReportBackstops(db, again); + await settle(f); + expect(deliveries(f.commentId)).toHaveLength(1); + }, 180_000); + // Chaos s9401/t72: a message queued its own run while the first run was // finishing. The first run completed the task, so the queued run was // cancelled as a stale continuation and the unread message was dropped. diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index 68c853cac8c..f50c4d9dd04 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -12,6 +12,7 @@ import { companies, createDb, environmentLeases, + heartbeatRunEvents, heartbeatRuns, issueComments, issues, @@ -467,6 +468,36 @@ describe("report-only ownership reconciliation", () => { expect(later.violations.filter(row => row.reason === "user_comment_without_wake")).toEqual([]); }); + it("finds a user comment that a restart left without a wake on a task its agent then closed", async () => { + const closedTask = async (actorType: "agent" | "user") => { + const issueId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: `Closed by ${actorType}`, status: "done", assigneeAgentId: agentId }); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "succeeded", invocationSource: "manual", + runtimeMode: "native", contextSnapshot: { issueId }, startedAt: minutesAgo(6), createdAt: minutesAgo(6), finishedAt: minutesAgo(1) }) + .returning(); + const [comment] = await db.insert(issueComments).values({ companyId, issueId, authorUserId: "user-1", + body: "also add a follow-up line", createdAt: minutesAgo(4) }).returning(); + await db.insert(activityLog).values({ companyId, actorType, actorId: actorType === "agent" ? agentId : "user-1", + action: "issue.updated", entityType: "issue", entityId: issueId, details: { toStatus: "done" }, createdAt: minutesAgo(1) }); + return { issueId, runId: run!.id, commentId: comment!.id }; + }; + const lost = await closedTask("agent"); + const steered = await closedTask("agent"); + const goalSteered = await closedTask("agent"); + const personClosed = await closedTask("user"); + await db.insert(heartbeatRunEvents).values([ + { companyId, runId: steered.runId, agentId, seq: 1, eventType: "item.completed", stream: "system", level: "info", + sourceEventId: randomUUID(), payload: { prpEvent: { itemId: `provider-item-9f2:steer:${steered.commentId}`, + payload: { kind: "steering_acknowledgement" } } } }, + { companyId, runId: goalSteered.runId, agentId, seq: 1, eventType: "item.started", stream: "system", level: "info", + sourceEventId: randomUUID(), payload: { prpEvent: { itemId: "msg_provider_generated_7c1", + payload: { kind: "userMessage", commandId: `goal_comment_${goalSteered.commentId}` } } } }, + ]); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(report.violations.filter(row => row.reason === "user_comment_without_wake").map(row => row.recordId)) + .toEqual([lost.commentId]); + }); + it("keeps a backstop deadline for a lifted pause still inside its grace window", async () => { const updatedAt = minutesAgo(1); await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", status: "deferred_issue_execution", diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index 7e22e0c20bf..d915c90cc55 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -840,8 +840,9 @@ export async function applyOwnershipReportBackstops(db: Db, report: OwnershipRep .where(and(eq(issues.companyId, violation.companyId), eq(issues.id, violation.issueId))); if (!issue?.assigneeAgentId) continue; const saved = await saveUndeliveredUserComments(db, { companyId: violation.companyId, issueId: violation.issueId, - agentId: issue.assigneeAgentId, commentIds: [violation.recordId], allowedStatuses: lost ? ["todo", "in_progress"] : ["done"], - ...(lost ? { executionWait: deferredWorkIntentWait("wake_lost", "This message's wake was lost; it is sent now.") } : {}) }); + agentId: issue.assigneeAgentId, commentIds: [violation.recordId], allowedStatuses: lost ? ["todo", "in_progress", "done"] : ["done"], + ...(lost ? { executionWait: deferredWorkIntentWait("wake_lost", "This message's wake was lost; it is sent now."), + reopenAgentClosedDone: true } : {}) }); if (saved) await db.transaction(tx => promoteDeferredWorkIntents(tx, { companyId: violation.companyId!, trigger: "reconciler_backstop", issueId: violation.issueId })); } catch (error) { diff --git a/server/src/services/recovery/ownership-invariant-report.ts b/server/src/services/recovery/ownership-invariant-report.ts index 7cc486578b0..fae6a0cd9c7 100644 --- a/server/src/services/recovery/ownership-invariant-report.ts +++ b/server/src/services/recovery/ownership-invariant-report.ts @@ -453,7 +453,18 @@ async function checkUserCommentsWithoutWake(db: Db, now: Date, limit: number): P and c.author_agent_id is null and c.created_by_run_id is null and c.created_at >= ${since}::timestamptz and i.assignee_agent_id is not null and i.conversation_agent_id is null - and i.status in ('todo', 'in_progress') and i.execution_run_id is null + and (i.status in ('todo', 'in_progress') + or (i.status = 'done' and ${taskClosedByAgentSql(sql`c.company_id`, sql`c.issue_id`)})) + and i.execution_run_id is null + and not exists (select 1 from heartbeat_runs steered join heartbeat_run_events steer_event + on steer_event.company_id = steered.company_id and steer_event.run_id = steered.id + where steered.company_id = c.company_id + and (steered.context_snapshot->>'issueId' = c.issue_id::text or steered.native_issue_id = c.issue_id) + and coalesce(steered.started_at, steered.created_at) <= c.created_at + and (steered.finished_at is null or steered.finished_at >= c.created_at) + and steer_event.event_type in ('item.started', 'item.completed') + and steer_event.created_at >= c.created_at + and strpos(steer_event.payload::text, c.id::text) > 0) and exists (select 1 from heartbeat_runs prior where prior.company_id = c.company_id and (prior.context_snapshot->>'issueId' = c.issue_id::text or prior.native_issue_id = c.issue_id)) and not exists (select 1 from heartbeat_runs r where r.company_id = c.company_id diff --git a/server/src/services/work-intent-outbox.ts b/server/src/services/work-intent-outbox.ts index 89034210e03..fa94d0cca6d 100644 --- a/server/src/services/work-intent-outbox.ts +++ b/server/src/services/work-intent-outbox.ts @@ -5,6 +5,7 @@ import { withQueuedCommentIdsInWakePayload } from "./issue-queued-comment-queue. import { deliveredQueuedCommentIds } from "./queued-comment-delivery.js"; import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; +import { persistActivity, publishActivity, type ActivityPublication } from "./activity-log.js"; /** * Work-intent outbox (durable ownership plan, section E; invariant 4). @@ -111,9 +112,11 @@ export async function saveUndeliveredUserComments(db: Db, input: { source?: string | null; triggerDetail?: string | null; /** Names why the receipt waits and what promotes it. */ executionWait?: Record; + reopenAgentClosedDone?: boolean; }): Promise { if (input.commentIds.length === 0) return null; - return db.transaction(async tx => { + let publication: ActivityPublication | undefined; + const savedId = await db.transaction(async tx => { const [issue] = await tx.select({ status: issues.status, assigneeAgentId: issues.assigneeAgentId, conversationAgentId: issues.conversationAgentId }).from(issues) .where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId))).for("update"); @@ -147,6 +150,18 @@ export async function saveUndeliveredUserComments(db: Db, input: { [DEFERRED_WAKE_CONTEXT_KEY]: { issueId: input.issueId, taskId: input.issueId, wakeReason: "issue_commented", source: "issue.comment", wakeCommentIds: remaining, wakeCommentId: latest } }, remaining), }).returning({ id: agentWakeupRequests.id }); + if (saved && issue.status === "done" && input.reopenAgentClosedDone) { + const { issueService } = await import("./issues.js"); + const updated = await issueService(tx as unknown as Db).updateForCompany(input.issueId, input.companyId, + { status: "todo", executionState: null }, tx); + if (!updated) throw new Error("undelivered_comment_task_changed"); + publication = (await persistActivity(tx as unknown as Db, { companyId: input.companyId, actorType: "system", + actorId: "ownership-reconciler", agentId: input.agentId, action: "issue.updated", entityType: "issue", + entityId: input.issueId, + details: { status: "todo", reopened: true, reopenedFrom: "done", source: "undelivered_comment_backstop" } })).publication; + } return saved?.id ?? null; }); + if (publication) publishActivity(publication); + return savedId; }