diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index b8d5a33bd5f..e1065ae4f18 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1910,6 +1910,54 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); }); + it("records operator interrupt cancellation metadata without changing terminal status", async () => { + const { runId, issueId } = await seedRunFixture({ + agentStatus: "running", + includeIssue: true, + }); + const heartbeat = heartbeatService(db); + + const cancelled = await heartbeat.cancelRun(runId, "Interrupted by board comment", { + errorCode: "operator_interrupted", + resultJson: { + operatorInterrupted: true, + interruptionSource: "issue_comment_interrupt", + interruptedIssueId: issueId, + }, + eventMessage: "run interrupted by board comment", + eventPayload: { + issueId, + source: "issue_comment_interrupt", + }, + }); + + expect(cancelled?.status).toBe("cancelled"); + expect(cancelled?.errorCode).toBe("operator_interrupted"); + expect(cancelled?.error).toBe("Interrupted by board comment"); + expect(cancelled?.resultJson).toMatchObject({ + stopReason: "cancelled", + operatorInterrupted: true, + interruptionSource: "issue_comment_interrupt", + interruptedIssueId: issueId, + }); + + const events = await db + .select() + .from(heartbeatRunEvents) + .where(eq(heartbeatRunEvents.runId, runId)); + expect(events).toHaveLength(1); + expect(events[0]).toMatchObject({ + eventType: "lifecycle", + stream: "system", + level: "warn", + message: "run interrupted by board comment", + payload: expect.objectContaining({ + issueId, + source: "issue_comment_interrupt", + }), + }); + }); + it("dispatches assigned todo work with no prior run as a normal assignment wake", async () => { const { companyId, agentId, issueId } = await seedAssignedTodoNoRunFixture(); const heartbeat = heartbeatService(db); diff --git a/server/src/__tests__/issue-comment-reopen-routes.test.ts b/server/src/__tests__/issue-comment-reopen-routes.test.ts index 4043d86a37b..a304e28a218 100644 --- a/server/src/__tests__/issue-comment-reopen-routes.test.ts +++ b/server/src/__tests__/issue-comment-reopen-routes.test.ts @@ -1281,7 +1281,23 @@ describe.sequential("issue comment reopen routes", () => { expect(res.status).toBe(200); expect(mockHeartbeatService.getRun).toHaveBeenCalledWith("run-1"); - expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("run-1"); + expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith( + "run-1", + "Interrupted by board comment", + expect.objectContaining({ + errorCode: "operator_interrupted", + resultJson: expect.objectContaining({ + operatorInterrupted: true, + interruptionSource: "issue_comment_interrupt", + interruptedIssueId: "11111111-1111-4111-8111-111111111111", + }), + eventMessage: "run interrupted by board comment", + eventPayload: expect.objectContaining({ + issueId: "11111111-1111-4111-8111-111111111111", + source: "issue_comment_interrupt", + }), + }), + ); expect(mockLogActivity).toHaveBeenCalledWith( expect.anything(), expect.objectContaining({ @@ -1289,6 +1305,8 @@ describe.sequential("issue comment reopen routes", () => { details: expect.objectContaining({ source: "issue_comment_interrupt", issueId: "11111111-1111-4111-8111-111111111111", + cancellationKind: "operator_interrupted", + operatorInterrupted: true, }), }), ); diff --git a/server/src/__tests__/issue-update-comment-wakeup-routes.test.ts b/server/src/__tests__/issue-update-comment-wakeup-routes.test.ts index 9ad4b0d7dc4..41b8c76a289 100644 --- a/server/src/__tests__/issue-update-comment-wakeup-routes.test.ts +++ b/server/src/__tests__/issue-update-comment-wakeup-routes.test.ts @@ -316,7 +316,23 @@ describe("issue update comment wakeups", () => { }); expect(res.status).toBe(200); - expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("run-1"); + expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith( + "run-1", + "Interrupted by board comment", + expect.objectContaining({ + errorCode: "operator_interrupted", + resultJson: expect.objectContaining({ + operatorInterrupted: true, + interruptionSource: "issue_comment_interrupt", + interruptedIssueId: existing.id, + }), + eventMessage: "run interrupted by board comment", + eventPayload: expect.objectContaining({ + issueId: existing.id, + source: "issue_comment_interrupt", + }), + }), + ); await vi.waitFor(() => expect(mockHeartbeatService.wakeup).toHaveBeenCalledTimes(1)); expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith( ASSIGNEE_AGENT_ID, @@ -386,7 +402,19 @@ describe("issue update comment wakeups", () => { }); expect(res.status).toBe(200); - expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("run-2"); + expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith( + "run-2", + "Interrupted by board comment", + expect.objectContaining({ + errorCode: "operator_interrupted", + resultJson: expect.objectContaining({ + operatorInterrupted: true, + interruptionSource: "issue_comment_interrupt", + interruptedIssueId: existing.id, + }), + eventMessage: "run interrupted by board comment", + }), + ); await vi.waitFor(() => expect(mockIssueService.findMentionedAgents).toHaveBeenCalledWith( existing.companyId, "stop here, I will take it", diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 6241b79f968..e600266e831 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -2106,6 +2106,26 @@ export function issueRoutes( return runToInterrupt?.status === "running" ? runToInterrupt : null; } + function operatorInterruptCancelOptions(input: { issueId: string; actor: ReturnType }) { + return { + errorCode: "operator_interrupted", + resultJson: { + operatorInterrupted: true, + interruptionSource: "issue_comment_interrupt", + interruptedIssueId: input.issueId, + interruptedByActorType: input.actor.actorType, + interruptedByActorId: input.actor.actorId, + }, + eventMessage: "run interrupted by board comment", + eventPayload: { + issueId: input.issueId, + source: "issue_comment_interrupt", + interruptedByActorType: input.actor.actorType, + interruptedByActorId: input.actor.actorId, + }, + }; + } + async function normalizeIssueAssigneeAgentReference( companyId: string, rawAssigneeAgentId: string | null | undefined, @@ -4761,7 +4781,11 @@ export function issueRoutes( const runToInterrupt = await resolveActiveIssueRun(existing); if (runToInterrupt) { - const cancelled = await heartbeat.cancelRun(runToInterrupt.id); + const cancelled = await heartbeat.cancelRun( + runToInterrupt.id, + "Interrupted by board comment", + operatorInterruptCancelOptions({ issueId: existing.id, actor }), + ); if (cancelled) { interruptedRunId = cancelled.id; await logActivity(db, { @@ -4773,7 +4797,13 @@ export function issueRoutes( action: "heartbeat.cancelled", entityType: "heartbeat_run", entityId: cancelled.id, - details: { agentId: cancelled.agentId, source: "issue_comment_interrupt", issueId: existing.id }, + details: { + agentId: cancelled.agentId, + source: "issue_comment_interrupt", + issueId: existing.id, + cancellationKind: "operator_interrupted", + operatorInterrupted: true, + }, }); } } @@ -6554,7 +6584,11 @@ export function issueRoutes( const runToInterrupt = await resolveActiveIssueRun(currentIssue); if (runToInterrupt) { - const cancelled = await heartbeat.cancelRun(runToInterrupt.id); + const cancelled = await heartbeat.cancelRun( + runToInterrupt.id, + "Interrupted by board comment", + operatorInterruptCancelOptions({ issueId: currentIssue.id, actor }), + ); if (cancelled) { interruptedRunId = cancelled.id; await logActivity(db, { @@ -6566,7 +6600,13 @@ export function issueRoutes( action: "heartbeat.cancelled", entityType: "heartbeat_run", entityId: cancelled.id, - details: { agentId: cancelled.agentId, source: "issue_comment_interrupt", issueId: currentIssue.id }, + details: { + agentId: cancelled.agentId, + source: "issue_comment_interrupt", + issueId: currentIssue.id, + cancellationKind: "operator_interrupted", + operatorInterrupted: true, + }, }); } } diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index e54bfdb4814..79a321e422d 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -10612,11 +10612,41 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return wakeupIds.length; } - async function cancelRunInternal(runId: string, reason = "Cancelled by control plane") { + type CancelRunOptions = { + errorCode?: string; + resultJson?: Record; + eventMessage?: string; + eventPayload?: Record; + }; + + async function cancelRunInternal(runId: string, reason = "Cancelled by control plane", options: CancelRunOptions = {}) { const run = await getRun(runId); if (!run) throw notFound("Heartbeat run not found"); if (!CANCELLABLE_HEARTBEAT_RUN_STATUSES.includes(run.status as (typeof CANCELLABLE_HEARTBEAT_RUN_STATUSES)[number])) return run; const agent = await getAgent(run.agentId); + const errorCode = options.errorCode ?? "cancelled"; + const resultJson = agent + ? { + ...mergeRunStopMetadataForAgent(agent, "cancelled", { + resultJson: parseObject(run.resultJson), + errorCode, + errorMessage: reason, + }), + ...(options.resultJson ?? {}), + } + : options.resultJson; + + const cancelled = await setRunStatus(run.id, "cancelled", { + finishedAt: new Date(), + error: reason, + errorCode, + ...(resultJson ? { resultJson } : {}), + }); + + await setWakeupStatus(run.wakeupRequestId, "cancelled", { + finishedAt: new Date(), + error: reason, + }); const running = runningProcesses.get(run.id); if (running) { @@ -10632,30 +10662,13 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }); } - const cancelled = await setRunStatus(run.id, "cancelled", { - finishedAt: new Date(), - error: reason, - errorCode: "cancelled", - ...(agent ? { - resultJson: mergeRunStopMetadataForAgent(agent, "cancelled", { - resultJson: parseObject(run.resultJson), - errorCode: "cancelled", - errorMessage: reason, - }), - } : {}), - }); - - await setWakeupStatus(run.wakeupRequestId, "cancelled", { - finishedAt: new Date(), - error: reason, - }); - if (cancelled) { await appendRunEvent(cancelled, 1, { eventType: "lifecycle", stream: "system", level: "warn", - message: "run cancelled", + message: options.eventMessage ?? "run cancelled", + ...(options.eventPayload ? { payload: options.eventPayload } : {}), }); await releaseIssueExecutionAndPromote(cancelled); } @@ -11100,7 +11113,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }; }, - cancelRun: (runId: string, reason?: string) => cancelRunInternal(runId, reason), + cancelRun: (runId: string, reason?: string, options?: CancelRunOptions) => cancelRunInternal(runId, reason, options), cancelActiveForAgent: (agentId: string, reason?: string) => cancelActiveForAgentInternal(agentId, reason),