mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-28 13:26:04 -06:00
fix(discord): keep activity receipts in adopted threads (#116119)
* fix(discord): keep active thread replies together * fix(discord): preserve adopted thread progress receipts * test(discord): prove adopted thread reply matching * fix(discord): preserve receipt on draft finalization failure
This commit is contained in:
Jason (Json)
@@ -8,6 +8,11 @@ const handleDiscordActionMock = vi
|
||||
.spyOn(runtimeModule, "handleDiscordAction")
|
||||
.mockResolvedValue({ content: [], details: { ok: true } });
|
||||
const { handleDiscordMessageAction } = await import("./handle-action.js");
|
||||
const {
|
||||
beginDiscordActiveTurnThreadRoute,
|
||||
notifyDiscordActiveTurnThreadCreated,
|
||||
notifyDiscordActiveTurnThreadReplyDelivered,
|
||||
} = await import("../active-turn-thread-route.js");
|
||||
const { beginDiscordInboundEventDeliveryCorrelation } =
|
||||
await import("../inbound-event-delivery.js");
|
||||
|
||||
@@ -618,6 +623,171 @@ describe("handleDiscordMessageAction", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("adopts a thread created from the active source message and confirms replies there", async () => {
|
||||
const sessionKey = "agent:main:discord:channel:channel-1";
|
||||
const onThreadAdopted = vi.fn();
|
||||
const onThreadReplyDelivered = vi.fn();
|
||||
const endRoute = beginDiscordActiveTurnThreadRoute(sessionKey, {
|
||||
accountId: "account-1",
|
||||
sourceChannelId: "channel-1",
|
||||
sourceMessageId: "message-1",
|
||||
onThreadAdopted,
|
||||
onThreadReplyDelivered,
|
||||
});
|
||||
try {
|
||||
expect(
|
||||
notifyDiscordActiveTurnThreadReplyDelivered({
|
||||
sessionKey,
|
||||
accountId: "account-1",
|
||||
}),
|
||||
).toBe(false);
|
||||
|
||||
handleDiscordActionMock.mockResolvedValueOnce({
|
||||
content: [],
|
||||
details: { thread: { id: "thread-1" } },
|
||||
});
|
||||
await handleDiscordMessageAction({
|
||||
action: "thread-create",
|
||||
params: {
|
||||
channelId: "channel-1",
|
||||
messageId: "message-1",
|
||||
threadName: "investigation",
|
||||
},
|
||||
cfg: discordConfig({ threads: true }),
|
||||
accountId: "account-1",
|
||||
sessionKey,
|
||||
});
|
||||
|
||||
expect(onThreadAdopted).toHaveBeenCalledWith("thread-1");
|
||||
|
||||
handleDiscordActionMock.mockResolvedValueOnce({
|
||||
content: [],
|
||||
details: { ok: true },
|
||||
});
|
||||
const mismatchedResult = await handleDiscordMessageAction({
|
||||
action: "thread-reply",
|
||||
params: {
|
||||
threadId: "thread-2",
|
||||
message: "unrelated",
|
||||
},
|
||||
cfg: discordConfig({ threads: true }),
|
||||
accountId: "account-1",
|
||||
sessionKey,
|
||||
});
|
||||
|
||||
expect(mismatchedResult.details).toEqual({ ok: true });
|
||||
expect(onThreadReplyDelivered).not.toHaveBeenCalled();
|
||||
|
||||
handleDiscordActionMock.mockResolvedValueOnce({
|
||||
content: [],
|
||||
details: { ok: true },
|
||||
});
|
||||
const result = await handleDiscordMessageAction({
|
||||
action: "thread-reply",
|
||||
params: {
|
||||
threadId: "thread-1",
|
||||
message: "done",
|
||||
},
|
||||
cfg: discordConfig({ threads: true }),
|
||||
accountId: "account-1",
|
||||
sessionKey,
|
||||
});
|
||||
|
||||
expect(result.details).toEqual({
|
||||
ok: true,
|
||||
sourceReplyRoute: "current-source",
|
||||
});
|
||||
expect(onThreadReplyDelivered).toHaveBeenCalledWith("thread-1");
|
||||
} finally {
|
||||
endRoute();
|
||||
}
|
||||
});
|
||||
|
||||
it("confirms only the matching adopted thread across concurrent routes", async () => {
|
||||
const sessionKey = "agent:main:discord:channel:channel-1";
|
||||
const firstReplyDelivered = vi.fn();
|
||||
const secondReplyDelivered = vi.fn();
|
||||
const endFirstRoute = beginDiscordActiveTurnThreadRoute(sessionKey, {
|
||||
accountId: "account-1",
|
||||
sourceChannelId: "channel-1",
|
||||
sourceMessageId: "message-1",
|
||||
onThreadAdopted: vi.fn(),
|
||||
onThreadReplyDelivered: firstReplyDelivered,
|
||||
});
|
||||
const endSecondRoute = beginDiscordActiveTurnThreadRoute(sessionKey, {
|
||||
accountId: "account-1",
|
||||
sourceChannelId: "channel-1",
|
||||
sourceMessageId: "message-2",
|
||||
onThreadAdopted: vi.fn(),
|
||||
onThreadReplyDelivered: secondReplyDelivered,
|
||||
});
|
||||
try {
|
||||
await notifyDiscordActiveTurnThreadCreated({
|
||||
sessionKey,
|
||||
accountId: "account-1",
|
||||
sourceChannelId: "channel-1",
|
||||
sourceMessageId: "message-1",
|
||||
threadId: "thread-1",
|
||||
});
|
||||
await notifyDiscordActiveTurnThreadCreated({
|
||||
sessionKey,
|
||||
accountId: "account-1",
|
||||
sourceChannelId: "channel-1",
|
||||
sourceMessageId: "message-2",
|
||||
threadId: "thread-2",
|
||||
});
|
||||
|
||||
expect(
|
||||
notifyDiscordActiveTurnThreadReplyDelivered({
|
||||
sessionKey,
|
||||
accountId: "account-1",
|
||||
threadId: "thread-2",
|
||||
}),
|
||||
).toBe(true);
|
||||
expect(firstReplyDelivered).not.toHaveBeenCalled();
|
||||
expect(secondReplyDelivered).toHaveBeenCalledWith("thread-2");
|
||||
} finally {
|
||||
endFirstRoute();
|
||||
endSecondRoute();
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps a successful thread-create result when progress migration fails", async () => {
|
||||
const sessionKey = "agent:main:discord:channel:channel-1";
|
||||
const onThreadAdoptionError = vi.fn();
|
||||
const endRoute = beginDiscordActiveTurnThreadRoute(sessionKey, {
|
||||
sourceChannelId: "channel-1",
|
||||
sourceMessageId: "message-1",
|
||||
onThreadAdopted: async () => {
|
||||
throw new Error("preview move failed");
|
||||
},
|
||||
onThreadAdoptionError,
|
||||
});
|
||||
try {
|
||||
const expectedResult = {
|
||||
content: [],
|
||||
details: { thread: { id: "thread-1" } },
|
||||
};
|
||||
handleDiscordActionMock.mockResolvedValueOnce(expectedResult);
|
||||
|
||||
const result = await handleDiscordMessageAction({
|
||||
action: "thread-create",
|
||||
params: {
|
||||
channelId: "channel-1",
|
||||
messageId: "message-1",
|
||||
threadName: "investigation",
|
||||
},
|
||||
cfg: discordConfig({ threads: true }),
|
||||
sessionKey,
|
||||
});
|
||||
|
||||
expect(result).toBe(expectedResult);
|
||||
expect(onThreadAdoptionError).toHaveBeenCalledWith(expect.any(Error));
|
||||
} finally {
|
||||
endRoute();
|
||||
}
|
||||
});
|
||||
|
||||
it("forwards top-level components on sends", async () => {
|
||||
const components = { blocks: [{ type: "text", text: "Pick one" }] };
|
||||
const cfg = discordConfig();
|
||||
|
||||
@@ -16,6 +16,10 @@ import {
|
||||
} from "openclaw/plugin-sdk/interactive-runtime";
|
||||
import { normalizeOptionalStringifiedId } from "openclaw/plugin-sdk/string-coerce-runtime";
|
||||
import { handleDiscordAction } from "../../action-runtime-api.js";
|
||||
import {
|
||||
notifyDiscordActiveTurnThreadCreated,
|
||||
notifyDiscordActiveTurnThreadReplyDelivered,
|
||||
} from "../active-turn-thread-route.js";
|
||||
import { notifyDiscordInboundEventOutboundSuccess } from "../inbound-event-delivery.js";
|
||||
import {
|
||||
DISCORD_PRESENTATION_CAPABILITIES,
|
||||
@@ -32,6 +36,17 @@ import { readDiscordAutoArchiveDurationParam } from "./runtime.shared.js";
|
||||
|
||||
const providerId = "discord";
|
||||
|
||||
function withCurrentSourceReplyRoute<T>(result: AgentToolResult<T>): AgentToolResult<T> {
|
||||
const details =
|
||||
result.details && typeof result.details === "object" && !Array.isArray(result.details)
|
||||
? result.details
|
||||
: {};
|
||||
return {
|
||||
...result,
|
||||
details: { ...details, sourceReplyRoute: "current-source" } as T,
|
||||
};
|
||||
}
|
||||
|
||||
function readCurrentDiscordTarget(
|
||||
toolContext: Pick<ChannelMessageActionContext, "toolContext">["toolContext"],
|
||||
): string | undefined {
|
||||
@@ -402,6 +417,18 @@ export async function handleDiscordMessageAction(
|
||||
cfg,
|
||||
actionOptions,
|
||||
);
|
||||
const details =
|
||||
result.details && typeof result.details === "object" && !Array.isArray(result.details)
|
||||
? (result.details as { thread?: { id?: unknown } })
|
||||
: undefined;
|
||||
const threadId = typeof details?.thread?.id === "string" ? details.thread.id : undefined;
|
||||
await notifyDiscordActiveTurnThreadCreated({
|
||||
sessionKey: ctx.sessionKey,
|
||||
accountId,
|
||||
sourceChannelId: resolveChannelId(),
|
||||
sourceMessageId: messageId,
|
||||
threadId,
|
||||
});
|
||||
notifyVisibleOutbound(resolveChannelId());
|
||||
return result;
|
||||
}
|
||||
@@ -451,7 +478,17 @@ export async function handleDiscordMessageAction(
|
||||
});
|
||||
if (adminResult !== undefined) {
|
||||
if (action === "thread-reply") {
|
||||
notifyVisibleOutbound(readStringParam(params, "threadId") ?? readTarget());
|
||||
const threadId = readStringParam(params, "threadId") ?? readTarget();
|
||||
notifyVisibleOutbound(threadId);
|
||||
if (
|
||||
notifyDiscordActiveTurnThreadReplyDelivered({
|
||||
sessionKey: ctx.sessionKey,
|
||||
accountId,
|
||||
threadId,
|
||||
})
|
||||
) {
|
||||
return withCurrentSourceReplyRoute(adminResult);
|
||||
}
|
||||
}
|
||||
return adminResult;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
type ActiveDiscordTurnThreadRoute = {
|
||||
accountId?: string;
|
||||
sourceChannelId: string;
|
||||
sourceMessageId: string;
|
||||
adoptedThreadId?: string;
|
||||
onThreadAdopted: (threadId: string) => Promise<void> | void;
|
||||
onThreadReplyDelivered?: (threadId: string) => void;
|
||||
onThreadAdoptionError?: (error: unknown) => void;
|
||||
};
|
||||
|
||||
const activeRoutes = new Map<string, Set<ActiveDiscordTurnThreadRoute>>();
|
||||
|
||||
function normalizeId(value: string | undefined): string | undefined {
|
||||
const trimmed = value?.trim();
|
||||
return trimmed || undefined;
|
||||
}
|
||||
|
||||
export function beginDiscordActiveTurnThreadRoute(
|
||||
sessionKey: string | undefined,
|
||||
route: ActiveDiscordTurnThreadRoute,
|
||||
): () => void {
|
||||
const key = normalizeId(sessionKey);
|
||||
if (!key) {
|
||||
return () => {};
|
||||
}
|
||||
const routes = activeRoutes.get(key) ?? new Set<ActiveDiscordTurnThreadRoute>();
|
||||
routes.add(route);
|
||||
activeRoutes.set(key, routes);
|
||||
return () => {
|
||||
routes.delete(route);
|
||||
if (routes.size === 0 && activeRoutes.get(key) === routes) {
|
||||
activeRoutes.delete(key);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
export async function notifyDiscordActiveTurnThreadCreated(params: {
|
||||
sessionKey?: string | null;
|
||||
accountId?: string | null;
|
||||
sourceChannelId?: string;
|
||||
sourceMessageId?: string;
|
||||
threadId?: string;
|
||||
}): Promise<boolean> {
|
||||
const key = normalizeId(params.sessionKey ?? undefined);
|
||||
const threadId = normalizeId(params.threadId);
|
||||
const sourceChannelId = normalizeId(params.sourceChannelId);
|
||||
const sourceMessageId = normalizeId(params.sourceMessageId);
|
||||
const route = key
|
||||
? Array.from(activeRoutes.get(key) ?? []).find(
|
||||
(candidate) =>
|
||||
sourceChannelId === candidate.sourceChannelId &&
|
||||
sourceMessageId === candidate.sourceMessageId &&
|
||||
(!candidate.accountId || !params.accountId || candidate.accountId === params.accountId),
|
||||
)
|
||||
: undefined;
|
||||
if (!route || !threadId) {
|
||||
return false;
|
||||
}
|
||||
route.adoptedThreadId = threadId;
|
||||
try {
|
||||
await route.onThreadAdopted(threadId);
|
||||
} catch (error) {
|
||||
route.onThreadAdoptionError?.(error);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
export function notifyDiscordActiveTurnThreadReplyDelivered(params: {
|
||||
sessionKey?: string | null;
|
||||
accountId?: string | null;
|
||||
threadId?: string;
|
||||
}): boolean {
|
||||
const route = findDiscordActiveTurnThreadReplyRoute(params);
|
||||
const threadId = normalizeId(params.threadId ?? undefined);
|
||||
if (!route || !threadId) {
|
||||
return false;
|
||||
}
|
||||
route.onThreadReplyDelivered?.(threadId);
|
||||
return true;
|
||||
}
|
||||
|
||||
function findDiscordActiveTurnThreadReplyRoute(params: {
|
||||
sessionKey?: string | null;
|
||||
accountId?: string | null;
|
||||
threadId?: string;
|
||||
}): ActiveDiscordTurnThreadRoute | undefined {
|
||||
const key = normalizeId(params.sessionKey ?? undefined);
|
||||
const threadId = normalizeId(params.threadId);
|
||||
if (!key || !threadId) {
|
||||
return undefined;
|
||||
}
|
||||
return Array.from(activeRoutes.get(key) ?? []).find(
|
||||
(route) =>
|
||||
Boolean(route.adoptedThreadId) &&
|
||||
route.adoptedThreadId === threadId &&
|
||||
(!route.accountId || !params.accountId || route.accountId === params.accountId),
|
||||
);
|
||||
}
|
||||
@@ -4,6 +4,83 @@ import { describe, expect, it, vi } from "vitest";
|
||||
import { createDiscordDraftStream } from "./draft-stream.js";
|
||||
|
||||
describe("createDiscordDraftStream", () => {
|
||||
it("moves the visible draft to a newly adopted thread", async () => {
|
||||
const rest = {
|
||||
post: vi
|
||||
.fn()
|
||||
.mockResolvedValueOnce({ id: "parent-draft" })
|
||||
.mockResolvedValueOnce({ id: "thread-draft" }),
|
||||
patch: vi.fn(async () => undefined),
|
||||
delete: vi.fn(async () => undefined),
|
||||
};
|
||||
const stream = createDiscordDraftStream({
|
||||
rest: rest as never,
|
||||
channelId: "parent",
|
||||
throttleMs: 250,
|
||||
});
|
||||
|
||||
stream.update("working");
|
||||
await stream.flush();
|
||||
stream.update("working harder");
|
||||
await stream.retarget("thread-1");
|
||||
|
||||
expect(rest.delete).toHaveBeenCalledWith("/channels/parent/messages/parent-draft");
|
||||
expect(rest.post).toHaveBeenLastCalledWith(
|
||||
"/channels/thread-1/messages",
|
||||
expect.objectContaining({ body: expect.objectContaining({ content: "working harder" }) }),
|
||||
);
|
||||
expect(stream.messageId()).toBe("thread-draft");
|
||||
});
|
||||
|
||||
it("retries cleanup for a parent draft left behind by retargeting", async () => {
|
||||
const rest = {
|
||||
post: vi
|
||||
.fn()
|
||||
.mockResolvedValueOnce({ id: "parent-draft" })
|
||||
.mockResolvedValueOnce({ id: "thread-draft" }),
|
||||
patch: vi.fn(async () => undefined),
|
||||
delete: vi.fn().mockRejectedValueOnce(new Error("transient")).mockResolvedValue(undefined),
|
||||
};
|
||||
const stream = createDiscordDraftStream({
|
||||
rest: rest as never,
|
||||
channelId: "parent",
|
||||
throttleMs: 250,
|
||||
});
|
||||
|
||||
stream.update("working");
|
||||
await stream.flush();
|
||||
await stream.retarget("thread-1");
|
||||
await stream.cleanupRetargeted();
|
||||
|
||||
expect(rest.delete).toHaveBeenNthCalledWith(1, "/channels/parent/messages/parent-draft");
|
||||
expect(rest.delete).toHaveBeenNthCalledWith(2, "/channels/parent/messages/parent-draft");
|
||||
});
|
||||
|
||||
it("keeps the parent draft when the thread replacement cannot be created", async () => {
|
||||
const rest = {
|
||||
post: vi
|
||||
.fn()
|
||||
.mockResolvedValueOnce({ id: "parent-draft" })
|
||||
.mockRejectedValueOnce(new Error("thread post failed")),
|
||||
patch: vi.fn(async () => undefined),
|
||||
delete: vi.fn(async () => undefined),
|
||||
};
|
||||
const stream = createDiscordDraftStream({
|
||||
rest: rest as never,
|
||||
channelId: "parent",
|
||||
throttleMs: 250,
|
||||
});
|
||||
|
||||
stream.update("working");
|
||||
await stream.flush();
|
||||
|
||||
await expect(stream.retarget("thread-1")).rejects.toThrow("retarget replacement failed");
|
||||
expect(rest.delete).not.toHaveBeenCalled();
|
||||
|
||||
await stream.cleanupRetargeted();
|
||||
expect(rest.delete).toHaveBeenCalledWith("/channels/parent/messages/parent-draft");
|
||||
});
|
||||
|
||||
it("holds the first preview until minInitialChars is reached", async () => {
|
||||
const rest = {
|
||||
post: vi.fn(async () => ({ id: "m1" })),
|
||||
|
||||
@@ -23,6 +23,10 @@ type DiscordDraftStream = {
|
||||
discardPending: () => Promise<void>;
|
||||
seal: () => Promise<void>;
|
||||
stop: () => Promise<void>;
|
||||
/** Move the active draft to another Discord channel, preserving its current text. */
|
||||
retarget: (channelId: string) => Promise<void>;
|
||||
/** Retry cleanup for drafts left behind by a failed retarget delete. */
|
||||
cleanupRetargeted: () => Promise<void>;
|
||||
/** Reset internal state so the next update creates a new message instead of editing. */
|
||||
forceNewMessage: (mode?: "preserve" | "discard") => void;
|
||||
};
|
||||
@@ -42,7 +46,7 @@ export function createDiscordDraftStream(params: {
|
||||
const maxChars = Math.min(params.maxChars ?? DISCORD_STREAM_MAX_CHARS, DISCORD_STREAM_MAX_CHARS);
|
||||
const throttleMs = Math.max(250, params.throttleMs ?? DEFAULT_THROTTLE_MS);
|
||||
const minInitialChars = params.minInitialChars;
|
||||
const channelId = params.channelId;
|
||||
let channelId = params.channelId;
|
||||
const rest = params.rest;
|
||||
const flags = resolveDiscordMessageFlags({ suppressEmbeds: params.suppressEmbeds });
|
||||
const resolveReplyToMessageId = () =>
|
||||
@@ -56,6 +60,7 @@ export function createDiscordDraftStream(params: {
|
||||
let streamGeneration = 0;
|
||||
let activeCreateGeneration: number | undefined;
|
||||
let discardActiveCreate = false;
|
||||
let retargetedCleanup: Array<{ channelId: string; messageId: string }> = [];
|
||||
|
||||
const sendOrEditStreamMessage = async (text: string): Promise<boolean> => {
|
||||
const generation = streamGeneration;
|
||||
@@ -185,6 +190,56 @@ export function createDiscordDraftStream(params: {
|
||||
loop.resetPending();
|
||||
loop.resetThrottleWindow();
|
||||
};
|
||||
const cleanupRetargeted = async () => {
|
||||
const pending = retargetedCleanup;
|
||||
retargetedCleanup = [];
|
||||
for (const stale of pending) {
|
||||
try {
|
||||
await deleteChannelMessage(rest, stale.channelId, stale.messageId);
|
||||
} catch (err) {
|
||||
retargetedCleanup.push(stale);
|
||||
params.warn?.(`discord stream preview retarget cleanup failed: ${formatErrorMessage(err)}`);
|
||||
}
|
||||
}
|
||||
};
|
||||
const retarget = async (nextChannelId: string) => {
|
||||
const normalized = nextChannelId.trim();
|
||||
if (!normalized || normalized === channelId) {
|
||||
return;
|
||||
}
|
||||
await loop.waitForInFlight();
|
||||
const pendingText = loop.takePending?.() ?? "";
|
||||
const previousChannelId = channelId;
|
||||
const previousMessageId = streamMessageId;
|
||||
const previousText = pendingText || lastSentText;
|
||||
streamGeneration += 1;
|
||||
channelId = normalized;
|
||||
streamMessageId = undefined;
|
||||
lastSentText = "";
|
||||
streamState.stopped = false;
|
||||
streamState.final = false;
|
||||
loop.resetThrottleWindow();
|
||||
if (previousText) {
|
||||
update(previousText);
|
||||
await loop.flush();
|
||||
}
|
||||
if (previousMessageId) {
|
||||
const stale = {
|
||||
channelId: previousChannelId,
|
||||
messageId: previousMessageId,
|
||||
};
|
||||
if (!streamMessageId) {
|
||||
retargetedCleanup.push(stale);
|
||||
throw new Error("discord stream preview retarget replacement failed");
|
||||
}
|
||||
try {
|
||||
await deleteChannelMessage(rest, previousChannelId, previousMessageId);
|
||||
} catch (err) {
|
||||
retargetedCleanup.push(stale);
|
||||
params.warn?.(`discord stream preview retarget cleanup failed: ${formatErrorMessage(err)}`);
|
||||
}
|
||||
}
|
||||
};
|
||||
const deleteCurrentMessage = async () => {
|
||||
loop.resetPending();
|
||||
await loop.waitForInFlight();
|
||||
@@ -213,6 +268,8 @@ export function createDiscordDraftStream(params: {
|
||||
discardPending,
|
||||
seal,
|
||||
stop,
|
||||
retarget,
|
||||
cleanupRetargeted,
|
||||
forceNewMessage,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -217,6 +217,31 @@ export function createDiscordDraftPreviewController(params: {
|
||||
markPreviewFinalized() {
|
||||
finalizedViaPreviewMessage = true;
|
||||
},
|
||||
async retarget(channelId: string) {
|
||||
await draftStream?.retarget(channelId);
|
||||
},
|
||||
async finalizeProgressReceipt(receiptLine: string) {
|
||||
if (!draftStream || discordStreamMode !== "progress") {
|
||||
return false;
|
||||
}
|
||||
const receipt = receiptLine.trim();
|
||||
if (!receipt) {
|
||||
return false;
|
||||
}
|
||||
const progressText = lastPartialText.trimEnd();
|
||||
const maxProgressChars = Math.max(0, draftMaxChars - receipt.length - 1);
|
||||
const fittedProgressText =
|
||||
progressText.length > maxProgressChars
|
||||
? progressText.slice(progressText.length - maxProgressChars).trimStart()
|
||||
: progressText;
|
||||
draftStream.update(fittedProgressText ? `${fittedProgressText}\n${receipt}` : receipt);
|
||||
await draftStream.stop();
|
||||
if (!draftStream.messageId()) {
|
||||
return false;
|
||||
}
|
||||
finalizedViaPreviewMessage = true;
|
||||
return true;
|
||||
},
|
||||
disableBlockStreamingForDraft: draftStream ? true : undefined,
|
||||
async pushToolProgress(
|
||||
line?: string | ChannelProgressDraftLine,
|
||||
@@ -388,6 +413,7 @@ export function createDiscordDraftPreviewController(params: {
|
||||
if (!finalizedViaPreviewMessage && draftStream?.messageId()) {
|
||||
await draftStream.clear();
|
||||
}
|
||||
await draftStream?.cleanupRetargeted();
|
||||
} catch (err) {
|
||||
params.log(`discord: draft cleanup failed: ${String(err)}`);
|
||||
}
|
||||
|
||||
@@ -6,6 +6,8 @@ import {
|
||||
import { resolveMarkdownTableMode } from "openclaw/plugin-sdk/markdown-table-runtime";
|
||||
import { resolveChunkMode } from "openclaw/plugin-sdk/reply-chunking";
|
||||
import { createChannelHistoryWindow } from "openclaw/plugin-sdk/reply-history";
|
||||
import { resolveSendableOutboundReplyParts } from "openclaw/plugin-sdk/reply-payload";
|
||||
import type { ReplyDispatchKind, ReplyPayload } from "openclaw/plugin-sdk/reply-runtime";
|
||||
import { logVerbose } from "openclaw/plugin-sdk/runtime-env";
|
||||
import { getSessionEntry, resolveStorePath } from "openclaw/plugin-sdk/session-store-runtime";
|
||||
import { readLatestAssistantTextByIdentity } from "openclaw/plugin-sdk/session-transcript-runtime";
|
||||
@@ -15,12 +17,63 @@ import type { RequestClient } from "../internal/discord.js";
|
||||
import { buildDiscordMessageProcessContext } from "./message-handler.context.js";
|
||||
import { createDiscordDraftPreviewController } from "./message-handler.draft-preview.js";
|
||||
import type { DiscordMessagePreflightContext } from "./message-handler.preflight.js";
|
||||
import { formatDiscordReplySkip } from "./reply-delivery.js";
|
||||
import { createDiscordReplyTypingFeedback } from "./reply-typing-feedback.js";
|
||||
|
||||
type DiscordMessageProcessContext = NonNullable<
|
||||
Awaited<ReturnType<typeof buildDiscordMessageProcessContext>>
|
||||
>;
|
||||
|
||||
export function formatDiscordReasoningQuote(quoteText: string): string | undefined {
|
||||
const lines = quoteText
|
||||
.split("\n")
|
||||
.map((line) => line.trim())
|
||||
.filter(Boolean);
|
||||
if (!lines.length) {
|
||||
return undefined;
|
||||
}
|
||||
lines[0] = `🧠 ${lines[0]}`;
|
||||
return lines.map((line) => `> ${line}`).join("\n");
|
||||
}
|
||||
|
||||
export function createDiscordBeforePayloadDelivery(params: {
|
||||
abortSignal?: AbortSignal;
|
||||
getDeliverTarget: () => string;
|
||||
sessionKey?: string;
|
||||
draftPreview: ReturnType<typeof createDiscordDraftPreviewController>;
|
||||
isFallbackOnlyToolWarningFinal: (payload: ReplyPayload) => boolean;
|
||||
}) {
|
||||
return (payload: ReplyPayload, info: { kind: ReplyDispatchKind }): ReplyPayload | null => {
|
||||
if (params.abortSignal?.aborted) {
|
||||
logVerbose(
|
||||
formatDiscordReplySkip({
|
||||
kind: info.kind,
|
||||
reason: "aborted before delivery",
|
||||
target: params.getDeliverTarget(),
|
||||
sessionKey: params.sessionKey,
|
||||
}),
|
||||
);
|
||||
return null;
|
||||
}
|
||||
if (payload.isReasoning || payload.isCommentary) {
|
||||
return payload;
|
||||
}
|
||||
if (
|
||||
params.draftPreview.draftStream &&
|
||||
params.draftPreview.isProgressMode &&
|
||||
info.kind === "block" &&
|
||||
!resolveSendableOutboundReplyParts(payload).hasMedia &&
|
||||
!payload.isError
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
if (info.kind === "final" && !params.isFallbackOnlyToolWarningFinal(payload)) {
|
||||
params.draftPreview.markFinalReplyStarted();
|
||||
}
|
||||
return payload;
|
||||
};
|
||||
}
|
||||
|
||||
export function createDiscordMessageReplyRuntime(params: {
|
||||
ctx: DiscordMessagePreflightContext;
|
||||
processContext: DiscordMessageProcessContext;
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
import { beginDiscordActiveTurnThreadRoute } from "../active-turn-thread-route.js";
|
||||
|
||||
type DiscordReplyReference = {
|
||||
peek: () => string | undefined;
|
||||
use: () => string | undefined;
|
||||
markSent: () => void;
|
||||
hasReplied: () => boolean;
|
||||
};
|
||||
|
||||
export async function finalizeDiscordAdoptedThreadProgressReceipt(
|
||||
hasProgressDraft: boolean,
|
||||
receipt: string,
|
||||
finalizeDraft: (receiptLine: string) => Promise<boolean>,
|
||||
deliverReceipt: (receiptLine: string) => Promise<unknown>,
|
||||
onFinalizeError?: (error: unknown) => void,
|
||||
) {
|
||||
const receiptLine = receipt.trim();
|
||||
if (!hasProgressDraft || !receiptLine) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
if (await finalizeDraft(receiptLine)) {
|
||||
return;
|
||||
}
|
||||
} catch (error) {
|
||||
onFinalizeError?.(error);
|
||||
}
|
||||
// Draft creation can fail independently from the model-owned thread reply.
|
||||
// Preserve the terminal activity record as a normal message in that thread.
|
||||
await deliverReceipt(receiptLine);
|
||||
}
|
||||
|
||||
export function createDiscordMessageActiveThreadRoute(params: {
|
||||
sessionKey?: string;
|
||||
accountId?: string;
|
||||
sourceChannelId: string;
|
||||
sourceMessageId: string;
|
||||
sourceReplyReference: DiscordReplyReference;
|
||||
log: (message: string) => void;
|
||||
}) {
|
||||
let adoptedThreadId: string | undefined;
|
||||
// Delivery is dispatch-scoped: the model may continue after the message tool
|
||||
// returns, but that continuation cannot undo the visible thread reply.
|
||||
let threadReplyDelivered = false;
|
||||
let onThreadAdopted: ((threadId: string) => Promise<void>) | undefined;
|
||||
const end = beginDiscordActiveTurnThreadRoute(params.sessionKey, {
|
||||
accountId: params.accountId,
|
||||
sourceChannelId: params.sourceChannelId,
|
||||
sourceMessageId: params.sourceMessageId,
|
||||
onThreadAdopted: async (threadId) => {
|
||||
// Route identity must change before migration so a successful thread
|
||||
// reply remains current-source even when moving the draft later fails.
|
||||
adoptedThreadId = threadId;
|
||||
await onThreadAdopted?.(threadId);
|
||||
},
|
||||
onThreadReplyDelivered: () => {
|
||||
threadReplyDelivered = true;
|
||||
},
|
||||
onThreadAdoptionError: (error) => {
|
||||
params.log(`discord: failed to move active progress into adopted thread (${String(error)})`);
|
||||
},
|
||||
});
|
||||
|
||||
return {
|
||||
replyReference: {
|
||||
peek: () => (adoptedThreadId ? undefined : params.sourceReplyReference.peek()),
|
||||
use: () => (adoptedThreadId ? undefined : params.sourceReplyReference.use()),
|
||||
markSent: () => params.sourceReplyReference.markSent(),
|
||||
hasReplied: () => Boolean(adoptedThreadId) || params.sourceReplyReference.hasReplied(),
|
||||
},
|
||||
bindThreadAdoption(callback: (threadId: string) => Promise<void>) {
|
||||
onThreadAdopted = callback;
|
||||
},
|
||||
get threadReplyDelivered() {
|
||||
return threadReplyDelivered;
|
||||
},
|
||||
end,
|
||||
};
|
||||
}
|
||||
@@ -1,5 +1,9 @@
|
||||
// Discord message processing coverage split by cohesive behavior.
|
||||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
notifyDiscordActiveTurnThreadCreated,
|
||||
notifyDiscordActiveTurnThreadReplyDelivered,
|
||||
} from "../active-turn-thread-route.js";
|
||||
import {
|
||||
createAutomaticSourceDeliveryContext,
|
||||
createNoQueuedDispatchResult,
|
||||
@@ -20,6 +24,175 @@ import {
|
||||
registerDiscordProcessTestLifecycle();
|
||||
|
||||
describe("processDiscordMessage draft streaming progress", () => {
|
||||
it("moves progress and final delivery into a thread created from the source message", async () => {
|
||||
const draftStream = createMockDraftStreamForTest();
|
||||
dispatchInboundMessage.mockImplementationOnce(async (params?: DispatchInboundParams) => {
|
||||
await params?.replyOptions?.onItemEvent?.({
|
||||
itemId: "preamble-1",
|
||||
kind: "preamble",
|
||||
progressText: "Investigating.",
|
||||
});
|
||||
await notifyDiscordActiveTurnThreadCreated({
|
||||
sessionKey: String(params?.ctx?.SessionKey),
|
||||
accountId: "default",
|
||||
sourceChannelId: "c1",
|
||||
sourceMessageId: "m1",
|
||||
threadId: "thread-1",
|
||||
});
|
||||
await params?.dispatcher.sendFinalReply({ text: "done" });
|
||||
return { queuedFinal: true, counts: { final: 1, tool: 0, block: 0 } };
|
||||
});
|
||||
const ctx = await createAutomaticSourceDeliveryContext({
|
||||
discordConfig: { streaming: { mode: "progress" } },
|
||||
});
|
||||
|
||||
await runProcessDiscordMessage(ctx);
|
||||
|
||||
expect(draftStream.retarget).toHaveBeenCalledWith("thread-1");
|
||||
expect(deliverDiscordReply).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
target: "channel:thread-1",
|
||||
replyToId: undefined,
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("keeps adopted-thread progress with terminal tool and timing receipts", async () => {
|
||||
const elapseProgressDraftStartDelay = useProgressDraftStartDelay();
|
||||
const draftStream = createMockDraftStreamForTest();
|
||||
dispatchInboundMessage.mockImplementationOnce(async (params?: DispatchInboundParams) => {
|
||||
await params?.replyOptions?.onToolStart?.({ name: "exec", phase: "start" });
|
||||
await params?.replyOptions?.onItemEvent?.({
|
||||
itemId: "tool-1",
|
||||
kind: "tool",
|
||||
progressText: "Checked the pipeline.",
|
||||
});
|
||||
await elapseProgressDraftStartDelay();
|
||||
await notifyDiscordActiveTurnThreadCreated({
|
||||
sessionKey: String(params?.ctx?.SessionKey),
|
||||
accountId: "default",
|
||||
sourceChannelId: "c1",
|
||||
sourceMessageId: "m1",
|
||||
threadId: "thread-1",
|
||||
});
|
||||
expect(
|
||||
notifyDiscordActiveTurnThreadReplyDelivered({
|
||||
sessionKey: String(params?.ctx?.SessionKey),
|
||||
accountId: "default",
|
||||
threadId: "thread-1",
|
||||
}),
|
||||
).toBe(true);
|
||||
return createNoQueuedDispatchResult();
|
||||
});
|
||||
const ctx = await createAutomaticSourceDeliveryContext({
|
||||
discordConfig: {
|
||||
streaming: { mode: "progress", progress: { label: "Investigating" } },
|
||||
},
|
||||
});
|
||||
|
||||
await runProcessDiscordMessage(ctx);
|
||||
|
||||
expect(draftStream.retarget).toHaveBeenCalledWith("thread-1");
|
||||
expect(draftStream.update).toHaveBeenLastCalledWith(
|
||||
expect.stringMatching(
|
||||
/^Investigating\n\n🛠️ Exec\n.*Checked the pipeline\.\n-# .*🛠️ 1 tool call.*⏱️ 5s$/,
|
||||
),
|
||||
);
|
||||
expect(draftStream.stop).toHaveBeenCalledTimes(1);
|
||||
expect(draftStream.clear).not.toHaveBeenCalled();
|
||||
expect(deliverDiscordReply).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("sends the adopted-thread receipt when the progress draft has no message id", async () => {
|
||||
const elapseProgressDraftStartDelay = useProgressDraftStartDelay();
|
||||
const draftStream = createMockDraftStreamForTest();
|
||||
draftStream.messageId.mockReturnValue(undefined);
|
||||
dispatchInboundMessage.mockImplementationOnce(async (params?: DispatchInboundParams) => {
|
||||
await params?.replyOptions?.onToolStart?.({ name: "exec", phase: "start" });
|
||||
await params?.replyOptions?.onItemEvent?.({
|
||||
itemId: "tool-1",
|
||||
kind: "tool",
|
||||
progressText: "Checked the pipeline.",
|
||||
});
|
||||
await elapseProgressDraftStartDelay();
|
||||
await notifyDiscordActiveTurnThreadCreated({
|
||||
sessionKey: String(params?.ctx?.SessionKey),
|
||||
accountId: "default",
|
||||
sourceChannelId: "c1",
|
||||
sourceMessageId: "m1",
|
||||
threadId: "thread-1",
|
||||
});
|
||||
notifyDiscordActiveTurnThreadReplyDelivered({
|
||||
sessionKey: String(params?.ctx?.SessionKey),
|
||||
accountId: "default",
|
||||
threadId: "thread-1",
|
||||
});
|
||||
return createNoQueuedDispatchResult();
|
||||
});
|
||||
const ctx = await createAutomaticSourceDeliveryContext({
|
||||
discordConfig: {
|
||||
streaming: { mode: "progress", progress: { label: "Investigating" } },
|
||||
},
|
||||
});
|
||||
|
||||
await runProcessDiscordMessage(ctx);
|
||||
|
||||
expect(deliverDiscordReply).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
target: "channel:thread-1",
|
||||
kind: "block",
|
||||
replies: [
|
||||
expect.objectContaining({
|
||||
text: expect.stringMatching(/🛠️ 1 tool call.*⏱️ 5s$/),
|
||||
}),
|
||||
],
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("sends the adopted-thread receipt when draft finalization rejects", async () => {
|
||||
const elapseProgressDraftStartDelay = useProgressDraftStartDelay();
|
||||
const draftStream = createMockDraftStreamForTest();
|
||||
draftStream.stop.mockRejectedValueOnce(new Error("draft stop failed"));
|
||||
dispatchInboundMessage.mockImplementationOnce(async (params?: DispatchInboundParams) => {
|
||||
await params?.replyOptions?.onToolStart?.({ name: "exec", phase: "start" });
|
||||
await elapseProgressDraftStartDelay();
|
||||
await notifyDiscordActiveTurnThreadCreated({
|
||||
sessionKey: String(params?.ctx?.SessionKey),
|
||||
accountId: "default",
|
||||
sourceChannelId: "c1",
|
||||
sourceMessageId: "m1",
|
||||
threadId: "thread-1",
|
||||
});
|
||||
notifyDiscordActiveTurnThreadReplyDelivered({
|
||||
sessionKey: String(params?.ctx?.SessionKey),
|
||||
accountId: "default",
|
||||
threadId: "thread-1",
|
||||
});
|
||||
return createNoQueuedDispatchResult();
|
||||
});
|
||||
const ctx = await createAutomaticSourceDeliveryContext({
|
||||
discordConfig: {
|
||||
streaming: { mode: "progress", progress: { label: "Investigating" } },
|
||||
},
|
||||
});
|
||||
|
||||
await runProcessDiscordMessage(ctx);
|
||||
|
||||
expect(deliverDiscordReply).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
target: "channel:thread-1",
|
||||
kind: "block",
|
||||
replies: [
|
||||
expect.objectContaining({
|
||||
text: expect.stringMatching(/🛠️ 1 tool call.*⏱️ 5s$/),
|
||||
}),
|
||||
],
|
||||
}),
|
||||
);
|
||||
expect(draftStream.clear).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("keeps opt-in commentary receipts independent from hidden tool progress", async () => {
|
||||
const draftStream = createMockDraftStreamForTest();
|
||||
|
||||
|
||||
@@ -47,6 +47,8 @@ export function createMockDraftStream() {
|
||||
discardPending: vi.fn(async () => {}),
|
||||
seal: vi.fn(async () => {}),
|
||||
stop: vi.fn(async () => {}),
|
||||
retarget: vi.fn(async () => {}),
|
||||
cleanupRetargeted: vi.fn(async () => {}),
|
||||
forceNewMessage: vi.fn(() => {
|
||||
messageId = undefined;
|
||||
}),
|
||||
|
||||
@@ -33,7 +33,15 @@ import { buildDiscordMessageProcessContext } from "./message-handler.context.js"
|
||||
import type { DiscordMessagePreflightContext } from "./message-handler.preflight.js";
|
||||
import { createDiscordMessageProgressRuntime } from "./message-handler.process-progress.js";
|
||||
import { createDiscordMessageReactionRuntime } from "./message-handler.process-reactions.js";
|
||||
import { createDiscordMessageReplyRuntime } from "./message-handler.process-reply-runtime.js";
|
||||
import {
|
||||
createDiscordBeforePayloadDelivery,
|
||||
createDiscordMessageReplyRuntime,
|
||||
formatDiscordReasoningQuote,
|
||||
} from "./message-handler.process-reply-runtime.js";
|
||||
import {
|
||||
createDiscordMessageActiveThreadRoute,
|
||||
finalizeDiscordAdoptedThreadProgressReceipt,
|
||||
} from "./message-handler.process-thread-route.js";
|
||||
import { completeDiscordSessionConflict } from "./message-handler.retry.js";
|
||||
import {
|
||||
deliverDiscordReply,
|
||||
@@ -89,6 +97,7 @@ async function processDiscordMessageInner(
|
||||
replyToMode,
|
||||
message,
|
||||
messageChannelId,
|
||||
canonicalMessageId,
|
||||
isGuildMessage,
|
||||
isDirectMessage,
|
||||
isGroupDm,
|
||||
@@ -154,10 +163,20 @@ async function processDiscordMessageInner(
|
||||
persistedSessionKey,
|
||||
turn,
|
||||
replyPlan,
|
||||
deliverTarget,
|
||||
deliverTarget: initialDeliverTarget,
|
||||
replyTarget,
|
||||
replyReference,
|
||||
replyReference: sourceReplyReference,
|
||||
} = processContext;
|
||||
let deliverTarget = initialDeliverTarget;
|
||||
const activeThreadRoute = createDiscordMessageActiveThreadRoute({
|
||||
sessionKey: ctxPayload.SessionKey,
|
||||
accountId,
|
||||
sourceChannelId: messageChannelId,
|
||||
sourceMessageId: canonicalMessageId ?? message.id,
|
||||
sourceReplyReference,
|
||||
log: logVerbose,
|
||||
});
|
||||
const replyReference = activeThreadRoute.replyReference;
|
||||
observer?.onReplyPlanResolved?.({
|
||||
createdThreadId: replyPlan.createdThreadId,
|
||||
sessionKey: persistedSessionKey,
|
||||
@@ -165,7 +184,7 @@ async function processDiscordMessageInner(
|
||||
|
||||
const replyRuntime = createDiscordMessageReplyRuntime({
|
||||
ctx,
|
||||
processContext,
|
||||
processContext: { ...processContext, replyReference },
|
||||
sourceRepliesAreToolOnly,
|
||||
shouldDisableCoreTypingKeepalive,
|
||||
isRoomEvent,
|
||||
@@ -182,10 +201,16 @@ async function processDiscordMessageInner(
|
||||
beginQueuedDeliveryCorrelation,
|
||||
endDeliveryCorrelation,
|
||||
resolveCurrentTurnTranscriptFinalText,
|
||||
deliverChannelId,
|
||||
deliverChannelId: initialDeliverChannelId,
|
||||
draftPreview,
|
||||
resolvedBlockStreamingEnabled,
|
||||
} = replyRuntime;
|
||||
let deliverChannelId = initialDeliverChannelId;
|
||||
activeThreadRoute.bindThreadAdoption(async (threadId) => {
|
||||
deliverTarget = `channel:${threadId}`;
|
||||
deliverChannelId = threadId;
|
||||
await draftPreview.retarget(threadId);
|
||||
});
|
||||
let finalReplyStartNotified = false;
|
||||
const notifyFinalReplyStart = () => {
|
||||
if (finalReplyStartNotified) {
|
||||
@@ -207,18 +232,6 @@ async function processDiscordMessageInner(
|
||||
draftPreview.markFinalReplyDelivered();
|
||||
observer?.onFinalReplyDelivered?.();
|
||||
};
|
||||
// Per-line quoting survives Discord chunking; blank quote rows render badly.
|
||||
const formatDiscordReasoningQuote = (quoteText: string): string | undefined => {
|
||||
const lines = quoteText
|
||||
.split("\n")
|
||||
.map((line) => line.trim())
|
||||
.filter(Boolean);
|
||||
if (!lines.length) {
|
||||
return undefined;
|
||||
}
|
||||
lines[0] = `🧠 ${lines[0]}`;
|
||||
return lines.map((line) => `> ${line}`).join("\n");
|
||||
};
|
||||
// Set when a progress draft collapses: the receipt appends to the final
|
||||
// answer text and the draft message deletes once that answer delivered.
|
||||
let progressReceiptLine: string | undefined;
|
||||
@@ -248,40 +261,21 @@ async function processDiscordMessageInner(
|
||||
await replyPipeline.typingCallbacks?.onReplyStart();
|
||||
await reactions.controller.setThinking();
|
||||
};
|
||||
const beforeDiscordPayloadDelivery = (
|
||||
payload: ReplyPayload,
|
||||
info: { kind: ReplyDispatchKind },
|
||||
): ReplyPayload | null => {
|
||||
if (isProcessAborted(abortSignal)) {
|
||||
logVerbose(
|
||||
formatDiscordReplySkip({
|
||||
kind: info.kind,
|
||||
reason: "aborted before delivery",
|
||||
target: deliverTarget,
|
||||
sessionKey: ctxPayload.SessionKey,
|
||||
}),
|
||||
);
|
||||
return null;
|
||||
}
|
||||
if (payload.isReasoning || payload.isCommentary) {
|
||||
return payload;
|
||||
}
|
||||
if (draftPreview.draftStream && draftPreview.isProgressMode && info.kind === "block") {
|
||||
const reply = resolveSendableOutboundReplyParts(payload);
|
||||
if (!reply.hasMedia && !payload.isError) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
if (info.kind === "final" && !isFallbackOnlyToolWarningFinal(payload)) {
|
||||
draftPreview.markFinalReplyStarted();
|
||||
}
|
||||
return payload;
|
||||
};
|
||||
const beforeDiscordPayloadDelivery = createDiscordBeforePayloadDelivery({
|
||||
abortSignal,
|
||||
getDeliverTarget: () => deliverTarget,
|
||||
sessionKey: ctxPayload.SessionKey,
|
||||
draftPreview,
|
||||
isFallbackOnlyToolWarningFinal,
|
||||
});
|
||||
|
||||
const deliverDiscordPayload = async (
|
||||
payload: ReplyPayload,
|
||||
info: { kind: ReplyDispatchKind },
|
||||
options?: { allowFallbackOnlyToolWarning?: boolean },
|
||||
options?: {
|
||||
allowFallbackOnlyToolWarning?: boolean;
|
||||
allowProgressBlock?: boolean;
|
||||
},
|
||||
) => {
|
||||
if (isProcessAborted(abortSignal)) {
|
||||
// Surface so operators don't chase missing replies when an abort
|
||||
@@ -384,7 +378,12 @@ async function processDiscordMessageInner(
|
||||
await onDiscordReplyStart();
|
||||
}
|
||||
const draftStream = draftPreview.draftStream;
|
||||
if (draftStream && draftPreview.isProgressMode && info.kind === "block") {
|
||||
if (
|
||||
draftStream &&
|
||||
draftPreview.isProgressMode &&
|
||||
info.kind === "block" &&
|
||||
!options?.allowProgressBlock
|
||||
) {
|
||||
const reply = resolveSendableOutboundReplyParts(deliverablePayload);
|
||||
if (!reply.hasMedia && !deliverablePayload.isError) {
|
||||
return { visibleReplySent: false };
|
||||
@@ -680,6 +679,23 @@ async function processDiscordMessageInner(
|
||||
dispatchAborted = true;
|
||||
return;
|
||||
}
|
||||
if (activeThreadRoute.threadReplyDelivered && !userFacingFinalDelivered) {
|
||||
draftPreview.markFinalReplyStarted();
|
||||
await finalizeDiscordAdoptedThreadProgressReceipt(
|
||||
draftPreview.hasProgressDraftToCollapse,
|
||||
progress.buildProgressSummaryLine(),
|
||||
(receiptLine) => draftPreview.finalizeProgressReceipt(receiptLine),
|
||||
(receiptText) =>
|
||||
deliverDiscordPayload(
|
||||
{ text: receiptText },
|
||||
{ kind: "block" },
|
||||
{ allowProgressBlock: true },
|
||||
),
|
||||
(error) =>
|
||||
logVerbose(`discord: failed to finalize adopted thread progress (${String(error)})`),
|
||||
);
|
||||
markUserFacingFinalDelivered();
|
||||
}
|
||||
} catch (err) {
|
||||
if (isProcessAborted(abortSignal)) {
|
||||
dispatchAborted = true;
|
||||
@@ -692,6 +708,7 @@ async function processDiscordMessageInner(
|
||||
}
|
||||
throw err;
|
||||
} finally {
|
||||
activeThreadRoute.end();
|
||||
endDeliveryCorrelation();
|
||||
await draftPreview.cleanup();
|
||||
const finalDeliveryFailed = (dispatchResult?.failedCounts?.final ?? 0) > 0;
|
||||
|
||||
@@ -340,6 +340,33 @@ describe("isDeliveredMessagingToolResult", () => {
|
||||
});
|
||||
|
||||
describe("isDeliveredMessageToolOnlySourceReplyResult", () => {
|
||||
it("accepts a confirmed adopted-thread reply outside message-tool-only mode", () => {
|
||||
expect(
|
||||
isDeliveredMessageToolOnlySourceReplyResult({
|
||||
sourceReplyDeliveryMode: "automatic",
|
||||
toolName: "message",
|
||||
args: { action: "thread-reply", threadId: "thread-1", message: "done" },
|
||||
result: {
|
||||
details: {
|
||||
ok: true,
|
||||
sourceReplyRoute: "current-source",
|
||||
},
|
||||
},
|
||||
}),
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it("rejects an unconfirmed thread reply outside message-tool-only mode", () => {
|
||||
expect(
|
||||
isDeliveredMessageToolOnlySourceReplyResult({
|
||||
sourceReplyDeliveryMode: "automatic",
|
||||
toolName: "message",
|
||||
args: { action: "thread-reply", threadId: "thread-1", message: "done" },
|
||||
result: { details: { ok: true } },
|
||||
}),
|
||||
).toBe(false);
|
||||
});
|
||||
|
||||
it("accepts only confirmed implicit message sends", () => {
|
||||
expect(
|
||||
isDeliveredMessageToolOnlySourceReplyResult({
|
||||
|
||||
@@ -60,7 +60,11 @@ function isMessageToolSourceReplyActionName(action: unknown): boolean {
|
||||
if (isMessageToolSendActionName(action)) {
|
||||
return true;
|
||||
}
|
||||
return typeof action === "string" && action.trim().toLowerCase() === "reply";
|
||||
if (typeof action !== "string") {
|
||||
return false;
|
||||
}
|
||||
const normalized = action.trim().toLowerCase();
|
||||
return normalized === "reply" || normalized === "thread-reply";
|
||||
}
|
||||
|
||||
function normalizeStatus(value: unknown): string | undefined {
|
||||
@@ -560,7 +564,10 @@ export function isDeliveredMessageToolOnlySourceReplyResult(params: {
|
||||
isError?: boolean;
|
||||
allowExplicitSourceRoute?: boolean;
|
||||
}): boolean {
|
||||
if (params.sourceReplyDeliveryMode !== "message_tool_only") {
|
||||
const confirmedCurrentSourceRoute =
|
||||
resultConfirmsCurrentSourceRoute(params.result) ||
|
||||
resultConfirmsCurrentSourceRoute(params.hookResult);
|
||||
if (params.sourceReplyDeliveryMode !== "message_tool_only" && !confirmedCurrentSourceRoute) {
|
||||
return false;
|
||||
}
|
||||
if (normalizeToolName(params.toolName) !== MESSAGE_TOOL_NAME) {
|
||||
@@ -568,12 +575,13 @@ export function isDeliveredMessageToolOnlySourceReplyResult(params: {
|
||||
}
|
||||
const args = asRecord(params.args);
|
||||
const sourceRouteReplyAction =
|
||||
params.allowExplicitSourceRoute === true && isMessageToolSourceReplyActionName(args.action);
|
||||
(params.allowExplicitSourceRoute === true || confirmedCurrentSourceRoute) &&
|
||||
isMessageToolSourceReplyActionName(args.action);
|
||||
if (!isMessageToolSendActionName(args.action) && !sourceRouteReplyAction) {
|
||||
return false;
|
||||
}
|
||||
const hasConfirmedExplicitSourceRoute =
|
||||
params.allowExplicitSourceRoute === true || resultConfirmsCurrentSourceRoute(params.result);
|
||||
params.allowExplicitSourceRoute === true || confirmedCurrentSourceRoute;
|
||||
if (hasExplicitMessageRoute(args) && !hasConfirmedExplicitSourceRoute) {
|
||||
return false;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user