From 3db33c131ef59d8d7ad8ce3dc9aae4af0ce5dc25 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 16 Oct 2026 16:05:46 -0500 Subject: [PATCH] perf(slack): overlap endpoint preparation and narrow credential reads (#16593) Prepare Slack endpoints in pairs with a per-call concurrency limit of two, and resolve only the current bot token for live tool authority. Preserve candidate-order errors, active-sibling settlement, credential rotation and authorization revocation. Verified with real PostgreSQL regression controls, full repository CI/E2E, security checks and current-head Greptile Plus5/5 with no unresolved findings. Co-Authored-By: Paperclip --- doc/connections/SLACK-TASK-TOOLS.md | 7 + maps/agent-access-map/chat-channels.md | 9 ++ maps/feature-map/chat-channels.md | 8 + .../__tests__/slack-tool-preparation.test.ts | 152 ++++++++++++++++++ .../chat-channels/slack/task-authority.ts | 5 +- .../src/services/connectors/slack-catalog.ts | 10 +- .../slack-endpoint-preparation.test.ts | 75 +++++++++ .../connectors/slack-endpoint-preparation.ts | 20 +++ server/src/services/connectors/slack.ts | 16 +- 9 files changed, 284 insertions(+), 18 deletions(-) create mode 100644 server/src/__tests__/slack-tool-preparation.test.ts create mode 100644 server/src/services/connectors/slack-endpoint-preparation.test.ts create mode 100644 server/src/services/connectors/slack-endpoint-preparation.ts diff --git a/doc/connections/SLACK-TASK-TOOLS.md b/doc/connections/SLACK-TASK-TOOLS.md index 892426ac34c..351cc8ef79e 100644 --- a/doc/connections/SLACK-TASK-TOOLS.md +++ b/doc/connections/SLACK-TASK-TOOLS.md @@ -44,6 +44,13 @@ signed approvals. Removing a tool profile binding does not get undone by resolvi a retained session. Disconnecting an identity or disabling a connection revokes access for retained runs too. +Assignment and tool discovery prepare at most two endpoints at a time. Each +endpoint still resolves live authority. Preparation waits for the active pair +before it returns or fails, keeps candidate order, and stops before the next +pair if either check fails. Tool authority resolves only the current bot token; +webhook signing and app tokens stay on their existing transport paths. No +credential or authorization cache is introduced. + ## Tools and scope upgrades The reviewed, strict method/scope/argument matrix is diff --git a/maps/agent-access-map/chat-channels.md b/maps/agent-access-map/chat-channels.md index 307039764ff..84e40384895 100644 --- a/maps/agent-access-map/chat-channels.md +++ b/maps/agent-access-map/chat-channels.md @@ -129,6 +129,15 @@ The pinned SDK transport regression is covered in [receipt tests](../../server/src/services/chat-slack-receipts.test.ts); live acknowledgement speed needs a separate provider test. +Slack assignment and tool discovery bound concurrent endpoint preparation to +two. Every endpoint keeps the live task, company, agent and linked-user checks; +403 denials omit only that endpoint. Preparation waits for both active checks +before it returns an ordered result or propagates an error. Tool authority reads +the current bot token without resolving unrelated webhook credentials. See the +[real-database preparation controls](../../server/src/__tests__/slack-tool-preparation.test.ts) +and [batch failure controls](../../server/src/services/connectors/slack-endpoint-preparation.test.ts). +These local controls do not establish live Slack latency or provider access. + New conversations resolve typed project/workspace defaults from their endpoint and destination under the task-creation transaction. Defaults do not authorize private sources: admission checks the linked user's authority and assigned agent. diff --git a/maps/feature-map/chat-channels.md b/maps/feature-map/chat-channels.md index 349778ad62c..3ff7a0f0789 100644 --- a/maps/feature-map/chat-channels.md +++ b/maps/feature-map/chat-channels.md @@ -60,6 +60,14 @@ and removes them when the final reply arrives. The pinned Slack SDK encodes a linear DM with an empty thread timestamp; receipt transport accepts that address while still validating the channel and the source message timestamp. +For an agent with two linked Slack endpoints, compare assignment preparation and +tool discovery in the run trace. Endpoint checks may overlap in pairs. Confirm +that both permitted tool sets remain available, then revoke one linked identity +and confirm that the next preparation excludes it. The +[database regression](../../server/src/__tests__/slack-tool-preparation.test.ts) +uses real PostgreSQL with one connection and controlled credential resolution; +it does not prove live provider latency or an end-to-end response target. + ### `endpoint-management` Automated: [channel integration](../../server/src/__tests__/chat-channels.integration.test.ts) covers shared endpoint/delivery behavior; live credential repair and provider ownership remain manual checks. diff --git a/server/src/__tests__/slack-tool-preparation.test.ts b/server/src/__tests__/slack-tool-preparation.test.ts new file mode 100644 index 00000000000..1475247632f --- /dev/null +++ b/server/src/__tests__/slack-tool-preparation.test.ts @@ -0,0 +1,152 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { eq } from "drizzle-orm"; +import { + agents, chatEndpoints, chatExternalPrincipals, chatIdentityLinks, companies, + companyMemberships, createDb, heartbeatRuns, issues, toolApplications, toolConnections, +} from "@paperclipai/db"; +import type { ToolCredentialSecretRef } from "@paperclipai/shared"; +import { createChatCredentials } from "../services/chat-channels/credentials.js"; +import { createSlackTaskAuthorityRegistration } from "../services/chat-channels/slack/task-authority.js"; +import { resolveSlackTaskAuthority } from "../services/connectors/slack-authority.js"; +import { slackAssignedResource } from "../services/connectors/slack.js"; +import { slackToolsForSession, syncSlackBotTools } from "../services/connectors/slack-catalog.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { initializeRunIdentity } from "../services/run-identity.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const describeDatabase = support.supported ? describe : describe.skip; + +function deferred() { + let resolve!: () => void; + const promise = new Promise(yes => { resolve = yes; }); + return { promise, resolve }; +} + +describeDatabase("Slack tool preparation", () => { + let database: Awaited>; + let db: ReturnType; + + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("paperclip-slack-preparation-"); + db = createDb(database.connectionString, { maxConnections: 1 }); + await instanceSettingsService(db).updateExperimental({ enableChatConnectors: true }); + }); + afterAll(async () => { await database?.cleanup(); }); + + async function fixture() { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(), userId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Slack preparation", issuePrefix: randomUUID().slice(0, 8).toUpperCase() }); + await db.insert(agents).values({ id: agentId, companyId, name: "Slack agent" }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, status: "active", membershipRole: "owner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Briefing", assigneeAgentId: agentId, responsibleUserId: userId }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", responsibleUserId: userId, contextSnapshot: { issueId, source: "routine" } }); + await initializeRunIdentity(db, { companyId, agentId, issueId, runId, responsibleUserId: userId, cause: "routine" }); + const refs: ToolCredentialSecretRef[] = []; + const endpoints: Array = []; + const links: Array = []; + for (let i = 0; i < 2; i++) { + const [application] = await db.insert(toolApplications).values({ companyId, name: `Slack ${i}`, type: "mcp_http" }).returning(); + const credentialSecretRefs = ["botToken", "signingSecret", "appToken"].map(key => ({ secretId: randomUUID(), configPath: `credentials.${key}`, versionSelector: "latest" as const, required: true })); + refs.push(...credentialSecretRefs); + const [connection] = await db.insert(toolConnections).values({ companyId, applicationId: application.id, name: `Slack ${i}`, uid: randomUUID(), transport: "mcp_remote", credentialSecretRefs }).returning(); + const [endpoint] = await db.insert(chatEndpoints).values({ companyId, connectionId: connection.id, assignedAgentId: agentId, provider: "slack", publicId: randomUUID(), providerAccountId: `T${i}`, status: "active" }).returning(); + endpoints.push(endpoint); + const [principal] = await db.insert(chatExternalPrincipals).values({ companyId, provider: "slack", providerAccountId: `T${i}`, externalId: `U${i}`, isBot: false }).returning(); + const [link] = await db.insert(chatIdentityLinks).values({ companyId, endpointId: endpoint.id, principalId: principal.id, paperclipUserId: userId, status: "linked" }).returning(); + links.push(link); + await db.transaction(tx => syncSlackBotTools(tx, endpoint, userId, true)); + } + const token = "current-bot-token"; + const resolveSecretValue = vi.fn(async (_companyId: string, secretId: string) => { + const ref = refs.find(ref => ref.secretId === secretId)!; + return ref.configPath === "credentials.botToken" ? token : "webhook-secret"; + }); + const credentials = createChatCredentials(db, { secrets: { resolveSecretValue } } as unknown as Parameters[1]); + const registration = createSlackTaskAuthorityRegistration(db, { + resolveCredentials: credentials.resolveCredentials, + authorizeInboundWakeup: async () => { throw new Error("This is an ordinary task"); }, + }); + return { binding: { companyId, agentId, issueId, runId }, endpoints, links, refs, resolveSecretValue, + ...registration }; + } + + const paths = { + assignments: async (binding: { companyId: string; agentId: string; runId: string; issueId: string }) => + (await slackAssignedResource(db, binding)).map(resource => resource.id), + tools: async (binding: { companyId: string; agentId: string; runId: string; issueId: string }) => + [...new Set((await slackToolsForSession(db, binding)).map(tool => String(tool.providerMetadata?.endpointId)))], + }; + + it.each(Object.entries(paths))("overlaps two independent endpoint checks for %s", async (_name, prepare) => { + const f = await fixture(); + const gate = deferred(); + let active = 0, peak = 0; + f.resolveSecretValue.mockImplementation(async (_companyId, secretId) => { + if (f.refs.find(ref => ref.secretId === secretId)!.configPath === "credentials.botToken") { + active++; peak = Math.max(peak, active); + await gate.promise; + active--; + } + return "credential"; + }); + const pending = prepare(f.binding); + try { + await vi.waitFor(() => expect(peak).toBe(2), { timeout: 1000, interval: 10 }); + } finally { + gate.resolve(); + await pending; + f.unregisterSlackTaskAuthority(); + } + expect((await pending).sort()).toEqual(f.endpoints.map(endpoint => endpoint.id).sort()); + }); + + it("resolves only current bot credentials and observes rotation without a cache", async () => { + const f = await fixture(); + f.resolveSecretValue.mockImplementation(async (_companyId, secretId) => { + const ref = f.refs.find(ref => ref.secretId === secretId)!; + if (ref.configPath !== "credentials.botToken") throw new Error("Webhook credential is unavailable"); + return f.resolveSecretValue.mock.calls.length === 1 ? "old-bot-token" : "rotated-bot-token"; + }); + try { + const binding = { ...f.binding, endpointId: f.endpoints[0].id }; + const before = await resolveSlackTaskAuthority(db, binding); + const after = await resolveSlackTaskAuthority(db, binding); + expect(before.botToken).toBe("old-bot-token"); + expect(after.botToken).toBe("rotated-bot-token"); + expect(after.revision).not.toBe(before.revision); + expect(f.resolveSecretValue).toHaveBeenCalledTimes(2); + } finally { f.unregisterSlackTaskAuthority(); } + }); + + it("preserves a failed or missing bot credential", async () => { + const f = await fixture(); + const binding = { ...f.binding, endpointId: f.endpoints[0].id }; + try { + const error = new Error("Bot credential is revoked"); + f.resolveSecretValue.mockRejectedValueOnce(error); + await expect(resolveSlackTaskAuthority(db, binding)).rejects.toBe(error); + f.resolveSecretValue.mockResolvedValueOnce(""); + await expect(resolveSlackTaskAuthority(db, binding)).rejects.toMatchObject({ status: 403 }); + } finally { f.unregisterSlackTaskAuthority(); } + }); + + it.each(Object.entries(paths))("rechecks identity revocation for %s without touching another company", async (_name, prepare) => { + const f = await fixture(); + const foreign = await fixture(); + foreign.unregisterSlackTaskAuthority(); + const registration = createSlackTaskAuthorityRegistration(db, { + resolveCredentials: createChatCredentials(db, { secrets: { resolveSecretValue: f.resolveSecretValue } } as unknown as Parameters[1]).resolveCredentials, + authorizeInboundWakeup: async () => { throw new Error("This is an ordinary task"); }, + }); + try { + expect((await prepare(f.binding)).sort()).toEqual(f.endpoints.map(endpoint => endpoint.id).sort()); + await db.update(chatIdentityLinks).set({ status: "revoked" }).where(eq(chatIdentityLinks.id, f.links[0].id)); + f.resolveSecretValue.mockClear(); + expect(await prepare(f.binding)).toEqual([f.endpoints[1].id]); + expect(f.resolveSecretValue).toHaveBeenCalledTimes(1); + expect(f.resolveSecretValue.mock.calls[0][0]).toBe(f.binding.companyId); + } finally { registration.unregisterSlackTaskAuthority(); } + }); +}); diff --git a/server/src/services/chat-channels/slack/task-authority.ts b/server/src/services/chat-channels/slack/task-authority.ts index 8fa8f85396b..1cd7243ff3f 100644 --- a/server/src/services/chat-channels/slack/task-authority.ts +++ b/server/src/services/chat-channels/slack/task-authority.ts @@ -1,4 +1,5 @@ import type { EndpointRow, DbOrTransaction } from "../runtime-types.js"; +import type { createChatCredentials } from "../credentials.js"; import { slackSearchActionToken } from "../../connectors/slack-search-context.js"; import { slackAuthorizationRevision } from "../../connectors/slack-revision.js"; @@ -45,7 +46,7 @@ type SlackTaskAuthorityRegistrationOptions = { endpoint: EndpointRow; conversation: typeof chatConversations.$inferSelect; }>; - resolveCredentials: (endpoint: EndpointRow) => Promise>; + resolveCredentials: ReturnType["resolveCredentials"]; }; export function createSlackTaskAuthorityRegistration(db: Db, options: SlackTaskAuthorityRegistrationOptions) { @@ -136,7 +137,7 @@ export function createSlackTaskAuthorityRegistration(db: Db, options: SlackTaskA if (!principal) throw forbidden("Slack sender is no longer available"); return { ...source, principalId: principal.id, slackUserId: principal.externalId }; }); - const credentials = await measureRunPhase("connectors.slack.credentials", () => resolveCredentials(resolved.endpoint)); + const credentials = await measureRunPhase("connectors.slack.credentials", () => resolveCredentials(resolved.endpoint, db, ["botToken"])); if (!credentials.botToken) throw forbidden("Slack bot credential is unavailable"); return { endpoint: resolved.endpoint, issueId: binding.issueId, conversation: resolved.conversation, diff --git a/server/src/services/connectors/slack-catalog.ts b/server/src/services/connectors/slack-catalog.ts index cd0b2d79f7d..00d6796d094 100644 --- a/server/src/services/connectors/slack-catalog.ts +++ b/server/src/services/connectors/slack-catalog.ts @@ -1,3 +1,4 @@ +import { prepareSlackEndpoints } from "./slack-endpoint-preparation.js"; import { createHash } from "node:crypto"; import { measureRunPhase } from "../execution-preparation-timing.js"; import { and, eq, isNull, sql } from "drizzle-orm"; @@ -271,12 +272,9 @@ export async function slackToolsForSession( session: Parameters[1], ): Promise { if (!session.agentId || !session.runId || !session.issueId) return []; - const tools: ToolGatewayDescriptor[] = []; - for (const endpoint of await slackEndpointCandidates(db, { + const endpoints = await slackEndpointCandidates(db, { companyId: session.companyId, agentId: session.agentId, - })) { - tools.push(...(await slackToolsForEndpoint(db, session, endpoint.id))); - } - return tools; + }); + return prepareSlackEndpoints(endpoints, endpoint => slackToolsForEndpoint(db, session, endpoint.id)); } diff --git a/server/src/services/connectors/slack-endpoint-preparation.test.ts b/server/src/services/connectors/slack-endpoint-preparation.test.ts new file mode 100644 index 00000000000..d606d27cce2 --- /dev/null +++ b/server/src/services/connectors/slack-endpoint-preparation.test.ts @@ -0,0 +1,75 @@ +import { describe, expect, it, vi } from "vitest"; +import { prepareSlackEndpoints } from "./slack-endpoint-preparation.js"; + +function deferred() { + let resolve!: (value: T) => void; + let reject!: (error: Error) => void; + const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); + return { promise, resolve, reject }; +} +const endpoints = Array.from({ length: 5 }, (_, i) => ({ id: String(i), connectionId: `connection-${i}` })); + +describe("Slack endpoint preparation batches", () => { + it("does not start work without candidates", async () => { + const resolve = vi.fn(); + expect(await prepareSlackEndpoints([], resolve)).toEqual([]); + expect(resolve).not.toHaveBeenCalled(); + }); + + it("keeps at most two checks active and preserves candidate order", async () => { + const pending = endpoints.map(() => deferred()); + const resolve = vi.fn((endpoint: typeof endpoints[number]) => pending[Number(endpoint.id)].promise); + const prepared = prepareSlackEndpoints(endpoints, resolve); + expect(resolve.mock.calls.map(([endpoint]) => endpoint.id)).toEqual(["0", "1"]); + pending[1].resolve(["second"]); + await Promise.resolve(); + expect(resolve).toHaveBeenCalledTimes(2); + pending[0].resolve(["first"]); + await vi.waitFor(() => expect(resolve).toHaveBeenCalledTimes(4)); + pending[3].resolve(["fourth"]); + pending[2].resolve([]); + await vi.waitFor(() => expect(resolve).toHaveBeenCalledTimes(5)); + pending[4].resolve(["fifth"]); + expect(await prepared).toEqual(["first", "second", "fourth", "fifth"]); + }); + + it("settles the active batch before rejecting and does not start later candidates", async () => { + const gate = deferred(); + const error = new Error("Endpoint unavailable"); + const resolve = vi.fn((endpoint: typeof endpoints[number]) => endpoint.id === "0" ? Promise.reject(error) : gate.promise); + let finished = false; + const prepared = prepareSlackEndpoints(endpoints, resolve); + const checked = expect(prepared).rejects.toBe(error).finally(() => { finished = true; }); + await Promise.resolve(); + expect(finished).toBe(false); + expect(resolve).toHaveBeenCalledTimes(2); + gate.resolve([]); + await checked; + expect(resolve).toHaveBeenCalledTimes(2); + }); + + it("keeps candidate-order failure precedence even if the second fails first", async () => { + const first = deferred(); + const firstError = new Error("First candidate failed"); + const secondError = new Error("Second candidate failed"); + const resolve = vi.fn((endpoint: typeof endpoints[number]) => endpoint.id === "0" ? first.promise : Promise.reject(secondError)); + const checked = expect(prepareSlackEndpoints(endpoints, resolve)).rejects.toBe(firstError); + await Promise.resolve(); + first.reject(firstError); + await checked; + expect(resolve).toHaveBeenCalledTimes(2); + }); + + it("settles a sibling after a synchronous resolver failure", async () => { + const gate = deferred(); + const error = new Error("Synchronous failure"); + const resolve = vi.fn((endpoint: typeof endpoints[number]) => { + if (endpoint.id === "0") throw error; + return gate.promise; + }); + const checked = expect(prepareSlackEndpoints(endpoints, resolve)).rejects.toBe(error); + expect(resolve).toHaveBeenCalledTimes(2); + gate.resolve([]); + await checked; + }); +}); diff --git a/server/src/services/connectors/slack-endpoint-preparation.ts b/server/src/services/connectors/slack-endpoint-preparation.ts new file mode 100644 index 00000000000..447966c42af --- /dev/null +++ b/server/src/services/connectors/slack-endpoint-preparation.ts @@ -0,0 +1,20 @@ +type SlackEndpoint = { id: string; connectionId: string }; + +export async function prepareSlackEndpoints( + endpoints: readonly SlackEndpoint[], + resolve: (endpoint: SlackEndpoint) => Promise, +): Promise { + const contributions: T[] = []; + for (let offset = 0; offset < endpoints.length; offset += 2) { + const results = await Promise.allSettled( + endpoints.slice(offset, offset + 2).map(async endpoint => resolve(endpoint)), + ); + for (const result of results) { + if (result.status === "rejected") throw result.reason; + } + for (const result of results) { + if (result.status === "fulfilled") contributions.push(...result.value); + } + } + return contributions; +} diff --git a/server/src/services/connectors/slack.ts b/server/src/services/connectors/slack.ts index 0298ec6b7da..4946b954e70 100644 --- a/server/src/services/connectors/slack.ts +++ b/server/src/services/connectors/slack.ts @@ -6,6 +6,7 @@ import { slackBotCatalogIsCurrent, syncSlackBotTools } from "./slack-catalog.js" import { SLACK_TOOLS } from "@paperclipai/shared"; import { measureRunPhase } from "../execution-preparation-timing.js"; import { measureRunTransaction } from "../execution-transaction-timing.js"; +import { prepareSlackEndpoints } from "./slack-endpoint-preparation.js"; import { toolProfiles, toolProfileBindings, @@ -104,16 +105,11 @@ export async function slackAssignedResource( db: Db, binding: Partial & { companyId: string; agentId: string }, ) { - const resources = []; - for (const endpoint of await measureRunPhase("connectors.slack.candidates", () => slackEndpointCandidates(db, binding))) { - resources.push( - ...(await slackAssignedEndpoint(db, { - ...binding, - endpointId: endpoint.id, - }, endpoint)), - ); - } - return resources; + const endpoints = await measureRunPhase("connectors.slack.candidates", () => slackEndpointCandidates(db, binding)); + return prepareSlackEndpoints(endpoints, endpoint => slackAssignedEndpoint(db, { + ...binding, + endpointId: endpoint.id, + }, endpoint)); } const messageView = (message: SlackObject, team: string, channel: string) => ({