diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 99000257b36..1e11ff657e7 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -232,6 +232,11 @@ by company and normalized session. Migration 0432 builds both indexes concurrently. A poll therefore reads a few index entries, not a run's full event history. +A subtree pause release finds the runs that a dispatch gate refused for that pause +through the partial index `heartbeat_runs_tree_hold_suppression_idx`. The index +covers only cancelled runs with a `treeHoldSuppression` marker, keyed by company and +hold. Migration 0436 builds it concurrently. + The chat wake-notice sweep compares UUID columns directly against guarded, canonical lowercase UUID payload values. Malformed and noncanonical payloads still match nothing. The query can use existing primary keys without adding a diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 6c6a28cfc13..91a64f1acb5 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1919,7 +1919,22 @@ future execution until Resume, and missing stop proof still blocks continuation. A run interrupted by a pause is an operator hold, not a failure. The pause's member snapshot names each interrupted run, and that work is owed one re-dispatch once no -pause covers the task. The obligation belongs to the run, so overlapping pauses keep +pause covers the task. Work a dispatch gate refused because a pause covered the task +is owed the same way: a scheduled retry that came due during the hold, or a queued +run that reached its claim during the hold. The gate cancels that run with +`issue_paused` and records the covering hold as `treeHoldSuppression.holdId`. +Release revisits the hold's snapshot members, the tree's current tasks, and every +task with a run refused for that hold. That covers a task created under the tree +after the snapshot, a task moved out before release, and a task whose work an +overlapping pause refused and released while this one still covered it. The scheduled-retry gate holds the pause row while it saves its +refusal, so a concurrent release either sees the refusal or the retry sees no pause. +The queued-run gate checks the hold again after saving its refusal and resumes the +task itself if release already finished. A resume restores the assignee's +execution only; like a wake that admission refuses during a hold, a non-assignee +reply or review run refused by a pause is not replayed. Among runs refused before +gates recorded the hold, only scheduled retries carry `issue_paused`, so only those +fall back to a snapshot pause that was active when they were cancelled. The +obligation belongs to the run, so overlapping pauses keep it until the last covering release. Releasing or resuming a hold re-dispatches owed work (`issue_tree_resumed`, with `holdId` and `interruptedRunId` in the wake context), even without `wakeAgents`. The interrupted run must still be the task's diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 6a7cb262907..d25597c4452 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -593,6 +593,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0419_slippery_impossible_man.sql", "0432_strong_mole_man.sql", "0435_premium_puff_adder.sql", + "0436_legal_reavers.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0436_legal_reavers.sql b/packages/db/src/migrations/0436_legal_reavers.sql new file mode 100644 index 00000000000..f6b223cca01 --- /dev/null +++ b/packages/db/src/migrations/0436_legal_reavers.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "heartbeat_runs_tree_hold_suppression_idx" ON "heartbeat_runs" USING btree ("company_id",("result_json"->'treeHoldSuppression'->>'holdId')) WHERE "heartbeat_runs"."status" = 'cancelled' and "heartbeat_runs"."result_json" ? 'treeHoldSuppression'; diff --git a/packages/db/src/migrations/meta/0431_snapshot.json b/packages/db/src/migrations/meta/0436_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0431_snapshot.json rename to packages/db/src/migrations/meta/0436_snapshot.json index 020d5e6aa2b..f89861310b7 100644 --- a/packages/db/src/migrations/meta/0431_snapshot.json +++ b/packages/db/src/migrations/meta/0436_snapshot.json @@ -1,6 +1,6 @@ { - "id": "9fcf29d1-a9a6-4ead-8d43-7e087a2b2b24", - "prevId": "b900b172-2cdc-45c6-a964-27e850bc4c74", + "id": "a80d63fe-5111-49e8-a39a-7a5589b3b31a", + "prevId": "dedfce21-7ff9-4079-b5da-4a74a856a251", "version": "7", "dialect": "postgresql", "tables": { @@ -4058,6 +4058,28 @@ "method": "btree", "with": {} }, + "agent_wakeup_requests_tree_hold_resume_uq": { + "name": "agent_wakeup_requests_tree_hold_resume_uq", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "idempotency_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "\"agent_wakeup_requests\".\"idempotency_key\" LIKE 'tree-hold-resume:%' AND \"agent_wakeup_requests\".\"status\" <> 'skipped'", + "concurrently": true, + "method": "btree", + "with": {} + }, "agent_wakeup_requests_chat_completion_uq": { "name": "agent_wakeup_requests_chat_completion_uq", "columns": [ @@ -34186,6 +34208,50 @@ "concurrently": false, "method": "btree", "with": {} + }, + "heartbeat_run_events_usage_item_run_idx": { + "name": "heartbeat_run_events_usage_item_run_idx", + "columns": [ + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_run_events\".\"event_type\" = 'item.completed' and (\"heartbeat_run_events\".\"payload\" -> 'prpEvent' -> 'payload' ->> 'kind') = 'usage'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "heartbeat_run_events_compaction_session_idx": { + "name": "heartbeat_run_events_compaction_session_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"payload\" -> 'prpEvent' ->> 'normalizedSessionId')", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_run_events\".\"event_type\" = 'context.compacted'", + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -35009,6 +35075,28 @@ "method": "btree", "with": {} }, + "heartbeat_runs_tree_hold_suppression_idx": { + "name": "heartbeat_runs_tree_hold_suppression_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"result_json\"->'treeHoldSuppression'->>'holdId')", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_runs\".\"status\" = 'cancelled' and \"heartbeat_runs\".\"result_json\" ? 'treeHoldSuppression'", + "concurrently": false, + "method": "btree", + "with": {} + }, "heartbeat_runs_cost_accounting_pending_idx": { "name": "heartbeat_runs_cost_accounting_pending_idx", "columns": [ @@ -42448,6 +42536,12 @@ "primaryKey": false, "notNull": false }, + "origin_issue_id": { + "name": "origin_issue_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, "origin_identity_context_id": { "name": "origin_identity_context_id", "type": "uuid", @@ -42891,6 +42985,27 @@ "method": "btree", "with": {} }, + "issues_company_origin_issue_idx": { + "name": "issues_company_origin_issue_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "origin_issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, "issues_company_project_workspace_idx": { "name": "issues_company_project_workspace_idx", "columns": [ @@ -43490,6 +43605,19 @@ "onDelete": "no action", "onUpdate": "no action" }, + "issues_origin_issue_id_issues_id_fk": { + "name": "issues_origin_issue_id_issues_id_fk", + "tableFrom": "issues", + "tableTo": "issues", + "columnsFrom": [ + "origin_issue_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + }, "issues_execution_workspace_id_execution_workspaces_id_fk": { "name": "issues_execution_workspace_id_execution_workspaces_id_fk", "tableFrom": "issues", @@ -61840,6 +61968,13 @@ "type": "boolean", "primaryKey": false, "notNull": true, + "default": true + }, + "classic_workspace_enabled": { + "name": "classic_workspace_enabled", + "type": "boolean", + "primaryKey": false, + "notNull": true, "default": false }, "created_at": { diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 0d04f0a001b..be513a9a67f 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -3039,6 +3039,13 @@ "when": 1793937048235, "tag": "0435_premium_puff_adder", "breakpoints": true + }, + { + "idx": 436, + "version": "7", + "when": 1793952565204, + "tag": "0436_legal_reavers", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/heartbeat_runs.ts b/packages/db/src/schema/heartbeat_runs.ts index 28443732d95..c989342ac96 100644 --- a/packages/db/src/schema/heartbeat_runs.ts +++ b/packages/db/src/schema/heartbeat_runs.ts @@ -140,6 +140,10 @@ export const heartbeatRuns = pgTable( ), scheduledRetryCancellationIdx: index("heartbeat_runs_scheduled_retry_cancellation_idx").on(table.id) .where(sql`${table.status} = 'scheduled_retry' and jsonb_typeof(${table.resultJson}->'startupCancellation') = 'object'`), + // Runs a dispatch gate refused for a pause; that hold's release resumes them. + treeHoldSuppressionIdx: index("heartbeat_runs_tree_hold_suppression_idx") + .on(table.companyId, sql`(${table.resultJson}->'treeHoldSuppression'->>'holdId')`) + .where(sql`${table.status} = 'cancelled' and ${table.resultJson} ? 'treeHoldSuppression'`), costAccountingPendingIdx: index("heartbeat_runs_cost_accounting_pending_idx").on(table.updatedAt, table.id).where(sql`${table.costAccountingPending} = true`), costDataQualityIdx: index("heartbeat_runs_cost_data_quality_idx") .on(table.companyId, sql`coalesce(${table.finishedAt}, ${table.createdAt}) desc`, table.id) diff --git a/server/src/__tests__/issue-tree-control-routes.test.ts b/server/src/__tests__/issue-tree-control-routes.test.ts index 28966145ad0..1aeee40dc81 100644 --- a/server/src/__tests__/issue-tree-control-routes.test.ts +++ b/server/src/__tests__/issue-tree-control-routes.test.ts @@ -14,6 +14,7 @@ const mockTreeControlService = vi.hoisted(() => ({ cancelIssueStatusesForHold: vi.fn(), restoreIssueStatusesForHold: vi.fn(), getHold: vi.fn(), + listTreeIssues: vi.fn(), getActivePauseHoldGate: vi.fn(), releaseHold: vi.fn(), listPauseInterruptedResumeTargets: vi.fn(), @@ -62,7 +63,8 @@ async function createApp( where: (predicate: unknown) => { mockReplayWhere(predicate); return mockReplayBlocks(); }, limit: mockReplayBlocks, }; - app.use("/api", issueTreeControlRoutes({ select: () => query } as any, { pluginWorkerManager })); + app.use("/api", issueTreeControlRoutes({ select: () => query, selectDistinct: () => query } as any, + { pluginWorkerManager })); app.use(errorHandler); return app; } @@ -74,6 +76,7 @@ describe("issue tree control routes", () => { mockReplayBlocks.mockResolvedValue([]); mockExecutionBlocker.mockResolvedValue(null); mockTreeControlService.getHold.mockResolvedValue(null); + mockTreeControlService.listTreeIssues.mockResolvedValue([]); mockTreeControlService.getActivePauseHoldGate.mockResolvedValue(null); mockIssueService.getById.mockResolvedValue({ id: "11111111-1111-4111-8111-111111111111", diff --git a/server/src/__tests__/issue-tree-pause-resume-run.test.ts b/server/src/__tests__/issue-tree-pause-resume-run.test.ts index 20288d1be4d..14a581bca36 100644 --- a/server/src/__tests__/issue-tree-pause-resume-run.test.ts +++ b/server/src/__tests__/issue-tree-pause-resume-run.test.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import express from "express"; import request from "supertest"; -import { and, eq } from "drizzle-orm"; +import { and, eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { activityLog, @@ -487,6 +487,292 @@ describeEmbeddedPostgres("subtree pause and release of an in-flight child run", expect(await getExecutionBlocker(db, fixture.companyId, fixture.childId)).toBeNull(); }); + // The staging S50b shape: a workspace_busy retry is parked when the board + // pauses the tree, comes due during the pause, and the dispatch gate cancels + // it with issue_paused. No operator Stop names that run in the hold snapshot. + async function seedParkedWorkspaceBusyRetry(fixture: Awaited>) { + await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, fixture.childId)); + const context = { issueId: fixture.childId, taskId: fixture.childId, wakeReason: "issue_assigned" }; + const busyRunId = randomUUID(); + const retryRunId = randomUUID(); + const createdAt = new Date(Date.now() - 60_000); + await db.insert(heartbeatRuns).values({ id: busyRunId, companyId: fixture.companyId, agentId: fixture.agentId, + invocationSource: "assignment", triggerDetail: "system", status: "cancelled", errorCode: "workspace_busy", + error: "Waiting for the shared execution workspace", contextSnapshot: context, createdAt, + startedAt: createdAt, finishedAt: new Date(createdAt.getTime() + 1_000), + resultJson: { executionRecovery: { kind: "workspace_wait", providerWorkStarted: false } } }); + await db.insert(heartbeatRuns).values({ id: retryRunId, companyId: fixture.companyId, agentId: fixture.agentId, + invocationSource: "assignment", triggerDetail: "system", status: "scheduled_retry", retryOfRunId: busyRunId, + scheduledRetryAttempt: 1, scheduledRetryReason: "workspace_busy", + scheduledRetryAt: new Date(Date.now() + 1_000), contextSnapshot: { ...context, wakeReason: "workspace_busy_retry" }, + createdAt: new Date(createdAt.getTime() + 1_000) }); + return { busyRunId, retryRunId }; + } + + // "legacy" models a retry refused before gates recorded the hold (the staging run). + for (const releaseVia of ["release", "resume", "legacy"] as const) { + it(`resumes a parked retry the pause suppressed exactly once after ${releaseVia}`, async () => { + const fixture = await seedTree(); + const { busyRunId, retryRunId } = await seedParkedWorkspaceBusyRetry(fixture); + const pause = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds`).send({ mode: "pause" }); + expect(pause.status).toBe(201); + const holdId = pause.body.hold.id as string; + + // The retry comes due during the hold and the dispatch gate suppresses it. + await db.update(heartbeatRuns).set({ scheduledRetryAt: new Date(Date.now() - 1_000) }) + .where(eq(heartbeatRuns.id, retryRunId)); + await heartbeat.promoteDueScheduledRetries(new Date()); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + const [suppressed] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, retryRunId)); + expect(suppressed).toMatchObject({ status: "cancelled", errorCode: "issue_paused" }); + expect(suppressed!.resultJson).toMatchObject({ treeHoldSuppression: { holdId } }); + if (releaseVia === "legacy") { + await db.update(heartbeatRuns).set({ resultJson: null }).where(eq(heartbeatRuns.id, retryRunId)); + } + expect(executedRunIds).toEqual([]); + // The periodic stranded-issue sweep must not block or re-dispatch held work. + await heartbeat.reconcileStrandedAssignedIssues(); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + expect(await childRuns(fixture.childId)).toHaveLength(2); + expect(await getExecutionBlocker(db, fixture.companyId, fixture.childId)).toBeNull(); + + const released = releaseVia !== "resume" + ? await request(app).post(`/api/issues/${fixture.parentId}/tree-holds/${holdId}/release`).send({}) + : await request(app).post(`/api/issues/${fixture.parentId}/tree-holds`).send({ mode: "resume" }); + expect(released.status).toBe(200); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + // Stranded-issue recovery must not add a second executor for the same work. + await heartbeat.reconcileStrandedAssignedIssues(); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + + const runs = await childRuns(fixture.childId); + expect(runs.map((run) => run.id).slice(0, 2)).toEqual([busyRunId, retryRunId]); + const resumed = runs.slice(2); + expect(resumed).toHaveLength(1); + expect(resumed[0]).toMatchObject({ status: "succeeded" }); + expect(resumed[0]!.contextSnapshot).toMatchObject({ + wakeReason: "issue_tree_resumed", holdId, interruptedRunId: retryRunId, + }); + expect(executedRunIds).toEqual([resumed[0]!.id]); + expect(await getExecutionBlocker(db, fixture.companyId, fixture.childId)).toBeNull(); + const [child] = await db.select().from(issues).where(eq(issues.id, fixture.childId)); + expect(child!.status).toBe("done"); + }); + } + + it("resumes a queued run the claim gate refused during the hold exactly once", async () => { + const fixture = await seedTree(); + await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, fixture.childId)); + const pause = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds`).send({ mode: "pause" }); + expect(pause.status).toBe(201); + const holdId = pause.body.hold.id as string; + + // A run queued just before the hold, missed by its snapshot, reaches the claim gate. + const queuedRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: queuedRunId, companyId: fixture.companyId, agentId: fixture.agentId, + invocationSource: "assignment", triggerDetail: "system", status: "queued", + contextSnapshot: { issueId: fixture.childId, taskId: fixture.childId, wakeReason: "issue_assigned" } }); + // The claim gate holds the agent start lock; refusing the run must not wait on it. + await heartbeat.resumeQueuedRuns(); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + const [refused] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, queuedRunId)); + expect(refused).toMatchObject({ status: "cancelled", errorCode: "issue_paused" }); + expect(refused!.resultJson).toMatchObject({ treeHoldSuppression: { holdId } }); + expect(executedRunIds).toEqual([]); + + const release = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds/${holdId}/release`).send({}); + expect(release.status).toBe(200); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + await heartbeat.reconcileStrandedAssignedIssues(); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + + const runs = await childRuns(fixture.childId); + expect(runs.map((run) => run.contextSnapshot?.wakeReason)).toEqual(["issue_assigned", "issue_tree_resumed"]); + expect(runs[1]).toMatchObject({ status: "succeeded" }); + expect(runs[1]!.contextSnapshot).toMatchObject({ holdId, interruptedRunId: queuedRunId }); + expect(executedRunIds).toEqual([runs[1]!.id]); + expect(await getExecutionBlocker(db, fixture.companyId, fixture.childId)).toBeNull(); + }); + + async function pauseAndSuppressRetry(fixture: Awaited>) { + const { retryRunId } = await seedParkedWorkspaceBusyRetry(fixture); + const pause = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds`).send({ mode: "pause" }); + expect(pause.status).toBe(201); + await db.update(heartbeatRuns).set({ scheduledRetryAt: new Date(Date.now() - 1_000) }) + .where(eq(heartbeatRuns.id, retryRunId)); + await heartbeat.promoteDueScheduledRetries(new Date()); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + const [suppressed] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, retryRunId)); + expect(suppressed).toMatchObject({ status: "cancelled", errorCode: "issue_paused" }); + return { retryRunId, holdId: pause.body.hold.id as string }; + } + + async function expectResumedOnce(childId: string, holdId: string, interruptedRunId: string) { + const resumed = (await childRuns(childId)).filter((run) => run.contextSnapshot?.wakeReason === "issue_tree_resumed"); + expect(resumed).toHaveLength(1); + expect(resumed[0]).toMatchObject({ status: "succeeded" }); + expect(resumed[0]!.contextSnapshot).toMatchObject({ holdId, interruptedRunId }); + expect(executedRunIds).toEqual([resumed[0]!.id]); + } + + it("resumes a refused retry on a task moved out of the paused tree before release", async () => { + const fixture = await seedTree(); + const { retryRunId, holdId } = await pauseAndSuppressRetry(fixture); + await db.update(issues).set({ parentId: null }).where(eq(issues.id, fixture.childId)); + + const release = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds/${holdId}/release`).send({}); + expect(release.status).toBe(200); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + await expectResumedOnce(fixture.childId, holdId, retryRunId); + }); + + it("resumes a refused retry on a task outside the snapshot when release also wakes members", async () => { + const fixture = await seedTree(); + // The parent is user-assigned, so wakeAgents has no member to wake; the + // child joins the tree after the snapshot. + await db.update(issues).set({ parentId: null }).where(eq(issues.id, fixture.childId)); + const pause = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds`).send({ mode: "pause" }); + expect(pause.status).toBe(201); + const holdId = pause.body.hold.id as string; + expect(pause.body.hold.members.map((member: { issueId: string }) => member.issueId)).not.toContain(fixture.childId); + await db.update(issues).set({ parentId: fixture.parentId }).where(eq(issues.id, fixture.childId)); + const { retryRunId } = await seedParkedWorkspaceBusyRetry(fixture); + await db.update(heartbeatRuns).set({ scheduledRetryAt: new Date(Date.now() - 1_000) }) + .where(eq(heartbeatRuns.id, retryRunId)); + await heartbeat.promoteDueScheduledRetries(new Date()); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + const [suppressed] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, retryRunId)); + expect(suppressed).toMatchObject({ status: "cancelled", errorCode: "issue_paused" }); + + const release = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds/${holdId}/release`) + .send({ metadata: { wakeAgents: true } }); + expect(release.status).toBe(200); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + await expectResumedOnce(fixture.childId, holdId, retryRunId); + }); + + it("resumes a task added after overlapping snapshots once the last covering pause is released", async () => { + const fixture = await seedTree(); + const midId = randomUUID(); + await db.insert(issues).values({ id: midId, companyId: fixture.companyId, parentId: fixture.parentId, title: "Mid", + status: "in_progress", priority: "medium", responsibleUserId: "responsible-user", assigneeUserId: "responsible-user", + issueNumber: 3, identifier: `${fixture.prefix}-3` }); + await db.update(issues).set({ parentId: null }).where(eq(issues.id, fixture.childId)); + const outer = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds`).send({ mode: "pause" }); + const inner = await request(app).post(`/api/issues/${midId}/tree-holds`).send({ mode: "pause" }); + expect([outer.status, inner.status]).toEqual([201, 201]); + // The task joins under both pauses after their snapshots; the nearer pause refuses its retry. + await db.update(issues).set({ parentId: midId }).where(eq(issues.id, fixture.childId)); + const { retryRunId } = await seedParkedWorkspaceBusyRetry(fixture); + await db.update(heartbeatRuns).set({ scheduledRetryAt: new Date(Date.now() - 1_000) }) + .where(eq(heartbeatRuns.id, retryRunId)); + await heartbeat.promoteDueScheduledRetries(new Date()); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + const [suppressed] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, retryRunId)); + expect(suppressed!.resultJson).toMatchObject({ treeHoldSuppression: { holdId: inner.body.hold.id } }); + + const innerRelease = await request(app).post(`/api/issues/${midId}/tree-holds/${inner.body.hold.id}/release`).send({}); + expect(innerRelease.status).toBe(200); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + expect(executedRunIds).toEqual([]); + + const outerRelease = await request(app) + .post(`/api/issues/${fixture.parentId}/tree-holds/${outer.body.hold.id}/release`).send({}); + expect(outerRelease.status).toBe(200); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + await expectResumedOnce(fixture.childId, inner.body.hold.id, retryRunId); + }); + + it("leaves a due retry parked when the pause is released while the gate refuses it", async () => { + const fixture = await seedTree(); + const { retryRunId } = await seedParkedWorkspaceBusyRetry(fixture); + const pause = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds`).send({ mode: "pause" }); + expect(pause.status).toBe(201); + const holdId = pause.body.hold.id as string; + await db.update(heartbeatRuns).set({ scheduledRetryAt: new Date(Date.now() - 1_000) }) + .where(eq(heartbeatRuns.id, retryRunId)); + + // A release has written but not committed when the gate reads the pause. + let commitRelease!: () => void; + const releaseCommitted = new Promise((resolve) => { commitRelease = resolve; }); + let releaseWritten!: () => void; + const written = new Promise((resolve) => { releaseWritten = resolve; }); + const release = db.transaction(async (tx) => { + await tx.update(issueTreeHolds).set({ status: "released", releasedAt: new Date() }) + .where(eq(issueTreeHolds.id, holdId)); + releaseWritten(); + await releaseCommitted; + }); + await written; + const promotion = heartbeat.promoteDueScheduledRetries(new Date()); + // Wait until the gate, having seen the pause, waits on the uncommitted release. + let waiting = false; + for (let attempt = 0; attempt < 200 && !waiting; attempt += 1) { + const rows = await db.execute(sql`select 1 from pg_locks where not granted limit 1`); + waiting = (rows as unknown as unknown[]).length > 0; + if (!waiting) await new Promise((resolve) => setTimeout(resolve, 25)); + } + expect(waiting).toBe(true); + commitRelease(); + await release; + await promotion; + + // The gate did not cancel the retry under a pause that no longer exists; + // the retry runs as ordinary work, once. + const [afterGate] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, retryRunId)); + expect(afterGate!.status).not.toBe("cancelled"); + await heartbeat.promoteDueScheduledRetries(new Date()); + await heartbeat.resumeQueuedRuns(); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + const [ran] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, retryRunId)); + expect(ran).toMatchObject({ status: "succeeded" }); + expect(executedRunIds).toEqual([retryRunId]); + }); + + it("resumes work a gate refused after the release that should have resumed it had finished", async () => { + const fixture = await seedTree(); + await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, fixture.childId)); + const pause = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds`).send({ mode: "pause" }); + expect(pause.status).toBe(201); + const holdId = pause.body.hold.id as string; + const release = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds/${holdId}/release`).send({}); + expect(release.status).toBe(200); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + expect(await childRuns(fixture.childId)).toEqual([]); + + // The claim gate read the pause before release and saved its refusal after. + const refusedRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: refusedRunId, companyId: fixture.companyId, agentId: fixture.agentId, + invocationSource: "assignment", triggerDetail: "system", status: "cancelled", errorCode: "issue_paused", + finishedAt: new Date(), resultJson: { treeHoldSuppression: { holdId, rootIssueId: fixture.parentId } }, + contextSnapshot: { issueId: fixture.childId, taskId: fixture.childId, wakeReason: "issue_assigned" } }); + const { resumeIfPauseAlreadyReleased } = await import("../services/issue-tree-resume.js"); + const wakeup = (agentId: string, opts: Parameters[1]) => heartbeat.wakeup(agentId, opts); + expect(await resumeIfPauseAlreadyReleased(db, { companyId: fixture.companyId, issueId: fixture.childId, holdId, wakeup })) + .toEqual([]); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + // A repeated check cannot dispatch the same work twice. + await resumeIfPauseAlreadyReleased(db, { companyId: fixture.companyId, issueId: fixture.childId, holdId, wakeup }); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + await expectResumedOnce(fixture.childId, holdId, refusedRunId); + }); + + it("does not resume a paused-code cancellation that no pause hold covered", async () => { + const fixture = await seedTree(); + const { retryRunId } = await seedParkedWorkspaceBusyRetry(fixture); + // A suppressed run with the pause error code but no covering pause owes nothing. + await db.update(heartbeatRuns).set({ status: "cancelled", errorCode: "issue_paused", finishedAt: new Date() }) + .where(eq(heartbeatRuns.id, retryRunId)); + const { resumePauseInterruptedWork } = await import("../services/issue-tree-resume.js"); + expect(await resumePauseInterruptedWork(db, { + companyId: fixture.companyId, issueIds: [fixture.childId], actor: { actorType: "system", actorId: "system" }, + wakeup: (agentId, opts) => heartbeat.wakeup(agentId, opts), + })).toEqual([]); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + expect(await childRuns(fixture.childId)).toHaveLength(2); + expect(executedRunIds).toEqual([]); + }); + it("does not resume work the board reassigned while the tree was paused", async () => { const fixture = await seedTree(); const prepared = new Promise((resolve) => { preparationGate.reached = resolve; }); diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index ca98461535b..d3a038b7468 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -27,6 +27,7 @@ import { heartbeatRuns, issueRecoveryActions, issueComments, + issueTreeHolds, issues, nativeRunFinalizations, projects, @@ -52,7 +53,9 @@ import { issueService } from "../../../services/issues.js"; import { issueTreeControlService, isVerifiedIssueTreeControlInteractionWake, + ISSUE_PAUSED_ERROR_CODE, ISSUE_TREE_CONTROL_INTERACTION_WAKE_REASONS, + pauseSuppressionResult, } from "../../../services/issue-tree-control.js"; import { continuationSummaryParksExecutor, @@ -745,6 +748,17 @@ export function createPostgresRunDispatchAdapter( tx: Db, input: CancelSuppressedRetryInput, ): Promise { + // A retry a pause refused is owed one re-dispatch when that pause is released. + const pauseHoldId = input.errorCode === ISSUE_PAUSED_ERROR_CODE ? readNonEmptyString(input.details.holdId) : null; + if (pauseHoldId) { + // Hold the pause against a concurrent release until this refusal commits, + // so release either sees the refused retry or the retry sees no pause. + // Under a released pause the retry stays parked for the next promotion. + const [active] = await tx.select({ id: issueTreeHolds.id }).from(issueTreeHolds).where(and( + eq(issueTreeHolds.companyId, input.companyId), eq(issueTreeHolds.id, pauseHoldId), + eq(issueTreeHolds.status, "active"))).for("share"); + if (!active) return { applied: false }; + } await notifyExecutionRunWork(tx, { companyId: input.companyId, runId: input.runId }); await notifyInstructionCollection(tx, { companyId: input.companyId, runId: input.runId }); await notifyExternalAgentSettlement(tx, { companyId: input.companyId, runId: input.runId }); @@ -752,6 +766,9 @@ export function createPostgresRunDispatchAdapter( await notifyScheduledRetryCancellation(tx, { companyId: input.companyId, runId: input.runId }); await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.issueId ?? undefined }); await notifyChatRunMilestones(tx, { companyId: input.companyId, issueId: input.issueId ?? undefined, runId: input.runId }); + const pauseSuppression = pauseHoldId ? pauseSuppressionResult({ + holdId: pauseHoldId, rootIssueId: readNonEmptyString(input.details.rootIssueId), + }) : null; const [row] = await tx .update(heartbeatRuns) .set({ @@ -759,6 +776,9 @@ export function createPostgresRunDispatchAdapter( finishedAt: input.now, error: input.reason, errorCode: input.errorCode, + ...(pauseSuppression ? { + resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify(pauseSuppression)}::jsonb`, + } : {}), updatedAt: input.now, }) .where( diff --git a/server/src/routes/issue-tree-control.ts b/server/src/routes/issue-tree-control.ts index eb90f99f59d..592ea2189ce 100644 --- a/server/src/routes/issue-tree-control.ts +++ b/server/src/routes/issue-tree-control.ts @@ -97,14 +97,23 @@ export function issueTreeControlRoutes( root: { id: string; companyId: string }, holdId: string, actor: ReturnType, + /** Snapshot members already received an explicit wake. */ + options: { skipMembers?: boolean } = {}, ) { // A pause Stop is an operator hold. Re-dispatch the work it interrupted // once, under current admission gates; never replay uncertain actions. const hold = await treeControlSvc.getHold(root.companyId, holdId); if (!hold || hold.mode !== "pause") return []; + // Tasks under the tree now include ones added after the snapshot, whose + // work an overlapping hold that was released earlier may have refused. + const currentTree = await treeControlSvc.listTreeIssues(root.companyId, hold.rootIssueId); return resumeInterruptedWork(db, { companyId: root.companyId, - issueIds: (hold.members ?? []).map((member) => member.issueId), + issueIds: [...(options.skipMembers ? [] : (hold.members ?? []).map((member) => member.issueId)), + ...currentTree.map((issue) => issue.id)], + // Gates also refused work for this hold on tasks moved out of the tree since. + suppressedByHoldId: hold.id, + excludeIssueIds: options.skipMembers ? (hold.members ?? []).map((member) => member.issueId) : [], rootIssueId: root.id, actor, wakeup: (agentId, opts) => heartbeat.wakeup(agentId, opts), @@ -571,9 +580,11 @@ export function issueTreeControlRoutes( } } - // An explicit wake above already covers every executable member. - if (hold.mode === "pause" && req.body.metadata?.wakeAgents !== true) { - wakeFailures.push(...await resumePauseInterruptedWork(root, hold.id, actor)); + // An explicit wake above already covers every executable member, but not + // tasks outside the snapshot whose work a gate refused for this hold. + if (hold.mode === "pause") { + wakeFailures.push(...await resumePauseInterruptedWork(root, hold.id, actor, + { skipMembers: req.body.metadata?.wakeAgents === true })); } res.json({ ...hold, ...(wakeFailures.length ? { wakeFailures } : {}) }); diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index dff68ca5b46..df1fc858f91 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -145,9 +145,12 @@ import { } from "../issue-rewake-throttle.js"; import { logActivity } from "../activity-log.js"; import { ISSUE_BLOCKERS_RESOLVED_WAKE_REASON } from "../issue-dependency-wakeups.js"; +import { resumeIfPauseAlreadyReleased } from "../issue-tree-resume.js"; import { + ISSUE_PAUSED_ERROR_CODE, ISSUE_TREE_CONTROL_INTERACTION_WAKE_REASONS, isVerifiedIssueTreeControlInteractionWake, + pauseSuppressionResult, } from "../issue-tree-control.js"; import { buildExecutionWorkspaceAdapterConfig, @@ -219,7 +222,8 @@ export interface HeartbeatQueueDependencies extends Pick, options?: { suppressImmediateRecovery?: boolean; resumeStrandedQueue?: boolean; deferredPostCommitEffects?: WakeQueuePostCommitEffect[] }) => Promise; applyWakeQueuePostCommitEffects: (effects: WakeQueuePostCommitEffect[]) => Promise; publishRunLifecyclePluginEvent: (run: HeartbeatRun) => void; - cancelRunInternal: (runId: string, reason?: string, options?: { errorCode?: string }) => Promise; + cancelRunInternal: (runId: string, reason?: string, + options?: { errorCode?: string; resultJson?: Record }) => Promise; cancelActiveForAgentInternal: (agentId: string, reason: string) => Promise; resumeExecutionWaitComments: (scope?: DeferredReceiptRecoveryOptions) => Promise; resumeQueuedCommentInterrupt: (companyId: string, queueId: string) => Promise; @@ -799,6 +803,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende run: typeof heartbeatRuns.$inferSelect, companyAgents?: AgentOrgRow[], deferredPostCommitEffects?: WakeQueuePostCommitEffect[], + deferredCancellations?: Array<() => Promise>, ) { if (run.status !== "queued") return run; const [caseHandoffReceipt] = run.wakeupRequestId ? await db.select().from(caseHandoffs).where(and( @@ -952,32 +957,43 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende contextSnapshot: context, })); if (activePauseHold && !treeHoldInteractionWake) { - await cancelRunInternal( - run.id, - "Cancelled because issue is held by an active subtree pause hold", - ); - await logActivity(db, { - companyId: run.companyId, - actorType: "system", - actorId: "system", - agentId: run.agentId, - runId: run.id, - action: "issue.tree_hold_run_interrupted", - entityType: "heartbeat_run", - entityId: run.id, - issueId: issueId, - details: { - issueId, - holdId: activePauseHold.holdId, - rootIssueId: activePauseHold.rootIssueId, - source: "heartbeat.claim_queued_run", - securityPrinciples: [ - "Complete Mediation", - "Fail Securely", - "Secure Defaults", - ], - }, - }); + const refuse = async () => { + // Record the covering hold so its release re-dispatches this work once. + await cancelRunInternal( + run.id, + "Cancelled because issue is held by an active subtree pause hold", + { errorCode: ISSUE_PAUSED_ERROR_CODE, resultJson: pauseSuppressionResult(activePauseHold) }, + ); + await logActivity(db, { + companyId: run.companyId, + actorType: "system", + actorId: "system", + agentId: run.agentId, + runId: run.id, + action: "issue.tree_hold_run_interrupted", + entityType: "heartbeat_run", + entityId: run.id, + issueId: issueId, + details: { + issueId, + holdId: activePauseHold.holdId, + rootIssueId: activePauseHold.rootIssueId, + source: "heartbeat.claim_queued_run", + securityPrinciples: [ + "Complete Mediation", + "Fail Securely", + "Secure Defaults", + ], + }, + }); + // A release that committed before this refusal did not see it. + await resumeIfPauseAlreadyReleased(db, { companyId: run.companyId, issueId, + holdId: activePauseHold.holdId, wakeup: (agentId, opts) => trackWakeup(agentId, opts) }); + }; + // Cancellation promotes the agent's next queued run under the agent + // start lock, so a caller holding that lock settles it after release. + if (deferredCancellations) deferredCancellations.push(refuse); + else await refuse(); return null; } @@ -1906,6 +1922,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende // agent's next queued run, which takes this same lock. const rejectedClaims: Array<{ run: typeof heartbeatRuns.$inferSelect; err: HttpError }> = []; const deferredPostCommitEffects: WakeQueuePostCommitEffect[] = []; + const deferredCancellations: Array<() => Promise> = []; let cancellationReason: string | undefined; return withAgentStartLock(agentId, async () => { @@ -2018,7 +2035,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende if (claimedRuns.length >= availableSlots) break; let claimed: typeof heartbeatRuns.$inferSelect | null; try { - claimed = await claimQueuedRun(queuedRun, companyAgents, deferredPostCommitEffects); + claimed = await claimQueuedRun(queuedRun, companyAgents, deferredPostCommitEffects, deferredCancellations); } catch (err) { if (isPermanentClaimRejection(err)) { rejectedClaims.push({ run: queuedRun, err }); @@ -2061,6 +2078,10 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende } finally { if (cancellationReason) await cancelActiveForAgentInternal(agentId, cancellationReason); await cancelRejectedQueuedRuns(rejectedClaims); + for (const cancel of deferredCancellations) { + await cancel().catch((err) => logger.warn({ err, agentId }, + "queued heartbeat run refused by a pause hold could not be cancelled; the next claim retries it")); + } } }); } diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index e66d7cb5e69..0c6cd97cd1e 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -83,6 +83,24 @@ type RestoreTreeStatusResult = TreeStatusUpdateResult & { const TERMINAL_ISSUE_STATUSES = new Set(["done", "cancelled"]); export const ISSUE_TREE_RESUME_EXECUTABLE_STATUSES = ["todo", "in_progress", "in_review"]; + +/** The error code a dispatch gate records when an active pause hold refuses a run. */ +export const ISSUE_PAUSED_ERROR_CODE = "issue_paused"; + +/** + * Recorded on a run a dispatch gate refused because a pause hold covered its + * task, so that hold's release can re-dispatch the refused work once. + */ +export function pauseSuppressionResult(hold: { holdId: string; rootIssueId?: string | null }) { + return { treeHoldSuppression: { holdId: hold.holdId, rootIssueId: hold.rootIssueId ?? null } }; +} + +function readPauseSuppressionHoldId(resultJson: Record | null) { + const marker = resultJson?.treeHoldSuppression; + if (!marker || typeof marker !== "object" || Array.isArray(marker)) return null; + const holdId = (marker as Record).holdId; + return typeof holdId === "string" && holdId.length > 0 ? holdId : null; +} const ACTIVE_RUN_STATUSES = ["queued", "running"] as const; const DEFAULT_RELEASE_POLICY: IssueTreeHoldReleasePolicy = { strategy: "manual" }; const MAX_PAUSE_HOLD_ANCESTOR_DEPTH = 100; @@ -1226,13 +1244,16 @@ export function issueTreeControlService(db: Db) { } /** - * A pause interrupts runs as an operator hold, not as a failure. A pause - * hold's member snapshot names each run it interrupted, and that pending work - * is owed one re-dispatch once no pause covers the task. The obligation - * belongs to the run, so overlapping holds keep it until the last release. - * Current state governs: the interrupted run must still be the task's latest - * run, the task must be executable by the same agent, and the claim marker - * makes the resume happen at most once. Execution blockers are the caller's. + * A pause interrupts runs as an operator hold, not as a failure. That pending + * work is owed one re-dispatch once no pause covers the task. A run is owed + * when a pause hold's member snapshot names it and an operator Stop ended it, + * or when a dispatch gate refused it because a pause covered the task (a + * scheduled retry that came due, or a queued run reaching its claim); the + * gate records the covering hold on the run. The obligation belongs to the + * run, so overlapping holds keep it until the last release. Current state + * governs: the interrupted run must still be the task's latest run, the task + * must be executable by the same agent, and the claim marker makes the + * resume happen at most once. Execution blockers are the caller's. */ async function findOwedPauseResume(companyId: string, issueId: string) { const [issue] = await db.select({ @@ -1242,25 +1263,40 @@ export function issueTreeControlService(db: Db) { || !ISSUE_TREE_RESUME_EXECUTABLE_STATUSES.includes(issue.status)) return null; const [run] = await db.select({ id: heartbeatRuns.id, agentId: heartbeatRuns.agentId, status: heartbeatRuns.status, - finishedAt: heartbeatRuns.finishedAt, resultJson: heartbeatRuns.resultJson, + finishedAt: heartbeatRuns.finishedAt, resultJson: heartbeatRuns.resultJson, errorCode: heartbeatRuns.errorCode, }).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, companyId), sql`coalesce(${heartbeatRuns.issueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${issueId}`, )).orderBy(sql`${heartbeatRuns.createdAt} desc`).limit(1); if (!run || run.status !== "cancelled" || !run.finishedAt || run.agentId !== issue.assigneeAgentId - || run.resultJson?.treeHoldResume !== undefined - || readRunCancellation(run.resultJson)?.source !== "operator") return null; - // Only a Stop recorded while a pause naming this run was active is owed. - const [hold] = await db.select({ id: issueTreeHolds.id }).from(issueTreeHoldMembers) + || run.resultJson?.treeHoldResume !== undefined) return null; + // Only a cancellation recorded while a covering pause was active is owed. + const activeAtCancellation = and( + eq(issueTreeHolds.companyId, companyId), eq(issueTreeHolds.mode, "pause"), + sql`${issueTreeHolds.createdAt} <= ${run.finishedAt.toISOString()}::timestamptz`, + or(isNull(issueTreeHolds.releasedAt), + sql`${issueTreeHolds.releasedAt} >= ${run.finishedAt.toISOString()}::timestamptz`), + ); + const memberHold = (snapshotRunId?: string) => db.select({ id: issueTreeHolds.id }).from(issueTreeHoldMembers) .innerJoin(issueTreeHolds, eq(issueTreeHolds.id, issueTreeHoldMembers.holdId)) .where(and( eq(issueTreeHoldMembers.companyId, companyId), eq(issueTreeHoldMembers.issueId, issueId), - eq(issueTreeHoldMembers.activeRunId, run.id), eq(issueTreeHoldMembers.skipped, false), - eq(issueTreeHolds.companyId, companyId), eq(issueTreeHolds.mode, "pause"), - sql`${issueTreeHolds.createdAt} <= ${run.finishedAt.toISOString()}::timestamptz`, - or(isNull(issueTreeHolds.releasedAt), - sql`${issueTreeHolds.releasedAt} >= ${run.finishedAt.toISOString()}::timestamptz`), + snapshotRunId ? eq(issueTreeHoldMembers.activeRunId, snapshotRunId) : undefined, + eq(issueTreeHoldMembers.skipped, false), activeAtCancellation, )).orderBy(asc(issueTreeHolds.createdAt)).limit(1); + let hold: { id: string } | undefined; + if (run.errorCode === ISSUE_PAUSED_ERROR_CODE) { + const suppressedBy = readPauseSuppressionHoldId(run.resultJson); + // The gate recorded the pause it observed. A run refused before gates + // recorded that hold falls back to a snapshot pause active at the time. + [hold] = suppressedBy + ? await db.select({ id: issueTreeHolds.id }).from(issueTreeHolds).where(and(eq(issueTreeHolds.id, suppressedBy), + eq(issueTreeHolds.companyId, companyId), eq(issueTreeHolds.mode, "pause"))).limit(1) + : await memberHold(); + } else if (readRunCancellation(run.resultJson)?.source === "operator") { + // A Stop is owed only when the pause snapshot named the stopped run. + [hold] = await memberHold(run.id); + } if (!hold) return null; if (await getActivePauseHoldGate(companyId, issueId)) return null; return { issueId, agentId: run.agentId, interruptedRunId: run.id, holdId: hold.id }; diff --git a/server/src/services/issue-tree-resume.ts b/server/src/services/issue-tree-resume.ts index ff73374fb6f..55162cef7d9 100644 --- a/server/src/services/issue-tree-resume.ts +++ b/server/src/services/issue-tree-resume.ts @@ -1,5 +1,5 @@ import { and, asc, eq, gt, ne, sql } from "drizzle-orm"; -import { agentWakeupRequests, heartbeatRuns, issues, type Db } from "@paperclipai/db"; +import { agentWakeupRequests, heartbeatRuns, issueTreeHolds, issues, type Db } from "@paperclipai/db"; import { isUniqueViolation } from "../db-errors.js"; import { logActivity } from "./activity-log.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -123,11 +123,19 @@ async function dispatch(db: Db, input: { */ export async function resumePauseInterruptedWork( db: Db, - input: { companyId: string; issueIds: string[]; actor: ResumeActor; wakeup: Wakeup; rootIssueId?: string | null }, + input: { + companyId: string; issueIds: string[]; actor: ResumeActor; wakeup: Wakeup; rootIssueId?: string | null; + /** Also revisit tasks whose work a gate refused for this hold, wherever they are now. */ + suppressedByHoldId?: string; excludeIssueIds?: string[]; + }, ) { const treeControl = issueTreeControlService(db); const failures: Array<{ issueId: string; message: string }> = []; - for (const issueId of [...new Set(input.issueIds)]) { + const suppressedIssueIds = input.suppressedByHoldId + ? await listPauseSuppressedIssueIds(db, input.companyId, input.suppressedByHoldId) : []; + const excluded = new Set(input.excludeIssueIds ?? []); + for (const issueId of new Set([...input.issueIds, ...suppressedIssueIds])) { + if (excluded.has(issueId)) continue; try { const target = await treeControl.findOwedPauseResume(input.companyId, issueId); if (!target) continue; @@ -154,6 +162,39 @@ export async function resumePauseInterruptedWork( return failures; } +/** + * Tasks with a run a dispatch gate refused for this hold. The gate also covers + * tasks created under the tree after the hold's snapshot, and a task can move + * out of the tree before release, so the refused runs name them. + */ +async function listPauseSuppressedIssueIds(db: Db, companyId: string, holdId: string) { + const rows = await db.selectDistinct({ + issueId: sql`coalesce(${heartbeatRuns.issueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId')`, + }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "cancelled"), + // Matches the partial index heartbeat_runs_tree_hold_suppression_idx. + sql`${heartbeatRuns.resultJson} ? 'treeHoldSuppression'`, + sql`${heartbeatRuns.resultJson}->'treeHoldSuppression'->>'holdId' = ${holdId}`, + )); + return rows.flatMap((row) => row.issueId ? [row.issueId] : []); +} + +/** + * A gate that refused work for a pause re-checks the hold after its refusal is + * saved. A release that committed in between did not see the refusal, so the + * gate resumes that task itself; the resume claim keeps it to one dispatch. + */ +export async function resumeIfPauseAlreadyReleased(db: Db, input: { + companyId: string; issueId: string; holdId: string; wakeup: Wakeup; +}) { + const [hold] = await db.select({ status: issueTreeHolds.status }).from(issueTreeHolds) + .where(and(eq(issueTreeHolds.companyId, input.companyId), eq(issueTreeHolds.id, input.holdId))); + if (hold?.status === "active") return []; + return resumePauseInterruptedWork(db, { + companyId: input.companyId, issueIds: [input.issueId], actor: SYSTEM_ACTOR, wakeup: input.wakeup, + }); +} + function pendingResumePredicate(staleBefore: Date) { return and(eq(heartbeatRuns.status, "cancelled"), sql`${heartbeatRuns.resultJson}->'treeHoldResume'->>'state' = 'pending'`,