diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 35ae06460b8..d5a3f0b0fb2 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -84,6 +84,12 @@ transaction's task and run locks. Read and comment access follows normal policy. Board-initiated reassignments, agent API keys, native executions, and handoffs to another agent still require the outgoing execution to stop before ownership changes. +A native Stop can take longer than the request. When its durable Stop fence is +committed, the reassignment applies at once and the new assignee's wake waits +behind the stopping run. When that run releases the task, its Stop still does not +promote the stopped agent's own queued work, but it does promote the current +assignee's waiting wake. A run that finished on its own before the Stop needs no +stop. The request refuses the reassignment only when no Stop was recorded. ## 3. Status Semantics diff --git a/server/src/__tests__/issue-agent-mutation-ownership-routes.test.ts b/server/src/__tests__/issue-agent-mutation-ownership-routes.test.ts index dd4b7091b48..2318870258c 100644 --- a/server/src/__tests__/issue-agent-mutation-ownership-routes.test.ts +++ b/server/src/__tests__/issue-agent-mutation-ownership-routes.test.ts @@ -3029,6 +3029,79 @@ describe("agent issue mutation checkout ownership", () => { }); }); + describe("board reassignment whose run stop settles after the request", () => { + const nativeRun = { + id: ownerRunId, + companyId, + agentId: ownerAgentId, + runtimeMode: "native", + status: "running", + contextSnapshot: { issueId }, + resultJson: {}, + }; + const reassign = { assigneeAgentId: peerAgentId }; + + beforeEach(() => { + mockIssueService.getById.mockResolvedValue(makeIssue({ executionRunId: ownerRunId, checkoutRunId: ownerRunId })); + mockAgentService.resolveByReference.mockImplementation(async (_companyId: string, id: string) => ({ + ambiguous: false, + agent: await mockAgentService.getById(id), + })); + }); + + it("applies the reassignment when the native stop is durably recorded but its owner has not settled yet", async () => { + const fenced = { ...nativeRun, resultJson: { reassignmentStopRequested: true } }; + mockHeartbeatService.getRun.mockResolvedValueOnce(nativeRun).mockResolvedValue(fenced); + mockHeartbeatService.cancelRun.mockResolvedValue(fenced); + + const res = await request(createApp(boardActor())).patch(`/api/issues/${issueId}`).send(reassign); + + expect(res.status, JSON.stringify(res.body)).toBe(200); + expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith(ownerRunId, "Cancelled before issue reassignment", + expect.objectContaining({ errorCode: "issue_reassigned" })); + expect(mockHeartbeatService.getRun).toHaveBeenLastCalledWith(ownerRunId, { includeExecutionEvidence: true }); + expect(mockIssueService.update).toHaveBeenCalledTimes(1); + expect(mockIssueService.update.mock.calls[0]?.[1]).toMatchObject(reassign); + expect(mockHeartbeatService.cancelRun.mock.invocationCallOrder[0]) + .toBeLessThan(mockIssueService.update.mock.invocationCallOrder[0]!); + }); + + it("applies the reassignment when the native owner's acknowledgement is still pending", async () => { + const { HttpError: CurrentHttpError } = await vi.importActual("../errors.js"); + mockHeartbeatService.getRun.mockResolvedValueOnce(nativeRun) + .mockResolvedValue({ ...nativeRun, resultJson: { reassignmentStopRequested: true } }); + mockHeartbeatService.cancelRun.mockRejectedValue(new CurrentHttpError(409, + "Stop is recorded; the native session is still starting and will stop when it opens.", + { code: "native_stop_pending", runId: ownerRunId })); + + const res = await request(createApp(boardActor())).patch(`/api/issues/${issueId}`).send(reassign); + + expect(res.status, JSON.stringify(res.body)).toBe(200); + expect(mockIssueService.update.mock.calls[0]?.[1]).toMatchObject(reassign); + }); + + it("applies the reassignment when the run finished on its own before the stop landed", async () => { + mockHeartbeatService.getRun.mockResolvedValueOnce(nativeRun).mockResolvedValue({ ...nativeRun, status: "succeeded" }); + mockHeartbeatService.cancelRun.mockResolvedValue({ ...nativeRun, status: "succeeded" }); + + const res = await request(createApp(boardActor())).patch(`/api/issues/${issueId}`).send(reassign); + + expect(res.status, JSON.stringify(res.body)).toBe(200); + expect(mockIssueService.update.mock.calls[0]?.[1]).toMatchObject(reassign); + }); + + it("still refuses the reassignment when no stop was recorded for a run that is still active", async () => { + mockHeartbeatService.getRun.mockResolvedValue(nativeRun); + mockHeartbeatService.cancelRun.mockResolvedValue(nativeRun); + + const res = await request(createApp(boardActor())).patch(`/api/issues/${issueId}`).send(reassign); + + expect(res.status, JSON.stringify(res.body)).toBe(409); + expect(res.body.details).toMatchObject({ code: "runner_goal_reassignment_stop_unconfirmed", runId: ownerRunId }); + expect(mockIssueService.update).not.toHaveBeenCalled(); + }); + }); + describe("native run changing its own task status", () => { const nativeRun = { id: ownerRunId, diff --git a/server/src/modules/wake-queue/adapters/postgres.test.ts b/server/src/modules/wake-queue/adapters/postgres.test.ts index 73fb36fc2a0..b4a4574d492 100644 --- a/server/src/modules/wake-queue/adapters/postgres.test.ts +++ b/server/src/modules/wake-queue/adapters/postgres.test.ts @@ -2,7 +2,7 @@ import { resolveExternalTestDatabaseUrl } from "../../../__tests__/helpers/exter import { DELIVERY_QUEUES } from "../../../services/delivery-work-notifications.js"; import { instanceSettingsService } from "../../../services/instance-settings.js"; import { randomBytes, randomUUID } from "node:crypto"; -import { eq } from "drizzle-orm"; +import { and, eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import type { Db } from "@paperclipai/db"; import { @@ -691,6 +691,46 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0); }); + it("runs the new assignee's parked wake when a reassignment committed while the native Stop was settling", async () => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const nextAgentId = await seedAgent({ companyId, name: "NextOwner" }); + // The board reassigned the task while the outgoing run's Stop was still + // being acknowledged by its owner, so the task already names the successor. + const issueId = await seedIssue({ companyId, assigneeAgentId: nextAgentId, status: "in_progress" }); + const runId = await seedRun({ companyId, agentId, status: "cancelled", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId, resultJson: { + reassignmentStopRequested: true, + nativeCancellation: { schema: "paperclip.native-cancellation.v1", runId, companyId, issueId, + scope: "run", reasonCode: "cancellation_run_only", dispatchState: "acknowledged", dispatched: true, + intentAuditId: randomUUID(), acknowledgementAuditId: randomUUID() }, + } }).where(eq(heartbeatRuns.id, runId)); + await db.update(issues).set({ executionRunId: runId, checkoutRunId: runId }).where(eq(issues.id, issueId)); + const outgoingWakeId = await seedDeferredWake({ companyId, agentId, issueId }); + const assignmentWakeId = await seedDeferredWake({ companyId, agentId: nextAgentId, issueId }); + await db.update(agentWakeupRequests).set({ source: "assignment", reason: "issue_assigned" }) + .where(eq(agentWakeupRequests.id, assignmentWakeId)); + const release = createReleaseIssueExecution({ issueLock: createPostgresWakeQueueAdapter(db, stubDeps), + recovery: { escalateStrandedAssignedIssue: async () => { throw new Error("must not escalate a stopped run"); }, + escalateStrandedRecoveryIssueInPlace: async () => { throw new Error("must not escalate a stopped run"); } }, + }); + + const result = await release({ companyId, runId, now: new Date() }); + + expect(result.outcome.kind).toBe("promoted"); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), + eq(heartbeatRuns.agentId, nextAgentId))); + expect(successors).toHaveLength(1); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, assignmentWakeId)))[0]) + .toMatchObject({ status: "queued", runId: successors[0]!.id }); + // The stopped agent's own queued work is still not promoted by its Stop. + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, outgoingWakeId)))[0].status) + .toBe("deferred_issue_execution"); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), + eq(heartbeatRuns.agentId, agentId)))).toHaveLength(1); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0); + }); + it("releases an acknowledged interrupt Stop without blocking or retiring the settling executor", async () => { const companyId = await seedCompany(); const agentId = await seedAgent({ companyId }); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 2161b80f28c..9a2c633643c 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -1275,8 +1275,16 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd !interruptedQueue), }; const preDrain = decidePreDrain(preDrainFacts); + // A deliberate stop does not promote the stopped agent's queued work. + // When a reassignment committed while that stop was still settling, the + // new assignee's wake parked behind this run; it is explicit new work + // and must run now, or the task is left with no run at all. + const handoffAssigneeAgentId = preDrain.kind === "released" && + decidePreDrain({ ...preDrainFacts, executionCancellationAcknowledged: false }).kind === "proceed" && + issueRow?.assigneeAgentId && !issueRow.assigneeUserId && issueRow.assigneeAgentId !== run.agentId + ? issueRow.assigneeAgentId : undefined; - if (preDrain.kind === "released") { + if (preDrain.kind === "released" && !handoffAssigneeAgentId) { return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot }; } @@ -1335,7 +1343,7 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd }; } - if (await recordNativeTerminalRecoveryIfNeeded(tx, run, issueRow, input.now)) { + if (!handoffAssigneeAgentId && await recordNativeTerminalRecoveryIfNeeded(tx, run, issueRow, input.now)) { return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot }; } @@ -1357,7 +1365,8 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot }; } - const locked: LockedIssueExecution = { primaryIssue: toIssueSnapshot(issueRow), run: runSnapshot, recoveryOnly }; + const locked: LockedIssueExecution = { primaryIssue: toIssueSnapshot(issueRow), run: runSnapshot, + recoveryOnly: recoveryOnly && !handoffAssigneeAgentId, ...(handoffAssigneeAgentId ? { handoffAssigneeAgentId } : {}) }; const result = await fn(locked, { host: buildHost(tx, deps), transaction: buildTransaction(tx, deps, db, run) }); // Explicit retry admission requires the exact source's task claim. // Preserve it only when the existing queue-first recovery policy diff --git a/server/src/modules/wake-queue/application/ports.ts b/server/src/modules/wake-queue/application/ports.ts index 500d1675bc2..927498cb7bb 100644 --- a/server/src/modules/wake-queue/application/ports.ts +++ b/server/src/modules/wake-queue/application/ports.ts @@ -14,6 +14,13 @@ export type { InvokableAgentSnapshot, IssueSnapshot, ReleaseRecoveryBlockedNotic export type LockedIssueExecution = { /** Plan bounded recovery without draining messages while the finishing owner cleans up. */ recoveryOnly?: boolean; + /** + * The finishing run was deliberately stopped, and the issue now belongs to + * this other agent (a reassignment committed while that stop was settling). + * Promote only that assignee's deferred wake; the stopped run's own queued + * work stays put and no recovery is planned for it. + */ + handoffAssigneeAgentId?: string; primaryIssue: IssueSnapshot; run: RunSnapshot; }; diff --git a/server/src/modules/wake-queue/application/use-cases.ts b/server/src/modules/wake-queue/application/use-cases.ts index 6484ee18b9a..ddc41582fa4 100644 --- a/server/src/modules/wake-queue/application/use-cases.ts +++ b/server/src/modules/wake-queue/application/use-cases.ts @@ -169,6 +169,11 @@ async function runReleaseDrain( } processedWakeIds.add(candidate.id); + if (locked.handoffAssigneeAgentId && candidate.agentId !== locked.handoffAssigneeAgentId) { + preservedWakeIds.push(candidate.id); + continue; + } + // A lead can post its closing comment before committing Done. Check // again when the source run releases its queue, using every original // comment ID so coalesced human or unrelated input is never hidden. @@ -318,6 +323,9 @@ async function runReleaseDrain( return promoted; } + // A stopped run never plans recovery for itself, including after a handoff. + if (locked.handoffAssigneeAgentId) return { outcome: { kind: "released" }, postCommitEffects }; + return runReleaseRecoveryTail(issue, run, ports.host, ports.transaction, input, postCommitEffects); } diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index f7b5513d3fb..2f1ebf6605b 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -14807,26 +14807,51 @@ export function issueRoutes( runToStopForReassignment.companyId === existing.companyId && runToStopForReassignment.contextSnapshot?.issueId === existing.id; if (runToStopForReassignment && !isHumanSelfHandoff) { - const cancelled = await heartbeat.cancelRun( - runToStopForReassignment.id, - "Cancelled before issue reassignment", - { - errorCode: "issue_reassigned", - resultJson: { reassignmentStopConfirmed: true, issueMutationStopId }, - eventMessage: "run cancelled before issue reassignment", - eventPayload: { issueId: existing.id }, - }, - ); - if (!cancelled || cancelled.status !== "cancelled") { - throw conflict( - "The active agent run could not be stopped before reassignment", + let cancelled: Awaited> = null; + try { + cancelled = await heartbeat.cancelRun( + runToStopForReassignment.id, + "Cancelled before issue reassignment", { - code: "runner_goal_reassignment_stop_unconfirmed", - runId: runToStopForReassignment.id, + errorCode: "issue_reassigned", + resultJson: { reassignmentStopConfirmed: true, issueMutationStopId }, + eventMessage: "run cancelled before issue reassignment", + eventPayload: { issueId: existing.id }, }, ); + } catch (error) { + // The durable Stop intent is committed; only its remote owner's + // acknowledgement is outstanding. Settle below like any pending Stop. + if (!(error instanceof HttpError && parseObject(error.details).code === "native_stop_pending")) throw error; } - interruptedRunId = cancelled.id; + if (!cancelled || cancelled.status !== "cancelled") { + // A native Stop can outlast this request. Once its durable fence is + // committed the stop will complete, so apply the reassignment now: + // the new assignee's wake defers behind the stopping run and is + // promoted when it releases the issue. Refusing here would stop the + // work but drop the reassignment. A run that already finished on its + // own needs no stop. A failed native run may still own a same-run + // retry, so it keeps requiring a confirmed stop. + // Read the server-owned Stop fence through the bounded evidence projection, + // which is available even where the presentation projection hides resultJson. + const current = await heartbeat.getRun(runToStopForReassignment.id, { includeExecutionEvidence: true }); + const finishedWithoutRetry = Boolean(current) && ( + ["succeeded", "cancelled", "timed_out", "interrupted"].includes(current!.status) || + (current!.status === "failed" && current!.runtimeMode !== "native")); + const stopDurablyRequested = current?.runtimeMode === "native" && + ["running", "queued"].includes(current.status) && + parseObject(current.resultJson).reassignmentStopRequested === true; + if (!finishedWithoutRetry && !stopDurablyRequested) { + throw conflict( + "The active agent run could not be stopped before reassignment", + { + code: "runner_goal_reassignment_stop_unconfirmed", + runId: runToStopForReassignment.id, + }, + ); + } + } + interruptedRunId = runToStopForReassignment.id; } } diff --git a/server/src/services/heartbeat/run-state.test.ts b/server/src/services/heartbeat/run-state.test.ts index 7769b779da8..956da72b4ce 100644 --- a/server/src/services/heartbeat/run-state.test.ts +++ b/server/src/services/heartbeat/run-state.test.ts @@ -88,6 +88,8 @@ describe("heartbeat run state module", () => { expect(projectedSql(utf8.columns[0].resultJson)).toContain("pg_column_size"); await asciiState.getRun("evidence", { includeExecutionEvidence: true }); expect(projectedSql(ascii.columns[2].resultJson)).toContain("startupPreparationSettledAt"); + // A pending reassignment Stop is read through this projection on SQL_ASCII. + expect(projectedSql(ascii.columns[2].resultJson)).toContain("reassignmentStopRequested"); await asciiState.getRunWorkspaceSettlement("settlement"); expect(projectedSql(ascii.columns[3].error)).toBe("NULL"); expect(ascii.execute).toHaveBeenCalledTimes(1); diff --git a/server/src/services/heartbeat/run-state.ts b/server/src/services/heartbeat/run-state.ts index 3c48a03114f..4584ea7fd8a 100644 --- a/server/src/services/heartbeat/run-state.ts +++ b/server/src/services/heartbeat/run-state.ts @@ -327,6 +327,7 @@ const heartbeatRunExecutionEvidenceColumn = sql | null>` 'nativeCancellation', ${heartbeatRuns.resultJson} -> 'nativeCancellation', 'cancelledByActorType', ${heartbeatRuns.resultJson} -> 'cancelledByActorType', 'cancelledByUserId', ${heartbeatRuns.resultJson} -> 'cancelledByUserId', + 'reassignmentStopRequested', ${heartbeatRuns.resultJson} -> 'reassignmentStopRequested', 'conversationContinuation', ${heartbeatRuns.resultJson} -> 'conversationContinuation', 'cancellation', ${heartbeatRuns.resultJson} -> 'cancellation', 'acpToolInventoryComplete', ${heartbeatRuns.resultJson} -> 'acpToolInventoryComplete',