diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index d008805a395..6dff82acc7b 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -212,6 +212,13 @@ The active-lock lifecycle is part of the checkout contract: Stale-lock recovery is crash recovery, not a retry loop. Paperclip must not clear or adopt locks held by non-terminal runs. After stale cleanup, a checkout `409` should mean a real live owner, status/assignee mismatch, unresolved blocker, or active gate still prevents checkout. Agents must treat that `409` as an ownership conflict and stop rather than retrying the same checkout. +Stopping a queued legacy run before runtime selection clears only its own task +locks. It preserves deferred messages without promoting another run. The stop +must have a recorded, expected operator cancellation and no runtime selection, +start timestamp, or process identity. A later eligible instruction can adopt the +saved messages through normal admission. This rule does not prove a dispatched +provider has stopped or change native cancellation acknowledgement. + ### Known execution waits at admission A known execution hold is a waiting condition, not a new execution attempt. Every issue wake must read the current effective reconciliation hold under the issue admission lock before creating a run. Resolved recovery bookkeeping can still carry a no-replay hold; only clearing the effective hold makes admission eligible again. The final dispatch gate remains required for changes after admission. diff --git a/maps/agent-access-map/recovery.md b/maps/agent-access-map/recovery.md index e6fc1098fe6..70615f26782 100644 --- a/maps/agent-access-map/recovery.md +++ b/maps/agent-access-map/recovery.md @@ -126,6 +126,13 @@ and must not create reader participation. No live recovery proof is claimed. Gaps: Permitted resolutions use native `call_api` under existing eligibility and ownership checks. Source/action policy determines whether continuation is allowed. +A recorded operator Stop before legacy runtime selection releases only that +run's task locks and preserves deferred input for normal later admission. This +does not grant an agent a new retry or provider-stop authority. Local +[database regressions](../../server/src/modules/wake-queue/adapters/queued-operator-stop-postgres.test.ts) +cover the release boundary and a competing active routine cycle; they are not +live provider-stop proof. + ## `exhaustion` Review: `baseline`. diff --git a/maps/feature-map/recovery.md b/maps/feature-map/recovery.md index 251c79327b1..7ca8f4be28b 100644 --- a/maps/feature-map/recovery.md +++ b/maps/feature-map/recovery.md @@ -203,6 +203,12 @@ Automated: [legacy continuation](../../server/src/services/recovery/legacy-conti cover their respective recovery paths. [Legacy failure continuation](../../tests/e2e/legacy-failure-continuation.spec.ts) adds a browser journey; it is not native runtime proof. +[Queued Stop database regressions](../../server/src/modules/wake-queue/adapters/queued-operator-stop-postgres.test.ts) +verify that an operator cancellation before runtime selection releases its own +locks, preserves deferred input without launching it, and leaves another active +cycle of the same routine unchanged. Provider cancellations and already selected +or dispatched runs retain their existing release behavior. + Manual: stop a disposable run, submit a distinguishable follow-up, and observe any cleanup wait. After confirmed stop, verify the follow-up reaches one eligible continuation and produces the requested answer. Repeat with a task/ancestor diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index d2624b2512f..caf6372b2f5 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -1,4 +1,5 @@ import { isAcknowledgedNativeReassignmentStop, isAcknowledgedNativeStop } from "../../../services/acknowledged-native-stop.js"; +import { readRunCancellation } from "../../../services/run-cancellation.js"; import { isCompletedOnboardingHandoffWake } from "../../../services/chat-completion-delivery.js"; import { instanceSettingsService } from "../../../services/instance-settings.js"; import { currentConversationCommentCondition } from "../../../services/agent-conversations.js"; @@ -1096,6 +1097,11 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd sql`${agentWakeupRequests.payload}->>'issueId' = ${issueRow.id}`, )).limit(1) : []; + const cancellation = readRunCancellation(run.resultJson); + const queuedOperatorStop = run.runtimeMode === "legacy" && run.status === "cancelled" && + run.runtimeModeResolvedAt === null && run.startedAt === null && run.processStartedAt === null && + run.processPid === null && run.processGroupId === null && + cancellation?.source === "operator" && cancellation.expected; const preDrainFacts: PreDrainFacts = { issueRowPresent: issueRow !== null, executionRunIdMatchesRun: !issueRow || !issueRow.executionRunId || issueRow.executionRunId === run.id, @@ -1111,7 +1117,7 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd executionCancellationAcknowledged: isAcknowledgedNativeReassignmentStop(run) || (run.status === "cancelled" && - (parseObject(run.resultJson?.executionCancellation).state === "acknowledged" || isAcknowledgedNativeStop(run)) && + (queuedOperatorStop || parseObject(run.resultJson?.executionCancellation).state === "acknowledged" || isAcknowledgedNativeStop(run)) && !interruptedQueue), }; const preDrain = decidePreDrain(preDrainFacts); diff --git a/server/src/modules/wake-queue/adapters/queued-operator-stop-postgres.test.ts b/server/src/modules/wake-queue/adapters/queued-operator-stop-postgres.test.ts new file mode 100644 index 00000000000..bd510d15f1b --- /dev/null +++ b/server/src/modules/wake-queue/adapters/queued-operator-stop-postgres.test.ts @@ -0,0 +1,95 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agentWakeupRequests, agents, companies, createDb, heartbeatRuns, issues } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../../__tests__/helpers/embedded-postgres.js"; +import { createReleaseIssueExecution } from "../application/use-cases.js"; +import { createPostgresWakeQueueAdapter } from "./postgres.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const describePostgres = support.supported ? describe : describe.skip; + +describePostgres("queued operator stop release", () => { + let db: ReturnType; + let database: Awaited>; + + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("paperclip-queued-operator-stop-"); + db = createDb(database.connectionString); + }, 20_000); + + afterAll(async () => { + await database?.cleanup(); + }); + + async function seed(input: { activeCycle?: boolean; source?: string; started?: boolean; selected?: boolean }) { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); + const originId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Stop regression", issuePrefix: `T${companyId.slice(0, 6)}`, + defaultResponsibleUserId: "operator", requireBoardApprovalForNewAgents: false }); + await db.insert(agents).values({ id: agentId, companyId, name: "Worker", role: "engineer", status: "active", + adapterType: "codex_local", adapterConfig: {}, runtimeConfig: { heartbeat: { wakeOnDemand: true } } }); + await db.insert(issues).values({ id: issueId, companyId, title: "Stopped cycle", status: "in_progress", + assigneeAgentId: agentId, responsibleUserId: "operator", originKind: "routine_execution", originId }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", status: "cancelled", + runtimeMode: "legacy", contextSnapshot: { issueId }, finishedAt: new Date(), + ...(input.started ? { startedAt: new Date(), runtimeModeResolvedAt: new Date() } : input.selected ? { runtimeModeResolvedAt: new Date() } : {}), + resultJson: { ...(!input.started ? { executionRecovery: { kind: "bootstrap", providerWorkStarted: false }, + startupCancellation: { requestedAt: new Date().toISOString(), beforeNativeSelection: false } } : {}), cancellation: { source: input.source ?? "operator", expected: input.source !== "provider", + initiator: { type: input.source === "provider" ? "provider" : "user", id: "operator" }, + reason: "Stopped", recordedAt: new Date().toISOString() } } }); + let activeCycleId: string | null = null; + let activeRunId: string | null = null; + if (input.activeCycle) { + activeCycleId = randomUUID(); activeRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: activeRunId, companyId, agentId, invocationSource: "on_demand", status: "running", + contextSnapshot: { issueId: activeCycleId }, startedAt: new Date() }); + await db.insert(issues).values({ id: activeCycleId, companyId, title: "Current cycle", status: "in_progress", + assigneeAgentId: agentId, originKind: "routine_execution", originId, executionRunId: activeRunId }); + } else { + await db.update(issues).set({ executionRunId: runId, checkoutRunId: runId }).where(eq(issues.id, issueId)); + } + const [wake] = await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", reason: "issue_commented", + status: "deferred_issue_execution", payload: { issueId }, requestedByActorType: "user", requestedByActorId: "operator" }).returning(); + const release = createReleaseIssueExecution({ + issueLock: createPostgresWakeQueueAdapter(db, { resolveResponsibleUserId: async () => "operator", + getRoutineEnv: async () => ({ routineId: null, env: null, responsibleUserId: null }), resolveSessionBeforeForWakeup: async () => null }), + recovery: { escalateStrandedAssignedIssue: async () => { throw Error("Unexpected recovery"); }, + escalateStrandedRecoveryIssueInPlace: async () => { throw Error("Unexpected recovery"); } }, + }); + return { companyId, agentId, issueId, runId, wakeId: wake.id, activeCycleId, activeRunId, release }; + } + + it.each([false, true])("preserves pending work after a pre-dispatch Stop (active cycle=%s)", async (activeCycle) => { + const f = await seed({ activeCycle }); + const before = (await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]; + const result = await f.release({ companyId: f.companyId, runId: f.runId, now: new Date() }); + expect(result).toEqual({ outcome: { kind: "released" }, postCommitEffects: [] }); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ status: "in_progress", + statusVersion: before.statusVersion, assigneeAgentId: f.agentId, executionRunId: null, checkoutRunId: null }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId)))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(activeCycle ? 2 : 1); + if (f.activeCycleId) expect((await db.select().from(issues).where(eq(issues.id, f.activeCycleId)))[0]) + .toMatchObject({ status: "in_progress", executionRunId: f.activeRunId }); + expect(await f.release({ companyId: f.companyId, runId: f.runId, now: new Date() })).toEqual(result); + }); + + it("keeps ordinary provider cancellation promotion", async () => { + const f = await seed({ source: "provider" }); + expect((await f.release({ companyId: f.companyId, runId: f.runId, now: new Date() })).outcome.kind).toBe("promoted"); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId)))[0].status).toBe("queued"); + }); + + it("does not replace stop acknowledgement for a dispatched run", async () => { + const f = await seed({ started: true }); + expect((await f.release({ companyId: f.companyId, runId: f.runId, now: new Date() })).outcome.kind).toBe("released"); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId)))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + }); + + it("keeps the existing release path when runtime selection has occurred", async () => { + const f = await seed({ selected: true }); + expect((await f.release({ companyId: f.companyId, runId: f.runId, now: new Date() })).outcome.kind).toBe("promoted"); + }); +});