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,