diff --git a/maps/agent-access-map/chat-channels.md b/maps/agent-access-map/chat-channels.md index 400aa48b951..961b84899b8 100644 --- a/maps/agent-access-map/chat-channels.md +++ b/maps/agent-access-map/chat-channels.md @@ -147,6 +147,23 @@ the current bot token without resolving unrelated webhook credentials. See the 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. +Native tool gateways use the durable run task binding for discovery and policy +checks, with legacy JSON task fields as a fallback. GitHub and Slack follow +accepted request provenance across task continuations. GitHub can retain the +originating conversation after a board follow-up from the same accepted user. +An inherited origin cannot cross a task, agent, company, or requester boundary. +Cycle checks reject invalid history without a fixed limit on valid continuations. +Endpoint settings, linked identity, tool profiles, and invocation policy still +apply on each call. The native tool contract rotates older provider sessions so +an empty retained catalog is not reused. PostgreSQL fixtures cover discovery, +review publication, Slack reads, and revocation in +[channel integration](../../server/src/__tests__/chat-channels.integration.test.ts) +and [provenance controls](../../server/src/services/chat-channels/run-origin.test.ts). +The [voice gateway fixture](../../server/src/__tests__/voice-session-store.integration.test.ts) +also verifies durable task binding for a later personal callback invocation. +Failure replies cannot replace final answers inherited from an earlier run. +These fixtures do not prove live provider delivery. + Catalog authorization projection (2026-10-17): tool discovery uses the same live service-owned task, company, agent, requester, endpoint and credential checks as full authority resolution. It omits the revision digest and search-action token, diff --git a/maps/feature-map/chat-channels.md b/maps/feature-map/chat-channels.md index 7b51a48d6e4..5288ba7ff47 100644 --- a/maps/feature-map/chat-channels.md +++ b/maps/feature-map/chat-channels.md @@ -75,6 +75,19 @@ identical catalog membership, absence of digest SQL during discovery, and revocation on the next preparation. A faster component is not proof of a faster Slack reaction or final reply; measure those separately on the deployed revision. +For GitHub and Slack tool continuity, start a task from the channel, then admit +a later monitor or retry run for the same task. Check that the later run can +discover and call its channel tools without a new inbound message. For GitHub, +also continue from a board message by the same accepted requester and complete +the review in the original conversation, including when the original source is +recorded only in legacy wake metadata. Revoke the linked identity and check +that the next call fails. Repeat the boundary checks with another task, agent, +company, and requester. The channel and provenance fixtures above cover these +contracts with synthetic providers; record a live journey separately. +For a long-lived task, repeat more than 100 continuations and check that tool +access remains available. Fail a later GitHub run after a final answer and +check that the earlier answer remains intact. + ### `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__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 539acdfca90..cd3b83ec975 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -27,6 +27,9 @@ import { githubResponseCommentService } from "../services/chat-github-response-c import { githubReviewCheckService } from "../services/chat-github-checks.js"; import { githubAutomaticReviewEvent } from "../services/chat-github-events.js"; import { githubBotToolsForSession, syncGitHubBotTools } from "../services/chat-github-tools.js"; +import { buildPaperclipRuntimeMcpServers } from "../services/heartbeat/run-preparation.js"; +import { createAssignedMcpTools } from "../services/native-runtime/assigned-mcp-tools.js"; +import { createToolGatewayService } from "../services/tool-gateway.js"; import { resolveGitHubOperationCredentials } from "../services/github-operation-credentials.js"; import { initializeRunIdentity } from "../services/run-identity.js"; import { chatGitHubRegistrations, chatGitHubReviews, chatGitHubConfigurations, toolCatalogEntries } from "@paperclipai/db"; @@ -2786,6 +2789,55 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { await expect(gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters: assessment, idempotencyKey })).rejects.toThrow(); expect(summaryAttempts).toBe(2); }, 25_000); + it.each([ + { boardFollowUp: false, legacyOrigin: false }, + { boardFollowUp: true, legacyOrigin: false }, + { boardFollowUp: false, legacyOrigin: true }, + { boardFollowUp: true, legacyOrigin: true }, + ])("keeps GitHub review tools callable through a later native task gateway (board follow-up: $boardFollowUp, legacy origin: $legacyOrigin)", async ({ boardFollowUp, legacyOrigin }) => { + vi.stubEnv("PAPERCLIP_API_URL", "https://paperclip.example.test"); + onTestFinished(() => vi.unstubAllEnvs()); + const f = await publicationRetryFixture(); + const identity = await initializeRunIdentity(db, { + companyId: f.companyId, runId: f.run.id, issueId: f.conversation.issueId!, + responsibleUserId: "owner-user", messageIds: legacyOrigin ? [] : [String(f.run.contextSnapshot?.wakeCommentId)], cause: "instruction", + }); + vi.stubGlobal("fetch", f.providerFetch); + onTestFinished(() => vi.unstubAllGlobals()); + const [agent] = await db.select().from(agents).where(eq(agents.id, f.assignedAgentId)); + const gateway = createToolGatewayService(db); + const first = await buildPaperclipRuntimeMcpServers({ db, agent: agent!, runId: f.run.id }); + const firstTools = await createAssignedMcpTools({ gateway, gatewayPublicId: new URL(first[0].url).pathname.split("/").at(-1)!, bearerToken: first[0].token }); + expect(await firstTools.execute({ tool: "paperclip_search_assigned_tools", arguments: { query: "begin_review" } })).toMatchObject({ tools: expect.arrayContaining([expect.objectContaining({ name: expect.stringContaining("begin_review") })]) }); + await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.run.id)); + const [later] = await db.insert(heartbeatRuns).values({ + companyId: f.companyId, agentId: f.assignedAgentId, status: "running", runtimeMode: "native", + nativeIssueId: f.conversation.issueId, contextSnapshot: { wakeReason: "issue_monitor_due" }, + }).returning(); + const [boardMessage] = boardFollowUp ? await db.insert(issueComments).values({ + companyId: f.companyId, issueId: f.conversation.issueId!, authorType: "user", authorUserId: "owner-user", body: "Finish the review", + }).returning() : []; + await initializeRunIdentity(db, { + companyId: f.companyId, runId: later.id, issueId: f.conversation.issueId!, + responsibleUserId: "owner-user", parentContextId: identity.id, messageIds: boardMessage ? [boardMessage.id] : [], cause: "monitor", + }); + const servers = await buildPaperclipRuntimeMcpServers({ db, agent: agent!, runId: later.id }); + expect(servers).toHaveLength(1); + await expect(gateway.createSession({ companyId: f.companyId, agentId: f.assignedAgentId, runId: later.id, issueId: randomUUID() })) + .rejects.toMatchObject({ reasonCode: "run_context_mismatch" }); + const assigned = await createAssignedMcpTools({ gateway, gatewayPublicId: new URL(servers[0].url).pathname.split("/").at(-1)!, bearerToken: servers[0].token }); + expect(await assigned.execute({ tool: "paperclip_search_assigned_tools", arguments: { query: "begin_review" } })).toMatchObject({ tools: expect.arrayContaining([expect.objectContaining({ name: expect.stringContaining("begin_review") })]) }); + const read = assigned.definitions().find(tool => String(tool.name).includes("read_pull_request"))!; + expect(await assigned.execute({ tool: String(read.name), arguments: { section: "metadata" } })).toMatchObject({ data: { pull: { number: 418 } } }); + const begin = assigned.definitions().find(tool => String(tool.name).includes("begin_review"))!; + await assigned.execute({ tool: String(begin.name), arguments: { reviewedCommit: f.head } }); + const submit = assigned.definitions().find(tool => String(tool.name).includes("submit_review"))!; + expect(await assigned.execute({ tool: String(submit.name), arguments: f.assessment })).toMatchObject({ data: { status: "processed", score: 3 } }); + expect(f.responseComments.size).toBe(1); + expect([...f.responseComments.values()][0].body).toContain("Advisory review"); + await db.update(chatIdentityLinks).set({ status: "revoked" }).where(eq(chatIdentityLinks.endpointId, f.endpoint.id)); + await expect(assigned.execute({ tool: String(read.name), arguments: { section: "metadata" } })).rejects.toThrow(); + }); it.each([403, 401])("preserves GitHub connection intent after publication HTTP %s", async status => { const f = await toolReplyFixture(); const initialEndpoint = await f.service.get(f.endpoint.id); @@ -3307,6 +3359,18 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(fallback.state).toBe(expected === "none" ? "published" : "cancelled"); }, ); + it("preserves an earlier final GitHub answer when a later continuation fails", async () => { + const f = await toolReplyFixture(); + const identity = await initializeRunIdentity(db, { companyId: f.companyId, runId: f.run.id, issueId: f.conversation.issueId!, responsibleUserId: "owner-user", messageIds: [String(f.run.contextSnapshot?.wakeCommentId)], cause: "instruction" }); + await githubChatReviewService(db, f.providerFetch).execute(f.session, "comment", { body: "Earlier final answer", idempotencyKey: randomUUID() }); + await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.run.id)); + const [later] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.assignedAgentId, status: "failed", nativeIssueId: f.conversation.issueId, contextSnapshot: {} }).returning(); + await initializeRunIdentity(db, { companyId: f.companyId, runId: later.id, issueId: f.conversation.issueId!, responsibleUserId: "owner-user", parentContextId: identity.id, cause: "monitor" }); + const writes = f.responseWrites.length; + await expect(githubResponseCommentService(db, f.providerFetch).failRun({ companyId: f.companyId, endpointId: f.endpoint.id, issueId: f.conversation.issueId!, runId: later.id, body: "Later run failed" })).rejects.toMatchObject({ status: 403 }); + expect(f.responseWrites).toHaveLength(writes); + expect([...f.responseComments.values()][0].body).toContain("Earlier final answer"); + }); it("rechecks a GitHub fallback after waiting for the credential lane", async () => { const f = await toolReplyFixture(); await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, f.run.id)); @@ -8228,6 +8292,40 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { await expect(executeConnectorTool(db, binding, "slack_post_message", { channel: "DROUTINE", text: "Revoked", idempotencyKey: randomUUID() })).rejects.toThrow(); }); + it("keeps Slack tools callable after a native task continuation without wake JSON", async () => { + const { resolveConnectorAssignments } = await import("../services/connector-runtime.js"); + const { executeSlackTool } = await import("../services/connectors/slack.js"); + await instanceSettingsService(db).updateExperimental({ enableChatConnectors: true }); + const f = await seedCompany(); + const { callbacks, endpoint, service } = await configuredSlackEndpoint(f, { allowUnlinkedPeople: false }); + const [principal] = await db.insert(chatExternalPrincipals).values({ companyId: f.companyId, provider: "slack", providerAccountId: "T-PAPERCLIP", externalId: "UCONTINUE" }).returning(); + await db.insert(chatIdentityLinks).values({ companyId: f.companyId, endpointId: endpoint.id, principalId: principal.id, paperclipUserId: "owner-user", status: "linked", confirmedAt: new Date() }); + const thread = makeThread({ channelId: "slack:CCONTINUE", id: "slack:CCONTINUE:7100.1" }); + await deliverMessage({ callbacks, endpointId: endpoint.id, thread: thread.thread, message: makeMessage({ id: "7100.1", text: "@maya read this channel", mentioned: true, userId: "UCONTINUE" }), trigger: "mention" }); + const [conversation] = await readChatConversationsByEndpointId(endpoint.id); + const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, endpoint.id), eq(chatActions.kind, "inbound_wakeup"))); + const [first] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.assignedAgentId, status: "running", contextSnapshot: { issueId: conversation.issueId } }).returning(); + const origin = await initializeRunIdentity(db, { companyId: f.companyId, runId: first.id, issueId: conversation.issueId!, responsibleUserId: "owner-user", messageIds: [String(action.payload.commentId)], cause: "instruction" }); + const binding = { companyId: f.companyId, agentId: f.assignedAgentId, issueId: conversation.issueId!, runId: first.id }; + expect((await resolveConnectorAssignments(db, binding)).map(assignment => assignment.key)).toEqual(["slack"]); + await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, first.id)); + const [later] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.assignedAgentId, status: "running", runtimeMode: "native", nativeIssueId: conversation.issueId, contextSnapshot: { wakeReason: "issue_monitor_due" } }).returning(); + await initializeRunIdentity(db, { companyId: f.companyId, runId: later.id, issueId: conversation.issueId!, responsibleUserId: "owner-user", parentContextId: origin.id, cause: "monitor" }); + const continued = { ...binding, runId: later.id }; + expect((await resolveConnectorAssignments(db, continued)).map(assignment => assignment.key)).toEqual(["slack"]); + const provider = vi.fn(async (input: RequestInfo | URL) => { + const method = String(input).split("/").at(-1); + if (method === "users.info") return Response.json({ ok: true, user: { id: "UCONTINUE", team_id: "T-PAPERCLIP" } }); + if (method === "conversations.info") return Response.json({ ok: true, channel: { id: "CCONTINUE", is_member: true, is_private: false } }); + if (method === "conversations.members") return Response.json({ ok: true, members: ["UCONTINUE"] }); + if (method === "conversations.history") return Response.json({ ok: true, messages: [{ ts: "7000.1", text: "Retained channel history" }] }); + throw new Error(`Unexpected Slack method ${method}`); + }); + expect(await executeSlackTool(db, continued, "slack_history", { channel: "CCONTINUE" }, provider as typeof fetch)).toMatchObject({ messages: [{ text: "Retained channel history" }] }); + await service.revokeLink(endpoint.id, principal.id, "owner-user"); + expect(await resolveConnectorAssignments(db, continued)).toEqual([]); + await expect(executeSlackTool(db, continued, "slack_history", { channel: "CCONTINUE" }, provider as typeof fetch)).rejects.toThrow(); + }); it("binds Slack tools to the admitted linked request, recovers its provenance and revokes retained access", async () => { const { resolveConnectorAssignments, executeConnectorTool, applyConnectorSkills, prepareConnectorSkillDelivery } = await import("../services/connector-runtime.js"); const { executeSlackTool } = await import("../services/connectors/slack.js"); @@ -8327,7 +8425,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }); cliApp.use("/api", slackToolRoutes(db, "https://paperclip.example")); cliApp.use(errorHandler); - await request(cliApp).post(`/api/companies/${binding.companyId}/slack/tasks/${binding.issueId}/tools`).send({ tool: "slack_channel_info", arguments: { channel: "COTHER" } }).expect(200); + await request(cliApp).post(`/api/companies/${binding.companyId}/slack/tasks/${binding.issueId}/tools`).send({ tool: "slack_channel_info", arguments: { channel: "COTHER" } }).expect(response => expect(response.status, response.text).toBe(200)); await request(cliApp).post(`/api/companies/${randomUUID()}/slack/tasks/${binding.issueId}/tools`).send({ tool: "slack_history", arguments: { channel: "COTHER" } }).expect(403); await request(cliApp).post(`/api/companies/${binding.companyId}/slack/tasks/${randomUUID()}/tools`).send({ tool: "slack_history", arguments: { channel: "COTHER" } }).expect(403); await request(cliApp).post(`/api/companies/${binding.companyId}/slack/tasks/${binding.issueId}/tools`).send({ tool: "slack_history", arguments: { channel: "COTHER", token: "forged" } }).expect(400); diff --git a/server/src/__tests__/voice-session-store.integration.test.ts b/server/src/__tests__/voice-session-store.integration.test.ts index 276bdb353f4..4b1d0dbf172 100644 --- a/server/src/__tests__/voice-session-store.integration.test.ts +++ b/server/src/__tests__/voice-session-store.integration.test.ts @@ -1301,8 +1301,14 @@ const support = await getEmbeddedPostgresTestSupport(); expect(removed.paperclipRuntimeSkills).toEqual([]); }); - it("makes call_my_phone discoverable through the governed gateway and reuses a durable attempt", async () => { + it.each([false, true])("makes call_my_phone discoverable through the governed gateway and reuses a durable attempt (continued: %s)", async continued => { const f = await phoneToolFixture(); + if (continued) { + await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId)); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running", nativeIssueId: f.issueId, contextSnapshot: {} }).returning(); + await initializeRunIdentity(db, { companyId: f.companyId, agentId: f.agentId, runId: run.id, issueId: f.issueId, responsibleUserId: f.callerId, parentRunId: f.runId, cause: "monitor" }); + f.gatewaySession = await f.gateway.createSession({ companyId: f.companyId, agentId: f.agentId, runId: run.id }); + } const tools = await f.gateway.listToolsForSession(f.gatewaySession.token); const tool = tools.find(tool => tool.providerType === "paperclip_speko_voice"); expect(tool).toMatchObject({ upstreamToolName: "call_my_phone", connectionId: f.connectionId }); diff --git a/server/src/services/chat-channels/run-origin.test.ts b/server/src/services/chat-channels/run-origin.test.ts new file mode 100644 index 00000000000..6289333d6dc --- /dev/null +++ b/server/src/services/chat-channels/run-origin.test.ts @@ -0,0 +1,130 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agents, companies, createDb, heartbeatRuns, issueComments, issues, runIdentityContexts } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { boundRunIssueId, boundRunIssueIdSql } from "../run-task-binding.js"; +import { chatRunSourceMessageId } from "./run-origin.js"; + +const support = await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("channel task provenance", () => { + let database: Awaited>; + let db: ReturnType; + + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("paperclip-channel-origin-"); + db = createDb(database.connectionString); + }, 30_000); + afterAll(async () => { await database?.cleanup(); }, 60_000); + + async function fixture() { + const [company] = await db.insert(companies).values({ name: "Channel continuation", issuePrefix: randomUUID().slice(0, 8) }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Channel agent", role: "general", adapterType: "codex_local" }).returning(); + const [issue] = await db.insert(issues).values({ companyId: company.id, title: "Channel task", assigneeAgentId: agent.id }).returning(); + const [source, continued] = await db.insert(heartbeatRuns).values([ + { companyId: company.id, agentId: agent.id, status: "succeeded", contextSnapshot: { issueId: issue.id }, responsibleUserId: "requester" }, + { companyId: company.id, agentId: agent.id, status: "running", nativeIssueId: issue.id, contextSnapshot: {}, responsibleUserId: "requester" }, + ]).returning(); + const messageId = randomUUID(); + const [origin] = await db.insert(runIdentityContexts).values({ companyId: company.id, runId: source.id, revision: 1, responsibleUserId: "requester", messageId, cause: "instruction", correlationId: "message" }).returning(); + const [context] = await db.insert(runIdentityContexts).values({ companyId: company.id, runId: continued.id, revision: 1, responsibleUserId: "requester", parentContextId: origin.id, cause: "monitor", correlationId: "dispatch" }).returning(); + const [run] = await db.update(heartbeatRuns).set({ activeIdentityContextId: context.id }).where(eq(heartbeatRuns.id, continued.id)).returning(); + return { company, agent, issue, source, run, origin, context, messageId }; + } + + it("recovers the accepted message across repeated native and legacy task continuations", async () => { + const f = await fixture(); + expect(await chatRunSourceMessageId(db, f.run)).toBe(f.messageId); + const [next] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, status: "running", issueId: f.issue.id, responsibleUserId: "requester" }).returning(); + const [context] = await db.insert(runIdentityContexts).values({ companyId: f.company.id, runId: next.id, revision: 1, responsibleUserId: "requester", parentContextId: f.context.id, cause: "retry", correlationId: "dispatch" }).returning(); + const [run] = await db.update(heartbeatRuns).set({ activeIdentityContextId: context.id }).where(eq(heartbeatRuns.id, next.id)).returning(); + expect(await chatRunSourceMessageId(db, run)).toBe(f.messageId); + }); + + it("keeps valid channel authority after more than 100 continuations", async () => { + const f = await fixture(); + const runs = Array.from({ length: 120 }, () => ({ id: randomUUID(), companyId: f.company.id, agentId: f.agent.id, status: "succeeded", nativeIssueId: f.issue.id, responsibleUserId: "requester" })); + await db.insert(heartbeatRuns).values(runs); + const ids = runs.map(() => randomUUID()); + await db.insert(runIdentityContexts).values(runs.map((run, index) => ({ id: ids[index], companyId: f.company.id, runId: run.id, revision: 1, responsibleUserId: "requester", parentContextId: index ? ids[index - 1] : f.context.id, cause: "monitor", correlationId: "dispatch" }))); + const [run] = await db.update(heartbeatRuns).set({ activeIdentityContextId: ids.at(-1)! }).where(eq(heartbeatRuns.id, runs.at(-1)!.id)).returning(); + expect(await chatRunSourceMessageId(db, run)).toBe(f.messageId); + }); + + it("stops at a new accepted message instead of borrowing an earlier person's origin", async () => { + const f = await fixture(); + const messageId = randomUUID(); + await db.update(runIdentityContexts).set({ messageId, responsibleUserId: "new-requester" }).where(eq(runIdentityContexts.id, f.context.id)); + const [run] = await db.update(heartbeatRuns).set({ responsibleUserId: "new-requester", contextSnapshot: { wakeCommentId: f.messageId } }).where(eq(heartbeatRuns.id, f.run.id)).returning(); + expect(await chatRunSourceMessageId(db, run, { allowLegacyWakeComment: true })).toBe(messageId); + }); + + it("recovers a legacy wake only through a verified same-requester parent run", async () => { + const f = await fixture(); + await db.update(runIdentityContexts).set({ messageId: null }).where(eq(runIdentityContexts.id, f.origin.id)); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issue.id, wakeCommentId: f.messageId } }).where(eq(heartbeatRuns.id, f.source.id)); + expect(await chatRunSourceMessageId(db, f.run)).toBeNull(); + expect(await chatRunSourceMessageId(db, f.run, { allowLegacyWakeComment: true })).toBe(f.messageId); + await db.update(heartbeatRuns).set({ responsibleUserId: "another-person" }).where(eq(heartbeatRuns.id, f.source.id)); + expect(await chatRunSourceMessageId(db, f.run, { allowLegacyWakeComment: true })).toBeNull(); + }); + + it("does not borrow GitHub origin from an earlier requester after a board message", async () => { + const f = await fixture(); + const [message] = await db.insert(issueComments).values({ companyId: f.company.id, issueId: f.issue.id, authorType: "user", authorUserId: "new-requester", body: "Continue" }).returning(); + await db.update(runIdentityContexts).set({ messageId: message.id, responsibleUserId: "new-requester" }).where(eq(runIdentityContexts.id, f.context.id)); + const [run] = await db.update(heartbeatRuns).set({ responsibleUserId: "new-requester", contextSnapshot: { wakeCommentId: f.messageId } }).where(eq(heartbeatRuns.id, f.run.id)).returning(); + expect(await chatRunSourceMessageId(db, run, { provider: "github", allowLegacyWakeComment: true })).toBeNull(); + }); + + it("recovers a legacy GitHub source after a verified same-user board message", async () => { + const f = await fixture(); + const [message] = await db.insert(issueComments).values({ companyId: f.company.id, issueId: f.issue.id, authorType: "user", authorUserId: "requester", body: "Continue" }).returning(); + await db.update(runIdentityContexts).set({ messageId: null }).where(eq(runIdentityContexts.id, f.origin.id)); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issue.id, wakeCommentId: f.messageId } }).where(eq(heartbeatRuns.id, f.source.id)); + await db.update(runIdentityContexts).set({ messageId: message.id }).where(eq(runIdentityContexts.id, f.context.id)); + const [run] = await db.update(heartbeatRuns).set({ contextSnapshot: { wakeCommentId: message.id } }).where(eq(heartbeatRuns.id, f.run.id)).returning(); + expect(await chatRunSourceMessageId(db, run, { provider: "github", allowLegacyWakeComment: true })).toBe(f.messageId); + await db.update(heartbeatRuns).set({ responsibleUserId: "another-person" }).where(eq(heartbeatRuns.id, f.source.id)); + expect(await chatRunSourceMessageId(db, run, { provider: "github", allowLegacyWakeComment: true })).toBeNull(); + }); + + it.each(["task", "agent", "company", "requester", "missing-parent", "pending-parent", "pending-current", "cycle"])( + "rejects a %s break in inherited authority", async fault => { + const f = await fixture(); + if (fault === "task") { + const [other] = await db.insert(issues).values({ companyId: f.company.id, title: "Other task" }).returning(); + const [otherRun] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, status: "succeeded", nativeIssueId: other.id }).returning(); + await db.update(runIdentityContexts).set({ runId: otherRun.id }).where(eq(runIdentityContexts.id, f.origin.id)); + } + if (fault === "agent") { + const [other] = await db.insert(agents).values({ companyId: f.company.id, name: "Other agent", role: "general", adapterType: "codex_local" }).returning(); + await db.update(heartbeatRuns).set({ agentId: other.id }).where(eq(heartbeatRuns.id, f.source.id)); + } + if (fault === "company") { + const [other] = await db.insert(companies).values({ name: "Other company", issuePrefix: randomUUID().slice(0, 8) }).returning(); + await db.update(runIdentityContexts).set({ companyId: other.id }).where(eq(runIdentityContexts.id, f.origin.id)); + } + if (fault === "requester") await db.update(runIdentityContexts).set({ responsibleUserId: "another-person" }).where(eq(runIdentityContexts.id, f.origin.id)); + if (fault === "missing-parent") await db.update(runIdentityContexts).set({ parentContextId: randomUUID() }).where(eq(runIdentityContexts.id, f.context.id)); + if (fault === "pending-parent") await db.update(runIdentityContexts).set({ status: "pending" }).where(eq(runIdentityContexts.id, f.origin.id)); + if (fault === "pending-current") await db.update(runIdentityContexts).set({ status: "pending" }).where(eq(runIdentityContexts.id, f.context.id)); + if (fault === "cycle") await db.update(runIdentityContexts).set({ messageId: null, parentContextId: f.context.id }).where(eq(runIdentityContexts.id, f.origin.id)); + await expect(chatRunSourceMessageId(db, f.run)).rejects.toMatchObject({ status: 403 }); + }, + ); + + it.each(["native", "issue-column", "legacy-issue", "legacy-task"])("uses the same %s task binding in SQL and runtime authority", async source => { + const f = await fixture(); + const values = { + nativeIssueId: source === "native" ? f.issue.id : null, + issueId: source === "issue-column" ? f.issue.id : null, + contextSnapshot: source === "legacy-task" ? { issueId: "", taskId: f.issue.id } + : { issueId: source === "native" ? randomUUID() : source === "issue-column" ? [] : f.issue.id }, + }; + const [run] = await db.update(heartbeatRuns).set(values).where(eq(heartbeatRuns.id, f.run.id)).returning(); + const [projection] = await db.select({ issueId: boundRunIssueIdSql() }).from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + expect(projection.issueId).toBe(f.issue.id); + expect(boundRunIssueId(run)).toBe(projection.issueId); + }); +}); diff --git a/server/src/services/chat-channels/run-origin.ts b/server/src/services/chat-channels/run-origin.ts new file mode 100644 index 00000000000..f73f9dca22c --- /dev/null +++ b/server/src/services/chat-channels/run-origin.ts @@ -0,0 +1,83 @@ +import { and, eq, isNull, or, sql } from "drizzle-orm"; +import { chatConversations, chatEndpoints, chatMessageLinks, heartbeatRuns, issueComments, runIdentityContexts, type Db } from "@paperclipai/db"; +import type { ChatProvider } from "@paperclipai/shared"; +import { forbidden } from "../../errors.js"; +import { boundRunIssueId, boundRunIssueIdSql } from "../run-task-binding.js"; + +type Run = Pick; +type Context = typeof runIdentityContexts.$inferSelect; + +export async function chatRunSourceMessageId(db: Db, run: Run, options: { context?: Context | null; allowLegacyWakeComment?: boolean; provider?: ChatProvider } = {}): Promise { + let legacyMessageId = options.allowLegacyWakeComment && typeof run.contextSnapshot?.wakeCommentId === "string" + ? run.contextSnapshot.wakeCommentId : null; + if (!run.activeIdentityContextId) { + return legacyMessageId; + } + let origin = options.context ?? (await db.select().from(runIdentityContexts).where(and( + eq(runIdentityContexts.id, run.activeIdentityContextId), + eq(runIdentityContexts.runId, run.id), + eq(runIdentityContexts.companyId, run.companyId), + eq(runIdentityContexts.status, "accepted"), + )).limit(1))[0]; + if (!origin || origin.id !== run.activeIdentityContextId || origin.runId !== run.id || + origin.companyId !== run.companyId || origin.status !== "accepted" || + origin.responsibleUserId !== run.responsibleUserId) { + throw forbidden("Chat requester provenance changed"); + } + const seen = new Set(); + let acceptedMessageSeen = false; + while (true) { + if (origin.messageId) { + acceptedMessageSeen = true; + if (!options.provider) return origin.messageId; + const [link] = await db.select({ issueId: chatConversations.issueId, provider: chatEndpoints.provider }).from(chatMessageLinks) + .innerJoin(chatConversations, and(eq(chatConversations.id, chatMessageLinks.conversationId), eq(chatConversations.companyId, run.companyId))) + .innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatConversations.endpointId), eq(chatEndpoints.companyId, run.companyId))) + .where(and(eq(chatMessageLinks.companyId, run.companyId), eq(chatMessageLinks.commentId, origin.messageId), eq(chatMessageLinks.direction, "inbound"))).limit(1); + if (link) { + if (link.issueId !== boundRunIssueId(run)) throw forbidden("Chat request belongs to another task"); + if (link.provider !== options.provider) return null; + return origin.messageId; + } + const [message] = await db.select({ authorUserId: issueComments.authorUserId }).from(issueComments).where(and( + eq(issueComments.id, origin.messageId), eq(issueComments.companyId, run.companyId), + eq(issueComments.issueId, boundRunIssueId(run) ?? "00000000-0000-0000-0000-000000000000"), + or(eq(issueComments.authorType, "user"), and(isNull(issueComments.authorType), isNull(issueComments.authorAgentId))), + isNull(issueComments.deletedAt), + )).limit(1); + if (!run.responsibleUserId || message?.authorUserId !== run.responsibleUserId) throw forbidden("Chat requester provenance changed"); + // This verified board message may continue its same-user channel history, + // but its own wake metadata is not the earlier channel request. + legacyMessageId = null; + } + if (!origin.parentContextId) return legacyMessageId; + if (seen.has(origin.id)) throw forbidden("Chat source provenance could not be verified"); + seen.add(origin.id); + const [parent] = await db.select().from(runIdentityContexts).where(and( + eq(runIdentityContexts.id, origin.parentContextId), + eq(runIdentityContexts.companyId, run.companyId), + eq(runIdentityContexts.status, "accepted"), + )).limit(1); + if (!parent) throw forbidden("Chat requester provenance changed"); + if (parent.responsibleUserId !== run.responsibleUserId) { + if (acceptedMessageSeen) return null; + throw forbidden("Chat requester provenance changed"); + } + const [sourceRun] = await db.select({ + issueId: boundRunIssueIdSql(), + responsibleUserId: heartbeatRuns.responsibleUserId, + wakeCommentId: sql`${heartbeatRuns.contextSnapshot}->'wakeCommentId'`, + }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, parent.runId), + eq(heartbeatRuns.companyId, run.companyId), + eq(heartbeatRuns.agentId, run.agentId), + )).limit(1); + if (!sourceRun || !boundRunIssueId(run) || sourceRun.issueId !== boundRunIssueId(run)) { + throw forbidden("Chat tools cannot be inherited by another task"); + } + if (options.allowLegacyWakeComment && sourceRun.responsibleUserId === run.responsibleUserId && + typeof sourceRun.wakeCommentId === "string") legacyMessageId ??= sourceRun.wakeCommentId; + origin = parent; + } +} diff --git a/server/src/services/chat-channels/slack/task-authority.ts b/server/src/services/chat-channels/slack/task-authority.ts index 4b4a086ad12..dab73c8951e 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 { boundRunIssueId } from "../../run-task-binding.js"; import type { createChatCredentials } from "../credentials.js"; import { slackSearchActionToken } from "../../connectors/slack-search-context.js"; @@ -58,7 +59,7 @@ export function createSlackTaskAuthorityRegistration(db: Db, options: SlackTaskA throw forbidden("Chat connectors are disabled"); const identity = await measureRunPhase("connectors.slack.identity", () => slackRunOrigin(db, binding)); const run = identity.run; - if ((run.contextSnapshot?.issueId ?? run.contextSnapshot?.taskId) !== binding.issueId || !run.responsibleUserId) + if (boundRunIssueId(run) !== binding.issueId || !run.responsibleUserId) throw forbidden("Slack tools require the bound task and an accepted linked user"); const resolved = await measureRunTransaction(db, "connectors.slack.authority", async (tx) => { const [action] = await tx.select().from(chatActions).where(and( diff --git a/server/src/services/chat-github-response-comments.ts b/server/src/services/chat-github-response-comments.ts index a3b97356b87..a20877fe3e0 100644 --- a/server/src/services/chat-github-response-comments.ts +++ b/server/src/services/chat-github-response-comments.ts @@ -15,6 +15,8 @@ import { projectSafeChatPublicationText } from "./chat-publication-projection.js import type { GitHubAutomaticEventContext } from "@paperclipai/shared"; import { githubRunReplyState } from "./chat-run-publications.js"; import { logActivity } from "./activity-log.js"; +import { boundRunIssueId } from "./run-task-binding.js"; +import { chatRunSourceMessageId } from "./chat-channels/run-origin.js"; type Transaction = Parameters[0]>[0]; type Api = { request(path: string, options?: { method?: string; body?: unknown }): Promise }; @@ -220,8 +222,9 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) { } async function requestsForRun(input: { companyId: string; endpointId: string; issueId: string; runId: string }) { const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId))); - if (!run || (run.nativeIssueId ?? run.contextSnapshot?.issueId) !== input.issueId) throw forbidden("GitHub run has no request binding"); - const ids = [...new Set([run.contextSnapshot?.wakeCommentId, ...(Array.isArray(run.contextSnapshot?.wakeCommentIds) ? run.contextSnapshot.wakeCommentIds : [])].filter((id): id is string => typeof id === "string"))]; + if (!run || boundRunIssueId(run) !== input.issueId) throw forbidden("GitHub run has no request binding"); + const originMessageId = await chatRunSourceMessageId(db, run, { allowLegacyWakeComment: true, provider: "github" }); + const ids = [...new Set([originMessageId, run.contextSnapshot?.wakeCommentId, ...(Array.isArray(run.contextSnapshot?.wakeCommentIds) ? run.contextSnapshot.wakeCommentIds : [])].filter((id): id is string => typeof id === "string"))]; if (!ids.length) throw forbidden("GitHub run has no request binding"); const links = await db.select({ deliveryId: chatMessageLinks.deliveryId }).from(chatMessageLinks) .innerJoin(chatConversations, and(eq(chatConversations.id, chatMessageLinks.conversationId), eq(chatConversations.companyId, input.companyId), @@ -274,14 +277,20 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) { let receipt: { id: string; url: string } | undefined; for (const link of links) { if (!link.deliveryId) continue; + const responseForRequest = async () => (await db.select({ result: chatActions.result }).from(chatActions).where(and( + eq(chatActions.companyId, input.companyId), eq(chatActions.endpointId, input.endpointId), + eq(chatActions.deliveryId, link.deliveryId!), eq(chatActions.kind, "github_response_comment"), + )).limit(1))[0]; + if ((await responseForRequest())?.result?.final === true) continue; const assertRun = async () => { const source = await context(link.deliveryId!); const [current] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, input.companyId))); if (!current || current.agentId !== source.endpoint.assignedAgentId || !["succeeded", "failed", "interrupted", "timed_out", "cancelled"].includes(current.status) || source.endpoint.id !== input.endpointId || source.conversation.issueId !== input.issueId || - (current.nativeIssueId ?? current.contextSnapshot?.issueId) !== input.issueId) + boundRunIssueId(current) !== input.issueId) throw forbidden("GitHub failure authority changed"); + if ((await responseForRequest())?.result?.final === true) throw forbidden("GitHub request already has a final response"); if (await githubRunReplyState(db, input) !== "none") throw conflict("GitHub reply is still being resolved"); }; receipt = await acknowledge(link.deliveryId, input.body, assertRun); diff --git a/server/src/services/chat-github-reviews.ts b/server/src/services/chat-github-reviews.ts index 9367e02f1b3..b187982bf70 100644 --- a/server/src/services/chat-github-reviews.ts +++ b/server/src/services/chat-github-reviews.ts @@ -1,5 +1,7 @@ import { githubRunReplyState } from "./chat-run-publications.js"; import { stageReceiptReactionRemovals } from "./chat-receipt-reactions.js"; +import { chatRunSourceMessageId } from "./chat-channels/run-origin.js"; +import { boundRunIssueId } from "./run-task-binding.js"; import { GitHubPublicationLeaseLost, withGitHubPublicationLease, @@ -144,10 +146,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { eq(heartbeatRuns.agentId, session.agentId), ), ); - const taskId = - run?.nativeIssueId ?? - run?.contextSnapshot?.issueId ?? - run?.contextSnapshot?.taskId; + const taskId = run ? boundRunIssueId(run) : null; if ( !run || taskId !== session.issueId || @@ -155,7 +154,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { (publishing && !["running", "succeeded"].includes(run.status)) ) throw forbidden("This task run is no longer authorized"); - const commentId = run.contextSnapshot?.wakeCommentId; + const commentId = await chatRunSourceMessageId(db, run, { allowLegacyWakeComment: true, provider: "github" }); const [source] = typeof commentId === "string" ? await db diff --git a/server/src/services/chat-github-tools.ts b/server/src/services/chat-github-tools.ts index 4c4c8f73583..f87a9f1c663 100644 --- a/server/src/services/chat-github-tools.ts +++ b/server/src/services/chat-github-tools.ts @@ -1,7 +1,7 @@ import { createHash } from "node:crypto"; import { z } from "zod"; import { githubReviewAssessmentSchema } from "@paperclipai/shared"; -import { and, eq, isNull } from "drizzle-orm"; +import { and, eq, isNull, sql } from "drizzle-orm"; import { chatDeliveries, chatMessageLinks, @@ -21,7 +21,8 @@ import type { ToolGatewayDescriptor, ToolGatewaySession, } from "./tool-gateway.js"; -import { runContextSnapshotString } from "./tool-access-policy.js"; +import { boundRunIssueIdSql } from "./run-task-binding.js"; +import { chatRunSourceMessageId } from "./chat-channels/run-origin.js"; const objectSchema = ( properties: Record, @@ -333,8 +334,7 @@ export async function githubBotConnectionIdsForRun( ): Promise> { const [run] = await db .select({ - issueId: heartbeatRuns.nativeIssueId, - snapshotIssueId: runContextSnapshotString("issueId"), + issueId: boundRunIssueIdSql(), }) .from(heartbeatRuns) .where( @@ -344,7 +344,7 @@ export async function githubBotConnectionIdsForRun( eq(heartbeatRuns.id, runId), ), ); - const issueId = run?.issueId ?? run?.snapshotIssueId; + const issueId = run?.issueId; if (typeof issueId !== "string") return new Set(); const rows = await db .select({ @@ -386,7 +386,16 @@ export async function githubGuestBotConnectionForSession( ): Promise { if (!session.runId || !session.agentId || !session.issueId) return null; const [run] = await db - .select({ wakeCommentId: runContextSnapshotString("wakeCommentId") }) + .select({ + id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, + responsibleUserId: heartbeatRuns.responsibleUserId, + activeIdentityContextId: heartbeatRuns.activeIdentityContextId, + nativeIssueId: heartbeatRuns.nativeIssueId, + issueId: boundRunIssueIdSql(), + contextSnapshot: sql>`jsonb_build_object('wakeCommentId', ${heartbeatRuns.contextSnapshot}->'wakeCommentId')`, + }) .from(heartbeatRuns) .where( and( @@ -395,7 +404,7 @@ export async function githubGuestBotConnectionForSession( eq(heartbeatRuns.id, session.runId), ), ); - const commentId = run?.wakeCommentId; + const commentId = run ? await chatRunSourceMessageId(db, run, { allowLegacyWakeComment: true, provider: "github" }) : null; if (typeof commentId !== "string") return null; const [source] = await db .select({ diff --git a/server/src/services/connectors/slack-authority.ts b/server/src/services/connectors/slack-authority.ts index 486b37ab62f..06da6455d35 100644 --- a/server/src/services/connectors/slack-authority.ts +++ b/server/src/services/connectors/slack-authority.ts @@ -10,6 +10,8 @@ import { import { and, eq, inArray } from "drizzle-orm"; import { captureRunIdentity } from "../run-identity.js"; import { forbidden } from "../../errors.js"; +import { boundRunIssueId } from "../run-task-binding.js"; +import { chatRunSourceMessageId } from "../chat-channels/run-origin.js"; export interface SlackTaskBinding { companyId: string; @@ -149,51 +151,13 @@ export async function slackRunOrigin(db: Db, binding: SlackTaskBinding) { ); } if ( - (captured.run.contextSnapshot?.issueId ?? - captured.run.contextSnapshot?.taskId) !== binding.issueId || + boundRunIssueId(captured.run) !== binding.issueId || !captured.run.responsibleUserId ) throw forbidden( "Slack tools require the bound task and an accepted linked user", ); - let origin = captured.context; - const seen = new Set(); - while (origin && !origin.messageId && origin.parentContextId) { - if (seen.has(origin.id) || seen.size >= 100) - throw forbidden("Slack source provenance could not be verified"); - seen.add(origin.id); - const [parent] = await db - .select() - .from(runIdentityContexts) - .where( - and( - eq(runIdentityContexts.id, origin.parentContextId), - eq(runIdentityContexts.companyId, binding.companyId), - eq(runIdentityContexts.status, "accepted"), - ), - ); - if (!parent || parent.responsibleUserId !== captured.run.responsibleUserId) - throw forbidden("Slack requester provenance changed"); - // Delegation to a different task does not grant that task Slack tools. - const [sourceRun] = await db - .select() - .from(heartbeatRuns) - .where( - and( - eq(heartbeatRuns.id, parent.runId), - eq(heartbeatRuns.companyId, binding.companyId), - eq(heartbeatRuns.agentId, binding.agentId), - ), - ); - if ( - !sourceRun || - (sourceRun.contextSnapshot?.issueId ?? - sourceRun.contextSnapshot?.taskId) !== binding.issueId - ) - throw forbidden("Slack tools cannot be inherited by another task"); - origin = parent; - } - return { ...captured, sourceMessageId: origin?.messageId ?? null }; + return { ...captured, sourceMessageId: await chatRunSourceMessageId(db, captured.run, { context: captured.context }) }; } /** Enumerate only this agent's connections; each still needs live run/user authority. */ diff --git a/server/src/services/heartbeat/run-preparation.ts b/server/src/services/heartbeat/run-preparation.ts index 3a41c987dbb..374412e1276 100644 --- a/server/src/services/heartbeat/run-preparation.ts +++ b/server/src/services/heartbeat/run-preparation.ts @@ -6,6 +6,7 @@ import { CONFIGURED_ENVIRONMENT_KEYS } from "../../vendor/paperclip-runner/index import { ASSIGNED_MCP_SERVER_NAME } from "../mcp-tool-names.js"; import { externalConversationStateSql } from "../slack-conversation-state.js"; import { githubBotConnectionIdsForRun } from "../chat-github-tools.js"; +import { boundRunIssueIdSql } from "../run-task-binding.js"; import { isBrowserUseConnection } from "../browser-use-client.js"; import { explicitOperatorRunIdentity } from "../run-identity.js"; import { @@ -935,7 +936,7 @@ export async function buildPaperclipRuntimeMcpServers(input: { ); const [runIdentity] = await input.db .select({ - contextSnapshot: heartbeatRuns.contextSnapshot, + issueId: boundRunIssueIdSql(), responsibleUserId: heartbeatRuns.responsibleUserId, activeIdentityContextId: heartbeatRuns.activeIdentityContextId, }) @@ -998,7 +999,7 @@ export async function buildPaperclipRuntimeMcpServers(input: { companyId: input.agent.companyId, agentId: input.agent.id, runId: input.runId, - issueId: readNonEmptyString(runIdentity?.contextSnapshot?.issueId ?? runIdentity?.contextSnapshot?.taskId), + issueId: runIdentity?.issueId ?? null, identityContextId: runIdentity?.activeIdentityContextId, })).map(tool => tool.connectionId)) : new Set(); diff --git a/server/src/services/native-runtime/native-session-resume.ts b/server/src/services/native-runtime/native-session-resume.ts index c8d8c031d28..3243de9980e 100644 --- a/server/src/services/native-runtime/native-session-resume.ts +++ b/server/src/services/native-runtime/native-session-resume.ts @@ -33,7 +33,7 @@ export function nativeToolContractFingerprintForTarget( return `sha256:${createHash("sha256") .update( JSON.stringify({ - schema: "paperclip.native-tool-contract.v16", + schema: "paperclip.native-tool-contract.v17", executionTargetKind, advertisementPolicy: { // Direct provider threads retain declarations from thread/start. @@ -48,6 +48,7 @@ export function nativeToolContractFingerprintForTarget( "always_advertised_run_issue_agent_binding_gated_current_task_description.v2", semanticCompletion: "finish_accessible_deliverable_evidence.v4", connectorTools: "assigned_resources_and_pinned_skill_bundle.v1", + chatTaskTools: "durable_task_binding_and_accepted_origin.v1", githubResponses: "bound_working_comment_progress_and_final.v1", githubInstructionSkills: "trusted_configuration_native_selection.v1", ...(executionTargetKind === "remote" diff --git a/server/src/services/native-runtime/runtime-context.test.ts b/server/src/services/native-runtime/runtime-context.test.ts index d0d81c292b9..8fd5e101e8b 100644 --- a/server/src/services/native-runtime/runtime-context.test.ts +++ b/server/src/services/native-runtime/runtime-context.test.ts @@ -428,7 +428,7 @@ it("pins channel tools only when the current task run is bound to that bot", asy it("pins a voice tool only when the current requester can call through that assigned endpoint", async () => { serviceMocks.getEffectiveProfilesForAgent.mockResolvedValue({ entries: [{ effect: "include", connectionId: "voice-connection" }], installedConnections: [{ id: "voice-connection", status: "active", enabled: true, healthStatus: "healthy", transport: "voice", config: { provider: "speko" }, transportConfig: {} }], allowedTools: [{ id: "phone-tool", connectionId: "voice-connection" }] }); - const limit = vi.fn().mockResolvedValue([{ contextSnapshot: { issueId: "task-1" }, activeIdentityContextId: "identity-1" }]); + const limit = vi.fn().mockResolvedValue([{ issueId: "task-1", activeIdentityContextId: "identity-1" }]); const db = { select: () => ({ from: () => ({ where: () => ({ limit }) }) }) } as unknown as Db; const input = { db, agent: { id: "agent-1", companyId: "company-1" }, runId: "run-1" }; expect((await resolveNativeRuntimeMcpSnapshot(input)).bindingId).toBeNull(); diff --git a/server/src/services/native-runtime/runtime-context.ts b/server/src/services/native-runtime/runtime-context.ts index edbfd890913..8c561aebbf2 100644 --- a/server/src/services/native-runtime/runtime-context.ts +++ b/server/src/services/native-runtime/runtime-context.ts @@ -1,6 +1,7 @@ import { measureRunPhase } from "../execution-preparation-timing.js"; import { spekoToolsForSession } from "../voice/speko-agent-tools.js"; import { githubBotConnectionIdsForRun } from "../chat-github-tools.js"; +import { boundRunIssueIdSql } from "../run-task-binding.js"; import { isBrowserUseConnection } from "../browser-use-client.js"; import { createHash, randomUUID } from "node:crypto"; import fs from "node:fs/promises"; @@ -205,7 +206,7 @@ export async function resolveNativeRuntimeMcpSnapshot(input: { db: Db; agent: Pi }); const hasVoiceConnection = effective.installedConnections.some(connection => connection.transport === "voice"); const [runIdentity] = hasGitHubConnection || hasVoiceConnection - ? await input.db.select({ contextSnapshot: heartbeatRuns.contextSnapshot, responsibleUserId: heartbeatRuns.responsibleUserId, activeIdentityContextId: heartbeatRuns.activeIdentityContextId }) + ? await input.db.select({ issueId: boundRunIssueIdSql(), responsibleUserId: heartbeatRuns.responsibleUserId, activeIdentityContextId: heartbeatRuns.activeIdentityContextId }) .from(heartbeatRuns) .where(and( eq(heartbeatRuns.id, input.runId), @@ -227,7 +228,7 @@ export async function resolveNativeRuntimeMcpSnapshot(input: { db: Db; agent: Pi // App access is optional runtime context. Keep usable assignments pinned, but // do not stop unrelated work because an assigned app needs attention. const githubBotConnectionIds = await githubBotConnectionIdsForRun(input.db, input.agent.companyId, input.agent.id, input.runId); - const spekoConnectionIds = hasVoiceConnection ? new Set((await spekoToolsForSession(input.db, { companyId: input.agent.companyId, agentId: input.agent.id, runId: input.runId, issueId: typeof runIdentity?.contextSnapshot?.issueId === "string" ? runIdentity.contextSnapshot.issueId : typeof runIdentity?.contextSnapshot?.taskId === "string" ? runIdentity.contextSnapshot.taskId : null, identityContextId: runIdentity?.activeIdentityContextId })).map(tool => tool.connectionId)) : new Set(); + const spekoConnectionIds = hasVoiceConnection ? new Set((await spekoToolsForSession(input.db, { companyId: input.agent.companyId, agentId: input.agent.id, runId: input.runId, issueId: runIdentity?.issueId ?? null, identityContextId: runIdentity?.activeIdentityContextId })).map(tool => tool.connectionId)) : new Set(); const availableConnectionIds = new Set(resolvedInstalledConnections.filter((connection) => permitted.has(connection.id) && connection.status === "active" diff --git a/server/src/services/run-identity.ts b/server/src/services/run-identity.ts index 496b58719d2..19ac75989ce 100644 --- a/server/src/services/run-identity.ts +++ b/server/src/services/run-identity.ts @@ -10,6 +10,7 @@ import { type Db, } from "@paperclipai/db"; import { conflict, forbidden } from "../errors.js"; +import { boundRunIssueId } from "./run-task-binding.js"; import { isUuidLike } from "@paperclipai/shared"; import { queuedCommentIdsFromRunContext, queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js"; @@ -71,19 +72,18 @@ async function lockIdentityTask( ) { const [run] = await executor .select({ - context: heartbeatRuns.contextSnapshot, - issueId: heartbeatRuns.nativeIssueId, + contextSnapshot: heartbeatRuns.contextSnapshot, + nativeIssueId: heartbeatRuns.nativeIssueId, + issueId: heartbeatRuns.issueId, }) .from(heartbeatRuns) .where( and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, companyId)), ); - const issueId = run?.context?.issueId ?? run?.context?.taskId ?? run?.issueId; - if ( - issueId !== undefined && - issueId !== null && - (typeof issueId !== "string" || !isUuidLike(issueId)) - ) { + const issueId = run?.nativeIssueId ?? run?.issueId ?? run?.contextSnapshot?.issueId ?? run?.contextSnapshot?.taskId; + const legacyIssueId = run?.contextSnapshot?.issueId ?? run?.contextSnapshot?.taskId; + if ([issueId, legacyIssueId].some(id => id !== undefined && id !== null && + (typeof id !== "string" || !isUuidLike(id)))) { throw forbidden("Run task identity is invalid"); } if (typeof issueId === "string") @@ -151,7 +151,7 @@ async function activate(executor: Executor, context: RunIdentityContext) { ), ) .returning(); - const issueId = run?.contextSnapshot?.issueId ?? run?.contextSnapshot?.taskId; + const issueId = run ? boundRunIssueId(run) : null; if (typeof issueId === "string") await executor .update(issues) @@ -319,8 +319,7 @@ export async function prepareSteeredIdentity( throw forbidden("Steering requires an authenticated message author"); if ( run.status !== "running" || - (run.contextSnapshot?.issueId !== input.issueId && - run.contextSnapshot?.taskId !== input.issueId) + boundRunIssueId(run) !== input.issueId ) { throw conflict("Steering targets a different or inactive execution"); } diff --git a/server/src/services/run-task-binding.ts b/server/src/services/run-task-binding.ts new file mode 100644 index 00000000000..67b89336bd2 --- /dev/null +++ b/server/src/services/run-task-binding.ts @@ -0,0 +1,21 @@ +import { sql } from "drizzle-orm"; +import { heartbeatRuns } from "@paperclipai/db"; + +type RunTaskBinding = { + nativeIssueId?: string | null; + issueId?: string | null; + contextSnapshot?: Record | null; +}; + +export function boundRunIssueId(run: RunTaskBinding): string | null { + const values = [run.nativeIssueId, run.issueId, run.contextSnapshot?.issueId, run.contextSnapshot?.taskId]; + return values.find((value): value is string => typeof value === "string" && value.trim().length > 0)?.trim() ?? null; +} + +export function boundRunIssueIdSql() { + return sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.issueId}::text, + case when jsonb_typeof(${heartbeatRuns.contextSnapshot}->'issueId') = 'string' + then nullif(btrim(${heartbeatRuns.contextSnapshot}->>'issueId'), '') end, + case when jsonb_typeof(${heartbeatRuns.contextSnapshot}->'taskId') = 'string' + then nullif(btrim(${heartbeatRuns.contextSnapshot}->>'taskId'), '') end)`; +} diff --git a/server/src/services/tool-access-policy.ts b/server/src/services/tool-access-policy.ts index 6357c9b5e87..05485b87c18 100644 --- a/server/src/services/tool-access-policy.ts +++ b/server/src/services/tool-access-policy.ts @@ -1,4 +1,5 @@ import { createHash } from "node:crypto"; +import { boundRunIssueIdSql } from "./run-task-binding.js"; import { and, asc, desc, eq, gt, inArray, isNull, ne, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { @@ -976,7 +977,7 @@ export function toolAccessPolicyService(db: Db) { .select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, - issueId: runContextSnapshotString("issueId"), + issueId: boundRunIssueIdSql(), projectId: runContextSnapshotString("projectId"), routineId: runContextSnapshotString("routineId"), }) diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index 393278dd4e1..95de0f8c244 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -12,6 +12,7 @@ import { claimSlackRateLimitRetry } from "./connectors/slack-retry.js"; import { resolveSlackTaskAuthority } from "./connectors/slack-authority.js"; import { SLACK_TOOLS } from "@paperclipai/shared"; import { slackToolsForSession } from "./connectors/slack-catalog.js"; +import { boundRunIssueIdSql } from "./run-task-binding.js"; import { executeSlackTool } from "./connectors/slack.js"; import { githubGuestBotConnectionForSession, githubBotToolsForSession } from "./chat-github-tools.js"; import { githubChatReviewService } from "./chat-github-reviews.js"; @@ -1664,7 +1665,7 @@ export function createToolGatewayService( companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, status: heartbeatRuns.status, - issueId: runContextSnapshotString("issueId"), + issueId: boundRunIssueIdSql(), projectId: runContextSnapshotString("projectId"), }) .from(heartbeatRuns) diff --git a/server/src/services/voice/speko-agent-tools.ts b/server/src/services/voice/speko-agent-tools.ts index 669a70a0c56..6332329fe35 100644 --- a/server/src/services/voice/speko-agent-tools.ts +++ b/server/src/services/voice/speko-agent-tools.ts @@ -3,6 +3,7 @@ import { and, eq, isNull, sql } from "drizzle-orm"; import { forbidden, HttpError } from "../../errors.js"; import { authorizationService } from "../authorization.js"; import { captureRunIdentity } from "../run-identity.js"; +import { boundRunIssueId } from "../run-task-binding.js"; import { chatVoiceCallbacks, instanceUserRoles, companyMemberships } from "@paperclipai/db"; import type { VoiceCaller } from "./voice-session-store.js"; import { z } from "zod"; @@ -19,7 +20,7 @@ async function resolveAuthority(db: Db, session: SpekoTaskSession, endpointId: s if (!(await instanceSettingsService(db).getExperimental()).enableChatConnectors) throw forbidden("Voice connections are disabled"); if (!runtimes.has(db) || !session.agentId || !session.runId || !session.issueId) throw forbidden("Speko tools require an active task run"); const { run, context } = await captureRunIdentity(db, { companyId: session.companyId, agentId: session.agentId, runId: session.runId }); - if ((run.contextSnapshot?.issueId ?? run.contextSnapshot?.taskId) !== session.issueId || !run.responsibleUserId || session.identityContextId && session.identityContextId !== context?.id) throw forbidden("The accepted requester or task changed"); + if (boundRunIssueId(run) !== session.issueId || !run.responsibleUserId || session.identityContextId && session.identityContextId !== context?.id) throw forbidden("The accepted requester or task changed"); const [endpoint] = await db.select().from(chatEndpoints).where(and(eq(chatEndpoints.companyId, session.companyId), eq(chatEndpoints.id, endpointId), eq(chatEndpoints.provider, "speko"), eq(chatEndpoints.assignedAgentId, session.agentId), eq(chatEndpoints.status, "active"))); if (!endpoint) throw forbidden("This agent has no active Speko connection"); const userId = run.responsibleUserId;