perf(slack): remove delays before mention acknowledgements (#16491)

Slack mentions now start eyes reactions after the authorized task/comment commits, independently of attachments, subscription and scheduler acceptance. Omit unused optional URL-preview polling while retaining message text, links and files. Keep receipt failures and missing-file warnings independent, including terminal cleanup of older multi-line diagnostics.

Verified with full current-head CI and E2E, Superagent and legacy security checks, Apex 5/5 with no unresolved findings, 19 selected real-PostgreSQL receipt/authorization/cleanup cases, and full frozen-source typecheck/build. The first CI attempt's unchanged Cursor probe timeout is retained in the PR; the one failed-only retry passed under unchanged deadlines. Live Slack latency still requires deployed acceptance.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-10-15 17:36:50 -05:00
1 parent 81cea89d6b
commit becdc3b08e
14 files changed
+458 -37

No files matched your search

@@ -1,4 +1,5 @@
import { idleWorkSnapshot } from "../services/task-admission.js";
import { stageReceiptReactionRemovals } from "../services/chat-receipt-reactions.js";
import type { Db } from "@paperclipai/db";
import * as activityLogModule from "../services/activity-log.js";
import * as connectionCredentialBindings from "../services/connection-credential-bindings.js";
@@ -1758,6 +1759,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
| "questionFormOpenAuthorizationBarrier"
| "questionResolutionPersistBarrier"
| "reactionReplayEndpointLockBarrier"
| "receiptReactionTransportBarrier"
| "reachAuthorizationBarrier"
| "resolveNativeQuestion"
| "renewCredentialMutationLease"
@@ -28007,6 +28009,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
const fixture = await seedCompany();
const statusCalls: string[] = [];
const ordinaryFetch = fakeSlackFetch();
let releaseReceipt!: () => void;
const receiptGate = new Promise<void>((resolve) => { releaseReceipt = resolve; });
let earlyFinal:
| ((
agentId: string,
@@ -28017,6 +28021,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
| undefined;
const context = await configuredSlackEndpoint(fixture, {
scheduleDeferredWork: () => undefined,
...(mode === "final_before_add" ? { receiptReactionTransportBarrier: () => receiptGate } : {}),
fetch: async (input, init) => {
if (
String(input) === "https://slack.com/api/agents.sessions.setStatus"
@@ -28089,8 +28094,9 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
eq(chatActions.kind, "receipt_reaction"),
),
),
).resolves.toHaveLength(0);
).resolves.toHaveLength(1);
await publishFinal();
releaseReceipt();
return { runId };
};
try {
@@ -28210,6 +28216,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
).toHaveLength(1);
expect(wakeup).toHaveBeenCalledTimes(mode === "newer_followup" ? 2 : 1);
} finally {
releaseReceipt();
// Retire the synthetic still-running retry's conversation so later
// globally-scanned milestone fixtures cannot inherit this test's work.
if (mode === "same_source_retry")
@@ -28420,6 +28427,152 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
}
});
it.each(["attachment", "subscription", "wakeup"] as const)("acknowledges a Slack mention while %s is blocked", async (phase) => {
const fixture = await seedCompany();
let release!: () => void;
const gate = new Promise<void>((resolve) => { release = resolve; });
const entered = vi.fn();
const blocked = async () => { entered(); await gate; return { accepted: true }; };
const { callbacks, endpoint, runtime, service, wakeup } = await configuredSlackEndpoint(fixture, {
...(phase === "wakeup" ? { wakeup: blocked } : {}),
...(phase === "attachment" ? { storage: createStorageService().storage } : {}),
});
const thread = makeThread({ channelId: "CFAST", id: "slack:CFAST:1740000088.1" });
if (phase === "subscription") thread.thread.subscribe = async () => { await blocked(); };
const message = makeMessage({ id: "1740000088.1", text: "@maya acknowledge immediately", mentioned: true,
...(phase === "attachment" ? { attachments: [{ type: "file", name: "receipt.txt", mimeType: "text/plain", fetchData: async () => { await blocked(); return Buffer.from("synthetic attachment"); } } as Attachment] } : {}),
});
const startedAt = performance.now();
const inbound = deliverMessage({ callbacks, endpointId: endpoint.id, thread: thread.thread, message, trigger: "mention" });
try {
await vi.waitFor(() => expect(entered).toHaveBeenCalled(), { timeout: 3_000, interval: 10 });
await vi.waitFor(() => expect(runtime.endpoints.get(endpoint.id)!.reactions).toHaveLength(1), { timeout: 1_000, interval: 10 });
const [delivery] = await readChatDeliveriesByEndpointId(endpoint.id);
await vi.waitFor(async () => {
const [receipt] = await readChatActionsByProviderActionId(`receipt_reaction:${delivery.id}`);
expect(receipt.status).toBe("processed");
}, { timeout: 1_000, interval: 10 });
const [receipt] = await readChatActionsByProviderActionId(`receipt_reaction:${delivery.id}`);
expect(await readIssuesByCompanyId(fixture.companyId)).toHaveLength(1);
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, fixture.companyId))).toHaveLength(0);
process.stdout.write(`Slack ${phase} blocked: eyes confirmed in ${(performance.now() - startedAt).toFixed(1)}ms; durable receipt to confirmed action ${receipt.updatedAt.getTime() - delivery.receivedAt.getTime()}ms (synthetic Slack, real PostgreSQL)\n`);
} finally {
release();
await inbound;
}
await deliverMessage({ callbacks, endpointId: endpoint.id, thread: thread.thread, message, trigger: "mention" });
expect(runtime.endpoints.get(endpoint.id)!.reactions).toHaveLength(1);
expect(wakeup).toHaveBeenCalledTimes(1);
await service.shutdown();
});
it.each(["attachment", "subscription"] as const)("preserves an early Slack receipt failure after %s acceptance", async phase => {
const fixture = await seedCompany();
let release!: () => void;
const gate = new Promise<void>(resolve => { release = resolve; });
const entered = vi.fn();
const block = async () => { entered(); await gate; };
const { callbacks, endpoint, runtime, service } = await configuredSlackEndpoint(fixture, {
...(phase === "attachment" ? { storage: createStorageService().storage } : {}),
});
const provider = runtime.endpoints.get(endpoint.id)!;
provider.reactionErrors.push(Object.assign(new Error("missing_scope"), { code: "slack_webapi_platform_error", data: { error: "missing_scope" } }));
const reaction = vi.spyOn(provider, "applySlackReceiptReaction");
const thread = makeThread({ channelId: "CEARLYERROR", id: "slack:CEARLYERROR:1740000091.1" });
if (phase === "subscription") thread.thread.subscribe = block;
const message = makeMessage({ id: "1740000091.1", text: "@maya preserve the receipt warning", mentioned: true,
...(phase === "attachment" ? { attachments: [{ type: "file", name: "receipt.txt", mimeType: "text/plain", fetchData: async () => { await block(); return Buffer.from("synthetic attachment"); } } as Attachment] } : {}),
});
const inbound = deliverMessage({ callbacks, endpointId: endpoint.id, thread: thread.thread, message, trigger: "mention" });
try {
await vi.waitFor(() => expect(entered).toHaveBeenCalled(), { timeout: 3_000, interval: 10 });
const [delivery] = await readChatDeliveriesByEndpointId(endpoint.id);
await vi.waitFor(async () => {
const [receipt] = await readChatActionsByProviderActionId(`receipt_reaction:${delivery.id}`);
expect(receipt).toMatchObject({ status: "failed", result: { redactedError: expect.stringContaining("missing_scope") } });
}, { timeout: 1_000, interval: 10 });
expect((await readChatDeliveriesByEndpointId(endpoint.id))[0].state).toBe("processing");
release();
await inbound;
expect((await readChatDeliveriesByEndpointId(endpoint.id))[0]).toMatchObject({
state: "processed", redactedError: expect.stringMatching(/Receipt reaction .*missing_scope/),
});
expect(reaction).toHaveBeenCalledTimes(1);
} finally { release(); await inbound; await service.shutdown(); }
});
it.each(["confirmed", "already reacted"] as const)("keeps an omitted-file warning after an early Slack receipt failure and %s retry", async outcome => {
const fixture = await seedCompany();
let release!: () => void;
const gate = new Promise<void>(resolve => { release = resolve; });
const entered = vi.fn();
const { callbacks, endpoint, runtime, service } = await configuredSlackEndpoint(fixture, { storage: createStorageService().storage });
const provider = runtime.endpoints.get(endpoint.id)!;
provider.reactionErrors.push(Object.assign(new Error("missing_scope\nsynthetic provider detail"), { code: "slack_webapi_platform_error", data: { error: "missing_scope" } }));
if (outcome === "already reacted") provider.reactionErrors.push(Object.assign(new Error("already reacted"), { code: "slack_webapi_platform_error", data: { error: "already_reacted" } }));
const reaction = vi.spyOn(provider, "applySlackReceiptReaction");
const thread = makeThread({ channelId: "COMITTEDERROR", id: "slack:COMITTEDERROR:1740000092.1" });
const inbound = deliverMessage({ callbacks, endpointId: endpoint.id, thread: thread.thread, trigger: "mention",
message: makeMessage({ id: "1740000092.1", text: "@maya retain both warnings", mentioned: true,
attachments: [{ type: "file", name: "omitted.txt", mimeType: "text/plain", fetchData: async () => { entered(); await gate; throw new Error("controlled file download failure"); } } as Attachment],
}),
});
try {
await vi.waitFor(() => expect(entered).toHaveBeenCalled(), { timeout: 3_000, interval: 10 });
const [delivery] = await readChatDeliveriesByEndpointId(endpoint.id);
await vi.waitFor(async () => {
const [receipt] = await readChatActionsByProviderActionId(`receipt_reaction:${delivery.id}`);
expect(receipt.status).toBe("failed");
}, { timeout: 1_000, interval: 10 });
release();
await inbound;
const before = (await readChatDeliveriesByEndpointId(endpoint.id))[0];
expect(before.redactedError).toContain("Receipt reaction");
expect(before.redactedError).toContain("1 external attachment was omitted");
expect(reaction).toHaveBeenCalledTimes(1);
const [receipt] = await readChatActionsByProviderActionId(`receipt_reaction:${delivery.id}`);
await db.update(chatActions).set({ result: { ...receipt.result, retryable: true, retryAt: new Date(0).toISOString() } }).where(eq(chatActions.id, receipt.id));
await service.processPendingReceiptReactions(1, receipt.id);
const after = (await readChatDeliveriesByEndpointId(endpoint.id))[0];
expect(after.redactedError).toContain("1 external attachment was omitted");
expect(after.redactedError).not.toContain("Receipt reaction");
expect(after.redactedError).not.toContain("synthetic provider detail");
expect(reaction).toHaveBeenCalledTimes(2);
expect((await readChatActionsById(receipt.id))[0].status).toBe("processed");
} finally { release(); await inbound; await service.shutdown(); }
});
it("acknowledges a signed URL mention through the real Slack SDK without waiting for an unfurl", async () => {
const fixture = await seedCompany();
const { endpoint, runtime, service, wakeup } = await configuredSlackEndpoint(fixture);
const pinned = createChatSdkEndpointRuntime({ ...runtime.configurations.get(endpoint.id)!, logger: "silent" });
const threadId = "slack:CFASTSDK:1740000090.1";
const adapter = pinned.thread(threadId).adapter as unknown as {
_client: { auth: { test: (...args: unknown[]) => Promise<unknown> }; users: { info: (...args: unknown[]) => Promise<unknown> } };
};
adapter._client.auth.test = vi.fn(async () => ({ ok: true, user_id: "UBOT", user: "maya", team_id: "T-PAPERCLIP" }));
adapter._client.users.info = vi.fn(async () => ({ ok: true, user: { id: "USENDER", profile: { display_name: "Test sender" } } }));
const startedAt = performance.now();
try {
const response = await pinned.handleWebhook(signedSlackWebhookRequest({
url: `https://paperclip.example/api/chat-webhooks/${endpoint.publicId}/slack`, contentType: "application/json",
body: JSON.stringify({ type: "event_callback", event_id: "EvFASTSDK", team_id: "T-PAPERCLIP",
event: { type: "app_mention", channel: "CFASTSDK", user: "USENDER", ts: "1740000090.1", text: "<@UBOT> review <https://example.com/review|this link>" },
}),
}));
expect(response.status).toBe(200);
await vi.waitFor(() => expect(runtime.endpoints.get(endpoint.id)!.reactions).toEqual([
{ threadId, messageId: "1740000090.1", emoji: "eyes" },
]), { timeout: 1_000, interval: 10 });
expect(wakeup).toHaveBeenCalledTimes(1);
expect(adapter._client.users.info).toHaveBeenCalled();
process.stdout.write(`Signed URL mention through real Slack SDK: eyes confirmed in ${(performance.now() - startedAt).toFixed(1)}ms (real PostgreSQL, simulated provider APIs)\n`);
} finally {
await pinned.shutdown();
await service.shutdown();
}
});
it("settles a retried Slack receipt reaction when the provider reports already_reacted", async () => {
const fixture = await seedCompany();
const { callbacks, endpoint, runtime, service } =
@@ -28504,6 +28657,37 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
await service.shutdown();
});
it.each([false, true])("clears an older multi-line Slack receipt diagnostic on terminal removal with attachment warning %s", async hasAttachmentWarning => {
const fixture = await seedCompany();
const { callbacks, endpoint, runtime, service } = await configuredSlackEndpoint(fixture);
try {
const thread = makeThread({ channelId: "COLDWARNING", id: "slack:COLDWARNING:1740000093.1" });
await deliverMessage({ callbacks, endpointId: endpoint.id, thread: thread.thread, trigger: "mention",
message: makeMessage({ id: "1740000093.1", text: "@maya clear an older receipt diagnostic", mentioned: true }),
});
const [delivery] = await readChatDeliveriesByEndpointId(endpoint.id);
const [receipt] = await readChatActionsByProviderActionId(`receipt_reaction:${delivery.id}`);
const diagnostic = "Receipt reaction failed after 1 attempt (endpoint_attention): missing_scope\nsynthetic provider detail";
const attachmentWarning = "1 external attachment was omitted";
await db.update(chatActions).set({ status: "failed", result: { attempts: 1, retryable: false, redactedError: diagnostic } }).where(eq(chatActions.id, receipt.id));
await db.update(chatDeliveries).set({ redactedError: diagnostic + (hasAttachmentWarning ? `\n${attachmentWarning}` : "") }).where(eq(chatDeliveries.id, delivery.id));
const [conversation] = await service.listConversations(endpoint.id);
const runId = randomUUID();
await db.insert(heartbeatRuns).values({ id: runId, companyId: fixture.companyId, agentId: fixture.assignedAgentId, status: "succeeded",
contextSnapshot: await chatWakeContext({ endpointId: endpoint.id, issueId: conversation.issueId, provider: "slack", providerMessageId: "1740000093.1" }),
});
const removalIds = await db.transaction(tx => stageReceiptReactionRemovals(tx as unknown as Db, {
endpoint, binding: { companyId: fixture.companyId, endpointId: endpoint.id, conversationId: conversation.id, issueId: conversation.issueId }, runId,
runtimeContext: { generation: Number(receipt.payload.runtimeGeneration), credentialFingerprint: String(receipt.payload.credentialFingerprint) },
}));
expect(removalIds).toHaveLength(1);
await service.processPendingReceiptReactions(1, removalIds[0]);
expect(runtime.endpoints.get(endpoint.id)!.removedReactions).toEqual([{ threadId: thread.thread.id, messageId: "1740000093.1", emoji: "eyes" }]);
expect((await readChatActionsById(removalIds[0]))[0].status).toBe("processed");
expect((await readChatDeliveriesByEndpointId(endpoint.id))[0].redactedError).toBe(hasAttachmentWarning ? attachmentWarning : null);
} finally { await service.shutdown(); }
});
it("canonicalizes prefixed Chat SDK channel ids onto provider inventory resources", async () => {
const fixture = await seedCompany();
const providerFetch = vi.fn(async (input: string | URL | Request) => {
+29 -6
View File
@@ -849,7 +849,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
attachmentOmissionDetail, runtimeContextForRecord, endpointRecord, stageProviderEffect, scheduleProviderEffect,
});
const { handleReaction, processPendingReactionDeliveries } = createInboundReactions(db, { lockCurrentPrincipalAuthorization, runtimeCallbackRecord, runtimeCallbackEndpoint, ensurePrincipal, durableExternalThreadIdentity, recordCurrentMicrosoftTeamsRoute, lifecycleRuntimeFence, deliveryReady, redactError, nonDirectDestinationAllowed, reactionLinkPreflightBarrier: () => options.reactionLinkPreflightBarrier?.() ?? Promise.resolve(), reactionReplayEndpointLockBarrier: () => options.reactionReplayEndpointLockBarrier?.() ?? Promise.resolve(), reactionReplayConversationLockBarrier: () => options.reactionReplayConversationLockBarrier?.() ?? Promise.resolve() });
const { addReceiptReaction, processReceiptReaction, processPendingReceiptReactions } = createReceiptReactions(db, { endpointRecord, runtimeContextForRecord, withCredentialMutationLease, runtimeFor, fetchImpl, runtime, runtimeGeneration, credentialFingerprint, redactError, MAX_ERROR_TEXT, receiptReactionTransportBarrier: () => options.receiptReactionTransportBarrier?.() ?? Promise.resolve() });
const { stageReceiptReaction, addReceiptReaction, processReceiptReaction, processPendingReceiptReactions } = createReceiptReactions(db, { endpointRecord, runtimeContextForRecord, withCredentialMutationLease, runtimeFor, fetchImpl, runtime, runtimeGeneration, credentialFingerprint, redactError, MAX_ERROR_TEXT, receiptReactionTransportBarrier: () => options.receiptReactionTransportBarrier?.() ?? Promise.resolve() });
const { admitDiscordRootMention, discordRootThreadPending, reconcileDiscordRootThread, promoteDiscordRootReceipt } = createDiscordRootThreads(db, { ensureDiscordGatewayRuntimeIsCurrent, withCredentialMutationLease, runtimeCallbackEndpoint, lockCurrentPrincipalAuthorization, processMessage, redactError, scheduleConversationDrain, lifecycleRuntimeFence, runtimeFor, DISCORD_ROOT_THREAD_CONFIRMATION_DELAY_MS, discordRootThreadTransportBarrier: () => options.discordRootThreadTransportBarrier?.() ?? Promise.resolve() });
const { handleDiscordGatewayEvent } = createDiscordGatewayEvents(db, { ensureDiscordGatewayRuntimeIsCurrent, discordGatewayRuntimeIsCurrent, runtimeCallbackRecord, runtimeCallbackEndpoint, applyProviderLifecycleEffect, runtimeGeneration, discordGatewayEventBarrier: (event) => options.discordGatewayEventBarrier?.(event) ?? Promise.resolve() });
const { handlePhotonFailure, handlePhotonEvent } = createPhotonEvents(db, { runtimeCallbackRecord, runtimeCallbackEndpoint, invalidateRuntime, ensurePrincipal, requireCurrentExternalActionAuthorization, isExternalActionAuthorizationChange, scheduleMessageProcessing, processPendingPublications, questionResponses, redactError, renewDiscordGatewayOwnershipForMessageAdmission, handleSdkMessage });
@@ -2319,10 +2319,19 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
principalId: principalResolution.principal.id,
actorUserId: taskUserId,
});
const receiptActionId = taskEndpoint.provider === "slack" && receiptReactionSupported && runtimeContext
? await stageReceiptReaction(taskTx, {
deliveryId: activeDelivery.id,
endpoint: taskEndpoint,
threadId: thread.id,
messageId: message.id,
runtimeContext,
})
: null;
if (taskEndpoint.provider === "imessage-photon") {
await recordPhotonInboundActivity(taskTx, taskEndpoint, issue, comment.id, taskUserId, inboundActivityPublications);
}
return { actorUserId: taskUserId, comment, conversation, issue };
return { actorUserId: taskUserId, comment, conversation, issue, receiptActionId };
};
const taskMutation = await db.transaction(async (tx) => {
// Admission and durable task mutation are separate so the provider
@@ -2462,6 +2471,19 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
);
});
if (!taskMutation) return;
// Slack's eyes mean the authorized message has been durably received.
// Dispatch immediately after commit, independently of file downloads,
// subscriptions and scheduler admission. The durable action owns retry;
// slow Slack I/O must not hold up the task's remaining acceptance work.
const earlyReceiptReaction = taskMutation.receiptActionId
? processReceiptReaction(taskMutation.receiptActionId).catch((error) => {
logger.warn({ error: redactError(error) }, "early chat receipt dispatch failed; durable worker will recover");
})
: null;
if (earlyReceiptReaction) {
backgroundMessageTasks.add(earlyReceiptReaction);
void earlyReceiptReaction.finally(() => backgroundMessageTasks.delete(earlyReceiptReaction));
}
// The open task can fetch the comment immediately, before attachments
// finish preparing or the agent starts. Never publish an uncommitted row.
for (const publication of inboundActivityPublications) {
@@ -2513,10 +2535,11 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
.limit(1)
.then((rows) => rows[0] ?? null);
if (acceptedWake?.status !== "processed") return;
// Provider-visible acknowledgement begins only after the task, external
// comment, durable wakeup request, delivery state, and message link commit.
// Other providers acknowledge accepted wakeup requests here. Slack's
// receipt was already staged with the authorized task/comment above;
// this idempotent reconciliation also covers older recovered deliveries.
await Promise.allSettled([
receiptReactionSupported
earlyReceiptReaction ?? (receiptReactionSupported
? addReceiptReaction({
deliveryId: activeDelivery.id,
endpoint,
@@ -2524,7 +2547,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
runtimeContext,
thread,
})
: Promise.resolve(),
: Promise.resolve()),
// Slack implements this through assistant.threads.setStatus, which
// requires assistant:write. The least-privilege Paperclip manifest
// deliberately does not request that scope; the coalesced lifecycle
@@ -116,7 +116,17 @@ export function createInboundWakeups(db: Db, options: WakeupOptions) {
.set({
state: "processed",
processedAt: acceptedAt,
redactedError: attachmentOmissionDetail(attachmentResult),
redactedError: sql<string | null>`nullif(concat_ws(${"\n"}, (
select replace(replace(${chatActions.result}->>'redactedError', ${"\n"}, ' '), ${"\r"}, ' ') from ${chatActions}
where ${chatActions.companyId} = ${delivery.companyId}
and ${chatActions.endpointId} = ${delivery.endpointId}
and ${chatActions.deliveryId} = ${deliveryId}
and ${chatActions.providerActionId} = ${`receipt_reaction:${deliveryId}`}
and ${chatActions.kind} = 'receipt_reaction'
and ${chatActions.status} = 'failed'
and ${chatActions.payload}->>'operation' = 'add'
limit 1
), ${attachmentOmissionDetail(attachmentResult)}::text), '')`,
nextAttemptAt: null,
updatedAt: new Date(),
})
@@ -1,5 +1,5 @@
import type { ChatRuntime } from "../chat-runtime.js";
import type { EndpointRow, RuntimeContext, LifecycleRuntimeFence } from "./runtime-types.js";
import type { EndpointRow, RuntimeContext, LifecycleRuntimeFence, DbOrTransaction } from "./runtime-types.js";
import { dueChatAction, PROVIDER_EFFECT_STALE_MS } from "../chat-action-work.js";
@@ -52,6 +52,7 @@ import type {
import type { EndpointRecord } from "./runtime-types.js";
import { isSlackReceiptReactionAlreadyApplied } from "./slack/messages.js";
import { beginSlackReceiptTiming } from "./slack/receipt-timing.js";
type ReceiptReactionsOptions = {
endpointRecord: (endpointId: string) => Promise<EndpointRecord | null>;
@@ -67,6 +68,17 @@ type ReceiptReactionsOptions = {
receiptReactionTransportBarrier?: () => Promise<void>;
};
function deliveryWarningWithoutReceipt(previousDiagnostic: unknown) {
const remaining = sql<string | null>`nullif(regexp_replace(${chatDeliveries.redactedError}, ${"^Receipt reaction [^\\n]*(\\n|$)"}, ''), '')`;
if (typeof previousDiagnostic !== "string" || !previousDiagnostic) return remaining;
return sql<string | null>`case
when ${chatDeliveries.redactedError} = ${previousDiagnostic}::text then null
when left(${chatDeliveries.redactedError}, length(${previousDiagnostic}::text) + 1) = ${previousDiagnostic}::text || ${"\n"}::text
then nullif(substring(${chatDeliveries.redactedError} from length(${previousDiagnostic}::text) + 2), '')
else ${remaining}
end`;
}
export function createReceiptReactions(db: Db, options: ReceiptReactionsOptions) {
const { endpointRecord, runtimeContextForRecord, withCredentialMutationLease, runtimeFor, fetchImpl, runtime, runtimeGeneration, credentialFingerprint, redactError, MAX_ERROR_TEXT } = options;
@@ -88,9 +100,29 @@ export function createReceiptReactions(db: Db, options: ReceiptReactionsOptions)
// endpoint again after a rotation and accidentally blessing an obsolete
// adapter with the new generation.
const context = input.runtimeContext ?? runtimeContextForRecord(record);
const [inserted] = await db.transaction(async (tx) => {
await notifyChatActionWork(tx, "receipt_reaction");
return tx
const actionId = await db.transaction((tx) => stageReceiptReaction(tx, {
deliveryId: input.deliveryId,
endpoint: input.endpoint,
threadId: input.thread.id,
messageId: input.message.id,
runtimeContext: context,
}));
if (!actionId) return;
await processReceiptReaction(actionId);
}
// Stage on the caller's task transaction: acknowledgement survives a crash
// after task acceptance, and a rollback can never leave a visible receipt.
async function stageReceiptReaction(tx: DbOrTransaction, input: {
deliveryId: string;
endpoint: EndpointRow;
threadId: string;
messageId: string;
runtimeContext: LifecycleRuntimeFence;
}): Promise<string | null> {
if (input.endpoint.provider === "github") return null;
await notifyChatActionWork(tx, "receipt_reaction");
const [inserted] = await tx
.insert(chatActions)
.values({
companyId: input.endpoint.companyId,
@@ -101,20 +133,19 @@ export function createReceiptReactions(db: Db, options: ReceiptReactionsOptions)
payload: {
version: 1,
operation: "add",
threadId: input.thread.id,
messageId: input.message.id,
threadId: input.threadId,
messageId: input.messageId,
reaction: "eyes",
runtimeGeneration: context.generation,
credentialFingerprint: context.credentialFingerprint,
runtimeGeneration: input.runtimeContext.generation,
credentialFingerprint: input.runtimeContext.credentialFingerprint,
} satisfies ReceiptReactionPayload,
status: "received",
})
.onConflictDoNothing()
.returning({ id: chatActions.id });
});
const actionId =
inserted?.id ??
(await db
(await tx
.select({ id: chatActions.id })
.from(chatActions)
.where(
@@ -127,15 +158,19 @@ export function createReceiptReactions(db: Db, options: ReceiptReactionsOptions)
),
)
.then((rows) => rows[0]?.id ?? null));
if (!actionId) return;
await options.receiptReactionTransportBarrier?.();
await processReceiptReaction(actionId);
return actionId;
}
async function processReceiptReaction(actionId: string): Promise<void> {
await options.receiptReactionTransportBarrier?.();
const initialAction = await db
.select({ endpointId: chatActions.endpointId })
.select({ endpointId: chatActions.endpointId, receivedAt: chatDeliveries.receivedAt })
.from(chatActions)
.leftJoin(chatDeliveries, and(
eq(chatDeliveries.id, chatActions.deliveryId),
eq(chatDeliveries.companyId, chatActions.companyId),
eq(chatDeliveries.endpointId, chatActions.endpointId),
))
.where(
and(
eq(chatActions.id, actionId),
@@ -151,13 +186,14 @@ export function createReceiptReactions(db: Db, options: ReceiptReactionsOptions)
// rotation, pause, and removal so credentials cannot become obsolete
// between authorization and transport.
await withCredentialMutationLease(initialEndpoint.endpoint, (lease) =>
processReceiptReactionWithCredentialLease(actionId, lease),
processReceiptReactionWithCredentialLease(actionId, lease, initialAction.receivedAt),
);
}
async function processReceiptReactionWithCredentialLease(
actionId: string,
credentialLease: CredentialMutationLeaseGuard,
receivedAt: Date | null,
): Promise<void> {
let action = await db
.select()
@@ -350,7 +386,14 @@ export function createReceiptReactions(db: Db, options: ReceiptReactionsOptions)
const adapter = endpointRuntime.thread(payload.threadId).adapter;
await credentialLease.assertOwned();
if (record.endpoint.provider === "slack") {
await endpointRuntime.applySlackReceiptReaction(payload, fetchImpl);
const finishTiming = beginSlackReceiptTiming({ operation: payload.operation, queuedAt: action.createdAt, receivedAt });
try {
await endpointRuntime.applySlackReceiptReaction(payload, fetchImpl);
finishTiming("confirmed");
} catch (error) {
finishTiming(payload.operation === "add" && isSlackReceiptReactionAlreadyApplied(error) ? "confirmed" : "failed");
throw error;
}
} else if (record.endpoint.provider === "github") {
const assertCurrent = async () => {
await credentialLease.assertOwned();
@@ -440,7 +483,7 @@ export function createReceiptReactions(db: Db, options: ReceiptReactionsOptions)
if (action.deliveryId) {
await tx
.update(chatDeliveries)
.set({ redactedError: null, updatedAt: new Date() })
.set({ redactedError: deliveryWarningWithoutReceipt(action.result?.redactedError), updatedAt: new Date() })
.where(
and(
eq(chatDeliveries.id, action.deliveryId),
@@ -486,7 +529,7 @@ export function createReceiptReactions(db: Db, options: ReceiptReactionsOptions)
if (action.deliveryId) {
await tx
.update(chatDeliveries)
.set({ redactedError: null, updatedAt: new Date() })
.set({ redactedError: deliveryWarningWithoutReceipt(action!.result?.redactedError), updatedAt: new Date() })
.where(
and(
eq(chatDeliveries.id, action.deliveryId),
@@ -510,7 +553,7 @@ export function createReceiptReactions(db: Db, options: ReceiptReactionsOptions)
const operationLabel =
payload.operation === "remove" ? " reaction removal" : " reaction";
const diagnostic =
`Receipt${operationLabel} ${retryable ? "deferred" : "failed"} after ${attempt} attempt${attempt === 1 ? "" : "s"} (${disposition.kind}): ${redactError(error)}`.slice(
`Receipt${operationLabel} ${retryable ? "deferred" : "failed"} after ${attempt} attempt${attempt === 1 ? "" : "s"} (${disposition.kind}): ${redactError(error).replace(/[\r\n]+/g, " ")}`.slice(
0,
MAX_ERROR_TEXT,
);
@@ -548,7 +591,7 @@ export function createReceiptReactions(db: Db, options: ReceiptReactionsOptions)
if (failed && action.deliveryId) {
await tx
.update(chatDeliveries)
.set({ redactedError: diagnostic, updatedAt: new Date() })
.set({ redactedError: sql`concat_ws(${"\n"}, ${diagnostic}::text, ${deliveryWarningWithoutReceipt(action.result?.redactedError)})`, updatedAt: new Date() })
.where(
and(
eq(chatDeliveries.id, action.deliveryId),
@@ -582,5 +625,5 @@ export function createReceiptReactions(db: Db, options: ReceiptReactionsOptions)
for (const action of actions) await processReceiptReaction(action.id);
return actions.length;
}
return { addReceiptReaction, processReceiptReaction, processPendingReceiptReactions };
return { stageReceiptReaction, addReceiptReaction, processReceiptReaction, processPendingReceiptReactions };
}
@@ -0,0 +1,41 @@
import { SlackAdapter, type SlackEvent } from "@chat-adapter/slack";
import { ConsoleLogger, type ChatInstance } from "chat";
import { afterEach, describe, expect, it, vi } from "vitest";
import { PaperclipSlackAdapter } from "./adapter.js";
afterEach(() => vi.useRealTimers());
describe("Slack inbound link latency", () => {
it("accepts URL text and files without the SDK's two-second unfurl wait", async () => {
vi.useFakeTimers();
const state = { get: vi.fn(async () => null) };
const chat = { getState: () => state } as unknown as ChatInstance;
class Baseline extends SlackAdapter {
parse(event: SlackEvent) { this.chat = chat; return this.parseSlackMessage(event, "slack:C123:1740000001.1"); }
}
class Candidate extends PaperclipSlackAdapter {
parse(event: SlackEvent) { this.chat = chat; return this.parseSlackMessage(event, "slack:C123:1740000001.1"); }
}
const event = { type: "app_mention", channel: "C123", ts: "1740000001.1", username: "Test sender",
text: "Review <https://example.com/review|this link>",
files: [{ id: "F123", name: "notes.txt", mimetype: "text/plain", url_private: "https://files.slack.com/notes.txt" }],
} as SlackEvent;
const config = { botToken: "xoxb-synthetic", signingSecret: "synthetic", userName: "test-bot", logger: new ConsoleLogger("silent") };
const received = await new Candidate(config).parse(event);
expect(received.text).toBe("Review this link");
expect(received.links).toMatchObject([{ url: "https://example.com/review" }]);
expect(received.attachments).toMatchObject([{ name: "notes.txt", fetchData: expect.any(Function) }]);
expect(state.get).not.toHaveBeenCalled();
let baselineComplete = false;
const baseline = new Baseline(config).parse(event).then((message) => { baselineComplete = true; return message; });
await vi.advanceTimersByTimeAsync(1_950);
expect(baselineComplete).toBe(false);
expect(state.get).toHaveBeenCalledWith("slack:unfurls:1740000001.1");
await vi.runAllTimersAsync();
const original = await baseline;
expect(received.text).toBe(original.text);
expect(received.formatted).toEqual(original.formatted);
expect(received.links).toEqual(original.links);
});
});
@@ -0,0 +1,11 @@
import { SlackAdapter } from "@chat-adapter/slack";
import type { LinkPreview } from "chat";
export class PaperclipSlackAdapter extends SlackAdapter {
// The SDK polls for an optional Slack unfurl for two seconds before it
// delivers a message containing a URL. Paperclip accepts the message text,
// links and files directly; preview metadata is not part of task admission.
protected override async enrichLinks(links: LinkPreview[]): Promise<LinkPreview[]> {
return links;
}
}
@@ -0,0 +1,41 @@
import { afterEach, describe, expect, it, vi } from "vitest";
const tracer = vi.hoisted(() => ({ startSpan: vi.fn() }));
vi.mock("@opentelemetry/api", () => ({ trace: { getTracer: () => tracer } }));
import { beginSlackReceiptTiming } from "./receipt-timing.js";
afterEach(() => { vi.unstubAllEnvs(); vi.restoreAllMocks(); });
const input = { operation: "add" as const, queuedAt: new Date(100), receivedAt: new Date(50) };
describe("Slack receipt timing", () => {
it("does no tracing without the operator endpoint", () => {
vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "");
beginSlackReceiptTiming(input)("confirmed");
expect(tracer.startSpan).not.toHaveBeenCalled();
});
it("records queue wait and delivery age without content or identifiers", () => {
vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "https://collector.example");
const span = { setAttribute: vi.fn(), end: vi.fn() };
tracer.startSpan.mockReturnValue(span);
vi.spyOn(Date, "now").mockReturnValueOnce(200).mockReturnValueOnce(300);
const finish = beginSlackReceiptTiming(input);
finish("confirmed");
expect(tracer.startSpan).toHaveBeenCalledWith("chat.slack.receipt", {
attributes: { "paperclip.chat.receipt.operation": "add", "paperclip.chat.receipt.queue_ms": 100 },
});
expect(span.setAttribute.mock.calls).toEqual([
["paperclip.chat.receipt.outcome", "confirmed"],
["paperclip.chat.receipt.delivery_age_ms", 250],
]);
expect(span.end).toHaveBeenCalledOnce();
});
it("contains tracer failures at creation and completion", () => {
vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "https://collector.example");
tracer.startSpan.mockImplementationOnce(() => { throw new Error("tracer unavailable"); });
expect(() => beginSlackReceiptTiming(input)("failed")).not.toThrow();
tracer.startSpan.mockReturnValue({ setAttribute() { throw new Error("sink unavailable"); } });
expect(() => beginSlackReceiptTiming(input)("confirmed")).not.toThrow();
});
});
@@ -0,0 +1,26 @@
import { trace, type Span } from "@opentelemetry/api";
// Closed, content-free Observability dimensions. No provider IDs or content.
export function beginSlackReceiptTiming(input: {
operation: "add" | "remove";
queuedAt: Date;
receivedAt: Date | null;
}) {
if (!process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim()) return () => {};
let span: Span | undefined;
try {
span = trace.getTracer("paperclip.chat").startSpan("chat.slack.receipt", {
attributes: {
"paperclip.chat.receipt.operation": input.operation,
"paperclip.chat.receipt.queue_ms": Math.max(0, Date.now() - input.queuedAt.getTime()),
},
});
} catch { /* Tracing cannot prevent the reaction. */ }
return (outcome: "confirmed" | "failed") => {
try {
span?.setAttribute("paperclip.chat.receipt.outcome", outcome);
if (input.receivedAt) span?.setAttribute("paperclip.chat.receipt.delivery_age_ms", Math.max(0, Date.now() - input.receivedAt.getTime()));
} catch { /* Tracing cannot change provider settlement. */ }
try { span?.end(); } catch { /* Best-effort span completion. */ }
};
}
@@ -202,6 +202,9 @@ export async function stageReceiptReactionRemovals(
credentialFingerprint: input.runtimeContext.credentialFingerprint,
} satisfies ReceiptReactionPayload,
status: "received",
result: typeof receipt.result?.redactedError === "string"
? { redactedError: receipt.result.redactedError }
: {},
},
];
});
+3 -3
View File
@@ -64,9 +64,9 @@ vi.mock("@chat-adapter/discord", () => ({
}));
vi.mock("@chat-adapter/slack", () => ({
createSlackAdapter: (config: Record<string, unknown>) => {
captures.slackConfigs.push(config);
return { name: "slack" };
SlackAdapter: class {
name = "slack";
constructor(config: Record<string, unknown>) { captures.slackConfigs.push(config); }
},
}));
+3 -5
View File
@@ -14,10 +14,8 @@ import {
type DiscordAdapter,
type DiscordAdapterConfig,
} from "@chat-adapter/discord";
import {
createSlackAdapter,
type SlackAdapterConfig,
} from "@chat-adapter/slack";
import type { SlackAdapterConfig } from "@chat-adapter/slack";
import { PaperclipSlackAdapter } from "./chat-channels/slack/adapter.js";
import {
createTeamsAdapter,
type TeamsAdapter,
@@ -1374,7 +1372,7 @@ function createProviderAdapter(
timeout: SLACK_WEB_API_TIMEOUT_MS,
},
};
const adapter = createSlackAdapter(adapterConfig);
const adapter = new PaperclipSlackAdapter(adapterConfig);
(
adapter as unknown as SlackAdapterInternals
).paperclipFileUploadReceiptContext = new AsyncLocalStorage<