diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index e84cedb5f46..d8221011a28 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -2000,6 +2000,7 @@ export { reconstructWorkspaceSchema, resumeWorkspaceFinalizationSchema, confirmNativeControllerStoppedSchema, + confirmRunStoppedSchema, confirmOwnerGoneSchema, issueReviewRequestSchema, issueExecutionWorkspaceSettingsSchema, diff --git a/packages/shared/src/types/execution-projection.ts b/packages/shared/src/types/execution-projection.ts index 052ad125886..39c7af04e52 100644 --- a/packages/shared/src/types/execution-projection.ts +++ b/packages/shared/src/types/execution-projection.ts @@ -129,7 +129,9 @@ export interface TaskInterruptionPresentation { canRetry: boolean; userAction: "approval" | "reconnect" | "assign_agent" | "choose_model" | "resume" | "budget" | "project_access" /** Board confirms that a previous server, which can no longer report, stopped its workspace copyback. */ - | "confirm_owner_stop" | null; + | "confirm_owner_stop" + /** Board confirms that a run whose server restarted before it recorded the agent's process has stopped. */ + | "confirm_run_stopped" | null; /** Only the task's own project id is exposed. Other source projects stay unnamed. */ projectAccess?: { agentId: string; projectId: string | null; ready: boolean }; sourceRunIds: string[]; diff --git a/packages/shared/src/validators/index.ts b/packages/shared/src/validators/index.ts index 248b0b48865..fa214eed3e7 100644 --- a/packages/shared/src/validators/index.ts +++ b/packages/shared/src/validators/index.ts @@ -452,6 +452,7 @@ export { reconstructWorkspaceSchema, resumeWorkspaceFinalizationSchema, confirmNativeControllerStoppedSchema, + confirmRunStoppedSchema, confirmOwnerGoneSchema, issueReviewRequestSchema, issueExecutionWorkspaceSettingsSchema, diff --git a/packages/shared/src/validators/issue.ts b/packages/shared/src/validators/issue.ts index 68ed55331c8..8a8affbcdcf 100644 --- a/packages/shared/src/validators/issue.ts +++ b/packages/shared/src/validators/issue.ts @@ -599,6 +599,11 @@ export const confirmNativeControllerStoppedSchema = z.object({ stopEvidence: z.string().trim().min(20).max(12000), }).strict(); +export const confirmRunStoppedSchema = z.object({ + runId: z.string().guid(), + stopEvidence: z.string().trim().min(20).max(12000), +}).strict(); + /** Board attestation that the owner of a record Paperclip cannot judge (saved * before owners were recorded) has stopped, so recovery may settle the record. */ export const confirmOwnerGoneSchema = z.object({ diff --git a/server/src/__tests__/interrupted-preparation-restore-retry.test.ts b/server/src/__tests__/interrupted-preparation-restore-retry.test.ts new file mode 100644 index 00000000000..84324ca9e57 --- /dev/null +++ b/server/src/__tests__/interrupted-preparation-restore-retry.test.ts @@ -0,0 +1,148 @@ +import { randomUUID } from "node:crypto"; +import { and, eq, inArray } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { + agents, + agentWakeupRequests, + authUsers, + companyMemberships, + companies, + createDb, + environmentLeases, + environments, + heartbeatRunEvents, + heartbeatRuns, + issueRecoveryActions, + issues, + workspaceOperations, +} from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { getExecutionBlocker } from "../services/execution-blocker.js"; +import { heartbeatService } from "../services/heartbeat.js"; +import { requestTaskRetry } from "../services/task-retry.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const suite = support.supported ? describe : describe.skip; + +/** + * Daytona storage baseline w-20261109T154810-s9911, task t06: a follow-up was + * restoring the saved remote workspace when a graceful shutdown interrupted it. + * No provider ran. Restore recovery saved the files, but the same agent stayed + * held, so a Retry was accepted and parked behind the hold with no run. + */ +suite("Retry after a shutdown interrupted workspace restore before provider work", () => { + let db!: ReturnType; + let temporary: Awaited> | null = null; + let companyId = "", agentId = "", environmentId = ""; + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("paperclip-restore-retry-"); + db = createDb(temporary.connectionString); + companyId = randomUUID(); agentId = randomUUID(); environmentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Paperclip", issuePrefix: "RRT", + requireBoardApprovalForNewAgents: false }); + await db.insert(agents).values({ id: agentId, companyId, name: "Coder", role: "engineer", status: "idle", + adapterType: "paperclip_runner", adapterConfig: {}, runtimeConfig: {}, permissions: {} }); + await db.insert(environments).values({ id: environmentId, name: "Sandbox", driver: "sandbox", status: "active", + config: { provider: "fake" } }); + await db.insert(authUsers).values({ id: "local-board", name: "Local operator", email: "local-board@example.invalid", + createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing(); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "local-board", membershipRole: "owner" }); + }, 30_000); + + afterAll(async () => { await temporary?.cleanup(); }); + + async function interruptedRestore(options: { cancellationSource?: string; terminationReceipt?: boolean; + controllerLeaseExpiresAt?: Date; recoveredSnapshotId?: string } = {}) { + const issueId = randomUUID(), runId = randomUUID(), leaseId = randomUUID(), providerLeaseId = `sandbox-${leaseId}`; + const finishedAt = new Date(Date.now() - 30_000); + const restoreRecovery = { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [leaseId] }; + const snapshotId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Follow-up", status: "blocked", + assigneeAgentId: agentId, statusVersion: 5 }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "automation", + issueId, status: "interrupted", errorCode: "server_shutdown_interrupted", runtimeMode: "legacy", + executionStage: "preparing", startedAt: new Date(finishedAt.getTime() - 3_000), finishedAt, nextEventSeq: 1, + contextSnapshot: { issueId, wakeReason: "issue_commented" }, + controllerLeaseExpiresAt: options.controllerLeaseExpiresAt ?? null, + runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" }, + remoteWorkspace: { schema: "paperclip.remote-workspace-run.v1", state: "saved", epoch: 2, leaseId, providerLeaseId, + snapshotId, workspaceId: randomUUID(), remoteCwd: "/home/daytona/workspace" } }, + resultJson: { stopReason: "interrupted", workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: restoreRecovery, + cancellation: { reason: "Server shutdown interrupted execution", source: options.cancellationSource ?? "shutdown", + expected: true, initiator: { type: "system" }, recordedAt: finishedAt.toISOString() } } }); + await db.insert(environmentLeases).values({ id: leaseId, companyId, environmentId, issueId, heartbeatRunId: runId, + status: "expired", cleanupStatus: "success", leasePolicy: "reuse_by_environment", provider: "fake", providerLeaseId, + releasedAt: new Date(finishedAt.getTime() + 2_000), + metadata: options.terminationReceipt === false ? {} : { remoteExecutionTermination: { schema: "paperclip.remote-termination.v1", + companyId, runId, leaseId, provider: "fake", providerLeaseId, state: "destroyed", confirmedAt: finishedAt.toISOString() } } }); + const holdId = randomUUID(), preservationId = randomUUID(); + await db.insert(workspaceOperations).values([ + { id: preservationId, companyId, issueId, heartbeatRunId: runId, phase: "workspace_recovery", status: "succeeded", + metadata: { recoveryActionId: holdId, storage: "remote_workspace", leaseId, providerLeaseId, + snapshotId: options.recoveredSnapshotId ?? snapshotId } }, + { companyId, issueId, heartbeatRunId: runId, phase: "workspace_finalize", status: "succeeded", + metadata: { owningService: "workspace_recovery", preservationOperationIds: [preservationId] } }, + ]); + await db.insert(issueRecoveryActions).values({ id: holdId, companyId, sourceIssueId: issueId, + kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", + fingerprint: `legacy-execution:${runId}`, status: "resolved", outcome: "blocked", ownerType: "system", + nextAction: "Verify the stop receipt and record workspace repair evidence.", resolvedAt: finishedAt, + evidence: { runId, attempt: 1, originalFailureCode: "server_shutdown_interrupted", continuationDelivery: "invalidated", + workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: restoreRecovery, + automaticRecovery: { policy: "preserve_without_replay_v1", runId, replay: "blocked", actionOutcome: "unknown" }, + nativeBootstrapFailureBlock: { runId, previousStatus: "in_progress", statusVersion: 5 }, + systemRecovery: { schema: "paperclip.system-recovery.v1", policyVersion: 1, sourceRunId: runId, state: "complete", + phase: "reconcile_effects", attempts: 1, retriesUsed: 0, preservationOperationIds: [preservationId] }, + workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", runId, status: "recovered", stage: "saving_receipts", + attempt: 1, attemptLimit: 6, retryable: true, requestIds: [] } } }); + return { issueId, runId, holdId }; + } + + const retry = (f: { issueId: string; runId: string }) => requestTaskRetry({ db, companyId, issueId: f.issueId, + actorId: "local-board", actorSource: "local_implicit", idempotencyKey: randomUUID(), observedRunId: f.runId }); + const successors = (f: { issueId: string; runId: string }) => db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.issueId, f.issueId), + inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]))); + + it("starts the Retry because no provider work can have an unknown outcome", async () => { + const heartbeat = heartbeatService(db); + const f = await interruptedRestore(); + expect(await getExecutionBlocker(db, companyId, f.issueId)).toBeNull(); + const receipt = await retry(f); + expect(receipt).toMatchObject({ accepted: true }); + await heartbeat.resumeTaskRetry(companyId, receipt.requestId!); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId!)); + expect(wake?.status).not.toBe("deferred_issue_execution"); + const [started] = await successors(f); + expect(started).toBeDefined(); + await vi.waitFor(async () => { + const [run] = await db.select({ status: heartbeatRuns.status }).from(heartbeatRuns).where(eq(heartbeatRuns.id, started!.id)); + expect(["queued", "running"]).not.toContain(run?.status); + }, { timeout: 30_000, interval: 200 }); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.holdId)))[0]) + .toMatchObject({ status: "resolved", outcome: "blocked", evidence: { automaticRecovery: { replay: "blocked" } } }); + }, 60_000); + + it.each([ + ["a provider turn started", async () => { + const f = await interruptedRestore(); + await db.insert(heartbeatRunEvents).values({ companyId, runId: f.runId, agentId, seq: 1, eventType: "turn.started", + stream: "system", level: "info", message: "turn started" }); + return f; + }], + ["a runner event was recorded", async () => { + const f = await interruptedRestore(); + await db.insert(heartbeatRunEvents).values({ companyId, runId: f.runId, agentId, seq: 1, eventType: "harness.ready", + stream: "system", level: "info", message: "harness ready", sourceEventId: randomUUID() }); + return f; + }], + ["an operator stopped it", () => interruptedRestore({ cancellationSource: "operator" })], + ["its sandbox has no stop receipt", () => interruptedRestore({ terminationReceipt: false })], + ["its controller may still be live", () => interruptedRestore({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) })], + ["recovery saved a different snapshot", () => interruptedRestore({ recoveredSnapshotId: randomUUID() })], + ] as const)("keeps the hold when %s", async (_label, arrange) => { + const f = await arrange(); + expect(await getExecutionBlocker(db, companyId, f.issueId)).toMatchObject({ recoveryActionId: f.holdId, runId: f.runId }); + }, 60_000); +}); diff --git a/server/src/__tests__/issue-confirm-run-stopped-routes.test.ts b/server/src/__tests__/issue-confirm-run-stopped-routes.test.ts new file mode 100644 index 00000000000..766af68a106 --- /dev/null +++ b/server/src/__tests__/issue-confirm-run-stopped-routes.test.ts @@ -0,0 +1,192 @@ +import { randomUUID } from "node:crypto"; +import { spawn } from "node:child_process"; +import { once } from "node:events"; +import os from "node:os"; +import express from "express"; +import request from "supertest"; +import { and, eq, inArray } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { + activityLog, agents, agentWakeupRequests, applyPendingMigrations, companies, companyMemberships, createDb, + environmentLeases, environments, heartbeatRunEvents, heartbeatRuns, issues, subscribeDatabaseWork, +} from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { errorHandler } from "../middleware/index.js"; +import { issueRoutes } from "../routes/issues.js"; +import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; +import { hasNativeLocalProcessStop, PROCESS_START_REQUESTED } from "../services/native-local-process-stop.js"; + +const LOCAL_PROCESS_STOPPED = "native.local_process_stopped"; +import { taskInterruptionPresentations } from "../services/task-interruption-presentation.js"; + +const externalTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_CONFIRM_RUN_STOPPED_TEST_DATABASE_URL"); +const support = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); + +describe.skipIf(!support.supported)("confirming that an unidentified local run stopped", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + + beforeAll(async () => { + if (externalTestDatabaseUrl) { + await applyPendingMigrations(externalTestDatabaseUrl); + db = createDb(externalTestDatabaseUrl); + } else { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-confirm-run-stopped-"); + db = createDb(tempDb.connectionString); + } + }, 30_000); + afterAll(async () => { + await db?.$client.end({ timeout: 1 }); + await tempDb?.cleanup(); + }); + + function app(actor: Express.Request["actor"]) { + const server = express(); + server.use(express.json()); + server.use((req, _res, next) => { req.actor = actor; next(); }); + server.use("/api", issueRoutes(db, {} as any)); + server.use(errorHandler); + return server; + } + const board = (companyId: string, membershipRole: "admin" | "viewer" = "admin"): Express.Request["actor"] => ({ + type: "board", userId: "board-user", companyIds: [companyId], + memberships: [{ companyId, membershipRole, status: "active" }], isInstanceAdmin: false, source: "session", + }); + + async function seed(options: { hostname?: string; launch?: boolean } = {}) { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Paperclip", issuePrefix: `S${companyId.slice(0, 5).toUpperCase()}` }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "board-user", + status: "active", membershipRole: "admin" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Codex", role: "engineer", adapterType: "codex_local", + adapterConfig: {}, runtimeConfig: {}, permissions: {} }); + await db.insert(issues).values({ id: issueId, companyId, title: "Chaos A1 brief", status: "blocked", priority: "medium", + assigneeAgentId: agentId, issueNumber: 1, identifier: `CHA-${issueId.slice(0, 8)}` }); + const finishedAt = new Date(Date.now() - 60_000); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, nativeIssueId: issueId, + runtimeMode: "native", status: "failed", errorCode: "adapter_failed", invocationSource: "assignment", + startedAt: new Date(finishedAt.getTime() - 90_000), finishedAt, contextSnapshot: { issueId } }); + if (options.launch !== false) await appendHeartbeatRunEvent(db, { companyId, runId, agentId, eventType: PROCESS_START_REQUESTED, + stream: "system", level: "info", message: "launch requested", payload: { hostname: options.hostname ?? os.hostname() } }); + const wakeId = randomUUID(); + const [wake] = await db.insert(agentWakeupRequests).values({ id: wakeId, companyId, agentId, source: "on_demand", triggerDetail: "manual", + reason: "task_retry", status: "deferred_issue_execution", requestedByActorType: "user", requestedByActorId: "board-user", + payload: { issueId, taskRetry: { schema: "paperclip.task-retry.v1", generationId: wakeId, actorId: "board-user", sourceRunId: runId, strategy: "continue" }, + executionWait: { reason: "process_identity_missing", message: "The previous run has no verified stop record." } }, + }).returning(); + return { companyId, agentId, issueId, runId, wakeId: wake!.id }; + } + + it("shows a board action instead of an endless wait, then records the confirmed stop (storage chaos ST1)", async () => { + const f = await seed(); + const [issue] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + const before = (await taskInterruptionPresentations(db, f.companyId, [issue!])).get(f.issueId); + expect(before).toMatchObject({ responsibility: "user", state: "waiting_for_user", canRetry: false, + userAction: "confirm_run_stopped", observedRunId: f.runId, retryWait: { reason: "process_identity_missing" } }); + expect(before?.retryWait?.message).toContain("a board operator can confirm"); + + const accountingWakes: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.accounting, event => { accountingWakes.push(event); }); + const res = await request(app(board(f.companyId))).post(`/api/issues/${f.issueId}/recovery-actions/confirm-run-stopped`) + .send({ runId: f.runId, stopEvidence: "No codex process for this task is running on the server host." }) + .then(response => { unsubscribe(); return response; }, error => { unsubscribe(); throw error; }); + + expect(res.status, JSON.stringify(res.body)).toBe(200); + expect(res.body).toMatchObject({ runId: f.runId, alreadyStopped: false }); + expect(accountingWakes).toContain("settled"); + expect(await hasNativeLocalProcessStop(db, f.companyId, f.runId)).toBe(true); + const [wakeAfter] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId)); + expect(wakeAfter!.status).toBe("coalesced"); + expect(wakeAfter!.runId).toBeTruthy(); + expect(wakeAfter!.runId).not.toBe(f.runId); + const [started] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wakeAfter!.runId!)); + expect(started).toMatchObject({ agentId: f.agentId }); + await vi.waitFor(async () => { + const [run] = await db.select({ status: heartbeatRuns.status }).from(heartbeatRuns).where(eq(heartbeatRuns.id, started!.id)); + expect(["queued", "running"]).not.toContain(run!.status); + }, { timeout: 30_000, interval: 200 }); + const [stop] = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.runId), + eq(heartbeatRunEvents.eventType, "native.local_process_stopped"))); + expect(stop!.payload).toMatchObject({ basis: "operator_attested", attestedByUserId: "board-user" }); + const [logged] = await db.select().from(activityLog).where(eq(activityLog.action, "issue.run_stop_confirmed")); + expect(logged).toMatchObject({ entityId: f.issueId, actorId: "board-user", runId: f.runId, + details: expect.objectContaining({ stopEvidence: "No codex process for this task is running on the server host." }) }); + }); + + it("refuses to confirm while a runner for the run is still running on this host", async () => { + const f = await seed(); + const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)", "--", "--run-id", f.runId], { stdio: "ignore" }); + try { + await once(child, "spawn"); + const res = await request(app(board(f.companyId))).post(`/api/issues/${f.issueId}/recovery-actions/confirm-run-stopped`) + .send({ runId: f.runId, stopEvidence: "I believe the previous agent has stopped working." }); + expect(res.status, JSON.stringify(res.body)).toBe(409); + expect(await hasNativeLocalProcessStop(db, f.companyId, f.runId)).toBe(false); + } finally { + const exited = once(child, "exit"); child.kill("SIGKILL"); await exited; + } + }); + + it.each([ + ["no_unidentified_launch", () => seed({ launch: false })], + ["launch_on_other_host", () => seed({ hostname: "another-server" })], + ["process_identity_recorded", async () => { + const f = await seed(); + await db.update(heartbeatRuns).set({ processPid: 424242 }).where(eq(heartbeatRuns.id, f.runId)); + return f; + }], + ["remote_environment", async () => { + const f = await seed(); + const environmentId = randomUUID(); + await db.insert(environments).values({ id: environmentId, name: "Sandbox", driver: "sandbox", + status: "active", config: { provider: "fake" } }); + await db.insert(environmentLeases).values({ companyId: f.companyId, environmentId, issueId: f.issueId, + heartbeatRunId: f.runId, status: "released", provider: "daytona", providerLeaseId: "sandbox-1", releasedAt: new Date() }); + return f; + }], + ] as const)("keeps the verification wait and refuses the confirmation when it cannot clear it (%s)", async (code, arrange) => { + const f = await arrange(); + const [issue] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + const presentation = (await taskInterruptionPresentations(db, f.companyId, [issue!])).get(f.issueId); + expect(presentation).toMatchObject({ userAction: null, canRetry: true, retryWait: { reason: "process_identity_missing" } }); + expect(presentation?.retryWait?.message).not.toContain("a board operator can confirm"); + const res = await request(app(board(f.companyId))).post(`/api/issues/${f.issueId}/recovery-actions/confirm-run-stopped`) + .send({ runId: f.runId, stopEvidence: "No codex process for this task is running on the server host." }); + expect(res.status, JSON.stringify(res.body)).toBe(409); + expect(res.body).toMatchObject({ details: { code } }); + expect(await hasNativeLocalProcessStop(db, f.companyId, f.runId)).toBe(false); + }); + + it("judges every parked task on a list page together", async () => { + const [a, b, c] = [await seed(), await seed(), await seed({ hostname: "another-server" })]; + const tasks = await db.select().from(issues).where(inArray(issues.id, [a.issueId, b.issueId, c.issueId])); + const byCompany = await Promise.all([a, b, c].map(f => taskInterruptionPresentations(db, f.companyId, + tasks.filter(task => task.companyId === f.companyId)))); + expect(byCompany.map((presentations, index) => presentations.get([a, b, c][index]!.issueId)?.userAction)) + .toEqual(["confirm_run_stopped", "confirm_run_stopped", null]); + }); + + it("reports a stop that was already recorded without recording another", async () => { + const f = await seed(); + await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, agentId: f.agentId, eventType: LOCAL_PROCESS_STOPPED, + stream: "system", level: "info", message: "stopped", payload: { processPid: null, processGroupId: null, processStartedAt: null } }); + const res = await request(app(board(f.companyId))).post(`/api/issues/${f.issueId}/recovery-actions/confirm-run-stopped`) + .send({ runId: f.runId, stopEvidence: "No codex process for this task is running on the server host." }); + expect(res.status, JSON.stringify(res.body)).toBe(200); + expect(res.body).toMatchObject({ runId: f.runId, alreadyStopped: true }); + expect(await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.runId), + eq(heartbeatRunEvents.eventType, LOCAL_PROCESS_STOPPED)))).toHaveLength(1); + }); + + it("lets only a board member with change access confirm the stop", async () => { + const f = await seed(); + const body = { runId: f.runId, stopEvidence: "No codex process for this task is running on the server host." }; + const agent: Express.Request["actor"] = { type: "agent", agentId: f.agentId, companyId: f.companyId, runId: randomUUID(), source: "agent_jwt" }; + expect((await request(app(agent)).post(`/api/issues/${f.issueId}/recovery-actions/confirm-run-stopped`).send(body)).status).toBe(403); + expect((await request(app(board(f.companyId, "viewer"))).post(`/api/issues/${f.issueId}/recovery-actions/confirm-run-stopped`) + .send(body)).status).toBe(403); + expect(await hasNativeLocalProcessStop(db, f.companyId, f.runId)).toBe(false); + }); +}); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 8e32f0a8ad8..8ffb36b5765 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -128,6 +128,7 @@ import { taskRetrySchema, resumeWorkspaceFinalizationSchema, confirmNativeControllerStoppedSchema, + confirmRunStoppedSchema, confirmOwnerGoneSchema, runnerGoalActionRequestSchema, feedbackTargetTypeSchema, @@ -10154,6 +10155,37 @@ export function issueRoutes( res.status(202).json(receipt); }); + router.post("/issues/:id/recovery-actions/confirm-run-stopped", validate(confirmRunStoppedSchema), async (req, res) => { + assertBoard(req); + const issue = await getAccessibleResource(req, res, svc.getById(req.params.id as string), "Issue not found"); + if (!issue || !(await assertIssueReadAllowed(req, res, issue))) return; + const decision = await access.decide({ actor: req.actor, action: "runtime:confirm_owner_stop", resource: { type: "company", companyId: issue.companyId } }); + if (!decision.allowed) throw forbidden(decision.explanation, authorizationDeniedDetails(decision)); + const { confirmUnidentifiedLocalRunnerStopped, UnidentifiedRunnerStopRefusedError } = await import("../services/native-local-process-stop.js"); + const actor = getActorInfo(req); + let receipt: Awaited>; + try { + receipt = await confirmUnidentifiedLocalRunnerStopped(db, { companyId: issue.companyId, issueId: issue.id, + runId: req.body.runId, actorId: actor.actorId }); + } catch (error) { + if (error instanceof UnidentifiedRunnerStopRefusedError) throw conflict(error.message, { code: error.code }); + throw error; + } + await logActivity(db, { companyId: issue.companyId, actorType: actor.actorType, actorId: actor.actorId, + action: "issue.run_stop_confirmed", entityType: "issue", entityId: issue.id, runId: receipt.runId, + details: { runId: receipt.runId, alreadyStopped: receipt.alreadyStopped, stopEvidence: req.body.stopEvidence } }); + const { promoteDeferredWorkIntents } = await import("../services/work-intent-outbox.js"); + await db.transaction(tx => promoteDeferredWorkIntents(tx, { companyId: issue.companyId, trigger: "lock_released", issueId: issue.id })); + const parked = await db.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, issue.companyId), eq(agentWakeupRequests.status, "deferred_issue_execution"), + eq(agentWakeupRequests.reason, "task_retry"), sql`${agentWakeupRequests.payload}->>'issueId' = ${issue.id}`)); + for (const wake of parked) { + await heartbeat.resumeTaskRetry(issue.companyId, wake.id).catch(err => + logger.warn({ err, issueId: issue.id, requestId: wake.id }, "Task Retry remains queued after the stop was confirmed")); + } + res.json({ ...receipt, presentation: await taskInterruptionPresentation(db, issue) }); + }); + router.post("/issues/:id/recovery-actions/confirm-controller-stopped", validate(confirmNativeControllerStoppedSchema), async (req, res) => { assertBoard(req); const issue = await getAccessibleResource(req, res, svc.getById(req.params.id as string), "Issue not found"); diff --git a/server/src/routes/openapi.ts b/server/src/routes/openapi.ts index 816c533ad41..9cd8fcc14b4 100644 --- a/server/src/routes/openapi.ts +++ b/server/src/routes/openapi.ts @@ -263,6 +263,7 @@ import { resumeWorkspaceFinalizationSchema, confirmNativeControllerStoppedSchema, confirmOwnerGoneSchema, + confirmRunStoppedSchema, cancelIssueThreadInteractionSchema, // Secret provider configs and remote import createSecretProviderConfigSchema, @@ -11136,6 +11137,15 @@ registerCurrentRoute({ responses: { 202: r.ok(), 400: r.badRequest, 401: r.unauthorized, 403: r.forbidden, 404: r.notFound, 409: r.conflict }, }); +registerCurrentRoute({ + method: "post", + path: "/api/issues/{id}/recovery-actions/confirm-run-stopped", + tags: ["issues"], + summary: "Confirm that a local run whose process was never recorded stopped, so its parked Retry can start", + body: confirmRunStoppedSchema, + responses: { 200: r.ok(), 400: r.badRequest, 401: r.unauthorized, 403: r.forbidden, 404: r.notFound, 409: r.conflict }, +}); + registerCurrentRoute({ method: "post", path: "/api/issues/{id}/recovery-actions/confirm-owner-gone", diff --git a/server/src/services/execution-blocker.ts b/server/src/services/execution-blocker.ts index 21c7406a06d..477a8735782 100644 --- a/server/src/services/execution-blocker.ts +++ b/server/src/services/execution-blocker.ts @@ -15,6 +15,7 @@ import { hasRecoveredCleanupStopReceipt, hasRemoteTerminationReceipt } from "./r import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; import { completedWorkspaceRecoveryCondition, hasVerifiedWorkspaceRepair, workspaceRecoveryDisplay } from "./workspace-recovery-receipt.js"; import { nativeManualAdmissionHoldsExecution } from "./native-runtime/native-runner-ownership.js"; +import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js"; const blockerContextKeys = ["issueId", "taskId", "source", "connectionIntentResolution", "interactionKind", "interactionStatus", "mutation", "interactionId"] as const; @@ -186,12 +187,57 @@ export function stoppedBeforeProviderTurnCondition() { where result.company_id = stopped.company_id and result.run_id = stopped.id))`; } +export function interruptedPreparationBeforeProviderWorkCondition() { + return sql`(stopped.status = 'interrupted' and stopped.runtime_mode = 'legacy' + and stopped.runtime_mode_resolved_at is null and stopped.execution_stage = 'preparing' + and stopped.runner_profile_json#>>'{adapterDispatch,adapterType}' = 'paperclip_runner' + and stopped.runner_profile_json->'nativeExecutionInput' is null + and stopped.process_pid is null and stopped.process_group_id is null and stopped.process_started_at is null + and stopped.session_id_after is null and stopped.native_issue_id is null and stopped.native_session_id is null + and (stopped.controller_lease_expires_at is null or stopped.controller_lease_expires_at <= now()) + and (stopped.runner_profile_json->'remoteWorkspace' is null or exists ( + select 1 from jsonb_array_elements_text(case when jsonb_typeof(${issueRecoveryActions.evidence}#>'{systemRecovery,preservationOperationIds}') = 'array' + then ${issueRecoveryActions.evidence}#>'{systemRecovery,preservationOperationIds}' else '[]'::jsonb end) saved(id) + join ${workspaceOperations} op on op.id::text = saved.id + where op.company_id = stopped.company_id and op.heartbeat_run_id = stopped.id and op.status = 'succeeded' + and op.metadata->>'snapshotId' = stopped.runner_profile_json#>>'{remoteWorkspace,snapshotId}')) + and coalesce(stopped.result_json#>'{executionRecovery,providerWorkStarted}', 'false'::jsonb) = 'false'::jsonb + and case when stopped.result_json->'cancellation' is null then stopped.error_code = 'server_shutdown_interrupted' + else stopped.result_json#>>'{cancellation,initiator,type}' = 'system' + and coalesce(stopped.result_json#>>'{cancellation,source}', '') <> 'operator' + and (stopped.result_json#>>'{cancellation,source}' = 'shutdown' + or stopped.result_json#>'{cancellation,expected}' = 'false'::jsonb) end + and not exists (select 1 from ${heartbeatRunEvents} work + where work.company_id = stopped.company_id and work.run_id = stopped.id + and (work.source_event_id is not null + or work.event_type in ('adapter.invoke', ${PROCESS_START_REQUESTED}, ${PROCESS_IDENTITY_RECORDED}) + or ${providerTurnWorkEventCondition(sql`work.event_type`, sql`work.source_event_id`)})) + and not exists (select 1 from ${nativeRunFinalizations} coordinator + where coordinator.company_id = stopped.company_id and coordinator.run_id = stopped.id) + and not exists (select 1 from ${nativeRunResults} result + where result.company_id = stopped.company_id and result.run_id = stopped.id) + and not exists (select 1 from ${environmentLeases} lease + where lease.company_id = stopped.company_id and lease.heartbeat_run_id = stopped.id + and not coalesce(lease.released_at is not null and lease.cleanup_status = 'success' + and lease.status in ('released', 'expired', 'failed') + and coalesce(lease.metadata->>'pendingCleanupInFlight', 'false') = 'false' + and (lease.provider = 'local' or ( + lease.metadata#>>'{remoteExecutionTermination,schema}' = 'paperclip.remote-termination.v1' + and lease.metadata#>>'{remoteExecutionTermination,companyId}' = lease.company_id::text + and lease.metadata#>>'{remoteExecutionTermination,runId}' = stopped.id::text + and lease.metadata#>>'{remoteExecutionTermination,leaseId}' = lease.id::text + and lease.metadata#>>'{remoteExecutionTermination,provider}' = lease.provider + and lease.metadata#>>'{remoteExecutionTermination,providerLeaseId}' = lease.provider_lease_id + and lease.metadata#>>'{remoteExecutionTermination,state}' in ('stopped', 'destroyed'))), false)))`; +} + /** * A file-preservation hold binds a newer task owner only until automatic * recovery saved the stopped run's files. That owner's turn is a fresh * assignment, not a replay: the stopped run's unknown outcomes stay with its - * own agent, which keeps the hold. A Stop fenced before any provider turn has - * no unknown outcome, so its hold releases for the same agent too. The proof is + * own agent, which keeps the hold. A Stop fenced before any provider turn, or a + * shutdown that interrupted preparation before any provider work, has no + * unknown outcome, so its hold releases for the same agent too. The proof is * the durable completed-recovery receipt bound to this action, so later reuse * of the workspace by the current owner (new epoch, writer or snapshot) cannot * re-arm the hold. @@ -217,7 +263,7 @@ export function restoreHoldReleasedForCurrentOwnerPredicate() { and task.execution_state->>'status' = 'pending' and task.execution_state->'currentParticipant'->>'type' = 'agent' and task.execution_state->'currentParticipant'->>'agentId' = stopped.agent_id::text, false)) - or ${stoppedBeforeProviderTurnCondition()}))`, + or ${stoppedBeforeProviderTurnCondition()} or ${interruptedPreparationBeforeProviderWorkCondition()}))`, completedWorkspaceRecoveryCondition(sourceRunId), // Copyback put the files back into the task workspace that the new owner // reuses. A snapshot-only save needs an explicit fresh-workspace request diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts index 47f4055bf11..2e30bc1f927 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/server/src/services/native-local-process-stop.ts @@ -143,6 +143,101 @@ export async function recordUnidentifiedLocalRunnerStop(db: Db, run: Pick; +type LaunchRefusal = { code: string; message: string } | "already_stopped" | null; + +export async function unidentifiedLocalLaunchRefusals(db: Db, runs: UnidentifiedLaunchRun[]): Promise> { + const result = new Map(); + const candidates: UnidentifiedLaunchRun[] = []; + for (const run of runs) { + if (run.runtimeMode !== "native" || !run.finishedAt || !FINISHED_RUN_STATUSES.has(run.status)) + result.set(run.id, { code: "run_not_finished", message: "Only a finished run's stop can be confirmed." }); + else if (run.processPid || run.processGroupId) + result.set(run.id, { code: "process_identity_recorded", message: "This run recorded its process. Paperclip verifies that process itself." }); + else candidates.push(run); + } + if (!candidates.length) return result; + const ids = candidates.map(run => run.id); + const companyIds = [...new Set(candidates.map(run => run.companyId))]; + const latest = await db.selectDistinctOn([heartbeatRunEvents.runId], { runId: heartbeatRunEvents.runId, + eventType: heartbeatRunEvents.eventType, payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and( + inArray(heartbeatRunEvents.companyId, companyIds), inArray(heartbeatRunEvents.runId, ids), + isNull(heartbeatRunEvents.sourceEventId), + inArray(heartbeatRunEvents.eventType, [PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED, LOCAL_PROCESS_STOPPED]), + )).orderBy(heartbeatRunEvents.runId, desc(heartbeatRunEvents.seq)); + const remote = new Set((await db.select({ runId: environmentLeases.heartbeatRunId }).from(environmentLeases).where(and( + inArray(environmentLeases.companyId, companyIds), inArray(environmentLeases.heartbeatRunId, ids), + sql`${environmentLeases.provider} is distinct from 'local'`, + ))).map(row => row.runId)); + const latestByRun = new Map(latest.map(row => [row.runId, row])); + for (const run of candidates) { + const event = latestByRun.get(run.id); + result.set(run.id, event?.eventType === LOCAL_PROCESS_STOPPED ? "already_stopped" + : event?.eventType !== PROCESS_START_REQUESTED + ? { code: "no_unidentified_launch", message: "This run has no unconfirmed runner launch to stop." } + : record(event.payload).hostname !== os.hostname() + ? { code: "launch_on_other_host", message: "The runner was launched on another host. Confirm the stop there." } + : remote.has(run.id) + ? { code: "remote_environment", message: "This run used a remote environment. Its environment cleanup confirms the stop." } + : null); + } + return result; +} + +export async function unidentifiedLocalLaunchRefusal(db: Db, run: UnidentifiedLaunchRun): Promise { + return (await unidentifiedLocalLaunchRefusals(db, [run])).get(run.id) ?? null; +} + +export async function confirmUnidentifiedLocalRunnerStopped(db: Db, input: { + companyId: string; issueId: string; runId: string; actorId: string; +}): Promise<{ runId: string; alreadyStopped: boolean }> { + const refuse = (code: string, message: string) => { throw new UnidentifiedRunnerStopRefusedError(code, message); }; + const [run] = await db.select().from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId))); + if (!run || ![run.issueId, run.nativeIssueId, record(run.contextSnapshot).issueId].includes(input.issueId)) + return refuse("run_not_found", "This run does not belong to the task."); + const refusal = await unidentifiedLocalLaunchRefusal(db, run); + if (refusal === "already_stopped") return { runId: run.id, alreadyStopped: true }; + if (refusal) return refuse(refusal.code, refusal.message); + const { noLocalRunnerProcessForRun } = await import("./native-runtime/native-session-executor.js"); + if (!await noLocalRunnerProcessForRun(run.id)) + return refuse("runner_still_running", "A runner for this run is still running on this host. Stop it first."); + await db.transaction(async tx => { + const [locked] = await tx.select().from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, run.id))).for("update"); + const current = locked ? await unidentifiedLocalLaunchRefusal(tx as unknown as Db, locked) : null; + if (!locked || current === "already_stopped" || current !== null || locked.finishedAt?.getTime() !== run.finishedAt?.getTime()) + return refuse("run_changed", "This run changed while its stop was being confirmed. Check it again."); + await appendHeartbeatRunEvent(tx as unknown as Db, { + companyId: run.companyId, + runId: run.id, + agentId: run.agentId, + eventType: LOCAL_PROCESS_STOPPED, + stream: "system", + level: "info", + message: "A board operator confirmed that this run's runner and its agent processes have stopped.", + payload: { processPid: null, processGroupId: null, processStartedAt: null, basis: "operator_attested", + attestedByUserId: input.actorId }, + }); + await notifyDeliveryWork(tx as unknown as Db, DELIVERY_QUEUES.accounting, run.companyId); + }); + return { runId: run.id, alreadyStopped: false }; +} + +const FINISHED_RUN_STATUSES = new Set(["failed", "cancelled", "timed_out", "succeeded", "interrupted"]); + +function record(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; +} + /** A native run that an admission fence refused before it requested any * runner launch (a quarantined prior-session cleanup is checked before the * provider starts) never had a process: there is nothing to stop. Any launch diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 452ee43d4ce..28285036118 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -2224,7 +2224,7 @@ export async function unidentifiedLocalRunnerProvenAbsent(input: { runId: string /** `paperclip-runnerd` is always launched with `--run-id `. A failed or * slow process listing proves nothing, so it keeps the caller blocked. */ -async function noLocalRunnerProcessForRun(runId: string): Promise { +export async function noLocalRunnerProcessForRun(runId: string): Promise { const listing = await new Promise(resolve => { execFile("ps", ["-axww", "-o", "command="], { encoding: "utf8", timeout: 5_000, maxBuffer: 64 * 1024 * 1024 }, (error, stdout) => resolve(error ? null : stdout)); diff --git a/server/src/services/task-interruption-presentation.ts b/server/src/services/task-interruption-presentation.ts index fab5116159c..4c77ab82c09 100644 --- a/server/src/services/task-interruption-presentation.ts +++ b/server/src/services/task-interruption-presentation.ts @@ -1,4 +1,5 @@ import { and, desc, eq, inArray, isNull, sql, type SQL } from "drizzle-orm"; +import { unidentifiedLocalLaunchRefusals } from "./native-local-process-stop.js"; import { agents, agentWakeupRequests, approvals, chatActions, companies, heartbeatRuns, issueApprovals, issueRecoveryActions, issues, issueTreeHoldMembers, issueTreeHolds, nativeRunFinalizations, projects, type Db } from "@paperclipai/db"; import { unionAll } from "drizzle-orm/pg-core"; @@ -36,7 +37,7 @@ type ProjectionIssue = Pick & Partial> & { + | "scheduledRetryReason" | "createdAt" | "finishedAt" | "processPid" | "processGroupId">> & { /** Due time of the native coordinator's own same-run resume, if one is pending. */ nativeResumeAt?: Date | null; /** When the native coordinator saved this run's terminal failure, if it did. */ @@ -63,6 +64,7 @@ type ProjectionState = { approval: boolean; /** A board-owned workspace-owner fence on this run that no automatic proof has released. */ ownerStopRequired?: boolean; + unidentifiedLaunchConfirmable?: boolean; /** The latest wake for a task with no run yet, when admission refused it. */ admissionWait?: { id: string; status: string; reason: string | null; agentId: string }; retryDependencies?: RetryDependencies; @@ -136,6 +138,9 @@ const retryWaitMessages: Record = { retry_source_changed: "Retry cannot continue because its saved message changed. Send a new message to continue from saved work.", }; +const UNIDENTIFIED_RUNNER_STOP_MESSAGE = "Paperclip can't confirm that the previous run stopped. The server restarted after it started the agent but before it recorded the agent's process. " + + "If no agent is still working on this task, a board operator can confirm that it stopped, and your Retry starts right away. Your message is saved."; + type RetryDependencies = { count: number; blocked: boolean; cancelled: boolean; savingFilesOnly: boolean }; function dependencyRetryWaitMessage(dependencies: RetryDependencies): string { @@ -266,6 +271,10 @@ export function projectTaskInterruption(issue: ProjectionIssue, input: Projectio if (retryWait?.sourceRunId === run.id && !retryWait.executionWaitReason && !retryWait.preparationStatus && dependencies?.count) return { ...base, intentId: retryWait.id, state: "working", retryWait: { reason: "dependency_blocked", message: dependencyRetryWaitMessage(dependencies) } }; + if (retryWait?.sourceRunId === run.id && retryWait.executionWaitReason === "process_identity_missing" + && input.unidentifiedLaunchConfirmable) + return { ...base, responsibility: "user", state: "waiting_for_user", canRetry: false, userAction: "confirm_run_stopped", + retryWait: { reason: "process_identity_missing", message: UNIDENTIFIED_RUNNER_STOP_MESSAGE } }; if (retryWait?.sourceRunId === run.id && retryWait.executionWaitReason) { const reason = Object.hasOwn(retryWaitMessages, retryWait.executionWaitReason) ? retryWait.executionWaitReason : "execution_recovery"; // Keep the latest input receipt so message submission can settle independently. @@ -311,6 +320,7 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i db.select({ issueId: requestedIssueId, id: heartbeatRuns.id, status: heartbeatRuns.status, createdAt: heartbeatRuns.createdAt, finishedAt: heartbeatRuns.finishedAt, agentId: heartbeatRuns.agentId, companyId: heartbeatRuns.companyId, nativeIssueId: heartbeatRuns.nativeIssueId, runtimeMode: heartbeatRuns.runtimeMode, startedAt: heartbeatRuns.startedAt, + processPid: heartbeatRuns.processPid, processGroupId: heartbeatRuns.processGroupId, agentAdapter: agents.adapterType, error: sql`case when ${heartbeatRuns.errorCode} = 'setup_failed' and ${heartbeatRuns.error} = 'Project is outside this low-trust boundary.' @@ -518,6 +528,12 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i const projectMap = new Map(projectRows.map(project => [project.id, project])); const pauses = new Set(pauseRows.map(row => row.issueId)), approvalsSet = new Set(approvalRows.map(row => row.issueId)); const admissionWaitMap = byIssue(admissionWaits); + const launchWaits = [...retryWaitMap].flatMap(([issueId, wait]) => { + const run = runMap.get(issueId); + return run && wait.sourceRunId === run.id && wait.executionWaitReason === "process_identity_missing" ? [{ issueId, run }] : []; + }); + const launchRefusals = await unidentifiedLocalLaunchRefusals(db, launchWaits.map(entry => entry.run)); + const confirmableLaunches = new Set(launchWaits.filter(entry => launchRefusals.get(entry.run.id) === null).map(entry => entry.issueId)); const now = Date.now(); for (const issue of selected) { const required = requirements.get(issue.id); @@ -532,6 +548,7 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i agent: agentMap.get(issue.assigneeAgentId ?? ""), project: projectMap.get(issue.projectId ?? ""), company: companyRows[0], paused: pauses.has(issue.id), approval: approvalsSet.has(issue.id), ownerStopRequired: ownerStopRequired.has(`${issue.id}:${facts.run?.id}`), + unidentifiedLaunchConfirmable: confirmableLaunches.has(issue.id), admissionWait: admissionWaitMap.get(issue.id), retryDependencies: retryDependencyMap.get(issue.id), }, now); diff --git a/ui/src/api/issues.ts b/ui/src/api/issues.ts index d80e60aa701..1be23279292 100644 --- a/ui/src/api/issues.ts +++ b/ui/src/api/issues.ts @@ -420,6 +420,9 @@ export const issuesApi = { `/issues/${id}/stalled-review-decision`, data, ), + confirmRunStopped: (id: string, data: { runId: string; stopEvidence: string }) => + api.post<{ runId: string; alreadyStopped: boolean; presentation: import("@paperclipai/shared").TaskInterruptionPresentation | null }>( + `/issues/${id}/recovery-actions/confirm-run-stopped`, data), confirmControllerStopped: (id: string, data: { actionId: string; runId: string; stopEvidence: string }) => api.post<{ runId: string; status: "queued" }>(`/issues/${id}/recovery-actions/confirm-controller-stopped`, data), confirmOwnerGone: (id: string, data: { actionId: string; recordId: string; stopEvidence: string; recordVersion?: string }) => diff --git a/ui/src/components/ExecutionBlockerNotice.confirm-run-stopped.test.tsx b/ui/src/components/ExecutionBlockerNotice.confirm-run-stopped.test.tsx new file mode 100644 index 00000000000..538ef5cf5ec --- /dev/null +++ b/ui/src/components/ExecutionBlockerNotice.confirm-run-stopped.test.tsx @@ -0,0 +1,83 @@ +// @vitest-environment jsdom +import { act } from "react"; +import { createRoot, type Root } from "react-dom/client"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { TaskInterruptionPresentation } from "@paperclipai/shared"; +import { queryKeys } from "../lib/queryKeys"; +import { ExecutionBlockerNotice } from "./ExecutionBlockerNotice"; + +const api = vi.hoisted(() => ({ confirmRunStopped: vi.fn() })); +vi.mock("../api/issues", () => ({ issuesApi: { confirmRunStopped: api.confirmRunStopped } })); +(globalThis as unknown as { IS_REACT_ACT_ENVIRONMENT: boolean }).IS_REACT_ACT_ENVIRONMENT = true; + +const message = "Paperclip can't confirm that the previous run stopped. If no agent is still working on this task, a board operator can confirm that it stopped, and your Retry starts right away. Your message is saved."; +const presentation: TaskInterruptionPresentation = { intentId: "intent", responsibility: "user", state: "waiting_for_user", + nextRetryAt: null, canRetry: false, userAction: "confirm_run_stopped", sourceRunIds: ["run-1"], observedRunId: "run-1", + retryWait: { reason: "process_identity_missing", message } }; + +describe("confirming that an unidentified previous run stopped", () => { + let root: Root, container: HTMLDivElement, client: QueryClient; + beforeEach(() => { + vi.resetAllMocks(); + client = new QueryClient({ defaultOptions: { queries: { retry: false, staleTime: Infinity }, mutations: { retry: false } } }); + client.setQueryData(queryKeys.access.currentBoardAccess, { userId: "board-1", source: "session", isInstanceAdmin: false, + companyIds: ["company-1"], memberships: [{ companyId: "company-1", membershipRole: "owner", status: "active" }] }); + container = document.createElement("div"); document.body.append(container); root = createRoot(container); + }); + afterEach(async () => { await act(async () => root.unmount()); container.remove(); client.clear(); }); + + async function mount(value: TaskInterruptionPresentation = presentation, issueId = "issue-1") { + await act(async () => root.render( + undefined} issueId={issueId} companyId="company-1" /> + )); + await act(async () => { await new Promise(resolve => setTimeout(resolve, 0)); }); + } + async function click(label: string) { + const button = Array.from(container.querySelectorAll("button")).find(candidate => candidate.textContent === label); + expect(button, label).toBeTruthy(); + await act(async () => button!.click()); + await act(async () => { await new Promise(resolve => setTimeout(resolve, 0)); }); + } + + it("explains the wait and confirms the stop instead of offering a Retry that cannot start (storage chaos ST1)", async () => { + api.confirmRunStopped.mockResolvedValue({ runId: "run-1", alreadyStopped: false, presentation: null }); + await mount(); + expect(container.textContent).toContain("a board operator can confirm"); + expect(Array.from(container.querySelectorAll("button")).some(button => button.textContent === "Retry")).toBe(false); + await click("Confirm it stopped"); + const textarea = container.querySelector("textarea")!; + await act(async () => { + const setter = Object.getOwnPropertyDescriptor(HTMLTextAreaElement.prototype, "value")!.set!; + setter.call(textarea, "No codex process for this task is running on the server host."); + textarea.dispatchEvent(new Event("input", { bubbles: true })); + }); + await click("Confirm and resume"); + expect(api.confirmRunStopped).toHaveBeenCalledExactlyOnceWith("issue-1", { runId: "run-1", + stopEvidence: "No codex process for this task is running on the server host." }); + expect(container.textContent).toContain("Starting your Retry"); + }); + + it("starts a fresh confirmation when the page moves to another task's run", async () => { + await mount(); + await click("Confirm it stopped"); + const textarea = container.querySelector("textarea")!; + await act(async () => { + const setter = Object.getOwnPropertyDescriptor(HTMLTextAreaElement.prototype, "value")!.set!; + setter.call(textarea, "Evidence written for the first task's run only."); + textarea.dispatchEvent(new Event("input", { bubbles: true })); + }); + await mount({ ...presentation, sourceRunIds: ["run-2"], observedRunId: "run-2" }, "issue-2"); + expect(container.querySelector("textarea")).toBeNull(); + await click("Confirm it stopped"); + expect(container.querySelector("textarea")!.value).toBe(""); + }); + + it("does not offer the confirmation to someone who cannot confirm a stop", async () => { + client.setQueryData(queryKeys.access.currentBoardAccess, { userId: "board-1", source: "session", isInstanceAdmin: false, + companyIds: ["company-1"], memberships: [{ companyId: "company-1", membershipRole: "viewer", status: "active" }] }); + await mount(); + expect(container.textContent).toContain("a board operator can confirm"); + expect(Array.from(container.querySelectorAll("button")).some(button => button.textContent === "Confirm it stopped")).toBe(false); + }); +}); diff --git a/ui/src/components/ExecutionBlockerNotice.tsx b/ui/src/components/ExecutionBlockerNotice.tsx index 7724d509f93..8a98297b3c9 100644 --- a/ui/src/components/ExecutionBlockerNotice.tsx +++ b/ui/src/components/ExecutionBlockerNotice.tsx @@ -9,6 +9,8 @@ import { useSecondTick } from "../hooks/useSecondTick"; import { TaskProjectAccessNotice } from "./TaskProjectAccessNotice"; import { ProviderFailureNotice } from "./ProviderFailureNotice"; import { LockedIssueChip, isLockedIssueStub } from "./LockedIssueChip"; +import { PreviousServerStopConfirmation } from "./WorkspaceExportRecovery"; +import { issuesApi } from "../api/issues"; const recoveryStepText: Record, string> = { checking_previous_run: "Checking that the previous attempt stopped", @@ -23,8 +25,9 @@ const blockerAction: Record = { }; /** One ordinary interruption surface. Technical records remain in diagnostics. */ -export function ExecutionBlockerNotice({ presentation, canRetry, onRetry, optimistic, takingLonger, error, className, issueId, agentName, projectName, blockers }: { +export function ExecutionBlockerNotice({ presentation, canRetry, onRetry, optimistic, takingLonger, error, className, issueId, companyId, agentName, projectName, blockers }: { presentation: TaskInterruptionPresentation | null | undefined; + companyId?: string; blockers?: IssueRelationIssueSummary[]; canRetry: boolean; onRetry: () => void; @@ -77,6 +80,15 @@ export function ExecutionBlockerNotice({ presentation, canRetry, onRetry, optimi : null} ; } + if (!optimistic && presentation?.userAction === "confirm_run_stopped" && presentation.observedRunId && issueId && companyId) { + const runId = presentation.observedRunId; + return issuesApi.confirmRunStopped(issueId, { runId, stopEvidence })} + copy={{ waiting: presentation.retryWait?.message ?? "Paperclip can't confirm that the previous run stopped. Your message is saved.", + question: "How did you confirm that no agent is still working on this task?", + placeholder: "For example: no codex or claude process for this task is running on the server host.", + resuming: "Starting your Retry…", confirm: "Confirm it stopped" }} />; + } if (optimistic || !presentation || presentation.responsibility !== "system" || !["failed", "retry_scheduled"].includes(presentation.state)) return null; return ( diff --git a/ui/src/components/WorkspaceExportRecovery.tsx b/ui/src/components/WorkspaceExportRecovery.tsx index 0c38829e65e..d9549d0857c 100644 --- a/ui/src/components/WorkspaceExportRecovery.tsx +++ b/ui/src/components/WorkspaceExportRecovery.tsx @@ -154,14 +154,15 @@ function retainedOwnerStopToken(action: IssueRecoveryAction): string | null { /** A previous server that can no longer report cannot confirm its own stop. * The board confirms it once; the saved result then finishes without another agent turn. */ -function PreviousServerStopConfirmation({ action, ownerToken, issueId, companyId, className, confirmStop, copy }: { - action: IssueRecoveryAction; issueId: string; companyId: string; className?: string; +export function PreviousServerStopConfirmation({ action, runId: givenRunId, formId: givenFormId, ownerToken, issueId, companyId, className, confirmStop, copy }: { + action?: IssueRecoveryAction; runId?: string; formId?: string; issueId: string; companyId: string; className?: string; /** Saved-work copyback owner; used when no custom confirmation is given. */ ownerToken?: string; confirmStop?: (stopEvidence: string) => Promise; - copy?: { waiting: string; question: string; placeholder: string; resuming: string }; + copy?: { waiting: string; question: string; placeholder: string; resuming: string; confirm?: string }; }) { - const runId = action.evidence.runId as string; + const runId = givenRunId ?? action!.evidence.runId as string; + const formId = givenFormId ?? action!.id; const client = useQueryClient(); const [open, setOpen] = useState(false); const [evidence, setEvidence] = useState(""); @@ -171,7 +172,7 @@ function PreviousServerStopConfirmation({ action, ownerToken, issueId, companyId const canConfirm = (accessView.kind === "ready" || accessView.kind === "stale") && canBoardConfirmOwnerStop(companyId, access.data); const confirm = useMutation({ retry: false, // query-error-ok: an explicit board confirmation; the error is shown below. mutationFn: () => confirmStop ? confirmStop(evidence.trim()) - : issuesApi.resumeWorkspaceFinalization(issueId, { actionId: action.id, runId, ownerToken: ownerToken!, + : issuesApi.resumeWorkspaceFinalization(issueId, { actionId: action!.id, runId, ownerToken: ownerToken!, controllerAndCopybackStopped: true, stopEvidence: evidence.trim() }), onSuccess: () => { setOpen(false); @@ -187,11 +188,11 @@ function PreviousServerStopConfirmation({ action, ownerToken, issueId, companyId ?

{accessView.kind === "loading" || accessView.kind === "reconnecting" ? "Checking access…" : "Paperclip is waiting to confirm that the previous server stopped."}

: !open ? + {copy?.confirm ?? "Confirm the previous server stopped"} :
{ event.preventDefault(); if (ready && !confirm.isPending) confirm.mutate(); }}> -