Skip to content

Commit 2c8f78e

Browse files
joshavantjoshavant
authored
Harden final delivery routing refresh (#83835)
* harden final delivery routing refresh * add changelog for final delivery hardening
1 parent a9eaf0c commit 2c8f78e

5 files changed

Lines changed: 221 additions & 68 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,7 @@ Docs: https://docs.openclaw.ai
8383
- LM Studio: resolve env-template API keys like `${LMSTUDIO_API_KEY}` through the standard SecretInput path instead of sending the raw template as the bearer token, and preserve header-auth and discovery-key precedence when the template is unset. Fixes #80495. (#80568) Thanks @MonkeyLeeT.
8484
- Discord/subagents: route the initial reply from thread-bound delegated sessions into the bound Discord thread instead of the parent channel. Fixes #83170. (#83172) Thanks @100menotu001.
8585
- Gateway/sessions: rotate failed agent sessions when their transcript file is missing instead of wedging per-channel lanes. Fixes #83488. (#83553) Thanks @LLagoon3.
86+
- Agents: refresh final-delivery routing from fresh session state before declaring a no-send failure, keeping recovered runs on the normal durable delivery path. (#83835) Thanks @joshavant.
8687
- Media: prevent image metadata probing from invoking external decoder delegates on unrecognized image bytes, and stop fallback chaining after real processing errors.
8788
- Media: install Sharp with the root package and fall back to sips, Windows native imaging, ImageMagick, GraphicsMagick, or ffmpeg for image resizing/conversion when Sharp is unavailable. Fixes #83401. Thanks @scotthuang.
8889
- Telegram: deliver generated media completions back into forum topics by preserving topic IDs across requester-agent handoff. (#83556) Thanks @fuller-stack-dev.

‎src/agents/agent-command.ts‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1524,13 +1524,29 @@ async function agentCommandInternal(
15241524
}
15251525

15261526
const { deliverAgentCommandResult } = await loadDeliveryRuntime();
1527+
const resolveFreshSessionEntryForDelivery =
1528+
sessionStore && sessionKey
1529+
? async (): Promise<SessionEntry | undefined> => {
1530+
const { loadSessionStore } = await loadSessionStoreRuntime();
1531+
const freshStore = loadSessionStore(storePath, {
1532+
skipCache: true,
1533+
clone: false,
1534+
});
1535+
const freshEntry = freshStore[sessionKey];
1536+
if (freshEntry) {
1537+
sessionStore[sessionKey] = freshEntry;
1538+
}
1539+
return freshEntry;
1540+
}
1541+
: undefined;
15271542
const deliveryResult = await deliverAgentCommandResult({
15281543
cfg,
15291544
deps: resolvedDeps,
15301545
runtime,
15311546
opts,
15321547
outboundSession,
15331548
sessionEntry,
1549+
resolveFreshSessionEntryForDelivery,
15341550
result,
15351551
payloads,
15361552
});

‎src/agents/command/delivery.test.ts‎

Lines changed: 67 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,9 @@ function latestNormalizerOptions(): MediaNormalizerOptions {
103103
}
104104

105105
function latestOutboundDeliveryArgs(): {
106+
channel?: string;
107+
to?: string;
108+
accountId?: string;
106109
payloads: ReplyPayload[];
107110
bestEffort?: boolean;
108111
queuePolicy?: string;
@@ -111,7 +114,14 @@ function latestOutboundDeliveryArgs(): {
111114
if (!args || typeof args !== "object") {
112115
throw new Error("expected outbound delivery arguments");
113116
}
114-
return args as { payloads: ReplyPayload[]; bestEffort?: boolean; queuePolicy?: string };
117+
return args as {
118+
channel?: string;
119+
to?: string;
120+
accountId?: string;
121+
payloads: ReplyPayload[];
122+
bestEffort?: boolean;
123+
queuePolicy?: string;
124+
};
115125
}
116126

117127
type DeliveryStatusLike = {
@@ -329,6 +339,62 @@ describe("normalizeAgentCommandReplyPayloads", () => {
329339
});
330340
});
331341

342+
it("refreshes stale implicit session routing before final delivery", async () => {
343+
deliverOutboundPayloadsMock.mockResolvedValue([{ channel: "slack", messageId: "msg-1" }]);
344+
const runtime = { log: vi.fn(), error: vi.fn() };
345+
const resolveFreshSessionEntryForDelivery = vi.fn(async () => ({
346+
sessionId: "session-1",
347+
updatedAt: 2,
348+
deliveryContext: {
349+
channel: "slack",
350+
to: "#fresh",
351+
accountId: "workspace-1",
352+
},
353+
}));
354+
355+
const delivered = await deliverAgentCommandResult({
356+
cfg: {
357+
agents: {
358+
list: [{ id: "tester", workspace: "/tmp/agent-workspace" }],
359+
},
360+
} as OpenClawConfig,
361+
deps: {} as CliDeps,
362+
runtime: runtime as never,
363+
opts: {
364+
message: "go",
365+
deliver: true,
366+
channel: "slack",
367+
sessionKey: "agent:tester:main",
368+
} as AgentCommandOpts,
369+
outboundSession: {
370+
key: "agent:tester:main",
371+
agentId: "tester",
372+
} as never,
373+
sessionEntry: {
374+
sessionId: "session-1",
375+
updatedAt: 1,
376+
},
377+
resolveFreshSessionEntryForDelivery,
378+
payloads: [{ text: "final answer" }],
379+
result: createResult(),
380+
});
381+
382+
expect(resolveFreshSessionEntryForDelivery).toHaveBeenCalledTimes(1);
383+
expect(deliverOutboundPayloadsMock).toHaveBeenCalledTimes(1);
384+
const deliverArgs = latestOutboundDeliveryArgs();
385+
expect(deliverArgs.channel).toBe("slack");
386+
expect(deliverArgs.to).toBe("#fresh");
387+
expect(deliverArgs.accountId).toBe("workspace-1");
388+
expect(delivered.deliverySucceeded).toBe(true);
389+
expectDeliveryStatusFields(delivered, {
390+
requested: true,
391+
attempted: true,
392+
status: "sent",
393+
succeeded: true,
394+
resultCount: 1,
395+
});
396+
});
397+
332398
it("does not report success when best-effort delivery records an error", async () => {
333399
deliverOutboundPayloadsMock.mockImplementationOnce(async (params: unknown) => {
334400
(

‎src/agents/command/delivery.ts‎

Lines changed: 136 additions & 67 deletions
Original file line numberDiff line numberDiff line change
@@ -338,6 +338,7 @@ export async function deliverAgentCommandResult(params: {
338338
opts: AgentCommandOpts;
339339
outboundSession: OutboundSessionContext | undefined;
340340
sessionEntry: SessionEntry | undefined;
341+
resolveFreshSessionEntryForDelivery?: () => Promise<SessionEntry | undefined>;
341342
result: RunResult;
342343
payloads: RunResult["payloads"];
343344
}): Promise<AgentCommandDeliveryResult> {
@@ -349,76 +350,144 @@ export async function deliverAgentCommandResult(params: {
349350
const turnSourceTo = opts.runContext?.currentChannelId ?? opts.to;
350351
const turnSourceAccountId = opts.runContext?.accountId ?? opts.accountId;
351352
const turnSourceThreadId = opts.runContext?.currentThreadTs ?? opts.threadId;
352-
const deliveryPlan = resolveAgentDeliveryPlan({
353-
sessionEntry,
354-
requestedChannel: opts.replyChannel ?? opts.channel,
355-
explicitTo: opts.replyTo ?? opts.to,
356-
explicitThreadId: opts.threadId,
357-
accountId: opts.replyAccountId ?? opts.accountId,
358-
wantsDelivery: deliver,
359-
turnSourceChannel,
360-
turnSourceTo,
361-
turnSourceAccountId,
362-
turnSourceThreadId,
363-
});
364-
let deliveryChannel = deliveryPlan.resolvedChannel;
365353
const explicitChannelHint = (opts.replyChannel ?? opts.channel)?.trim();
366-
if (deliver && isInternalMessageChannel(deliveryChannel) && !explicitChannelHint) {
367-
try {
368-
const selection = await resolveMessageChannelSelection({ cfg });
369-
deliveryChannel = selection.channel;
370-
} catch {
371-
// Keep the internal channel marker; error handling below reports the failure.
354+
const resolveDeliveryRouting = async (candidateSessionEntry: SessionEntry | undefined) => {
355+
const deliveryPlan = resolveAgentDeliveryPlan({
356+
sessionEntry: candidateSessionEntry,
357+
requestedChannel: opts.replyChannel ?? opts.channel,
358+
explicitTo: opts.replyTo ?? opts.to,
359+
explicitThreadId: opts.threadId,
360+
accountId: opts.replyAccountId ?? opts.accountId,
361+
wantsDelivery: deliver,
362+
turnSourceChannel,
363+
turnSourceTo,
364+
turnSourceAccountId,
365+
turnSourceThreadId,
366+
});
367+
let deliveryChannel = deliveryPlan.resolvedChannel;
368+
if (deliver && isInternalMessageChannel(deliveryChannel) && !explicitChannelHint) {
369+
try {
370+
const selection = await resolveMessageChannelSelection({ cfg });
371+
deliveryChannel = selection.channel;
372+
} catch {
373+
// Keep the internal channel marker; error handling below reports the failure.
374+
}
372375
}
373-
}
374-
const effectiveDeliveryPlan =
375-
deliveryChannel === deliveryPlan.resolvedChannel
376-
? deliveryPlan
377-
: {
378-
...deliveryPlan,
379-
resolvedChannel: deliveryChannel,
380-
};
381-
// Channel docking: delivery channels are resolved via plugin registry.
382-
const deliveryPlugin =
383-
deliver && !isInternalMessageChannel(deliveryChannel)
384-
? getChannelPlugin(normalizeChannelId(deliveryChannel) ?? deliveryChannel)
385-
: undefined;
386-
387-
const isDeliveryChannelKnown =
388-
isInternalMessageChannel(deliveryChannel) || Boolean(deliveryPlugin);
376+
const effectiveDeliveryPlan =
377+
deliveryChannel === deliveryPlan.resolvedChannel
378+
? deliveryPlan
379+
: {
380+
...deliveryPlan,
381+
resolvedChannel: deliveryChannel,
382+
};
383+
// Channel docking: delivery channels are resolved via plugin registry.
384+
const deliveryPlugin =
385+
deliver && !isInternalMessageChannel(deliveryChannel)
386+
? getChannelPlugin(normalizeChannelId(deliveryChannel) ?? deliveryChannel)
387+
: undefined;
388+
const isDeliveryChannelKnown =
389+
isInternalMessageChannel(deliveryChannel) || Boolean(deliveryPlugin);
390+
const targetMode =
391+
opts.deliveryTargetMode ??
392+
effectiveDeliveryPlan.deliveryTargetMode ??
393+
(opts.to ? "explicit" : "implicit");
394+
const resolvedAccountId = effectiveDeliveryPlan.resolvedAccountId;
395+
const resolved =
396+
deliver && isDeliveryChannelKnown && deliveryChannel
397+
? resolveAgentOutboundTarget({
398+
cfg,
399+
plan: effectiveDeliveryPlan,
400+
targetMode,
401+
validateExplicitTarget: true,
402+
})
403+
: {
404+
resolvedTarget: null,
405+
resolvedTo: effectiveDeliveryPlan.resolvedTo,
406+
targetMode,
407+
};
408+
const resolvedThreadId = deliveryPlan.resolvedThreadId ?? opts.threadId;
409+
const replyTransport =
410+
deliveryPlugin?.threading?.resolveReplyTransport?.({
411+
cfg,
412+
accountId: resolvedAccountId,
413+
threadId: resolvedThreadId,
414+
}) ?? null;
415+
return {
416+
deliveryPlan,
417+
deliveryChannel,
418+
effectiveDeliveryPlan,
419+
deliveryPlugin,
420+
isDeliveryChannelKnown,
421+
targetMode,
422+
resolvedAccountId,
423+
resolved,
424+
resolvedTarget: resolved.resolvedTarget,
425+
deliveryTarget: resolved.resolvedTo,
426+
resolvedThreadId,
427+
resolvedReplyToId: replyTransport?.replyToId ?? undefined,
428+
resolvedThreadTarget:
429+
replyTransport && Object.hasOwn(replyTransport, "threadId")
430+
? (replyTransport.threadId ?? null)
431+
: (resolvedThreadId ?? null),
432+
};
433+
};
434+
const deliveryRoutingFailureReason = (
435+
route: Awaited<ReturnType<typeof resolveDeliveryRouting>>,
436+
): string | undefined => {
437+
if (!deliver) {
438+
return undefined;
439+
}
440+
if (isInternalMessageChannel(route.deliveryChannel)) {
441+
return "channel_resolved_to_internal";
442+
}
443+
if (!route.isDeliveryChannelKnown) {
444+
return "unknown_channel";
445+
}
446+
if (route.resolvedTarget && !route.resolvedTarget.ok) {
447+
return "invalid_delivery_target";
448+
}
449+
if (!route.deliveryTarget) {
450+
return "no_delivery_target";
451+
}
452+
return undefined;
453+
};
454+
const isRetryableFreshSessionRoutingFailure = (
455+
route: Awaited<ReturnType<typeof resolveDeliveryRouting>>,
456+
): boolean => {
457+
const reason = deliveryRoutingFailureReason(route);
458+
if (!reason) {
459+
return false;
460+
}
461+
if (reason === "unknown_channel") {
462+
return false;
463+
}
464+
return true;
465+
};
389466

390-
const targetMode =
391-
opts.deliveryTargetMode ??
392-
effectiveDeliveryPlan.deliveryTargetMode ??
393-
(opts.to ? "explicit" : "implicit");
394-
const resolvedAccountId = effectiveDeliveryPlan.resolvedAccountId;
395-
const resolved =
396-
deliver && isDeliveryChannelKnown && deliveryChannel
397-
? resolveAgentOutboundTarget({
398-
cfg,
399-
plan: effectiveDeliveryPlan,
400-
targetMode,
401-
validateExplicitTarget: true,
402-
})
403-
: {
404-
resolvedTarget: null,
405-
resolvedTo: effectiveDeliveryPlan.resolvedTo,
406-
targetMode,
407-
};
408-
const resolvedTarget = resolved.resolvedTarget;
409-
const deliveryTarget = resolved.resolvedTo;
410-
const resolvedThreadId = deliveryPlan.resolvedThreadId ?? opts.threadId;
411-
const replyTransport =
412-
deliveryPlugin?.threading?.resolveReplyTransport?.({
413-
cfg,
414-
accountId: resolvedAccountId,
415-
threadId: resolvedThreadId,
416-
}) ?? null;
417-
const resolvedReplyToId = replyTransport?.replyToId ?? undefined;
418-
const resolvedThreadTarget =
419-
replyTransport && Object.hasOwn(replyTransport, "threadId")
420-
? (replyTransport.threadId ?? null)
421-
: (resolvedThreadId ?? null);
467+
let deliveryRouting = await resolveDeliveryRouting(sessionEntry);
468+
if (isRetryableFreshSessionRoutingFailure(deliveryRouting)) {
469+
const freshSessionEntry = await params.resolveFreshSessionEntryForDelivery?.();
470+
if (freshSessionEntry && freshSessionEntry !== sessionEntry) {
471+
const freshRouting = await resolveDeliveryRouting(freshSessionEntry);
472+
if (!deliveryRoutingFailureReason(freshRouting)) {
473+
if (!opts.json) {
474+
runtime.log(
475+
`[delivery] refreshed session routing before final delivery (session=${effectiveSessionKey ?? "unknown"} channel=${freshRouting.deliveryChannel})`,
476+
);
477+
}
478+
deliveryRouting = freshRouting;
479+
}
480+
}
481+
}
482+
const {
483+
deliveryChannel,
484+
isDeliveryChannelKnown,
485+
resolvedAccountId,
486+
resolvedTarget,
487+
deliveryTarget,
488+
resolvedReplyToId,
489+
resolvedThreadTarget,
490+
} = deliveryRouting;
422491

423492
let deliveryLoggedError = false;
424493
const logDeliveryError = (err: unknown) => {

‎src/agents/command/session-store.runtime.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1 +1,2 @@
11
export { updateSessionStoreAfterAgentRun } from "./session-store.js";
2+
export { loadSessionStore } from "../../config/sessions.js";

0 commit comments

Comments
 (0)