mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 07:10:34 +01:00
fix(tree-control): resume work a pause gate refused after release (#19421)
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work.
> - The board can pause an issue tree. A pause stops work. Resume or
release must continue that work once.
> - PR #19394 made release re-dispatch a run that a pause Stop
cancelled.
> - That rule covers only a run that an operator Stop ended and that the
hold snapshot named.
> - A scheduled retry that comes due during a pause is not stopped. The
dispatch gate cancels it with `issue_paused`. The release did not see it
as owed work, so the task stalled with no run.
> - This pull request makes every gate that refuses work for a pause
record the covering hold. Release then re-dispatches that work exactly
once.
> - The benefit is that a paused task with a parked retry or a queued
run continues after release, with no manual wake.
## Linked Issues or Issue Description
Refs #19394
**What happened?**
A child task had a `workspace_busy` scheduled retry. The board paused
the parent tree. The retry came due during the pause. The
scheduled-retry gate cancelled it with `issue_paused` ("Scheduled retry
suppressed because the issue is held by an active subtree pause hold").
The board released the hold. Four minutes later the child was still
`in_progress` with no run, no execution blocker, and no new activity.
**Expected behavior**
Release (or `mode: "resume"`) re-dispatches the refused work exactly
once. No reconciliation and no user message are necessary.
Stranded-issue recovery does not add a second run.
**Steps to reproduce**
1. Give an assigned child task a parked `workspace_busy` scheduled
retry.
2. Pause the parent tree.
3. Let the retry come due while the pause is active. The gate cancels it
with `issue_paused`.
4. Release the pause. No run starts for the child.
**Paperclip version or commit**
master at 4d49a31286 (includes #19394).
**Agent adapter(s) involved**
Not adapter-specific (core bug).
## What Changed
- `issue-tree-control.ts`: add `ISSUE_PAUSED_ERROR_CODE` and
`pauseSuppressionResult()`. A gate records `treeHoldSuppression: {
holdId, rootIssueId }` on a run that it refused for a pause.
- `findOwedPauseResume` now treats the latest run as owed when it is an
`issue_paused` cancellation under a pause hold. It uses the recorded
hold. For an unmarked run (a scheduled retry cancelled before this
change), it uses a snapshot pause that was active at cancellation. The
operator Stop rule does not change.
- Scheduled-retry gate (`run-dispatch/adapters/postgres.ts`,
`cancelSuppressedRetryInTx`): locks the pause row `FOR SHARE` and
records the covering hold. A concurrent release either sees the refusal,
or the retry sees no pause and runs normally.
- Queued-run claim gate (`heartbeat/queue.ts`): cancels with
`issue_paused` and the hold marker, after the agent start lock is
released. Before, the cancel re-entered that lock from inside it, and
the agent's queue stalled until the 30-second stale-lock timeout. After
the refusal is saved, the gate re-checks the hold and resumes the task
itself if release already finished (`resumeIfPauseAlreadyReleased`).
- Release (`issue-tree-resume.ts`, route): revisits the snapshot members
and every task with a run refused for that hold, also when `wakeAgents`
is set. This covers tasks created under the tree after the snapshot and
tasks moved out before release.
- `packages/db`: partial index
`heartbeat_runs_tree_hold_suppression_idx` (migration 0436, `CREATE
INDEX CONCURRENTLY`, registered as a top-level migration) for that
lookup.
- Docs: `doc/execution-semantics.md` (gate-refused work, race handling,
non-assignee and pre-fix scope) and `doc/DATABASE.md` (index).
## Verification
- Red before the fix, green after: `pnpm --filter @paperclipai/server
exec vitest run src/__tests__/issue-tree-pause-resume-run.test.ts` (21
tests).
- "resumes a parked retry the pause suppressed exactly once after
release / resume / legacy": before, `expected [] to have a length of 1`
(no resumed run). After, one `issue_tree_resumed` run with
`interruptedRunId` = the suppressed retry, no execution blocker, and the
task finished. `reconcileStrandedAssignedIssues` added no run during the
hold or after release.
- "resumes a queued run the claim gate refused during the hold exactly
once": before, the test timed out at 15 s (start-lock re-entry). After,
the run is cancelled with `issue_paused` and the hold marker, and
release re-dispatches it once.
- "resumes a refused retry on a task moved out of the paused tree before
release" and "... on a task outside the snapshot when release also wakes
members": review findings, one resume each.
- "leaves a due retry parked when the pause is released while the gate
refuses it": the test waits until the gate blocks on an uncommitted
release. The retry is not cancelled and runs once. Without the `FOR
SHARE` lock the gate does not wait and the test fails.
- "resumes work a gate refused after the release that should have
resumed it had finished": the late-refusal re-check resumes once, and a
repeated check does not dispatch again.
- "does not resume a paused-code cancellation that no pause hold
covered": negative case.
- Related suites pass locally: `src/__tests__/issue-tree*`,
`src/modules/run-dispatch`, `src/services/heartbeat/queue.test.ts`,
`heartbeat-dependency-scheduling`, `heartbeat-workspace-busy`, and
`packages/db` `client.test.ts`.
- I did not run a live staging check here. Staging APIs are not
reachable from this environment.
## Risks
- Migration 0436 adds a partial index on `heartbeat_runs`. It is built
concurrently and covers only cancelled runs with the marker, so it is
small.
- The scheduled-retry gate now takes a `FOR SHARE` row lock on the pause
hold before it cancels. Release updates that row, so the two serialize
briefly. Release does not lock issues or runs, so the lock order cannot
deadlock with the gate's issue-then-run locks.
- The queued-run claim gate now records `issue_paused` in place of the
generic `cancelled`. Stranded recovery already classifies `issue_paused`
as non-retryable. The pause resume owns that work, so recovery does not
add a second run.
- Scope limits (documented): a non-assignee run refused by a pause is
not replayed, the same as non-assignee wakes that admission drops during
a hold. Among pre-fix rows, only scheduled retries carry `issue_paused`,
so only those fall back to the snapshot. A pre-fix row whose hold is
already released does not self-recover; the staging task that showed
this bug needs one manual wake, or a new pause and release.
> For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and
discuss it in `#dev` before opening the PR. Feature PRs that overlap
with planned core work may need to be redirected — check the roadmap
first. See `CONTRIBUTING.md`.
## Model Used
- Claude Opus 5.5 (Anthropic), model ID `claude-opus-5-5`, used through
Claude Code with tool use (shell, file edits, test execution).
## Checklist
- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
2c7ee84221
commit
7e3caa7bac
14 files changed
+643
-57
No files matched your search
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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';
|
||||
+137
-2
@@ -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": {
|
||||
+7
@@ -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
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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<ReturnType<typeof seedTree>>) {
|
||||
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<ReturnType<typeof 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);
|
||||
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<void>((resolve) => { commitRelease = resolve; });
|
||||
let releaseWritten!: () => void;
|
||||
const written = new Promise<void>((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<typeof heartbeat.wakeup>[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<void>((resolve) => { preparationGate.reached = resolve; });
|
||||
|
||||
@@ -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<CancelSuppressedRetryResult> {
|
||||
// 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(
|
||||
|
||||
@@ -97,14 +97,23 @@ export function issueTreeControlRoutes(
|
||||
root: { id: string; companyId: string },
|
||||
holdId: string,
|
||||
actor: ReturnType<typeof getActorInfo>,
|
||||
/** 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 } : {}) });
|
||||
|
||||
@@ -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<HeartbeatRetryDependenc
|
||||
releaseIssueExecutionAndPromote: (run: Pick<HeartbeatRun, "id" | "companyId">, options?: { suppressImmediateRecovery?: boolean; resumeStrandedQueue?: boolean; deferredPostCommitEffects?: WakeQueuePostCommitEffect[] }) => Promise<unknown>;
|
||||
applyWakeQueuePostCommitEffects: (effects: WakeQueuePostCommitEffect[]) => Promise<void>;
|
||||
publishRunLifecyclePluginEvent: (run: HeartbeatRun) => void;
|
||||
cancelRunInternal: (runId: string, reason?: string, options?: { errorCode?: string }) => Promise<unknown>;
|
||||
cancelRunInternal: (runId: string, reason?: string,
|
||||
options?: { errorCode?: string; resultJson?: Record<string, unknown> }) => Promise<unknown>;
|
||||
cancelActiveForAgentInternal: (agentId: string, reason: string) => Promise<unknown>;
|
||||
resumeExecutionWaitComments: (scope?: DeferredReceiptRecoveryOptions) => Promise<unknown>;
|
||||
resumeQueuedCommentInterrupt: (companyId: string, queueId: string) => Promise<unknown>;
|
||||
@@ -799,6 +803,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
run: typeof heartbeatRuns.$inferSelect,
|
||||
companyAgents?: AgentOrgRow[],
|
||||
deferredPostCommitEffects?: WakeQueuePostCommitEffect[],
|
||||
deferredCancellations?: Array<() => Promise<void>>,
|
||||
) {
|
||||
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<void>> = [];
|
||||
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"));
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@@ -83,6 +83,24 @@ type RestoreTreeStatusResult = TreeStatusUpdateResult & {
|
||||
|
||||
const TERMINAL_ISSUE_STATUSES = new Set<IssueStatus>(["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<string, unknown> | null) {
|
||||
const marker = resultJson?.treeHoldSuppression;
|
||||
if (!marker || typeof marker !== "object" || Array.isArray(marker)) return null;
|
||||
const holdId = (marker as Record<string, unknown>).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 };
|
||||
|
||||
@@ -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<string | null>`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'`,
|
||||
|
||||
Reference in new issue
Block a user