Files
PaperClipAI/server/src/services/execution-continuation.ts
T
DottaandClaude Opus 5.5 5d1bb5192f fix(retry): carry the original wake to every re-queued run (#20989)
## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - A run serves one wake: a comment, an answer, an assignment, or a
one-shot task monitor check (`issue_monitor_due`).
> - When a run fails, Paperclip often starts a successor run by itself:
the bounded transient retry, the uncharged `interrupted_preparation`
re-queue, workspace or connection waits, missing-comment repair,
capacity retries and recovery successors. Each successor links to the
failed run through `retry_of_run_id`.
> - Most of these re-queues replace the wake reason with their own (for
example `transient_failure_retry`). The agent then does not know which
wake it must handle.
> - #20970 fixed this for the board Retry only. A no-fault chaos wave
showed the same loss on an automatic retry: the monitor run failed
before provider work, the automatic retry succeeded, and the agent did
not treat it as the monitor check. The monitor was already consumed, so
the task rested `in_progress` with nothing pending.
> - This pull request covers all re-queue paths at one shared point: the
continuation builder that every issue dispatch uses
(`buildExecutionContinuation`, called from `heartbeat/task-context.ts`).
It walks the re-queue chain to the first run that was not a re-queue and
records that run's wake reason as `trigger.retriedReason`. The wake
prompt already shows that field (#20970).
> - The benefit is that every successor of a failed run handles the same
wake as the run it replaces.

## Linked Issues or Issue Description

**What happened?**
No-fault restart-chaos wave on master `87fe4d9793`, volume-free mode
(`w-20261112T010036-s6801`, Claude, task t01, workload W2). The monitor
wake started run `78e27ef9`. It failed before provider work
(`native_predecessor_retirement_unproven`). The
`interrupted_preparation` re-queue (attempt 1/12) started run `0cd387a0`
on a fresh sandbox, and that run succeeded. The agent reported that no
board comment arrived and that the monitor wake was not visible. It
yielded `response_wake` again. The monitor was `triggered` with no
`nextCheckAt`. The issue rested `in_progress` with `canRetry=false`, no
live run, no wake and no monitor. The harness graded `R1
rested_not_done` and `R3 missing_retry`.

**Expected behavior**
A successor of a failed run tells the agent which wake it re-delivers. A
re-queued monitor check is handled as the monitor check.

**Steps to reproduce**
1. Wake a native agent with `issue_monitor_due` on a task.
2. Make the run fail before provider work, for example when the previous
run's sandbox session cannot be retired.
3. Let the `interrupted_preparation` re-queue run.
4. Before this change, the successor's continuation trigger has the
re-queue reason only, and its prompt does not name the monitor check.

**Additional context**
Follow-up to #20970.

Refs #20970
Refs #20935
Refs #20863

## What Changed

- `server/src/services/execution-continuation.ts`: new
`requeuedWakeReason`.
- It starts at the current run's `retry_of_run_id` column. If that is
empty, it uses the context `retryOfRunId`, or the source run of a board
Retry.
- It follows the chain within the same company and issue, at most 32
links, until it finds a run with no re-queue link. It returns that run's
wake reason.
- A `trigger.retriedReason` already saved on a run in the chain is
reused.
- `trigger.retriedReason` is set only when the original reason differs
from the current reason and is not itself a Retry reason. A re-queue
that keeps the original wake reason (`preserveWakeContext`) is
unchanged.
- This one point covers every successor that links with
`retry_of_run_id`: bounded transient retries, `interrupted_preparation`
(#20863), workspace and AI-connection waits, missing-comment repair
(`run-lifecycle.ts`), native capacity retries, safe replacement, and the
board Retry.
- Tests:
- `server/src/__tests__/heartbeat-native-predispatch-retry.test.ts`: the
assembled heartbeat service runs a monitor-due wake, a
predecessor-retirement failure before provider work, and the automatic
`interrupted_preparation` re-queue. It asserts the successor's saved
continuation and wake prompt.
- `server/src/services/execution-continuation.test.ts`: a two-link chain
of automatic re-queues, and a re-queue that keeps the original reason.

## Verification

- Red on master: `cd server && npx vitest run
src/__tests__/heartbeat-native-predispatch-retry.test.ts -t "delivers
the monitor check"`. The successor's trigger had `sourceRunId` = the
failed monitor run, but no `retriedReason`.
- Green: `heartbeat-native-predispatch-retry.test.ts`,
`execution-continuation.test.ts` and `heartbeat/retries.test.ts` pass
(78 tests).
- `npx tsc --noEmit -p server` passes.
- Chaos: the existing no-fault W2 cell in volume-free mode reproduces
this when the monitor run fails before provider work. I did not rerun a
wave.

## Risks

- Low risk. The change adds one optional trigger field to more wakes. It
changes no admission, status or authority path.
- The walk costs at most one indexed primary-key read per chain link,
only for runs that are re-queues.
- The walk follows only same-company, same-issue runs. Old saved runs
work, because they already store `wakeReason` and the `retry_of_run_id`
column.
- The resting state "agent yielded `response_wake` after the monitor was
consumed" is still possible when an agent really waits. #20935 shows it
on the run projection as "Waiting for a reply". The ownership
reconciler's I7 check still counts it as `no_owner_no_wake`. This PR
does not change I7.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude
Code with tool use (shell, file edits, local 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>
2026-11-11 19:52:21 -06:00

624 lines
30 KiB
TypeScript

import { createHash } from "node:crypto";
import { and, asc, desc, eq, inArray, isNotNull, isNull, or, sql } from "drizzle-orm";
import { z } from "zod";
import {
agentWakeupRequests,
heartbeatRuns,
issueComments,
issueRecoveryActions,
issueThreadInteractions,
issues,
type Db,
} from "@paperclipai/db";
import type { ExecutionContinuationEnvelope } from "@paperclipai/shared";
import { sanitizeQuarantinedCommentForHigherTrust } from "./source-trust.js";
import { hasConversationContinuationPolicy } from "./conversation-continuation.js";
import { queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js";
import { queuedInteractionId } from "./queued-interaction-response.js";
import { childReviewOutcomes } from "./native-runtime/child-review-outcomes.js";
import { isCompletedOnboardingHandoffWake } from "./chat-completion-delivery.js";
import { githubRecoveryReceipts } from "./chat-channels/github/recovery-receipts.js";
import { readTaskRetryIntent } from "./task-retry.js";
const retryWakeReasons = new Set(["task_retry", "retry_failed_run"]);
const maxRequeueChain = 32;
const savedRecordOperations = new Set([
"PUT /api/issues/{id}/documents/{key}",
"POST /api/issues/{id}/comments",
"PATCH /api/issues/{id}",
]);
/** Project duplicate saved content, never the original execution receipt or user input. */
function continuationApiResult(operationId: string, result: unknown, companyId: string): unknown {
const response = object(result);
const data = object(response.data);
if (!savedRecordOperations.has(operationId) || response.operationId !== operationId
|| response.ok !== true || typeof response.status !== "number" || response.status < 200 || response.status >= 300
|| data.companyId !== companyId || !z.string().uuid().safeParse(data.id).success) return result;
const serialized = JSON.stringify(data);
if (Buffer.byteLength(serialized) <= 4 * 1024) return result;
const reference: Record<string, unknown> = {};
for (const key of ["id", "companyId", "issueId", "key", "latestRevisionId", "latestRevisionNumber",
"identifier", "status", "statusVersion", "createdAt", "updatedAt"]) {
const value = data[key];
if (value === null || typeof value === "string" && value.length <= 256
|| typeof value === "number" && Number.isSafeInteger(value)) reference[key] = value;
}
return { ...response, data: { ...reference, contentOmitted: true,
contentSha256: `sha256:${createHash("sha256").update(serialized).digest("hex")}` } };
}
export class StaleExecutionContinuationError extends Error {
constructor(readonly code: "continuation_task_ownership_changed") {
super(code);
this.name = "StaleExecutionContinuationError";
}
}
const object = (v: unknown): Record<string, unknown> =>
v && typeof v === "object" && !Array.isArray(v)
? (v as Record<string, unknown>)
: {};
const string = (v: unknown) =>
typeof v === "string" && v.length > 0 ? v : null;
const historyResult = sql<Record<string, unknown> | null>`case
when jsonb_typeof(${heartbeatRuns.resultJson}) = 'object' then (
select coalesce(jsonb_object_agg(field.key, case
when field.key = 'apiToolReceipts' and jsonb_typeof(field.value) = 'object' then (
select coalesce(jsonb_object_agg(receipt.key, (
select jsonb_object_agg(detail.key, detail.value)
from jsonb_each(receipt.value) as detail(key, value)
where detail.key in ('state', 'operationId', 'result')
)), '{}'::jsonb)
from jsonb_each(field.value) as receipt(key, value)
where receipt.value->'state' = '"completed"'::jsonb
and jsonb_typeof(receipt.value->'operationId') = 'string'
)
when field.key = 'workspaceRestoreRecovery' and jsonb_typeof(field.value) = 'object' then (
select coalesce(jsonb_object_agg(detail.key, detail.value), '{}'::jsonb)
from jsonb_each(field.value) as detail(key, value)
where detail.key = 'schema'
)
else field.value end), '{}'::jsonb)
from jsonb_each(${heartbeatRuns.resultJson}) as field(key, value)
where field.key in ('apiToolReceipts', 'conversationContinuation', 'workspaceRestoreFailure', 'workspaceRestoreRecovery')
) else ${heartbeatRuns.resultJson} end`;
export function continuationOriginCommentIds(context: unknown): string[] {
const c = object(context);
const prior = object(c.executionContinuation);
return [
...new Set(
[
c.commentId,
c.latestCommentId,
...(Array.isArray(c.commentIds) ? c.commentIds : []),
...(Array.isArray(c.wakeCommentIds) ? c.wakeCommentIds : []),
...(Array.isArray(prior.originCommentIds)
? prior.originCommentIds
: []),
].filter((v): v is string => typeof v === "string" && v.length > 0),
),
];
}
/**
* Return the comment IDs explicitly represented in a delivered continuation.
* An absent list is intentionally different from an empty list: historical or
* third-party snapshots cannot prove what the provider received.
*/
export function deliveredContinuationCommentIds(context: unknown): {
known: boolean;
ids: Set<string>;
} {
const c = object(context);
const wake = object(c.paperclipWake);
const continuation = object(c.executionContinuation);
const ids = new Set<string>();
let known = false;
const collect = (value: unknown) => {
if (!Array.isArray(value)) return;
known = true;
for (const item of value) {
const id = typeof item === "string" ? item : string(object(item).id);
if (id) ids.add(id);
}
};
collect(wake.comments);
collect(continuation.messages);
return { known, ids };
}
/** Keep service/tool results and generated summaries out of human authority. */
export function projectHumanInteractionResponse(row: {
id: string; kind: string; status: string; result: unknown;
resolvedByUserId: string | null; resolvedByAgentId: string | null;
resolvedByRunId: string | null; resolvedAt: Date | null;
}): NonNullable<ExecutionContinuationEnvelope["humanResponses"]>[number] | null {
if (!row.resolvedByUserId || row.resolvedByAgentId || row.resolvedByRunId || !row.resolvedAt) return null;
const result = object(row.result);
let response: Record<string, unknown>;
if (row.kind === "ask_user_questions" && row.status === "answered" && Array.isArray(result.answers)) {
response = { answers: result.answers.map(value => {
const answer = object(value);
return { questionId: answer.questionId, optionIds: answer.optionIds, otherText: answer.otherText };
}) };
} else if (["request_confirmation", "request_checkbox_confirmation"].includes(row.kind)
&& ["accepted", "rejected"].includes(row.status) && result.outcome === row.status) {
response = { outcome: result.outcome, reason: result.reason,
...(Array.isArray(result.selectedOptionIds) ? { selectedOptionIds: result.selectedOptionIds } : {}) };
} else return null;
return { id: row.id, kind: row.kind, status: row.status, resolvedByUserId: row.resolvedByUserId,
resolvedAt: row.resolvedAt.toISOString(), result: response };
}
/** Also retain user direction delivered after the source run's initial wake. */
export async function currentContinuationOrigins(
db: Db,
companyId: string,
issueId: string,
context: unknown,
): Promise<string[]> {
const candidates = continuationOriginCommentIds(context);
// A run may create an interaction on another task. Its own comments do not
// become authority on that task. Keep unknown references for dispatch to
// reject, rather than silently claiming complete context.
const foreignComments = candidates.length
? await db.select({ id: issueComments.id }).from(issueComments).where(and(
eq(issueComments.companyId, companyId),
sql`${issueComments.issueId} != ${issueId}`,
inArray(sql<string>`${issueComments.id}::text`, candidates),
))
: [];
const foreignIds = new Set(foreignComments.map(row => row.id));
const [latest] = await db
.select({ id: issueComments.id })
.from(issueComments)
.where(
and(
eq(issueComments.companyId, companyId),
eq(issueComments.issueId, issueId),
isNotNull(issueComments.authorUserId),
isNull(issueComments.createdByRunId),
isNull(issueComments.authorAgentId),
isNull(issueComments.deletedAt),
),
)
.orderBy(desc(issueComments.createdAt), desc(issueComments.id))
.limit(1);
return [
...new Set([
...candidates.filter(id => !foreignIds.has(id)),
...(latest ? [latest.id] : []),
]),
];
}
/** Re-read task scope at dispatch, including messages already delivered to an earlier provider session. */
function requeueSourceRunId(context: Record<string, unknown>, retryOfRunId: string | null): string | null {
if (retryOfRunId) return retryOfRunId;
const retryOf = string(context.retryOfRunId);
if (retryOf) return retryOf;
if (!retryWakeReasons.has(string(context.wakeReason) ?? "")) return null;
return string(object(context.explicitUserContinuation).previousRunId) ?? string(context.previousRunId);
}
async function requeuedWakeReason(db: Db, input: {
companyId: string; issueId: string; runId?: string; context: Record<string, unknown>;
}): Promise<string | null> {
const [current] = input.runId ? await db.select({ retryOfRunId: heartbeatRuns.retryOfRunId }).from(heartbeatRuns)
.where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId))) : [];
let next = requeueSourceRunId(input.context, current?.retryOfRunId ?? null);
const visited = new Set<string>();
while (next && !visited.has(next) && visited.size < maxRequeueChain) {
visited.add(next);
const [run] = await db.select({ context: heartbeatRuns.contextSnapshot, retryOfRunId: heartbeatRuns.retryOfRunId })
.from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, next)));
const context = object(run?.context);
if (!run || context.issueId !== input.issueId) return null;
const carried = string(object(object(context.executionContinuation).trigger).retriedReason);
if (carried) return carried;
next = requeueSourceRunId(context, run.retryOfRunId);
if (!next) return string(context.wakeReason);
}
return null;
}
export async function buildExecutionContinuation(input: {
db: Db;
companyId: string;
issueId: string;
agentId: string;
context: Record<string, unknown>;
previousContextRunId?: string | null;
/** Server-owned current run identity when validating dispatch authority. */
runId?: string;
summary: string | null;
exposeLowTrustRaw: boolean;
prospectiveTaskStatus?: "todo";
}): Promise<ExecutionContinuationEnvelope> {
const { db, companyId, issueId } = input;
const [issue] = await db
.select()
.from(issues)
.where(and(eq(issues.companyId, companyId), eq(issues.id, issueId)));
if (
!issue ||
issue.assigneeAgentId !== input.agentId ||
issue.status === "cancelled" ||
(issue.status === "done" && input.prospectiveTaskStatus !== "todo" && !await isCompletedOnboardingHandoffWake(db, {
companyId, issueId, agentId: input.agentId,
reason: string(input.context.wakeReason), contextSnapshot: input.context,
}))
)
throw new StaleExecutionContinuationError("continuation_task_ownership_changed");
const rows = await db
.select()
.from(issueComments)
.where(
and(
eq(issueComments.companyId, companyId),
eq(issueComments.issueId, issueId),
),
)
.orderBy(asc(issueComments.createdAt), asc(issueComments.id));
const interactions = await db
.select()
.from(issueThreadInteractions)
.where(
and(
eq(issueThreadInteractions.companyId, companyId),
eq(issueThreadInteractions.issueId, issueId),
),
)
.orderBy(
asc(issueThreadInteractions.createdAt),
asc(issueThreadInteractions.id),
);
const triggerInteraction = interactions.find(
(row) => row.id === input.context.interactionId,
);
const explicitContinuation = object(input.context.explicitUserContinuation);
const explicitUserSource = string(explicitContinuation.previousRunId);
const interruptedSource = string(input.context.interruptedRunId);
const resumeSourceRunId =
explicitUserSource ??
string(input.context.retryOfRunId) ??
string(input.context.previousRunId) ??
interruptedSource;
const producerRunId = triggerInteraction?.sourceRunId ?? null;
const requestedSourceRunId = resumeSourceRunId ?? producerRunId;
const loadRun = async (id: string) => (await db
.select({ context: heartbeatRuns.contextSnapshot, result: heartbeatRuns.resultJson })
.from(heartbeatRuns)
.where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, id)))
)[0];
const candidate = requestedSourceRunId ? await loadRun(requestedSourceRunId) : null;
// An interaction producer is provenance. Explicit resume history must still
// belong to this task, and only task-scoped content can enter the envelope.
const sourceRun = object(candidate?.context).issueId === issueId ? candidate : null;
// interruptedRunId is a best-effort hint written when a run is interrupted.
// After a cross-issue reassign it can name another task's run or a deleted
// row. Unlike an explicit user resume or a retry chain, a hint that does
// not resolve to this task must not fail the whole wake: degrade to "no
// resume source" and continue.
const staleInterruptedSource =
interruptedSource !== null
&& resumeSourceRunId === interruptedSource
&& explicitUserSource === null
&& string(input.context.retryOfRunId) === null
&& string(input.context.previousRunId) === null
&& sourceRun === null;
// A degraded hint falls back to the interaction producer so interaction
// wakes keep their source attribution.
const sourceRunId = staleInterruptedSource ? (producerRunId ?? null) : requestedSourceRunId;
if (
!staleInterruptedSource
&& ((requestedSourceRunId && !candidate) || (resumeSourceRunId && !sourceRun))
)
throw new Error(explicitUserSource ? "continuation_user_authorization_missing" : "continuation_source_context_missing");
const producer = producerRunId === requestedSourceRunId ? candidate
: producerRunId ? await loadRun(producerRunId) : null;
if (producerRunId && !producer) throw new Error("continuation_source_context_missing");
const producerIssueId = string(object(producer?.context).issueId);
const producerOrigins = new Set(continuationOriginCommentIds(producer?.context));
const recordedOrigins = triggerInteraction?.originCommentIds ?? [];
const inheritedOrigins = recordedOrigins.filter(id => producerOrigins.has(id));
// Older interactions copied the producer's origins without task scope.
// Ignore only references proven to be comments on that producer's other
// task. Missing rows, unrelated references, and explicit wake origins keep
// the existing fail-closed check below.
const inheritedForeignComments = producerIssueId && producerIssueId !== issueId && inheritedOrigins.length
? await db.select({ id: issueComments.id }).from(issueComments).where(and(
eq(issueComments.companyId, companyId),
sql`${issueComments.issueId}::text = ${producerIssueId}`,
inArray(sql<string>`${issueComments.id}::text`, inheritedOrigins),
))
: [];
const inheritedForeignIds = new Set(inheritedForeignComments.map(row => row.id));
const originCommentIds = [
...new Set([
...continuationOriginCommentIds(input.context),
...continuationOriginCommentIds(sourceRun?.context),
...recordedOrigins.filter(id => !inheritedForeignIds.has(id)),
...(triggerInteraction?.sourceCommentId
? [triggerInteraction.sourceCommentId]
: []),
]),
];
// Missing source rows cannot silently become a claim of complete context.
if (originCommentIds.some((id) => !rows.some((row) => row.id === id)))
throw new Error("continuation_source_context_missing");
const messages = rows.map((row) => {
const safe = input.exposeLowTrustRaw
? row
: sanitizeQuarantinedCommentForHigherTrust(row);
return {
id: row.id,
authorType:
row.authorType ??
(row.authorUserId ? "user" : row.authorAgentId ? "agent" : "system"),
authorId: row.authorUserId ?? row.authorAgentId,
createdByRunId: row.createdByRunId,
body: row.deletedAt ? "" : safe.body,
createdAt: row.createdAt.toISOString(),
updatedAt: row.updatedAt.toISOString(),
deleted: row.deletedAt !== null,
sourceTrust: row.sourceTrust,
};
});
const previousRun = input.previousContextRunId
? (
await db
.select({ context: heartbeatRuns.contextSnapshot })
.from(heartbeatRuns)
.where(
and(
eq(heartbeatRuns.companyId, companyId),
eq(heartbeatRuns.agentId, input.agentId),
eq(heartbeatRuns.id, input.previousContextRunId),
sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`,
),
)
)[0]
: null;
const priorEnvelope = object(previousRun?.context?.executionContinuation);
const deliveredMessages = Array.isArray(priorEnvelope.messages)
? priorEnvelope.messages.map(object)
: null;
const resumeDelta =
deliveredMessages && input.previousContextRunId
? {
baseRunId: input.previousContextRunId,
messages: messages.filter(
(message) =>
originCommentIds.includes(message.id) ||
!deliveredMessages.some(
(prior) =>
prior.id === message.id &&
prior.updatedAt === message.updatedAt &&
prior.body === message.body &&
prior.deleted === message.deleted &&
prior.authorId === message.authorId &&
(prior.createdByRunId ?? null) === message.createdByRunId &&
JSON.stringify(prior.sourceTrust) ===
JSON.stringify(message.sourceTrust),
),
),
}
: undefined;
const latestRequest = messages.findLast(
(row) =>
row.authorType === "user" && !row.createdByRunId && !row.deleted && row.body.trim().length > 0,
);
const hashObjectiveSource = (value: string) => createHash("sha256").update(value.trim()).digest("hex");
const objectiveSource = latestRequest
? { kind: "comment" as const, id: latestRequest.id, revision: latestRequest.updatedAt }
: issue.description !== null && issue.description !== undefined
? { kind: "description" as const, id: issue.id, revision: hashObjectiveSource(issue.description) }
: { kind: "title" as const, id: issue.id, revision: hashObjectiveSource(issue.title) };
const priorRuns = await db
.select({ id: heartbeatRuns.id, result: historyResult, status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode, runtimeMode: heartbeatRuns.runtimeMode, retryOfRunId: heartbeatRuns.retryOfRunId })
.from(heartbeatRuns)
.where(
and(
eq(heartbeatRuns.companyId, companyId),
or(eq(heartbeatRuns.agentId, input.agentId),
sourceRunId ? eq(heartbeatRuns.id, sourceRunId) : undefined),
sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`,
),
)
.orderBy(asc(heartbeatRuns.createdAt), asc(heartbeatRuns.id));
const completedActions = priorRuns.flatMap((run) =>
Object.entries(object(object(run.result).apiToolReceipts)).flatMap(
([receiptId, receipt]) => {
const value = object(receipt);
return value.state === "completed" &&
typeof value.operationId === "string"
? [
{
runId: run.id,
receiptId,
operationId: value.operationId,
result: continuationApiResult(value.operationId, value.result, companyId),
},
]
: [];
},
),
);
completedActions.push(...await githubRecoveryReceipts(db, companyId, issueId));
const reconciliations = await db
.select({
id: issueRecoveryActions.id,
evidence: issueRecoveryActions.evidence,
})
.from(issueRecoveryActions)
.where(
and(
eq(issueRecoveryActions.companyId, companyId),
eq(issueRecoveryActions.sourceIssueId, issueId),
eq(issueRecoveryActions.status, "resolved"),
),
);
const lastTerminal = priorRuns.findLast((run) =>
["succeeded", "failed", "timed_out", "interrupted", "cancelled"].includes(run.status) &&
!(run.status === "cancelled" && run.errorCode === "execution_reconciliation_required"),
);
if (explicitUserSource) {
const predecessor = priorRuns.find(run => run.id === explicitUserSource &&
["failed", "timed_out", "interrupted", "cancelled"].includes(run.status));
const failedRunId = string(explicitContinuation.failedRunId);
const retryWakes = failedRunId ? await db.select().from(agentWakeupRequests).where(and(
eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, input.agentId),
eq(agentWakeupRequests.reason, "retry_failed_run"), eq(agentWakeupRequests.requestedByActorType, "user"),
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
)) : [];
// Admission records the board operator's authority separately from the
// message author. At dispatch, prove that exact queue was adopted by this
// run; caller-supplied continuation context cannot grant this authority.
const queueIntentId = z.string().guid().safeParse(explicitContinuation.queuedCommentInterruptId);
const nativeQueuedContinuation = predecessor?.runtimeMode === "native" && ["cancelled", "failed"].includes(predecessor.status);
const [recordedQueue] = nativeQueuedContinuation && input.runId && queueIntentId.success ? await db.select().from(agentWakeupRequests).where(and(
eq(agentWakeupRequests.id, queueIntentId.data), eq(agentWakeupRequests.companyId, companyId),
eq(agentWakeupRequests.agentId, input.agentId), eq(agentWakeupRequests.status, "coalesced"),
eq(agentWakeupRequests.runId, input.runId), sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
)) : [];
const taskRetryId = string(explicitContinuation.taskRetryRequestId);
const [taskRetryWake] = taskRetryId && z.string().guid().safeParse(taskRetryId).success
? await db.select().from(agentWakeupRequests).where(and(
eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.id, taskRetryId),
eq(agentWakeupRequests.agentId, input.agentId), eq(agentWakeupRequests.reason, "task_retry"),
eq(agentWakeupRequests.requestedByActorType, "user"), eq(agentWakeupRequests.status, "coalesced"),
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
)) : [];
const taskRetry = readTaskRetryIntent(taskRetryWake?.payload);
const retryAuthorization = taskRetry && taskRetry.generationId === taskRetryWake?.id
&& taskRetry.actorId === taskRetryWake.requestedByActorId
? object(object(taskRetryWake.payload?.taskRetry).continuation) : {};
const continuationAuthorizations = [...reconciliations.map(row => object(row.evidence.explicitUserContinuation)),
...(recordedQueue ? [object(object(recordedQueue.payload?.queuedCommentInterrupt).continuation)] : []),
retryAuthorization]
.filter(value => value.previousRunId === explicitUserSource &&
(!input.runId || value.runId === input.runId) &&
value.commentId === explicitContinuation.commentId &&
priorRuns.some(run => run.id === value.runId));
const interruptQueueIds = [...new Set(continuationAuthorizations.flatMap(value => {
const parsed = z.string().guid().safeParse(value.queuedCommentInterruptId);
return parsed.success ? [parsed.data] : [];
}))];
const interruptQueues = interruptQueueIds.length
? await db.select().from(agentWakeupRequests).where(and(
inArray(agentWakeupRequests.id, interruptQueueIds),
eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, input.agentId),
eq(agentWakeupRequests.status, "coalesced"),
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is not null`,
input.runId ? eq(agentWakeupRequests.runId, input.runId) : undefined,
)) : [];
const authorization = continuationAuthorizations.find(value => taskRetryId
? value.taskRetryRequestId === taskRetryId && taskRetryWake?.requestedByActorId === value.actorId && (
// The Retry receipt stays bound to the run it admitted. A bounded
// automatic retry of that run carries its own admission receipt.
value.automaticRetry
? taskRetry?.generationId === taskRetryWake.id && taskRetry.actorId === value.actorId &&
object(value.automaticRetry).sourceRunId === explicitUserSource && issue.executionRunId === input.runId &&
priorRuns.some(run => run.id === value.runId && run.retryOfRunId === explicitUserSource)
: taskRetryWake.runId === value.runId &&
object(taskRetryWake.payload?.taskRetry).sourceRunId === value.previousRunId)
: failedRunId
? value.failedRunId === failedRunId && retryWakes.some(wake =>
wake.runId === value.runId && wake.requestedByActorId === value.actorId &&
priorRuns.some(run => run.id === wake.runId && run.retryOfRunId === failedRunId))
: value.interactionId && value.queuedCommentInterruptId
? interactions.some(interaction => interaction.id === value.interactionId && interaction.id === triggerInteraction?.id &&
["accepted", "answered", "rejected"].includes(interaction.status) &&
interruptQueues.some(queue => queue.id === value.queuedCommentInterruptId && queue.runId === value.runId &&
object(queue.payload?.queuedCommentInterrupt).actorId === value.actorId &&
queuedInteractionId(queue.payload) === interaction.id && queue.payload?.interactionStatus === interaction.status))
: value.interactionId
? (!input.runId || value.runId === input.runId) && priorRuns.some(run => run.id === value.runId) &&
interactions.some(interaction => interaction.id === value.interactionId &&
["accepted", "answered", "rejected"].includes(interaction.status) &&
interaction.resolvedByUserId === value.actorId && !interaction.resolvedByAgentId)
: rows.some(comment => comment.id === value.commentId &&
comment.authorType === "user" &&
(!("commentUpdatedAt" in value || "commentBodyHash" in value || value.automaticRetry) || (
value.commentUpdatedAt === comment.updatedAt.toISOString() &&
value.commentBodyHash === createHash("sha256").update(comment.body).digest("hex")
)) &&
(!value.automaticRetry || (
object(value.automaticRetry).sourceRunId === explicitUserSource &&
comment.body.trim().length > 0 && issue.executionRunId === input.runId &&
priorRuns.some(run => run.id === value.runId && run.retryOfRunId === explicitUserSource)
)) &&
(value.queuedCommentInterruptId
? interruptQueues.some(queue => queue.id === value.queuedCommentInterruptId &&
queue.runId === value.runId &&
object(object(queue.payload).queuedCommentInterrupt).actorId === value.actorId &&
queuedCommentIdsFromWakePayload(queue.payload).includes(comment.id))
: comment.authorUserId === value.actorId) &&
!comment.createdByRunId && !comment.deletedAt));
if (!predecessor || !authorization || explicitUserSource !== sourceRunId)
throw new Error("continuation_user_authorization_missing");
}
const interruptedRunId = explicitUserSource ?? (staleInterruptedSource ? undefined : interruptedSource) ?? (lastTerminal && lastTerminal.status !== "succeeded" &&
(hasConversationContinuationPolicy(lastTerminal.result) ||
lastTerminal.status === "interrupted" || lastTerminal.errorCode === "process_lost")
? lastTerminal.id : undefined);
const triggerReason = string(input.context.wakeReason) ?? "task_execution";
const retriedCandidate = await requeuedWakeReason(db, { companyId, issueId, runId: input.runId, context: input.context });
const retriedReason = retriedCandidate && retriedCandidate !== triggerReason && !retryWakeReasons.has(retriedCandidate)
? retriedCandidate : null;
return {
...(interruptedRunId ? { interruptedRunId } : {}),
...(resumeDelta ? { resumeDelta } : {}),
recoveryOutcomes: reconciliations
.filter((row) => row.evidence.executionReconciliation)
.map((row) => ({
recoveryActionId: row.id,
decision: row.evidence.executionReconciliation,
})),
version: 1,
companyId,
issueId,
trigger: {
reason: triggerReason,
interactionId: triggerInteraction?.id ?? null,
sourceRunId,
...(retriedReason ? { retriedReason } : {}),
},
originCommentIds,
objective: latestRequest?.body ?? issue.description ?? issue.title,
objectiveSource,
messages,
humanResponses: interactions.flatMap(row => {
const response = projectHumanInteractionResponse(row);
return response ? [response] : [];
}),
interactionOutcomes: [...interactions
.filter((row) => row.status !== "pending")
.map((row) => ({
id: row.id,
kind: row.kind,
status: row.status,
result: row.result,
})), ...await childReviewOutcomes(db, companyId, issueId, { type: "agent", agentId: input.agentId, companyId, onBehalfOfUserId: issue.responsibleUserId })],
// Low-trust evidence only: renderPaperclipWakePrompt removes completedWork
// from requestContext and encodes it in the fenced, non-authoritative
// continuation-evidence section. It cannot supply objective or authority.
completedWork: input.summary ??
string(object(object(sourceRun?.result).nativeResult).summary)?.slice(0, 32_000) ??
string(object(sourceRun?.result).summary)?.slice(0, 32_000) ?? null,
completedActions,
unresolvedInteractionIds: interactions
.filter((row) => row.status === "pending")
.map((row) => row.id),
coverage: {
kind: "full_task_history",
throughCommentId: messages.at(-1)?.id ?? null,
summaryThroughCommentId: null,
},
};
}