github/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-04-19 09:38:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: unify typing dispatch lifecycle and policy boundaries

Browse Source
This commit is contained in:
Peter Steinberger
2026-02-26 17:36:09 +01:00
parent 6fd9ec97de
commit 273973d374
19 changed files with 420 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
extensions/bluebubbles/src/monitor.test.ts
View File

@@ -162,6 +162,24 @@ function createMockRuntime(): PluginRuntime {
vi.fn() as unknown as PluginRuntime["channel"]["reply"]["resolveHumanDelayConfig"],
dispatchReplyFromConfig:
vi.fn() as unknown as PluginRuntime["channel"]["reply"]["dispatchReplyFromConfig"],
withReplyDispatcher: vi.fn(
async ({
dispatcher,
run,
onSettled,
}: Parameters<PluginRuntime["channel"]["reply"]["withReplyDispatcher"]>[0]) => {
try {
return await run();
} finally {
dispatcher.markComplete();
try {
await dispatcher.waitForIdle();
} finally {
await onSettled?.();
}
}
},
) as unknown as PluginRuntime["channel"]["reply"]["withReplyDispatcher"],
finalizeInboundContext: vi.fn(
(ctx: Record<string, unknown>) => ctx,
) as unknown as PluginRuntime["channel"]["reply"]["finalizeInboundContext"],

19
extensions/feishu/src/bot.test.ts
View File

@@ -90,6 +90,24 @@ describe("handleFeishuMessage command authorization", () => {
const mockDispatchReplyFromConfig = vi
.fn()
.mockResolvedValue({ queuedFinal: false, counts: { final: 1 } });
const mockWithReplyDispatcher = vi.fn(
async ({
dispatcher,
run,
onSettled,
}: Parameters<PluginRuntime["channel"]["reply"]["withReplyDispatcher"]>[0]) => {
try {
return await run();
} finally {
dispatcher.markComplete();
try {
await dispatcher.waitForIdle();
} finally {
await onSettled?.();
}
}
},
);
const mockResolveCommandAuthorizedFromAuthorizers = vi.fn(() => false);
const mockShouldComputeCommandAuthorized = vi.fn(() => true);
const mockReadAllowFromStore = vi.fn().mockResolvedValue([]);
@@ -127,6 +145,7 @@ describe("handleFeishuMessage command authorization", () => {
formatAgentEnvelope: vi.fn((params: { body: string }) => params.body),
finalizeInboundContext: mockFinalizeInboundContext,
dispatchReplyFromConfig: mockDispatchReplyFromConfig,
withReplyDispatcher: mockWithReplyDispatcher,
},
commands: {
shouldComputeCommandAuthorized: mockShouldComputeCommandAuthorized,

48
extensions/feishu/src/bot.ts
View File

@@ -943,33 +943,31 @@ export async function handleFeishuMessage(params: {
});
log(`feishu[${account.accountId}]: dispatching to agent (session=${route.sessionKey})`);
try {
const { queuedFinal, counts } = await core.channel.reply.dispatchReplyFromConfig({
ctx: ctxPayload,
cfg,
dispatcher,
replyOptions,
});
if (isGroup && historyKey && chatHistories) {
clearHistoryEntriesIfEnabled({
historyMap: chatHistories,
historyKey,
limit: historyLimit,
});
}
log(
`feishu[${account.accountId}]: dispatch complete (queuedFinal=${queuedFinal}, replies=${counts.final})`,
);
} finally {
dispatcher.markComplete();
try {
await dispatcher.waitForIdle();
} finally {
const { queuedFinal, counts } = await core.channel.reply.withReplyDispatcher({
dispatcher,
onSettled: () => {
markDispatchIdle();
}
},
run: () =>
core.channel.reply.dispatchReplyFromConfig({
ctx: ctxPayload,
cfg,
dispatcher,
replyOptions,
}),
});
if (isGroup && historyKey && chatHistories) {
clearHistoryEntriesIfEnabled({
historyMap: chatHistories,
historyKey,
limit: historyLimit,
});
}
log(
`feishu[${account.accountId}]: dispatch complete (queuedFinal=${queuedFinal}, replies=${counts.final})`,
);
} catch (err) {
error(`feishu[${account.accountId}]: failed to dispatch message: ${String(err)}`);
}

62
extensions/matrix/src/matrix/monitor/handler.ts
View File

@@ -655,39 +655,37 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
},
});
try {
const { queuedFinal, counts } = await core.channel.reply.dispatchReplyFromConfig({
ctx: ctxPayload,
cfg,
dispatcher,
replyOptions: {
...replyOptions,
skillFilter: roomConfig?.skills,
onModelSelected,
},
});
if (!queuedFinal) {
return;
}
didSendReply = true;
const finalCount = counts.final;
logVerboseMessage(
`matrix: delivered ${finalCount} reply${finalCount === 1 ? "" : "ies"} to ${replyTarget}`,
);
if (didSendReply) {
const previewText = bodyText.replace(/\s+/g, " ").slice(0, 160);
core.system.enqueueSystemEvent(`Matrix message from ${senderName}: ${previewText}`, {
sessionKey: route.sessionKey,
contextKey: `matrix:message:${roomId}:${messageId || "unknown"}`,
});
}
} finally {
dispatcher.markComplete();
try {
await dispatcher.waitForIdle();
} finally {
const { queuedFinal, counts } = await core.channel.reply.withReplyDispatcher({
dispatcher,
onSettled: () => {
markDispatchIdle();
}
},
run: () =>
core.channel.reply.dispatchReplyFromConfig({
ctx: ctxPayload,
cfg,
dispatcher,
replyOptions: {
...replyOptions,
skillFilter: roomConfig?.skills,
onModelSelected,
},
}),
});
if (!queuedFinal) {
return;
}
didSendReply = true;
const finalCount = counts.final;
logVerboseMessage(
`matrix: delivered ${finalCount} reply${finalCount === 1 ? "" : "ies"} to ${replyTarget}`,
);
if (didSendReply) {
const previewText = bodyText.replace(/\s+/g, " ").slice(0, 160);
core.system.enqueueSystemEvent(`Matrix message from ${senderName}: ${previewText}`, {
sessionKey: route.sessionKey,
contextKey: `matrix:message:${roomId}:${messageId || "unknown"}`,
});
}
} catch (err) {
runtime.error?.(`matrix handler failed: ${String(err)}`);

48
extensions/mattermost/src/mattermost/monitor.ts
View File

@@ -772,32 +772,30 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {}
},
});
try {
await core.channel.reply.dispatchReplyFromConfig({
ctx: ctxPayload,
cfg,
dispatcher,
replyOptions: {
...replyOptions,
disableBlockStreaming:
typeof account.blockStreaming === "boolean" ? !account.blockStreaming : undefined,
onModelSelected,
},
});
if (historyKey) {
clearHistoryEntriesIfEnabled({
historyMap: channelHistories,
historyKey,
limit: historyLimit,
});
}
} finally {
dispatcher.markComplete();
try {
await dispatcher.waitForIdle();
} finally {
await core.channel.reply.withReplyDispatcher({
dispatcher,
onSettled: () => {
markDispatchIdle();
}
},
run: () =>
core.channel.reply.dispatchReplyFromConfig({
ctx: ctxPayload,
cfg,
dispatcher,
replyOptions: {
...replyOptions,
disableBlockStreaming:
typeof account.blockStreaming === "boolean" ? !account.blockStreaming : undefined,
onModelSelected,
},
}),
});
if (historyKey) {
clearHistoryEntriesIfEnabled({
historyMap: channelHistories,
historyKey,
limit: historyLimit,
});
}
};

56
extensions/msteams/src/monitor-handler/message-handler.ts
View File

@@ -533,30 +533,23 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) {
log.info("dispatching to agent", { sessionKey: route.sessionKey });
try {
try {
const { queuedFinal, counts } = await core.channel.reply.dispatchReplyFromConfig({
ctx: ctxPayload,
cfg,
dispatcher,
replyOptions,
});
const { queuedFinal, counts } = await core.channel.reply.withReplyDispatcher({
dispatcher,
onSettled: () => {
markDispatchIdle();
},
run: () =>
core.channel.reply.dispatchReplyFromConfig({
ctx: ctxPayload,
cfg,
dispatcher,
replyOptions,
}),
});
log.info("dispatch complete", { queuedFinal, counts });
log.info("dispatch complete", { queuedFinal, counts });
if (!queuedFinal) {
if (isRoomish && historyKey) {
clearHistoryEntriesIfEnabled({
historyMap: conversationHistories,
historyKey,
limit: historyLimit,
});
}
return;
}
const finalCount = counts.final;
logVerboseMessage(
`msteams: delivered ${finalCount} reply${finalCount === 1 ? "" : "ies"} to ${teamsTo}`,
);
if (!queuedFinal) {
if (isRoomish && historyKey) {
clearHistoryEntriesIfEnabled({
historyMap: conversationHistories,
@@ -564,13 +557,18 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) {
limit: historyLimit,
});
}
} finally {
dispatcher.markComplete();
try {
await dispatcher.waitForIdle();
} finally {
markDispatchIdle();
}
return;
}
const finalCount = counts.final;
logVerboseMessage(
`msteams: delivered ${finalCount} reply${finalCount === 1 ? "" : "ies"} to ${teamsTo}`,
);
if (isRoomish && historyKey) {
clearHistoryEntriesIfEnabled({
historyMap: conversationHistories,
historyKey,
limit: historyLimit,
});
}
} catch (err) {
log.error("dispatch failed", { error: String(err) });