diff --git a/server/src/__tests__/approval-routes-idempotency.test.ts b/server/src/__tests__/approval-routes-idempotency.test.ts index 656aa152501..426603feda2 100644 --- a/server/src/__tests__/approval-routes-idempotency.test.ts +++ b/server/src/__tests__/approval-routes-idempotency.test.ts @@ -260,6 +260,48 @@ describe("approval routes idempotent retries", () => { expect(mockApprovalService.reject).toHaveBeenCalledWith("approval-5", "user-1", "not now"); }); + it("wakes the requesting agent when an approval is rejected", async () => { + mockApprovalService.getById.mockResolvedValue({ + id: "approval-7", + companyId: "company-1", + type: "request_board_approval", + status: "pending", + payload: {}, + requestedByAgentId: "agent-1", + }); + mockApprovalService.reject.mockResolvedValue({ + approval: { + id: "approval-7", + companyId: "company-1", + type: "request_board_approval", + status: "rejected", + payload: {}, + requestedByAgentId: "agent-1", + }, + applied: true, + }); + mockIssueApprovalService.listIssuesForApproval.mockResolvedValue([{ id: "issue-1" }, { id: "issue-2" }]); + + const res = await request(await createApp()) + .post("/api/approvals/approval-7/reject") + .send({ decisionNote: "not approved" }); + + expect(res.status).toBe(200); + expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith("agent-1", expect.objectContaining({ + reason: "approval_rejected", + payload: expect.objectContaining({ + approvalId: "approval-7", + approvalStatus: "rejected", + issueId: "issue-1", + issueIds: ["issue-1", "issue-2"], + }), + contextSnapshot: expect.objectContaining({ + wakeReason: "approval_rejected", + taskId: "issue-1", + }), + })); + }); + it("derives approval attribution from the authenticated actor on request revision", async () => { mockApprovalService.getById.mockResolvedValue({ id: "approval-6", diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index c5c2b9e676e..e9f10739cb5 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -7,6 +7,7 @@ import { agents, agentRuntimeState, agentWakeupRequests, + approvals, budgetPolicies, companySkills, companies, @@ -17,6 +18,7 @@ import { heartbeatRunEvents, heartbeatRuns, issueComments, + issueApprovals, issueDocuments, issueRelations, issueThreadInteractions, @@ -312,15 +314,18 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { await db.delete(costEvents); await db.delete(issueComments); await db.delete(issueThreadInteractions); + await db.delete(issueApprovals); await db.delete(issueDocuments); await db.delete(documentRevisions); await db.delete(documents); + await db.delete(approvals); await db.delete(issueRelations); await db.delete(issueTreeHoldMembers); await db.delete(issueTreeHolds); for (let attempt = 0; attempt < 5; attempt += 1) { await db.delete(issueComments); await db.delete(issueThreadInteractions); + await db.delete(issueApprovals); await db.delete(issueDocuments); try { await db.delete(issues); @@ -343,6 +348,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { } await db.delete(agentWakeupRequests); await db.delete(budgetPolicies); + await db.delete(approvals); for (let attempt = 0; attempt < 5; attempt += 1) { await db.delete(agentRuntimeState); try { @@ -2401,7 +2407,10 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { issueId, kind: "request_confirmation", status: "pending", - payload: {}, + payload: { + version: 1, + prompt: "Approve the plan?", + }, createdByAgentId: agentId, sourceRunId: runId, }); @@ -2431,6 +2440,219 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(refreshedRun?.livenessReason).toContain("continuation suppressed by explicit interaction or approval wait"); }); + it("does not re-dispatch assigned work while a live pending confirmation owns the next step", async () => { + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "todo", + runStatus: "failed", + }); + await db.insert(issueThreadInteractions).values({ + companyId, + issueId, + kind: "request_confirmation", + status: "pending", + payload: { + version: 1, + prompt: "Approve the next action?", + }, + createdByAgentId: agentId, + sourceRunId: runId, + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.dispatchRequeued).toBe(0); + expect(result.continuationRequeued).toBe(0); + expect(result.skipped).toBeGreaterThan(0); + + const wakeups = await db + .select() + .from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "issue_assignment_recovery"))); + expect(wakeups).toHaveLength(0); + }); + + it("expires a stale confirmation target and then queues bounded liveness continuation", async () => { + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "succeeded", + livenessState: "plan_only", + livenessReason: "Run produced a plan and needs approval", + nextAction: "Continue from the current plan revision.", + }); + const documentId = randomUUID(); + const oldRevisionId = randomUUID(); + const currentRevisionId = randomUUID(); + const interactionId = randomUUID(); + await db.insert(documents).values({ + id: documentId, + companyId, + title: "Plan", + format: "markdown", + latestBody: "current plan", + latestRevisionId: currentRevisionId, + latestRevisionNumber: 2, + }); + await db.insert(issueDocuments).values({ + companyId, + issueId, + documentId, + key: "plan", + }); + await db.insert(documentRevisions).values([ + { + id: oldRevisionId, + companyId, + documentId, + revisionNumber: 1, + title: "Plan", + format: "markdown", + body: "old plan", + }, + { + id: currentRevisionId, + companyId, + documentId, + revisionNumber: 2, + title: "Plan", + format: "markdown", + body: "current plan", + }, + ]); + await db.insert(issueThreadInteractions).values({ + id: interactionId, + companyId, + issueId, + kind: "request_confirmation", + status: "pending", + continuationPolicy: "wake_assignee_on_accept", + payload: { + version: 1, + prompt: "Approve the plan?", + target: { + type: "issue_document", + issueId, + documentId, + key: "plan", + revisionId: oldRevisionId, + revisionNumber: 1, + }, + }, + createdByAgentId: agentId, + sourceRunId: runId, + }); + const heartbeat = heartbeatService(db); + const sourceRun = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then((rows) => rows[0] ?? null); + if (!sourceRun) throw new Error("Expected source run"); + + await heartbeat.handleRunLivenessContinuationForTest(sourceRun); + + const [interaction] = await db + .select() + .from(issueThreadInteractions) + .where(eq(issueThreadInteractions.id, interactionId)); + expect(interaction?.status).toBe("expired"); + expect((interaction?.result as Record | null)?.outcome).toBe("stale_target"); + + const wakeups = await db + .select() + .from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "run_liveness_continuation"))); + expect(wakeups).toHaveLength(1); + }); + + it("expires a confirmation superseded by a user comment and then queues bounded liveness continuation", async () => { + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "succeeded", + livenessState: "plan_only", + livenessReason: "Run asked the board to confirm the next step", + nextAction: "Continue after incorporating the board comment.", + }); + const interactionId = randomUUID(); + await db.insert(issueThreadInteractions).values({ + id: interactionId, + companyId, + issueId, + kind: "request_confirmation", + status: "pending", + payload: { + version: 1, + prompt: "Approve the next step?", + supersedeOnUserComment: true, + }, + createdByAgentId: agentId, + sourceRunId: runId, + createdAt: new Date("2026-03-19T00:06:00.000Z"), + }); + const commentId = randomUUID(); + await db.insert(issueComments).values({ + id: commentId, + companyId, + issueId, + authorUserId: "board", + body: "Use the narrower implementation path.", + createdAt: new Date("2026-03-19T00:07:00.000Z"), + }); + const heartbeat = heartbeatService(db); + const sourceRun = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then((rows) => rows[0] ?? null); + if (!sourceRun) throw new Error("Expected source run"); + + await heartbeat.handleRunLivenessContinuationForTest(sourceRun); + + const [interaction] = await db + .select() + .from(issueThreadInteractions) + .where(eq(issueThreadInteractions.id, interactionId)); + expect(interaction?.status).toBe("expired"); + expect(interaction?.result).toMatchObject({ + outcome: "superseded_by_comment", + commentId, + }); + + const wakeups = await db + .select() + .from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "run_liveness_continuation"))); + expect(wakeups).toHaveLength(1); + }); + + it("queues bounded liveness continuation once a linked approval is resolved", async () => { + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "succeeded", + livenessState: "plan_only", + livenessReason: "Run requested approval before continuing", + nextAction: "Continue after the approval decision.", + }); + const approvalId = randomUUID(); + await db.insert(approvals).values({ + id: approvalId, + companyId, + type: "request_board_approval", + requestedByAgentId: agentId, + status: "rejected", + payload: {}, + decidedByUserId: "board", + decidedAt: new Date("2026-03-19T00:06:00.000Z"), + }); + await db.insert(issueApprovals).values({ + companyId, + issueId, + approvalId, + }); + const heartbeat = heartbeatService(db); + const sourceRun = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then((rows) => rows[0] ?? null); + if (!sourceRun) throw new Error("Expected source run"); + + await heartbeat.handleRunLivenessContinuationForTest(sourceRun); + + const wakeups = await db + .select() + .from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "run_liveness_continuation"))); + expect(wakeups).toHaveLength(1); + }); + it("queues a bounded liveness continuation when no pause hold suppresses it", async () => { const { companyId, runId } = await seedStrandedIssueFixture({ status: "in_progress", diff --git a/server/src/routes/approvals.ts b/server/src/routes/approvals.ts index 9c837ba124a..d81bea6467b 100644 --- a/server/src/routes/approvals.ts +++ b/server/src/routes/approvals.ts @@ -240,6 +240,10 @@ export function approvalRoutes( const { approval, applied } = await svc.reject(id, decidedByUserId, req.body.decisionNote); if (applied) { + const linkedIssues = await issueApprovalsSvc.listIssuesForApproval(approval.id); + const linkedIssueIds = linkedIssues.map((issue) => issue.id); + const primaryIssueId = linkedIssueIds[0] ?? null; + await logActivity(db, { companyId: approval.companyId, actorType: "user", @@ -247,8 +251,75 @@ export function approvalRoutes( action: "approval.rejected", entityType: "approval", entityId: approval.id, - details: { type: approval.type }, + details: { + type: approval.type, + requestedByAgentId: approval.requestedByAgentId, + linkedIssueIds, + }, }); + + if (approval.requestedByAgentId) { + try { + const wakeRun = await heartbeat.wakeup(approval.requestedByAgentId, { + source: "automation", + triggerDetail: "system", + reason: "approval_rejected", + payload: { + approvalId: approval.id, + approvalStatus: approval.status, + issueId: primaryIssueId, + issueIds: linkedIssueIds, + }, + requestedByActorType: "user", + requestedByActorId: req.actor.userId ?? "board", + contextSnapshot: { + source: "approval.rejected", + approvalId: approval.id, + approvalStatus: approval.status, + issueId: primaryIssueId, + issueIds: linkedIssueIds, + taskId: primaryIssueId, + wakeReason: "approval_rejected", + }, + }); + + await logActivity(db, { + companyId: approval.companyId, + actorType: "user", + actorId: req.actor.userId ?? "board", + action: "approval.requester_wakeup_queued", + entityType: "approval", + entityId: approval.id, + details: { + requesterAgentId: approval.requestedByAgentId, + wakeRunId: wakeRun?.id ?? null, + linkedIssueIds, + }, + }); + } catch (err) { + logger.warn( + { + err, + approvalId: approval.id, + requestedByAgentId: approval.requestedByAgentId, + }, + "failed to queue requester wakeup after approval rejection", + ); + await logActivity(db, { + companyId: approval.companyId, + actorType: "user", + actorId: req.actor.userId ?? "board", + action: "approval.requester_wakeup_failed", + entityType: "approval", + entityId: approval.id, + details: { + requesterAgentId: approval.requestedByAgentId, + linkedIssueIds, + error: err instanceof Error ? err.message : String(err), + }, + }); + } + } } res.json(redactApprovalPayload(approval)); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 3faacafc388..25800282bac 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -80,6 +80,7 @@ import { sanitizeRuntimeServiceBaseEnv, } from "./workspace-runtime.js"; import { issueService } from "./issues.js"; +import { issueThreadInteractionService } from "./issue-thread-interactions.js"; import { parseIssueExecutionState } from "./issue-execution-policy.js"; import { ISSUE_TREE_CONTROL_INTERACTION_WAKE_REASONS, @@ -2774,6 +2775,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } async function hasExplicitIssueWait(companyId: string, issueId: string) { + await issueThreadInteractionService(db).expireInvalidPendingRequestConfirmationsForIssue( + { id: issueId, companyId }, + { agentId: null, userId: null }, + ); + const [interaction, approval] = await Promise.all([ db .select({ id: issueThreadInteractions.id }) diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index ac26df38f19..b5e216160fa 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -1,5 +1,5 @@ import { isDeepStrictEqual } from "node:util"; -import { and, asc, eq, inArray } from "drizzle-orm"; +import { and, asc, eq, gt, inArray, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { documents, @@ -463,6 +463,55 @@ async function expireStaleRequestConfirmationTarget(db: Db | any, args: { return hydrateInteraction(updated); } +async function expireRequestConfirmationSupersededByLatestUserComment(db: Db | any, args: { + row: IssueThreadInteractionRow; + actor: InteractionActor; +}): Promise { + if (args.row.kind !== "request_confirmation" || args.row.status !== "pending") return null; + const interaction = hydrateInteraction(args.row) as RequestConfirmationInteraction; + if (interaction.payload.supersedeOnUserComment !== true) return null; + + const [comment] = await db + .select({ id: issueComments.id }) + .from(issueComments) + .where(and( + eq(issueComments.companyId, args.row.companyId), + eq(issueComments.issueId, args.row.issueId), + gt(issueComments.createdAt, args.row.createdAt), + sql`${issueComments.authorUserId} is not null`, + )) + .orderBy(asc(issueComments.createdAt)) + .limit(1); + if (!comment) return null; + + const now = new Date(); + const [updated] = await db + .update(issueThreadInteractions) + .set({ + status: "expired", + result: { + version: 1, + outcome: "superseded_by_comment", + commentId: comment.id, + }, + resolvedByAgentId: args.actor.agentId ?? null, + resolvedByUserId: args.actor.userId ?? null, + resolvedAt: now, + updatedAt: now, + }) + .where(and( + eq(issueThreadInteractions.id, args.row.id), + eq(issueThreadInteractions.status, "pending"), + )) + .returning(); + + if (!updated) { + throw conflict("Interaction has already been resolved"); + } + await touchIssue(db, args.row.issueId); + return hydrateInteraction(updated); +} + export function issueThreadInteractionService(db: Db) { async function getIdempotentInteraction(args: { issueId: string; @@ -1133,6 +1182,33 @@ export function issueThreadInteractionService(db: Db) { return expired; }, + expireInvalidPendingRequestConfirmationsForIssue: async ( + issue: { id: string; companyId: string }, + actor: InteractionActor, + ) => { + const rows = await db + .select() + .from(issueThreadInteractions) + .where(and( + eq(issueThreadInteractions.companyId, issue.companyId), + eq(issueThreadInteractions.issueId, issue.id), + eq(issueThreadInteractions.kind, "request_confirmation"), + eq(issueThreadInteractions.status, "pending"), + )); + + const expired: IssueThreadInteraction[] = []; + for (const row of rows) { + const staleTarget = await expireStaleRequestConfirmationTarget(db, { row, actor }); + if (staleTarget) { + expired.push(staleTarget); + continue; + } + const superseded = await expireRequestConfirmationSupersededByLatestUserComment(db, { row, actor }); + if (superseded) expired.push(superseded); + } + return expired; + }, + answerQuestions: async ( issue: { id: string; companyId: string }, interactionId: string, diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index c1051346479..51423ed0c22 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -32,6 +32,7 @@ import { budgetService } from "../budgets.js"; import { instanceSettingsService } from "../instance-settings.js"; import { issueTreeControlService } from "../issue-tree-control.js"; import { issueService } from "../issues.js"; +import { issueThreadInteractionService } from "../issue-thread-interactions.js"; import { getRunLogStore } from "../run-log-store.js"; import { RECOVERY_ORIGIN_KINDS, @@ -398,6 +399,10 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) if (issue.status === "done" || issue.status === "cancelled") return true; if (issue.status === "blocked" || issue.status === "in_review") return true; if (issue.executionState) return true; + await issueThreadInteractionService(db).expireInvalidPendingRequestConfirmationsForIssue( + { id: issue.id, companyId: issue.companyId }, + { agentId: null, userId: null }, + ); const [interaction, approval, recoveryIssue] = await Promise.all([ db @@ -1793,6 +1798,11 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) continue; } + if (await hasExplicitWaitingPath(issue)) { + result.skipped += 1; + continue; + } + const latestRun = await getLatestIssueRun(issue.companyId, issue.id); if (isStrandedIssueRecoveryIssue(issue) && isUnsuccessfulTerminalIssueRun(latestRun)) { const updated = await escalateStrandedRecoveryIssueInPlace({