From 349cdb6a91ed21cac1c38e7cffe5a1123763858e Mon Sep 17 00:00:00 2001 From: nguyenm7 <13559011+nguyenm7@users.noreply.github.com> Date: Tue, 22 Sep 2026 16:58:30 -0700 Subject: [PATCH] fix: recover missing continuation wakes on secret binding replay Retry continuation delivery for terminal receipts and share the durable interaction wake key across REST and gateway. Cover executed and failed receipts, with repeated replay leaving proposals and bindings unchanged. Co-Authored-By: Paperclip --- .../issue-thread-interaction-routes.test.ts | 48 +++++++++++++++++++ .../plugin-orchestration-apis.test.ts | 39 +++++++++++++++ server/src/routes/issues.ts | 19 ++++---- server/src/services/plugin-host-services.ts | 23 +++++---- 4 files changed, 111 insertions(+), 18 deletions(-) diff --git a/server/src/__tests__/issue-thread-interaction-routes.test.ts b/server/src/__tests__/issue-thread-interaction-routes.test.ts index c543172e079..5513f99652f 100644 --- a/server/src/__tests__/issue-thread-interaction-routes.test.ts +++ b/server/src/__tests__/issue-thread-interaction-routes.test.ts @@ -1543,6 +1543,54 @@ describe.sequential("issue thread interaction routes", () => { ); }); + it.each(["executed", "failed"] as const)("restores the continuation for a persisted %s receipt without executing the binding again", async (executionStatus) => { + const proposalId = "66666666-6666-4666-8666-666666666666"; + const acceptedWithoutReceipt = { + id: "interaction-secret-proposal-replay", + companyId: "company-1", + issueId: "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", + kind: "request_confirmation", + status: "accepted", + continuationPolicy: "wake_assignee", + requestedResolverPolicy: "human_only", + effectiveResolverPolicy: "human_only", + resolverPolicyProvenance: "explicit", + effectiveResolverPolicySource: "governed_action", + payload: { + version: 1, + prompt: "Create the binding?", + secretProposal: { + version: 1, + proposalId, + configPath: "access.NEW_ALIAS", + }, + }, + result: { version: 1, outcome: "accepted", secretProposal: { status: executionStatus } }, + }; + mockInteractionService.getForIssue.mockResolvedValueOnce(acceptedWithoutReceipt); + const approveSecretProposal = vi.fn().mockResolvedValue({ status: "approved" }); + const app = await createApp(undefined, { approveSecretProposal }); + + const res = await request(app) + .post("/api/issues/aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa/interactions/interaction-secret-proposal-replay/accept") + .send({}); + + expect(res.status).toBe(200); + expect(mockInteractionService.acceptInteraction).not.toHaveBeenCalled(); + expect(approveSecretProposal).not.toHaveBeenCalled(); + expect(mockInteractionService.recordSecretProposalExecutionResult).not.toHaveBeenCalled(); + expect(res.body.result.secretProposal).toMatchObject({ status: executionStatus }); + expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith( + ASSIGNEE_AGENT_ID, + expect.objectContaining({ + idempotencyKey: "interaction:interaction-secret-proposal-replay:accepted", + payload: expect.objectContaining({ + secretProposal: expect.objectContaining({ proposalId, executionStatus }), + }), + }), + ); + }); + it("records a failed secret-proposal execution and posts a thread comment", async () => { const proposalId = "55555555-5555-4555-8555-555555555555"; const approveSecretProposal = vi.fn().mockRejectedValue(new Error("binding failed")); diff --git a/server/src/__tests__/plugin-orchestration-apis.test.ts b/server/src/__tests__/plugin-orchestration-apis.test.ts index 2012617e583..34bbe200590 100644 --- a/server/src/__tests__/plugin-orchestration-apis.test.ts +++ b/server/src/__tests__/plugin-orchestration-apis.test.ts @@ -1332,6 +1332,45 @@ describeEmbeddedPostgres("plugin orchestration APIs", () => { .resolves.toMatchObject({ contextSnapshot: expect.objectContaining({ secretProposal: expect.objectContaining({ executionStatus: "failed" }) }) }); }); + it.each([false, true])("respondInteraction restores a terminal receipt's missing wake exactly once (failed=%s)", async (conflictingConfig) => { + const fixture = await seedGovernedSecretBinding({ conflictingConfig }); + const services = buildHostServices(db, "plugin-record-id", "paperclip.gateway", createEventBusStub(), undefined, { heartbeatRuntimeEnv: {} }); + const params = { + issueId: fixture.issueId, + interactionId: fixture.interactionId, + companyId: fixture.companyId, + action: "accept" as const, + actorUserId: fixture.actorUserId, + }; + // Produce a real terminal execution receipt without its continuation, + // matching the persisted state after a crash between receipt and wake. + await db.update(issueThreadInteractions).set({ continuationPolicy: "none" }) + .where(eq(issueThreadInteractions.id, fixture.interactionId)); + const first = await services.issues.respondInteraction(params); + const executionStatus = conflictingConfig ? "failed" : "executed"; + expect(first.interaction.result).toMatchObject({ secretProposal: { status: executionStatus } }); + const proposalsBeforeReplay = await db.select().from(companySecretProposals); + const bindingsBeforeReplay = await db.select().from(companySecretBindings); + const wakeKey = `interaction:${fixture.interactionId}:accepted`; + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.idempotencyKey, wakeKey))).toHaveLength(0); + await db.update(issueThreadInteractions).set({ continuationPolicy: "wake_assignee" }) + .where(eq(issueThreadInteractions.id, fixture.interactionId)); + + const replay = await services.issues.respondInteraction(params); + await services.issues.respondInteraction(params); + + expect(replay).toMatchObject({ applied: false, interaction: { result: { secretProposal: { status: executionStatus } } } }); + await waitForSecretProposalWake(fixture.companyId, fixture.agentId, executionStatus); + const wakes = await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, fixture.companyId), + eq(agentWakeupRequests.idempotencyKey, wakeKey), + )); + expect(wakes).toHaveLength(1); + expect(wakes[0]?.payload).toMatchObject({ secretProposal: { executionStatus } }); + expect(await db.select().from(companySecretProposals)).toEqual(proposalsBeforeReplay); + expect(await db.select().from(companySecretBindings)).toEqual(bindingsBeforeReplay); + }); + it("respondInteraction idempotently repairs an accepted card that has no execution receipt", async () => { const fixture = await seedGovernedSecretBinding({ acceptedWithoutReceipt: true }); const services = buildHostServices( diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 99ee2d76763..76f8f8fbf54 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -15991,15 +15991,18 @@ export function issueRoutes( executionDisposition: replay.disposition, }, }); - await queueResolvedInteractionContinuationWakeup({ - db, - heartbeat, - issue, - interaction: replay.interaction, - actor, - source: "issue.interaction.accept.reconcile", - }); } + // Receipt persistence can precede a crash before the wake is queued. + // Reconcile delivery even for terminal receipts; the stable interaction + // idempotency key deduplicates a continuation that already exists. + await queueResolvedInteractionContinuationWakeup({ + db, + heartbeat, + issue, + interaction: replay.interaction, + actor, + source: "issue.interaction.accept.reconcile", + }); res.json(replay.interaction); return; } diff --git a/server/src/services/plugin-host-services.ts b/server/src/services/plugin-host-services.ts index 062e81d08dd..fe976857a48 100644 --- a/server/src/services/plugin-host-services.ts +++ b/server/src/services/plugin-host-services.ts @@ -927,7 +927,7 @@ export function buildHostServices( * cards never produce. Failure-tolerant: a wake failure is logged, never * thrown back to the plugin (the decision itself already applied). */ - const queuePluginInteractionContinuationWakeup = (args: { + const queuePluginInteractionContinuationWakeup = async (args: { issue: { id: string; assigneeAgentId: string | null; status: string }; interaction: { id: string; @@ -941,7 +941,7 @@ export function buildHostServices( }; actorUserId: string; source: string; - }): void => { + }): Promise => { const { interaction, issue } = args; if ( interaction.continuationPolicy !== "wake_assignee" @@ -975,7 +975,7 @@ export function buildHostServices( } const secretProposal = readSecretProposalContinuationContext(interaction); - void heartbeat.wakeup(issue.assigneeAgentId, { + await heartbeat.wakeup(issue.assigneeAgentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", @@ -990,6 +990,7 @@ export function buildHostServices( ...(secretProposal ? { secretProposal } : {}), mutation: "interaction", }, + ...(secretProposal ? { idempotencyKey: `interaction:${interaction.id}:${interaction.status}` } : {}), requestedByActorType: "user", requestedByActorId: args.actorUserId, contextSnapshot: { @@ -2572,13 +2573,15 @@ export function buildHostServices( executionDisposition: replay.disposition, }, }); - queuePluginInteractionContinuationWakeup({ - issue, - interaction: replay.interaction as typeof current, - actorUserId: params.actorUserId, - source: `plugin:${pluginKey}:interaction.accept.reconcile`, - }); } + // A terminal receipt does not prove its continuation was queued. + // Use the same durable interaction key as REST for safe replay. + await queuePluginInteractionContinuationWakeup({ + issue, + interaction: replay.interaction as typeof current, + actorUserId: params.actorUserId, + source: `plugin:${pluginKey}:interaction.accept.reconcile`, + }); return { interaction: replay.interaction as any, applied: false }; } return { interaction: current as any, applied: false }; @@ -2649,7 +2652,7 @@ export function buildHostServices( }, }); - queuePluginInteractionContinuationWakeup({ + await queuePluginInteractionContinuationWakeup({ issue: continuationTarget, interaction: resolved, actorUserId: params.actorUserId,