mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 00:56:52 +01:00
## Thinking Path > - Paperclip manages agents and their work. > - Task messages remain in a durable queue while an earlier run saves its result. > - That result can mark the task done before the user sends the saved messages. > - Send records current user intent, but its wake guard allowed only active task states. > - The old queue therefore remained pending with an obsolete finalization message. > - This change admits that exact Send receipt and reopens the task through the issue service in the successor transaction. > - The agent can receive the saved input once while the completed result remains unchanged. ## Linked Issues or Issue Description **What happened?** Send on a saved message can leave the message queued after its native predecessor succeeds and commits. The old wait message remains visible. Allowing admission alone also fails: continuation construction rejects a task that remains done. **Expected behavior** Explicit Send must deliver the saved input once through normal admission. It must retain the previous run and its result. **Steps to reproduce** 1. Save a follow-up while a native run finishes its result. 2. Let that run succeed, commit its accepted result, and mark the task done. 3. Send the saved follow-up. Before this change, the receipt stays deferred. Related public work: Refs #19542. That change selects recovered workspace files. This change handles the later saved-message admission. Refs #19602 for the independent workflow fixture repair required by CI. ## What Changed - Allow the exact explicit Send wake guard to accept a done task. Keep automatic recovery and cancelled task guards unchanged. - Reopen the task through the company-scoped issue service after admission gates pass. Commit reopening, queue consumption and the successor together. Publish the audit after commit. - Add a native succeeded/committed predecessor regression with unchanged saved work and one successor after two resume calls. - Verify real heartbeat startup under the clicking operator. Keep automatic input, cancelled tasks and paused agents held. - Check workspace preflight for the prospective reopened task. Use the normal company-scoped transition if preflight blocks a done task, so its completion metadata is cleared. - Check a successful native predecessor for its committed accepted result, current ownership and pending cleanup. Keep approvals and questions effective even without a recovery action. - Document the saved Send contract and update the queue test dependency. ## Verification - Final merge qualification for `675914f0fa9b5eb1d2c9c8fcd448e0669d0c4836`: all 56 applicable status checks pass, with no pending or failed checks. [Final CI](https://github.com/paperclipai/paperclip/actions/runs/42307623559) includes typecheck, build, all applicable test and E2E shards, and aggregate verification. Current-head Greptile is 5/5; all five review threads are resolved. Any paused or queued application review is not claimed as a new completed review. - Final base qualification at `675914f0fa9b5eb1d2c9c8fcd448e0669d0c4836`: official preflight passes (1.5s); all six integrated model-normalization cases pass (19.07s); both affected test typecheck layers pass. The integration changes no production server/shared/Runner/adapter inputs, so the preceding 12-case Send proof, two ordinary status guards, and unchanged production typecheck layers remain applicable. Fresh final-head CI supplies broad coverage. - New review regression: both failed and cancelled native predecessors on a task manually marked Done fail before the repair with `continuation_task_ownership_changed` through real `resumeQueuedCommentInterrupt` admission. At `bfb0eff7941090ac3ac43ce4366ba2e3d46612e9`, all 12 affected Send/source/cleanup/workspace/decision cases pass (246.80s), ordinary Done/cancelled rejection cases pass (2; 100.31s), and all seven server typecheck layers pass. History validation uses the prospective reopened status only for an already-verified explicit native Send. No status is changed during a dry run. - Prior `e64e4bb1` CI passed the affected serialized server suites, including the complete saved-message route suite, typecheck and build. General server shard 4 still failed two unrelated stale default-model expectations (`gpt-5.6-sol` versus the new `gpt-6.1-sol` default). The equivalent repair has already merged in #19610 as `8621bcf53ee2e4bbefb7d193add42dd544ca06e8`; the separately prepared #19645 is closed as superseded. This source fix does not weaken or edit those assertions. - Original RED: the committed-source and completed-task startup cases left the saved receipt deferred. - Review RED: Done plus an unrunnable isolated worktree created one successor instead of zero. A successful committed predecessor with pending lease cleanup also created one instead of zero. The new Done-to-blocked path retained stale completion time. Pending approvals and questions each allowed a successor. All failures were captured before their respective repairs. - Integrated production revision `99018264dd1ef002ff3500378184bcd167d9206b`: 417 affected cases pass across the complete native continuation, responsible-user startup, and workspace preflight suites (386 + 18 + 13). These include the original successful startup case and six cleanup/finalization faults that hold the exact saved receipt until settlement, then deliver it once. - `pnpm --filter @paperclipai/server typecheck` passes all seven layers on that integrated head, including the local Runner/SDK dependency build. Module-boundary, whitespace, and added-code-comment checks pass. - Normal master integrations preserve branch history. No force push or assertion weakening was used. Fresh CI and automated review are running on `675914f0fa9b5eb1d2c9c8fcd448e0669d0c4836`, which normally integrates the already-merged #19610 fixture repair after the reproduced inherited CI failure. - The earlier CI attempt failed in five unchanged workflow fixture cases because IMAGE_ONLY was unset. The merged #19602 repairs that fixture. A later inherited duplicate default in that independent fixture has a separate repair; this PR does not edit it. - Follow-up route-fixture RED: three existing native queue delivery cases failed because their fixture marked a run successful without an accepted result or committed finalization. The same three cases pass after the fixture records the complete native success contract (42.08s); delivery assertions and production gates are unchanged. The affected test-tree TypeScript layer also passes. The full route run passed all 119 assertions but failed its 30-second database teardown hook, so it is not recorded as a suite pass. That result is retained. One unchanged bounded rerun was stopped after more than 15 minutes without a final verdict; it is inconclusive, not a pass. - Final integrated head `e64e4bb1`: all 13 focused route and completed-task admission cases pass across two real-PostgreSQL suites (188.95s), including the three repaired route fixtures and every new cleanup/finalization/decision hold. Fresh preflight passes; all seven server typecheck layers pass. Required final CI supplies the broader route and server coverage. - These tests use real PostgreSQL and normal admission, cleanup, queue consumption, and startup services. The startup suite controls the provider boundary. Hosted provider acceptance remains pending normal deployment. ## Risks - Explicit Send can reopen a done task. This is limited to its validated saved receipt and current assignee. A cancelled task remains held. - Reopening uses the existing company-scoped issue transition and shares the successor transaction. A failed admission does not reopen the task. Provider Stop, workspace, approval, pause and budget checks retain their existing owners. - Cleanup and pending decisions retain the done task and saved receipt. An unrunnable workspace uses the existing blocked-task diagnostic. No schema or runtime package changes. ## Model Used - OpenAI Codex, GPT-6 family; exact serving snapshot/context window not exposed; repository analysis, code editing, tool execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
582 lines
28 KiB
TypeScript
582 lines
28 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;
|
|
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
|
|
? 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,
|
|
},
|
|
};
|
|
}
|