From 04ad66b23dc3ef0ced004d91b00709d0bfd6fb5a Mon Sep 17 00:00:00 2001 From: anyech Date: Tue, 2 Jun 2026 01:43:52 +0000 Subject: [PATCH] fix(tasks): deliver ACP completions to bound Discord threads --- src/tasks/task-registry.test.ts | 144 +++++++++++++++++++++++++++++++- src/tasks/task-registry.ts | 32 ++++++- 2 files changed, 171 insertions(+), 5 deletions(-) diff --git a/src/tasks/task-registry.test.ts b/src/tasks/task-registry.test.ts index d66ea82386bf..9173fa7b7453 100644 --- a/src/tasks/task-registry.test.ts +++ b/src/tasks/task-registry.test.ts @@ -131,7 +131,7 @@ vi.mock("../agents/subagent-control.js", () => ({ vi.mock("../utils/message-channel.js", () => ({ isDeliverableMessageChannel: (channel: string) => - channel === "notifychat" || channel === "guildchat", + channel === "notifychat" || channel === "guildchat" || channel === "discord", })); function configureTaskRegistryMaintenanceRuntimeForTest(params: { @@ -1337,6 +1337,148 @@ describe("task-registry", () => { }); }); + it("delivers delegated ACP completion directly to an explicitly bound Discord thread", async () => { + await withTaskRegistryTempDir(async (root) => { + process.env.OPENCLAW_STATE_DIR = root; + resetTaskRegistryForTests(); + const runId = "run-bound-discord-thread-terminal"; + hoisted.sendMessageMock.mockResolvedValue({ + channel: "discord", + to: "channel:parent-channel", + via: "direct", + }); + + createTaskRecord({ + runtime: "acp", + ownerKey: "agent:main:discord:guild-123:channel-parent-channel", + scopeKind: "session", + requesterOrigin: { + channel: "discord", + to: "channel:parent-channel", + threadId: "thread-84022", + }, + childSessionKey: "agent:main:acp:child", + runId, + task: "Investigate thread-bound ACP delivery", + status: "running", + deliveryStatus: "pending", + terminalSummary: "ACP final answer", + startedAt: 100, + }); + + emitAgentEvent({ + runId, + stream: "lifecycle", + data: { + phase: "end", + endedAt: 250, + }, + }); + + await waitForAssertion(() => { + const task = findTaskByRunId(runId); + if (!task) { + throw new Error(`Expected task for run ${runId}`); + } + expect(task.status).toBe("succeeded"); + expect(task.deliveryStatus).toBe("delivered"); + }); + await waitForAssertion(() => expect(hoisted.sendMessageMock).toHaveBeenCalledTimes(1)); + const message = sentMessageCall(); + expectRecordFields(message, { + channel: "discord", + to: "channel:parent-channel", + threadId: "thread-84022", + }); + expect(String(message.content)).toContain( + "Background task ready for review: ACP background task", + ); + expect(String(message.content)).toContain("ACP final answer"); + expect(String(message.content)).toContain( + "Next: parent will review/verify before calling it done.", + ); + expect(peekSystemEvents("agent:main:discord:guild-123:channel-parent-channel")).toStrictEqual( + [], + ); + }); + }); + + it.each([ + { + id: "missing-thread", + requesterOrigin: { + channel: "discord", + to: "channel:parent-channel", + }, + }, + { + id: "non-channel-target", + requesterOrigin: { + channel: "discord", + to: "user:U123", + threadId: "thread-84022", + }, + }, + { + id: "non-discord-channel", + requesterOrigin: { + channel: "guildchat", + to: "guildchat:channel:parent-channel", + threadId: "thread-84022", + }, + }, + ])( + "keeps delegated ACP completion queued without an explicit bound Discord thread ($id)", + async ({ requesterOrigin }) => { + await withTaskRegistryTempDir(async (root) => { + process.env.OPENCLAW_STATE_DIR = root; + resetTaskRegistryForTests(); + const runId = `run-non-bound-discord-thread-terminal-${requesterOrigin.channel}-${requesterOrigin.to}`; + hoisted.sendMessageMock.mockResolvedValue({ + channel: requesterOrigin.channel, + to: requesterOrigin.to, + via: "direct", + }); + + createTaskRecord({ + runtime: "acp", + ownerKey: "agent:main:discord:guild-123:channel-parent-channel", + scopeKind: "session", + requesterOrigin, + childSessionKey: "agent:main:acp:child", + runId, + task: "Investigate thread-bound ACP delivery", + status: "running", + deliveryStatus: "pending", + terminalSummary: "ACP final answer", + startedAt: 100, + }); + + emitAgentEvent({ + runId, + stream: "lifecycle", + data: { + phase: "end", + endedAt: 250, + }, + }); + + await waitForAssertion(() => { + const task = findTaskByRunId(runId); + if (!task) { + throw new Error(`Expected task for run ${runId}`); + } + expect(task.status).toBe("succeeded"); + expect(task.deliveryStatus).toBe("session_queued"); + }); + expect(hoisted.sendMessageMock).not.toHaveBeenCalled(); + expect(peekSystemEvents("agent:main:discord:guild-123:channel-parent-channel")).toEqual([ + expect.stringContaining("Background task ready for review: ACP background task"), + ]); + }); + }, + ); + it.each([ { id: "channel", diff --git a/src/tasks/task-registry.ts b/src/tasks/task-registry.ts index d70c5b2efdbc..414ff7f24e68 100644 --- a/src/tasks/task-registry.ts +++ b/src/tasks/task-registry.ts @@ -1233,12 +1233,34 @@ function canDeliverTaskToRequesterOrigin(task: TaskRecord): boolean { if (shouldRouteCompletionThroughRequesterSession(owner.sessionKey)) { return false; } - const origin = owner.requesterOrigin; + return canDeliverToRequesterOrigin(owner.requesterOrigin); +} + +function canDeliverToRequesterOrigin(origin: TaskDeliveryState["requesterOrigin"]): boolean { const channel = origin?.channel?.trim(); const to = origin?.to?.trim(); return Boolean(channel && to && isDeliverableMessageChannel(channel)); } +function canDeliverParentReviewTaskToBoundDiscordThread(task: TaskRecord): boolean { + if (!shouldUseParentReviewTaskTerminalMessage(task)) { + return false; + } + const owner = resolveTaskDeliveryOwner(task); + const origin = owner.requesterOrigin; + const channel = origin?.channel?.trim().toLowerCase(); + const to = origin?.to?.trim().toLowerCase(); + const threadId = String(origin?.threadId ?? "").trim(); + // This is a narrow transport exception for explicitly bound Discord threads, + // not a general parent-review direct-delivery relaxation. + return Boolean( + channel === "discord" && + to?.startsWith("channel:") && + threadId && + canDeliverToRequesterOrigin(origin), + ); +} + function resolveMissingOwnerDeliveryStatus(task: TaskRecord): TaskDeliveryStatus { return task.scopeKind === "system" ? "not_applicable" : "parent_missing"; } @@ -1322,13 +1344,15 @@ export async function maybeDeliverTaskTerminalUpdate(taskId: string): Promise