diff --git a/doc/plans/2026-09-22-slack-origin-tools.md b/doc/plans/2026-09-22-slack-origin-tools.md index be2eacc0925..30529ecd243 100644 --- a/doc/plans/2026-09-22-slack-origin-tools.md +++ b/doc/plans/2026-09-22-slack-origin-tools.md @@ -65,9 +65,21 @@ must be reported accurately and retain bounded history search. endpoint lock race) are fixed and retested. Channel discovery subsequently found a listed channel returning `channel_not_found`; `d23bb857a` omits only definite access denials. Its unit/PG regressions, full typecheck and build pass. Deployment - and live discovery retest are in progress. -- Remaining live checks include governed actions and disabled-response channel - reads/writes. Optional OAuth, private multi-user boundaries, and CLI execution + and live discovery retest passed on `653c9ad2b`: two pages exhausted, both + accessible channels returned, and the natural-language search succeeded. +- Live disabled-response boundary passed: from the requester's bot DM, channel + reading succeeded while a write was denied. The channel's original enabled + setting was restored afterward. +- Explicit posting and editing passed without a duplicate send. Governed channel + creation executed only after approval; the new channel remained disabled for + responses and no invitations were sent. Leaf's independent signing secret was + configured through its managed environment with explicit operator permission. +- Approval continuation exposed a missing return path: its result appeared in + Paperclip but not Slack. The fix verifies the durable tool-review wake and every + referenced action's source before authorizing publication, and separates wakes + from different source runs. Authority/dedup integration and 81 gateway/response + tests pass; the live approval-result retest is pending. +- Optional OAuth, private multi-user boundaries, and CLI execution currently have automated coverage, not complete live qualification. ## Explicit runtime/provider limitations diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 0b580154069..b18efb8e0ee 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -4043,6 +4043,16 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { await expect(executeConnectorTool(db, binding, "slack_post_message", send)).resolves.toMatchObject({ status: "completed" }); await executeConnectorTool(db, binding, "slack_post_message", send); expect(postFetch.mock.calls.filter(call => String(call[0]).endsWith("/chat.postMessage"))).toHaveLength(1); + const { slackExplicitPublicationDuplicate } = await import("../services/connectors/slack-publication.js"); + const [finalComment] = await db.insert(issueComments).values({ companyId: binding.companyId, issueId: binding.issueId, authorAgentId: binding.agentId, createdByRunId: runId, body: send.text }).returning(); + const publication = { companyId: binding.companyId, issueId: binding.issueId, endpointId: endpoint.id, conversationId: conversation.id, commentId: finalComment.id, payload: { text: send.text } } as Parameters[1]; + await expect(slackExplicitPublicationDuplicate(db, publication)).resolves.toBe("delivered"); + await expect(slackExplicitPublicationDuplicate(db, { ...publication, payload: { text: "A distinct outcome summary" } })).resolves.toBeNull(); + await expect(slackExplicitPublicationDuplicate(db, { ...publication, companyId: randomUUID() })).resolves.toBeNull(); + const [explicitSend] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, endpoint.id), eq(chatActions.providerActionId, `slack-tool:${binding.issueId}:${send.idempotencyKey}`))); + await db.update(chatActions).set({ status: "uncertain" }).where(eq(chatActions.id, explicitSend.id)); + await expect(slackExplicitPublicationDuplicate(db, publication)).resolves.toBe("unresolved"); + await db.update(chatActions).set({ status: "processed" }).where(eq(chatActions.id, explicitSend.id)); await expect(executeConnectorTool(db, binding, "slack_post_message", { ...send, text: "Different operation" })).rejects.toThrow("different Slack operation"); // Definite rate-limit failures can retry after their deadline; uncertain // deliveries are reconciled without sending another message. @@ -4088,6 +4098,32 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(invocation, invocation.errorMessage ?? undefined).toMatchObject({ status: "succeeded", approvalState: "approved" }); expect(postFetch.mock.calls.filter(call => String(call[0]).endsWith("/conversations.create"))).toHaveLength(1); expect(await db.select().from(chatEndpointResources).where(and(eq(chatEndpointResources.endpointId, endpoint.id), eq(chatEndpointResources.providerResourceId, "CNEWCHANNEL"), eq(chatEndpointResources.enabled, true)))).toEqual([]); + // The governed result must return to its exact originating Slack thread, + // even though approval delivery has a different wake key from normal input. + const { toolActionDeliveryService } = await import("../services/tool-action-delivery.js"); + const { resolveChatRunPresentationAuthorizationReason, CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON } = await import("../services/chat-run-publications.js"); + await db.update(heartbeatRuns).set({ contextSnapshot: { + issueId: conversation.issueId, source: "chat:slack", endpointId: endpoint.id, + wakeCommentIds: [String(action.payload.commentId)], + } }).where(eq(heartbeatRuns.id, runId)); + let continuationRunId = ""; + const approvalWake = vi.fn(async (agentId: string, input: any) => { + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: fixture.companyId, + agentId, source: input.source, status: "completed", idempotencyKey: input.idempotencyKey, + payload: input.payload }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: fixture.companyId, + agentId, status: "succeeded", wakeupRequestId: wake.id, contextSnapshot: input.contextSnapshot }).returning(); + continuationRunId = run.id; + await db.update(agentWakeupRequests).set({ runId: run.id }).where(eq(agentWakeupRequests.id, wake.id)); + return run; + }); + await toolActionDeliveryService(db, { wakeup: approvalWake as any }).deliver(approval.id); + expect(approvalWake).toHaveBeenCalledWith(fixture.assignedAgentId, expect.objectContaining({ allowRunCoalescing: false })); + const presentation = { companyId: fixture.companyId, issueId: conversation.issueId, runId: continuationRunId }; + await expect(resolveChatRunPresentationAuthorizationReason(db, presentation)).resolves.toBe(CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON); + await expect(resolveChatRunPresentationAuthorizationReason(db, { ...presentation, companyId: randomUUID() })).resolves.toBe("internal_agent_write"); + await db.update(agentWakeupRequests).set({ payload: { ...approvalWake.mock.calls[0][1].payload, sourceRunId: randomUUID() } }).where(eq(agentWakeupRequests.runId, continuationRunId)); + await expect(resolveChatRunPresentationAuthorizationReason(db, presentation)).resolves.toBe("internal_agent_write"); globalThis.fetch = originalFetch; const recoveredRun = randomUUID(); await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, runId)); diff --git a/server/src/__tests__/heartbeat-run-summary.test.ts b/server/src/__tests__/heartbeat-run-summary.test.ts index 1b139d20f6a..5b89f61b69b 100644 --- a/server/src/__tests__/heartbeat-run-summary.test.ts +++ b/server/src/__tests__/heartbeat-run-summary.test.ts @@ -872,6 +872,7 @@ describe("resolveHeartbeatRunResponse", () => { expect(isExternalChatPresentationContext({ source: "chatty:github" })).toBe( false, ); + expect(isExternalChatPresentationContext({ source: "tool_action_review" })).toBe(true); expect(isExternalChatPresentationContext(null)).toBe(false); }); }); diff --git a/server/src/__tests__/tool-gateway-service.test.ts b/server/src/__tests__/tool-gateway-service.test.ts index 6d8420561f6..116c2f74dd0 100644 --- a/server/src/__tests__/tool-gateway-service.test.ts +++ b/server/src/__tests__/tool-gateway-service.test.ts @@ -432,6 +432,32 @@ describeEmbeddedPostgres("tool gateway service", () => { expect(native).toHaveLength(2); }); + it("keeps reviewed results from different source runs in separate non-coalescing wakes", async () => { + const { company, agent, issue, run } = await createRunFixture(db); + const [secondRun] = await db.insert(heartbeatRuns).values({ companyId: company.id, + agentId: agent.id, status: "running", contextSnapshot: { issueId: issue.id } }).returning(); + await db.insert(toolPolicies).values({ companyId: company.id, name: "Ask first", policyType: "require_approval", selectors: { toolName: "mcp-remote-fixture:update_note" } }); + const gateway = createTestToolGatewayService(db); + for (const sourceRun of [run, secondRun]) { + const session = await gateway.createSession({ companyId: company.id, agentId: agent.id, runId: sourceRun.id }); + await expect(gateway.executeTool({ sessionToken: session.token, tool: "mcp-remote-fixture:update_note", parameters: { noteId: sourceRun.id, body: "separate origin" } })).rejects.toMatchObject({ reasonCode: "approval_required" }); + } + const requests = await db.select().from(toolActionRequests); + for (const action of requests) await gateway.declineActionRequest({ companyId: company.id, issueId: issue.id, + interactionId: action.interactionId!, actionRequestId: action.id, actor: { userId: "reviewer" } }); + await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.companyId, company.id)); + const wakeup = vi.fn(async (agentId: string, input: any) => (await db.insert(agentWakeupRequests).values({ companyId: company.id, + agentId, source: input.source, idempotencyKey: input.idempotencyKey, payload: input.payload }).returning())[0] as any); + await toolActionDeliveryService(db, { wakeup }).sweepPending(); + expect(wakeup).toHaveBeenCalledTimes(2); + expect(new Set(wakeup.mock.calls.map(([, input]) => input.payload.sourceRunId))).toEqual(new Set([run.id, secondRun.id])); + for (const [, input] of wakeup.mock.calls) { + expect(input.allowRunCoalescing).toBe(false); + expect(input.payload.toolActionRequestIds).toHaveLength(1); + } + expect((await db.select().from(toolActionDeliveries)).every(row => row.deliveredAt)).toBe(true); + }); + it("bounds many outcomes and recovers their full-result reference after wake commit", async () => { const { company, agent, run } = await createRunFixture(db); await db.insert(toolPolicies).values({ companyId: company.id, name: "Ask first", policyType: "require_approval", selectors: { toolName: "mcp-remote-fixture:update_note" } }); diff --git a/server/src/services/heartbeat-run-summary.ts b/server/src/services/heartbeat-run-summary.ts index 7235929d4be..897ff29721f 100644 --- a/server/src/services/heartbeat-run-summary.ts +++ b/server/src/services/heartbeat-run-summary.ts @@ -191,6 +191,7 @@ export function isExternalChatPresentationContext( typeof context.source === "string" ? context.source.trim() : ""; return ( source.startsWith("chat:") || + source === "tool_action_review" || context.externalChatContinuation === true || wake.externalInteractionContinuation === true ); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 87a36078024..fa7a88297c7 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -64,6 +64,8 @@ import { issueReadStates, issueThreadInteractions, toolActionRequests, + toolActionDeliveries, + toolInvocations, issues, labels, projectWorkspaces, @@ -1302,6 +1304,51 @@ export async function resolveChatOriginPublicationBindings( } const contextSource = readStringFromRecord(snapshot, "source"); + if (contextSource === "tool_action_review") { + // Review results use their own durable wake, not the ordinary interaction + // response key. Attest the entire batch before following its origin; the + // model's context/sourceRunId alone never grants publication authority. + const [wake] = await dbOrTx.select({ payload: agentWakeupRequests.payload, + idempotencyKey: agentWakeupRequests.idempotencyKey }) + .from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.agentId, lineageAgentId!), + eq(agentWakeupRequests.runId, originRunId), + like(agentWakeupRequests.idempotencyKey, "tool-action-response:%"), + )).limit(1); + const requestIds = wake?.payload?.toolActionRequestIds; + const sourceRunId = readStringFromRecord(snapshot, "sourceRunId"); + if (!Array.isArray(requestIds) || requestIds.length === 0 || + requestIds.some((id: unknown) => typeof id !== "string" || !isUuidLike(id)) || + !sourceRunId || !isUuidLike(sourceRunId) || sourceRunId !== wake.payload.sourceRunId || + wake.idempotencyKey !== `tool-action-response:${requestIds[0]}` || + snapshot.interactionId !== wake.payload.interactionId) return []; + const receipts = await dbOrTx.select({ id: toolActionRequests.id }) + .from(toolActionRequests) + .innerJoin(toolInvocations, and( + eq(toolInvocations.id, toolActionRequests.invocationId), + eq(toolInvocations.companyId, companyId), + eq(toolInvocations.issueId, issueId), + eq(toolInvocations.agentId, lineageAgentId!), + eq(toolInvocations.runId, sourceRunId), + )) + .innerJoin(toolActionDeliveries, and( + eq(toolActionDeliveries.actionRequestId, toolActionRequests.id), + eq(toolActionDeliveries.companyId, companyId), + eq(toolActionDeliveries.issueId, issueId), + eq(toolActionDeliveries.interactionId, toolActionRequests.interactionId), + )) + .where(and( + eq(toolActionRequests.companyId, companyId), + eq(toolActionRequests.issueId, issueId), + eq(toolActionRequests.requestedByAgentId, lineageAgentId!), + inArray(toolActionRequests.id, requestIds), + inArray(toolActionRequests.status, ["executed", "failed", "rejected", "expired", "cancelled"]), + )); + if (receipts.length !== requestIds.length) return []; + originRunId = sourceRunId; + continue; + } if (contextSource?.startsWith("chat:")) { contextSnapshot = snapshot; break; diff --git a/server/src/services/tool-action-delivery.ts b/server/src/services/tool-action-delivery.ts index 5a2d4b7f890..63f217de9a0 100644 --- a/server/src/services/tool-action-delivery.ts +++ b/server/src/services/tool-action-delivery.ts @@ -115,7 +115,7 @@ export function toolActionDeliveryService( ) .limit(1); if (pending) return false; - const outcomes = await tx + const readyOutcomes = await tx .select({ receipt: toolActionDeliveries, receiptCreatedAtText: sql`${toolActionDeliveries.createdAt}::text`, @@ -163,7 +163,14 @@ export function toolActionDeliveryService( asc(toolActionDeliveries.createdAt), asc(toolActionDeliveries.actionRequestId), ); - if (!outcomes.length) return false; + if (!readyOutcomes.length) return false; + // A task can have reviews from different external conversations. Never + // combine their results into a wake that would publish to only one origin. + const sourceOutcome = readyOutcomes.find((row) => row.request.id === source.actionRequestId); + if (!sourceOutcome) return false; + const outcomes = readyOutcomes.filter( + (row) => row.invocation.runId === sourceOutcome.invocation.runId, + ); const sourceRunIds = outcomes.flatMap((row) => row.invocation.runId ? [row.invocation.runId] : [], ); @@ -297,6 +304,7 @@ export function toolActionDeliveryService( triggerDetail: "system", reason: "issue_commented", idempotencyKey, + allowRunCoalescing: false, issueStateGuard: { statuses: [issue.status], assigneeAgentId: agent.id,