diff --git a/server/src/services/deferred-execution-wait.test.ts b/server/src/services/deferred-execution-wait.test.ts new file mode 100644 index 00000000000..1736eded083 --- /dev/null +++ b/server/src/services/deferred-execution-wait.test.ts @@ -0,0 +1,82 @@ +import { randomUUID } from "node:crypto"; +import { activityLog, agents, agentWakeupRequests, companies, createDb, issues } from "@paperclipai/db"; +import type { LiveEvent } from "@paperclipai/shared"; +import { and, eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; +import { publishActivity, type ActivityPublication } from "./activity-log.js"; +import { writeDeferredExecutionWait } from "./deferred-execution-wait.js"; +import { subscribeCompanyLiveEvents } from "./live-events.js"; + +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("deferred execution wait live refresh", () => { + let database: Awaited>; + let db: ReturnType; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("deferred-execution-wait-"); + db = createDb(database.connectionString); + }, 60_000); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + + /** A person's Retry, saved as a deferred request that admission has not started. */ + async function deferredRetry() { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Wait fixture", issuePrefix: companyId.slice(0, 8) }); + await db.insert(agents).values({ id: agentId, companyId, name: "Morgan", adapterType: "process", status: "active" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Website task", status: "blocked", assigneeAgentId: agentId }); + const [request] = await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "on_demand", + reason: "task_retry", status: "deferred_issue_execution", payload: { issueId } }).returning(); + return { companyId, agentId, issueId, requestId: request!.id }; + } + + function listen(companyId: string) { + const events: LiveEvent[] = []; + const stop = subscribeCompanyLiveEvents(companyId, event => events.push(event)); + const waiting = () => events.filter(event => event.type === "activity.logged" && event.payload.action === "issue.execution_waiting"); + return { waiting, stop }; + } + + const budget = { reason: "budget.blocked", message: "Retry is waiting for the budget to allow new work." }; + + it("refreshes an open task page once when admission records why a Retry waits", async () => { + const f = await deferredRetry(), live = listen(f.companyId); + try { + expect(await writeDeferredExecutionWait(db, { companyId: f.companyId, requestId: f.requestId }, budget)).toBe(true); + expect(live.waiting()).toHaveLength(1); + expect(live.waiting()[0]!.payload).toMatchObject({ entityType: "issue", entityId: f.issueId, issueId: f.issueId, + actorType: "system", agentId: f.agentId, details: { requestId: f.requestId, reason: "budget.blocked" } }); + // The same reason again changes nothing a reader can see: no repeated refresh. + await writeDeferredExecutionWait(db, { companyId: f.companyId, requestId: f.requestId }, budget); + expect(live.waiting()).toHaveLength(1); + // A different gate does refresh the page. + await writeDeferredExecutionWait(db, { companyId: f.companyId, requestId: f.requestId }, + { reason: "issue_tree_hold_active", message: "This task is paused." }, { replace: true }); + expect(live.waiting()).toHaveLength(2); + const rows = await db.select().from(activityLog).where(and(eq(activityLog.companyId, f.companyId), + eq(activityLog.action, "issue.execution_waiting"))); + expect(rows.map(row => (row.details as { reason: string }).reason)).toEqual(["budget.blocked", "issue_tree_hold_active"]); + } finally { live.stop(); } + }); + + it("publishes only after the caller's transaction commits when it passes its publication list", async () => { + const f = await deferredRetry(), live = listen(f.companyId), publications: ActivityPublication[] = []; + try { + await db.transaction(async tx => { + await writeDeferredExecutionWait(tx as unknown as typeof db, { companyId: f.companyId, requestId: f.requestId }, budget, { publications }); + expect(live.waiting()).toHaveLength(0); + }); + expect(publications).toHaveLength(1); + for (const publication of publications) publishActivity(publication); + expect(live.waiting()).toHaveLength(1); + } finally { live.stop(); } + }); + + it("records nothing for a request that is no longer deferred", async () => { + const f = await deferredRetry(), live = listen(f.companyId); + try { + await db.update(agentWakeupRequests).set({ status: "completed" }).where(eq(agentWakeupRequests.id, f.requestId)); + expect(await writeDeferredExecutionWait(db, { companyId: f.companyId, requestId: f.requestId }, budget)).toBe(false); + expect(live.waiting()).toHaveLength(0); + } finally { live.stop(); } + }); +}); diff --git a/server/src/services/deferred-execution-wait.ts b/server/src/services/deferred-execution-wait.ts index 6924cb4e1ae..08f974c144a 100644 --- a/server/src/services/deferred-execution-wait.ts +++ b/server/src/services/deferred-execution-wait.ts @@ -2,6 +2,7 @@ import { HttpError } from "../errors.js"; import { and, eq, sql } from "drizzle-orm"; import { agentWakeupRequests, type Db } from "@paperclipai/db"; import { notifyDeferredExecutionWork } from "./deferred-execution-notifications.js"; +import { persistActivity, publishActivity, type ActivityPublication } from "./activity-log.js"; /** Admission has saved this refusal on the caller's durable deferred receipt. * HTTP callers still receive the existing 409; receipt recovery waits for its @@ -10,17 +11,27 @@ export class PersistedExecutionWaitRefusal extends HttpError { constructor(message: string, details?: unknown) { super(409, message, details); } } +/** + * Save why a deferred request cannot start yet. When the reason changes, the + * request's task also records an `issue.execution_waiting` activity. Its live + * event refreshes an open task page, so a person who just pressed Retry sees + * the kept failure and the named wait (for example the budget) without + * reloading. Pass `publications` when the caller's transaction is still open; + * they must be published after it commits. + */ export async function writeDeferredExecutionWait(db: Db, scope: { companyId: string; requestId: string }, - condition: Record, options: { replace?: boolean } = {}) { - return db.transaction(async tx => { - const [current] = await tx.select({ payload: agentWakeupRequests.payload }).from(agentWakeupRequests) + condition: Record, options: { replace?: boolean; publications?: ActivityPublication[] } = {}) { + const pending: ActivityPublication[] = []; + const saved = await db.transaction(async tx => { + const [current] = await tx.select({ payload: agentWakeupRequests.payload, agentId: agentWakeupRequests.agentId }).from(agentWakeupRequests) .where(and(eq(agentWakeupRequests.companyId, scope.companyId), eq(agentWakeupRequests.id, scope.requestId), eq(agentWakeupRequests.status, "deferred_issue_execution"))).for("update"); if (!current) return false; const previous = current.payload?.executionWait; const previousReason = previous && typeof previous === "object" && !Array.isArray(previous) ? (previous as Record).reason : undefined; - if (!previous || previousReason !== condition.reason) await notifyDeferredExecutionWork(tx, scope); + const reasonChanged = !previous || previousReason !== condition.reason; + if (reasonChanged) await notifyDeferredExecutionWork(tx, scope); const [saved] = await tx.update(agentWakeupRequests).set({ payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}', ${options.replace ? sql`'{}'::jsonb` : sql`coalesce(${agentWakeupRequests.payload}->'executionWait', '{}'::jsonb)`} @@ -28,6 +39,16 @@ export async function writeDeferredExecutionWait(db: Db, scope: { companyId: str updatedAt: new Date(), }).where(and(eq(agentWakeupRequests.companyId, scope.companyId), eq(agentWakeupRequests.id, scope.requestId), eq(agentWakeupRequests.status, "deferred_issue_execution"))).returning({ id: agentWakeupRequests.id }); + const issueId = current.payload?.issueId; + if (saved && reasonChanged && typeof issueId === "string" && issueId && typeof condition.reason === "string") { + const { publication } = await persistActivity(tx as unknown as Db, { companyId: scope.companyId, actorType: "system", + actorId: "execution-admission", action: "issue.execution_waiting", entityType: "issue", entityId: issueId, + agentId: current.agentId, details: { requestId: scope.requestId, reason: condition.reason } }); + pending.push(publication); + } return Boolean(saved); }); + if (options.publications) options.publications.push(...pending); + else for (const publication of pending) publishActivity(publication); + return saved; } diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 4a230a73c3a..230c913591c 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -3107,7 +3107,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende // so the task explains it and a new click can replace this intent. const declineRetry = async (reason: string, message: string) => { await writeDeferredExecutionWait(tx as unknown as Db, { companyId: agent.companyId, - requestId: pending.id }, { reason, message }, { replace: true }); + requestId: pending.id }, { reason, message }, { replace: true, publications: issueActivityPublications }); logger.warn({ requestId: pending.id, issueId, reason }, "Task Retry could not start"); return { kind: "deferred" as const }; }; @@ -3581,7 +3581,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende const condition = { recoveryActionId: executionBlocker.recoveryActionId, ...continuationWait }; if (executionWaitRequestId) { await writeDeferredExecutionWait(tx as unknown as Db, { companyId: agent.companyId, - requestId: executionWaitRequestId }, condition, { replace: true }); + requestId: executionWaitRequestId }, condition, { replace: true, publications: issueActivityPublications }); return { kind: "deferred" as const, admitted: true }; } if (durableRequest || wakeCommentId || diff --git a/ui/src/components/ExecutionBlockerNotice.test.tsx b/ui/src/components/ExecutionBlockerNotice.test.tsx index 0788bef6054..ad264030ef4 100644 --- a/ui/src/components/ExecutionBlockerNotice.test.tsx +++ b/ui/src/components/ExecutionBlockerNotice.test.tsx @@ -38,6 +38,23 @@ describe("ordinary task interruption", () => { expect(container.querySelector("a, details, input, textarea")).toBeNull(); expect(container.textContent).not.toMatch(/service attention|process|receipt|workspace/i); }); + it("keeps the failure visible and names the gate while a saved Retry waits on the budget", async () => { + await render({ ...failed, responsibility: "user", state: "waiting_for_user", userAction: "budget", canRetry: false }); + const notice = container.querySelector('[data-testid="task-execution-waiting"]'); + expect(notice?.textContent).toContain("Couldn’t finish this attempt. Your message is saved."); + expect(notice?.textContent).toContain("waiting for the budget to allow new work"); + // The saved Retry starts by itself; a second Retry would only duplicate it. + expect(container.querySelector("button")).toBeNull(); + }); + + it("does not report a failed attempt for a task the budget held before its first run", async () => { + await render({ ...failed, responsibility: "user", state: "waiting_for_user", userAction: "budget", canRetry: false, + sourceRunIds: [], observedRunId: null }); + const notice = container.querySelector('[data-testid="task-execution-waiting"]'); + expect(notice?.textContent).toContain("This task has not started."); + expect(notice?.textContent).not.toContain("Couldn’t finish this attempt"); + }); + it("shows persisted backoff with a usable Retry", async () => { vi.useFakeTimers(); vi.setSystemTime(new Date("2026-10-20T00:00:00Z")); await render({ ...failed, state: "retry_scheduled", nextRetryAt: "2026-10-20T00:00:30Z" }); @@ -144,6 +161,14 @@ describe("ordinary task interruption", () => { expect(container.textContent).toContain(retryWait.message); expect(container.querySelector("button")).toBeNull(); }); + it("names the budget wait alongside a provider failure while a saved Retry is held", async () => { + const providerFailure = { schema: "paperclip.provider-failure.v1" as const, provider: "openai", phase: "startup" as const, + category: "overloaded" as const, retryability: "transient" as const, message: "Provider overloaded" }; + await render({ ...failed, providerFailure, responsibility: "user", state: "waiting_for_user", userAction: "budget", canRetry: false }); + expect(container.textContent).toContain("Provider overloaded"); + expect(container.querySelector('[data-testid="task-execution-waiting"]')?.textContent).toContain("waiting for the budget to allow new work"); + expect(container.querySelector("button")).toBeNull(); + }); it("shows an unknown provider's original explanation as text and preserves read-only access", async () => { const providerFailure = { schema: "paperclip.provider-failure.v1" as const, provider: "new-provider", phase: "startup" as const, category: "unknown" as const, retryability: "unknown" as const, httpStatus: 429, diff --git a/ui/src/components/ExecutionBlockerNotice.tsx b/ui/src/components/ExecutionBlockerNotice.tsx index 8a98297b3c9..b4c12e44574 100644 --- a/ui/src/components/ExecutionBlockerNotice.tsx +++ b/ui/src/components/ExecutionBlockerNotice.tsx @@ -18,6 +18,17 @@ const recoveryStepText: Record + presentation?.state === "waiting_for_user" && presentation.userAction === "budget"; + const blockerAction: Record = { blocked: "Blocked: open it to Retry", cancelled: "Cancelled: remove or replace it", @@ -55,6 +66,7 @@ export function ExecutionBlockerNotice({ presentation, canRetry, onRetry, optimi return Retry : undefined}> {presentation.state === "retry_scheduled" &&

{step ? `${step}. ${retryIn}` : "Paperclip will retry automatically."}

} + {isBudgetWait(presentation) &&

It is waiting for the budget to allow new work. Raise the budget in Costs.

} ; } if (step && presentation?.state === "working") return
; } + // Other person-owned decisions have their own card on the task (approval, + // project access, ...). A budget hold has none, so name it here and keep the + // failure visible. The saved attempt starts by itself once the budget allows. + if (!optimistic && presentation && isBudgetWait(presentation)) return ( +
+
+
+
+ ); if (optimistic || !presentation || presentation.responsibility !== "system" || !["failed", "retry_scheduled"].includes(presentation.state)) return null; return (