Implement backend interrupt handoff semantics

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-06-08 20:21:21 +00:00
1 parent 99607ff5c0
commit 189a4e1430
5 files changed
+175 -28

No files matched your search

@@ -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);
@@ -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,
}),
}),
);
@@ -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",
+44 -4
View File
@@ -2106,6 +2106,26 @@ export function issueRoutes(
return runToInterrupt?.status === "running" ? runToInterrupt : null;
}
function operatorInterruptCancelOptions(input: { issueId: string; actor: ReturnType<typeof getActorInfo> }) {
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,
},
});
}
}
+34 -21
View File
@@ -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<string, unknown>;
eventMessage?: string;
eventPayload?: Record<string, unknown>;
};
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),