mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 01:50:13 +01:00
fix(tree-control): resume work a subtree pause interrupted (#19394)
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work
> - Task tree control lets the board pause a task tree and then resume
it.
> - A pause stops the runs that are active in the tree. The release must
then continue the work that was pending.
> - A child task whose run was active at pause time did not start again
after release. The child became blocked and asked for a user message.
> - A pause is an operator hold, not a failure. It must not strand work,
and it must not require reconciliation for a run that never started.
> - This pull request re-dispatches interrupted work one time on
release, and it settles a legacy Stop that never entered the adapter.
> - The benefit is that pause and resume keep a task tree moving without
manual recovery, and real safety gates stay in force.
## Linked Issues or Issue Description
**What happened?**
A parent task delegated two children. One child's run was active (in
host workspace preparation, before the adapter started) when the board
paused the parent's tree. The pause cancelled that run. After the board
released the hold, no new run started for that child. The child became
`blocked` with the execution blocker cause
`legacy_execution_requires_reconciliation`. Later the next action became
"Send a message to continue." A sibling child that had no active run at
pause time resumed normally.
**Expected behavior**
A run that a subtree pause cancels is an operator hold. On release, the
pending work starts again one time, automatically. It does not go
through execution reconciliation when the run never entered its adapter.
It does not wait for a user message.
**Steps to reproduce**
1. Create a parent task and a child task that is assigned to an agent.
2. Wake the child. While its run is in host preparation (or queued),
call `POST /api/issues/:parentId/tree-holds` with `{ "mode": "pause" }`.
3. Call `POST /api/issues/:parentId/tree-holds/:holdId/release` without
`wakeAgents`.
4. Before this change, no new run starts, and `getExecutionBlocker`
reports `legacy_execution_requires_reconciliation` for the child.
**Root cause**
- (a) The pause records each interrupted run in the hold member snapshot
(`activeRunId`). Nothing used that record on release.
- (b) Release woke tasks only when `metadata.wakeAgents === true`.
Without it, release relied on stranded-issue recovery. That recovery
stands down when the latest run was cancelled by an operator
(`isOperatorCancelledRun`). So the interrupted child had no owner.
- (c) The pause Stop of a legacy run in preparation is written directly
as `cancelled`. Legacy terminalization classifies a cancelled run with
no bootstrap evidence as an unknown outcome. It records
`legacy_execution_requires_reconciliation`, although the gated dispatch
never entered the adapter.
- Native runs: the verified native Stop does not create the
reconciliation hold, but gap (b) applies to native runs too. The release
change is runtime-independent.
## What Changed
- `issueTreeControlService.findOwedPauseResume` decides if a task is
owed one resume. The task's latest run must be an operator Stop that a
pause hold member snapshot names, recorded while that hold was active.
The task must still be executable by the same agent, and no pause may
still cover it. The obligation belongs to the run, so overlapping pauses
keep it until the last covering release.
- New `services/issue-tree-resume.ts` re-dispatches owed work with wake
reason `issue_tree_resumed` (`holdId` and `interruptedRunId` in the
context). A compare-and-set claim (`treeHoldResume`) on the interrupted
run makes the resume happen at most once. A failed wake returns the
claim.
- Pause release and `mode: "resume"` call it for every member of the
released hold, also without `wakeAgents`. When `wakeAgents` is true, the
existing loop wakes every executable member.
- A task that still has an execution blocker is not replayed. The
release writes `issue.tree_resume_wake_skipped`, the blocker stays
visible, and the obligation stays unclaimed.
- `releaseHold` updates only an `active` hold. A repeated release gets
`409`.
- `settleCancelledLegacyPreparation`: when the executor that owned a
cancelled legacy run proves that dispatch never entered the adapter, it
records bootstrap evidence and resolves that run's
`legacy-execution:<runId>` hold as `false_positive`. It also corrects an
automatic no-replay disposition and restores a `blocked` status only
when that disposition made the change. Then it re-dispatches owed work,
for a release that happened while the Stop was unwinding. It requires a
requested Stop with no existing recovery receipt, and no process,
session, native finalization, unreleased lease, or execution-start
event.
- `settleUnrecoverableExecutions` waits while a legacy executor still
holds its controller lease, and records a `legacyCancellationBlock`
receipt when it blocks a task for a cancelled legacy run.
- Docs: `doc/execution-semantics.md` and `doc/composer-stop.md` describe
the resume rule.
## Verification
- New regression
`server/src/__tests__/issue-tree-pause-resume-run.test.ts` uses embedded
Postgres, the real heartbeat, and the real tree-control routes:
- A run paused during preparation: nothing runs during the hold, there
is no blocker, and release creates exactly one `issue_tree_resumed` run
that succeeds. A second release returns `409` and creates no run.
- Release before the stopped preparation finishes unwinding (with the
periodic disposition run in between): exactly one resumed run after the
executor settles.
- An automatic no-replay disposition that ran first: the blocker clears,
the status returns to `in_progress`, and release resumes once.
- Overlapping child and parent pauses: no resume while the parent pause
remains, then exactly one resume after it is released.
- A queued run cancelled by the pause: release creates exactly one
resumed run.
- A task reassigned to a user during the hold: no resume.
- A mid-turn legacy Stop with unknown outcome: no replay, the
reconciliation blocker remains, and `issue.tree_resume_wake_skipped` is
logged.
- Red before each fix: the first 4 cases failed on master; the 3 review
cases failed on the first PR commit. All 7 pass now.
- Also passed locally: `issue-tree-control-routes`,
`issue-tree-control-service`, `heartbeat-workspace-busy`,
`heartbeat-undispatched-execution-wait`, `execution-control-work`,
`legacy-workspace-restore-recovery`, `legacy-continuation-authority`,
`task-retry`, `legacy-execution-recovery`, `agent-conversations`,
`heartbeat-retry-scheduling`, `satisfied-connection-intents`.
- Not verified here: a live staging pause/resume. Re-check on staging
after deploy.
## Risks
- Behavior change: release now starts the work that the pause
interrupted, also without `wakeAgents`. Before, that work stayed idle.
Other idle tasks still wake only with `wakeAgents`.
- The settlement trusts the executor's adapter-entry flag. The flag is
set only inside the atomic dispatch gate, which refuses a cancelled run.
The settlement also requires no process, session, lease, or execution
event.
- A native run whose Stop has not finalized at release time is not a
target. Stranded recovery and explicit wake still apply.
## Model Used
- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude
Code with tool use (shell, file edit, 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
- [ ] All Paperclip CI gates are green
- [ ] 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
6ca7b151f7
commit
5c185db77c
17 files changed
+1131
-7
No files matched your search
@@ -18,7 +18,9 @@ terminal tasks. Resume/restore retain the optional wake-agents checkbox.
|
||||
|
||||
## Resume and execution recovery
|
||||
|
||||
Resume releases a hold. Waking agents is optional and only applies to tasks in
|
||||
Resume releases a hold. Work the pause interrupted is re-dispatched once
|
||||
automatically, under current admission gates (see `doc/execution-semantics.md`).
|
||||
Waking other agents is optional and only applies to tasks in
|
||||
`todo`, `in_progress`, or `in_review`; parked and terminal tasks stay untouched.
|
||||
The current execution-recovery policy requires verified outcomes before a stopped
|
||||
provider can restart. If any affected task still needs that reconciliation,
|
||||
|
||||
@@ -1879,6 +1879,30 @@ interrupt. A verified native stop must not become an unexplained failure simply
|
||||
because it came from a subtree action. The explicit pause hold still prevents
|
||||
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
|
||||
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
|
||||
latest run, assigned to the same agent, and executable; normal admission gates still
|
||||
apply. A compare-and-set claim on the interrupted run makes the resume happen at most
|
||||
once. The claim is saved as `pending` and arms the execution-continuation queue in
|
||||
the same commit; the wake uses the stable idempotency key `tree-hold-resume:<runId>`.
|
||||
If the process stops before that wake is saved, the queue's sweep finds the stale
|
||||
claim, rechecks current state, and dispatches it once. A task that still needs execution reconciliation is not replayed: the release
|
||||
records `issue.tree_resume_wake_skipped`, the blocker stays visible, and the
|
||||
obligation stays unclaimed.
|
||||
|
||||
A legacy run stopped before its gated dispatch entered the adapter has no provider
|
||||
actions to reconcile. Once its executor confirms no process, session, run-owned lease
|
||||
or execution event, it records bootstrap evidence and resolves that run's
|
||||
reconciliation hold as a false positive, including an automatic no-replay
|
||||
disposition. It restores a `blocked` status only when that disposition made the
|
||||
change (`legacyCancellationBlock` receipt). The executor then re-dispatches owed work
|
||||
if the pause was already released. The automatic disposition waits while a legacy
|
||||
executor still holds its controller lease. A hold can be released only once.
|
||||
|
||||
### Late provider usage
|
||||
|
||||
Each run measures its usage from the total the previous run accounted for, whether
|
||||
|
||||
@@ -592,6 +592,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
|
||||
"0414_closed_edwin_jarvis.sql",
|
||||
"0419_slippery_impossible_man.sql",
|
||||
"0432_strong_mole_man.sql",
|
||||
"0435_premium_puff_adder.sql",
|
||||
|
||||
|
||||
"0384_funny_adam_warlock.sql",
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS "agent_wakeup_requests_tree_hold_resume_uq" ON "agent_wakeup_requests" USING btree ("company_id","idempotency_key") WHERE "agent_wakeup_requests"."idempotency_key" LIKE 'tree-hold-resume:%' AND "agent_wakeup_requests"."status" <> 'skipped';
|
||||
+127
-2
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"id": "b900b172-2cdc-45c6-a964-27e850bc4c74",
|
||||
"prevId": "809726a1-762c-4c9d-a7fd-55670f1c856f",
|
||||
"id": "dedfce21-7ff9-4079-b5da-4a74a856a251",
|
||||
"prevId": "5b40e9ac-12aa-4757-aac7-36218c7500a8",
|
||||
"version": "7",
|
||||
"dialect": "postgresql",
|
||||
"tables": {
|
||||
@@ -713,6 +713,12 @@
|
||||
"notNull": true,
|
||||
"default": "'{}'::jsonb"
|
||||
},
|
||||
"previous_usage_json": {
|
||||
"name": "previous_usage_json",
|
||||
"type": "jsonb",
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"reason": {
|
||||
"name": "reason",
|
||||
"type": "text",
|
||||
@@ -4052,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": [
|
||||
@@ -22033,6 +22061,12 @@
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"reported_usage_json": {
|
||||
"name": "reported_usage_json",
|
||||
"type": "jsonb",
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"pricing_provenance": {
|
||||
"name": "pricing_provenance",
|
||||
"type": "jsonb",
|
||||
@@ -34174,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": {
|
||||
@@ -42436,6 +42514,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",
|
||||
@@ -42879,6 +42963,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": [
|
||||
@@ -43478,6 +43583,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",
|
||||
@@ -61828,6 +61946,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
@@ -3032,6 +3032,13 @@
|
||||
"when": 1793924947593,
|
||||
"tag": "0434_brave_leo",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 435,
|
||||
"version": "7",
|
||||
"when": 1793937048235,
|
||||
"tag": "0435_premium_puff_adder",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -90,6 +90,12 @@ export const agentWakeupRequests = pgTable(
|
||||
toolActionDeliveryIdempotencyUq: uniqueIndex("agent_wakeup_requests_tool_action_delivery_uq")
|
||||
.on(table.companyId, table.idempotencyKey)
|
||||
.where(sql`${table.idempotencyKey} LIKE 'tool-action-response:%' AND ${table.status} NOT IN ('skipped', 'failed', 'cancelled')`),
|
||||
// A subtree pause owes its interrupted run one resume wake, ever. A stale
|
||||
// dispatcher that saves late must not repeat work a takeover already ran.
|
||||
treeHoldResumeIdempotencyUq: uniqueIndex("agent_wakeup_requests_tree_hold_resume_uq")
|
||||
.on(table.companyId, table.idempotencyKey)
|
||||
.concurrently()
|
||||
.where(sql`${table.idempotencyKey} LIKE 'tree-hold-resume:%' AND ${table.status} <> 'skipped'`),
|
||||
chatCompletionIdempotencyUq: uniqueIndex("agent_wakeup_requests_chat_completion_uq")
|
||||
.on(table.companyId, table.idempotencyKey)
|
||||
.where(sql`${table.idempotencyKey} LIKE 'chat-completion:%'`),
|
||||
|
||||
@@ -16,6 +16,7 @@ const mockTreeControlService = vi.hoisted(() => ({
|
||||
getHold: vi.fn(),
|
||||
getActivePauseHoldGate: vi.fn(),
|
||||
releaseHold: vi.fn(),
|
||||
listPauseInterruptedResumeTargets: vi.fn(),
|
||||
cancelUnclaimedWakeupsForTree: vi.fn(),
|
||||
}));
|
||||
|
||||
@@ -79,6 +80,7 @@ describe("issue tree control routes", () => {
|
||||
companyId: "company-2",
|
||||
});
|
||||
mockTreeControlService.cancelUnclaimedWakeupsForTree.mockResolvedValue([]);
|
||||
mockTreeControlService.listPauseInterruptedResumeTargets.mockResolvedValue([]);
|
||||
mockTreeControlService.cancelIssueStatusesForHold.mockResolvedValue({
|
||||
updatedIssueIds: [],
|
||||
updatedIssues: [],
|
||||
|
||||
@@ -0,0 +1,527 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import express from "express";
|
||||
import request from "supertest";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
activityLog,
|
||||
agentRuntimeState,
|
||||
agentWakeupRequests,
|
||||
agents,
|
||||
authUsers,
|
||||
closeRegisteredClients,
|
||||
companies,
|
||||
companyMemberships,
|
||||
companySkills,
|
||||
costEvents,
|
||||
createDb,
|
||||
environmentLeases,
|
||||
heartbeatRunEvents,
|
||||
heartbeatRuns,
|
||||
issueRecoveryActions,
|
||||
issueTreeHoldMembers,
|
||||
issueTreeHolds,
|
||||
issues,
|
||||
} from "@paperclipai/db";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js";
|
||||
import { registerServerAdapter, unregisterServerAdapter } from "../adapters/index.ts";
|
||||
|
||||
// Hold one run in host preparation, before the legacy adapter is entered, so a
|
||||
// subtree pause can interrupt it exactly where the staging retry was stopped.
|
||||
const preparationGate = vi.hoisted(() => ({
|
||||
blockNext: false,
|
||||
blockedRunId: null as string | null,
|
||||
reached: null as null | (() => void),
|
||||
release: null as null | (() => void),
|
||||
}));
|
||||
vi.mock("../services/heartbeat/run-preparation.js", async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import("../services/heartbeat/run-preparation.js")>();
|
||||
return {
|
||||
...actual,
|
||||
buildPaperclipRuntimeMcpServers: async (input: Parameters<typeof actual.buildPaperclipRuntimeMcpServers>[0]) => {
|
||||
if (preparationGate.blockNext) {
|
||||
preparationGate.blockNext = false;
|
||||
preparationGate.blockedRunId = input.runId;
|
||||
await new Promise<void>((resolve) => {
|
||||
preparationGate.release = resolve;
|
||||
preparationGate.reached?.();
|
||||
});
|
||||
}
|
||||
return actual.buildPaperclipRuntimeMcpServers(input);
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
const externalTestDatabaseUrl = process.env.PAPERCLIP_TEST_DATABASE_URL?.trim();
|
||||
const embeddedPostgresSupport = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
const ADAPTER = "tree_pause_resume_test";
|
||||
|
||||
describeEmbeddedPostgres("subtree pause and release of an in-flight child run", () => {
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
let heartbeat!: ReturnType<typeof import("../services/heartbeat.ts").heartbeatService>;
|
||||
let getExecutionBlocker!: typeof import("../services/execution-blocker.js").getExecutionBlocker;
|
||||
let app!: express.Express;
|
||||
const executedRunIds: string[] = [];
|
||||
const adapterGate = { block: false, entered: null as null | (() => void) };
|
||||
|
||||
beforeAll(async () => {
|
||||
if (externalTestDatabaseUrl) db = createDb(externalTestDatabaseUrl);
|
||||
else {
|
||||
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-tree-pause-resume-");
|
||||
db = createDb(tempDb.connectionString);
|
||||
}
|
||||
const [{ heartbeatService }, blocker, { issueTreeControlRoutes }, { errorHandler }] = await Promise.all([
|
||||
import("../services/heartbeat.ts"),
|
||||
import("../services/execution-blocker.js"),
|
||||
import("../routes/issue-tree-control.js"),
|
||||
import("../middleware/index.js"),
|
||||
]);
|
||||
heartbeat = heartbeatService(db);
|
||||
getExecutionBlocker = blocker.getExecutionBlocker;
|
||||
registerServerAdapter({
|
||||
type: ADAPTER,
|
||||
execute: async (input) => {
|
||||
executedRunIds.push(input.runId);
|
||||
if (adapterGate.block) {
|
||||
adapterGate.entered?.();
|
||||
await new Promise<void>((resolve) => {
|
||||
if (input.signal?.aborted) return resolve();
|
||||
input.signal?.addEventListener("abort", () => resolve(), { once: true });
|
||||
});
|
||||
}
|
||||
// Finish the task so liveness follow-ups do not add unrelated runs.
|
||||
const issueId = (input.context as { issueId?: unknown }).issueId;
|
||||
if (typeof issueId === "string" && !input.signal?.aborted) {
|
||||
await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, issueId));
|
||||
}
|
||||
return { exitCode: 0, signal: null, timedOut: false, resultJson: {} };
|
||||
},
|
||||
testEnvironment: async () => ({ adapterType: ADAPTER, status: "pass", checks: [], testedAt: new Date().toISOString() }),
|
||||
});
|
||||
app = express();
|
||||
app.use(express.json());
|
||||
app.use((req, _res, next) => {
|
||||
(req as any).actor = { type: "board", userId: "board-user", source: "local_implicit", isInstanceAdmin: false };
|
||||
next();
|
||||
});
|
||||
app.use("/api", issueTreeControlRoutes(db));
|
||||
app.use(errorHandler);
|
||||
}, 30_000);
|
||||
|
||||
afterEach(async () => {
|
||||
preparationGate.release?.();
|
||||
Object.assign(preparationGate, { blockNext: false, blockedRunId: null, reached: null, release: null });
|
||||
adapterGate.block = false;
|
||||
adapterGate.entered = null;
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
executedRunIds.length = 0;
|
||||
for (const table of [costEvents, activityLog, issueRecoveryActions, issueTreeHoldMembers, issueTreeHolds,
|
||||
environmentLeases, heartbeatRunEvents]) await db.delete(table);
|
||||
await db.update(issues).set({ executionRunId: null, checkoutRunId: null });
|
||||
for (const table of [issues, heartbeatRuns, agentWakeupRequests, agentRuntimeState, agents, companySkills,
|
||||
companyMemberships, companies]) await db.delete(table);
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
unregisterServerAdapter(ADAPTER);
|
||||
if (externalTestDatabaseUrl) await closeRegisteredClients(externalTestDatabaseUrl);
|
||||
await db?.$client.end();
|
||||
await tempDb?.cleanup();
|
||||
});
|
||||
|
||||
async function seedTree() {
|
||||
const companyId = randomUUID();
|
||||
const agentId = randomUUID();
|
||||
const parentId = randomUUID();
|
||||
const childId = randomUUID();
|
||||
const prefix = `P${companyId.replace(/-/g, "").slice(0, 5).toUpperCase()}`;
|
||||
const now = new Date();
|
||||
await db.insert(companies).values({ id: companyId, name: "Pause co", issuePrefix: prefix,
|
||||
requireBoardApprovalForNewAgents: false, defaultResponsibleUserId: "responsible-user" });
|
||||
await db.insert(authUsers).values({ id: "responsible-user", name: "Responsible", email: "tree-pause@example.test",
|
||||
createdAt: now, updatedAt: now }).onConflictDoNothing();
|
||||
await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "responsible-user",
|
||||
status: "active", membershipRole: "member" });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Nibbler", role: "engineer", status: "idle",
|
||||
adapterType: ADAPTER, adapterConfig: {},
|
||||
runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } }, permissions: {} });
|
||||
await db.insert(issues).values({ id: parentId, companyId, title: "Parent", status: "in_progress", priority: "medium",
|
||||
responsibleUserId: "responsible-user", assigneeUserId: "responsible-user", issueNumber: 1, identifier: `${prefix}-1` });
|
||||
await db.insert(issues).values({ id: childId, companyId, parentId, title: "Child", status: "todo", priority: "medium",
|
||||
responsibleUserId: "responsible-user", assigneeAgentId: agentId, issueNumber: 2, identifier: `${prefix}-2` });
|
||||
return { companyId, agentId, parentId, childId, prefix };
|
||||
}
|
||||
|
||||
function wakeChild(fixture: Awaited<ReturnType<typeof seedTree>>) {
|
||||
return heartbeat.wakeup(fixture.agentId, {
|
||||
source: "assignment", triggerDetail: "system", reason: "issue_assigned",
|
||||
payload: { issueId: fixture.childId }, requestedByActorType: "system", requestedByActorId: "system",
|
||||
contextSnapshot: { issueId: fixture.childId, taskId: fixture.childId, wakeReason: "issue_assigned" },
|
||||
});
|
||||
}
|
||||
|
||||
async function childRuns(childId: string) {
|
||||
const rows = await db.select().from(heartbeatRuns);
|
||||
return rows.filter((run) => run.contextSnapshot?.issueId === childId)
|
||||
.sort((a, b) => a.createdAt.getTime() - b.createdAt.getTime());
|
||||
}
|
||||
|
||||
async function pauseAndRelease(
|
||||
fixture: Awaited<ReturnType<typeof seedTree>>,
|
||||
hooks: { afterPause?: () => Promise<void>; whilePaused?: () => Promise<void> } = {},
|
||||
) {
|
||||
const pause = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds`)
|
||||
.send({ mode: "pause", reason: "soak pause" });
|
||||
expect(pause.status).toBe(201);
|
||||
await hooks.afterPause?.();
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
await hooks.whilePaused?.();
|
||||
const release = await request(app)
|
||||
.post(`/api/issues/${fixture.parentId}/tree-holds/${pause.body.hold.id}/release`)
|
||||
.send({ reason: "soak resume" });
|
||||
expect(release.status).toBe(200);
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
return pause.body.hold.id as string;
|
||||
}
|
||||
|
||||
it("re-dispatches a child whose run was paused during preparation exactly once, without reconciliation", async () => {
|
||||
const fixture = await seedTree();
|
||||
const prepared = new Promise<void>((resolve) => { preparationGate.reached = resolve; });
|
||||
preparationGate.blockNext = true;
|
||||
await wakeChild(fixture);
|
||||
await prepared;
|
||||
const interruptedRunId = preparationGate.blockedRunId!;
|
||||
|
||||
const holdId = await pauseAndRelease(fixture, {
|
||||
// The executor observes the operator Stop at its dispatch gate.
|
||||
afterPause: async () => { preparationGate.release?.(); },
|
||||
whilePaused: async () => {
|
||||
// Nothing runs during the hold, and the Stop needs no reconciliation.
|
||||
expect((await childRuns(fixture.childId)).map((run) => run.status)).toEqual(["cancelled"]);
|
||||
expect(await getExecutionBlocker(db, fixture.companyId, fixture.childId)).toBeNull();
|
||||
},
|
||||
});
|
||||
|
||||
const runs = await childRuns(fixture.childId);
|
||||
expect(runs[0]).toMatchObject({ id: interruptedRunId, status: "cancelled" });
|
||||
expect(executedRunIds).not.toContain(interruptedRunId);
|
||||
const resumed = runs.slice(1);
|
||||
expect(resumed).toHaveLength(1);
|
||||
expect(resumed[0]).toMatchObject({ status: "succeeded" });
|
||||
expect(resumed[0]!.contextSnapshot).toMatchObject({ wakeReason: "issue_tree_resumed", holdId });
|
||||
expect(executedRunIds).toEqual([resumed[0]!.id]);
|
||||
const holds = await db.select().from(issueRecoveryActions).where(and(
|
||||
eq(issueRecoveryActions.sourceIssueId, fixture.childId), eq(issueRecoveryActions.status, "active")));
|
||||
expect(holds).toEqual([]);
|
||||
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");
|
||||
|
||||
// The released hold owes its work once; a repeated release cannot replay it.
|
||||
const again = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds/${holdId}/release`)
|
||||
.send({ reason: "duplicate resume" });
|
||||
expect(again.status).toBe(409);
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
expect(await childRuns(fixture.childId)).toHaveLength(2);
|
||||
});
|
||||
|
||||
async function pauseWhilePreparing(fixture: Awaited<ReturnType<typeof seedTree>>, rootId = fixture.parentId) {
|
||||
const prepared = new Promise<void>((resolve) => { preparationGate.reached = resolve; });
|
||||
preparationGate.blockNext = true;
|
||||
await wakeChild(fixture);
|
||||
await prepared;
|
||||
const pause = await request(app).post(`/api/issues/${rootId}/tree-holds`).send({ mode: "pause" });
|
||||
expect(pause.status).toBe(201);
|
||||
return { interruptedRunId: preparationGate.blockedRunId!, holdId: pause.body.hold.id as string };
|
||||
}
|
||||
|
||||
it("resumes once when the pause is released before the stopped preparation finishes unwinding", async () => {
|
||||
const fixture = await seedTree();
|
||||
const { interruptedRunId, holdId } = await pauseWhilePreparing(fixture);
|
||||
// The periodic disposition must not block a Stop its executor still owns.
|
||||
const { settleUnrecoverableExecutions } = await import("../services/execution-recovery-resolution.js");
|
||||
await settleUnrecoverableExecutions(db);
|
||||
const release = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds/${holdId}/release`).send({});
|
||||
expect(release.status).toBe(200);
|
||||
expect((await childRuns(fixture.childId)).map((run) => run.status)).toEqual(["cancelled"]);
|
||||
const [held] = await db.select().from(issues).where(eq(issues.id, fixture.childId));
|
||||
expect(held!.status).not.toBe("blocked");
|
||||
|
||||
preparationGate.release?.();
|
||||
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 });
|
||||
expect(await getExecutionBlocker(db, fixture.companyId, fixture.childId)).toBeNull();
|
||||
});
|
||||
|
||||
it("clears an automatic no-replay block recorded before the executor proved no adapter started", async () => {
|
||||
const fixture = await seedTree();
|
||||
const { interruptedRunId, holdId } = await pauseWhilePreparing(fixture);
|
||||
// Model a disposition that ran after the controller lease lapsed.
|
||||
await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(0) }).where(eq(heartbeatRuns.id, interruptedRunId));
|
||||
const { settleUnrecoverableExecutions } = await import("../services/execution-recovery-resolution.js");
|
||||
await settleUnrecoverableExecutions(db);
|
||||
const [blocked] = await db.select().from(issues).where(eq(issues.id, fixture.childId));
|
||||
expect(blocked!.status).toBe("blocked");
|
||||
|
||||
preparationGate.release?.();
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
expect(await getExecutionBlocker(db, fixture.companyId, fixture.childId)).toBeNull();
|
||||
const [restored] = await db.select().from(issues).where(eq(issues.id, fixture.childId));
|
||||
expect(restored!.status).toBe("in_progress");
|
||||
|
||||
const release = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds/${holdId}/release`).send({});
|
||||
expect(release.status).toBe(200);
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
const runs = await childRuns(fixture.childId);
|
||||
expect(runs.map((run) => run.contextSnapshot?.wakeReason)).toEqual(["issue_assigned", "issue_tree_resumed"]);
|
||||
});
|
||||
|
||||
it("keeps the interrupted work owed across overlapping pauses until the last one is released", async () => {
|
||||
const fixture = await seedTree();
|
||||
const { interruptedRunId, holdId: childHoldId } = await pauseWhilePreparing(fixture, fixture.childId);
|
||||
preparationGate.release?.();
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
const parentPause = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds`).send({ mode: "pause" });
|
||||
expect(parentPause.status).toBe(201);
|
||||
|
||||
const childRelease = await request(app)
|
||||
.post(`/api/issues/${fixture.childId}/tree-holds/${childHoldId}/release`).send({});
|
||||
expect(childRelease.status).toBe(200);
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
expect(await childRuns(fixture.childId)).toHaveLength(1);
|
||||
|
||||
const parentRelease = await request(app)
|
||||
.post(`/api/issues/${fixture.parentId}/tree-holds/${parentPause.body.hold.id}/release`).send({});
|
||||
expect(parentRelease.status).toBe(200);
|
||||
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]!.contextSnapshot).toMatchObject({ holdId: childHoldId, interruptedRunId });
|
||||
});
|
||||
|
||||
it("lets only one of two concurrent releases resume the interrupted work", async () => {
|
||||
const fixture = await seedTree();
|
||||
const { holdId } = await pauseWhilePreparing(fixture);
|
||||
preparationGate.release?.();
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
|
||||
const url = `/api/issues/${fixture.parentId}/tree-holds/${holdId}/release`;
|
||||
const responses = await Promise.all([request(app).post(url).send({}), request(app).post(url).send({})]);
|
||||
expect(responses.map((response) => response.status).sort()).toEqual([200, 409]);
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
const runs = await childRuns(fixture.childId);
|
||||
expect(runs.map((run) => run.contextSnapshot?.wakeReason)).toEqual(["issue_assigned", "issue_tree_resumed"]);
|
||||
});
|
||||
|
||||
it("finishes a resume whose wake was not saved before a restart, exactly once", async () => {
|
||||
const fixture = await seedTree();
|
||||
const { interruptedRunId, holdId } = await pauseWhilePreparing(fixture);
|
||||
preparationGate.release?.();
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
// The release claimed the work, then the process stopped before the wake was saved.
|
||||
const claimedAt = new Date(Date.now() - 5 * 60_000).toISOString();
|
||||
await db.update(issueTreeHolds).set({ status: "released", releasedAt: new Date() }).where(eq(issueTreeHolds.id, holdId));
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, interruptedRunId));
|
||||
await db.update(heartbeatRuns).set({ resultJson: { ...run!.resultJson, treeHoldResume: { holdId, state: "pending", claimedAt } } })
|
||||
.where(eq(heartbeatRuns.id, interruptedRunId));
|
||||
|
||||
const { deliverPendingPauseResumes, hasPendingPauseResume } = await import("../services/issue-tree-resume.js");
|
||||
expect(await hasPendingPauseResume(db)).toBe(true);
|
||||
await deliverPendingPauseResumes(db, (agentId, opts) => heartbeat.wakeup(agentId, opts));
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
await deliverPendingPauseResumes(db, (agentId, opts) => heartbeat.wakeup(agentId, opts), new Date(Date.now() + 5 * 60_000));
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
|
||||
const runs = await childRuns(fixture.childId);
|
||||
expect(runs.map((row) => row.contextSnapshot?.wakeReason)).toEqual(["issue_assigned", "issue_tree_resumed"]);
|
||||
expect(runs[1]).toMatchObject({ status: "succeeded" });
|
||||
expect(await hasPendingPauseResume(db)).toBe(false);
|
||||
});
|
||||
|
||||
it("does not let a stalled dispatcher repeat a resume the sweep already took over", async () => {
|
||||
const fixture = await seedTree();
|
||||
const { holdId } = await pauseWhilePreparing(fixture);
|
||||
preparationGate.release?.();
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
await db.update(issueTreeHolds).set({ status: "released", releasedAt: new Date() }).where(eq(issueTreeHolds.id, holdId));
|
||||
|
||||
// The first dispatcher claims the work, then stalls before its wake is saved.
|
||||
const { resumePauseInterruptedWork, deliverPendingPauseResumes } = await import("../services/issue-tree-resume.js");
|
||||
let resumeStalled!: () => void;
|
||||
const stalled = new Promise<void>((resolve) => { resumeStalled = resolve; });
|
||||
let continueStalled!: () => void;
|
||||
const gate = new Promise<void>((resolve) => { continueStalled = resolve; });
|
||||
const slow = resumePauseInterruptedWork(db, {
|
||||
companyId: fixture.companyId, issueIds: [fixture.childId], actor: { actorType: "system", actorId: "system" },
|
||||
wakeup: async (agentId, opts) => { resumeStalled(); await gate; return heartbeat.wakeup(agentId, opts); },
|
||||
});
|
||||
await stalled;
|
||||
|
||||
// The sweep takes over the stale claim and the replacement run finishes
|
||||
// while the task stays open.
|
||||
await deliverPendingPauseResumes(db, (agentId, opts) => heartbeat.wakeup(agentId, opts), new Date(Date.now() + 5 * 60_000));
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
await db.update(issues).set({ status: "in_progress", completedAt: null }).where(eq(issues.id, fixture.childId));
|
||||
|
||||
continueStalled();
|
||||
expect(await slow).toEqual([]);
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
const runs = await childRuns(fixture.childId);
|
||||
expect(runs.map((run) => run.contextSnapshot?.wakeReason)).toEqual(["issue_assigned", "issue_tree_resumed"]);
|
||||
const wakes = await db.select().from(agentWakeupRequests)
|
||||
.where(eq(agentWakeupRequests.idempotencyKey, `tree-hold-resume:${runs[0]!.id}`));
|
||||
expect(wakes).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("keeps a resume owed when a new pause lands before its wake is admitted", async () => {
|
||||
const fixture = await seedTree();
|
||||
const { holdId } = await pauseWhilePreparing(fixture);
|
||||
preparationGate.release?.();
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
await db.update(issueTreeHolds).set({ status: "released", releasedAt: new Date() }).where(eq(issueTreeHolds.id, holdId));
|
||||
|
||||
// A second pause lands between the eligibility check and queue admission.
|
||||
const { resumePauseInterruptedWork, deliverPendingPauseResumes } = await import("../services/issue-tree-resume.js");
|
||||
let secondHoldId = "";
|
||||
expect(await resumePauseInterruptedWork(db, {
|
||||
companyId: fixture.companyId, issueIds: [fixture.childId], actor: { actorType: "system", actorId: "system" },
|
||||
wakeup: async (agentId, opts) => {
|
||||
const pause = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds`).send({ mode: "pause" });
|
||||
expect(pause.status).toBe(201);
|
||||
secondHoldId = pause.body.hold.id;
|
||||
return heartbeat.wakeup(agentId, opts);
|
||||
},
|
||||
})).toEqual([]);
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
expect(await childRuns(fixture.childId)).toHaveLength(1);
|
||||
|
||||
const release = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds/${secondHoldId}/release`).send({});
|
||||
expect(release.status).toBe(200);
|
||||
await deliverPendingPauseResumes(db, (agentId, opts) => heartbeat.wakeup(agentId, opts), new Date(Date.now() + 5 * 60_000));
|
||||
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" });
|
||||
});
|
||||
|
||||
it("visits every stale resume claim even when the first page is still paused", async () => {
|
||||
const fixture = await seedTree();
|
||||
const { interruptedRunId, holdId } = await pauseWhilePreparing(fixture);
|
||||
preparationGate.release?.();
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
await db.update(issueTreeHolds).set({ status: "released", releasedAt: new Date() }).where(eq(issueTreeHolds.id, holdId));
|
||||
|
||||
// A full page of claims on tasks under another active pause sorts first.
|
||||
const pausedRootId = randomUUID();
|
||||
await db.insert(issues).values({ id: pausedRootId, companyId: fixture.companyId, title: "Paused root", status: "in_progress",
|
||||
priority: "medium", responsibleUserId: "responsible-user", assigneeUserId: "responsible-user", issueNumber: 10,
|
||||
identifier: `${fixture.prefix}-10` });
|
||||
const pausedChildIds = Array.from({ length: 25 }, () => randomUUID());
|
||||
for (const [index, id] of pausedChildIds.entries()) {
|
||||
await db.insert(issues).values({ id, companyId: fixture.companyId, parentId: pausedRootId, title: `Paused ${index}`,
|
||||
status: "in_progress", priority: "medium", responsibleUserId: "responsible-user", assigneeAgentId: fixture.agentId,
|
||||
issueNumber: 11 + index, identifier: `${fixture.prefix}-${11 + index}` });
|
||||
}
|
||||
const otherPause = await request(app).post(`/api/issues/${pausedRootId}/tree-holds`).send({ mode: "pause" });
|
||||
expect(otherPause.status).toBe(201);
|
||||
const claimedAt = new Date(Date.now() - 5 * 60_000).toISOString();
|
||||
const pendingClaim = { treeHoldResume: { holdId: otherPause.body.hold.id, state: "pending", claimedAt } };
|
||||
await db.insert(heartbeatRuns).values(pausedChildIds.map((issueId, index) => ({
|
||||
id: `00000000-0000-4000-8000-${String(index).padStart(12, "0")}`, companyId: fixture.companyId, agentId: fixture.agentId,
|
||||
status: "cancelled", finishedAt: new Date(Date.now() - 10 * 60_000), createdAt: new Date(Date.now() - 10 * 60_000),
|
||||
contextSnapshot: { issueId }, resultJson: pendingClaim,
|
||||
})));
|
||||
// The runnable claim sorts after that page.
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, interruptedRunId));
|
||||
await db.update(heartbeatRuns).set({ resultJson: { ...run!.resultJson, treeHoldResume: { holdId, state: "pending", claimedAt } } })
|
||||
.where(eq(heartbeatRuns.id, interruptedRunId));
|
||||
expect(interruptedRunId > `00000000-0000-4000-8000-${String(24).padStart(12, "0")}`).toBe(true);
|
||||
|
||||
const { deliverPendingPauseResumes } = await import("../services/issue-tree-resume.js");
|
||||
await deliverPendingPauseResumes(db, (agentId, opts) => heartbeat.wakeup(agentId, opts));
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
|
||||
const runs = await childRuns(fixture.childId);
|
||||
expect(runs.map((row) => row.contextSnapshot?.wakeReason)).toEqual(["issue_assigned", "issue_tree_resumed"]);
|
||||
});
|
||||
|
||||
it("re-dispatches a queued child run that the pause cancelled before it started", async () => {
|
||||
const fixture = await seedTree();
|
||||
// Occupy the agent's only slot so the child's run stays queued.
|
||||
const otherIssueId = randomUUID();
|
||||
await db.insert(issues).values({ id: otherIssueId, companyId: fixture.companyId, title: "Other", status: "todo",
|
||||
priority: "medium", responsibleUserId: "responsible-user", assigneeAgentId: fixture.agentId, issueNumber: 3,
|
||||
identifier: `${fixture.prefix}-3` });
|
||||
const prepared = new Promise<void>((resolve) => { preparationGate.reached = resolve; });
|
||||
preparationGate.blockNext = true;
|
||||
await heartbeat.wakeup(fixture.agentId, {
|
||||
source: "assignment", triggerDetail: "system", reason: "issue_assigned", payload: { issueId: otherIssueId },
|
||||
requestedByActorType: "system", requestedByActorId: "system",
|
||||
contextSnapshot: { issueId: otherIssueId, taskId: otherIssueId, wakeReason: "issue_assigned" },
|
||||
});
|
||||
await prepared;
|
||||
const queued = await wakeChild(fixture);
|
||||
expect(queued).toMatchObject({ status: "queued" });
|
||||
|
||||
const holdId = await pauseAndRelease(fixture, {
|
||||
afterPause: async () => {
|
||||
expect((await childRuns(fixture.childId)).map((run) => run.status)).toEqual(["cancelled"]);
|
||||
preparationGate.release?.();
|
||||
},
|
||||
whilePaused: async () => {
|
||||
expect((await childRuns(fixture.childId)).map((run) => run.status)).toEqual(["cancelled"]);
|
||||
},
|
||||
});
|
||||
|
||||
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: queued!.id });
|
||||
expect(await getExecutionBlocker(db, fixture.companyId, fixture.childId)).toBeNull();
|
||||
});
|
||||
|
||||
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; });
|
||||
preparationGate.blockNext = true;
|
||||
await wakeChild(fixture);
|
||||
await prepared;
|
||||
|
||||
await pauseAndRelease(fixture, {
|
||||
afterPause: async () => { preparationGate.release?.(); },
|
||||
whilePaused: async () => {
|
||||
await db.update(issues).set({ assigneeAgentId: null, assigneeUserId: "responsible-user" })
|
||||
.where(eq(issues.id, fixture.childId));
|
||||
},
|
||||
});
|
||||
|
||||
expect((await childRuns(fixture.childId)).map((run) => run.status)).toEqual(["cancelled"]);
|
||||
expect(executedRunIds).toEqual([]);
|
||||
});
|
||||
|
||||
it("keeps an uncertain mid-turn Stop held for reconciliation instead of replaying it on release", async () => {
|
||||
const fixture = await seedTree();
|
||||
const entered = new Promise<void>((resolve) => { adapterGate.entered = resolve; });
|
||||
adapterGate.block = true;
|
||||
const run = await wakeChild(fixture);
|
||||
await entered;
|
||||
|
||||
await pauseAndRelease(fixture, { whilePaused: async () => { adapterGate.block = false; } });
|
||||
|
||||
const runs = await childRuns(fixture.childId);
|
||||
expect(runs.map((row) => row.id)).toEqual([run!.id]);
|
||||
expect(runs[0]).toMatchObject({ status: "cancelled" });
|
||||
expect(await getExecutionBlocker(db, fixture.companyId, fixture.childId))
|
||||
.toMatchObject({ cause: "legacy_execution_requires_reconciliation", runId: run!.id });
|
||||
const skipped = await db.select().from(activityLog).where(eq(activityLog.action, "issue.tree_resume_wake_skipped"));
|
||||
expect(skipped).toHaveLength(1);
|
||||
expect(skipped[0]!.details).toMatchObject({ interruptedRunId: run!.id, cause: "legacy_execution_requires_reconciliation" });
|
||||
});
|
||||
});
|
||||
+5
-1
@@ -80,6 +80,7 @@ import { verifyStoppedNativeSessionForReplacement } from "./services/native-runt
|
||||
import { embeddedPostgresOwnerPort } from "./embedded-postgres-owner.js";
|
||||
import { deliverExecutionStatuses } from "./services/execution-status-delivery.js";
|
||||
import { deliverReconciledExecutions, settleUnrecoverableExecutions } from "./services/execution-recovery-resolution.js";
|
||||
import { deliverPendingPauseResumes } from "./services/issue-tree-resume.js";
|
||||
import { reconcileSafeNativeReplacements } from "./services/native-runtime/native-safe-replacement.js";
|
||||
import { reconcileAbandonedExecutionControl } from "./services/execution-control-reconciliation.js";
|
||||
import { acknowledgeNativeStartupCleanup } from "./services/native-runtime/native-startup-cleanup.js";
|
||||
@@ -1324,7 +1325,10 @@ async function startServerWithDatabaseTeardown(
|
||||
{ queue: DELIVERY_QUEUES.executionDisposition, run: () => settleUnrecoverableExecutions(db), next: executionSchedule.disposition },
|
||||
...(heartbeat ? [
|
||||
{ queue: DELIVERY_QUEUES.executionReplacement, run: () => reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession: run => verifyStoppedNativeSessionForReplacement(db, run) }), next: executionSchedule.replacement },
|
||||
{ queue: DELIVERY_QUEUES.executionContinuation, run: () => deliverReconciledExecutions(db, heartbeat.wakeup), next: executionSchedule.continuation },
|
||||
{ queue: DELIVERY_QUEUES.executionContinuation, run: async () => {
|
||||
await deliverReconciledExecutions(db, heartbeat.wakeup);
|
||||
await deliverPendingPauseResumes(db, heartbeat.wakeup);
|
||||
}, next: executionSchedule.continuation },
|
||||
] : []),
|
||||
], () => !heartbeatSchedulerStopped);
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@ import {
|
||||
} from "@paperclipai/db";
|
||||
import { and, eq, inArray, isNotNull } from "drizzle-orm";
|
||||
import { getExecutionBlocker } from "../services/execution-blocker.js";
|
||||
import { resumePauseInterruptedWork as resumeInterruptedWork } from "../services/issue-tree-resume.js";
|
||||
import { ISSUE_TREE_RESUME_EXECUTABLE_STATUSES } from "../services/issue-tree-control.js";
|
||||
import { conflict } from "../errors.js";
|
||||
import {
|
||||
@@ -92,6 +93,24 @@ export function issueTreeControlRoutes(
|
||||
return root;
|
||||
}
|
||||
|
||||
async function resumePauseInterruptedWork(
|
||||
root: { id: string; companyId: string },
|
||||
holdId: string,
|
||||
actor: ReturnType<typeof getActorInfo>,
|
||||
) {
|
||||
// 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 [];
|
||||
return resumeInterruptedWork(db, {
|
||||
companyId: root.companyId,
|
||||
issueIds: (hold.members ?? []).map((member) => member.issueId),
|
||||
rootIssueId: root.id,
|
||||
actor,
|
||||
wakeup: (agentId, opts) => heartbeat.wakeup(agentId, opts),
|
||||
});
|
||||
}
|
||||
|
||||
router.post("/issues/:id/tree-control/preview", validate(previewIssueTreeControlSchema), async (req, res) => {
|
||||
assertBoard(req);
|
||||
const root = await getAccessibleResource(req, res, resolveRootIssue(req), "Root issue not found");
|
||||
@@ -356,6 +375,14 @@ export function issueTreeControlRoutes(
|
||||
}
|
||||
}
|
||||
|
||||
if (result.hold.mode === "resume") {
|
||||
const wakeFailures: Array<{ issueId: string; message: string }> = [];
|
||||
for (const pauseHoldId of result.resumedPauseHoldIds ?? []) {
|
||||
wakeFailures.push(...await resumePauseInterruptedWork(root, pauseHoldId, actor));
|
||||
}
|
||||
if (wakeFailures.length) result = { ...result, wakeFailures } as typeof result;
|
||||
}
|
||||
|
||||
res
|
||||
.status(result.hold.mode === "restore" || result.hold.mode === "resume" ? 200 : 201)
|
||||
.json(result);
|
||||
@@ -544,6 +571,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));
|
||||
}
|
||||
|
||||
res.json({ ...hold, ...(wakeFailures.length ? { wakeFailures } : {}) });
|
||||
},
|
||||
);
|
||||
|
||||
@@ -6,6 +6,7 @@ import { type DeliveryQueue } from "./delivery-work-notifications.js";
|
||||
import { EXECUTION_RECONCILIATION_INTERVAL_MS } from "./execution-control-deadline.js";
|
||||
import { obsoleteExecutionConversationHold, unresolvedExecutionDispositions } from "./execution-recovery-resolution.js";
|
||||
import { nativeReplacementPending } from "./native-runtime/native-safe-replacement.js";
|
||||
import { hasPendingPauseResume } from "./issue-tree-resume.js";
|
||||
|
||||
export function registerExecutionControlWork(
|
||||
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
|
||||
@@ -81,7 +82,7 @@ export function executionControlSchedule(db: Db) {
|
||||
eq(issueRecoveryActions.status, "resolved"),
|
||||
sql`${issueRecoveryActions.evidence}->>'continuationDelivery' = 'pending'`,
|
||||
)).limit(1);
|
||||
return rows.length ? retry() : null;
|
||||
return rows.length || await hasPendingPauseResume(db) ? retry() : null;
|
||||
},
|
||||
async status() {
|
||||
const rows = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
|
||||
|
||||
@@ -600,6 +600,9 @@ export async function settleUnrecoverableExecutions(
|
||||
});
|
||||
return;
|
||||
}
|
||||
// A legacy executor still holding its controller lease may be unwinding
|
||||
// a Stop and can prove that no adapter started. Let it settle first.
|
||||
if (run.runtimeMode === "legacy" && run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > now) return;
|
||||
// Adoption transfers maintenance to the retained-workspace owner.
|
||||
// This compatibility resolver must not close its hold or relabel it
|
||||
// as a user repair merely because old return-owner metadata is absent.
|
||||
@@ -631,6 +634,7 @@ export async function settleUnrecoverableExecutions(
|
||||
await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: task.companyId, issueId: task.id });
|
||||
let nativeFailureBlock = action.evidence.nativeFailureBlock;
|
||||
let nativeBootstrapFailureBlock = action.evidence.nativeBootstrapFailureBlock;
|
||||
let legacyCancellationBlock = action.evidence.legacyCancellationBlock;
|
||||
if (current) {
|
||||
await notifyNativeStatusReassessment(tx, { companyId: task.companyId, issueId: task.id });
|
||||
if (task.status !== "blocked") await notifyChatResponseAvailability(tx, { companyId: task.companyId, issueId: task.id });
|
||||
@@ -651,6 +655,9 @@ export async function settleUnrecoverableExecutions(
|
||||
if (task.status !== "blocked" && run.runtimeMode === "native") {
|
||||
nativeFailureBlock = { runId: run.id, statusVersion: projected!.statusVersion };
|
||||
}
|
||||
if (task.status !== "blocked" && run.runtimeMode === "legacy" && run.status === "cancelled") {
|
||||
legacyCancellationBlock = { runId: run.id, statusVersion: projected!.statusVersion, previousStatus: task.status };
|
||||
}
|
||||
if (task.status !== "blocked" && isInterruptedNativePreparationCandidate(run)) {
|
||||
nativeBootstrapFailureBlock = { runId: run.id, statusVersion: projected!.statusVersion, previousStatus: task.status };
|
||||
}
|
||||
@@ -670,6 +677,7 @@ export async function settleUnrecoverableExecutions(
|
||||
...action.evidence,
|
||||
...(nativeFailureBlock ? { nativeFailureBlock } : {}),
|
||||
...(nativeBootstrapFailureBlock ? { nativeBootstrapFailureBlock } : {}),
|
||||
...(legacyCancellationBlock ? { legacyCancellationBlock } : {}),
|
||||
automaticRecovery: {
|
||||
policy: "preserve_without_replay_v1",
|
||||
runId: run.id,
|
||||
|
||||
@@ -303,7 +303,9 @@ import {
|
||||
|
||||
import {
|
||||
settleInterruptedNativeBootstrap,
|
||||
settleCancelledLegacyPreparation,
|
||||
} from "./legacy-execution-recovery.js";
|
||||
import { resumePauseInterruptedWork } from "./issue-tree-resume.js";
|
||||
import {
|
||||
adapterExecutionControls,
|
||||
createAdapterExecutionControl,
|
||||
@@ -3962,6 +3964,19 @@ export function heartbeatService(
|
||||
providerDispatchStarted: legacyAdapterEntered || nativeDispatchStarted || nativeOwnershipHeld,
|
||||
}) ?? latestRun;
|
||||
}
|
||||
if (latestRun?.status === "cancelled" && latestRun.runtimeMode === "legacy") {
|
||||
const settled = await settleCancelledLegacyPreparation(db, { run: latestRun,
|
||||
providerDispatchStarted: legacyAdapterEntered || nativeDispatchStarted || nativeOwnershipHeld,
|
||||
});
|
||||
latestRun = settled ?? latestRun;
|
||||
// A pause may already have been released while this Stop unwound.
|
||||
const settledIssueId = settled?.issueId ?? settled?.contextSnapshot?.issueId;
|
||||
if (typeof settledIssueId === "string") {
|
||||
await resumePauseInterruptedWork(db, { companyId: settled!.companyId, issueIds: [settledIssueId],
|
||||
actor: { actorType: "system", actorId: "system" }, wakeup: enqueueWakeup,
|
||||
}).catch(error => logger.warn({ err: error, runId: run.id }, "Could not resume work after a released pause"));
|
||||
}
|
||||
}
|
||||
if (latestRun?.status === "cancelled" && !nativeDispatchStarted && !nativeOwnershipHeld &&
|
||||
(latestRun.runtimeMode === "native" ||
|
||||
parseObject(latestRun.resultJson?.startupCancellation).beforeNativeSelection === true)) {
|
||||
|
||||
@@ -34,6 +34,7 @@ import {
|
||||
import { conflict, notFound, unprocessable } from "../errors.js";
|
||||
import type { IssuePostCommitAction } from "./issues.js";
|
||||
import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js";
|
||||
import { readRunCancellation } from "./run-cancellation.js";
|
||||
|
||||
type IssueRow = typeof issues.$inferSelect;
|
||||
type HoldRow = typeof issueTreeHolds.$inferSelect;
|
||||
@@ -1208,9 +1209,12 @@ export function issueTreeControlService(db: Db) {
|
||||
releaseMetadata: input.metadata ?? null,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(and(eq(issueTreeHolds.id, holdId), eq(issueTreeHolds.companyId, companyId)))
|
||||
.where(and(eq(issueTreeHolds.id, holdId), eq(issueTreeHolds.companyId, companyId),
|
||||
eq(issueTreeHolds.status, "active")))
|
||||
.returning();
|
||||
});
|
||||
// A concurrent release won; it alone owns the resumed work.
|
||||
if (!updated) throw conflict("Issue tree hold is already released");
|
||||
|
||||
const members = await db
|
||||
.select()
|
||||
@@ -1221,6 +1225,47 @@ export function issueTreeControlService(db: Db) {
|
||||
return toHold(updated, members);
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
async function findOwedPauseResume(companyId: string, issueId: string) {
|
||||
const [issue] = await db.select({
|
||||
status: issues.status, assigneeAgentId: issues.assigneeAgentId, assigneeUserId: issues.assigneeUserId,
|
||||
}).from(issues).where(and(eq(issues.companyId, companyId), eq(issues.id, issueId)));
|
||||
if (!issue?.assigneeAgentId || issue.assigneeUserId
|
||||
|| !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,
|
||||
}).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)
|
||||
.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`),
|
||||
)).orderBy(asc(issueTreeHolds.createdAt)).limit(1);
|
||||
if (!hold) return null;
|
||||
if (await getActivePauseHoldGate(companyId, issueId)) return null;
|
||||
return { issueId, agentId: run.agentId, interruptedRunId: run.id, holdId: hold.id };
|
||||
}
|
||||
|
||||
async function cancelUnclaimedWakeupsForTree(companyId: string, rootIssueId: string, reason: string) {
|
||||
const treeIssues = await listTreeIssues(companyId, rootIssueId);
|
||||
const issueIds = treeIssues.map((issue) => issue.id);
|
||||
@@ -1268,6 +1313,7 @@ export function issueTreeControlService(db: Db) {
|
||||
listHolds,
|
||||
getActivePauseHoldGate,
|
||||
releaseHold,
|
||||
findOwedPauseResume,
|
||||
cancelUnclaimedWakeupsForTree,
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,239 @@
|
||||
import { and, asc, eq, gt, ne, sql } from "drizzle-orm";
|
||||
import { agentWakeupRequests, heartbeatRuns, 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";
|
||||
import { getExecutionBlocker } from "./execution-blocker.js";
|
||||
import { ISSUE_TREE_RESUME_EXECUTABLE_STATUSES, issueTreeControlService } from "./issue-tree-control.js";
|
||||
|
||||
type ResumeActor = {
|
||||
actorType: "user" | "agent" | "system";
|
||||
actorId: string;
|
||||
agentId?: string | null;
|
||||
runId?: string | null;
|
||||
agentApiKeyId?: string | null;
|
||||
};
|
||||
|
||||
type Wakeup = (agentId: string, opts: {
|
||||
source: "assignment";
|
||||
triggerDetail: "system";
|
||||
reason: string;
|
||||
payload: Record<string, unknown>;
|
||||
requestedByActorType: ResumeActor["actorType"];
|
||||
requestedByActorId: string;
|
||||
idempotencyKey: string;
|
||||
contextSnapshot: Record<string, unknown>;
|
||||
}) => Promise<unknown>;
|
||||
|
||||
/** A claim older than this with no saved wake belongs to an interrupted dispatcher. */
|
||||
export const PAUSE_RESUME_CLAIM_STALE_MS = 60_000;
|
||||
|
||||
/** Queue admission accepts one saved wake per interrupted run under this key. */
|
||||
export const PAUSE_RESUME_WAKE_INDEX = "agent_wakeup_requests_tree_hold_resume_uq";
|
||||
|
||||
const SWEEP_PAGE_SIZE = 25;
|
||||
|
||||
export function pauseResumeWakeKey(runId: string) {
|
||||
return `tree-hold-resume:${runId}`;
|
||||
}
|
||||
|
||||
const SYSTEM_ACTOR: ResumeActor = { actorType: "system", actorId: "system" };
|
||||
|
||||
function activity(db: Db, companyId: string, actor: ResumeActor, action: string, issueId: string,
|
||||
details: Record<string, unknown>) {
|
||||
return logActivity(db, {
|
||||
companyId, actorType: actor.actorType, actorId: actor.actorId, agentId: actor.agentId ?? null,
|
||||
runId: actor.runId ?? null, agentApiKeyId: actor.agentApiKeyId ?? null, action,
|
||||
entityType: "issue", entityId: issueId, details,
|
||||
});
|
||||
}
|
||||
|
||||
/** Compare-and-set the run's resume record. Only the writer that wins dispatches. */
|
||||
async function setResumeState(db: Db, input: {
|
||||
companyId: string; runId: string; next: Record<string, unknown>; expectedClaimedAt?: string | null;
|
||||
}) {
|
||||
const guard = input.expectedClaimedAt === undefined
|
||||
? sql`not (coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) ? 'treeHoldResume')`
|
||||
: sql`${heartbeatRuns.resultJson}->'treeHoldResume'->>'claimedAt' is not distinct from ${input.expectedClaimedAt}`;
|
||||
return db.transaction(async tx => {
|
||||
// A pending claim arms the recovery sweep in the same commit, so a crash
|
||||
// before the wake is saved is retried. Intent is registered before the write.
|
||||
if (input.next.state === "pending") await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionContinuation);
|
||||
const [row] = await tx.update(heartbeatRuns).set({
|
||||
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ treeHoldResume: input.next })}::jsonb`,
|
||||
}).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId),
|
||||
eq(heartbeatRuns.status, "cancelled"), guard)).returning({ id: heartbeatRuns.id });
|
||||
return Boolean(row);
|
||||
});
|
||||
}
|
||||
|
||||
/** A saved, admitted resume wake (queued, deferred, coalesced, or run) for the interrupted run. */
|
||||
async function findPauseResumeReceipt(db: Db, companyId: string, agentId: string, runId: string) {
|
||||
const [receipt] = await db.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, agentId),
|
||||
eq(agentWakeupRequests.idempotencyKey, pauseResumeWakeKey(runId)),
|
||||
ne(agentWakeupRequests.status, "skipped"))).limit(1);
|
||||
return Boolean(receipt);
|
||||
}
|
||||
|
||||
async function dispatch(db: Db, input: {
|
||||
companyId: string; issueId: string; agentId: string; runId: string; holdId: string;
|
||||
claimedAt: string; rootIssueId: string | null; actor: ResumeActor; wakeup: Wakeup;
|
||||
}) {
|
||||
const saved = await input.wakeup(input.agentId, {
|
||||
source: "assignment",
|
||||
triggerDetail: "system",
|
||||
reason: "issue_tree_resumed",
|
||||
payload: { issueId: input.issueId, rootIssueId: input.rootIssueId, holdId: input.holdId, interruptedRunId: input.runId },
|
||||
requestedByActorType: input.actor.actorType,
|
||||
requestedByActorId: input.actor.actorId,
|
||||
// The saved wake request is the durable receipt the recovery sweep checks.
|
||||
idempotencyKey: pauseResumeWakeKey(input.runId),
|
||||
contextSnapshot: {
|
||||
issueId: input.issueId,
|
||||
taskId: input.issueId,
|
||||
wakeReason: "issue_tree_resumed",
|
||||
source: "issue.tree_resume",
|
||||
rootIssueId: input.rootIssueId,
|
||||
holdId: input.holdId,
|
||||
interruptedRunId: input.runId,
|
||||
},
|
||||
}).then((run) => run ? "saved" as const : "not_admitted" as const, (error: unknown) => {
|
||||
// Another dispatcher already saved this run's resume wake; it owns the work.
|
||||
if (isUniqueViolation(error, PAUSE_RESUME_WAKE_INDEX)) return "duplicate" as const;
|
||||
throw error;
|
||||
});
|
||||
if (saved === "duplicate") return;
|
||||
// Admission can skip the wake, for example when a new pause landed after the
|
||||
// eligibility check. A skipped request is no receipt, so the claim stays
|
||||
// pending and the sweep retries once the task is runnable again.
|
||||
if (saved === "not_admitted" && !await findPauseResumeReceipt(db, input.companyId, input.agentId, input.runId)) return;
|
||||
await setResumeState(db, { companyId: input.companyId, runId: input.runId, expectedClaimedAt: input.claimedAt,
|
||||
next: { holdId: input.holdId, state: "delivered", claimedAt: input.claimedAt, deliveredAt: new Date().toISOString() } });
|
||||
}
|
||||
|
||||
/**
|
||||
* Re-dispatch, once, the work a subtree pause interrupted on each task that is
|
||||
* no longer paused. Both a hold release and the executor that settles a
|
||||
* cancelled preparation call this, so neither ordering loses the work. A task
|
||||
* that still needs execution reconciliation keeps its blocker and is skipped;
|
||||
* its obligation stays unclaimed for a later settlement or release. A claim
|
||||
* whose wake was never saved (for example after a restart) stays pending and
|
||||
* is finished by `deliverPendingPauseResumes`.
|
||||
*/
|
||||
export async function resumePauseInterruptedWork(
|
||||
db: Db,
|
||||
input: { companyId: string; issueIds: string[]; actor: ResumeActor; wakeup: Wakeup; rootIssueId?: string | null },
|
||||
) {
|
||||
const treeControl = issueTreeControlService(db);
|
||||
const failures: Array<{ issueId: string; message: string }> = [];
|
||||
for (const issueId of [...new Set(input.issueIds)]) {
|
||||
try {
|
||||
const target = await treeControl.findOwedPauseResume(input.companyId, issueId);
|
||||
if (!target) continue;
|
||||
const blocker = await getExecutionBlocker(db, input.companyId, issueId);
|
||||
if (blocker) {
|
||||
await activity(db, input.companyId, input.actor, "issue.tree_resume_wake_skipped", issueId, {
|
||||
holdId: target.holdId, interruptedRunId: target.interruptedRunId, cause: blocker.cause,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
const claimedAt = new Date().toISOString();
|
||||
if (!await setResumeState(db, { companyId: input.companyId, runId: target.interruptedRunId,
|
||||
next: { holdId: target.holdId, state: "pending", claimedAt } })) continue;
|
||||
await dispatch(db, { companyId: input.companyId, issueId, agentId: target.agentId, runId: target.interruptedRunId,
|
||||
holdId: target.holdId, claimedAt, rootIssueId: input.rootIssueId ?? null, actor: input.actor, wakeup: input.wakeup });
|
||||
} catch (error) {
|
||||
// A pending claim stays recoverable; the sweep retries it after it goes stale.
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
failures.push({ issueId, message });
|
||||
await Promise.resolve(activity(db, input.companyId, input.actor, "issue.tree_resume_wake_failed", issueId,
|
||||
{ error: message })).catch(() => null);
|
||||
}
|
||||
}
|
||||
return failures;
|
||||
}
|
||||
|
||||
function pendingResumePredicate(staleBefore: Date) {
|
||||
return and(eq(heartbeatRuns.status, "cancelled"),
|
||||
sql`${heartbeatRuns.resultJson}->'treeHoldResume'->>'state' = 'pending'`,
|
||||
sql`(${heartbeatRuns.resultJson}->'treeHoldResume'->>'claimedAt')::timestamptz <= ${staleBefore.toISOString()}::timestamptz`);
|
||||
}
|
||||
|
||||
/** True while a claimed resume has no durable receipt yet. */
|
||||
export async function hasPendingPauseResume(db: Db) {
|
||||
const [row] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
|
||||
.where(sql`${heartbeatRuns.status} = 'cancelled' and ${heartbeatRuns.resultJson}->'treeHoldResume'->>'state' = 'pending'`)
|
||||
.limit(1);
|
||||
return Boolean(row);
|
||||
}
|
||||
|
||||
/**
|
||||
* Finish resume claims whose dispatcher stopped between claiming and saving the
|
||||
* wake (for example a restart). A saved wake with the stable key is the receipt.
|
||||
* Otherwise the task's current state is checked again before one dispatch.
|
||||
*/
|
||||
export async function deliverPendingPauseResumes(db: Db, wakeup: Wakeup, now = new Date()) {
|
||||
const staleBefore = new Date(now.getTime() - PAUSE_RESUME_CLAIM_STALE_MS);
|
||||
// Page through the stale claims: a claim still waiting on a pause or blocker
|
||||
// is skipped without a write, so any fixed batch could starve the rest. Each
|
||||
// pass visits every stale claim in bounded pages.
|
||||
let after: string | null = null;
|
||||
for (;;) {
|
||||
const rows: PendingResumeRow[] = await db.select({
|
||||
id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
|
||||
createdAt: heartbeatRuns.createdAt, resultJson: heartbeatRuns.resultJson,
|
||||
issueId: sql<string | null>`coalesce(${heartbeatRuns.issueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId')`,
|
||||
}).from(heartbeatRuns)
|
||||
.where(and(pendingResumePredicate(staleBefore), after ? gt(heartbeatRuns.id, after) : undefined))
|
||||
.orderBy(asc(heartbeatRuns.id)).limit(SWEEP_PAGE_SIZE);
|
||||
if (!rows.length) return;
|
||||
after = rows[rows.length - 1]!.id;
|
||||
await deliverPendingPage(db, wakeup, now, rows);
|
||||
if (rows.length < SWEEP_PAGE_SIZE) return;
|
||||
}
|
||||
}
|
||||
|
||||
type PendingResumeRow = {
|
||||
id: string; companyId: string; agentId: string; createdAt: Date;
|
||||
resultJson: Record<string, unknown> | null; issueId: string | null;
|
||||
};
|
||||
|
||||
async function deliverPendingPage(db: Db, wakeup: Wakeup, now: Date, rows: PendingResumeRow[]) {
|
||||
for (const run of rows) {
|
||||
const claim = run.resultJson?.treeHoldResume as { holdId?: unknown; claimedAt?: unknown } | undefined;
|
||||
if (typeof claim?.claimedAt !== "string" || typeof claim.holdId !== "string" || !run.issueId) continue;
|
||||
const finish = (state: "delivered" | "abandoned", reason?: string) => setResumeState(db, {
|
||||
companyId: run.companyId, runId: run.id, expectedClaimedAt: claim.claimedAt as string,
|
||||
next: { holdId: claim.holdId, state, claimedAt: claim.claimedAt, settledAt: now.toISOString(), ...(reason ? { reason } : {}) },
|
||||
});
|
||||
try {
|
||||
if (await findPauseResumeReceipt(db, run.companyId, run.agentId, run.id)) { await finish("delivered"); continue; }
|
||||
const [issue] = await db.select({ status: issues.status, assigneeAgentId: issues.assigneeAgentId,
|
||||
assigneeUserId: issues.assigneeUserId }).from(issues)
|
||||
.where(and(eq(issues.companyId, run.companyId), eq(issues.id, run.issueId)));
|
||||
const [newer] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, run.companyId),
|
||||
sql`coalesce(${heartbeatRuns.issueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${run.issueId}`,
|
||||
// JavaScript dates drop microseconds, so exclude the run itself by id.
|
||||
sql`${heartbeatRuns.id} <> ${run.id}`,
|
||||
sql`${heartbeatRuns.createdAt} >= ${run.createdAt.toISOString()}::timestamptz`)).limit(1);
|
||||
if (!issue || issue.assigneeAgentId !== run.agentId || issue.assigneeUserId || newer
|
||||
|| !ISSUE_TREE_RESUME_EXECUTABLE_STATUSES.includes(issue.status)) {
|
||||
await finish("abandoned", "task_changed");
|
||||
continue;
|
||||
}
|
||||
// A paused or blocked task keeps its claim; release or settlement revisits it.
|
||||
if (await issueTreeControlService(db).getActivePauseHoldGate(run.companyId, run.issueId)
|
||||
|| await getExecutionBlocker(db, run.companyId, run.issueId)) continue;
|
||||
const claimedAt = now.toISOString();
|
||||
if (!await setResumeState(db, { companyId: run.companyId, runId: run.id, expectedClaimedAt: claim.claimedAt,
|
||||
next: { holdId: claim.holdId, state: "pending", claimedAt } })) continue;
|
||||
await dispatch(db, { companyId: run.companyId, issueId: run.issueId, agentId: run.agentId, runId: run.id,
|
||||
holdId: claim.holdId, claimedAt, rootIssueId: null, actor: SYSTEM_ACTOR, wakeup });
|
||||
} catch (error) {
|
||||
await Promise.resolve(activity(db, run.companyId, SYSTEM_ACTOR, "issue.tree_resume_wake_failed", run.issueId,
|
||||
{ error: error instanceof Error ? error.message : String(error), interruptedRunId: run.id })).catch(() => null);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -16,7 +16,10 @@ import { claimedAdapterType, hasConversationContinuationPolicy } from "./convers
|
||||
import { canonicalExecutionUuid } from "./execution-identity.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, desc, eq, getTableColumns, inArray, isNull, or, sql, type SQL } from "drizzle-orm";
|
||||
import { environmentLeases, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { environmentLeases, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { executionStartedEventPredicate } from "./pre-dispatch-execution-evidence.js";
|
||||
import { readRunCancellation } from "./run-cancellation.js";
|
||||
import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js";
|
||||
import { issueRecoveryActionService } from "./issue-recovery-actions.js";
|
||||
import { parseIssueExecutionState } from "./issue-execution-policy.js";
|
||||
import { executionFailureRetryCount } from "./execution-recovery-attempt.js";
|
||||
@@ -382,6 +385,87 @@ export async function settleInterruptedNativeBootstrap(
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* The executor that owned a cancelled legacy run proves that its gated
|
||||
* dispatch never entered the adapter. Such a Stop (for example a subtree pause
|
||||
* during workspace preparation) left no provider actions to reconcile, so the
|
||||
* hold that terminalization conservatively recorded is a false positive. Every
|
||||
* retained execution reference must agree; otherwise the hold stays in force.
|
||||
*/
|
||||
export async function settleCancelledLegacyPreparation(
|
||||
db: Db, input: { run: Run; providerDispatchStarted: boolean },
|
||||
): Promise<Run | null> {
|
||||
if (input.providerDispatchStarted || input.run.runtimeMode !== "legacy" || input.run.status !== "cancelled") return null;
|
||||
const issueId = input.run.issueId ?? input.run.contextSnapshot?.issueId;
|
||||
if (typeof issueId !== "string") return null;
|
||||
return db.transaction(async tx => {
|
||||
const [task] = await tx.select().from(issues)
|
||||
.where(and(eq(issues.id, issueId), eq(issues.companyId, input.run.companyId))).for("update");
|
||||
const [run] = await tx.select().from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.id, input.run.id), eq(heartbeatRuns.companyId, input.run.companyId))).for("update");
|
||||
const recovery = run?.resultJson?.executionRecovery as Record<string, unknown> | undefined;
|
||||
// Only a requested Stop qualifies, and only while it is still classified as
|
||||
// an unknown outcome: existing wait or bootstrap receipts already decide it.
|
||||
if (!task || !run || run.runtimeMode !== "legacy" || run.status !== "cancelled" || !run.finishedAt
|
||||
|| readRunCancellation(run.resultJson)?.expected !== true || recovery !== undefined
|
||||
|| !legacyExecutionNeedsReconciliation(run)
|
||||
|| run.processPid || run.processGroupId || run.processStartedAt || run.sessionIdAfter
|
||||
|| run.nativeIssueId || run.nativeSessionId
|
||||
|| hasWorkspaceRestoreFailure(run.resultJson) || hasRequiredWorkspaceRecovery(run.resultJson)
|
||||
|| run.runnerProfileJson?.nativeExecutionInput != null) return null;
|
||||
const [native] = await tx.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations)
|
||||
.where(and(eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.runId, run.id))).limit(1);
|
||||
const leases = await tx.select().from(environmentLeases)
|
||||
.where(and(eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id)));
|
||||
const [execution] = await tx.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and(
|
||||
eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), executionStartedEventPredicate(),
|
||||
)).limit(1);
|
||||
if (native || execution || leases.some(lease => lease.provider === "local"
|
||||
? !lease.releasedAt || lease.status === "pending_cleanup" || lease.cleanupStatus === "failed"
|
||||
: !hasRemoteTerminationReceipt(lease))) return null;
|
||||
const now = new Date();
|
||||
const [settled] = await tx.update(heartbeatRuns).set({ resultJson: { ...run.resultJson,
|
||||
executionRecovery: { kind: "bootstrap", providerWorkStarted: false, preparationSettledAt: now.toISOString() },
|
||||
}, updatedAt: now }).where(eq(heartbeatRuns.id, run.id)).returning();
|
||||
// The automatic disposition may already have closed the incident with
|
||||
// replay blocked, and moved the task to blocked. Correct both verdicts.
|
||||
const holds = await tx.select().from(issueRecoveryActions).where(and(
|
||||
eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, task.id),
|
||||
eq(issueRecoveryActions.cause, LEGACY_RECOVERY_CAUSE), eq(issueRecoveryActions.fingerprint, `legacy-execution:${run.id}`),
|
||||
sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`,
|
||||
sql`not (${issueRecoveryActions.evidence} ? 'executionReconciliation')`,
|
||||
or(inArray(issueRecoveryActions.status, ["active", "escalated"]), and(
|
||||
eq(issueRecoveryActions.status, "resolved"), eq(issueRecoveryActions.outcome, "blocked"),
|
||||
sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'policy' = 'preserve_without_replay_v1'`,
|
||||
sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`,
|
||||
)),
|
||||
)).for("update");
|
||||
const ownedBlock = holds.map(hold => hold.evidence.legacyCancellationBlock as Record<string, unknown> | undefined)
|
||||
.find(receipt => receipt?.runId === run.id && receipt.statusVersion === task.statusVersion &&
|
||||
["todo", "in_progress", "in_review"].includes(String(receipt.previousStatus)));
|
||||
if (task.status === "blocked" && ownedBlock && task.assigneeAgentId === run.agentId &&
|
||||
!isSupersededConversationRun(task, run) && !task.executionRunId && !task.checkoutRunId) {
|
||||
// Restore only the unchanged projection made by this exact incident.
|
||||
await issueService(tx as unknown as Db).update(task.id, { status: String(ownedBlock.previousStatus) }, tx);
|
||||
}
|
||||
const resolved = holds.length ? await tx.update(issueRecoveryActions).set({ status: "resolved", outcome: "false_positive",
|
||||
resolutionNote: "The run was stopped before its adapter started; there are no provider actions to reconcile.",
|
||||
resolvedAt: now, updatedAt: now,
|
||||
evidence: sql`case when ${issueRecoveryActions.evidence} ? 'automaticRecovery'
|
||||
then jsonb_set(${issueRecoveryActions.evidence}, '{automaticRecovery,replay}', '"allowed"'::jsonb, false)
|
||||
else ${issueRecoveryActions.evidence} end`,
|
||||
}).where(inArray(issueRecoveryActions.id, holds.map(hold => hold.id))).returning({ id: issueRecoveryActions.id }) : [];
|
||||
if (resolved.length) {
|
||||
await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: run.companyId, issueId: task.id });
|
||||
await logActivity(tx as unknown as Db, { companyId: run.companyId, actorType: "system", actorId: "system",
|
||||
action: "issue.recovery_action_resolved", entityType: "issue", entityId: task.id, agentId: run.agentId, runId: run.id,
|
||||
details: { reason: "legacy_cancelled_before_adapter", recoveryActionIds: resolved.map(r => r.id) },
|
||||
});
|
||||
}
|
||||
return settled;
|
||||
});
|
||||
}
|
||||
|
||||
/** Capture the required-file obligation before unrelated host writes. Keeps
|
||||
* execution ownership while the adapter's host finalization is still running. */
|
||||
export async function recordLegacyWorkspaceRestoreFailure(db: Db, run: Run, evidence: Record<string, unknown>, source?: LegacyWorkspaceRestoreSource) {
|
||||
|
||||
Reference in new issue
Block a user