From 9aeeae1c48f95fe3c45cf791169e9168b5d8668b Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sun, 30 Aug 2026 03:48:14 -0700 Subject: [PATCH] Commonly: forward delivery nonce on acknowledgement --- extensions/commonly/src/channel.ts | 54 +++++++++++++++----------- extensions/commonly/src/client.test.ts | 18 +++++++++ extensions/commonly/src/client.ts | 3 +- extensions/commonly/src/events.ts | 3 ++ 4 files changed, 55 insertions(+), 23 deletions(-) diff --git a/extensions/commonly/src/channel.ts b/extensions/commonly/src/channel.ts index a5453de664531..0a2a8bf9576f6 100644 --- a/extensions/commonly/src/channel.ts +++ b/extensions/commonly/src/channel.ts @@ -1,3 +1,4 @@ +import { readFileSync } from "node:fs"; import { buildChannelConfigSchema, createReplyPrefixContext, @@ -5,12 +6,10 @@ import { type ChannelPlugin, type ReplyPayload, } from "openclaw/plugin-sdk"; - import { CommonlyClient } from "./client.js"; -import { CommonlyWebSocket } from "./websocket.js"; -import type { CommonlyEvent } from "./events.js"; - import { CommonlyConfigSchema } from "./config-schema.js"; +import { parseInlineDirectives } from "./directive-tags.js"; +import type { CommonlyEvent } from "./events.js"; import { getCommonlyRuntime } from "./runtime.js"; import { listCommonlyAccountIds, @@ -18,8 +17,7 @@ import { resolveDefaultCommonlyAccountId, type ResolvedCommonlyAccount, } from "./types.js"; -import { parseInlineDirectives } from "./directive-tags.js"; -import { readFileSync } from "node:fs"; +import { CommonlyWebSocket } from "./websocket.js"; type CommonlyConnection = { ws: CommonlyWebSocket; @@ -29,7 +27,10 @@ type CommonlyConnection = { const activeConnections = new Map(); const normalizePodId = (raw: string) => - raw.replace(/^commonly:/i, "").replace(/^pod:/i, "").trim(); + raw + .replace(/^commonly:/i, "") + .replace(/^pod:/i, "") + .trim(); const buildSummaryMessage = (summary?: CommonlyEvent["payload"]["summary"]): string => { if (!summary) return ""; @@ -69,9 +70,11 @@ const formatEnsembleTurnBody = (event: CommonlyEvent): string => { const lines: string[] = []; lines.push(`Ensemble topic: ${context.topic}`); lines.push(`Turn: ${context.turnNumber} (round ${context.roundNumber})`); - lines.push(context.isStarter - ? "You are the starter. Provide the opening message." - : "You are responding to the ongoing discussion."); + lines.push( + context.isStarter + ? "You are the starter. Provide the opening message." + : "You are responding to the ongoing discussion.", + ); const participants = event.payload?.participants || []; if (participants.length > 0) { @@ -179,7 +182,10 @@ export const commonlyPlugin: ChannelPlugin = { deliveryMode: "direct", textChunkLimit: 8000, sendText: async ({ to, text, threadId, accountId }) => { - const account = resolveCommonlyAccount({ cfg: getCommonlyRuntime().config.loadConfig(), accountId }); + const account = resolveCommonlyAccount({ + cfg: getCommonlyRuntime().config.loadConfig(), + accountId, + }); const client = new CommonlyClient({ baseUrl: account.baseUrl, runtimeToken: account.runtimeToken, @@ -198,7 +204,10 @@ export const commonlyPlugin: ChannelPlugin = { return { channel: "commonly", messageId: `${podId}:${Date.now()}` }; }, sendMedia: async ({ to, text, mediaUrl, threadId, accountId }) => { - const account = resolveCommonlyAccount({ cfg: getCommonlyRuntime().config.loadConfig(), accountId }); + const account = resolveCommonlyAccount({ + cfg: getCommonlyRuntime().config.loadConfig(), + accountId, + }); const client = new CommonlyClient({ baseUrl: account.baseUrl, runtimeToken: account.runtimeToken, @@ -207,10 +216,9 @@ export const commonlyPlugin: ChannelPlugin = { instanceId: account.instanceId, }); const podId = normalizePodId(to); - const message = [ - sanitizeOutboundText(text ?? ""), - mediaUrl?.trim() || "", - ].filter(Boolean).join("\n"); + const message = [sanitizeOutboundText(text ?? ""), mediaUrl?.trim() || ""] + .filter(Boolean) + .join("\n"); if (threadId) { await client.postThreadComment(String(threadId), message); return { channel: "commonly", messageId: String(threadId) }; @@ -345,7 +353,7 @@ export const commonlyPlugin: ChannelPlugin = { ); } if (event._id) { - await client.ackEvent(event._id); + await client.ackEvent(event._id, event.payload?.deliveryId); ctx.log?.info?.(`[${connectionKey}] summary.request acked id=${eventId}`); } return; @@ -353,9 +361,11 @@ export const commonlyPlugin: ChannelPlugin = { const rawContent = resolveInboundBody(event); if (!rawContent) { - ctx.log?.info?.(`[${connectionKey}] event skipped (empty body) id=${eventId} type=${event.type}`); + ctx.log?.info?.( + `[${connectionKey}] event skipped (empty body) id=${eventId} type=${event.type}`, + ); if (event._id) { - await client.ackEvent(event._id); + await client.ackEvent(event._id, event.payload?.deliveryId); ctx.log?.info?.(`[${connectionKey}] empty-body acked id=${eventId}`); } return; @@ -495,8 +505,8 @@ export const commonlyPlugin: ChannelPlugin = { heartbeatTrigger: event.payload?.trigger, }); ctx.log?.info?.( - `[${connectionKey}] message posted id=${eventId} pod=${podId} chars=${message.length} ` - + `postedId=${String(posted?.id || "n/a")}`, + `[${connectionKey}] message posted id=${eventId} pod=${podId} chars=${message.length} ` + + `postedId=${String(posted?.id || "n/a")}`, ); if (event.type === "ensemble.turn" && !ensembleResponseSent) { const ensembleId = event.payload?.ensembleId; @@ -526,7 +536,7 @@ export const commonlyPlugin: ChannelPlugin = { }); if (event._id) { - await client.ackEvent(event._id); + await client.ackEvent(event._id, event.payload?.deliveryId); ctx.log?.info?.(`[${connectionKey}] event acked id=${eventId} type=${event.type}`); } }); diff --git a/extensions/commonly/src/client.test.ts b/extensions/commonly/src/client.test.ts index 4dc5c13e13583..5fba4246e7682 100644 --- a/extensions/commonly/src/client.test.ts +++ b/extensions/commonly/src/client.test.ts @@ -36,6 +36,24 @@ describe("CommonlyClient", () => { ); }); + it("echoes the claimed deliveryId when acknowledging", async () => { + fetchMock.mockResolvedValue(createResponse()); + const client = new CommonlyClient({ + baseUrl: "http://localhost:5000", + runtimeToken: "rt", + }); + + await client.ackEvent("event-1", "claimed-child"); + + expect(fetchMock).toHaveBeenCalledWith( + "http://localhost:5000/api/agents/runtime/events/event-1/ack", + expect.objectContaining({ + method: "POST", + body: JSON.stringify({ deliveryId: "claimed-child" }), + }), + ); + }); + it("uses user token for user endpoints when provided", async () => { fetchMock.mockResolvedValue(createResponse({ results: [] })); const client = new CommonlyClient({ diff --git a/extensions/commonly/src/client.ts b/extensions/commonly/src/client.ts index baaf92ab75e3e..b79dfc31598dc 100644 --- a/extensions/commonly/src/client.ts +++ b/extensions/commonly/src/client.ts @@ -172,10 +172,11 @@ export class CommonlyClient { /** * Acknowledge an event */ - async ackEvent(eventId: string): Promise { + async ackEvent(eventId: string, deliveryId?: string): Promise { const res = await fetch(`${this.config.baseUrl}/api/agents/runtime/events/${eventId}/ack`, { method: "POST", headers: this.runtimeHeaders, + body: JSON.stringify(typeof deliveryId === "string" && deliveryId ? { deliveryId } : {}), }); if (!res.ok) { throw new Error(`Failed to ack event: ${res.status}`); diff --git a/extensions/commonly/src/events.ts b/extensions/commonly/src/events.ts index a50d65a02d615..337730776fbd7 100644 --- a/extensions/commonly/src/events.ts +++ b/extensions/commonly/src/events.ts @@ -6,6 +6,9 @@ export type CommonlyEventType = | (string & {}); export type CommonlyEventPayload = { + // ADR-026 D6: minted by the kernel when this event is claimed and echoed + // with the acknowledgement so a stale delivery cannot settle a replacement. + deliveryId?: string; messageId?: string; content?: string; userId?: string;