Skip to content

Commit c67dc59

Browse files
sandieman2steipete
sandieman2
and
steipete
authored
fix(reply): deliver final reply when queued follow-up claims session; scope dedupe to routed thread (#90943)
* fix(reply): deliver final reply when queued follow-up claims session; scope dedupe to routed thread Two core bugs caused composed replies to be silently dropped (no delivery, no error) when a second message arrived in the same thread mid-run: 1. dispatch-from-config: ensureDispatchReplyOperation only kept the dispatch-owned operation authoritative while it had no result. Once runReplyAgent completed the operation to drain queued follow-ups, a second same-thread inbound could claim the session and the first final reply would try to re-acquire the lane instead of finishing delivery, deadlocking behind the queued work. Keep the dispatch-owned operation authoritative through final delivery. 2. reply-payloads-dedupe: messaging-tool reply dedupe compared only the channel target, not the routed thread, so a send in one thread could suppress a later reply in a different thread. Thread the routed thread id through buildReplyPayloads + follow-up delivery and only fall back to channel-only matching for providers without a thread-aware suppression matcher when neither side carries thread evidence. Adds regression tests; existing Telegram topic-suppression behavior is preserved by gating the thread guard to providers lacking a plugin matcher. * fix(reply): preserve threaded message delivery evidence * fix(reply): dedupe final payloads by delivery route * fix(slack): preserve native send thread evidence * fix(reply): preserve explicit reply thread evidence * fix(reply): align explicit reply route dedupe * fix(reply): preserve delivery lane through final dispatch * fix(mattermost): preserve threaded tool send routes * chore(plugin-sdk): refresh API baseline * fix(reply): align final delivery route dedupe * fix(reply): gate followups on final delivery * fix(reply): keep send receipts private * fix(reply): infer implicit message provider * fix(reply): align routed threading policy * fix(reply): preserve queued delivery context * fix(reply): hydrate queued system event routes * fix(reply): hydrate queued execution routes * fix(reply): scope final delivery barriers * fix(slack): preserve DM target aliases * fix(reply): mirror resolved source thread routes * fix(mattermost): retain delayed delivery barrier * fix(codex): separate message routing from tool policy * fix(reply): consume normalized Slack DM targets once * fix(slack): remove stale target alias * style(reply): satisfy changed lint gates * fix(mattermost): preserve explicit reply targets * test: align Slack reply branch checks * fix(reply): persist overflow summaries to admitted session --------- Co-authored-by: Peter Steinberger <[email protected]>
1 parent e174418 commit c67dc59

123 files changed

Lines changed: 8157 additions & 653 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎docs/.generated/plugin-sdk-api-baseline.sha256‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,2 +1,2 @@
1-
40b3c841849fbc29938a3bbb990e28a5db30142941c8ef0c081a94cee4c78331 plugin-sdk-api-baseline.json
2-
40ee8e1bbf112e768d4944776443f90b2441b02e3e950726e4112015cd106108 plugin-sdk-api-baseline.jsonl
1+
b121079a0912b3051a9fc319a675ef920da9db23364ca0c0ccd3c9f0a05a3a49 plugin-sdk-api-baseline.json
2+
61a0108da670e0f44ba4b861c002eb6eaa5cf63e392d4e7e7de42044cbe7d115 plugin-sdk-api-baseline.jsonl

‎extensions/codex/src/app-server/dynamic-tool-build.test.ts‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ import {
1515
includeForcedCodexDynamicToolAllow,
1616
resetOpenClawCodingToolsFactoryForTests,
1717
resolveOpenClawCodingToolsSessionKeys,
18+
resolveCodexMessageToolProvider,
1819
setOpenClawCodingToolsFactoryForTests,
1920
shouldEnableCodexAppServerNativeToolSurface,
2021
shouldForceMessageTool,
@@ -132,6 +133,15 @@ describe("Codex app-server dynamic tool build", () => {
132133
await fs.rm(tempDir, { recursive: true, force: true });
133134
});
134135

136+
it("uses the message tool channel before a differing ingress provider", () => {
137+
expect(
138+
resolveCodexMessageToolProvider({
139+
messageChannel: "discord",
140+
messageProvider: "discord-voice",
141+
}),
142+
).toBe("discord");
143+
});
144+
135145
it("filters Codex-native dynamic tools from app-server tool exposure", () => {
136146
const tools = [
137147
"read",
@@ -549,6 +559,28 @@ describe("Codex app-server dynamic tool build", () => {
549559
);
550560
});
551561

562+
it("passes native and routable channel targets into Codex dynamic tools", async () => {
563+
const sessionFile = path.join(tempDir, "session.jsonl");
564+
const workspaceDir = path.join(tempDir, "workspace");
565+
const params = createParams(sessionFile, workspaceDir);
566+
params.disableTools = false;
567+
params.currentChannelId = "D123";
568+
params.currentMessagingTarget = "user:U123";
569+
params.runtimePlan = createCodexRuntimePlanFixture();
570+
const factoryOptions: unknown[] = [];
571+
setOpenClawCodingToolsFactoryForTests((options) => {
572+
factoryOptions.push(options);
573+
return [];
574+
});
575+
576+
await buildDynamicToolsForTest(params, workspaceDir, { sandbox: null as never });
577+
578+
expect(factoryOptions[0]).toMatchObject({
579+
currentChannelId: "D123",
580+
currentMessagingTarget: "user:U123",
581+
});
582+
});
583+
552584
it("passes runtime config into Codex exec dynamic tool construction", async () => {
553585
const sessionFile = path.join(tempDir, "session.jsonl");
554586
const workspaceDir = path.join(tempDir, "workspace");

‎extensions/codex/src/app-server/dynamic-tool-build.ts‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,13 @@ export function resolveOpenClawCodingToolsSessionKeys(
9696
};
9797
}
9898

99+
/** Returns the canonical channel used for Codex message routing and receipts. */
100+
export function resolveCodexMessageToolProvider(
101+
params: Pick<EmbeddedRunAttemptParams, "messageChannel" | "messageProvider">,
102+
): string | undefined {
103+
return params.messageChannel ?? params.messageProvider;
104+
}
105+
99106
/** Resolves the channel id that hook events should target for this Codex app-server turn. */
100107
export function resolveCodexAppServerHookChannelId(
101108
params: EmbeddedRunAttemptParams,
@@ -209,7 +216,8 @@ export async function buildDynamicTools(input: DynamicToolBuildParams) {
209216
elevated: params.bashElevated,
210217
},
211218
sandbox: input.sandbox,
212-
messageProvider: params.messageChannel ?? params.messageProvider,
219+
messageProvider: resolveCodexMessageToolProvider(params),
220+
toolPolicyMessageProvider: params.messageProvider ?? params.messageChannel,
213221
agentAccountId: params.agentAccountId,
214222
messageTo: params.messageTo,
215223
messageThreadId: params.messageThreadId,
@@ -258,6 +266,7 @@ export async function buildDynamicTools(input: DynamicToolBuildParams) {
258266
),
259267
suppressManagedWebSearch: false,
260268
currentChannelId: params.currentChannelId,
269+
currentMessagingTarget: params.currentMessagingTarget,
261270
hookChannelId: resolveCodexAppServerHookChannelId(params, input.sandboxSessionKey),
262271
currentThreadTs: params.currentThreadTs,
263272
currentMessageId: params.currentMessageId,

‎extensions/codex/src/app-server/dynamic-tools.test.ts‎

Lines changed: 158 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import {
1818
import {
1919
createEmptyPluginRegistry,
2020
createMockPluginRegistry,
21+
createTestRegistry,
2122
setActivePluginRegistry,
2223
} from "openclaw/plugin-sdk/plugin-test-runtime";
2324
import { afterEach, describe, expect, it, vi } from "vitest";
@@ -798,6 +799,163 @@ describe("createCodexDynamicToolBridge", () => {
798799
]);
799800
});
800801

802+
it("records the current provider and transport thread for implicit message sends", async () => {
803+
const hasRepliedRef = { value: false };
804+
setActivePluginRegistry(
805+
createTestRegistry([
806+
{
807+
pluginId: "slack",
808+
plugin: {
809+
id: "slack",
810+
messaging: { normalizeTarget: (raw: string) => raw.trim().toLowerCase() },
811+
threading: {
812+
resolveAutoThreadId: ({
813+
to,
814+
toolContext,
815+
}: {
816+
to: string;
817+
toolContext?: {
818+
currentChannelId?: string;
819+
currentMessagingTarget?: string;
820+
currentThreadTs?: string;
821+
replyToMode?: "off" | "first" | "all" | "batched";
822+
hasRepliedRef?: { value: boolean };
823+
};
824+
}) => {
825+
if (
826+
to !== toolContext?.currentMessagingTarget &&
827+
to !== toolContext?.currentChannelId
828+
) {
829+
return undefined;
830+
}
831+
if (
832+
(toolContext?.replyToMode === "first" ||
833+
toolContext?.replyToMode === "batched") &&
834+
!toolContext.hasRepliedRef?.value
835+
) {
836+
return toolContext.currentThreadTs;
837+
}
838+
return undefined;
839+
},
840+
},
841+
},
842+
source: "test",
843+
},
844+
]),
845+
);
846+
const bridge = createCodexDynamicToolBridge({
847+
tools: [
848+
createTool({
849+
name: "message",
850+
execute: vi.fn(async () => {
851+
hasRepliedRef.value = true;
852+
return textToolResult("Sent.");
853+
}),
854+
}),
855+
],
856+
signal: new AbortController().signal,
857+
hookContext: {
858+
currentChannelProvider: "slack",
859+
currentChannelId: "D1",
860+
currentMessagingTarget: "user:u1",
861+
currentThreadId: "171.222",
862+
replyToMode: "first",
863+
hasRepliedRef,
864+
},
865+
});
866+
867+
await handleMessageToolCall(bridge, {
868+
action: "send",
869+
to: "user:U1",
870+
text: "hello from Codex",
871+
});
872+
873+
expect(bridge.telemetry.messagingToolSentTargets).toEqual([
874+
{
875+
tool: "message",
876+
provider: "slack",
877+
to: "user:u1",
878+
threadId: "171.222",
879+
threadImplicit: true,
880+
text: "hello from Codex",
881+
},
882+
]);
883+
});
884+
885+
it("records the provider-confirmed route for successful message sends", async () => {
886+
const registry = createTestRegistry([
887+
{
888+
pluginId: "mattermost",
889+
plugin: {
890+
id: "mattermost",
891+
messaging: { normalizeTarget: (raw: string) => raw.trim().toLowerCase() },
892+
actions: {
893+
extractToolSend: ({ args }: { args: Record<string, unknown> }) =>
894+
args.action === "send" && typeof args.to === "string"
895+
? { to: args.to, threadImplicit: true }
896+
: null,
897+
extractToolSendResult: ({ result }: { result: unknown }) => {
898+
const details = requireRecord(
899+
requireRecord(result, "message result").details,
900+
"message details",
901+
);
902+
const toolSend = requireRecord(details.toolSend, "tool send details");
903+
return {
904+
to: String(toolSend.to),
905+
threadId: String(toolSend.threadId),
906+
};
907+
},
908+
},
909+
},
910+
source: "test",
911+
},
912+
]);
913+
const middleware = vi.fn(async (event: { result: AgentToolResult<unknown> }) => {
914+
const details = requireRecord(event.result.details, "middleware details");
915+
const toolSend = requireRecord(details.toolSend, "middleware tool send");
916+
toolSend.to = "channel:corrupted";
917+
toolSend.threadId = "corrupted-root";
918+
return undefined;
919+
});
920+
registry.agentToolResultMiddlewares.push({
921+
pluginId: "route-details-stripper",
922+
pluginName: "Route details stripper",
923+
rawHandler: middleware,
924+
handler: middleware,
925+
runtimes: ["codex"],
926+
source: "test",
927+
});
928+
setActivePluginRegistry(registry);
929+
const bridge = createBridgeWithToolResult(
930+
"message",
931+
textToolResult("Sent.", {
932+
toolSend: {
933+
to: "channel:resolved-id",
934+
threadId: "root-post-id",
935+
},
936+
}),
937+
);
938+
939+
await handleMessageToolCall(bridge, {
940+
action: "send",
941+
provider: "mattermost",
942+
to: "town-square",
943+
text: "hello from Codex",
944+
});
945+
946+
expect(bridge.telemetry.messagingToolSentTargets).toEqual([
947+
{
948+
tool: "message",
949+
provider: "mattermost",
950+
to: "channel:resolved-id",
951+
threadId: "root-post-id",
952+
threadImplicit: undefined,
953+
threadSuppressed: undefined,
954+
text: "hello from Codex",
955+
},
956+
]);
957+
});
958+
801959
it("records message tool media attachment aliases as delivery evidence", async () => {
802960
const toolResult = {
803961
content: [{ type: "text", text: "Sent." }],

‎extensions/codex/src/app-server/dynamic-tools.ts‎

Lines changed: 55 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,8 @@ import type { AgentToolResult } from "openclaw/plugin-sdk/agent-core";
66
import {
77
createAgentToolResultMiddlewareRunner,
88
createCodexAppServerToolResultExtensionRunner,
9+
extractMessagingToolSend,
10+
extractMessagingToolSendResult,
911
extractToolResultMediaArtifact,
1012
filterToolResultMediaUrls,
1113
HEARTBEAT_RESPONSE_TOOL_NAME,
@@ -51,6 +53,12 @@ type CodexDynamicToolHookContext = {
5153
sessionKey?: string;
5254
runId?: string;
5355
channelId?: string;
56+
currentChannelProvider?: string;
57+
currentChannelId?: string;
58+
currentMessagingTarget?: string;
59+
currentThreadId?: string;
60+
replyToMode?: "off" | "first" | "all" | "batched";
61+
hasRepliedRef?: { value: boolean };
5462
};
5563

5664
type CodexToolResultHookContext = Omit<CodexDynamicToolHookContext, "config">;
@@ -67,6 +75,22 @@ type CodexDynamicToolSchemaQuarantine = {
6775
violations: readonly string[];
6876
};
6977

78+
function applyCurrentMessageProvider(
79+
toolName: string,
80+
args: Record<string, unknown>,
81+
currentProvider: string | undefined,
82+
): Record<string, unknown> {
83+
const hasProvider =
84+
typeof args.provider === "string" && args.provider.trim().length > 0
85+
? true
86+
: typeof args.channel === "string" && args.channel.trim().length > 0;
87+
const provider = currentProvider?.trim();
88+
if (toolName !== "message" || hasProvider || !provider) {
89+
return args;
90+
}
91+
return { ...args, provider };
92+
}
93+
7094
/** Runtime bridge returned to Codex app-server attempt code. */
7195
export type CodexDynamicToolBridge = {
7296
availableSpecs: CodexDynamicToolSpec[];
@@ -213,9 +237,30 @@ export function createCodexDynamicToolBridge(params: {
213237
// Prepare before marking side-effect evidence; argument preparation can
214238
// fail without the target tool actually starting.
215239
const preparedArgs = tool.prepareArguments ? tool.prepareArguments(args) : args;
240+
const telemetryArgs = isRecord(preparedArgs) ? preparedArgs : args;
241+
const messagingTelemetryArgs = applyCurrentMessageProvider(
242+
toolName,
243+
telemetryArgs,
244+
params.hookContext?.currentChannelProvider,
245+
);
246+
const messagingTarget =
247+
isMessagingTool(toolName) && isMessagingToolSendAction(toolName, telemetryArgs)
248+
? extractMessagingToolSend(toolName, messagingTelemetryArgs, {
249+
config: params.hookContext?.config,
250+
currentChannelId: params.hookContext?.currentChannelId,
251+
currentMessagingTarget: params.hookContext?.currentMessagingTarget,
252+
currentThreadId: params.hookContext?.currentThreadId,
253+
replyToMode: params.hookContext?.replyToMode,
254+
hasRepliedRef: params.hookContext?.hasRepliedRef,
255+
})
256+
: undefined;
216257
didStartExecution = true;
217258
const rawResult = await tool.execute(call.callId, preparedArgs, signal);
218259
const rawIsError = isCodexToolResultError(rawResult);
260+
const confirmedMessagingTarget =
261+
!rawIsError && messagingTarget
262+
? extractMessagingToolSendResult(messagingTarget, rawResult)
263+
: messagingTarget;
219264
const middlewareResult = await middlewareRunner.applyToolResultMiddleware({
220265
threadId: call.threadId,
221266
turnId: call.turnId,
@@ -237,11 +282,12 @@ export function createCodexDynamicToolBridge(params: {
237282
notifyAgentToolResult(options?.onAgentToolResult, toolName, result, resultIsError);
238283
collectToolTelemetry({
239284
toolName,
240-
args,
285+
args: telemetryArgs,
241286
result,
242287
mediaTrustResult: rawResult,
243288
telemetry,
244289
isError: resultIsError,
290+
messagingTarget: confirmedMessagingTarget,
245291
});
246292
void runAgentHarnessAfterToolCallHook({
247293
toolName,
@@ -631,6 +677,7 @@ function collectToolTelemetry(params: {
631677
mediaTrustResult?: AgentToolResult<unknown>;
632678
telemetry: CodexDynamicToolBridge["telemetry"];
633679
isError: boolean;
680+
messagingTarget?: MessagingToolSend;
634681
}): void {
635682
if (params.isError) {
636683
return;
@@ -683,11 +730,13 @@ function collectToolTelemetry(params: {
683730
const mediaUrls = collectMediaUrls(params.args);
684731
params.telemetry.messagingToolSentMediaUrls.push(...mediaUrls);
685732
params.telemetry.messagingToolSentTargets.push({
686-
tool: params.toolName,
687-
provider: readFirstString(params.args, ["provider", "channel"]) ?? params.toolName,
688-
accountId: readFirstString(params.args, ["accountId", "account_id"]),
689-
to: readFirstString(params.args, ["to", "target", "recipient"]),
690-
threadId: readFirstString(params.args, ["threadId", "thread_id", "messageThreadId"]),
733+
...(params.messagingTarget ?? {
734+
tool: params.toolName,
735+
provider: readFirstString(params.args, ["provider", "channel"]) ?? params.toolName,
736+
accountId: readFirstString(params.args, ["accountId", "account_id"]),
737+
to: readFirstString(params.args, ["to", "target", "recipient"]),
738+
threadId: readFirstString(params.args, ["threadId", "thread_id", "messageThreadId"]),
739+
}),
691740
...(text ? { text } : {}),
692741
...(mediaUrls.length > 0 ? { mediaUrls } : {}),
693742
});

‎extensions/codex/src/app-server/run-attempt.ts‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -149,6 +149,7 @@ import {
149149
includeForcedCodexDynamicToolAllow,
150150
isCodexNativeExecutionBlockedByNodeExecHost,
151151
resolveCodexAppServerHookChannelId,
152+
resolveCodexMessageToolProvider,
152153
resolveOpenClawCodingToolsSessionKeys,
153154
resetOpenClawCodingToolsFactoryForTests,
154155
setOpenClawCodingToolsFactoryForTests,
@@ -724,6 +725,12 @@ export async function runCodexAppServerAttempt(
724725
sessionKey: sandboxSessionKey,
725726
runId: params.runId,
726727
channelId: hookChannelId,
728+
currentChannelProvider: resolveCodexMessageToolProvider(params),
729+
currentChannelId: params.currentChannelId,
730+
currentMessagingTarget: params.currentMessagingTarget,
731+
currentThreadId: params.currentThreadTs,
732+
replyToMode: params.replyToMode,
733+
hasRepliedRef: params.hasRepliedRef,
727734
},
728735
});
729736
const hadSessionFile = await pathExists(activeSessionFile);

0 commit comments

Comments
 (0)