fix(chat): return governed tool outcomes to their originating thread

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-09-22 13:52:27 -05:00
1 parent 653c9ad2bf
commit bef1a0fada
7 files changed
+136 -5

No files matched your search

+15 -3
View File
@@ -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
@@ -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<typeof slackExplicitPublicationDuplicate>[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));
@@ -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);
});
});
@@ -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" } });
@@ -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
);
+47
View File
@@ -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;
+10 -2
View File
@@ -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<string>`${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,