diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index b83ae20c147..0da0511c1f7 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1771,7 +1771,7 @@ turn is never replayed. Activity records the request as `restart-recovery` on behalf of that user. Admission keeps every gate: saved-work recovery, stop proof, approvals, pauses, budget and ownership. A Retry that the server requested gets no second automatic attempt. A system wake (for example a -monitor or an automatic assignment) continues this way only with a positive +monitor) continues this way only with a positive record that no turn was ever issued: the refused reattachment proved the Runner process gone, and the run's controller journal, which records each turn command before it is sent, holds no turn command and no turn event @@ -1779,11 +1779,15 @@ before it is sent, holds no turn command and no turn event must also have no turn or tool event, and the Retry runs under the run's responsible user. Missing events alone never qualify, because a turn can reach the provider before its events are stored. An automatic retry of owed work (a -system wake whose run has `retryOfRunId`, such as `transient_failure_retry`) -also continues under the run's responsible user once the refusal proved its -Runner process gone, because losing the Runner to a restart is not the task's -failure. Other system wakes, and in-flight provider turns without that proof, -do not continue automatically. For these failed runs, a +system wake whose run has `retryOfRunId`, such as `transient_failure_retry`, a +system assignment such as a routine's dispatch (`issue_assigned`), or a comment +or external chat message wake whose message is still on the task and was not +written by a run) also continues under the run's responsible user once the +refusal proved its Runner process gone, because losing the Runner to a restart +is not the task's failure. A linked person's external chat message continues +under that person once the message is still on the task. Monitors and other +system wakes, and in-flight provider turns without that proof, do not continue +automatically. Each chain of runs gets at most one such automatic Retry. For these failed runs, a recovered workspace hold offers Retry (`canRetry`) and says "Retry or send a message to continue." An intentional Stop keeps "send a message" as its only continuation. diff --git a/server/src/app.ts b/server/src/app.ts index 11e12eaefd8..9d1a3960523 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -226,6 +226,7 @@ import { chatInboundWakeNoticeWork } from "./services/chat-inbound-wake-notice-w import { chatFailedRetryNoticeWork } from "./services/chat-failed-retry-notice-work.js"; import { chatRunMilestoneWork } from "./services/chat-run-milestone-work.js"; import { heartbeatService } from "./services/heartbeat.js"; +import { registerInterruptedTurnChatRetry } from "./services/heartbeat/restart-interrupted-turn.js"; import { pluginLifecycleManager } from "./services/plugin-lifecycle.js"; import { registerPluginLifecycleDelivery } from "./services/plugin-lifecycle-delivery.js"; import { createPluginJobCoordinator } from "./services/plugin-job-coordinator.js"; @@ -603,6 +604,8 @@ export async function createApp( storage: opts.storageService, }); if (opts.budgetResumeDelivery) opts.budgetResumeDelivery.prepareChatRetry = chatChannels.prepareFailedChatRunRetry; + // A restart's automatic Retry of a connected conversation needs the same callback. + registerInterruptedTurnChatRetry(chatChannels.prepareFailedChatRunRetry); // Voice capabilities are verified by the signed callback route, not by // board/agent bearer authentication. Mount before actorMiddleware so its // per-session Bearer cannot be mistaken for an agent API key. diff --git a/server/src/services/heartbeat/restart-interrupted-turn.ts b/server/src/services/heartbeat/restart-interrupted-turn.ts index 628a2840974..533d7b3cdc7 100644 --- a/server/src/services/heartbeat/restart-interrupted-turn.ts +++ b/server/src/services/heartbeat/restart-interrupted-turn.ts @@ -4,6 +4,7 @@ import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import type { NativeRestartRecoveryClaim } from "../native-runtime/native-restart-recovery.js"; import { isUserCommentContinuationReason } from "../explicit-native-continuation.js"; import { requestTaskRetry } from "../task-retry.js"; +import { EXTERNAL_CHAT_MESSAGE_WAKE_REASON } from "../queued-comment-delivery.js"; import { HttpError } from "../../errors.js"; import { logger } from "../../middleware/logger.js"; import { NATIVE_REMOTE_RECOVERY_LEASE_MISMATCH } from "../native-runtime/remote-recovery-lease-mismatch.js"; @@ -107,7 +108,30 @@ async function chainUsedAutomaticRetry(db: Db, run: { companyId: string; id: str return visited.size >= MAX_RETRY_CHAIN; } -type Outcome = { requested: true; requestId: string } | { requested: false; reason: string }; +type PrepareChatRetry = NonNullable[0]["prepareChatRetry"]>; + +/** A connected conversation's Retry goes through the chat service, which the + * app registers at startup. Without it, a connected task keeps its Retry. */ +let interruptedTurnChatRetry: PrepareChatRetry | undefined; +export function registerInterruptedTurnChatRetry(prepare: PrepareChatRetry | undefined) { + interruptedTurnChatRetry = prepare; +} + +/** The chat or comment message that woke the run is still on the task and was + * not written by a run, so the run still owes it an answer. */ +async function wakeMessageStillOwed(db: Db, companyId: string, issueId: string, reason: string | null | undefined, + context: Record, wake: { payload: Record | null }) { + if (reason !== EXTERNAL_CHAT_MESSAGE_WAKE_REASON && !isUserCommentContinuationReason(reason)) return false; + const commentId = typeof context.wakeCommentId === "string" ? context.wakeCommentId + : typeof wake.payload?.commentId === "string" ? wake.payload.commentId : null; + if (!commentId) return false; + const [comment] = await db.select({ id: issueComments.id }).from(issueComments).where(and( + eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId), + sql`${issueComments.id}::text = ${commentId}`, isNull(issueComments.deletedAt), isNull(issueComments.createdByRunId))); + return comment !== undefined; +} + +type Outcome ={ requested: true; requestId: string } | { requested: false; reason: string }; /** * Reattachment runs after a server restart and for a same-run retry of a remote @@ -118,9 +142,9 @@ type Outcome = { requested: true; requestId: string } | { requested: false; reas * proven gone; an in-flight turn's outcome is then unknown and recorded, and the * Retry's fresh session never replays it. Without that proof the in-flight turn * keeps the visible Retry. A system wake continues this way only when the - * controller journal proved that no turn was ever issued, or when it was an - * automatic retry of owed work whose Runner is proven gone, under the run's - * responsible user. Admission keeps every gate, including stop proof and + * controller journal proved that no turn was ever issued, or when it carried + * owed work (an automatic retry, a system assignment, or a chat message still + * on the task) whose Runner is proven gone, under the run's responsible user. Admission keeps every gate, including stop proof and * saved-work recovery before the next turn. */ export async function requestRestartInterruptedTurnRetry(db: Db, claim: NativeRestartRecoveryClaim): Promise { @@ -155,7 +179,12 @@ export async function requestInterruptedNativeTurnRetry(db: Db, runId: string): // ever issued: nothing the run did can be repeated, so the Retry runs once // under the run's responsible user. Missing events alone prove nothing, // because a turn can reach the provider before its events are stored. - const owedRedelivery = run.retryOfRunId !== null && runnerExit !== null; + // Work the wake still owes the task also continues once the Runner is + // proven gone: an automatic retry of earlier owed work, a system + // assignment (a routine's dispatch), or a chat message that is still on + // the task. Monitors and other system wakes keep their policy. + const owedRedelivery = runnerExit !== null && wake !== undefined && (run.retryOfRunId !== null + || reason === "issue_assigned" || await wakeMessageStillOwed(db, run.companyId, issueId, reason, context, wake)); if (!wake || reason === "task_retry" || !run.responsibleUserId || !(owedRedelivery || await hasProvenNoProviderTurn(db, run, runnerExit))) return { requested: false, reason: "not_user_request" }; @@ -177,6 +206,11 @@ export async function requestInterruptedNativeTurnRetry(db: Db, runId: string): eq(issueComments.authorUserId, userActorId), isNull(issueComments.deletedAt), isNull(issueComments.createdByRunId), )) : []; if (!comment && reason !== "issue_assigned") return { requested: false, reason: "message_unavailable" }; + } else if (reason === EXTERNAL_CHAT_MESSAGE_WAKE_REASON) { + // A linked person's chat message is stored by the chat bridge, not always + // as that person's own comment, so the message itself must still be there. + if (!await wakeMessageStillOwed(db, run.companyId, issueId, reason, context, wake!)) + return { requested: false, reason: "message_unavailable" }; } else return { requested: false, reason: "not_user_request" }; if (!actorId) return { requested: false, reason: "not_user_request" }; const [issue] = await db.select({ status: issues.status, assigneeAgentId: issues.assigneeAgentId }).from(issues) @@ -187,7 +221,8 @@ export async function requestInterruptedNativeTurnRetry(db: Db, runId: string): try { const receipt = await requestTaskRetry({ db, companyId: run.companyId, issueId, actorId, actorSource: actorId === "local-board" ? "local_implicit" : "session", - idempotencyKey: `${RESTART_RETRY_KEY_PREFIX}${run.id}`, observedRunId: run.id, initiatedBy: "restart_recovery" }); + idempotencyKey: `${RESTART_RETRY_KEY_PREFIX}${run.id}`, observedRunId: run.id, initiatedBy: "restart_recovery", + prepareChatRetry: interruptedTurnChatRetry }); return receipt.accepted && receipt.requestId ? { requested: true, requestId: receipt.requestId } : { requested: false, reason: "retry_stale" }; } catch (error) { diff --git a/server/src/services/runner-lost-mid-turn-continuation.test.ts b/server/src/services/runner-lost-mid-turn-continuation.test.ts index 73c942bf518..83d4eaf4110 100644 --- a/server/src/services/runner-lost-mid-turn-continuation.test.ts +++ b/server/src/services/runner-lost-mid-turn-continuation.test.ts @@ -4,10 +4,12 @@ import path from "node:path"; import { and, eq } from "drizzle-orm"; import { describe, expect, it, vi } from "vitest"; import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; -import { agentWakeupRequests, heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db"; +import { agentWakeupRequests, chatConversations, chatEndpoints, heartbeatRunEvents, heartbeatRuns, issueComments, + toolApplications, toolConnections } from "@paperclipai/db"; import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; import { heartbeatService } from "./heartbeat.js"; -import { requestInterruptedNativeTurnRetry, requestRestartInterruptedTurnRetry } from "./heartbeat/restart-interrupted-turn.js"; +import { registerInterruptedTurnChatRetry, requestInterruptedNativeTurnRetry, requestRestartInterruptedTurnRetry } + from "./heartbeat/restart-interrupted-turn.js"; import { readSystemRecovery } from "./system-recovery-state.js"; import { reconcileRetainedWorkspaces } from "./workspace-recovery.js"; import { useWorkspaceRecoveryFixture, action, db, run, seed } from "./__tests__/workspace-recovery-fixture.js"; @@ -31,14 +33,24 @@ const stagingRefusal = (reason: "process_not_running" | "process_identity_change describe("a remote Runner that died in the middle of a provider turn", () => { useWorkspaceRecoveryFixture(targets); - async function runnerLostMidTurn(input: { wake: "user" | "user_retry" | "system_retry" | "system_monitor"; + async function runnerLostMidTurn(input: { wake: "user" | "user_retry" | "system_retry" | "system_monitor" + | "system_assignment" | "system_chat" | "user_chat"; refusal?: "process_not_running" | "process_identity_changed" | null; + chatMessage?: "present" | "deleted" | "written_by_run"; events?: string[]; refusalPayload?: Record }) { const f = await seed("native"); const original = await run(f.runId); - const system = input.wake !== "user" && input.wake !== "user_retry"; - const reason = input.wake === "user" ? "issue_assigned" : input.wake === "user_retry" ? "task_retry" + const system = input.wake !== "user" && input.wake !== "user_retry" && input.wake !== "user_chat"; + const chat = input.wake === "system_chat" || input.wake === "user_chat"; + const reason = input.wake === "user" || input.wake === "system_assignment" ? "issue_assigned" + : input.wake === "user_retry" ? "task_retry" + : chat ? "External chat message received" : input.wake === "system_retry" ? "transient_failure_retry" : "issue_monitor_due"; + // The chat bridge stores the inbound message as a task comment before it wakes the agent. + const [chatComment] = chat ? await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + body: "Can you also check the release notes?", authorType: "system", + ...(input.chatMessage === "deleted" ? { deletedAt: new Date() } : {}), + ...(input.chatMessage === "written_by_run" ? { createdByRunId: f.runId } : {}) }).returning() : []; const [retryIntent] = input.wake === "user_retry" ? await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: original.agentId, source: "on_demand", triggerDetail: "manual", reason: "task_retry", status: "completed", idempotencyKey: `task-retry:${randomUUID()}`, requestedByActorType: "user", requestedByActorId: "local-board", @@ -48,13 +60,16 @@ describe("a remote Runner that died in the middle of a provider turn", () => { finishedAt: new Date(Date.now() - 50 * 60_000), errorCode: "environment_provider_unavailable", contextSnapshot: { issueId: f.issueId, wakeReason: "issue_assigned" } }).returning() : []; const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: original.agentId, - source: system ? "automation" : "assignment", triggerDetail: "system", reason, status: "failed", runId: f.runId, + source: system && input.wake !== "system_assignment" && !chat ? "automation" : "assignment", triggerDetail: "system", + reason, status: "failed", runId: f.runId, requestedByActorType: system ? "system" : "user", requestedByActorId: system ? null : "local-board", - finishedAt: new Date(), payload: { issueId: f.issueId } }).returning(); + finishedAt: new Date(), payload: { issueId: f.issueId, ...(chatComment ? { commentId: chatComment.id } : {}) } }).returning(); await db.update(heartbeatRuns).set({ errorCode: "native_continuation_requires_reconciliation", error: "runner_remote_process_identity_unavailable", wakeupRequestId: wake.id, responsibleUserId: "local-board", retryOfRunId: predecessor?.id ?? null, startedAt: new Date(Date.now() - 45 * 60_000), - contextSnapshot: { issueId: f.issueId, wakeReason: reason, ...(retryIntent ? { taskRetryRequestId: retryIntent.id, + contextSnapshot: { issueId: f.issueId, wakeReason: reason, + ...(chatComment ? { wakeCommentId: chatComment.id, commentId: chatComment.id, wakeSource: "assignment" } : {}), + ...(retryIntent ? { taskRetryRequestId: retryIntent.id, forceFreshSession: true } : {}), ...(predecessor ? { retryOfRunId: predecessor.id, retryReason: "transient_failure", scheduledRetryAttempt: 1 } : {}) }, }).where(eq(heartbeatRuns.id, f.runId)); @@ -85,6 +100,11 @@ describe("a remote Runner that died in the middle of a provider turn", () => { { title: "a user's assignment", wake: "user" as const, viaClaim: true }, { title: "an automatic retry of the assignment (the staging shape)", wake: "system_retry" as const, viaClaim: true }, { title: "a same-run retry of a user's assignment", wake: "user" as const, viaClaim: false }, + // Staging after the upgrade fix: a routine's system assignment and an + // external chat message, neither a retry, stayed blocked for a person. + { title: "a routine's system assignment", wake: "system_assignment" as const, viaClaim: true }, + { title: "an external chat message from an unlinked sender", wake: "system_chat" as const, viaClaim: true }, + { title: "an external chat message from a linked person", wake: "user_chat" as const, viaClaim: false }, ])("continues $title once in a fresh session and records that the turn's outcome is unknown", async ({ wake, viaClaim }) => { const f = await runnerLostMidTurn({ wake }); const request = () => viaClaim ? requestRestartInterruptedTurnRetry(db, f.claim) : requestInterruptedNativeTurnRetry(db, f.runId); @@ -150,14 +170,60 @@ describe("a remote Runner that died in the middle of a provider turn", () => { eq(agentWakeupRequests.reason, "task_retry")))).filter(retry => retry.id !== hookIntent.id)).toHaveLength(0); }, 30_000); + // The staging chat task runs through a connected Slack conversation, whose + // endpoint restricts external execution: its Retry goes through the chat service. + describe("an external chat message in a connected conversation", () => { + async function connectConversation(f: { companyId: string; issueId: string; agentId: string }) { + const [application] = await db.insert(toolApplications).values({ companyId: f.companyId, name: "Slack", type: "chat" }).returning(); + const [connection] = await db.insert(toolConnections).values({ companyId: f.companyId, applicationId: application.id, + name: "Slack", uid: randomUUID(), transport: "chat_sdk", connectionPurpose: "channel" }).returning(); + const [endpoint] = await db.insert(chatEndpoints).values({ companyId: f.companyId, connectionId: connection.id, + provider: "slack", publicId: randomUUID(), assignedAgentId: f.agentId, status: "active" }).returning(); + expect(endpoint.externalExecutionPolicy).toBe("restricted"); + await db.insert(chatConversations).values({ companyId: f.companyId, endpointId: endpoint.id, issueId: f.issueId, + externalConversationId: "channel", externalThreadId: randomUUID(), externalLabel: "#work", state: "waiting" }); + } + + it("continues through the chat service's Retry", async () => { + const f = await runnerLostMidTurn({ wake: "system_chat" }); + await connectConversation(f); + const actionId = randomUUID(); + const prepared: Array<{ failedRunId: string; initiatedByUserId: string }> = []; + registerInterruptedTurnChatRetry(async (_tx, scope) => { prepared.push(scope); return { actionId }; }); + try { + const outcome = await requestRestartInterruptedTurnRetry(db, f.claim); + expect(outcome).toEqual({ requested: true, requestId: expect.any(String) }); + expect(prepared).toEqual([expect.objectContaining({ failedRunId: f.runId, initiatedByUserId: "local-board" })]); + const [retry] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, + (outcome as { requestId: string }).requestId)); + expect(retry.payload).toMatchObject({ taskRetry: { chatActionId: actionId } }); + expect(await unknownOutcomeEvents(f.runId)).toHaveLength(1); + } finally { registerInterruptedTurnChatRetry(undefined); } + }, 30_000); + + it("keeps the visible Retry when the chat service is not available", async () => { + const f = await runnerLostMidTurn({ wake: "system_chat" }); + await connectConversation(f); + expect(await requestRestartInterruptedTurnRetry(db, f.claim)).toEqual({ requested: false, reason: "retry_refused" }); + }, 30_000); + }); + it.each([ { title: "the refusal did not prove the Runner process gone", wake: "user" as const, refusal: "process_identity_changed" as const, reason: "provider_turn_unsettled" }, { title: "no reattachment refusal was recorded", wake: "user" as const, refusal: null, reason: "provider_turn_unsettled" }, { title: "the source was a monitor wake, not a retry of owed work", wake: "system_monitor" as const, refusal: "process_not_running" as const, reason: "not_user_request" }, - ])("keeps the visible Retry and records nothing when $title", async ({ wake, refusal, reason }) => { - const f = await runnerLostMidTurn({ wake, refusal }); + { title: "a system chat wake's message was deleted", wake: "system_chat" as const, refusal: "process_not_running" as const, + chatMessage: "deleted" as const, reason: "not_user_request" }, + { title: "a system chat wake's message was written by a run", wake: "system_chat" as const, + refusal: "process_not_running" as const, chatMessage: "written_by_run" as const, reason: "not_user_request" }, + { title: "a linked person's chat message was deleted", wake: "user_chat" as const, refusal: "process_not_running" as const, + chatMessage: "deleted" as const, reason: "message_unavailable" }, + { title: "a system assignment's Runner was not proven gone", wake: "system_assignment" as const, + refusal: "process_identity_changed" as const, reason: "provider_turn_unsettled" }, + ])("keeps the visible Retry and records nothing when $title", async ({ wake, refusal, reason, ...rest }) => { + const f = await runnerLostMidTurn({ wake, refusal, chatMessage: "chatMessage" in rest ? rest.chatMessage : undefined }); expect(await requestRestartInterruptedTurnRetry(db, f.claim)).toEqual({ requested: false, reason }); expect(await unknownOutcomeEvents(f.runId)).toHaveLength(0); expect(await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId),