mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-08 12:33:05 +01:00
## Thinking Path > - Paperclip manages agent tasks and their saved execution history. > - A continuation carries completed API receipts to prevent duplicate work. > - Repeated saves can copy large record bodies into each new provider prompt. > - The resulting command can exceed the durable PRP frame limit before useful work starts. > - This change projects large successful saved-record responses as compact references. > - The original receipts and authenticated user direction remain available. ## Linked Issues or Issue Description **What happened?** A task with many saved document and comment responses can fail with `command_payload_too_large`. Same-run recovery can then show a runner identity failure that hides this first cause. Completed response bodies repeat content already stored in task records. **Expected behavior** Continue from saved records with completed-action evidence that fits the runner command. Keep user messages, answers and approvals intact. Keep the full original receipts. **Steps to reproduce** 1. Save 39 successful document, comment and task-write responses in a run's API receipt ledger. 2. Give each response a large record body. 3. Build and render a continuation, then queue it through the real durable PRP journal. 4. Before this fix, the regression throws `command_payload_too_large` with a 4.2 MB prompt. Related work: #16980 pages oversized assigned tool results. This change addresses earlier response bodies repeated in a later continuation. It does not change tool-result paging or protocol limits. ## What Changed - Project response data over 4 KiB for three known successful record-write operations. Keep record, revision and status metadata, HTTP outcome, receipt identity and a data digest. - Keep small, failed, unknown, mismatched and foreign-company results in their existing form. - Explain the reference format in shared prompt guidance and the execution contract. - Add database and real-journal regression coverage. Verify the original receipt ledger remains unchanged. The continuation service owns the projection. The shared adapter helper only explains the reference to native and legacy prompt consumers. There are no new dependencies, provider calls, timers or background scans. ## Verification - The unfixed database regression fails at the real PRP command limit with `command_payload_too_large`. - Continuation builder, native execution input and native continuation suites: 118 tests pass. - Shared prompt consumers: 142 tests pass. - Module boundaries and diff whitespace checks pass. - Greptile: 5/5 on `8109e7361c5b5e38f9aad35af9bf4ad7bdc5f338`, with no unresolved review threads. All CI checks pass. - Full workspace `pnpm -r typecheck` and `pnpm build`: passed. - Hosted verification: pending deployment. The source fix does not itself resume the affected hosted task. Local setup failures remain accounted for: the fresh checkout lacked an ignored root SDK link and embedded PostgreSQL library symlinks after an offline install that skipped lifecycle scripts. The maintained workspace-link helper, SDK build and platform hydration script repaired those prerequisites. A later test expectation assumed JSON insertion order across PostgreSQL JSONB; it now checks the digest against the unchanged stored receipt. These failures did not change runtime limits or recovery guards. ## Risks Large responses from these three successful operations now use explicit saved-record references in continuation context. Agents must read the record or source receipt when full content is needed. This is not a universal prompt-size limit: user history, unknown operations and other inputs retain their existing representation. No migration, permission, credential, provider-stop or replay policy changes. ## Model Used OpenAI GPT-6.1 Sol (`gpt-6.1-sol`), xhigh reasoning, through Codex in T3 Code. Used for investigation, implementation and local verification. No subagents. ## 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 linked existing issues or described the issue in-PR following the issue template - [x] I have not referenced internal/instance-local Paperclip issues or links - [x] My branch name describes the change and contains no internal task id - [x] I have run the targeted tests locally and they pass - [x] I have added tests and updated relevant documentation - [x] I have considered and documented risks - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open findings - [x] I will address review comments before requesting merge Co-authored-by: Paperclip <noreply@paperclip.ing>
576 lines
27 KiB
TypeScript
576 lines
27 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 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. */
|
|
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;
|
|
}): 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" && !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}`,
|
|
input.runId ? eq(agentWakeupRequests.runId, input.runId) : undefined,
|
|
)) : [];
|
|
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?.runId === value.runId &&
|
|
taskRetryWake.requestedByActorId === value.actorId &&
|
|
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
|
|
? 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))
|
|
: 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);
|
|
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: string(input.context.wakeReason) ?? "task_execution",
|
|
interactionId: triggerInteraction?.id ?? null,
|
|
sourceRunId,
|
|
},
|
|
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,
|
|
},
|
|
};
|
|
}
|