feat: let user messages continue stopped native tasks (#13239)
## Thinking Path > - Paperclip manages AI agents and their tasks. > - A failed native run can leave a durable execution hold. > - The hold prevents automatic replay of actions with unknown outcomes. > - It can also prevent the agent from answering a new user message. > - A new user message should authorize a fresh turn after the prior execution stops. > - This pull request adds that admission path and preserves the existing execution gates. > - Users can continue the conversation without certifying every past action. ## Linked Issues or Issue Description **Subsystem affected** Server task wake admission and native execution recovery. **Problem or motivation** A native task can remain blocked after automatic recovery stops. A new user message is saved, but its run is cancelled before the agent can answer. **Proposed solution** Use a new authenticated user comment to authorize a fresh turn. Check stopped predecessor ownership and available history. Retain uncertain action outcomes. Commit the new run and hold retirement together. **Roadmap alignment** This is a focused improvement to the existing self-healing runs and recovery behavior. Builds on merged #13237, which covers legacy conversation continuation. This PR adds native admission and preserves native automatic-recovery eligibility and budgets. ## What Changed - Admit a fresh native turn for a new user comment after every held predecessor has stopped. - Validate the comment author, task, timing, process ownership, controller, and cleanup leases. - Preserve failed runs, unknown action outcomes, and the failed incident's attempt count. - Record the new comment and run in the existing recovery audit history. - Validate the saved continuation source and discard consumed user-wake authority from later automatic replacements. - Keep pause, approval, budget, ownership, and dependency interaction rules. - Add database and actual wake-path regressions. Update the execution contract. ## Verification - [Full CI run 34626750213](https://github.com/paperclipai/paperclip/actions/runs/34626750213) passed on `c58e6c087e0df7530c747d80b27d491da925a9c4`: all 31 reported checks passed, including all server/workspace suites, browser shards, native runner verification, build, typecheck, release dry run, and aggregate gates. The two conditional Storybook checks were skipped. - Greptile reviewed this exact head at 5/5. All review threads are resolved, and security checks passed. - All 218 local targeted tests passed across explicit native continuation, continuation history, safe replacement, durable chat wakeups, wake queue, issue liveness, native session resume, and run dispatch. The native implementation is unchanged by the final rebase onto master. - Full workspace `pnpm -r typecheck` and `pnpm build` passed on the final head. Complete test coverage is supplied by the green CI suites; local tests used the targeted suites above. - Regressions cover scoped authorization, concurrent delivery, live ownership, later admission gates, retained message receipts, and automatic replacement after terminal-task or reviewer changes. ## Risks - A fresh model turn can choose to repeat an action. Paperclip preserves prior history and does not replay recorded calls. - Missing process identity and remote ownership without a target-aware stop proof retain the hold. A terminal database row alone does not prove that execution stopped. - No schema or dependency changes. Existing historical tasks are not awakened by deployment. ## Model Used OpenAI GPT-6 through Codex, with reasoning, repository tools, code execution, and test execution. This session does not expose a more specific model build ID or context-window size. ## 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 --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
GitHub
parent
b1efd65edc
commit
5545f6d166
|
|
@ -1564,3 +1564,12 @@ Export/import behavior in V1:
|
|||
- import supports preview (dry-run) before apply
|
||||
- import preview reports skill-policy and legacy-grant mappings before apply and rejects unknown policy schema versions
|
||||
- GitHub imports warn on unpinned refs instead of blocking
|
||||
|
||||
### User messages after native execution recovery stops
|
||||
|
||||
An authenticated user message can start a fresh native conversation turn once
|
||||
the prior execution is confirmed stopped. Retain the source history and uncertain
|
||||
action outcomes; do not replay tool calls or reset the failed incident's automatic
|
||||
retry budget. Existing pause, approval, budget, ownership, and dependency gates
|
||||
remain in effect. See `doc/execution-semantics.md` for admission and stop-proof
|
||||
requirements.
|
||||
|
|
|
|||
|
|
@ -857,6 +857,32 @@ new run. Preserve the baseline across recovery of the same run and start a new
|
|||
delta when attaching a new run. Other stale-event and authority checks remain.
|
||||
|
||||
|
||||
### Explicit user continuation after a native failure
|
||||
|
||||
An execution recovery hold blocks automatic replay. A new authenticated user
|
||||
comment can authorize a fresh native conversation turn after the predecessor's
|
||||
execution is confirmed stopped. This is a new request, not another automatic
|
||||
attempt in the failed incident. The old attempt count and unknown action outcomes
|
||||
remain unchanged.
|
||||
|
||||
Admission validates the persisted comment's author, task, and time against every
|
||||
held predecessor. An agent-authored comment, an old queued request, or a generic
|
||||
system wake cannot release a hold. The source task keeps its assignee. Process
|
||||
ownership, active controllers, cleanup leases, pause, approval, budget, and normal
|
||||
execution gates still apply. Dependency-blocked interaction mode remains limited
|
||||
to its existing answer/triage contract.
|
||||
|
||||
The hold retirement, audit record, and new run commit together under the task
|
||||
lock. The new turn uses a fresh provider session and retains the latest user
|
||||
request, task history, completed work, and the interruption notice. It receives
|
||||
no instruction to repeat old tool calls. Later messages cannot reset the old
|
||||
incident's retry budget or create another automatic replacement for it.
|
||||
|
||||
The initial native admission path verifies local process identities. Missing
|
||||
process identity or remote ownership without a target-aware stop proof remains a
|
||||
hold; a terminal database status or a PID check on the wrong host is insufficient.
|
||||
No historical task is automatically awakened by this change.
|
||||
|
||||
### Explicit Recovery Action
|
||||
|
||||
Paperclip opens an explicit recovery action when the system can identify a problem but cannot safely complete the work itself.
|
||||
|
|
|
|||
|
|
@ -209,7 +209,7 @@ export async function buildExecutionContinuation(input: {
|
|||
row.authorType === "user" && !row.createdByRunId && !row.deleted && row.body.trim().length > 0,
|
||||
);
|
||||
const priorRuns = await db
|
||||
.select({ id: heartbeatRuns.id, result: heartbeatRuns.resultJson, status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode })
|
||||
.select({ id: heartbeatRuns.id, result: heartbeatRuns.resultJson, status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode, runtimeMode: heartbeatRuns.runtimeMode })
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
|
|
@ -254,10 +254,25 @@ export async function buildExecutionContinuation(input: {
|
|||
["succeeded", "failed", "timed_out", "interrupted", "cancelled"].includes(run.status) &&
|
||||
!(run.status === "cancelled" && run.errorCode === "execution_reconciliation_required"),
|
||||
);
|
||||
const interruptedRunId = lastTerminal && lastTerminal.status !== "succeeded" &&
|
||||
const explicitContinuation = object(input.context.explicitUserContinuation);
|
||||
const explicitUserSource = string(explicitContinuation.previousRunId);
|
||||
if (explicitUserSource) {
|
||||
const predecessor = priorRuns.find(run => run.id === explicitUserSource &&
|
||||
run.runtimeMode === "native" && ["failed", "timed_out", "interrupted", "cancelled"].includes(run.status));
|
||||
const authorization = reconciliations.map(row => object(row.evidence.explicitUserContinuation))
|
||||
.find(value => value.previousRunId === explicitUserSource &&
|
||||
value.commentId === explicitContinuation.commentId &&
|
||||
priorRuns.some(run => run.id === value.runId) &&
|
||||
rows.some(comment => comment.id === value.commentId &&
|
||||
comment.authorType === "user" && comment.authorUserId === value.actorId &&
|
||||
!comment.createdByRunId && !comment.deletedAt));
|
||||
if (!predecessor || !authorization || explicitUserSource !== sourceRunId)
|
||||
throw new Error("continuation_user_authorization_missing");
|
||||
}
|
||||
const interruptedRunId = explicitUserSource ?? (lastTerminal && lastTerminal.status !== "succeeded" &&
|
||||
(hasConversationContinuationPolicy(lastTerminal.result) ||
|
||||
lastTerminal.status === "interrupted" || lastTerminal.errorCode === "process_lost")
|
||||
? lastTerminal.id : undefined;
|
||||
? lastTerminal.id : undefined);
|
||||
return {
|
||||
...(interruptedRunId ? { interruptedRunId } : {}),
|
||||
...(resumeDelta ? { resumeDelta } : {}),
|
||||
|
|
|
|||
|
|
@ -0,0 +1,195 @@
|
|||
import { randomUUID } from "node:crypto";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { beforeAll, afterAll, describe, it, expect } from "vitest";
|
||||
import {
|
||||
approvals, issueApprovals, issueThreadInteractions,
|
||||
agentWakeupRequests, agents, companies, createDb, heartbeatRuns, issueComments, issueRecoveryActions,
|
||||
issues, nativeRunFinalizations, environmentLeases, environments, issueRelations, issueTreeHolds, issueTreeHoldMembers,
|
||||
} from "@paperclipai/db";
|
||||
import { startEmbeddedPostgresTestDatabase, getEmbeddedPostgresTestSupport } from "../__tests__/helpers/embedded-postgres.js";
|
||||
import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js";
|
||||
import { buildExecutionContinuation } from "./execution-continuation.js";
|
||||
import { heartbeatService } from "./heartbeat.js";
|
||||
import { getExecutionBlocker } from "./execution-blocker.js";
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
(support.supported ? describe : describe.skip)("explicit native conversation continuation", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("explicit-native-message-"); db = createDb(database.connectionString); }, 30000);
|
||||
afterAll(async () => { await database?.cleanup(); });
|
||||
async function seed() {
|
||||
const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID();
|
||||
const sourceRunId = randomUUID(), successorRunId = randomUUID();
|
||||
const commentId: string = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Explicit turn", defaultResponsibleUserId: "board", issuePrefix: `E${companyId.slice(0, 6)}` });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Native", role: "engineer", adapterType: "paperclip_runner", status: "idle", runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } });
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Deploy", status: "blocked", assigneeAgentId: agentId });
|
||||
await db.insert(heartbeatRuns).values({ id: sourceRunId, companyId, agentId,
|
||||
nativeIssueId: issueId, runtimeMode: "native", status: "failed", processPid: 999999999,
|
||||
contextSnapshot: { issueId }, finishedAt: new Date("2026-09-11T10:00:00Z") });
|
||||
await db.insert(nativeRunFinalizations).values({ runId: sourceRunId, companyId, issueId,
|
||||
phase: "terminal_failure", attempt: 3, failureDetail: { replacementDenied: "uncertain_external_action" } });
|
||||
await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId,
|
||||
kind: "active_run_watchdog", cause: "uncertain_external_action", fingerprint: sourceRunId,
|
||||
status: "resolved", outcome: "blocked", nextAction: "Automatic recovery stopped.",
|
||||
evidence: { runId: sourceRunId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } });
|
||||
await db.insert(issueComments).values({ id: commentId, companyId, issueId, authorType: "user",
|
||||
authorUserId: "board", body: "What happened?", createdAt: new Date("2026-09-11T11:00:00Z") });
|
||||
return { companyId, issueId, agentId, sourceRunId, commentId, successorRunId,
|
||||
actorType: "user", actorId: "board", reason: "issue_commented" };
|
||||
}
|
||||
type Fixture = Awaited<ReturnType<typeof seed>>;
|
||||
const admit = (f: Fixture, dryRun = false) => db.transaction(async tx => {
|
||||
await tx.select().from(issues).where(eq(issues.id, f.issueId)).for("update");
|
||||
const result = await admitExplicitNativeContinuation({ ...f, dryRun, db: tx as unknown as typeof db });
|
||||
if (result && !dryRun) await tx.insert(heartbeatRuns).values({ id: f.successorRunId, companyId: f.companyId,
|
||||
agentId: f.agentId, status: "queued", contextSnapshot: { issueId: f.issueId, previousRunId: result.previousRunId, forceFreshSession: true } });
|
||||
return result;
|
||||
});
|
||||
it("queues the actual user wake with a fresh session and retained source context", async () => {
|
||||
const f = await seed();
|
||||
// Occupy this agent's only slot so this admission test never starts a provider.
|
||||
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" });
|
||||
const wake = await heartbeatService(db).wakeup(f.agentId, {
|
||||
source: "automation", triggerDetail: "system", reason: "issue_commented",
|
||||
requestedByActorType: "user", requestedByActorId: "board",
|
||||
payload: { issueId: f.issueId, commentId: f.commentId },
|
||||
contextSnapshot: { issueId: f.issueId, wakeCommentId: f.commentId },
|
||||
});
|
||||
expect(wake).toMatchObject({ status: "queued", retryOfRunId: null,
|
||||
contextSnapshot: { forceFreshSession: true, previousRunId: f.sourceRunId,
|
||||
explicitUserContinuation: { previousRunId: f.sourceRunId, commentId: f.commentId } } });
|
||||
const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
|
||||
expect(action.evidence.explicitUserContinuation).toMatchObject({ runId: wake!.id });
|
||||
const envelope = await buildExecutionContinuation({ db, companyId: f.companyId, issueId: f.issueId,
|
||||
agentId: f.agentId, context: wake!.contextSnapshot!, summary: "Deployment completed.", exposeLowTrustRaw: false });
|
||||
expect(envelope.interruptedRunId).toBe(f.sourceRunId);
|
||||
expect(envelope.objective).toBe("What happened?");
|
||||
expect(envelope.completedWork).toBe("Deployment completed.");
|
||||
});
|
||||
|
||||
it.each(["pause", "dependency", "state"])("keeps the existing %s gate on the actual user wake", async gate => {
|
||||
const f = await seed();
|
||||
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" });
|
||||
if (gate === "pause") {
|
||||
const holdId = randomUUID();
|
||||
await db.insert(issueTreeHolds).values({ id: holdId, companyId: f.companyId, rootIssueId: f.issueId, mode: "pause", status: "active" });
|
||||
await db.insert(issueTreeHoldMembers).values({ companyId: f.companyId, holdId, issueId: f.issueId, depth: 0, issueTitle: "Deploy", issueStatus: "blocked" });
|
||||
} else if (gate === "dependency") {
|
||||
const blockerId = randomUUID();
|
||||
await db.insert(issues).values({ id: blockerId, companyId: f.companyId, title: "Required approval", status: "todo" });
|
||||
await db.insert(issueRelations).values({ companyId: f.companyId, issueId: blockerId, relatedIssueId: f.issueId, type: "blocks" });
|
||||
}
|
||||
const wake = await heartbeatService(db).wakeup(f.agentId, {
|
||||
source: "automation", triggerDetail: "system", reason: "issue_commented", requestedByActorType: "user", requestedByActorId: "board",
|
||||
...(gate === "state" ? { issueStateGuard: { statuses: ["todo"], assigneeAgentId: f.agentId } } : {}),
|
||||
payload: { issueId: f.issueId, commentId: f.commentId }, contextSnapshot: { issueId: f.issueId, wakeCommentId: f.commentId },
|
||||
});
|
||||
const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
|
||||
if (gate === "pause" || gate === "state") {
|
||||
expect(wake).toBeNull();
|
||||
expect(action.evidence.explicitUserContinuation).toBeUndefined();
|
||||
} else {
|
||||
expect(wake).toMatchObject({ contextSnapshot: { dependencyBlockedInteraction: true, unresolvedBlockerCount: 1 } });
|
||||
expect(await db.select().from(issueRelations).where(eq(issueRelations.companyId, f.companyId))).toHaveLength(1);
|
||||
}
|
||||
});
|
||||
|
||||
it("preflights eligibility without retiring the hold or creating a successor", async () => {
|
||||
const f = await seed();
|
||||
expect(await admit(f, true)).toMatchObject({ previousRunId: f.sourceRunId });
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull();
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.successorRunId))).toHaveLength(0);
|
||||
const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
|
||||
expect(action.evidence.explicitUserContinuation).toBeUndefined();
|
||||
});
|
||||
it("retains the message receipt without a phantom run when ownership is still live", async () => {
|
||||
const f = await seed();
|
||||
await db.update(heartbeatRuns).set({ processPid: process.pid }).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
const wake = await heartbeatService(db).wakeup(f.agentId, {
|
||||
source: "automation", triggerDetail: "system", reason: "issue_commented",
|
||||
requestedByActorType: "user", requestedByActorId: "board",
|
||||
payload: { issueId: f.issueId, commentId: f.commentId },
|
||||
contextSnapshot: { issueId: f.issueId, wakeCommentId: f.commentId },
|
||||
});
|
||||
expect(wake).toBeNull();
|
||||
const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId));
|
||||
expect(receipt).toMatchObject({ status: "deferred_issue_execution", requestedByActorId: "board", runId: null });
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1);
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull();
|
||||
});
|
||||
it("lets a new human message continue after exhausted recovery without certifying old actions", async () => {
|
||||
const f = await seed();
|
||||
expect(await admit(f)).toEqual({ previousRunId: f.sourceRunId, commentId: f.commentId });
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
|
||||
const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
|
||||
expect(action.evidence).toMatchObject({ automaticRecovery: { actionOutcome: "unknown" }, explicitUserContinuation: { runId: f.successorRunId, commentId: f.commentId } });
|
||||
const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
expect(source.status).toBe("failed");
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
|
||||
expect(coordinator.attempt).toBe(3);
|
||||
expect(coordinator.failureDetail?.replacementDenied).toBe("explicit_user_continuation");
|
||||
});
|
||||
it.each(["live_process", "missing_process", "lease", "coordinator", "successor", "agent_message", "old_comment", "wrong_author", "run_authored", "reassigned", "automatic", "approval", "question", "malformed_comment", "legacy_owner"])("keeps the hold for %s", async kind => {
|
||||
const f = await seed();
|
||||
if (kind === "legacy_owner") await db.insert(heartbeatRuns).values({ companyId: f.companyId,
|
||||
agentId: f.agentId, status: "failed", runtimeMode: "legacy", processPid: process.pid,
|
||||
contextSnapshot: { issueId: f.issueId }, resultJson: { conversationContinuation: "continue_conversation_v1" } });
|
||||
if (kind === "live_process") await db.update(heartbeatRuns).set({ processPid: process.pid }).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
if (kind === "missing_process") await db.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
if (kind === "coordinator") await db.update(nativeRunFinalizations).set({ leaseOwner: "active-controller" }).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
|
||||
if (kind === "successor") await db.update(nativeRunFinalizations).set({ failureDetail: { successorRunId: randomUUID() } }).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
|
||||
if (kind === "lease") {
|
||||
const [environment] = await db.select().from(environments).where(eq(environments.driver, "local"));
|
||||
const environmentId = environment.id;
|
||||
await db.insert(environmentLeases).values({ companyId: f.companyId, environmentId, heartbeatRunId: f.sourceRunId, issueId: f.issueId, status: "active", leasePolicy: "ephemeral", provider: "local" });
|
||||
}
|
||||
if (kind === "question") await db.insert(issueThreadInteractions).values({
|
||||
companyId: f.companyId, issueId: f.issueId, kind: "ask_user_questions", status: "pending", payload: { version: 1, questions: [] },
|
||||
});
|
||||
if (kind === "approval") {
|
||||
const approvalId = randomUUID();
|
||||
await db.insert(approvals).values({ id: approvalId, companyId: f.companyId, type: "hire_agent", status: "pending", payload: {} });
|
||||
await db.insert(issueApprovals).values({ companyId: f.companyId, issueId: f.issueId, approvalId });
|
||||
}
|
||||
if (kind === "malformed_comment") f.commentId = "not-a-uuid";
|
||||
if (kind === "agent_message") f.actorType = "agent";
|
||||
if (kind === "automatic") f.reason = "issue_continuation_needed";
|
||||
if (kind === "wrong_author") f.actorId = "someone-else";
|
||||
if (kind === "run_authored") await db.update(issueComments).set({ createdByRunId: f.sourceRunId }).where(eq(issueComments.id, f.commentId));
|
||||
if (kind === "old_comment") await db.update(issueComments).set({ createdAt: new Date("2026-09-11T09:00:00Z") }).where(eq(issueComments.id, f.commentId));
|
||||
if (kind === "reassigned") await db.update(issues).set({ assigneeAgentId: null }).where(eq(issues.id, f.issueId));
|
||||
expect(await admit(f)).toBeNull();
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull();
|
||||
});
|
||||
it.each(["foreign_source", "missing_authorization", "nonterminal_source"])("rejects unverified interruption context: %s", async kind => {
|
||||
const f = await seed();
|
||||
let previousRunId: string = f.sourceRunId;
|
||||
if (kind === "foreign_source") previousRunId = (await seed()).sourceRunId;
|
||||
if (kind === "nonterminal_source") {
|
||||
await admit(f);
|
||||
await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
}
|
||||
await expect(buildExecutionContinuation({ db, companyId: f.companyId, issueId: f.issueId,
|
||||
agentId: f.agentId, context: { previousRunId: f.sourceRunId,
|
||||
explicitUserContinuation: { previousRunId, commentId: f.commentId } },
|
||||
summary: null, exposeLowTrustRaw: false })).rejects.toThrow("continuation_user_authorization_missing");
|
||||
});
|
||||
it("keeps one new turn under concurrent delivery of the same message", async () => {
|
||||
const f = await seed();
|
||||
const results = await Promise.all([admit(f), admit(f)]);
|
||||
expect(results.filter(Boolean)).toHaveLength(1);
|
||||
expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")))).toHaveLength(1);
|
||||
});
|
||||
it("retains the source incident through prior rejected message admissions", async () => {
|
||||
const f = await seed(), rejectedRunId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({ id: rejectedRunId, companyId: f.companyId, agentId: f.agentId,
|
||||
status: "cancelled", errorCode: "execution_reconciliation_required", contextSnapshot: { issueId: f.issueId },
|
||||
finishedAt: new Date("2026-09-11T10:30:00Z") });
|
||||
await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId,
|
||||
kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", fingerprint: rejectedRunId,
|
||||
status: "resolved", outcome: "blocked", nextAction: "Could not start", evidence: { runId: rejectedRunId, automaticRecovery: { replay: "blocked" } } });
|
||||
expect(await admit(f)).toMatchObject({ previousRunId: f.sourceRunId });
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,139 @@
|
|||
import { z } from "zod";
|
||||
import { and, eq, inArray, isNull, ne, or, sql } from "drizzle-orm";
|
||||
import {
|
||||
approvals, issueApprovals, issueThreadInteractions,
|
||||
environmentLeases, heartbeatRuns, issueComments, issueRecoveryActions,
|
||||
issues, nativeRunFinalizations, type Db,
|
||||
} from "@paperclipai/db";
|
||||
import { executionBlockerPredicate, getExecutionBlocker } from "./execution-blocker.js";
|
||||
import { buildExecutionContinuation } from "./execution-continuation.js";
|
||||
import { adapterExecutionControls } from "./adapter-execution-control.js";
|
||||
import { persistActivity } from "./activity-log.js";
|
||||
|
||||
type Run = typeof heartbeatRuns.$inferSelect;
|
||||
const terminal = ["failed", "interrupted", "timed_out", "cancelled"];
|
||||
|
||||
function processStopped(pid: number): boolean {
|
||||
try { process.kill(pid, 0); return false; }
|
||||
catch (error) { return (error as NodeJS.ErrnoException).code === "ESRCH"; }
|
||||
}
|
||||
|
||||
/** Called under the issue lock, in the transaction that creates the new turn.
|
||||
* A user request authorizes a new conversation, not replay of the failed run.
|
||||
* Unknown action outcomes and all prior records stay intact.
|
||||
*/
|
||||
export async function admitExplicitNativeContinuation(input: {
|
||||
db: Db; companyId: string; issueId: string; agentId: string;
|
||||
actorType: string | null | undefined; actorId: string | null | undefined;
|
||||
reason: string | null; commentId: string | null; successorRunId: string;
|
||||
dryRun?: boolean;
|
||||
}): Promise<{ previousRunId: string; commentId: string } | null> {
|
||||
const { db, companyId, issueId, agentId, actorId, commentId } = input;
|
||||
if (input.actorType !== "user" || !actorId || !commentId ||
|
||||
!["issue_commented", "issue_reopened_via_comment"].includes(input.reason ?? "")) return null;
|
||||
if (!z.string().guid().safeParse(commentId).success) return null;
|
||||
const [task] = await db.select().from(issues).where(and(
|
||||
eq(issues.companyId, companyId), eq(issues.id, issueId),
|
||||
));
|
||||
if (!task || task.assigneeAgentId !== agentId || ["done", "cancelled"].includes(task.status)) return null;
|
||||
const [comment] = await db.select().from(issueComments).where(and(
|
||||
eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId),
|
||||
eq(issueComments.id, commentId), eq(issueComments.authorType, "user"),
|
||||
eq(issueComments.authorUserId, actorId), isNull(issueComments.createdByRunId),
|
||||
isNull(issueComments.deletedAt),
|
||||
));
|
||||
if (!comment?.body.trim()) return null;
|
||||
const actions = await db.select().from(issueRecoveryActions).where(and(
|
||||
eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, issueId),
|
||||
executionBlockerPredicate(),
|
||||
)).for("update");
|
||||
if (!actions.length) return null;
|
||||
const blocker = await getExecutionBlocker(db, companyId, issueId);
|
||||
if (blocker && blocker.recoveryActionId === null) return null;
|
||||
const [pendingInteraction] = await db.select({ id: issueThreadInteractions.id }).from(issueThreadInteractions).where(and(
|
||||
eq(issueThreadInteractions.companyId, companyId), eq(issueThreadInteractions.issueId, issueId),
|
||||
eq(issueThreadInteractions.status, "pending"),
|
||||
)).limit(1);
|
||||
const [pendingApproval] = await db.select({ id: approvals.id }).from(issueApprovals).innerJoin(approvals, and(
|
||||
eq(approvals.id, issueApprovals.approvalId), eq(approvals.companyId, companyId),
|
||||
)).where(and(eq(issueApprovals.companyId, companyId), eq(issueApprovals.issueId, issueId),
|
||||
inArray(approvals.status, ["pending", "revision_requested"]))).limit(1);
|
||||
if (pendingInteraction || pendingApproval) return null;
|
||||
|
||||
const sources: Run[] = [];
|
||||
for (const action of actions) {
|
||||
const runId = action.evidence.runId ?? action.evidence.sourceRunId;
|
||||
if (typeof runId !== "string") return null;
|
||||
// Text comparison keeps malformed historical evidence a hold, not a UUID cast error.
|
||||
const [run] = await db.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), sql`${heartbeatRuns.id}::text = ${runId}`,
|
||||
));
|
||||
if (!run || run.agentId !== agentId || !terminal.includes(run.status) ||
|
||||
(run.nativeIssueId ?? run.contextSnapshot?.issueId) !== issueId ||
|
||||
!run.finishedAt || comment.createdAt <= run.finishedAt) return null;
|
||||
if (adapterExecutionControls.has(run.id)) return null;
|
||||
const unusedAdmission = run.status === "cancelled" && !run.startedAt &&
|
||||
run.errorCode === "execution_reconciliation_required" &&
|
||||
!run.processPid && !run.processGroupId && !run.nativeSessionId;
|
||||
if (run.runtimeMode !== "native" && !unusedAdmission) return null;
|
||||
if (!unusedAdmission) {
|
||||
// A missing process identity is not evidence that a provider exited.
|
||||
if (!run.processPid && !run.processGroupId) return null;
|
||||
if (run.processPid && !processStopped(run.processPid)) return null;
|
||||
if (run.processGroupId && !processStopped(-run.processGroupId)) return null;
|
||||
}
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, run.id),
|
||||
)).for("update");
|
||||
if (coordinator && (coordinator.phase !== "terminal_failure" || coordinator.leaseOwner ||
|
||||
coordinator.resultId || coordinator.failureDetail?.successorRunId)) return null;
|
||||
const leases = await db.select({ provider: environmentLeases.provider, releasedAt: environmentLeases.releasedAt })
|
||||
.from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, run.id),
|
||||
));
|
||||
// A PID on another host cannot be checked with this server's process table.
|
||||
// Remote execution retains its hold until a target-aware stop proof exists.
|
||||
if (leases.some(lease => !lease.releasedAt || lease.provider !== "local")) return null;
|
||||
sources.push(run);
|
||||
}
|
||||
const nativeSources = sources.filter(run => run.runtimeMode === "native");
|
||||
if (!nativeSources.length) return null;
|
||||
const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId),
|
||||
or(eq(heartbeatRuns.nativeIssueId, issueId), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`),
|
||||
inArray(heartbeatRuns.status, ["running", "queued", "scheduled_retry"]),
|
||||
ne(heartbeatRuns.id, input.successorRunId),
|
||||
)).limit(1);
|
||||
if (active) return null;
|
||||
const previous = nativeSources.sort((a, b) => b.createdAt.getTime() - a.createdAt.getTime())[0]!;
|
||||
// Prove required task history is available before retiring any hold.
|
||||
await buildExecutionContinuation({ db, companyId, issueId, agentId,
|
||||
context: { previousRunId: previous.id, wakeCommentId: commentId },
|
||||
summary: null, exposeLowTrustRaw: false });
|
||||
if (input.dryRun) return { previousRunId: previous.id, commentId };
|
||||
const authorization = { actorId, commentId, runId: input.successorRunId,
|
||||
previousRunId: previous.id, recordedAt: new Date().toISOString() };
|
||||
await db.update(nativeRunFinalizations).set({
|
||||
failureDetail: sql`coalesce(${nativeRunFinalizations.failureDetail}, '{}'::jsonb) || ${JSON.stringify({ replacementDenied: "explicit_user_continuation" })}::jsonb`,
|
||||
updatedAt: new Date(),
|
||||
}).where(and(eq(nativeRunFinalizations.companyId, companyId), inArray(nativeRunFinalizations.runId, nativeSources.map(run => run.id))));
|
||||
for (const action of actions) {
|
||||
await db.update(issueRecoveryActions).set({
|
||||
status: "resolved", outcome: "cancelled", resolvedAt: new Date(), updatedAt: new Date(),
|
||||
nextAction: "A new user message starts a fresh conversation turn. Prior action outcomes remain recorded.",
|
||||
resolutionNote: "The user continued after the prior execution stopped. No action outcomes were inferred.",
|
||||
wakePolicy: null, monitorPolicy: null,
|
||||
evidence: { ...action.evidence, explicitUserContinuation: authorization,
|
||||
...(action.evidence.automaticRecovery ? { automaticRecovery: {
|
||||
...(action.evidence.automaticRecovery as Record<string, unknown>), replay: "explicit_user_continuation",
|
||||
} } : {}),
|
||||
},
|
||||
}).where(eq(issueRecoveryActions.id, action.id));
|
||||
}
|
||||
await persistActivity(db, { companyId, actorType: "user", actorId,
|
||||
action: "issue.execution_recovery_settled", entityType: "issue", entityId: issueId,
|
||||
details: { continuation: "explicit_user_message", ...authorization,
|
||||
recoveryActionIds: actions.map(action => action.id), previousRunIds: sources.map(run => run.id) },
|
||||
});
|
||||
return { previousRunId: previous.id, commentId };
|
||||
}
|
||||
|
|
@ -1,3 +1,4 @@
|
|||
import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js";
|
||||
import { getExecutionBlocker } from "./execution-blocker.js";
|
||||
import { CONVERSATION_CONTINUATION_POLICY, runUsedConversationAdapter, hasConversationContinuationPolicy, isConversationAdapter } from "./conversation-continuation.js";
|
||||
import { recordExecutionWait } from "./execution-wait.js";
|
||||
|
|
@ -25538,14 +25539,9 @@ export function heartbeatService(
|
|||
reconciledSourceRunId = sourceRunId;
|
||||
}
|
||||
|
||||
// All wake producers share this admission gate. A resolved recovery
|
||||
// action can still prohibit replay; its durable evidence owns the wait.
|
||||
// Reconciliation wakes have already proved their authority above and
|
||||
// must still respect any other effective hold on the same issue.
|
||||
const executionBlocker = await getExecutionBlocker(
|
||||
tx as unknown as Db, issue.companyId, issue.id,
|
||||
);
|
||||
if (executionBlocker) {
|
||||
const deferBlockedExecution = async (
|
||||
executionBlocker: NonNullable<Awaited<ReturnType<typeof getExecutionBlocker>>>,
|
||||
) => {
|
||||
const condition = { recoveryActionId: executionBlocker.recoveryActionId };
|
||||
if (durableRequest || wakeCommentId || hasInteractionContinuationWakeContext(enrichedContextSnapshot)) {
|
||||
await tx.insert(agentWakeupRequests).values({
|
||||
|
|
@ -25581,7 +25577,19 @@ export function heartbeatService(
|
|||
});
|
||||
}
|
||||
return { kind: "deferred" as const };
|
||||
}
|
||||
};
|
||||
const explicitContinuationRunId = randomUUID();
|
||||
const executionBlocker = await getExecutionBlocker(
|
||||
tx as unknown as Db, issue.companyId, issue.id,
|
||||
);
|
||||
// Prove eligibility without retiring the hold. Later gates can still
|
||||
// decline this wake; hold retirement and successor creation stay atomic.
|
||||
if (executionBlocker && !(await admitExplicitNativeContinuation({
|
||||
db: tx as unknown as Db, companyId: issue.companyId, issueId: issue.id,
|
||||
agentId, actorType: opts.requestedByActorType, actorId: opts.requestedByActorId,
|
||||
reason, commentId: wakeCommentId ?? null, successorRunId: explicitContinuationRunId,
|
||||
dryRun: true,
|
||||
}))) return deferBlockedExecution(executionBlocker);
|
||||
|
||||
const issueStateGuard = opts.issueStateGuard;
|
||||
if (
|
||||
|
|
@ -26332,6 +26340,18 @@ export function heartbeatService(
|
|||
return { kind: "skipped" as const };
|
||||
}
|
||||
|
||||
const explicitContinuation = await admitExplicitNativeContinuation({
|
||||
db: tx as unknown as Db, companyId: issue.companyId, issueId: issue.id,
|
||||
agentId, actorType: opts.requestedByActorType, actorId: opts.requestedByActorId,
|
||||
reason, commentId: wakeCommentId ?? null, successorRunId: explicitContinuationRunId,
|
||||
});
|
||||
if (!explicitContinuation && executionBlocker) return deferBlockedExecution(executionBlocker);
|
||||
if (explicitContinuation) {
|
||||
enrichedContextSnapshot.forceFreshSession = true;
|
||||
enrichedContextSnapshot.previousRunId = explicitContinuation.previousRunId;
|
||||
enrichedContextSnapshot.explicitUserContinuation = explicitContinuation;
|
||||
}
|
||||
|
||||
const wakeupRequest = await tx
|
||||
.insert(agentWakeupRequests)
|
||||
.values({
|
||||
|
|
@ -26394,6 +26414,7 @@ export function heartbeatService(
|
|||
const newRun = await tx
|
||||
.insert(heartbeatRuns)
|
||||
.values({
|
||||
...(explicitContinuation ? { id: explicitContinuationRunId } : {}),
|
||||
companyId: agent.companyId,
|
||||
agentId,
|
||||
invocationSource: source,
|
||||
|
|
@ -26410,7 +26431,7 @@ export function heartbeatService(
|
|||
adoptedCommentIds,
|
||||
)
|
||||
: enrichedContextSnapshot,
|
||||
sessionIdBefore: sessionBefore,
|
||||
sessionIdBefore: explicitContinuation ? null : sessionBefore,
|
||||
continuationAttempt,
|
||||
...(reconciledSourceRunId
|
||||
? { retryOfRunId: reconciledSourceRunId }
|
||||
|
|
|
|||
|
|
@ -1,3 +1,5 @@
|
|||
import { createRunDispatch, deriveCommentId } from "../../modules/run-dispatch/index.js";
|
||||
import { buildExecutionContinuation } from "../execution-continuation.js";
|
||||
import { activityService } from "../activity.js";
|
||||
import { buildPaperclipWakePayload, heartbeatService } from "../heartbeat.js";
|
||||
import { legacyExecutionNeedsReconciliation, terminalizeLegacyExecution } from "../legacy-execution-recovery.js";
|
||||
|
|
@ -20,6 +22,7 @@ import {
|
|||
heartbeatRunEvents,
|
||||
heartbeatRuns,
|
||||
issueRecoveryActions,
|
||||
issueComments,
|
||||
issues,
|
||||
nativeRunFinalizations,
|
||||
} from "@paperclipai/db";
|
||||
|
|
@ -276,6 +279,50 @@ const support = externalDatabaseUrl
|
|||
"provider_failure_meaning_unverified",
|
||||
);
|
||||
});
|
||||
it.each(["done", "cancelled", "review"])("does not reuse consumed wake authority after a later %s transition", async transition => {
|
||||
const source = await seed();
|
||||
const previousRunId = randomUUID(), commentId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({ id: previousRunId, companyId: source.companyId,
|
||||
agentId: source.agentId, runtimeMode: "native", status: "failed", contextSnapshot: { issueId: source.issueId } });
|
||||
await db.insert(issueComments).values({ id: commentId, companyId: source.companyId,
|
||||
issueId: source.issueId, authorType: "user", authorUserId: "board", body: "Continue" });
|
||||
await db.insert(issueRecoveryActions).values({ companyId: source.companyId, sourceIssueId: source.issueId,
|
||||
kind: "active_run_watchdog", cause: "native_provider_terminal_failed", fingerprint: previousRunId,
|
||||
status: "resolved", outcome: "cancelled", nextAction: "User continued", evidence: {
|
||||
explicitUserContinuation: { previousRunId, commentId, actorId: "board", runId: source.runId },
|
||||
} });
|
||||
await db.update(heartbeatRuns).set({ contextSnapshot: {
|
||||
issueId: source.issueId, previousRunId,
|
||||
wakeCommentId: commentId, wakeCommentIds: [commentId], commentId,
|
||||
resumeIntent: true, followUpRequested: true,
|
||||
explicitUserContinuation: { previousRunId, commentId },
|
||||
} }).where(eq(heartbeatRuns.id, source.runId));
|
||||
await reconcileSafeNativeReplacements(db);
|
||||
const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, source.runId));
|
||||
expect(successor.contextSnapshot?.explicitUserContinuation).toBeUndefined();
|
||||
expect(deriveCommentId(successor.contextSnapshot)).toBeNull();
|
||||
expect(successor.contextSnapshot?.resumeIntent).toBeUndefined();
|
||||
expect(successor.contextSnapshot?.followUpRequested).toBeUndefined();
|
||||
const envelope = await buildExecutionContinuation({ db, companyId: source.companyId,
|
||||
issueId: source.issueId, agentId: source.agentId, context: successor.contextSnapshot!,
|
||||
summary: null, exposeLowTrustRaw: false });
|
||||
expect(envelope.trigger.sourceRunId).toBe(source.runId);
|
||||
expect(envelope.objective).toBe("Continue");
|
||||
if (transition === "review") {
|
||||
await db.update(issues).set({ status: "in_review", executionState: {
|
||||
status: "pending", currentStageId: randomUUID(), currentStageIndex: 0, currentStageType: "review",
|
||||
currentParticipant: { type: "agent", agentId: randomUUID(), userId: null },
|
||||
returnAssignee: { type: "agent", agentId: source.agentId, userId: null },
|
||||
reviewRequest: null, completedStageIds: [], lastDecisionId: null, lastDecisionOutcome: null,
|
||||
} }).where(eq(issues.id, source.issueId));
|
||||
} else {
|
||||
await db.update(issues).set({ status: transition }).where(eq(issues.id, source.issueId));
|
||||
}
|
||||
await db.update(heartbeatRuns).set({ status: "queued" }).where(eq(heartbeatRuns.id, successor.id));
|
||||
expect(await createRunDispatch(db).cancelStaleQueuedRun({ companyId: source.companyId,
|
||||
runId: successor.id, expectedStatus: "queued" })).toMatchObject({ outcome: "cancelled",
|
||||
errorCode: transition === "review" ? "issue_review_participant_changed" : "issue_terminal_status" });
|
||||
});
|
||||
it("persists exactly one linked successor under competing sweepers and restarts", async () => {
|
||||
const source = await seed(2);
|
||||
const now = new Date();
|
||||
|
|
|
|||
|
|
@ -316,8 +316,16 @@ export async function reconcileSafeNativeReplacements(
|
|||
return false;
|
||||
const successorRunId = randomUUID();
|
||||
const dueAt = new Date(now.getTime() + 30_000);
|
||||
const predecessorContext = { ...record(run.contextSnapshot) };
|
||||
// History comes from the failed source run. Consumed wake fields must
|
||||
// not grant this automatic retry fresh comment/resume authority.
|
||||
for (const key of [
|
||||
"explicitUserContinuation", "wakeCommentId", "wakeCommentIds", "commentId",
|
||||
"commentIds", "latestCommentId", "resumeIntent", "followUpRequested",
|
||||
"paperclipWake", "paperclipWakeComment", "paperclipTaskMarkdown", "paperclipTaskMarkdownCompact",
|
||||
]) delete predecessorContext[key];
|
||||
const context = {
|
||||
...record(run.contextSnapshot),
|
||||
...predecessorContext,
|
||||
issueId: task.id,
|
||||
retryOfRunId: run.id,
|
||||
wakeReason: NATIVE_SAFE_REPLACEMENT_REASON,
|
||||
|
|
|
|||
Loading…
Reference in New Issue