mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native execution can save an accepted result before workspace export fails. > - Recover must preserve the files and settle that same result without another agent turn. > - The export failure can own the task's Blocked transition and its recovery notice. > - Recovery must prove preservation, stop, source identity and current authority before settlement. > - Recover can also transfer repair ownership to the system, which must keep explicit board export repair available. > - This change restores those paths while preserving later task state, live owners and unknown action outcomes. ## Linked Issues or Issue Description Related #17916 protects unfinished exports. Related #19004 preserves deliberate Retry authority. This change repairs accepted-result finalization and its recovery notice. **What happened?** A native run accepted its result, then exhausted workspace export. Normal Recover could save the files but leave the result and old recovery notice unsettled. A saved Retry could also use a different generation from preservation. Separately, Recover changed the action owner from board to system. Explicit export Retry then rejected that same source as stale, even when its accepted result and stopped allocation still matched. **Expected behavior** Preserve files, finish the original accepted result and close only its obsolete notice after exact proof. Keep explicit board export repair available after system adoption. Preserve current authority, Stop, later task changes, newer execution and every physical ownership fence. **Steps to reproduce** 1. Accept a native result and exhaust its workspace export. 2. Request ordinary Recover, optionally with an existing saved Retry. 3. Let file preservation complete. Inspect the task and source recovery notice. 4. To reproduce the explicit-repair defect, call export Retry after Recover adopts the action. Before this repair it rejects the valid system-owned source before any provider call. ## What Changed - Record the exact accepted result, contract, task status version and prior decision when export exhaustion creates Blocked. - Bind passive finalization to complete, source-bound preservation and the current chat/review authority. Recheck the proof under status commit locks. - Keep the saved Retry identity separate from its preservation generation. Retain coalescing and reject malformed or mismatched history. - Preserve an existing published GitHub assessment for the no-publication status-unproven wait. Keep all semantic, result, authority and publication checks. - Close an old export notice only after its own accepted result, export or preservation receipts, effects, projection and old Runner ownership settle. Keep newer work and files unchanged. - Permit explicit board export Retry after Recover transfers the exact source to the system. Require valid system and workspace recovery records for that same run, plus every existing result, stop, allocation, task and shared physical-lock check. Reject missing proof, another source and agent ownership before provider access. - Merge current master while retaining the stronger Stop, source and verified preservation checks. - Update the execution contract. No schema, endpoint, Runner protocol, model, trust or provider artifact change. ## Verification - Current head: `9be5a73489801095c9f39c9bfcb285a22b055912`. - The explicit-repair regression fails before the new guard change: ordinary Recover adopts the real database action, then export Retry rejects it as stale. It passes after the repair. - On the merged source, three owning PostgreSQL/file suites pass **220 tests**: `preserved-native-response-wait.integration.test.ts`, `native-workspace-export-retry.test.ts` and `idle-completed-history.test.ts`. The export suite includes four unrelated-owner/source rejection controls and both board-owned and system-adopted active-exporter controls. Results and run counts prove that repair creates no provider turn. - Server layered typecheck, module boundaries, `git diff --check`, and the added-line secret/private-incident scan pass. - Full workspace typecheck (`pnpm -r typecheck`) and build (`pnpm build`) pass. The complete local `pnpm test:run` is still running; its completed serialized chat partitions pass. Current-head CI is green (53 successful checks, 3 expected skips). Greptile reviewed `9be5a73489801095c9f39c9bfcb285a22b055912` at 2026-11-05 00:58 UTC with confidence **5/5**; all review threads are resolved. - Original qualification remains supporting evidence: the accepted-result fixture exercises HTTP Recover, real files, the finalizer, saved Retry coalescing, eight commit races, current principal/resource/head changes, Stop, later task changes, notice closure and repeated delivery. No accepted response, provider turn or publication is replayed. - Earlier failures remain recorded. Original red controls exposed wrong-action preservation, unrelated finalization barriers, generation mismatch and a lost GitHub assessment; the original repair added controls for them. The prior CI run [41908572248](https://github.com/paperclipai/paperclip/actions/runs/41908572248) failed an unchanged completed-history runtime fixture because its loopback service refused connections through the original ten-second readiness window. That owning suite passes unchanged in the 220-case run above. The earlier readiness failure's cause remains unproven; no wait or assertion was weakened. - Local and CI tests do not establish hosted task recovery. The serving revision and original-task outcomes require separate verification after rollout. ## Risks - Historical failures without the exact export-owned Blocked receipt keep their current task status. A saved result does not grant execution or publication authority. - Explicit export repair still resumes only the same retained allocation under the shared physical lock. It cannot take over an active exporter, create another agent turn, choose a replacement sandbox or discard saved files. - Retiring a source notice does not release physical resources or modify a newer task owner, later Done state or saved snapshot history. - The proof adds bounded reads only for concrete recovery obligations. It adds no recurring worker or polling timer. ## Model Used OpenAI GPT-6 through Codex for the original repair. OpenAI GPT-6.1 Sol (`gpt-6.1-sol`), xhigh reasoning, for the merge, integration verification and system-adopted export repair. Both used repository tools, code execution and PostgreSQL tests. The runtime does not expose a context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used - [x] I have checked ROADMAP.md and searched for duplicate or related work - [x] I have described the issue and linked related public PRs - [x] I have included no internal task identifiers or private incident links - [x] I have added behavioral and safety regression tests - [x] I have documented verification limits and risks - [x] I have updated the execution contract - [x] All Paperclip CI gates are green on the current head - [x] Greptile is 5/5 on the current head with no unresolved comments - [x] I will address review comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing> Co-authored-by: Devin Foley <devin@paperclip.ing>
1687 lines
58 KiB
TypeScript
1687 lines
58 KiB
TypeScript
import { notifyIssueRecovery } from "../issue-recovery-notifications.js";
|
|
import { createHash } from "node:crypto";
|
|
|
|
import { and, desc, eq, inArray, isNull, sql } from "drizzle-orm";
|
|
import type { Db } from "@paperclipai/db";
|
|
import {
|
|
assets,
|
|
agents,
|
|
chatConversations,
|
|
chatDeliveries,
|
|
chatEndpointResources,
|
|
chatEndpoints,
|
|
chatExternalPrincipals,
|
|
chatIdentityLinks,
|
|
chatMessageLinks,
|
|
chatPublications,
|
|
companyMemberships,
|
|
heartbeatRuns,
|
|
issueAttachments,
|
|
issueComments,
|
|
issues,
|
|
issueWorkProducts,
|
|
runIdentityContexts,
|
|
} from "@paperclipai/db";
|
|
|
|
import {
|
|
isAllowedContentType,
|
|
MAX_ATTACHMENT_BYTES,
|
|
normalizeContentType,
|
|
} from "../../attachment-types.js";
|
|
import { getStorageService } from "../../storage/index.js";
|
|
import type { StorageService } from "../../storage/types.js";
|
|
import { loadConfig } from "../../config.js";
|
|
import { HttpError } from "../../errors.js";
|
|
import { voiceSessionStore, type VoiceTransaction } from "../voice/voice-session-store.js";
|
|
import { issueService } from "../issues.js";
|
|
import { readInheritedGitHubReviewWait } from "../chat-channels/github/review-wait.js";
|
|
import { boundExternalChatProvider } from "./external-chat-provider.js";
|
|
import { resolveExternalChatQuestionResponse } from "./external-chat-question-response.js";
|
|
import { githubRunReplyState } from "../chat-run-publications.js";
|
|
import { readPreservedNativeResultFinalization } from "./preserved-native-result-finalization.js";
|
|
|
|
// This boundary reads identity and wake authority, never launch renderings,
|
|
// saved provider profiles, or run output. Keep the nested wake proof narrow too.
|
|
const responseWaitContextKeys = ["source", "issueId", "wakeReason", "wakeCommentId", "wakeCommentIds",
|
|
"interactionId", "interactionKind", "interactionStatus", "sourceRunId", "sourceCommentId",
|
|
"externalChatContinuation", "paperclipExternalChatQuestionResponse", "paperclipHarnessCheckedOut",
|
|
"paperclipExternalChatExecutionBound"] as const;
|
|
const responseWaitRun = {
|
|
id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
|
|
nativeIssueId: heartbeatRuns.nativeIssueId, issueId: heartbeatRuns.issueId, status: heartbeatRuns.status,
|
|
responsibleUserId: heartbeatRuns.responsibleUserId, activeIdentityContextId: heartbeatRuns.activeIdentityContextId,
|
|
contextSnapshot: sql<Record<string, unknown> | null>`case when jsonb_typeof(${heartbeatRuns.contextSnapshot}) = 'object'
|
|
then (select coalesce(jsonb_object_agg(field.key, field.value), '{}'::jsonb)
|
|
from jsonb_each(${heartbeatRuns.contextSnapshot}) as field(key, value)
|
|
where field.key in (${sql.join(responseWaitContextKeys.map(key => sql`${key}`), sql`, `)}))
|
|
|| jsonb_build_object('paperclipWake', jsonb_build_object(
|
|
'checkedOutByHarness', ${heartbeatRuns.contextSnapshot}#>'{paperclipWake,checkedOutByHarness}',
|
|
'externalChatExecutionBound', ${heartbeatRuns.contextSnapshot}#>'{paperclipWake,externalChatExecutionBound}',
|
|
'externalChatProvider', ${heartbeatRuns.contextSnapshot}#>'{paperclipWake,externalChatProvider}',
|
|
'issue', jsonb_build_object('id', ${heartbeatRuns.contextSnapshot}#>'{paperclipWake,issue,id}'),
|
|
'commentIds', ${heartbeatRuns.contextSnapshot}#>'{paperclipWake,commentIds}'))
|
|
else null end`,
|
|
};
|
|
|
|
export const LIST_CHAT_ATTACHMENTS_TOOL_NAME = "list_chat_attachments";
|
|
export const REUSE_CHAT_ATTACHMENT_TOOL_NAME = "reuse_chat_attachment";
|
|
|
|
export const LIST_CHAT_ATTACHMENTS_TOOL_DEFINITION = Object.freeze({
|
|
name: LIST_CHAT_ATTACHMENTS_TOOL_NAME,
|
|
description:
|
|
"List metadata for files already received or delivered in this same external-chat conversation. Call with {} for the latest page (default 20), or set limit to an integer from 1 through 50. Page through nextCursor to find older files, or filter by an exact sourceCommentId when known. File bytes and private storage locations are never returned.",
|
|
inputSchema: {
|
|
type: "object",
|
|
properties: {
|
|
sourceCommentId: {
|
|
type: ["string", "null"],
|
|
description:
|
|
"Optional exact sourceCommentId returned by this tool on an earlier page.",
|
|
},
|
|
limit: { type: "integer", minimum: 1, maximum: 50, default: 20 },
|
|
cursor: {
|
|
type: ["string", "null"],
|
|
maxLength: 1024,
|
|
description: "Opaque nextCursor from the immediately preceding page.",
|
|
},
|
|
},
|
|
required: [],
|
|
additionalProperties: false,
|
|
},
|
|
annotations: {
|
|
semanticContract: "paperclip.server-chat-attachment-reuse.v1",
|
|
operationId: LIST_CHAT_ATTACHMENTS_TOOL_NAME,
|
|
version: 1,
|
|
exposure: "run_scoped",
|
|
requiredClaims: [],
|
|
},
|
|
});
|
|
|
|
export const REUSE_CHAT_ATTACHMENT_TOOL_DEFINITION = Object.freeze({
|
|
name: REUSE_CHAT_ATTACHMENT_TOOL_NAME,
|
|
description:
|
|
"Prepare one exact file previously received or delivered in this same external-chat conversation for the current response. All four arguments are required: copy sourceCommentId and attachmentId from list_chat_attachments, provide a nonempty title, and choose a nonempty idempotencyKey of at most 200 characters once for this reuse; send that same key unchanged on every retry. This queues a verified copy for final publication; it does not confirm provider delivery.",
|
|
inputSchema: {
|
|
type: "object",
|
|
properties: {
|
|
idempotencyKey: { type: "string", minLength: 1, maxLength: 200 },
|
|
sourceCommentId: { type: "string", format: "uuid" },
|
|
attachmentId: { type: "string", format: "uuid" },
|
|
title: { type: "string", minLength: 1, maxLength: 500 },
|
|
},
|
|
required: ["idempotencyKey", "sourceCommentId", "attachmentId", "title"],
|
|
additionalProperties: false,
|
|
},
|
|
annotations: {
|
|
semanticContract: "paperclip.server-chat-attachment-reuse.v1",
|
|
operationId: REUSE_CHAT_ATTACHMENT_TOOL_NAME,
|
|
version: 1,
|
|
exposure: "run_scoped",
|
|
requiredClaims: [],
|
|
},
|
|
});
|
|
|
|
export type ChatReuseBinding = {
|
|
companyId: string;
|
|
issueId: string;
|
|
runId: string;
|
|
agentId: string;
|
|
};
|
|
|
|
type AuthorizedConversation = {
|
|
conversationId: string;
|
|
endpointId: string;
|
|
};
|
|
|
|
export type ExternalChatResponseWaitAuthorization =
|
|
| "authorized"
|
|
| "revoked"
|
|
| "not_applicable";
|
|
|
|
type AuthorizationLockMode = "blocking" | "nonblocking" | "read";
|
|
|
|
type AuthorizationEndpoint = Pick<typeof chatEndpoints.$inferSelect,
|
|
"id" | "companyId" | "provider" | "providerAccountId" | "status" | "setup"
|
|
| "allowDirectMessages" | "allowGroupChats" | "allowUnlinkedPeople" | "sponsorUserId"
|
|
>;
|
|
type AuthorizationConversation = Pick<typeof chatConversations.$inferSelect,
|
|
"id" | "resourceId" | "isDirectMessage"
|
|
>;
|
|
type AuthorizationResource = Pick<typeof chatEndpointResources.$inferSelect,
|
|
"availability" | "type" | "enabled"
|
|
>;
|
|
|
|
class ExternalChatWaitAuthorizationContentionError extends Error {
|
|
constructor() {
|
|
super("paperclip_external_chat_wait_authorization_contended");
|
|
this.name = "ExternalChatWaitAuthorizationContentionError";
|
|
}
|
|
}
|
|
|
|
export function isExternalChatWaitAuthorizationContention(
|
|
error: unknown,
|
|
): boolean {
|
|
let current: unknown = error;
|
|
for (let depth = 0; depth < 4 && current; depth += 1) {
|
|
if (current instanceof ExternalChatWaitAuthorizationContentionError) {
|
|
return true;
|
|
}
|
|
const value = record(current);
|
|
if (value.code === "55P03") return true;
|
|
current = value.cause;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
export type ChatAttachmentReuseSource = {
|
|
sourceCommentId: string;
|
|
attachmentId: string;
|
|
filename: string;
|
|
contentType: string;
|
|
byteSize: number;
|
|
sha256: string;
|
|
objectKey: string;
|
|
createdAt: Date;
|
|
};
|
|
|
|
export type ListedChatAttachment = Omit<
|
|
ChatAttachmentReuseSource,
|
|
"objectKey" | "createdAt"
|
|
> & {
|
|
createdAt: string;
|
|
contentAccess: "metadata_only";
|
|
};
|
|
|
|
export type ListedChatAttachmentPage = {
|
|
attachments: ListedChatAttachment[];
|
|
nextCursor: string | null;
|
|
complete: boolean;
|
|
};
|
|
|
|
type ListCursor = {
|
|
schema: "paperclip.chat-attachment-list-cursor.v1";
|
|
conversationId: string;
|
|
sourceCommentId: string | null;
|
|
createdAt: string;
|
|
attachmentId: string;
|
|
sourceCommentIdTieBreak: string;
|
|
};
|
|
|
|
export type PreparedReusedChatAttachment = {
|
|
result: {
|
|
commandId: string;
|
|
disposition: "applied";
|
|
stateRevision: number;
|
|
entityRefs: string[];
|
|
scheduledWakeIds: never[];
|
|
source: {
|
|
commentId: string;
|
|
attachmentId: string;
|
|
sha256: string;
|
|
};
|
|
prepared: {
|
|
attachmentId: string;
|
|
workProductId: string;
|
|
commentId: string;
|
|
filename: string;
|
|
byteSize: number;
|
|
sha256: string;
|
|
};
|
|
};
|
|
rollbackDefinitePreCommitFailure: (() => Promise<void>) | null;
|
|
};
|
|
|
|
function record(value: unknown): Record<string, unknown> {
|
|
return typeof value === "object" && value !== null && !Array.isArray(value)
|
|
? (value as Record<string, unknown>)
|
|
: {};
|
|
}
|
|
|
|
function wakeCommentIds(contextSnapshot: unknown): string[] {
|
|
const context = record(contextSnapshot);
|
|
const raw = context.wakeCommentIds;
|
|
if (!Array.isArray(raw)) return [];
|
|
return [
|
|
...new Set(
|
|
raw
|
|
.filter(
|
|
(value): value is string =>
|
|
typeof value === "string" && value.trim().length > 0,
|
|
)
|
|
.map((value) => value.trim()),
|
|
),
|
|
];
|
|
}
|
|
|
|
function publicationAttachmentIds(payload: unknown): string[] {
|
|
const raw = record(payload).attachmentIds;
|
|
return Array.isArray(raw)
|
|
? raw.filter((value): value is string => typeof value === "string")
|
|
: [];
|
|
}
|
|
|
|
function isUuid(value: unknown): value is string {
|
|
return (
|
|
typeof value === "string" &&
|
|
/^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/iu.test(
|
|
value,
|
|
)
|
|
);
|
|
}
|
|
|
|
function encodeListCursor(cursor: ListCursor): string {
|
|
return Buffer.from(JSON.stringify(cursor), "utf8").toString("base64url");
|
|
}
|
|
|
|
function decodeListCursor(
|
|
value: unknown,
|
|
conversationId: string,
|
|
sourceCommentId: string | null,
|
|
): ListCursor | null {
|
|
if (value === null || value === undefined) return null;
|
|
if (typeof value !== "string" || value.length === 0 || value.length > 1024) {
|
|
throw new Error("paperclip_runner_chat_attachment_cursor_invalid");
|
|
}
|
|
try {
|
|
const parsed = record(
|
|
JSON.parse(Buffer.from(value, "base64url").toString("utf8")),
|
|
);
|
|
const createdAt =
|
|
typeof parsed.createdAt === "string" ? parsed.createdAt : "";
|
|
const parsedDate = new Date(createdAt);
|
|
if (
|
|
parsed.schema !== "paperclip.chat-attachment-list-cursor.v1" ||
|
|
parsed.conversationId !== conversationId ||
|
|
(parsed.sourceCommentId ?? null) !== sourceCommentId ||
|
|
Number.isNaN(parsedDate.getTime()) ||
|
|
!isUuid(parsed.attachmentId) ||
|
|
!isUuid(parsed.sourceCommentIdTieBreak)
|
|
) {
|
|
throw new Error("invalid");
|
|
}
|
|
return parsed as ListCursor;
|
|
} catch {
|
|
throw new Error("paperclip_runner_chat_attachment_cursor_invalid");
|
|
}
|
|
}
|
|
|
|
function destinationAllowed(
|
|
endpoint: AuthorizationEndpoint,
|
|
conversation: AuthorizationConversation,
|
|
resource: AuthorizationResource | null,
|
|
): boolean {
|
|
if (conversation.isDirectMessage) return endpoint.allowDirectMessages;
|
|
if (!resource || resource.availability !== "available") return false;
|
|
if (
|
|
endpoint.provider === "microsoft-teams" &&
|
|
resource.type === "group_chat"
|
|
) {
|
|
return endpoint.allowGroupChats;
|
|
}
|
|
return resource.enabled;
|
|
}
|
|
|
|
function currentSetupTestAcceptsBoundDeliveries(
|
|
endpoint: AuthorizationEndpoint,
|
|
deliveries: Array<{
|
|
normalizedEvent: unknown;
|
|
processedAt: Date | null;
|
|
receivedAt: Date;
|
|
}>,
|
|
): boolean {
|
|
if (endpoint.status === "active") return true;
|
|
if (endpoint.status !== "verifying" || deliveries.length === 0) return false;
|
|
const setup = record(endpoint.setup);
|
|
const testStartedAt =
|
|
typeof setup.testStartedAt === "string"
|
|
? new Date(setup.testStartedAt)
|
|
: null;
|
|
const generation = setup.runtimeGeneration;
|
|
if (
|
|
setup.step !== "test" ||
|
|
!testStartedAt ||
|
|
Number.isNaN(testStartedAt.getTime()) ||
|
|
typeof generation !== "number" ||
|
|
!Number.isSafeInteger(generation) ||
|
|
generation < 0
|
|
) {
|
|
return false;
|
|
}
|
|
return deliveries.every((delivery) => {
|
|
const runtimeContext = record(
|
|
record(delivery.normalizedEvent).runtimeContext,
|
|
);
|
|
return (
|
|
delivery.receivedAt >= testStartedAt &&
|
|
delivery.processedAt !== null &&
|
|
delivery.processedAt >= testStartedAt &&
|
|
delivery.processedAt >= delivery.receivedAt &&
|
|
runtimeContext.generation === generation &&
|
|
typeof runtimeContext.credentialFingerprint === "string" &&
|
|
/^[a-f0-9]{64}$/u.test(runtimeContext.credentialFingerprint)
|
|
);
|
|
});
|
|
}
|
|
|
|
async function principalAuthorized(
|
|
tx: Db,
|
|
endpoint: AuthorizationEndpoint,
|
|
principalId: string,
|
|
lockMode: AuthorizationLockMode = "blocking",
|
|
): Promise<boolean> {
|
|
if (lockMode === "blocking") {
|
|
await tx.execute(
|
|
sql`select pg_advisory_xact_lock(hashtextextended(${`chat-identity:${endpoint.companyId}:${principalId}`}, 0))`,
|
|
);
|
|
} else if (lockMode === "nonblocking") {
|
|
const [lock] = (await tx.execute(
|
|
sql`select pg_try_advisory_xact_lock(hashtextextended(${`chat-identity:${endpoint.companyId}:${principalId}`}, 0)) as acquired`,
|
|
)) as unknown as Array<{ acquired: boolean }>;
|
|
if (!lock?.acquired) {
|
|
throw new ExternalChatWaitAuthorizationContentionError();
|
|
}
|
|
}
|
|
const principalQuery = tx
|
|
.select({ id: chatExternalPrincipals.id, externalId: chatExternalPrincipals.externalId })
|
|
.from(chatExternalPrincipals)
|
|
.where(
|
|
and(
|
|
eq(chatExternalPrincipals.id, principalId),
|
|
eq(chatExternalPrincipals.companyId, endpoint.companyId),
|
|
eq(chatExternalPrincipals.provider, endpoint.provider),
|
|
eq(
|
|
chatExternalPrincipals.providerAccountId,
|
|
endpoint.providerAccountId ?? "",
|
|
),
|
|
),
|
|
);
|
|
const [principal] = await (lockMode === "read"
|
|
? principalQuery.limit(1)
|
|
: lockMode === "nonblocking"
|
|
? principalQuery.for("update", { noWait: true }).limit(1)
|
|
: principalQuery.for("update").limit(1));
|
|
if (!principal) return false;
|
|
if (endpoint.provider === "speko") {
|
|
try {
|
|
await voiceSessionStore(tx, { allowLocalBoard: loadConfig().deploymentMode === "local_trusted", nonblockingAuthority: lockMode === "nonblocking" })
|
|
.authorizePrincipal(tx as unknown as VoiceTransaction, endpoint.companyId, endpoint.id, principal.externalId);
|
|
return true;
|
|
} catch (error) {
|
|
if (error instanceof HttpError && [403, 404].includes(error.status)) return false;
|
|
throw error;
|
|
}
|
|
}
|
|
const linkQuery = tx
|
|
.select({
|
|
status: chatIdentityLinks.status,
|
|
userId: chatIdentityLinks.paperclipUserId,
|
|
})
|
|
.from(chatIdentityLinks)
|
|
.where(
|
|
and(
|
|
eq(chatIdentityLinks.companyId, endpoint.companyId),
|
|
eq(chatIdentityLinks.endpointId, endpoint.id),
|
|
eq(chatIdentityLinks.principalId, principalId),
|
|
),
|
|
);
|
|
const [link] = await (lockMode === "read"
|
|
? linkQuery.limit(1)
|
|
: lockMode === "nonblocking"
|
|
? linkQuery.for("update", { noWait: true }).limit(1)
|
|
: linkQuery.for("update").limit(1));
|
|
const activeMember = async (userId: string) => {
|
|
const membershipQuery = tx
|
|
.select({
|
|
status: companyMemberships.status,
|
|
role: companyMemberships.membershipRole,
|
|
})
|
|
.from(companyMemberships)
|
|
.where(
|
|
and(
|
|
eq(companyMemberships.companyId, endpoint.companyId),
|
|
eq(companyMemberships.principalType, "user"),
|
|
eq(companyMemberships.principalId, userId),
|
|
),
|
|
);
|
|
const [membership] = await (lockMode === "read"
|
|
? membershipQuery.limit(1)
|
|
: lockMode === "nonblocking"
|
|
? membershipQuery.for("update", { noWait: true }).limit(1)
|
|
: membershipQuery.for("update").limit(1));
|
|
return membership?.status === "active" && membership.role !== "viewer";
|
|
};
|
|
if (link?.status === "linked") {
|
|
return Boolean(link.userId && (await activeMember(link.userId)));
|
|
}
|
|
if (!endpoint.allowUnlinkedPeople) return false;
|
|
return endpoint.sponsorUserId ? activeMember(endpoint.sponsorUserId) : true;
|
|
}
|
|
|
|
function prepareBoundRunLinks(tx: Db, lockMode: AuthorizationLockMode) {
|
|
const linksQuery = tx
|
|
.select({
|
|
commentId: chatMessageLinks.commentId,
|
|
conversation: {
|
|
id: chatConversations.id,
|
|
resourceId: chatConversations.resourceId,
|
|
isDirectMessage: chatConversations.isDirectMessage,
|
|
},
|
|
endpoint: {
|
|
id: chatEndpoints.id,
|
|
companyId: chatEndpoints.companyId,
|
|
provider: chatEndpoints.provider,
|
|
providerAccountId: chatEndpoints.providerAccountId,
|
|
status: chatEndpoints.status,
|
|
allowDirectMessages: chatEndpoints.allowDirectMessages,
|
|
allowGroupChats: chatEndpoints.allowGroupChats,
|
|
allowUnlinkedPeople: chatEndpoints.allowUnlinkedPeople,
|
|
sponsorUserId: chatEndpoints.sponsorUserId,
|
|
setup: sql<AuthorizationEndpoint["setup"]>`jsonb_build_object(
|
|
'step', ${chatEndpoints.setup}->'step',
|
|
'testStartedAt', ${chatEndpoints.setup}->'testStartedAt',
|
|
'runtimeGeneration', ${chatEndpoints.setup}->'runtimeGeneration'
|
|
)`,
|
|
},
|
|
principalId: chatDeliveries.principalId,
|
|
normalizedEvent: sql<unknown>`jsonb_build_object('runtimeContext', jsonb_build_object(
|
|
'generation', ${chatDeliveries.normalizedEvent}->'runtimeContext'->'generation',
|
|
'credentialFingerprint', ${chatDeliveries.normalizedEvent}->'runtimeContext'->'credentialFingerprint'
|
|
))`,
|
|
processedAt: chatDeliveries.processedAt,
|
|
receivedAt: chatDeliveries.receivedAt,
|
|
})
|
|
.from(chatMessageLinks)
|
|
.innerJoin(
|
|
chatDeliveries,
|
|
and(
|
|
eq(chatDeliveries.id, chatMessageLinks.deliveryId),
|
|
eq(chatDeliveries.companyId, chatMessageLinks.companyId),
|
|
eq(chatDeliveries.endpointId, chatMessageLinks.endpointId),
|
|
eq(chatDeliveries.conversationId, chatMessageLinks.conversationId),
|
|
),
|
|
)
|
|
.innerJoin(
|
|
chatConversations,
|
|
and(
|
|
eq(chatConversations.id, chatMessageLinks.conversationId),
|
|
eq(chatConversations.companyId, chatMessageLinks.companyId),
|
|
eq(chatConversations.endpointId, chatMessageLinks.endpointId),
|
|
),
|
|
)
|
|
.innerJoin(
|
|
chatEndpoints,
|
|
and(
|
|
eq(chatEndpoints.id, chatConversations.endpointId),
|
|
eq(chatEndpoints.companyId, chatConversations.companyId),
|
|
),
|
|
)
|
|
.where(
|
|
and(
|
|
eq(chatMessageLinks.companyId, sql.placeholder("companyId")),
|
|
eq(chatMessageLinks.direction, "inbound"),
|
|
sql`${chatMessageLinks.commentId} = any(${sql.placeholder("commentIds")}::uuid[])`,
|
|
eq(chatDeliveries.state, "processed"),
|
|
eq(chatConversations.issueId, sql.placeholder("issueId")),
|
|
inArray(chatConversations.state, ["active", "waiting"]),
|
|
eq(chatEndpoints.provider, sql.placeholder("provider")),
|
|
eq(chatEndpoints.assignedAgentId, sql.placeholder("agentId")),
|
|
inArray(chatEndpoints.status, ["active", "verifying"]),
|
|
),
|
|
);
|
|
return (lockMode === "read"
|
|
? linksQuery
|
|
: lockMode === "nonblocking"
|
|
? linksQuery.for("update", { noWait: true })
|
|
: linksQuery.for("update")).prepare("chat_bound_run_links");
|
|
}
|
|
|
|
const boundRunLinksTemplates = new WeakMap<
|
|
Db,
|
|
Partial<Record<AuthorizationLockMode, ReturnType<typeof prepareBoundRunLinks>>>
|
|
>();
|
|
|
|
function boundRunLinksTemplate(tx: Db, lockMode: AuthorizationLockMode) {
|
|
let templates = boundRunLinksTemplates.get(tx);
|
|
if (!templates) {
|
|
templates = {};
|
|
boundRunLinksTemplates.set(tx, templates);
|
|
}
|
|
return templates[lockMode] ??= prepareBoundRunLinks(tx, lockMode);
|
|
}
|
|
|
|
/** Caller must independently verify the current run/issue execution owner. */
|
|
export async function authorizeChatConversationForBoundRun(
|
|
tx: Db,
|
|
binding: ChatReuseBinding,
|
|
contextSnapshot: unknown,
|
|
lockMode: AuthorizationLockMode = "blocking",
|
|
): Promise<AuthorizedConversation> {
|
|
let context = record(contextSnapshot);
|
|
if (context.source === "issue.interaction.respond") {
|
|
const answer = await resolveExternalChatQuestionResponse(tx, binding, context, lockMode);
|
|
if (!answer) throw new Error("paperclip_runner_chat_attachment_binding_denied");
|
|
context = answer.authorizationContext;
|
|
}
|
|
const source = typeof context.source === "string" ? context.source : "";
|
|
const provider = boundExternalChatProvider(source);
|
|
const commentIds = wakeCommentIds(context);
|
|
if (
|
|
!provider ||
|
|
(context.paperclipHarnessCheckedOut !== true &&
|
|
context.paperclipExternalChatExecutionBound !== true) ||
|
|
commentIds.length === 0
|
|
) {
|
|
throw new Error("paperclip_runner_chat_attachment_binding_denied");
|
|
}
|
|
const links = await boundRunLinksTemplate(tx, lockMode).execute({
|
|
companyId: binding.companyId,
|
|
issueId: binding.issueId,
|
|
agentId: binding.agentId,
|
|
provider,
|
|
commentIds,
|
|
});
|
|
const linkedCommentIds = new Set(links.map((row) => row.commentId));
|
|
const conversationIds = new Set(links.map((row) => row.conversation.id));
|
|
const endpointIds = new Set(links.map((row) => row.endpoint.id));
|
|
if (
|
|
links.length === 0 ||
|
|
conversationIds.size !== 1 ||
|
|
endpointIds.size !== 1 ||
|
|
!commentIds.every((id) => linkedCommentIds.has(id)) ||
|
|
links.some((row) => !row.principalId)
|
|
) {
|
|
throw new Error("paperclip_runner_chat_attachment_binding_denied");
|
|
}
|
|
const conversation = links[0]!.conversation;
|
|
const endpoint = links[0]!.endpoint;
|
|
if (!currentSetupTestAcceptsBoundDeliveries(endpoint, links)) {
|
|
throw new Error("paperclip_runner_chat_attachment_binding_denied");
|
|
}
|
|
const resource = conversation.resourceId
|
|
? await (() => {
|
|
const query = tx
|
|
.select({
|
|
availability: chatEndpointResources.availability,
|
|
type: chatEndpointResources.type,
|
|
enabled: chatEndpointResources.enabled,
|
|
})
|
|
.from(chatEndpointResources)
|
|
.where(
|
|
and(
|
|
eq(chatEndpointResources.id, conversation.resourceId!),
|
|
eq(chatEndpointResources.companyId, binding.companyId),
|
|
eq(chatEndpointResources.endpointId, endpoint.id),
|
|
),
|
|
);
|
|
return lockMode === "read"
|
|
? query.then((rows) => rows[0] ?? null)
|
|
: lockMode === "nonblocking"
|
|
? query
|
|
.for("update", { noWait: true })
|
|
.then((rows) => rows[0] ?? null)
|
|
: query.for("update").then((rows) => rows[0] ?? null);
|
|
})()
|
|
: null;
|
|
if (!destinationAllowed(endpoint, conversation, resource)) {
|
|
throw new Error("paperclip_runner_chat_attachment_destination_denied");
|
|
}
|
|
for (const principalId of new Set(links.map((row) => row.principalId!))) {
|
|
if (!(await principalAuthorized(tx, endpoint, principalId, lockMode))) {
|
|
throw new Error("paperclip_runner_chat_attachment_principal_denied");
|
|
}
|
|
}
|
|
return { conversationId: conversation.id, endpointId: endpoint.id };
|
|
}
|
|
|
|
function externalChatWaitCandidate(
|
|
contextSnapshot: unknown,
|
|
binding: ChatReuseBinding,
|
|
): { provider: string; commentIds: string[] } | null {
|
|
const context = record(contextSnapshot);
|
|
const source = typeof context.source === "string" ? context.source : "";
|
|
const provider = boundExternalChatProvider(source);
|
|
const commentIds = wakeCommentIds(context);
|
|
const wake = record(context.paperclipWake);
|
|
const wakeIssue = record(wake.issue);
|
|
const payloadCommentIds = Array.isArray(wake.commentIds)
|
|
? wake.commentIds.filter(
|
|
(value): value is string =>
|
|
typeof value === "string" && value.trim().length > 0,
|
|
)
|
|
: [];
|
|
if (
|
|
!provider ||
|
|
!(
|
|
(context.paperclipHarnessCheckedOut === true &&
|
|
wake.checkedOutByHarness === true) ||
|
|
(context.paperclipExternalChatExecutionBound === true &&
|
|
wake.externalChatExecutionBound === true)
|
|
) ||
|
|
commentIds.length === 0 ||
|
|
wake.externalChatProvider !== provider ||
|
|
wakeIssue.id !== binding.issueId ||
|
|
payloadCommentIds.length !== commentIds.length ||
|
|
payloadCommentIds.some((id, index) => id !== commentIds[index])
|
|
) {
|
|
return null;
|
|
}
|
|
return { provider, commentIds };
|
|
}
|
|
|
|
/**
|
|
* Authorize a no-work external-chat wait from the same durable boundary used
|
|
* by historical file reuse. The nested wake payload proves that this was a
|
|
* server-built chat turn; current destination and principal policy remain
|
|
* authoritative at the instant the status decision commits.
|
|
*/
|
|
export async function resolveExternalChatResponseWaitAuthorizationInTransaction(
|
|
tx: Db,
|
|
binding: ChatReuseBinding,
|
|
lockMode: AuthorizationLockMode = "blocking",
|
|
): Promise<ExternalChatResponseWaitAuthorization> {
|
|
// Preserve the native mutation lock order (issue -> run -> actor). The
|
|
// status committer already holds the issue lock; PostgreSQL treats this
|
|
// repeated lock as a no-op.
|
|
const issueQuery = tx
|
|
.select({
|
|
assigneeAgentId: issues.assigneeAgentId,
|
|
executionRunId: issues.executionRunId,
|
|
})
|
|
.from(issues)
|
|
.where(
|
|
and(
|
|
eq(issues.id, binding.issueId),
|
|
eq(issues.companyId, binding.companyId),
|
|
),
|
|
);
|
|
const [issue] = await (lockMode === "read"
|
|
? issueQuery.limit(1)
|
|
: lockMode === "nonblocking"
|
|
? issueQuery.for("update", { noWait: true }).limit(1)
|
|
: issueQuery.for("update").limit(1));
|
|
|
|
const runQuery = tx
|
|
.select(responseWaitRun)
|
|
.from(heartbeatRuns)
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.id, binding.runId),
|
|
eq(heartbeatRuns.companyId, binding.companyId),
|
|
eq(heartbeatRuns.agentId, binding.agentId),
|
|
eq(heartbeatRuns.nativeIssueId, binding.issueId),
|
|
eq(heartbeatRuns.runtimeMode, "native"),
|
|
),
|
|
);
|
|
const [run] = await (lockMode === "read"
|
|
? runQuery.limit(1)
|
|
: lockMode === "nonblocking"
|
|
? runQuery.for("update", { noWait: true }).limit(1)
|
|
: runQuery.for("update").limit(1));
|
|
const answerContext =
|
|
run && record(run.contextSnapshot).source === "issue.interaction.respond"
|
|
? await resolveExternalChatQuestionResponse(
|
|
tx,
|
|
binding,
|
|
run.contextSnapshot,
|
|
lockMode,
|
|
)
|
|
: null;
|
|
if (
|
|
run &&
|
|
record(run.contextSnapshot).source === "issue.interaction.respond" &&
|
|
record(run.contextSnapshot).paperclipExternalChatQuestionResponse &&
|
|
!answerContext
|
|
)
|
|
return "revoked";
|
|
const directContext = answerContext?.authorizationContext ?? run?.contextSnapshot;
|
|
const inheritedReviewWait = run && !externalChatWaitCandidate(directContext, binding)
|
|
? await readInheritedGitHubReviewWait(tx, run) : null;
|
|
if (inheritedReviewWait?.kind === "revoked") return "revoked";
|
|
const authorizationContext = inheritedReviewWait?.context ?? directContext;
|
|
const candidate = run ? externalChatWaitCandidate(authorizationContext, binding) : null;
|
|
if (!candidate) return "not_applicable";
|
|
|
|
const actorQuery = tx
|
|
.select({ status: agents.status })
|
|
.from(agents)
|
|
.where(
|
|
and(
|
|
eq(agents.id, binding.agentId),
|
|
eq(agents.companyId, binding.companyId),
|
|
),
|
|
);
|
|
const [actor] = await (lockMode === "read"
|
|
? actorQuery.limit(1)
|
|
: lockMode === "nonblocking"
|
|
? actorQuery.for("update", { noWait: true }).limit(1)
|
|
: actorQuery.for("update").limit(1));
|
|
const preserved = run && run.status !== "running"
|
|
? await readPreservedNativeResultFinalization(tx, binding) : null;
|
|
if (
|
|
!issue ||
|
|
!actor ||
|
|
(!preserved && run.status !== "running") ||
|
|
issue.assigneeAgentId !== binding.agentId ||
|
|
(!preserved && issue.executionRunId !== binding.runId) ||
|
|
["paused", "terminated", "pending_approval", "error"].includes(
|
|
actor.status,
|
|
)
|
|
) {
|
|
return "revoked";
|
|
}
|
|
|
|
const links = await tx
|
|
.select({
|
|
commentId: chatMessageLinks.commentId,
|
|
conversationId: chatConversations.id,
|
|
endpointId: chatEndpoints.id,
|
|
principalId: chatDeliveries.principalId,
|
|
})
|
|
.from(chatMessageLinks)
|
|
.innerJoin(
|
|
chatDeliveries,
|
|
and(
|
|
eq(chatDeliveries.id, chatMessageLinks.deliveryId),
|
|
eq(chatDeliveries.companyId, chatMessageLinks.companyId),
|
|
eq(chatDeliveries.endpointId, chatMessageLinks.endpointId),
|
|
eq(chatDeliveries.conversationId, chatMessageLinks.conversationId),
|
|
),
|
|
)
|
|
.innerJoin(
|
|
chatConversations,
|
|
and(
|
|
eq(chatConversations.id, chatMessageLinks.conversationId),
|
|
eq(chatConversations.companyId, chatMessageLinks.companyId),
|
|
eq(chatConversations.endpointId, chatMessageLinks.endpointId),
|
|
),
|
|
)
|
|
.innerJoin(
|
|
chatEndpoints,
|
|
and(
|
|
eq(chatEndpoints.id, chatConversations.endpointId),
|
|
eq(chatEndpoints.companyId, chatConversations.companyId),
|
|
),
|
|
)
|
|
.where(
|
|
and(
|
|
eq(chatMessageLinks.companyId, binding.companyId),
|
|
eq(chatMessageLinks.direction, "inbound"),
|
|
inArray(chatMessageLinks.commentId, candidate.commentIds),
|
|
eq(chatDeliveries.state, "processed"),
|
|
eq(chatConversations.issueId, binding.issueId),
|
|
eq(
|
|
chatEndpoints.provider,
|
|
candidate.provider as typeof chatEndpoints.$inferSelect.provider,
|
|
),
|
|
eq(chatEndpoints.assignedAgentId, binding.agentId),
|
|
),
|
|
);
|
|
const linkedCommentIds = new Set(links.map((row) => row.commentId));
|
|
if (
|
|
links.length === 0 ||
|
|
new Set(links.map((row) => row.conversationId)).size !== 1 ||
|
|
new Set(links.map((row) => row.endpointId)).size !== 1 ||
|
|
links.some((row) => !row.principalId) ||
|
|
!candidate.commentIds.every((id) => linkedCommentIds.has(id))
|
|
) {
|
|
return "revoked";
|
|
}
|
|
try {
|
|
await authorizeChatConversationForBoundRun(
|
|
tx,
|
|
binding,
|
|
authorizationContext,
|
|
lockMode,
|
|
);
|
|
return "authorized";
|
|
} catch (error) {
|
|
if (
|
|
error instanceof Error &&
|
|
[
|
|
"paperclip_runner_chat_attachment_binding_denied",
|
|
"paperclip_runner_chat_attachment_destination_denied",
|
|
"paperclip_runner_chat_attachment_principal_denied",
|
|
].includes(error.message)
|
|
) {
|
|
return "revoked";
|
|
}
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
export async function resolveExternalChatResponseWaitAuthorization(input: {
|
|
db: Db;
|
|
binding: ChatReuseBinding;
|
|
}): Promise<ExternalChatResponseWaitAuthorization> {
|
|
return input.db.transaction((tx) =>
|
|
resolveExternalChatResponseWaitAuthorizationInTransaction(
|
|
tx as unknown as Db,
|
|
input.binding,
|
|
"read",
|
|
),
|
|
);
|
|
}
|
|
|
|
const publishedGitHubReviewWaitKey = "github-review-response:";
|
|
|
|
export function isPublishedGitHubReviewCompletionContinuation(value: unknown): boolean {
|
|
const continuation = record(value);
|
|
return continuation.kind === "response_wake" && typeof continuation.idempotencyKey === "string" &&
|
|
continuation.idempotencyKey.startsWith(publishedGitHubReviewWaitKey);
|
|
}
|
|
|
|
/** Project a wait from the same saved publication and current authority used
|
|
* by finalization. Inherited waits need an accepted tree resume or a confirmed
|
|
* reply from this run to its exact current source. An arbitrary old review does
|
|
* not assign an unanswered new turn a wait. */
|
|
export async function readPublishedGitHubReviewCompletionContinuationInTransaction(
|
|
tx: Db, binding: ChatReuseBinding, lockMode: AuthorizationLockMode = "read",
|
|
) {
|
|
const [run] = await tx.select(responseWaitRun).from(heartbeatRuns).where(and(
|
|
eq(heartbeatRuns.id, binding.runId), eq(heartbeatRuns.companyId, binding.companyId),
|
|
eq(heartbeatRuns.agentId, binding.agentId), eq(heartbeatRuns.nativeIssueId, binding.issueId),
|
|
eq(heartbeatRuns.runtimeMode, "native"),
|
|
)).limit(1);
|
|
if (!run || run.status !== "running" && !await readPreservedNativeResultFinalization(tx, binding)) return null;
|
|
const wait = await readInheritedGitHubReviewWait(tx, run);
|
|
if (wait?.kind !== "published") return null;
|
|
if (wait.reviewRunId !== run.id) {
|
|
const [identity] = await tx.select({ cause: runIdentityContexts.cause, parentContextId: runIdentityContexts.parentContextId })
|
|
.from(runIdentityContexts).where(and(eq(runIdentityContexts.id, run.activeIdentityContextId!),
|
|
eq(runIdentityContexts.companyId, run.companyId), eq(runIdentityContexts.runId, run.id),
|
|
eq(runIdentityContexts.status, "accepted"))).limit(1);
|
|
const resumed = run.contextSnapshot?.source === "issue.tree_resume" && identity?.cause === "issue_tree_resumed" &&
|
|
Boolean(identity.parentContextId);
|
|
if (!resumed && !(run.contextSnapshot?.source === "chat:github" &&
|
|
await githubRunReplyState(tx, { ...binding, endpointId: wait.endpointId,
|
|
sourceCommentId: wait.context.wakeCommentId, requireSettled: true }) === "confirmed")) return null;
|
|
}
|
|
if (await resolveExternalChatResponseWaitAuthorizationInTransaction(tx, binding, lockMode) !== "authorized") return null;
|
|
return { kind: "response_wake" as const, summary: "Wait for the next authorized GitHub response to the published review.",
|
|
idempotencyKey: `${publishedGitHubReviewWaitKey}${run.id}:${wait.reviewId}:${wait.headSha}` };
|
|
}
|
|
|
|
export async function readPublishedGitHubReviewCompletionContinuation(input: { db: Db; binding: ChatReuseBinding }) {
|
|
return input.db.transaction(tx => readPublishedGitHubReviewCompletionContinuationInTransaction(tx as unknown as Db, input.binding));
|
|
}
|
|
|
|
async function sourceLineageExists(
|
|
tx: Db,
|
|
binding: ChatReuseBinding,
|
|
conversation: AuthorizedConversation,
|
|
source: Pick<
|
|
ChatAttachmentReuseSource,
|
|
"sourceCommentId" | "attachmentId"
|
|
> & {
|
|
parentCommentId: string;
|
|
},
|
|
): Promise<boolean> {
|
|
const [inbound] = await tx
|
|
.select({ providerEventId: chatDeliveries.providerEventId })
|
|
.from(chatMessageLinks)
|
|
.innerJoin(
|
|
chatDeliveries,
|
|
and(
|
|
eq(chatDeliveries.id, chatMessageLinks.deliveryId),
|
|
eq(chatDeliveries.companyId, chatMessageLinks.companyId),
|
|
eq(chatDeliveries.endpointId, chatMessageLinks.endpointId),
|
|
eq(chatDeliveries.conversationId, chatMessageLinks.conversationId),
|
|
),
|
|
)
|
|
.where(
|
|
and(
|
|
eq(chatMessageLinks.companyId, binding.companyId),
|
|
eq(chatMessageLinks.endpointId, conversation.endpointId),
|
|
eq(chatMessageLinks.conversationId, conversation.conversationId),
|
|
eq(chatMessageLinks.direction, "inbound"),
|
|
eq(chatMessageLinks.commentId, source.sourceCommentId),
|
|
eq(chatDeliveries.state, "processed"),
|
|
),
|
|
)
|
|
.limit(1);
|
|
if (inbound && source.parentCommentId === source.sourceCommentId) {
|
|
const [lifecycle] = await tx
|
|
.select({ id: chatDeliveries.id })
|
|
.from(chatDeliveries)
|
|
.where(
|
|
and(
|
|
eq(chatDeliveries.companyId, binding.companyId),
|
|
eq(chatDeliveries.endpointId, conversation.endpointId),
|
|
eq(chatDeliveries.conversationId, conversation.conversationId),
|
|
inArray(chatDeliveries.eventKind, [
|
|
"message_updated",
|
|
"message_deleted",
|
|
"message_restored",
|
|
]),
|
|
eq(chatDeliveries.state, "processed"),
|
|
sql`${chatDeliveries.normalizedEvent}->'message'->>'targetProviderEventId' = ${inbound.providerEventId}`,
|
|
),
|
|
)
|
|
.limit(1);
|
|
return !lifecycle;
|
|
}
|
|
const publications = await tx
|
|
.select({
|
|
payload: chatPublications.payload,
|
|
providerMessageId: chatPublications.providerMessageId,
|
|
publishedAt: chatPublications.publishedAt,
|
|
})
|
|
.from(chatPublications)
|
|
.where(
|
|
and(
|
|
eq(chatPublications.companyId, binding.companyId),
|
|
eq(chatPublications.endpointId, conversation.endpointId),
|
|
eq(chatPublications.conversationId, conversation.conversationId),
|
|
eq(chatPublications.issueId, binding.issueId),
|
|
eq(chatPublications.commentId, source.sourceCommentId),
|
|
eq(chatPublications.state, "published"),
|
|
),
|
|
);
|
|
const matchingPublications = publications.filter(
|
|
(row) =>
|
|
row.providerMessageId !== null &&
|
|
row.publishedAt !== null &&
|
|
publicationAttachmentIds(row.payload).includes(source.attachmentId),
|
|
);
|
|
if (matchingPublications.length === 0) return false;
|
|
for (const publication of matchingPublications) {
|
|
const [lifecycle] = await tx
|
|
.select({ id: chatDeliveries.id })
|
|
.from(chatDeliveries)
|
|
.where(
|
|
and(
|
|
eq(chatDeliveries.companyId, binding.companyId),
|
|
eq(chatDeliveries.endpointId, conversation.endpointId),
|
|
eq(chatDeliveries.conversationId, conversation.conversationId),
|
|
inArray(chatDeliveries.eventKind, [
|
|
"message_updated",
|
|
"message_deleted",
|
|
"message_restored",
|
|
]),
|
|
eq(chatDeliveries.state, "processed"),
|
|
eq(
|
|
sql<string>`${chatDeliveries.normalizedEvent}->'message'->>'providerMessageId'`,
|
|
publication.providerMessageId!,
|
|
),
|
|
),
|
|
)
|
|
.limit(1);
|
|
if (!lifecycle) return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
const HISTORICAL_ATTACHMENT_FILE_EXTENSIONS: ReadonlyMap<string, string> =
|
|
new Map([
|
|
["image/png", "png"],
|
|
["image/jpeg", "jpg"],
|
|
["image/jpg", "jpg"],
|
|
["image/webp", "webp"],
|
|
["image/gif", "gif"],
|
|
["audio/mpeg", "mp3"],
|
|
["audio/mp4", "m4a"],
|
|
["audio/ogg", "ogg"],
|
|
["audio/wav", "wav"],
|
|
["audio/webm", "webm"],
|
|
["video/mp4", "mp4"],
|
|
["video/webm", "webm"],
|
|
["video/quicktime", "mov"],
|
|
["video/x-m4v", "m4v"],
|
|
["application/pdf", "pdf"],
|
|
["application/zip", "zip"],
|
|
["application/json", "json"],
|
|
["text/plain", "txt"],
|
|
["text/markdown", "md"],
|
|
["text/csv", "csv"],
|
|
["text/html", "html"],
|
|
]);
|
|
|
|
async function loadSource(
|
|
tx: Db,
|
|
binding: ChatReuseBinding,
|
|
conversation: AuthorizedConversation,
|
|
sourceCommentId: string,
|
|
attachmentId: string,
|
|
allowEmpty = false,
|
|
): Promise<ChatAttachmentReuseSource> {
|
|
const [sourceComment] = await tx
|
|
.select({ id: issueComments.id })
|
|
.from(issueComments)
|
|
.where(
|
|
and(
|
|
eq(issueComments.id, sourceCommentId),
|
|
eq(issueComments.companyId, binding.companyId),
|
|
eq(issueComments.issueId, binding.issueId),
|
|
isNull(issueComments.deletedAt),
|
|
),
|
|
)
|
|
.for("update")
|
|
.limit(1);
|
|
if (!sourceComment) {
|
|
throw new Error("paperclip_runner_chat_attachment_source_denied");
|
|
}
|
|
const [row] = await tx
|
|
.select({
|
|
attachmentId: issueAttachments.id,
|
|
parentCommentId: issueComments.id,
|
|
filename: assets.originalFilename,
|
|
contentType: assets.contentType,
|
|
byteSize: assets.byteSize,
|
|
sha256: assets.sha256,
|
|
objectKey: assets.objectKey,
|
|
createdAt: issueAttachments.createdAt,
|
|
})
|
|
.from(issueAttachments)
|
|
.innerJoin(
|
|
assets,
|
|
and(
|
|
eq(assets.id, issueAttachments.assetId),
|
|
eq(assets.companyId, binding.companyId),
|
|
),
|
|
)
|
|
.innerJoin(
|
|
issueComments,
|
|
and(
|
|
eq(issueComments.id, issueAttachments.issueCommentId),
|
|
eq(issueComments.companyId, binding.companyId),
|
|
eq(issueComments.issueId, binding.issueId),
|
|
isNull(issueComments.deletedAt),
|
|
),
|
|
)
|
|
.where(
|
|
and(
|
|
eq(issueAttachments.id, attachmentId),
|
|
eq(issueAttachments.companyId, binding.companyId),
|
|
eq(issueAttachments.issueId, binding.issueId),
|
|
),
|
|
)
|
|
.for("update")
|
|
.limit(1);
|
|
if (
|
|
!row ||
|
|
row.byteSize < (allowEmpty ? 0 : 1) ||
|
|
row.byteSize > MAX_ATTACHMENT_BYTES ||
|
|
!isAllowedContentType(normalizeContentType(row.contentType)) ||
|
|
!/^[a-f0-9]{64}$/iu.test(row.sha256)
|
|
) {
|
|
throw new Error("paperclip_runner_chat_attachment_source_denied");
|
|
}
|
|
const sourceWithParent = {
|
|
sourceCommentId,
|
|
attachmentId: row.attachmentId,
|
|
parentCommentId: row.parentCommentId,
|
|
// Provider photos/voice messages legitimately have no original filename.
|
|
// Derive display metadata from the already-validated MIME and opaque id;
|
|
// never infer a storage path or change the selected source bytes.
|
|
filename: row.filename?.trim()
|
|
? row.filename
|
|
: `attachment-${row.attachmentId}.${HISTORICAL_ATTACHMENT_FILE_EXTENSIONS.get(normalizeContentType(row.contentType)) ?? "bin"}`,
|
|
contentType: normalizeContentType(row.contentType),
|
|
byteSize: row.byteSize,
|
|
sha256: row.sha256,
|
|
objectKey: row.objectKey,
|
|
createdAt: row.createdAt,
|
|
};
|
|
if (
|
|
!(await sourceLineageExists(tx, binding, conversation, sourceWithParent))
|
|
) {
|
|
throw new Error("paperclip_runner_chat_attachment_source_denied");
|
|
}
|
|
const { parentCommentId: _parentCommentId, ...source } = sourceWithParent;
|
|
return source;
|
|
}
|
|
|
|
export async function listAuthorizedChatAttachments(input: {
|
|
db: Db;
|
|
binding: ChatReuseBinding;
|
|
sourceCommentId?: string | null;
|
|
cursor?: string | null;
|
|
limit: number;
|
|
}): Promise<ListedChatAttachmentPage> {
|
|
return input.db.transaction(async (tx) => {
|
|
const [current] = await tx
|
|
.select({ run: heartbeatRuns, actorStatus: agents.status })
|
|
.from(heartbeatRuns)
|
|
.innerJoin(
|
|
issues,
|
|
and(
|
|
eq(issues.id, input.binding.issueId),
|
|
eq(issues.companyId, heartbeatRuns.companyId),
|
|
),
|
|
)
|
|
.innerJoin(
|
|
agents,
|
|
and(
|
|
eq(agents.id, input.binding.agentId),
|
|
eq(agents.companyId, heartbeatRuns.companyId),
|
|
),
|
|
)
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.id, input.binding.runId),
|
|
eq(heartbeatRuns.companyId, input.binding.companyId),
|
|
eq(heartbeatRuns.agentId, input.binding.agentId),
|
|
eq(heartbeatRuns.nativeIssueId, input.binding.issueId),
|
|
eq(heartbeatRuns.runtimeMode, "native"),
|
|
eq(heartbeatRuns.status, "running"),
|
|
eq(issues.assigneeAgentId, input.binding.agentId),
|
|
eq(issues.executionRunId, input.binding.runId),
|
|
),
|
|
)
|
|
.for("update")
|
|
.limit(1);
|
|
if (
|
|
!current ||
|
|
["paused", "terminated", "pending_approval", "error"].includes(
|
|
current.actorStatus,
|
|
)
|
|
) {
|
|
throw new Error("paperclip_runner_tool_binding_not_authorized");
|
|
}
|
|
const conversation = await authorizeChatConversationForBoundRun(
|
|
tx as unknown as Db,
|
|
input.binding,
|
|
current.run.contextSnapshot,
|
|
);
|
|
const sourceFilter = input.sourceCommentId ?? null;
|
|
const cursor = decodeListCursor(
|
|
input.cursor,
|
|
conversation.conversationId,
|
|
sourceFilter,
|
|
);
|
|
const rawCandidates = await tx.execute(sql<{
|
|
attachment_id: string;
|
|
source_comment_id: string;
|
|
created_at: Date | string;
|
|
}>`
|
|
with candidate_pairs as (
|
|
select
|
|
attachment.id as attachment_id,
|
|
link.comment_id as source_comment_id,
|
|
attachment.created_at
|
|
from issue_attachments attachment
|
|
join issue_comments parent_comment
|
|
on parent_comment.id = attachment.issue_comment_id
|
|
and parent_comment.company_id = ${input.binding.companyId}::uuid
|
|
and parent_comment.issue_id = ${input.binding.issueId}::uuid
|
|
and parent_comment.deleted_at is null
|
|
join chat_message_links link
|
|
on link.company_id = ${input.binding.companyId}::uuid
|
|
and link.endpoint_id = ${conversation.endpointId}::uuid
|
|
and link.conversation_id = ${conversation.conversationId}::uuid
|
|
and link.direction = 'inbound'
|
|
and link.comment_id = attachment.issue_comment_id
|
|
join chat_deliveries delivery
|
|
on delivery.id = link.delivery_id
|
|
and delivery.company_id = link.company_id
|
|
and delivery.endpoint_id = link.endpoint_id
|
|
and delivery.conversation_id = link.conversation_id
|
|
and delivery.state = 'processed'
|
|
where attachment.company_id = ${input.binding.companyId}::uuid
|
|
and attachment.issue_id = ${input.binding.issueId}::uuid
|
|
and (${sourceFilter}::uuid is null or link.comment_id = ${sourceFilter}::uuid)
|
|
and not exists (
|
|
select 1
|
|
from chat_deliveries lifecycle
|
|
where lifecycle.company_id = ${input.binding.companyId}::uuid
|
|
and lifecycle.endpoint_id = ${conversation.endpointId}::uuid
|
|
and lifecycle.conversation_id = ${conversation.conversationId}::uuid
|
|
and lifecycle.state = 'processed'
|
|
and lifecycle.event_kind in ('message_updated', 'message_deleted', 'message_restored')
|
|
and lifecycle.normalized_event->'message'->>'targetProviderEventId' = delivery.provider_event_id
|
|
)
|
|
|
|
union all
|
|
|
|
select
|
|
attachment.id as attachment_id,
|
|
publication.comment_id as source_comment_id,
|
|
attachment.created_at
|
|
from issue_attachments attachment
|
|
join issue_comments parent_comment
|
|
on parent_comment.id = attachment.issue_comment_id
|
|
and parent_comment.company_id = ${input.binding.companyId}::uuid
|
|
and parent_comment.issue_id = ${input.binding.issueId}::uuid
|
|
and parent_comment.deleted_at is null
|
|
join chat_publications publication
|
|
on publication.company_id = ${input.binding.companyId}::uuid
|
|
and publication.endpoint_id = ${conversation.endpointId}::uuid
|
|
and publication.conversation_id = ${conversation.conversationId}::uuid
|
|
and publication.issue_id = ${input.binding.issueId}::uuid
|
|
and publication.state = 'published'
|
|
and publication.published_at is not null
|
|
and publication.provider_message_id is not null
|
|
and publication.comment_id is not null
|
|
and publication.payload->'attachmentIds' ? attachment.id::text
|
|
join issue_comments source_comment
|
|
on source_comment.id = publication.comment_id
|
|
and source_comment.company_id = publication.company_id
|
|
and source_comment.issue_id = publication.issue_id
|
|
and source_comment.deleted_at is null
|
|
where attachment.company_id = ${input.binding.companyId}::uuid
|
|
and attachment.issue_id = ${input.binding.issueId}::uuid
|
|
and (${sourceFilter}::uuid is null or publication.comment_id = ${sourceFilter}::uuid)
|
|
and not exists (
|
|
select 1
|
|
from chat_deliveries lifecycle
|
|
where lifecycle.company_id = ${input.binding.companyId}::uuid
|
|
and lifecycle.endpoint_id = ${conversation.endpointId}::uuid
|
|
and lifecycle.conversation_id = ${conversation.conversationId}::uuid
|
|
and lifecycle.state = 'processed'
|
|
and lifecycle.event_kind in ('message_updated', 'message_deleted', 'message_restored')
|
|
and lifecycle.normalized_event->'message'->>'providerMessageId' = publication.provider_message_id
|
|
)
|
|
), candidates as (
|
|
select distinct on (attachment_id)
|
|
attachment_id,
|
|
source_comment_id,
|
|
created_at
|
|
from candidate_pairs
|
|
order by attachment_id, created_at desc, source_comment_id desc
|
|
)
|
|
select attachment_id, source_comment_id, created_at
|
|
from candidates
|
|
where (
|
|
${cursor?.createdAt ?? null}::timestamptz is null
|
|
or (created_at, attachment_id, source_comment_id) <
|
|
(${cursor?.createdAt ?? null}::timestamptz, ${cursor?.attachmentId ?? null}::uuid, ${cursor?.sourceCommentIdTieBreak ?? null}::uuid)
|
|
)
|
|
order by created_at desc, attachment_id desc, source_comment_id desc
|
|
limit ${input.limit + 1}
|
|
`);
|
|
const candidates = (
|
|
Array.from(rawCandidates) as Array<{
|
|
attachment_id: string;
|
|
source_comment_id: string;
|
|
created_at: Date | string;
|
|
}>
|
|
).map((row) => ({
|
|
sourceCommentId: row.source_comment_id,
|
|
attachmentId: row.attachment_id,
|
|
createdAt:
|
|
row.created_at instanceof Date
|
|
? row.created_at
|
|
: new Date(row.created_at),
|
|
}));
|
|
const listed: ListedChatAttachment[] = [];
|
|
const scanned = candidates.slice(0, input.limit);
|
|
for (const candidate of scanned) {
|
|
if (listed.length >= input.limit) break;
|
|
try {
|
|
const source = await loadSource(
|
|
tx as unknown as Db,
|
|
input.binding,
|
|
conversation,
|
|
candidate.sourceCommentId,
|
|
candidate.attachmentId,
|
|
true,
|
|
);
|
|
listed.push({
|
|
sourceCommentId: source.sourceCommentId,
|
|
attachmentId: source.attachmentId,
|
|
filename: source.filename,
|
|
contentType: source.contentType,
|
|
byteSize: source.byteSize,
|
|
sha256: source.sha256,
|
|
createdAt: source.createdAt.toISOString(),
|
|
contentAccess: "metadata_only",
|
|
});
|
|
} catch (error) {
|
|
// Metadata listing omits deleted, oversized, or no-longer-lineaged files.
|
|
if (
|
|
!(error instanceof Error) ||
|
|
error.message !== "paperclip_runner_chat_attachment_source_denied"
|
|
) {
|
|
throw error;
|
|
}
|
|
}
|
|
}
|
|
const lastScanned = scanned.at(-1) ?? null;
|
|
const hasMore = candidates.length > input.limit;
|
|
return {
|
|
attachments: listed,
|
|
nextCursor:
|
|
hasMore && lastScanned
|
|
? encodeListCursor({
|
|
schema: "paperclip.chat-attachment-list-cursor.v1",
|
|
conversationId: conversation.conversationId,
|
|
sourceCommentId: sourceFilter,
|
|
createdAt: lastScanned.createdAt.toISOString(),
|
|
attachmentId: lastScanned.attachmentId,
|
|
sourceCommentIdTieBreak: lastScanned.sourceCommentId,
|
|
})
|
|
: null,
|
|
complete: !hasMore,
|
|
};
|
|
});
|
|
}
|
|
|
|
export async function authorizeChatAttachmentReuse(input: {
|
|
db: Db;
|
|
binding: ChatReuseBinding;
|
|
contextSnapshot: unknown;
|
|
sourceCommentId: string;
|
|
attachmentId: string;
|
|
/** Inspection may open empty files; publication retains its nonempty bound. */
|
|
allowEmpty?: boolean;
|
|
}): Promise<ChatAttachmentReuseSource> {
|
|
const conversation = await authorizeChatConversationForBoundRun(
|
|
input.db,
|
|
input.binding,
|
|
input.contextSnapshot,
|
|
);
|
|
return loadSource(
|
|
input.db,
|
|
input.binding,
|
|
conversation,
|
|
input.sourceCommentId,
|
|
input.attachmentId,
|
|
input.allowEmpty,
|
|
);
|
|
}
|
|
|
|
async function readSourceBytes(
|
|
storage: StorageService,
|
|
companyId: string,
|
|
source: ChatAttachmentReuseSource,
|
|
timeoutMs: number,
|
|
): Promise<Buffer> {
|
|
let acquisitionTimedOut = false;
|
|
let rejectAcquisition!: (error: Error) => void;
|
|
const acquisitionTimeout = new Promise<never>((_resolve, reject) => {
|
|
rejectAcquisition = reject;
|
|
});
|
|
const acquisitionTimer = setTimeout(() => {
|
|
acquisitionTimedOut = true;
|
|
rejectAcquisition(
|
|
new Error("paperclip_runner_chat_attachment_source_read_timed_out"),
|
|
);
|
|
}, timeoutMs);
|
|
acquisitionTimer.unref?.();
|
|
const objectPromise = storage
|
|
.getObject(companyId, source.objectKey)
|
|
.then((object) => {
|
|
if (acquisitionTimedOut) object.stream.destroy();
|
|
return object;
|
|
});
|
|
let object: Awaited<ReturnType<StorageService["getObject"]>>;
|
|
try {
|
|
object = await Promise.race([objectPromise, acquisitionTimeout]);
|
|
} finally {
|
|
clearTimeout(acquisitionTimer);
|
|
}
|
|
const timeout = setTimeout(() => {
|
|
object.stream.destroy(
|
|
new Error("paperclip_runner_chat_attachment_source_read_timed_out"),
|
|
);
|
|
}, timeoutMs);
|
|
timeout.unref?.();
|
|
const chunks: Buffer[] = [];
|
|
let total = 0;
|
|
try {
|
|
for await (const chunk of object.stream) {
|
|
const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
|
|
total += buffer.length;
|
|
if (total > source.byteSize || total > MAX_ATTACHMENT_BYTES) {
|
|
object.stream.destroy();
|
|
throw new Error(
|
|
"paperclip_runner_chat_attachment_source_size_mismatch",
|
|
);
|
|
}
|
|
chunks.push(buffer);
|
|
}
|
|
} finally {
|
|
clearTimeout(timeout);
|
|
}
|
|
const body = Buffer.concat(chunks);
|
|
if (
|
|
body.length !== source.byteSize ||
|
|
createHash("sha256").update(body).digest("hex") !==
|
|
source.sha256.toLowerCase()
|
|
) {
|
|
if (!object.stream.destroyed) {
|
|
object.stream.destroy();
|
|
}
|
|
throw new Error(
|
|
"paperclip_runner_chat_attachment_source_integrity_mismatch",
|
|
);
|
|
}
|
|
return body;
|
|
}
|
|
|
|
const DEFAULT_STORAGE_TIMEOUT_MS = 10_000;
|
|
|
|
async function deleteStorageObjectWithin(
|
|
storage: StorageService,
|
|
companyId: string,
|
|
objectKey: string,
|
|
timeoutMs: number,
|
|
): Promise<void> {
|
|
const deletion = storage
|
|
.deleteObject(companyId, objectKey)
|
|
.catch(() => undefined);
|
|
let timer: ReturnType<typeof setTimeout> | null = null;
|
|
try {
|
|
await Promise.race([
|
|
deletion,
|
|
new Promise<void>((resolve) => {
|
|
timer = setTimeout(resolve, timeoutMs);
|
|
timer.unref?.();
|
|
}),
|
|
]);
|
|
} finally {
|
|
if (timer) clearTimeout(timer);
|
|
}
|
|
}
|
|
|
|
async function putStorageObjectWithin(
|
|
storage: StorageService,
|
|
input: Parameters<StorageService["putFile"]>[0],
|
|
timeoutMs: number,
|
|
): Promise<Awaited<ReturnType<StorageService["putFile"]>>> {
|
|
let timedOut = false;
|
|
let rejectTimeout!: (error: Error) => void;
|
|
const timeout = new Promise<never>((_resolve, reject) => {
|
|
rejectTimeout = reject;
|
|
});
|
|
const timer = setTimeout(() => {
|
|
timedOut = true;
|
|
rejectTimeout(
|
|
new Error("paperclip_runner_chat_attachment_storage_write_timed_out"),
|
|
);
|
|
}, timeoutMs);
|
|
timer.unref?.();
|
|
const write = storage.putFile(input).then((stored) => {
|
|
if (timedOut) {
|
|
void deleteStorageObjectWithin(
|
|
storage,
|
|
input.companyId,
|
|
stored.objectKey,
|
|
timeoutMs,
|
|
);
|
|
}
|
|
return stored;
|
|
});
|
|
try {
|
|
return await Promise.race([write, timeout]);
|
|
} finally {
|
|
clearTimeout(timer);
|
|
}
|
|
}
|
|
|
|
export async function prepareReusedChatAttachment(input: {
|
|
db: Db;
|
|
binding: ChatReuseBinding;
|
|
source: ChatAttachmentReuseSource;
|
|
title: string;
|
|
storage?: StorageService;
|
|
storageTimeoutMs?: number;
|
|
}): Promise<PreparedReusedChatAttachment> {
|
|
const [issue] = await input.db
|
|
.select({ statusVersion: issues.statusVersion })
|
|
.from(issues)
|
|
.innerJoin(
|
|
heartbeatRuns,
|
|
and(
|
|
eq(heartbeatRuns.id, input.binding.runId),
|
|
eq(heartbeatRuns.companyId, issues.companyId),
|
|
eq(heartbeatRuns.nativeIssueId, issues.id),
|
|
),
|
|
)
|
|
.where(
|
|
and(
|
|
eq(issues.id, input.binding.issueId),
|
|
eq(issues.companyId, input.binding.companyId),
|
|
eq(issues.assigneeAgentId, input.binding.agentId),
|
|
eq(issues.executionRunId, input.binding.runId),
|
|
eq(heartbeatRuns.agentId, input.binding.agentId),
|
|
eq(heartbeatRuns.status, "running"),
|
|
eq(heartbeatRuns.runtimeMode, "native"),
|
|
),
|
|
)
|
|
.limit(1);
|
|
if (!issue)
|
|
throw new Error("paperclip_runner_chat_attachment_binding_denied");
|
|
const storage = input.storage ?? getStorageService();
|
|
const storageTimeoutMs =
|
|
typeof input.storageTimeoutMs === "number" &&
|
|
Number.isFinite(input.storageTimeoutMs) &&
|
|
input.storageTimeoutMs > 0
|
|
? input.storageTimeoutMs
|
|
: DEFAULT_STORAGE_TIMEOUT_MS;
|
|
const body = await readSourceBytes(
|
|
storage,
|
|
input.binding.companyId,
|
|
input.source,
|
|
storageTimeoutMs,
|
|
);
|
|
const stored = await putStorageObjectWithin(
|
|
storage,
|
|
{
|
|
companyId: input.binding.companyId,
|
|
namespace: `issues/${input.binding.issueId}`,
|
|
originalFilename: input.source.filename,
|
|
contentType: input.source.contentType,
|
|
body,
|
|
},
|
|
storageTimeoutMs,
|
|
);
|
|
try {
|
|
if (
|
|
stored.byteSize !== body.length ||
|
|
stored.sha256.toLowerCase() !== input.source.sha256.toLowerCase() ||
|
|
stored.contentType !== input.source.contentType
|
|
) {
|
|
throw new Error("paperclip_runner_chat_attachment_storage_mismatch");
|
|
}
|
|
const attachment = await issueService(input.db).createAttachment({
|
|
issueId: input.binding.issueId,
|
|
provider: stored.provider,
|
|
objectKey: stored.objectKey,
|
|
contentType: stored.contentType,
|
|
byteSize: stored.byteSize,
|
|
sha256: stored.sha256,
|
|
originalFilename: stored.originalFilename,
|
|
createdByAgentId: input.binding.agentId,
|
|
createdByRunId: input.binding.runId,
|
|
});
|
|
if (
|
|
!attachment.artifactWorkProductId ||
|
|
attachment.originatingRunId !== input.binding.runId
|
|
) {
|
|
throw new Error("paperclip_runner_chat_attachment_origin_not_persisted");
|
|
}
|
|
const [workProduct] = await input.db
|
|
.select({ metadata: issueWorkProducts.metadata })
|
|
.from(issueWorkProducts)
|
|
.where(
|
|
and(
|
|
eq(issueWorkProducts.id, attachment.artifactWorkProductId),
|
|
eq(issueWorkProducts.companyId, input.binding.companyId),
|
|
eq(issueWorkProducts.issueId, input.binding.issueId),
|
|
eq(issueWorkProducts.createdByRunId, input.binding.runId),
|
|
),
|
|
)
|
|
.for("update")
|
|
.limit(1);
|
|
if (!workProduct) {
|
|
throw new Error("paperclip_runner_chat_attachment_origin_not_persisted");
|
|
}
|
|
await input.db.transaction(async tx => {
|
|
await notifyIssueRecovery(tx, { companyId: input.binding.companyId });
|
|
await tx
|
|
.update(issueWorkProducts)
|
|
.set({
|
|
title: input.title,
|
|
metadata: {
|
|
...record(workProduct.metadata),
|
|
reusedFromAttachmentId: input.source.attachmentId,
|
|
reusedFromCommentId: input.source.sourceCommentId,
|
|
reusedFromSha256: input.source.sha256,
|
|
},
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(
|
|
and(
|
|
eq(issueWorkProducts.id, attachment.artifactWorkProductId),
|
|
eq(issueWorkProducts.companyId, input.binding.companyId),
|
|
eq(issueWorkProducts.issueId, input.binding.issueId),
|
|
eq(issueWorkProducts.createdByRunId, input.binding.runId),
|
|
),
|
|
);
|
|
});
|
|
const comment = await issueService(input.db).addComment(
|
|
input.binding.issueId,
|
|
`Prepared ${input.title} for this response.`,
|
|
{ agentId: input.binding.agentId, runId: input.binding.runId },
|
|
{
|
|
attachmentIds: [attachment.id],
|
|
authorizationReason: "paperclip_runner_protocol",
|
|
},
|
|
input.db,
|
|
);
|
|
return {
|
|
result: {
|
|
commandId: `chat-attachment-reused:${attachment.id}`,
|
|
disposition: "applied",
|
|
stateRevision: issue.statusVersion,
|
|
entityRefs: [
|
|
attachment.id,
|
|
attachment.artifactWorkProductId,
|
|
comment.id,
|
|
],
|
|
scheduledWakeIds: [],
|
|
source: {
|
|
commentId: input.source.sourceCommentId,
|
|
attachmentId: input.source.attachmentId,
|
|
sha256: input.source.sha256,
|
|
},
|
|
prepared: {
|
|
attachmentId: attachment.id,
|
|
workProductId: attachment.artifactWorkProductId,
|
|
commentId: comment.id,
|
|
filename: input.source.filename,
|
|
byteSize: body.length,
|
|
sha256: attachment.sha256,
|
|
},
|
|
},
|
|
rollbackDefinitePreCommitFailure: async () => {
|
|
await deleteStorageObjectWithin(
|
|
storage,
|
|
input.binding.companyId,
|
|
stored.objectKey,
|
|
storageTimeoutMs,
|
|
);
|
|
},
|
|
};
|
|
} catch (error) {
|
|
await deleteStorageObjectWithin(
|
|
storage,
|
|
input.binding.companyId,
|
|
stored.objectKey,
|
|
storageTimeoutMs,
|
|
);
|
|
throw error;
|
|
}
|
|
}
|