From 5809e5b97e97122e1be1a969a180405dedbc4cd5 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 11 Sep 2026 17:56:04 -0500 Subject: [PATCH] fix(recovery): resume saved messages after cleanup resolves blockers Keep terminal stop receipts intact when a late process callback arrives. Reconsider saved execution-wait comments independently of recovery-action state, validate their author under the task lock, and retain admission gates and retry deduplication. Co-Authored-By: Paperclip --- .../explicit-native-continuation.test.ts | 49 +++++++++++++++++++ server/src/services/heartbeat.ts | 42 +++++++++++++--- 2 files changed, 85 insertions(+), 6 deletions(-) diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index b7eea198fe..47faf765d8 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -73,6 +73,55 @@ const support = await getEmbeddedPostgresTestSupport(); expect(await admit(f, true)).toBeNull(); }); + it("ignores late process callbacks without invalidating terminal stop evidence", async () => { + const f = await seed(); + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)); + expect(await recordNativeLocalProcessStop(db, source)).toBe(true); + await db.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, source.id)); + expect(await persistHeartbeatRunProcessMetadata(db, source.id, { + pid: 999999999, processGroupId: null, startedAt: new Date().toISOString(), + })).toBeNull(); + expect(await hasNativeLocalProcessStop(db, f.companyId, source.id)).toBe(true); + expect(await admit(f, true)).toMatchObject({ previousRunId: source.id }); + }); + + it.each(["available", "deleted", "different-author"])("reconsiders saved messages after cleanup resolves the blocker: %s", async state => { + const f = await seed(); + // Keep admission queued so the test never launches a real provider. + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + await db.update(heartbeatRuns).set({ processPid: process.pid }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await heartbeatService(db).wakeup(f.agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", + requestedByActorType: "user", requestedByActorId: "board", payload: { issueId: f.issueId, commentId: f.commentId }, + contextSnapshot: { issueId: f.issueId, wakeCommentId: f.commentId } }); + const [waiting] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId)); + expect(waiting.payload?.executionWait).toMatchObject({ reason: "process_running" }); + await db.update(heartbeatRuns).set({ processPid: 999999999 }).where(eq(heartbeatRuns.id, f.sourceRunId)); + const [stopped] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)); + expect(await recordNativeLocalProcessStop(db, stopped)).toBe(true); + await db.update(issueRecoveryActions).set({ status: "resolved", evidence: { runId: f.sourceRunId } }) + .where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + if (state === "deleted") await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, f.commentId)); + if (state === "different-author") await db.update(issueComments).set({ authorUserId: "someone-else" }).where(eq(issueComments.id, f.commentId)); + const makeDue = () => db.update(agentWakeupRequests).set({ updatedAt: new Date(0) }).where(eq(agentWakeupRequests.id, waiting.id)); + const holdId = randomUUID(); + await db.insert(issueTreeHolds).values({ id: holdId, companyId: f.companyId, rootIssueId: f.issueId, mode: "pause", status: "active" }); + await db.insert(issueTreeHoldMembers).values({ companyId: f.companyId, holdId, issueId: f.issueId, depth: 0, issueTitle: "Deploy", issueStatus: "blocked" }); + await makeDue(); + await heartbeatService(db).resumeExecutionWaitComments(); + const [held] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, waiting.id)); + expect(held.status).toBe("deferred_issue_execution"); + expect(held.payload?.executionWait).toMatchObject({ reason: "issue_tree_hold_active" }); + await db.update(issueTreeHolds).set({ status: "released" }).where(eq(issueTreeHolds.id, holdId)); + await makeDue(); + await Promise.all([heartbeatService(db).resumeExecutionWaitComments(), heartbeatService(db).resumeExecutionWaitComments()]); + const runs = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued"))); + expect(runs).toHaveLength(state === "available" ? 1 : 0); + const [after] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, waiting.id)); + expect(after.status).toBe(state === "available" ? "coalesced" : "deferred_issue_execution"); + if (state === "available") expect(after.runId).toBe(runs[0].id); + }); + it.each(["live", "remote", "provider_event"])("does not accept invalid local stop proof: %s", async kind => { const f = await seed(); if (kind === "remote") { diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 7b5bcba828..eac438cdd5 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -8615,7 +8615,13 @@ export async function persistHeartbeatRunProcessMetadata( : startedAt, updatedAt: new Date(), }) - .where(eq(heartbeatRuns.id, runId)) + // Late callbacks must not rewrite completed process history or revoke + // its stop evidence after asynchronous identity lookup. + .where(and( + eq(heartbeatRuns.id, runId), + eq(heartbeatRuns.status, "running"), + isNull(heartbeatRuns.finishedAt), + )) .returning() .then((rows) => rows[0] ?? null); if (run?.runtimeMode === "native") await appendHeartbeatRunEvent(tx as unknown as Db, { @@ -10028,10 +10034,7 @@ export function heartbeatService( sql`${issues.id}::text = ${agentWakeupRequests.payload}->>'issueId'`, eq(issues.assigneeAgentId, agentWakeupRequests.agentId))) .innerJoin(companies, and(eq(companies.id, issues.companyId), eq(companies.status, "active"))) - .where(and(exists(db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( - eq(issueRecoveryActions.companyId, issues.companyId), eq(issueRecoveryActions.sourceIssueId, issues.id), - executionBlockerPredicate(), - ))), eq(agentWakeupRequests.status, "deferred_issue_execution"), + .where(and(eq(agentWakeupRequests.status, "deferred_issue_execution"), eq(agentWakeupRequests.requestedByActorType, "user"), sql`${agentWakeupRequests.payload}->'executionWait' is not null`, lte(agentWakeupRequests.updatedAt, new Date(Date.now() - 30_000)), @@ -10053,7 +10056,26 @@ export function heartbeatService( // Match normal admission's deterministic current blocker selection. An // arbitrary historical action must not choose the retry's source run. const blocker = await getExecutionBlocker(db, wake.companyId, issueId); - const sourceId = blocker?.runId; + if (!blocker) { + // Cleanup may already have resolved the recovery action. The saved + // receipt remains work; normal admission rechecks ownership, holds, + // budgets and any blocker created since this read under the issue lock. + if (wake.idempotencyKey?.startsWith("chat-inbound:") || + !["issue_commented", "issue_reopened_via_comment"].includes(wake.reason ?? "")) continue; + const payload = parseObject(wake.payload); + const context = parseObject(payload[DEFERRED_WAKE_CONTEXT_KEY]); + await enqueueWakeup(wake.agentId, { + source: wake.source as WakeupOptions["source"], + triggerDetail: (wake.triggerDetail ?? undefined) as WakeupOptions["triggerDetail"], + reason: wake.reason, payload, contextSnapshot: context, + requestedByActorType: "user", requestedByActorId: wake.requestedByActorId, + idempotencyKey: `execution-wait-comment:${wake.id}`, + }, wake.id).catch(err => { + logger.warn({ err, requestId: wake.id }, "failed to resume saved execution-wait message after recovery"); + }); + continue; + } + const sourceId = blocker.runId; if (!sourceId || !isUuidLike(sourceId)) continue; const run = await getRun(sourceId); if (!run || run.companyId !== wake.companyId || run.agentId !== wake.agentId) continue; @@ -25505,6 +25527,14 @@ export function heartbeatService( if (!pending || !wakeCommentId || !queuedCommentIdsFromWakePayload(pending.payload).includes(wakeCommentId)) { return { kind: "deferred" as const }; } + const [comment] = await tx.select({ id: issueComments.id }).from(issueComments).where(and( + eq(issueComments.companyId, agent.companyId), eq(issueComments.issueId, issueId), + sql`${issueComments.id}::text = ${wakeCommentId}`, eq(issueComments.authorType, "user"), + eq(issueComments.authorUserId, opts.requestedByActorId ?? ""), + isNull(issueComments.deletedAt), isNull(issueComments.createdByRunId), + sql`length(trim(${issueComments.body})) > 0`, + )); + if (!comment) return { kind: "deferred" as const }; } let automaticParentRunId: string | null = null; if (