vrr/openclaw
2
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-10-25 16:37:51 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 30

refactor(channels): deslop channel lifecycle (#165431)

Browse source 
* refactor(channels): deslop channel lifecycle

Share account-scoped binding projection and adapter dispatch, native accepted
receipt accumulation, and Markdown UTF-16 edit projection. Keep platform
storage, expiry, streaming and delivery policy with their existing owners.
Remove redundant Teams settlement state and Mattermost preview wrappers.

Fix Zalo retaining a replaced runtime's media store and Synology releasing
replacement-runtime media limits from older in-flight operations. Preserve
persisted namespaces, capacities, expiry and response behavior.

* refactor(plugin-sdk): remove unused public helper exports

* refactor(channels): finish lifecycle caller cutover

* fix(channels): preserve binding serialization and repair CI contracts

* chore: prune repaired Mattermost mock baseline

* fix(channels): settle CI boundaries and finish SDK offsets

* test(codex): simplify cleanup fixture scheduling

* test(codex): use the native promise scheduler
This commit is contained in:
Peter Steinberger 2026-10-05 05:56:48 -07:00 • committed by GitHub
parent 044859704e
commit 2fec39f573
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
55 changed files with 825 additions and 1081 deletions
Download patch file Download diff file Expand all files Collapse all files

1
config/test-mock-exports-baseline.txt
View file

@ -1107,7 +1107,6 @@
["extensions/mattermost/src/mattermost/monitor-slash.test.ts","./slash-commands.js","7a044f1c0cfc629a7724bbdff1facf3b90f9739957158834f26b95f1e33f50a7",1]
["extensions/mattermost/src/mattermost/monitor-slash.test.ts","./slash-state.js","b874789c472dc18402977ba1108d5b0e64d2eb615e531a4a44f4fc29e8c6f08f",1]
["extensions/mattermost/src/mattermost/monitor.inbound-system-event.test.ts","../runtime.js","109e7a1a5f483040b7544c1214db4f7d5a5648492ad49eb806f50b8a48680057",1]
["extensions/mattermost/src/mattermost/monitor.inbound-system-event.test.ts","./draft-stream.js","5f489871a4645a60df67a255c84948465ebc9762e63cc8f19bfb4b39c85e6efb",1]
["extensions/mattermost/src/mattermost/monitor.inbound-system-event.test.ts","./monitor-slash.js","b2bdf86c8ec9db3b5f7648679107c0e5981a2e7761d14f6939766a55bc6a25f6",1]
["extensions/mattermost/src/mattermost/monitor.reaction-thread-session.test.ts","../runtime.js","f3b19ed3246a6d6bf97e0169c1de390e2afbc133f98710c379e85bbc2e20070a",1]
["extensions/mattermost/src/mattermost/send.loopback.test.ts","../runtime.js","4305da0537101acf8c38c6a59b2a23e2dc30c1b3ab21f9362e91cd076750f2fb",1]

7
docs/concepts/markdown-formatting.md
View file

@ -146,6 +146,13 @@ content is hidden or lost.
5. **Test** with format tests plus an outbound delivery test if the channel
chunks.
For disjoint text replacements and insertions, `openclaw/plugin-sdk/text-chunking`
exports `applyMarkdownTextEdits(text, edits)`. Edits use UTF-16 source coordinates;
the result contains the edited `text` and `mapOffset` for boundaries outside
replaced interiors. Each edit shifts offsets at and after its end. Insertions at
the same position retain caller order and precede a replacement starting there.
Renderers decide which styles include replacement text.
## Common gotchas
- Slack angle-bracket tokens (`<@U123>`, `<#C123>`, `<https://...>`) must

12
docs/plugins/sdk-channel-plugins/message-adapter.md
View file

@ -39,8 +39,16 @@ A logical result's receipt takes precedence over its legacy message IDs.
The result carries a receipt, `messageIds` (including an empty array), and
`visibleReplySent: true`; routing fields stay in the receipt. Optional `content`
is passed through, and `kind` and `replyToId` use the receipt builder's rules.
Keep acceptance side effects, content joining,
suppression, and whether an identityless outcome needs a receipt in the adapter.
For batches that join accepted text with newlines, use
`createChannelDeliveryAccumulator({ kind?, replyToId? })` from
`openclaw/plugin-sdk/channel-outbound`.
Call `add(source, acceptedText?)` only after each physical send succeeds. `size`
counts accepted sends, and `result()` returns their combined receipt and nonempty
text, or a `no_visible_result` suppression for an empty batch. On failure, throw
`partialError(error)` to retain earlier sends and a nested partial-delivery
error's accepted subset; failures before any acceptance pass through unchanged.
Keep transport acceptance side effects, other content-joining rules, and whether
an identityless outcome needs a receipt in the adapter.
Channel actions and adapter capabilities come from the selected plugin
registration. An omitted `actions`, `message`, or `outbound` surface is not

8
docs/plugins/sdk-channel-plugins/sessions-and-bindings.md
View file

@ -130,6 +130,14 @@ pass prepared `inactivityExpiresAt` and `maxAgeExpiresAt` values to
deadline and its reason, preferring idle expiration on ties; omitted deadlines
are disabled. The plugin still owns timestamp validation and duration defaults.
On the same subpath, `projectThreadBindingRecord(...)` projects prepared
conversation identity and expiry, with an explicit metadata merge callback.
`createAccountScopedBindingAdapter(...)` shares list, resolve, touch, and unbind
dispatch for account-prefixed binding IDs. The plugin's manager still owns
storage, mutation admission and settlement, child creation, and shutdown.
Platforms with different identity grammars, such as Matrix parent-room keys,
can use the projection without the account-prefixed adapter.
Preserve opaque plugin ownership metadata when projecting binding records.
Plugin-owned targets do not require an OpenClaw agent id; use
`isPluginOwnedSessionBindingRecord(...)` from

35
extensions/codex/src/app-server/run-attempt-thread-cleanup.test.ts
View file

@ -1,5 +1,6 @@
// Codex tests cover run attempt thread cleanup plugin behavior.
import path from "node:path";
import { setImmediate as yieldEventLoop } from "node:timers/promises";
import type { EmbeddedRunAttemptParamsV2 as EmbeddedRunAttemptParams } from "openclaw/plugin-sdk/agent-harness-runtime";
import { createDeferred } from "openclaw/plugin-sdk/extension-shared";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
@ -734,8 +735,11 @@ describe("Codex app-server main thread cleanup", () => {
? { id: interrupt.id, error: { code: -32_000, message: "startup interrupt failed" } }
: { id: interrupt.id, result: {} },
);
// Cancellation revokes native row admission, so cleanup retires the exact
// client rather than sending an unsubscribe under the canceled owner.
if (!interruptFails) {
const unsubscribe = await waitForHarnessRequest(harness, "thread/unsubscribe");
expect(unsubscribe.params).toEqual({ threadId: "thread-1" });
harness.send({ id: unsubscribe.id, result: {} });
}
await expect(failure).resolves.toMatchObject({
message: "turn/start aborted: cancelled",
cause: "cancelled",
@ -750,8 +754,9 @@ describe("Codex app-server main thread cleanup", () => {
"thread/start",
"turn/start",
"turn/interrupt",
...(!interruptFails ? ["thread/unsubscribe"] : []),
]);
expect(harness.stdinDestroyed).toBe(true);
expect(harness.stdinDestroyed).toBe(interruptFails);
},
);
@ -838,9 +843,7 @@ describe("Codex app-server main thread cleanup", () => {
harness.send({ id: threadStart.id, result: threadStartResult() });
const turnStart = await waitForHarnessRequest(harness, "turn/start");
harness.send({ id: turnStart.id, result: turnStartResult() });
await new Promise<void>((resolve) => {
setImmediate(resolve);
});
await yieldEventLoop();
abort.abort("cancelled");
const interrupt = await waitForHarnessRequest(harness, "turn/interrupt");
@ -849,9 +852,7 @@ describe("Codex app-server main thread cleanup", () => {
params: { threadId: "thread-1", turnId: "turn-1" },
});
harness.send({ id: interrupt.id, result: {} });
await new Promise<void>((resolve) => {
setImmediate(resolve);
});
await yieldEventLoop();
expect(settled).toBe(false);
expect(harness.writes.map((entry) => JSON.parse(entry).method)).not.toContain(
"thread/unsubscribe",
@ -909,16 +910,22 @@ describe("Codex app-server main thread cleanup", () => {
);
harness.send({ id: confirmation.id, result: { data: [], nextCursor: null } });
}
const unsubscribe = await waitForHarnessRequest(harness, "thread/unsubscribe");
expect(unsubscribe.params).toEqual({ threadId: "thread-1" });
expect(settled).toBe(false);
harness.send({ id: unsubscribe.id, result: {} });
if (rejected) {
await rejected;
} else {
expect(readAttemptTerminal(await run)).toMatchObject({ aborted: true, timedOut: false });
}
expect(harness.writes.map((entry) => JSON.parse(entry).method)).not.toContain(
"thread/unsubscribe",
);
expect(close).toHaveBeenCalledOnce();
expect(harness.stdinDestroyed).toBe(true);
expect(
harness.writes
.map((entry) => JSON.parse(entry).method)
.filter((method) => method === "thread/unsubscribe"),
).toEqual(["thread/unsubscribe"]);
expect(close).not.toHaveBeenCalled();
expect(harness.stdinDestroyed).toBe(false);
},
);

29
extensions/copilot/src/isolated-completion.ts
View file

@ -80,21 +80,22 @@ async function awaitWithinCompletionBoundary<T>(params: {
};
// Start only after the abort listener exists. Pool/session factories may
// synchronously trip cancellation before returning their promise.
const operation = Promise.resolve()
.then(() => {
assertCurrent();
return params.start(remainingMs);
})
.then((value) => {
try {
const operation = () =>
Promise.resolve()
.then(() => {
assertCurrent();
return value;
} catch (error) {
// Retirement can reject an acquired resource before its caller owns cleanup.
startBestEffortCleanup(async () => await params.cleanupLate?.(value));
throw error;
}
});
return params.start(remainingMs);
})
.then((value) => {
try {
assertCurrent();
return value;
} catch (error) {
// Retirement can reject an acquired resource before its caller owns cleanup.
startBestEffortCleanup(async () => await params.cleanupLate?.(value));
throw error;
}
});
try {
return await raceWithTimeout(
operation,

93
extensions/discord/src/monitor/thread-bindings.session-adapter.ts
View file

@ -1,7 +1,6 @@
import {
registerSessionBindingAdapter,
unregisterSessionBindingAdapter,
resolveThreadBindingConversationIdFromBindingId,
type SessionBindingAdapter,
type SessionBindingRecord,
} from "openclaw/plugin-sdk/conversation-runtime";
@ -11,6 +10,10 @@ import {
asOptionalObjectRecord,
normalizeOptionalString,
} from "openclaw/plugin-sdk/string-coerce-runtime";
import {
createAccountScopedBindingAdapter,
projectThreadBindingRecord,
} from "openclaw/plugin-sdk/thread-bindings-session-runtime";
import {
normalizeDiscordBindingChannelId,
resolveChannelIdForBinding,
@ -42,31 +45,22 @@ function toSessionBindingRecord(
threadId: record.threadId,
}) ?? `${record.accountId}:${record.threadId}`;
const lifecycle = resolvePreparedThreadBindingLifecycle({ record, ...defaults });
return {
bindingId,
targetSessionKey: record.targetSessionKey,
targetKind: record.targetKind === "subagent" ? "subagent" : "session",
return projectThreadBindingRecord(record, {
conversation: {
channel: "discord",
accountId: record.accountId,
conversationId: record.threadId,
parentConversationId: record.channelId,
},
status: "active",
boundAt: record.boundAt,
expiresAt: lifecycle.expiresAt,
metadata: {
agentId: record.agentId,
label: record.label,
bindingId,
targetKind: record.targetKind === "subagent" ? "subagent" : "session",
lifecycle,
metadata: (lifecycleMetadata) => ({
...lifecycleMetadata,
webhookId: record.webhookId,
webhookToken: record.webhookToken,
boundBy: record.boundBy,
lastActivityAt: record.lastActivityAt,
idleTimeoutMs: lifecycle.idleTimeoutMs,
maxAgeMs: lifecycle.maxAgeMs,
...record.metadata,
},
};
}),
});
}
export function createThreadBindingSessionAdapter(params: {
@ -79,7 +73,7 @@ export function createThreadBindingSessionAdapter(params: {
const serializeBinding = (entry: ThreadBindingRecord) =>
toSessionBindingRecord(entry, params.defaults);
return {
return createAccountScopedBindingAdapter({
channel: "discord",
accountId: params.accountId,
capabilities: {
@ -143,56 +137,17 @@ export function createThreadBindingSessionAdapter(params: {
});
return bound ? serializeBinding(bound) : null;
},
listBySession: (targetSessionKey) =>
params.manager.listBySessionKey(targetSessionKey).map(serializeBinding),
resolveByConversation: (ref) => {
if (ref.channel !== "discord") {
return null;
}
const binding = params.manager.getByThreadId(ref.conversationId);
return binding ? serializeBinding(binding) : null;
},
touch: (bindingId, at) => {
const threadId = resolveThreadBindingConversationIdFromBindingId({
accountId: params.accountId,
bindingId,
});
if (threadId) {
params.manager.touchThreadSync({ threadId, at, persist: true });
}
},
touchAsync: async (bindingId, at) => {
const threadId = resolveThreadBindingConversationIdFromBindingId({
accountId: params.accountId,
bindingId,
});
if (!threadId) {
return;
}
await params.manager.touchThread({ threadId, at, persist: true });
},
unbind: async (input) => {
if (input.targetSessionKey?.trim()) {
const removed = await params.manager.unbindBySessionKey({
targetSessionKey: input.targetSessionKey,
reason: input.reason,
});
return removed.map(serializeBinding);
}
const threadId = resolveThreadBindingConversationIdFromBindingId({
accountId: params.accountId,
bindingId: input.bindingId,
});
if (!threadId) {
return [];
}
const removed = await params.manager.unbindThread({
threadId,
reason: input.reason,
});
return removed ? [serializeBinding(removed)] : [];
},
};
project: serializeBinding,
listBySessionKey: params.manager.listBySessionKey,
getByConversation: (ref) => params.manager.getByThreadId(ref.conversationId),
touchConversation: (threadId, at) =>
params.manager.touchThreadSync({ threadId, at, persist: true }),
touchConversationAsync: (threadId, at) =>
params.manager.touchThread({ threadId, at, persist: true }),
unbindConversation: (threadId, reason) => params.manager.unbindThread({ threadId, reason }),
unbindBySessionKey: (targetSessionKey, reason) =>
params.manager.unbindBySessionKey({ targetSessionKey, reason }),
});
}
/** Disabled bindings have a live empty owner; retirement still makes that owner unavailable. */

7
extensions/discord/src/monitor/thread-bindings.session-shared.ts
View file

@ -19,13 +19,6 @@ import type {
ThreadBindingTargetKind,
} from "./thread-bindings.types.js";
export function normalizeNonNegativeMs(raw: number): number {
if (!Number.isFinite(raw)) {
return 0;
}
return Math.max(0, Math.floor(raw));
}
export function resolveBindingIdsForTargetSession(params: {
targetSessionKey: string;
accountId?: string;

10
extensions/discord/src/monitor/thread-bindings.session-updates.ts
View file

@ -1,5 +1,5 @@
import { resolveNonNegativeIntegerOption } from "openclaw/plugin-sdk/number-runtime";
import {
normalizeNonNegativeMs,
resolveBindingIdsForTargetSession,
mutateBindingsForTargetSession,
updateBindingsForTargetSessionSync,
@ -12,7 +12,7 @@ export async function setThreadBindingIdleTimeoutBySessionKeyAsync(input: {
idleTimeoutMs: number;
}): Promise<ThreadBindingRecord[]> {
const params = { ...input };
const idleTimeoutMs = normalizeNonNegativeMs(params.idleTimeoutMs);
const idleTimeoutMs = resolveNonNegativeIntegerOption(params.idleTimeoutMs, 0);
return mutateBindingsForTargetSession(params, (existing, now) => ({
...existing,
idleTimeoutMs,
@ -26,7 +26,7 @@ export async function setThreadBindingMaxAgeBySessionKeyAsync(input: {
maxAgeMs: number;
}): Promise<ThreadBindingRecord[]> {
const params = { ...input };
const maxAgeMs = normalizeNonNegativeMs(params.maxAgeMs);
const maxAgeMs = resolveNonNegativeIntegerOption(params.maxAgeMs, 0);
return mutateBindingsForTargetSession(params, (existing, now) => ({
...existing,
maxAgeMs,
@ -40,7 +40,7 @@ export function setThreadBindingIdleTimeoutBySessionKey(
params: Parameters<typeof setThreadBindingIdleTimeoutBySessionKeyAsync>[0],
): ThreadBindingRecord[] {
const ids = resolveBindingIdsForTargetSession(params);
const idleTimeoutMs = normalizeNonNegativeMs(params.idleTimeoutMs);
const idleTimeoutMs = resolveNonNegativeIntegerOption(params.idleTimeoutMs, 0);
return updateBindingsForTargetSessionSync(ids, (existing, now) => ({
...existing,
idleTimeoutMs,
@ -53,7 +53,7 @@ export function setThreadBindingMaxAgeBySessionKey(
params: Parameters<typeof setThreadBindingMaxAgeBySessionKeyAsync>[0],
): ThreadBindingRecord[] {
const ids = resolveBindingIdsForTargetSession(params);
const maxAgeMs = normalizeNonNegativeMs(params.maxAgeMs);
const maxAgeMs = resolveNonNegativeIntegerOption(params.maxAgeMs, 0);
return updateBindingsForTargetSessionSync(ids, (existing, now) => ({
...existing,
maxAgeMs,

7
extensions/feishu/src/reply-delivery-result.ts
View file

@ -87,12 +87,7 @@ export function mergeFeishuReplyDeliveryResults(
return createFeishuReplyDeliveryResult({
results: visible,
visibleReplySent: visible.length > 0,
content:
content === undefined
? acceptedContent.length > 0
? acceptedContent.join("\n\n")
: undefined
: content,
content: content ?? (acceptedContent.length > 0 ? acceptedContent.join("\n\n") : undefined),
});
}

88
extensions/feishu/src/thread-bindings.ts
View file

@ -5,32 +5,26 @@ import {
resolveThreadBindingIdleTimeoutMsForChannel,
resolveThreadBindingMaxAgeMsForChannel,
registerSessionBindingAdapter,
resolveThreadBindingConversationIdFromBindingId,
unregisterSessionBindingAdapter,
type BindingTargetKind,
type SessionBindingAdapter,
type SessionBindingRecord,
} from "openclaw/plugin-sdk/conversation-runtime";
import { resolveGlobalSingleton } from "openclaw/plugin-sdk/global-singleton";
import { isFutureDateTimestampMs } from "openclaw/plugin-sdk/number-runtime";
import { normalizeAccountId } from "openclaw/plugin-sdk/routing";
import { normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runtime";
import type { AccountScopedConversationBindingRecord } from "openclaw/plugin-sdk/thread-bindings-session-runtime";
import {
createAccountScopedBindingAdapter,
projectThreadBindingRecord,
} from "openclaw/plugin-sdk/thread-bindings-session-runtime";
type FeishuBindingTargetKind = "subagent" | "acp";
type FeishuThreadBindingRecord = {
accountId: string;
conversationId: string;
type FeishuThreadBindingRecord = AccountScopedConversationBindingRecord<FeishuBindingTargetKind> & {
parentConversationId?: string;
deliveryTo?: string;
deliveryThreadId?: string;
targetKind: FeishuBindingTargetKind;
targetSessionKey: string;
agentId?: string;
label?: string;
boundBy?: string;
boundAt: number;
lastActivityAt: number;
metadata?: Record<string, unknown>;
};
@ -81,31 +75,21 @@ function toSessionBindingRecord(
idleExpiresAt != null && maxAgeExpiresAt != null
? Math.min(idleExpiresAt, maxAgeExpiresAt)
: (idleExpiresAt ?? maxAgeExpiresAt);
return {
bindingId: resolveBindingKey(record),
targetSessionKey: record.targetSessionKey,
targetKind: record.targetKind === "subagent" ? "subagent" : "session",
return projectThreadBindingRecord(record, {
conversation: {
channel: "feishu",
accountId: record.accountId,
conversationId: record.conversationId,
parentConversationId: record.parentConversationId,
},
status: "active",
boundAt: record.boundAt,
expiresAt,
metadata: {
targetKind: record.targetKind === "subagent" ? "subagent" : "session",
lifecycle: { ...defaults, expiresAt },
metadata: (lifecycleMetadata) => ({
...record.metadata,
agentId: record.agentId,
label: record.label,
boundBy: record.boundBy,
...lifecycleMetadata,
deliveryTo: record.deliveryTo,
deliveryThreadId: record.deliveryThreadId,
lastActivityAt: record.lastActivityAt,
idleTimeoutMs: defaults.idleTimeoutMs,
maxAgeMs: defaults.maxAgeMs,
},
};
}),
});
}
export function createFeishuThreadBindingManager(params: {
@ -263,7 +247,7 @@ export function createFeishuThreadBindingManager(params: {
},
};
const sessionBindingAdapter: SessionBindingAdapter = {
const sessionBindingAdapter = createAccountScopedBindingAdapter({
channel: "feishu",
accountId,
capabilities: {
@ -282,43 +266,13 @@ export function createFeishuThreadBindingManager(params: {
});
return bound ? toSessionBindingRecord(bound, bindingTimeouts) : null;
},
listBySession: (targetSessionKey) =>
manager
.listBySessionKey(targetSessionKey)
.map((entry) => toSessionBindingRecord(entry, bindingTimeouts)),
resolveByConversation: (ref) => {
if (ref.channel !== "feishu") {
return null;
}
const found = manager.getByConversationId(ref.conversationId);
return found ? toSessionBindingRecord(found, bindingTimeouts) : null;
},
touch: (bindingId, at) => {
const conversationId = resolveThreadBindingConversationIdFromBindingId({
accountId,
bindingId,
});
if (conversationId) {
manager.touchConversation(conversationId, at);
}
},
unbind: async (input) => {
if (input.targetSessionKey?.trim()) {
return manager
.unbindBySessionKey(input.targetSessionKey.trim())
.map((entry) => toSessionBindingRecord(entry, bindingTimeouts));
}
const conversationId = resolveThreadBindingConversationIdFromBindingId({
accountId,
bindingId: input.bindingId,
});
if (!conversationId) {
return [];
}
const removed = manager.unbindConversation(conversationId);
return removed ? [toSessionBindingRecord(removed, bindingTimeouts)] : [];
},
};
project: (record: FeishuThreadBindingRecord) => toSessionBindingRecord(record, bindingTimeouts),
listBySessionKey: manager.listBySessionKey,
getByConversation: (ref) => manager.getByConversationId(ref.conversationId),
touchConversation: manager.touchConversation,
unbindConversation: manager.unbindConversation,
unbindBySessionKey: (sessionKey) => manager.unbindBySessionKey(sessionKey.trim()),
});
registerSessionBindingAdapter(sessionBindingAdapter);

26
extensions/imessage/src/markdown-format.ts
View file

@ -1,4 +1,5 @@
import {
applyMarkdownTextEdits,
FormatCapabilityProfile,
markdownToIR,
renderMarkdownWithAttributedRanges,
@ -57,35 +58,14 @@ function restoreCodeMarkers(
if (codeRanges.length === 0) {
return { text, ranges };
}
let rendered = "";
let cursor = 0;
// The code-only renderer merges and sorts these non-overlapping ranges.
// Record each cumulative UTF-16 shift at the existing end-inclusive boundary.
const edits = codeRanges.map((range) => {
const end = range.start + range.length;
const content = text.slice(range.start, end);
const marker = codeDelimiter(content);
const padding = content.startsWith("`") || content.endsWith("`") ? " " : "";
rendered +=
text.slice(cursor, range.start) + `${marker}${padding}${content}${padding}${marker}`;
cursor = end;
return { end, shift: rendered.length - end };
return { start: range.start, end, text: `${marker}${padding}${content}${padding}${marker}` };
});
rendered += text.slice(cursor);
const mapOffset = (offset: number) => {
let low = 0;
let high = edits.length;
while (low < high) {
const middle = low + Math.floor((high - low) / 2);
const edit = edits[middle];
if (edit && edit.end <= offset) {
low = middle + 1;
} else {
high = middle;
}
}
return offset + (edits[low - 1]?.shift ?? 0);
};
const { text: rendered, mapOffset } = applyMarkdownTextEdits(text, edits);
return {
text: rendered,
ranges: ranges.map((range) => {

47
extensions/imessage/src/monitor/deliver.ts
View file

@ -1,8 +1,4 @@
import {
createAcceptedChannelDeliveryResult,
createChannelPartialDeliveryError,
isChannelPartialDeliveryError,
} from "openclaw/plugin-sdk/channel-inbound";
import { createChannelDeliveryAccumulator } from "openclaw/plugin-sdk/channel-outbound";
import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
import { resolveMarkdownTableMode } from "openclaw/plugin-sdk/markdown-table-runtime";
import { chunkMarkdownTextWithMode, resolveChunkMode } from "openclaw/plugin-sdk/reply-chunking";
@ -40,7 +36,9 @@ export async function deliverIMessageReply(params: {
const reply = resolveSendableOutboundReplyParts(payload, {
text: convertMarkdownTables(rawText, tableMode),
});
const accepted: Awaited<ReturnType<typeof sendMessageIMessage>>[] = [];
const accepted = createChannelDeliveryAccumulator({
kind: reply.mediaUrls.length > 0 ? "media" : "text",
});
const sendAccepted = async (text: string, mediaUrl?: string) => {
const sent = await sendMessageIMessage(target, text, {
config: cfg,
@ -49,7 +47,7 @@ export async function deliverIMessageReply(params: {
accountId,
replyToId: payload.replyToId,
});
accepted.push(sent);
accepted.add({ receipt: sent.receipt }, sent.sentText);
const echoText = sent.echoText ?? (sent.sentText || undefined);
sentMessageCache?.remember(scope, {
...(echoText ? { text: echoText } : {}),
@ -67,39 +65,12 @@ export async function deliverIMessageReply(params: {
sendMedia: ({ mediaUrl, caption }) => sendAccepted(caption ?? "", mediaUrl),
});
} catch (error: unknown) {
const partial = isChannelPartialDeliveryError(error) ? error.deliveryResult : undefined;
if (accepted.length === 0 && partial?.visibleReplySent !== true) {
throw error;
}
// A native attachment can settle before its caption rejects; preserve every
// previously accepted receipt plus that nested provider-visible subset.
throw createChannelPartialDeliveryError(
error,
createAcceptedChannelDeliveryResult({
results: accepted.map((result) => ({ receipt: result.receipt })),
deliveryResults: partial ? [partial] : [],
kind: reply.mediaUrls.length > 0 ? "media" : "text",
content: [...accepted.map((result) => result.sentText), partial?.content]
.filter(Boolean)
.join("\n"),
}),
);
throw accepted.partialError(error);
}
if (delivered === "empty") {
return {
visibleReplySent: false as const,
suppression: { reason: "no_visible_result" as const },
};
const deliveryResult = accepted.result();
if (delivered !== "empty") {
runtime.log?.(`imessage: delivered reply to ${target}`);
}
const deliveryResult = createAcceptedChannelDeliveryResult({
results: accepted.map((result) => ({ receipt: result.receipt })),
kind: delivered,
content: accepted
.map((result) => result.sentText)
.filter(Boolean)
.join("\n"),
});
runtime.log?.(`imessage: delivered reply to ${target}`);
return deliveryResult;
}

3
extensions/matrix/src/matrix/draft-stream.ts
View file

@ -64,9 +64,6 @@ export function createMatrixDraftStream(params: {
);
return false;
}
if (sendFailed) {
return false;
}
if (preparedText.trimmedText === lastSentText) {
return true;
}

25
extensions/matrix/src/matrix/monitor/replies.ts
View file

@ -3,7 +3,10 @@ import {
createChannelPartialDeliveryError,
isChannelPartialDeliveryError,
} from "openclaw/plugin-sdk/channel-inbound";
import type { LivePreviewDeliveryResult } from "openclaw/plugin-sdk/channel-outbound";
import {
createChannelDeliveryAccumulator,
type LivePreviewDeliveryResult,
} from "openclaw/plugin-sdk/channel-outbound";
import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
import { resolveSendableOutboundReplyParts } from "openclaw/plugin-sdk/reply-payload";
import type { ReplyPayload } from "openclaw/plugin-sdk/reply-runtime";
@ -60,18 +63,6 @@ export function toMatrixPartialDeliveryError(
: error;
}
function createMatrixReplyDeliveryResult(
results: readonly MatrixSendResult[],
): MatrixReplyDeliveryResult {
if (results.length === 0) {
return mergeMatrixReplyDeliveryResults([]);
}
return createAcceptedChannelDeliveryResult({
results: results.map((result) => ({ receipt: result.receipt })),
content: joinMatrixVisibleContent(results.map((result) => result.content)),
});
}
function resolveVisibleMatrixReplyText(text?: string): string | undefined {
if (typeof text !== "string") {
return undefined;
@ -107,7 +98,7 @@ export async function deliverMatrixReplies(params: {
}
};
const hasRepliedRef = params.hasRepliedRef ?? { value: false };
const acceptedResults: MatrixSendResult[] = [];
const accepted = createChannelDeliveryAccumulator();
try {
for (const reply of params.replies) {
const visibleText = resolveVisibleMatrixReplyText(reply.text);
@ -140,7 +131,7 @@ export async function deliverMatrixReplies(params: {
: undefined;
const onDeliveryResult = (result: MatrixSendResult) => {
// A concrete event consumes the first-reply slot even when a later event fails.
acceptedResults.push(result);
accepted.add({ receipt: result.receipt }, result.content);
if (replyToIdForReply) {
hasRepliedRef.value = true;
}
@ -173,7 +164,7 @@ export async function deliverMatrixReplies(params: {
}
}
} catch (error: unknown) {
throw toMatrixPartialDeliveryError(error, [createMatrixReplyDeliveryResult(acceptedResults)]);
throw toMatrixPartialDeliveryError(error, [accepted.result()]);
}
return createMatrixReplyDeliveryResult(acceptedResults);
return accepted.result();
}

59
extensions/matrix/src/matrix/thread-bindings-shared.ts
View file

@ -1,25 +1,18 @@
import type {
BindingTargetKind,
SessionBindingRecord,
import {
projectThreadBindingRecord,
resolveThreadBindingLifecycle,
type AccountScopedConversationBindingRecord,
type SessionBindingRecord,
} from "openclaw/plugin-sdk/thread-bindings-session-runtime";
import { resolveThreadBindingLifecycle } from "openclaw/plugin-sdk/thread-bindings-session-runtime";
type MatrixThreadBindingTargetKind = "subagent" | "acp";
export type MatrixThreadBindingRecord = {
accountId: string;
conversationId: string;
parentConversationId?: string;
targetKind: MatrixThreadBindingTargetKind;
targetSessionKey: string;
agentId?: string;
label?: string;
boundBy?: string;
boundAt: number;
lastActivityAt: number;
idleTimeoutMs?: number;
maxAgeMs?: number;
};
export type MatrixThreadBindingRecord =
AccountScopedConversationBindingRecord<MatrixThreadBindingTargetKind> & {
parentConversationId?: string;
idleTimeoutMs?: number;
maxAgeMs?: number;
};
export type MatrixThreadBindingManager = {
accountId: string;
@ -60,14 +53,6 @@ export function resolveBindingKey(params: {
return `${params.accountId}:${params.parentConversationId?.trim() || "-"}:${params.conversationId}`;
}
function toSessionBindingTargetKind(raw: MatrixThreadBindingTargetKind): BindingTargetKind {
return raw === "subagent" ? "subagent" : "session";
}
export function toMatrixBindingTargetKind(raw: BindingTargetKind): MatrixThreadBindingTargetKind {
return raw === "subagent" ? "subagent" : "acp";
}
export function toSessionBindingRecord(
record: MatrixThreadBindingRecord,
defaults: { idleTimeoutMs: number; maxAgeMs: number },
@ -80,28 +65,16 @@ export function toSessionBindingRecord(
const idleTimeoutMs =
typeof record.idleTimeoutMs === "number" ? record.idleTimeoutMs : defaults.idleTimeoutMs;
const maxAgeMs = typeof record.maxAgeMs === "number" ? record.maxAgeMs : defaults.maxAgeMs;
return {
bindingId: resolveBindingKey(record),
targetSessionKey: record.targetSessionKey,
targetKind: toSessionBindingTargetKind(record.targetKind),
return projectThreadBindingRecord(record, {
conversation: {
channel: "matrix",
accountId: record.accountId,
conversationId: record.conversationId,
parentConversationId: record.parentConversationId,
},
status: "active",
boundAt: record.boundAt,
expiresAt: lifecycle.expiresAt,
metadata: {
agentId: record.agentId,
label: record.label,
boundBy: record.boundBy,
lastActivityAt: record.lastActivityAt,
idleTimeoutMs,
maxAgeMs,
},
};
bindingId: resolveBindingKey(record),
targetKind: record.targetKind === "subagent" ? "subagent" : "session",
lifecycle: { ...lifecycle, idleTimeoutMs, maxAgeMs },
});
}
export function setBindingRecord(record: MatrixThreadBindingRecord): void {

14
extensions/matrix/src/matrix/thread-bindings.ts
View file

@ -28,7 +28,6 @@ import {
resolveBindingKey,
setBindingRecord,
setMatrixThreadBindingManagerEntry,
toMatrixBindingTargetKind,
toSessionBindingRecord,
type MatrixThreadBindingManager,
type MatrixThreadBindingRecord,
@ -414,8 +413,8 @@ export async function createMatrixThreadBindingManager(params: {
reason: string | ((record: MatrixThreadBindingRecord) => string | undefined),
) => {
await Promise.all(
removed.map(async (record) => {
await sendFarewellMessage({
removed.map((record) =>
sendFarewellMessage({
cfg: params.cfg,
client: params.client,
accountId: params.accountId,
@ -423,8 +422,8 @@ export async function createMatrixThreadBindingManager(params: {
defaultIdleTimeoutMs: defaults.idleTimeoutMs,
defaultMaxAgeMs: defaults.maxAgeMs,
reason: typeof reason === "function" ? reason(record) : reason,
});
}),
}),
),
);
};
const unbindRecords = async (records: MatrixThreadBindingRecord[], reason: string) => {
@ -477,7 +476,7 @@ export async function createMatrixThreadBindingManager(params: {
accountId: params.accountId,
conversationId: boundConversationId,
...(boundParentConversationId ? { parentConversationId: boundParentConversationId } : {}),
targetKind: toMatrixBindingTargetKind(input.targetKind),
targetKind: input.targetKind === "subagent" ? "subagent" : "acp",
targetSessionKey,
agentId:
normalizeOptionalString(input.metadata?.agentId) ??
@ -522,7 +521,7 @@ export async function createMatrixThreadBindingManager(params: {
manager.touchBinding(bindingId, at);
},
unbind: async (input) => {
const removed = await unbindRecords(
return unbindRecords(
listBindingsForAccount(params.accountId).filter((record) => {
if (input.bindingId?.trim()) {
return resolveBindingKey(record) === input.bindingId.trim();
@ -534,7 +533,6 @@ export async function createMatrixThreadBindingManager(params: {
}),
input.reason,
);
return removed;
},
};

20
extensions/mattermost/src/delivery-trace.test.ts
View file

@ -1,7 +1,7 @@
// Mattermost delivery trace goldens: replayable wire-level lifecycle recordings.
//
// Wires the real draft-stream path the monitor uses (createMattermostDraftStream,
// preview boundary controller, deliverMattermostReplyWithDraftPreview,
// its forceNewMessage boundary, deliverMattermostReplyWithDraftPreview,
// deliverMattermostReplyPayload) against a recording MattermostClient, so OUT
// events are the raw REST calls (POST/PUT/DELETE /posts). The monitor's
// per-activity glue (partial dedupe, boundary rotation) is replicated inline in
@ -31,10 +31,7 @@ import type { ReplyPayload } from "openclaw/plugin-sdk/reply-runtime";
import { convertMarkdownTables } from "openclaw/plugin-sdk/text-chunking";
import { describe, it, vi } from "vitest";
import { createMattermostPost, type MattermostClient } from "./mattermost/client.js";
import {
createMattermostDraftPreviewBoundaryController,
createMattermostDraftStream,
} from "./mattermost/draft-stream.js";
import { createMattermostDraftStream } from "./mattermost/draft-stream.js";
import { resolveMattermostReplyRootId } from "./mattermost/monitor-context.js";
import { deliverMattermostReplyWithDraftPreview } from "./mattermost/monitor-draft-delivery.js";
import {
@ -118,12 +115,6 @@ function setupMattermostTrace(recorder: WireRecorder) {
chunkText: (value) =>
chunkMarkdownTextWithMode(convertMarkdownTables(value, tableMode), textLimit, chunkMode),
});
const previewBoundary = createMattermostDraftPreviewBoundaryController({
enabled: true,
forceNewMessage: async () => {
await draftStream.forceNewMessage();
},
});
const previewLifecycle = createLivePreviewLifecycle<ReplyPayload, string>({
draft: { ...draftStream, id: draftStream.postId },
});
@ -255,14 +246,15 @@ function setupMattermostTrace(recorder: WireRecorder) {
}
lastPartialText = cleaned;
draftStream.updateAssistantText(cleaned);
previewBoundary.noteUpdate();
break;
}
case "block-final":
// Block boundary = assistant message boundary: partial snapshots reset
// and the block-mode preview rotates to a fresh post.
lastPartialText = "";
await previewBoundary.noteBoundary();
if (lastPartialText) {
lastPartialText = "";
await draftStream.forceNewMessage();
}
break;
case "final":
// Final resolution may edit the confirmed preview post in place; join

106
extensions/mattermost/src/mattermost/draft-stream.test.ts
View file

@ -8,10 +8,7 @@ import { createDeferred } from "openclaw/plugin-sdk/extension-shared";
import type { ReplyPayload } from "openclaw/plugin-sdk/reply-runtime";
import { describe, expect, it, vi } from "vitest";
import type { MattermostClient } from "./client.js";
import {
createMattermostDraftPreviewBoundaryController,
createMattermostDraftStream,
} from "./draft-stream.js";
import { createMattermostDraftStream } from "./draft-stream.js";
import { deliverMattermostReplyWithDraftPreview } from "./monitor-draft-delivery.js";
type RequestRecord = {
@ -977,104 +974,3 @@ describe("createMattermostDraftStream forceNewMessage", () => {
]);
});
});
describe("createMattermostDraftPreviewBoundaryController", () => {
it("calls forceNewMessage on boundary when enabled and content was streamed", async () => {
const forceNewMessage = vi.fn();
const controller = createMattermostDraftPreviewBoundaryController({
enabled: true,
forceNewMessage,
});
controller.noteUpdate();
await controller.noteBoundary();
expect(forceNewMessage).toHaveBeenCalledTimes(1);
});
it("skips forceNewMessage when no content was streamed since the last boundary", async () => {
const forceNewMessage = vi.fn();
const controller = createMattermostDraftPreviewBoundaryController({
enabled: true,
forceNewMessage,
});
await controller.noteBoundary();
await controller.noteBoundary();
controller.noteUpdate();
await controller.noteBoundary();
await controller.noteBoundary();
expect(forceNewMessage).toHaveBeenCalledTimes(1);
});
it("never calls forceNewMessage when disabled", async () => {
const forceNewMessage = vi.fn();
const controller = createMattermostDraftPreviewBoundaryController({
enabled: false,
forceNewMessage,
});
controller.noteUpdate();
await controller.noteBoundary();
controller.noteUpdate();
await controller.noteBoundary();
expect(forceNewMessage).not.toHaveBeenCalled();
});
it("awaits the forceNewMessage promise before resolving the boundary", async () => {
let releaseForce: (() => void) | undefined;
const forcePending = new Promise<void>((resolve) => {
releaseForce = resolve;
});
const forceNewMessage = vi.fn(async () => {
await forcePending;
});
const controller = createMattermostDraftPreviewBoundaryController({
enabled: true,
forceNewMessage,
});
controller.noteUpdate();
let resolved = false;
const boundary = controller.noteBoundary().then(() => {
resolved = true;
});
await Promise.resolve();
await Promise.resolve();
expect(resolved).toBe(false);
releaseForce?.();
await boundary;
expect(resolved).toBe(true);
expect(forceNewMessage).toHaveBeenCalledTimes(1);
});
it("splits the next boundary when a noteUpdate arrives while the prior boundary is pending", async () => {
const releases: Array<() => void> = [];
const forceNewMessage = vi.fn(
() =>
new Promise<void>((resolve) => {
releases.push(resolve);
}),
);
const controller = createMattermostDraftPreviewBoundaryController({
enabled: true,
forceNewMessage,
});
controller.noteUpdate();
const firstBoundary = controller.noteBoundary();
controller.noteUpdate();
releases[0]?.();
await firstBoundary;
const secondBoundary = controller.noteBoundary();
releases[1]?.();
await secondBoundary;
expect(forceNewMessage).toHaveBeenCalledTimes(2);
});
});

19
extensions/mattermost/src/mattermost/draft-stream.ts
View file

@ -56,25 +56,6 @@ function consumeMattermostPublishedChunk(params: {
return params.source.startsWith(chunk, offset) ? offset + chunk.length : undefined;
}
export function createMattermostDraftPreviewBoundaryController(params: {
enabled: boolean;
forceNewMessage: () => void | Promise<void>;
}) {
let hasStreamedContent = false;
return {
noteUpdate() {
hasStreamedContent = true;
},
async noteBoundary() {
if (!params.enabled || !hasStreamedContent) {
return;
}
hasStreamedContent = false;
await params.forceNewMessage();
},
};
}
export function createMattermostDraftStream(params: {
client: MattermostClient;
channelId: string;

71
extensions/mattermost/src/mattermost/monitor-turn.ts
View file

@ -14,10 +14,7 @@ import { getGlobalHookRunner } from "openclaw/plugin-sdk/plugin-runtime";
import type { finalizeInboundContext } from "openclaw/plugin-sdk/reply-runtime";
import { resolveInboundLastRouteSessionKey } from "openclaw/plugin-sdk/routing";
import type { MattermostPost } from "./client.js";
import {
createMattermostDraftPreviewBoundaryController,
createMattermostDraftStream,
} from "./draft-stream.js";
import { createMattermostDraftStream } from "./draft-stream.js";
import { normalizeMattermostAllowEntry } from "./ingress-identity.js";
import {
formatMattermostFinalDeliveryOutcomeLog,
@ -49,24 +46,6 @@ type MattermostInboundTurnParams = {
turnAdoptionLifecycle?: MattermostIngressLifecycle;
};
function createDisabledMattermostDraftStream(): ReturnType<typeof createMattermostDraftStream> {
const noopAsync = async () => {};
return {
update: () => {},
updateAssistantText: () => {},
flush: noopAsync,
postId: () => undefined,
clear: noopAsync,
deleteCurrentMessage: noopAsync,
discardPending: noopAsync,
seal: noopAsync,
stop: noopAsync,
forceNewMessage: noopAsync,
settleBoundaries: noopAsync,
resolveFinalText: (text) => ({ kind: "full", text, publishedParts: [] }),
};
}
export async function dispatchMattermostInboundTurn(
monitor: MattermostMonitorContext,
params: MattermostInboundTurnParams,
@ -126,11 +105,8 @@ export async function dispatchMattermostInboundTurn(
log: monitor.logVerboseMessage,
warn: monitor.logVerboseMessage,
})
: createDisabledMattermostDraftStream();
const previewBoundaryController = createMattermostDraftPreviewBoundaryController({
enabled: draftPreviewEnabled && account.streamingMode === "block",
forceNewMessage: draftStream.forceNewMessage,
});
: undefined;
let hasStreamedPreview = false;
let lastPartialText = "";
let firstAssistantPreviewPrefix: string | undefined;
let firstAssistantPreviewPrefixPending = true;
@ -145,12 +121,12 @@ export async function dispatchMattermostInboundTurn(
seed: `${account.accountId}:${channelId}`,
shouldStartNow: (line) => typeof line === "object" && line.kind === "item",
update: async (previewText, options) => {
draftStream.update(previewText);
draftStream?.update(previewText);
if (options?.flush) {
await draftStream.flush();
await draftStream?.flush();
}
},
deleteCurrent: () => draftStream.deleteCurrentMessage(),
deleteCurrent: () => draftStream?.deleteCurrentMessage(),
});
const enterBlockPreviewActivity = (activity: "reasoning" | "text" | "tool") => {
if (account.streamingMode !== "block") {
@ -170,9 +146,11 @@ export async function dispatchMattermostInboundTurn(
if (startsNewGeneration) {
currentAssistantPreviewUsesPrefix = false;
}
const boundarySettled = startsNewGeneration
? previewBoundaryController.noteBoundary()
: undefined;
let boundarySettled: Promise<void> | undefined;
if (startsNewGeneration && draftStream && hasStreamedPreview) {
hasStreamedPreview = false;
boundarySettled = draftStream.forceNewMessage();
}
// Message-start is only a candidate boundary: consecutive tools stay together, while the first visible text or reasoning starts a new block.
if (!continuesCurrentActivity) {
progressDraft.resetActivity();
@ -185,7 +163,7 @@ export async function dispatchMattermostInboundTurn(
return boundarySettled;
};
const previewLifecycle = createLivePreviewLifecycle<ReplyPayload, string>({
draft: draftPreviewEnabled
draft: draftStream
? {
flush: draftStream.flush,
id: draftStream.postId,
@ -201,7 +179,12 @@ export async function dispatchMattermostInboundTurn(
});
const resolvePreviewFinalText = (text?: string): MattermostPreviewFinalResolution | undefined => {
const resolution = draftStream.resolveFinalText(typeof text === "string" ? text : "");
const finalText = typeof text === "string" ? text : "";
const resolution = draftStream?.resolveFinalText(finalText) ?? {
kind: "full",
text: finalText,
publishedParts: [],
};
const confirmedDelivery =
resolution.publishedParts.length > 0
? (() => {
@ -280,8 +263,8 @@ export async function dispatchMattermostInboundTurn(
? cleaned
: `${firstAssistantPreviewPrefix} ${cleaned}`
: cleaned;
draftStream.updateAssistantText(previewText);
previewBoundaryController.noteUpdate();
draftStream?.updateAssistantText(previewText);
hasStreamedPreview = true;
return boundarySettled;
};
@ -296,7 +279,7 @@ export async function dispatchMattermostInboundTurn(
if (info.kind === "final") {
await enterBlockPreviewActivity("text");
// Final text uses only confirmed-visible generations, so join prior boundary work before deciding whether to edit in place.
await draftStream.settleBoundaries();
await draftStream?.settleBoundaries();
}
// A visible same-thread final can be a send or an in-place draft edit; either path records participation.
let threadParticipationRecorded = false;
@ -328,7 +311,7 @@ export async function dispatchMattermostInboundTurn(
info.kind === "final" &&
!payloadToDeliver.isError &&
typeof payloadToDeliver.text === "string"
? draftStream.resolveFinalText(payloadToDeliver.text)
? draftStream?.resolveFinalText(payloadToDeliver.text)
: undefined;
const resolvedPayload = finalTextResolution
? {
@ -490,8 +473,8 @@ export async function dispatchMattermostInboundTurn(
}
if (!lastPartialText) {
const boundarySettled = enterBlockPreviewActivity("reasoning");
draftStream.update("Thinking…");
previewBoundaryController.noteUpdate();
draftStream?.update("Thinking…");
hasStreamedPreview = true;
await boundarySettled;
}
return false;
@ -505,7 +488,7 @@ export async function dispatchMattermostInboundTurn(
explanation: payloadValue.explanation,
explanationFormat: payloadValue.explanationFormat,
});
previewBoundaryController.noteUpdate();
hasStreamedPreview = true;
const [, visible] = await Promise.all([boundarySettled, progressSettled]);
return visible;
},
@ -519,7 +502,7 @@ export async function dispatchMattermostInboundTurn(
}
const boundarySettled = enterBlockPreviewActivity("tool");
const progressSettled = progressDraft.pushItemEvent(payloadLocal);
previewBoundaryController.noteUpdate();
hasStreamedPreview = true;
const [, visible] = await Promise.all([boundarySettled, progressSettled]);
return visible;
},
@ -529,7 +512,7 @@ export async function dispatchMattermostInboundTurn(
});
} finally {
try {
await draftStream.stop();
await draftStream?.stop();
await previewLifecycle.cleanup();
} catch (err) {
monitor.logVerboseMessage(`mattermost draft preview cleanup failed: ${String(err)}`);

12
extensions/mattermost/src/mattermost/monitor.inbound-system-event.test.ts
View file

@ -132,14 +132,10 @@ vi.mock("./client.js", async () => {
};
});
vi.mock("./draft-stream.js", async () => {
const actual = await vi.importActual<typeof import("./draft-stream.js")>("./draft-stream.js");
return {
createMattermostDraftStream: mockState.createMattermostDraftStream,
createMattermostDraftPreviewBoundaryController:
actual.createMattermostDraftPreviewBoundaryController,
};
});
vi.mock("./draft-stream.js", async (importOriginal) => ({
...(await importOriginal<typeof import("./draft-stream.js")>()),
createMattermostDraftStream: mockState.createMattermostDraftStream,
}));
vi.mock("./monitor-resources.js", async (importOriginal) => ({
...(await importOriginal<typeof import("./monitor-resources.js")>()),

17
extensions/mattermost/src/mattermost/monitor.preview-policy.test-support.ts
View file

@ -7,8 +7,11 @@ export function registerMattermostPreviewPolicyTests(harness: {
receivePost: (post: { id: string; message: string }, config?: OpenClawConfig) => Promise<unknown>;
mockState: {
runtimeCore: unknown;
abortController: AbortController | undefined;
createMattermostDraftStream: Mock;
dispatchInboundMessage: Mock;
createReplyDispatcherWithTyping: Mock;
sendMessageMattermost: Mock;
getGlobalHookRunner: Mock;
};
}) {
@ -23,9 +26,23 @@ export function registerMattermostPreviewPolicyTests(harness: {
},
};
mockState.runtimeCore = createRuntimeCore(config);
mockState.dispatchInboundMessage.mockImplementation(async (params) => {
await params.replyOptions?.onPartialReply?.({ text: "Partial answer" });
await params.replyOptions?.onReasoningEnd?.();
await params.replyOptions?.onReasoningStream?.({ text: "Private reasoning" });
const dispatcherOptions =
mockState.createReplyDispatcherWithTyping.mock.results.at(-1)?.value?.options;
await dispatcherOptions?.deliver({ text: "Complete answer" }, { kind: "final" });
mockState.abortController?.abort();
});
await receivePost({ id: "post-streaming-off", message: "stream this in blocks" }, config);
expect(mockState.dispatchInboundMessage).toHaveBeenCalledTimes(1);
expect(mockState.createMattermostDraftStream).not.toHaveBeenCalled();
expect(mockState.sendMessageMattermost).toHaveBeenCalledExactlyOnceWith(
"channel:chan-1",
"Complete answer",
expect.objectContaining({ accountId: "default" }),
);
const replyOptions = mockState.dispatchInboundMessage.mock.calls.at(0)?.[0].replyOptions;
expect(replyOptions?.disableBlockStreaming).toBe(false);
expect(replyOptions?.preserveProgressCallbackStartOrder).toBeUndefined();

51
extensions/mattermost/src/mattermost/reply-delivery.ts
View file

@ -1,9 +1,7 @@
import {
createAcceptedChannelDeliveryResult,
createChannelPartialDeliveryError,
isChannelPartialDeliveryError,
} from "openclaw/plugin-sdk/channel-inbound";
import type { MessageReceipt } from "openclaw/plugin-sdk/channel-outbound";
createChannelDeliveryAccumulator,
type MessageReceipt,
} from "openclaw/plugin-sdk/channel-outbound";
import type { OpenClawConfig, PluginRuntime } from "openclaw/plugin-sdk/core";
import { getAgentScopedMediaLocalRoots } from "openclaw/plugin-sdk/media-runtime";
import {
@ -13,7 +11,7 @@ import {
} from "openclaw/plugin-sdk/reply-payload";
import type { ReplyPayload } from "openclaw/plugin-sdk/reply-runtime";
import { requiresMattermostMediaUpload, resolveMattermostPresentation } from "../normalize.js";
import type { MattermostSendResult, sendMessageMattermost } from "./send.js";
import type { sendMessageMattermost } from "./send.js";
type MarkdownTableMode = Parameters<PluginRuntime["channel"]["text"]["convertMarkdownTables"]>[1];
@ -68,8 +66,10 @@ export async function deliverMattermostReplyPayload(params: {
"mattermost",
params.accountId,
);
const results: MattermostSendResult[] = [];
const acceptedContents: string[] = [];
const accepted = createChannelDeliveryAccumulator({
kind: reply.mediaUrls.length > 0 ? "media" : "text",
...(params.replyToId ? { replyToId: params.replyToId } : {}),
});
const sendAccepted = async (text: string, mediaUrl?: string) => {
const result = await params.sendMessage(`channel:${params.channelId}`, text, {
cfg: params.cfg,
@ -77,13 +77,12 @@ export async function deliverMattermostReplyPayload(params: {
...(mediaUrl ? { mediaUrl, mediaLocalRoots } : {}),
// Local media must upload successfully instead of silently posting only its caption.
...(requiresMattermostMediaUpload(mediaUrl) ? { requireMediaUpload: true } : {}),
...(results.length === 0 && reply.mediaUrls.length < 2 && presentation.buttons.length
...(accepted.size === 0 && reply.mediaUrls.length < 2 && presentation.buttons.length
? { buttons: presentation.buttons }
: {}),
replyToId: params.replyToId,
});
results.push(result);
acceptedContents.push(result.content);
accepted.add({ receipt: result.receipt }, result.content);
};
let outcome: Exclude<MattermostReplyDeliveryOutcome, "reasoning_skipped">;
try {
@ -96,36 +95,10 @@ export async function deliverMattermostReplyPayload(params: {
sendMedia: ({ mediaUrl, caption }) => sendAccepted(caption ?? "", mediaUrl),
});
} catch (error: unknown) {
const failedPartial = isChannelPartialDeliveryError(error) ? error.deliveryResult : undefined;
if (results.length === 0 && failedPartial?.visibleReplySent !== true) {
throw error;
}
throw createChannelPartialDeliveryError(
error,
createAcceptedChannelDeliveryResult({
results: results.map((result) => ({ receipt: result.receipt })),
deliveryResults: failedPartial ? [failedPartial] : [],
kind: reply.mediaUrls.length > 0 ? "media" : "text",
...(params.replyToId ? { replyToId: params.replyToId } : {}),
content: joinMattermostVisibleContent([...acceptedContents, failedPartial?.content]),
}),
);
}
if (outcome === "empty") {
return {
outcome,
visibleReplySent: false,
suppression: { reason: "no_visible_result" },
};
throw accepted.partialError(error);
}
return {
outcome,
...createAcceptedChannelDeliveryResult({
results: results.map((result) => ({ receipt: result.receipt })),
kind: outcome,
...(params.replyToId ? { replyToId: params.replyToId } : {}),
content: joinMattermostVisibleContent(acceptedContents),
}),
...accepted.result(),
};
}

35
extensions/msteams/src/format.ts
View file

@ -2,9 +2,11 @@ import { randomUUID } from "node:crypto";
import { expectDefined } from "openclaw/plugin-sdk/expect-runtime";
import { getMarkdownTableSource } from "openclaw/plugin-sdk/markdown-table-runtime";
import {
applyMarkdownTextEdits,
convertMarkdownTables,
FormatCapabilityProfile,
type MarkdownIR,
type MarkdownTextEdit,
markdownToIR,
markdownToIRWithMeta,
renderMarkdownWithMarkers,
@ -61,36 +63,9 @@ function restoreTokens(text: string, prefix: string, values: readonly string[]):
);
}
type TextEdit = { start: number; end: number; text: string };
function rewriteMarkdownIR(ir: MarkdownIR, edits: readonly TextEdit[]): MarkdownIR {
const ordered = edits.toSorted((a, b) => a.start - b.start);
const cumulativeDeltas: number[] = [];
const exactEdits = new Map<string, TextEdit>();
let text = "";
let cursor = 0;
let delta = 0;
for (const edit of ordered) {
text += ir.text.slice(cursor, edit.start) + edit.text;
cursor = edit.end;
delta += edit.text.length - (edit.end - edit.start);
cumulativeDeltas.push(delta);
exactEdits.set(`${edit.start}:${edit.end}`, edit);
}
text += ir.text.slice(cursor);
const mapOffset = (offset: number): number => {
let low = 0;
let high = ordered.length;
while (low < high) {
const middle = low + Math.floor((high - low) / 2);
if ((ordered[middle]?.end ?? Number.POSITIVE_INFINITY) <= offset) {
low = middle + 1;
} else {
high = middle;
}
}
return offset + (low > 0 ? (cumulativeDeltas[low - 1] ?? 0) : 0);
};
function rewriteMarkdownIR(ir: MarkdownIR, edits: readonly MarkdownTextEdit[]): MarkdownIR {
const { text, mapOffset } = applyMarkdownTextEdits(ir.text, edits);
const exactEdits = new Map(edits.map((edit) => [`${edit.start}:${edit.end}`, edit]));
const mapRange = <T extends { start: number; end: number }>(range: T): T => {
const exact = exactEdits.get(`${range.start}:${range.end}`);
const start = mapOffset(range.start);

2
extensions/msteams/src/reply-dispatcher.test.ts
View file

@ -683,7 +683,7 @@ describe("createMSTeamsReplyDispatcher", () => {
const latest = String(stream.update.mock.calls.at(-1)?.[0]);
expect(latest.match(/Checking/g)).toHaveLength(1);
dispatcher.replyOptions.onReasoningEnd?.();
await dispatcher.replyOptions.onReasoningEnd?.();
await dispatcher.replyOptions.onReasoningStream?.({
text: "Next thought",
isReasoningSnapshot,

145
extensions/msteams/src/reply-dispatcher.ts
View file

@ -4,13 +4,13 @@ import {
type ChannelInboundTurnPlan,
} from "openclaw/plugin-sdk/channel-inbound";
import {
normalizeAgentPlanSteps,
resolveChannelPreviewStreamMode,
resolveChannelStreamingBlockEnabled,
} from "openclaw/plugin-sdk/channel-outbound";
import { PlatformMessageNotDispatchedError } from "openclaw/plugin-sdk/error-runtime";
import { createDeferred } from "openclaw/plugin-sdk/extension-shared";
import { getGlobalHookRunner } from "openclaw/plugin-sdk/plugin-runtime";
import type { GetReplyOptions } from "openclaw/plugin-sdk/reply-runtime";
import { normalizeOptionalLowercaseString } from "openclaw/plugin-sdk/string-coerce-runtime";
import {
createChannelMessageReplyPipeline,
@ -172,7 +172,6 @@ export function createMSTeamsReplyDispatcher(params: {
content?: string;
nativeResult?: AcceptedDeliveryPart;
blockResults: AcceptedDeliveryPart[];
native: boolean;
nativeSettled: boolean;
blockSettled: boolean;
settled: boolean;
@ -184,31 +183,7 @@ export function createMSTeamsReplyDispatcher(params: {
// before another payload can mutate or overtake the native segment.
let pendingSettlement: Promise<void> | undefined;
const findPendingNativeDelivery = () =>
pendingDeliveries.find((candidate) => candidate.native && !candidate.nativeSettled);
const joinAcceptedContents = (contents: readonly (string | undefined)[]): string =>
contents.filter((content): content is string => Boolean(content)).join("\n");
const sendMessages = async (messages: MSTeamsRenderedMessage[]): Promise<string[]> => {
return sendMSTeamsMessages({
replyStyle: params.replyStyle,
app: params.app,
conversationRef: params.conversationRef,
context: params.context,
messages,
onRetry: (event) => {
params.log.debug?.("retrying send", {
replyStyle: params.replyStyle,
...event,
});
},
tokenProvider: params.tokenProvider,
sharePointSiteId: params.sharePointSiteId,
mediaMaxBytes,
feedbackLoopEnabled,
serviceUrlBoundary: resolveMSTeamsSdkCloudOptions(msteamsCfg),
});
};
pendingDeliveries.find((candidate) => !candidate.nativeSettled);
const queueDeliveryFailureSystemEvent = (failure: {
failed: number;
@ -249,7 +224,11 @@ export function createMSTeamsReplyDispatcher(params: {
});
};
const deliveryOutcome = (delivery: PendingDelivery): DeliveryOutcome => {
const settlePendingDelivery = (delivery: PendingDelivery) => {
if (delivery.settled || !delivery.blockSettled || !delivery.nativeSettled) {
return;
}
delivery.settled = true;
const acceptedParts = [
...(delivery.nativeResult ? [delivery.nativeResult] : []),
...delivery.blockResults,
@ -257,25 +236,16 @@ export function createMSTeamsReplyDispatcher(params: {
const messageIds = acceptedParts.flatMap((part) => part.messageIds);
const content =
delivery.errors.length > 0
? joinAcceptedContents(acceptedParts.map((part) => part.content))
? acceptedParts
.map((part) => part.content)
.filter(Boolean)
.join("\n")
: delivery.content;
return {
const outcome: DeliveryOutcome = {
visibleReplySent: acceptedParts.length > 0,
...(messageIds.length > 0 ? { messageIds } : {}),
...(acceptedParts.length > 0 && content !== undefined ? { content } : {}),
};
};
const settlePendingDelivery = (delivery: PendingDelivery) => {
if (
delivery.settled ||
!delivery.blockSettled ||
(delivery.native && !delivery.nativeSettled)
) {
return;
}
delivery.settled = true;
const outcome = deliveryOutcome(delivery);
if (delivery.errors.length === 0) {
delivery.finalization.resolve(outcome);
return;
@ -295,27 +265,6 @@ export function createMSTeamsReplyDispatcher(params: {
);
};
const queueReplyPayload = (
payload: ReplyPayload,
messages: MSTeamsRenderedMessage[],
native: boolean,
): PendingDelivery => {
const finalization = createDeferred<DeliveryOutcome>();
const delivery: PendingDelivery = {
messages,
finalization,
content: payload.text,
blockResults: [],
native,
nativeSettled: !native,
blockSettled: messages.length === 0,
settled: false,
errors: [],
};
pendingDeliveries.push(delivery);
return delivery;
};
const flushPendingMessages = async () => {
for (const delivery of pendingDeliveries) {
if (delivery.blockSettled) {
@ -328,7 +277,24 @@ export function createMSTeamsReplyDispatcher(params: {
const sentIds: string[] = [];
for (const msg of toSend) {
try {
const msgIds = await sendMessages([msg]);
const msgIds = await sendMSTeamsMessages({
replyStyle: params.replyStyle,
app: params.app,
conversationRef: params.conversationRef,
context: params.context,
messages: [msg],
onRetry: (event) => {
params.log.debug?.("retrying send", {
replyStyle: params.replyStyle,
...event,
});
},
tokenProvider: params.tokenProvider,
sharePointSiteId: params.sharePointSiteId,
mediaMaxBytes,
feedbackLoopEnabled,
serviceUrlBoundary: resolveMSTeamsSdkCloudOptions(msteamsCfg),
});
const validIds = msgIds.filter((id) => id.trim() && id !== "unknown");
if (msgIds.length > 0) {
delivery.blockResults.push({
@ -404,7 +370,17 @@ export function createMSTeamsReplyDispatcher(params: {
};
}
const pending = queueReplyPayload(payload, messages, native);
const pending: PendingDelivery = {
messages,
finalization: createDeferred<DeliveryOutcome>(),
content: payload.text,
blockResults: [],
nativeSettled: !native,
blockSettled: messages.length === 0,
settled: false,
errors: [],
};
pendingDeliveries.push(pending);
// When block streaming is enabled, flush immediately so blocks are
// delivered progressively instead of batching until markDispatchIdle.
@ -501,14 +477,19 @@ export function createMSTeamsReplyDispatcher(params: {
const shouldSuppressDefaultToolProgressMessages =
streamController.hasStream() && teamsStreamMode === "progress";
type PipelinePayload = Record<string, unknown>;
const progressCallbacks = streamController.hasStream()
const progressCallbacks: Pick<
GetReplyOptions,
| "onReasoningStream"
| "onReasoningEnd"
| "onToolStart"
| "onItemEvent"
| "onPlanUpdate"
| "onApprovalEvent"
> = streamController.hasStream()
? {
onReasoningStream: async (payload: PipelinePayload) => {
const text = typeof payload?.text === "string" ? payload.text : undefined;
await streamController.pushReasoningProgress(text, {
snapshot: payload?.isReasoningSnapshot === true,
onReasoningStream: async (payload) => {
await streamController.pushReasoningProgress(payload.text, {
snapshot: payload.isReasoningSnapshot === true,
});
return false;
},
@ -518,25 +499,15 @@ export function createMSTeamsReplyDispatcher(params: {
},
onToolStart: streamController.pushToolEvent,
onItemEvent: streamController.pushItemEvent,
onPlanUpdate: async (payload: PipelinePayload) => {
if (payload?.phase !== "update") {
onPlanUpdate: async (payload) => {
if (payload.phase !== "update") {
return false;
}
await streamController.pushPlanProgress(normalizeAgentPlanSteps(payload.steps), {
explanation: typeof payload.explanation === "string" ? payload.explanation : undefined,
explanationFormat: payload.explanationFormat === "plain" ? "plain" : undefined,
});
await streamController.pushPlanProgress(payload.steps, payload);
return false;
},
onApprovalEvent: async (payload: PipelinePayload) => {
await streamController.pushApprovalEvent({
...(typeof payload?.phase === "string" ? { phase: payload.phase } : {}),
...(typeof payload?.approvalId === "string" ? { approvalId: payload.approvalId } : {}),
...(typeof payload?.title === "string" ? { title: payload.title } : {}),
...(typeof payload?.command === "string" ? { command: payload.command } : {}),
...(typeof payload?.reason === "string" ? { reason: payload.reason } : {}),
...(typeof payload?.message === "string" ? { message: payload.message } : {}),
});
onApprovalEvent: async (payload) => {
await streamController.pushApprovalEvent(payload);
return false;
},
}

44
extensions/msteams/src/reply-stream-controller.ts
View file

@ -32,10 +32,6 @@ type MSTeamsNativeDeliveryFinalization = {
postNativePayloads?: ReplyPayload[];
};
type DeferredReplacementEntry =
| { kind: "payload"; payload: ReplyPayload }
| { kind: "replacement"; payload: ReplyPayload };
// The SDK throws StreamCancelledError synchronously from stream.emit/update
// when the user pressed Stop in Teams (Teams replies 403 to the next chunk
// update and the SDK flips _canceled). Match by `name` rather than importing
@ -84,7 +80,7 @@ export function createTeamsReplyStreamController(params: {
let replacementEmitFailed = false;
let replacementSettlementPending = false;
let replacementTextAwaitingAcknowledgement: { text: string; logicalText: string } | undefined;
let deferredReplacementEntries: DeferredReplacementEntry[] = [];
let deferredReplacementEntries: { kind: "payload" | "replacement"; payload: ReplyPayload }[] = [];
let queuedFinalActivity: ReturnType<typeof finalStreamActivity> | undefined;
let failedSegmentFallbackPrepared = false;
const streamEvents = (stream as { events?: TeamsStreamChunkEvents } | undefined)?.events;
@ -183,10 +179,10 @@ export function createTeamsReplyStreamController(params: {
return content || undefined;
};
const takeDeferredReplacementPayloads = (
const deferredReplacementPayloads = (
replacementFallback: ReplyPayload | undefined,
): ReplyPayload[] => {
const payloads = deferredReplacementEntries.flatMap((entry) => {
): ReplyPayload[] =>
deferredReplacementEntries.flatMap((entry) => {
if (entry.kind === "payload") {
return [entry.payload];
}
@ -197,23 +193,17 @@ export function createTeamsReplyStreamController(params: {
}
return [{ ...entry.payload, text: text || undefined }];
});
deferredReplacementEntries = [];
return payloads;
};
const finalizeWithoutReceipt = (
logicalContent?: string,
canceled = false,
): MSTeamsNativeDeliveryFinalization => {
const pending = pendingFinalPayload;
pendingFinalPayload = undefined;
const fallback =
pending && !canceled ? fallbackPayloadAfterAcknowledgedText(pending) : undefined;
pendingFinalPayload && !canceled
? fallbackPayloadAfterAcknowledgedText(pendingFinalPayload)
: undefined;
const postNativePayloads =
replacementSettlementPending && !canceled ? takeDeferredReplacementPayloads(fallback) : [];
if (canceled) {
deferredReplacementEntries = [];
}
replacementSettlementPending && !canceled ? deferredReplacementPayloads(fallback) : [];
return {
...acknowledgedNativeDelivery(),
...(!canceled && logicalContent ? { logicalContent } : {}),
@ -445,13 +435,7 @@ export function createTeamsReplyStreamController(params: {
}
let logicalContent: string | undefined;
try {
if (wasCanceled()) {
pendingFinalPayload = undefined;
deferredReplacementEntries = [];
streamFinalizationPending = false;
return acknowledgedNativeDelivery();
}
if (!streamFinalizationPending) {
if (wasCanceled() || !streamFinalizationPending) {
return acknowledgedNativeDelivery();
}
// A media-only replacement payload may arrive before its text payload.
@ -495,7 +479,6 @@ export function createTeamsReplyStreamController(params: {
}
}
const result = await stream.close();
streamFinalizationPending = false;
if (!result) {
return finalizeWithoutReceipt(logicalContent, wasCanceled());
}
@ -503,10 +486,9 @@ export function createTeamsReplyStreamController(params: {
replacementSettlementPending && replacementEmitFailed && pendingFinalPayload
? fallbackPayloadAfterAcknowledgedText(pendingFinalPayload)
: undefined;
pendingFinalPayload = undefined;
const messageId = extractMessageId(result) ?? acknowledgedStreamId;
const postNativePayloads = replacementSettlementPending
? takeDeferredReplacementPayloads(replacementFallback)
? deferredReplacementPayloads(replacementFallback)
: [];
const nativeContent = replacementEmitFailed ? acknowledgedText || undefined : content;
return {
@ -519,9 +501,6 @@ export function createTeamsReplyStreamController(params: {
} catch (err) {
if (isStreamCancelledError(err)) {
canceledLocally = true;
pendingFinalPayload = undefined;
deferredReplacementEntries = [];
streamFinalizationPending = false;
return acknowledgedNativeDelivery();
}
// Non-cancel failure during the closing emit/close. Preserve only a
@ -530,11 +509,12 @@ export function createTeamsReplyStreamController(params: {
// swallow the error — a thrown finalize would otherwise blow up
// the reply pipeline after the user already saw the response.
streamFailed = true;
streamFinalizationPending = false;
params.log?.warn?.(`msteams stream finalize failed: ${coerceErrorMessage(err)}`);
return finalizeWithoutReceipt(logicalContent);
} finally {
// This segment's acknowledged-prefix fallback has been consumed.
pendingFinalPayload = undefined;
streamFinalizationPending = false;
failedSegmentFallbackPrepared = true;
queuedFinalActivity = undefined;
replacementEmitFailed = false;

56
extensions/synology-chat/src/outbound-media.test.ts
View file

@ -555,30 +555,48 @@ describe("Synology Chat hosted outbound media", () => {
}
});
it("keeps serving limits when a fresh runtime reopens persisted capabilities", async () => {
const account = createAccount();
const prepared = await prepareSynologyHostedMedia({
account,
mediaUrl: "https://files.example.com/report.pdf",
});
installRuntime();
const requestUrl = internalCapabilityUrl(prepared.url);
const stalled = Array.from({ length: 5 }, () => makeRes({ finishOnEnd: false }));
for (const response of stalled) {
it("does not release a new runtime's serving slot when an old response finishes", async () => {
vi.useFakeTimers();
vi.setSystemTime(1_700_000_000_000);
const previous = makeRes({ finishOnEnd: false });
const current = Array.from({ length: 4 }, () => makeRes({ finishOnEnd: false }));
try {
const account = createAccount();
const prepared = await prepareSynologyHostedMedia({
account,
mediaUrl: "https://files.example.com/report.pdf",
});
const requestUrl = internalCapabilityUrl(prepared.url);
await tryHandleSynologyHostedMediaRequest(
makeReq("GET", "", { url: requestUrl }),
response,
previous,
account,
);
}
expect(previous.statusCode).toBe(200);
installRuntime();
for (const response of current) {
await tryHandleSynologyHostedMediaRequest(
makeReq("GET", "", { url: requestUrl }),
response,
account,
);
expect(response.statusCode).toBe(200);
}
previous.emit("finish");
expect(stalled.slice(0, 4).map((response) => response.statusCode)).toEqual([
200, 200, 200, 200,
]);
expect(stalled[4]?.statusCode).toBe(503);
for (const response of stalled.slice(0, 4)) {
response.emit("close");
const blocked = makeRes();
await tryHandleSynologyHostedMediaRequest(
makeReq("GET", "", { url: requestUrl }),
blocked,
account,
);
expect(blocked.statusCode).toBe(503);
} finally {
previous.emit("close");
for (const response of current) {
response.emit("close");
}
vi.useRealTimers();
}
});

54
extensions/synology-chat/src/outbound-media.ts
View file

@ -58,23 +58,32 @@ type PreparedSynologyHostedMedia = {
cleanup: () => Promise<void>;
};
const preparationLimiter = createWebhookInFlightLimiter({
maxInFlightPerKey: SYNOLOGY_OUTBOUND_MEDIA_MAX_PREPARATIONS,
maxTrackedKeys: 128,
});
const servingLimiter = createWebhookInFlightLimiter({
maxInFlightPerKey: SYNOLOGY_OUTBOUND_MEDIA_MAX_SERVES,
maxTrackedKeys: 128,
});
type ServedByteWindow = { startedAt: number; bytes: number };
function createHostedMediaLimits() {
return {
preparationLimiter: createWebhookInFlightLimiter({
maxInFlightPerKey: SYNOLOGY_OUTBOUND_MEDIA_MAX_PREPARATIONS,
maxTrackedKeys: 128,
}),
servingLimiter: createWebhookInFlightLimiter({
maxInFlightPerKey: SYNOLOGY_OUTBOUND_MEDIA_MAX_SERVES,
maxTrackedKeys: 128,
}),
servedByteWindows: new Map<string, ServedByteWindow>(),
};
}
let hostedMediaLimits = createHostedMediaLimits();
const hostedMediaStores = new Map<string, HostedOutboundMediaStore>();
const servedByteWindows = new Map<string, { startedAt: number; bytes: number }>();
let hostedMediaRuntime: ReturnType<typeof getSynologyRuntime> | undefined;
function reserveServedBytes(
servedByteWindows: Map<string, ServedByteWindow>,
accountId: string,
byteLength: number,
now = Date.now(),
): (() => void) | undefined {
const now = Date.now();
const existing = servedByteWindows.get(accountId);
const active =
existing && now - existing.startedAt < SYNOLOGY_OUTBOUND_MEDIA_SERVED_BYTES_WINDOW_MS
@ -103,7 +112,7 @@ function reserveServedBytes(
function holdServingLeaseUntilResponseDone(
res: ServerResponse,
accountId: string,
releaseServingSlot: () => void,
): { isActive: () => boolean; release: () => void } {
let released = false;
const release = () => {
@ -114,7 +123,7 @@ function holdServingLeaseUntilResponseDone(
clearTimeout(timeout);
res.off("finish", release);
res.off("close", release);
servingLimiter.release(accountId);
releaseServingSlot();
};
// `res.end()` only queues the body. Keep the account slot until the socket
// finishes or closes so slow readers cannot bypass the response concurrency cap.
@ -190,9 +199,7 @@ function getHostedMediaStore(accountId: string): HostedOutboundMediaStore {
if (hostedMediaRuntime !== runtime) {
hostedMediaRuntime = runtime;
hostedMediaStores.clear();
preparationLimiter.clear();
servingLimiter.clear();
servedByteWindows.clear();
hostedMediaLimits = createHostedMediaLimits();
}
const existing = hostedMediaStores.get(accountId);
if (existing) {
@ -446,10 +453,10 @@ export async function prepareSynologyHostedMedia(params: {
mediaReadFile?: (filePath: string) => Promise<Buffer>;
}): Promise<PreparedSynologyHostedMedia> {
const route = resolveSynologyHostedMediaRoute(params.account);
// Synchronize runtime-owned stores and counters before admitting work. A
// runtime change clears stale leases, so doing this after acquisition would
// erase the current request's slot.
// Capture limits after synchronizing the runtime; old work settles against
// the same counter owner that admitted it, even after runtime replacement.
const store = getHostedMediaStore(params.account.accountId);
const { preparationLimiter } = hostedMediaLimits;
if (!preparationLimiter.tryAcquire(params.account.accountId)) {
throw new Error(
"Synology Chat attachment preparation is busy. Retry after the current attachments finish preparing.",
@ -533,6 +540,7 @@ export async function tryHandleSynologyHostedMediaRequest(
// Runtime replacement resets the process-local limiter state. Resolve the
// matching store first so the lease acquired below belongs to that runtime.
const store = getHostedMediaStore(account.accountId);
const { servingLimiter, servedByteWindows } = hostedMediaLimits;
if (!servingLimiter.tryAcquire(account.accountId)) {
res.statusCode = 503;
res.setHeader("Retry-After", "1");
@ -542,7 +550,9 @@ export async function tryHandleSynologyHostedMediaRequest(
let responseOwnsServingLease = false;
let rollbackServedBytes: (() => void) | undefined;
let entry: HostedOutboundMediaEntry | null | undefined;
const servingLease = holdServingLeaseUntilResponseDone(res, account.accountId);
const servingLease = holdServingLeaseUntilResponseDone(res, () =>
servingLimiter.release(account.accountId),
);
try {
const routePath = toSynologyHostedMediaStoreRoutePath(url.pathname);
const metadata = await store.readMetadata(candidate.id);
@ -562,7 +572,11 @@ export async function tryHandleSynologyHostedMediaRequest(
if (method === "GET") {
// Authenticate and reserve from metadata before reading stored chunks.
// Rejected over-budget requests must not force SQLite payload reads.
rollbackServedBytes = reserveServedBytes(account.accountId, metadata.byteLength);
rollbackServedBytes = reserveServedBytes(
servedByteWindows,
account.accountId,
metadata.byteLength,
);
if (!rollbackServedBytes) {
res.statusCode = 429;
res.setHeader("Retry-After", "60");

42
extensions/telegram/src/chunk-delivery.ts
View file

@ -20,36 +20,20 @@ export function mergeTelegramPartialDeliveryError(
error: unknown,
priorDeliveryResult: PartialDeliveryResult,
): ReturnType<typeof createChannelPartialDeliveryError> {
if (!isChannelPartialDeliveryError(error)) {
return createChannelPartialDeliveryError(error, {
...priorDeliveryResult,
...(priorDeliveryResult.receipt
? {
messageIds: [
...new Set([
...(priorDeliveryResult.messageIds ?? []),
...listMessageReceiptPlatformIds(priorDeliveryResult.receipt),
]),
],
}
: {}),
});
}
const currentDeliveryResult = error.deliveryResult;
const currentDeliveryResult = isChannelPartialDeliveryError(error)
? error.deliveryResult
: undefined;
const messageIds = [
...new Set([
...(priorDeliveryResult.messageIds ?? []),
...(priorDeliveryResult.receipt
? listMessageReceiptPlatformIds(priorDeliveryResult.receipt)
: []),
...(currentDeliveryResult.messageIds ?? []),
...(currentDeliveryResult.receipt
? listMessageReceiptPlatformIds(currentDeliveryResult.receipt)
: []),
]),
...new Set(
[priorDeliveryResult, currentDeliveryResult].flatMap((result) =>
(result?.messageIds ?? []).concat(
result?.receipt ? listMessageReceiptPlatformIds(result.receipt) : [],
),
),
),
];
let receipt = currentDeliveryResult.receipt ?? priorDeliveryResult.receipt;
if (priorDeliveryResult.receipt && currentDeliveryResult.receipt) {
let receipt = currentDeliveryResult?.receipt ?? priorDeliveryResult.receipt;
if (priorDeliveryResult.receipt && currentDeliveryResult?.receipt) {
receipt = createMessageReceiptFromOutboundResults({
results: [
{ receipt: priorDeliveryResult.receipt },
@ -70,7 +54,7 @@ export function mergeTelegramPartialDeliveryError(
return createChannelPartialDeliveryError(error, {
...priorDeliveryResult,
...currentDeliveryResult,
...(messageIds.length > 0 ? { messageIds } : {}),
...((currentDeliveryResult ? messageIds.length > 0 : receipt) ? { messageIds } : {}),
...(receipt ? { receipt } : {}),
visibleReplySent: true,
});

47
extensions/telegram/src/thread-bindings-session.ts
View file

@ -5,6 +5,7 @@ import {
type SessionBindingRecord,
} from "openclaw/plugin-sdk/conversation-runtime";
import { normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runtime";
import { projectThreadBindingRecord } from "openclaw/plugin-sdk/thread-bindings-session-runtime";
import type { TelegramThreadBindingRecord } from "./thread-bindings-store.js";
export function resolveBindingKey(params: { accountId: string; conversationId: string }): string {
@ -15,30 +16,18 @@ export function toSessionBindingRecord(
record: TelegramThreadBindingRecord,
defaults: { idleTimeoutMs: number; maxAgeMs: number },
): SessionBindingRecord {
return {
bindingId: resolveBindingKey({
accountId: record.accountId,
conversationId: record.conversationId,
}),
targetSessionKey: record.targetSessionKey,
targetKind: record.targetKind === "subagent" ? "subagent" : "session",
return projectThreadBindingRecord(record, {
conversation: {
channel: "telegram",
accountId: record.accountId,
conversationId: record.conversationId,
},
status: "active",
boundAt: record.boundAt,
expiresAt: resolveThreadBindingEffectiveExpiresAt({
record,
defaultIdleTimeoutMs: defaults.idleTimeoutMs,
defaultMaxAgeMs: defaults.maxAgeMs,
}),
metadata: {
agentId: record.agentId,
label: record.label,
boundBy: record.boundBy,
lastActivityAt: record.lastActivityAt,
targetKind: record.targetKind === "subagent" ? "subagent" : "session",
lifecycle: {
expiresAt: resolveThreadBindingEffectiveExpiresAt({
record,
defaultIdleTimeoutMs: defaults.idleTimeoutMs,
defaultMaxAgeMs: defaults.maxAgeMs,
}),
idleTimeoutMs:
typeof record.idleTimeoutMs === "number"
? Math.max(0, Math.floor(record.idleTimeoutMs))
@ -47,9 +36,9 @@ export function toSessionBindingRecord(
typeof record.maxAgeMs === "number"
? Math.max(0, Math.floor(record.maxAgeMs))
: defaults.maxAgeMs,
...record.metadata,
},
};
metadata: (lifecycleMetadata) => ({ ...lifecycleMetadata, ...record.metadata }),
});
}
export function fromSessionBindingInput(params: {
@ -118,17 +107,3 @@ export function summarizeLifecycleForLog(
const maxAgeLabel = formatThreadBindingDurationLabel(Math.max(0, Math.floor(maxAgeMs)));
return `idle=${idleLabel} maxAge=${maxAgeLabel}`;
}
export function normalizeDurationMs(raw: unknown, fallback: number): number {
if (typeof raw !== "number" || !Number.isFinite(raw)) {
return fallback;
}
return Math.max(0, Math.floor(raw));
}
export function normalizeTimestampMs(raw: unknown): number {
if (typeof raw !== "number" || !Number.isFinite(raw)) {
return Date.now();
}
return Math.max(0, Math.floor(raw));
}

21
extensions/telegram/src/thread-bindings-store.ts
View file

@ -1,25 +1,18 @@
import { createHash } from "node:crypto";
import { normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runtime";
import type { AccountScopedConversationBindingRecord } from "openclaw/plugin-sdk/thread-bindings-session-runtime";
export const TELEGRAM_THREAD_BINDINGS_NAMESPACE = "telegram.thread-bindings";
export const TELEGRAM_THREAD_BINDINGS_MAX_ENTRIES = 5_000;
type TelegramBindingTargetKind = "subagent" | "acp";
export type TelegramThreadBindingRecord = {
accountId: string;
conversationId: string;
targetKind: TelegramBindingTargetKind;
targetSessionKey: string;
agentId?: string;
label?: string;
boundBy?: string;
boundAt: number;
lastActivityAt: number;
idleTimeoutMs?: number;
maxAgeMs?: number;
metadata?: Record<string, unknown>;
};
export type TelegramThreadBindingRecord =
AccountScopedConversationBindingRecord<TelegramBindingTargetKind> & {
idleTimeoutMs?: number;
maxAgeMs?: number;
metadata?: Record<string, unknown>;
};
export type TelegramThreadBindingManager = {
accountId: string;

102
extensions/telegram/src/thread-bindings.ts
View file

@ -1,16 +1,16 @@
import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
import {
registerSessionBindingAdapter,
resolveThreadBindingConversationIdFromBindingId,
resolveThreadBindingLifecycle,
unregisterSessionBindingAdapter,
type SessionBindingAdapter,
} from "openclaw/plugin-sdk/conversation-runtime";
import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
import { resolveNonNegativeIntegerOption } from "openclaw/plugin-sdk/number-runtime";
import { normalizeAccountId } from "openclaw/plugin-sdk/routing";
import { logVerbose } from "openclaw/plugin-sdk/runtime-env";
import { runQueuedStoreWrite } from "openclaw/plugin-sdk/sqlite-runtime";
import { normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runtime";
import { createAccountScopedBindingAdapter } from "openclaw/plugin-sdk/thread-bindings-session-runtime";
import { loadTelegramSendModule } from "./send-runtime.js";
import {
loadBindingsFromStore,
@ -20,8 +20,6 @@ import {
import { reconcileTelegramAcpBindingsOnStartup } from "./thread-bindings-reconcile.js";
import {
fromSessionBindingInput,
normalizeDurationMs,
normalizeTimestampMs,
resolveBindingKey,
summarizeLifecycleForLog,
toSessionBindingRecord,
@ -82,11 +80,14 @@ async function initializeThreadBindingManager(
}
const persist = params.persist ?? true;
const idleTimeoutMs = normalizeDurationMs(
const idleTimeoutMs = resolveNonNegativeIntegerOption(
params.idleTimeoutMs,
DEFAULT_THREAD_BINDING_IDLE_TIMEOUT_MS,
);
const maxAgeMs = normalizeDurationMs(params.maxAgeMs, DEFAULT_THREAD_BINDING_MAX_AGE_MS);
const maxAgeMs = resolveNonNegativeIntegerOption(
params.maxAgeMs,
DEFAULT_THREAD_BINDING_MAX_AGE_MS,
);
const loaded = await loadBindingsFromStore(accountId);
for (const entry of loaded) {
@ -166,7 +167,7 @@ async function initializeThreadBindingManager(
if (!existingLocal) {
return null;
}
const requestedActivityAt = normalizeTimestampMs(activityAt);
const requestedActivityAt = resolveNonNegativeIntegerOption(activityAt, Date.now());
const nextRecord: TelegramThreadBindingRecord = {
...existingLocal,
lastActivityAt:
@ -287,7 +288,7 @@ async function initializeThreadBindingManager(
const projectSessionBinding = (record: TelegramThreadBindingRecord) =>
toSessionBindingRecord(record, { idleTimeoutMs, maxAgeMs });
const sessionBindingAdapter: SessionBindingAdapter = {
const sessionBindingAdapter = createAccountScopedBindingAdapter({
channel: "telegram",
accountId,
capabilities: {
@ -405,68 +406,29 @@ async function initializeThreadBindingManager(
return projectSessionBinding(record);
});
},
listBySession: (targetSessionKey) =>
manager.listBySessionKey(targetSessionKey).map(projectSessionBinding),
resolveByConversation: (ref) => {
if (ref.channel !== "telegram") {
return null;
}
const conversationId = normalizeOptionalString(ref.conversationId);
if (!conversationId) {
return null;
}
const record = manager.getByConversationId(conversationId);
return record ? projectSessionBinding(record) : null;
project: projectSessionBinding,
listBySessionKey: manager.listBySessionKey,
getByConversation: (ref) => manager.getByConversationId(ref.conversationId),
touchConversation: (conversationId, at) => {
const lastActivityAt = resolveNonNegativeIntegerOption(at, Date.now());
manager.updateConversationSync(conversationId, (current) => ({ ...current, lastActivityAt }));
},
touch: (bindingId, at) => {
const conversationId = resolveThreadBindingConversationIdFromBindingId({
accountId,
bindingId,
});
if (conversationId) {
const lastActivityAt = normalizeTimestampMs(at ?? Date.now());
manager.updateConversationSync(conversationId, (current) => ({
...current,
lastActivityAt,
}));
}
},
touchAsync: async (bindingId, at) => {
const conversationId = resolveThreadBindingConversationIdFromBindingId({
accountId,
bindingId,
});
if (!conversationId) {
return;
}
await manager.touchConversation(conversationId, at);
},
unbind: async (input) => {
if (input.targetSessionKey?.trim()) {
const removed = await manager.unbindBySessionKey({
targetSessionKey: input.targetSessionKey,
reason: input.reason,
sendFarewell: false,
throwOnPersistError: true,
});
return removed.map(projectSessionBinding);
}
const conversationId = resolveThreadBindingConversationIdFromBindingId({
accountId,
bindingId: input.bindingId,
});
if (!conversationId) {
return [];
}
const removed = await manager.unbindConversation({
touchConversationAsync: manager.touchConversation,
unbindConversation: (conversationId, reason) =>
manager.unbindConversation({
conversationId,
reason: input.reason,
reason,
sendFarewell: false,
throwOnPersistError: true,
});
return removed ? [projectSessionBinding(removed)] : [];
},
};
}),
unbindBySessionKey: (targetSessionKey, reason) =>
manager.unbindBySessionKey({
targetSessionKey,
reason,
sendFarewell: false,
throwOnPersistError: true,
}),
});
registerSessionBindingAdapter(sessionBindingAdapter);
@ -553,7 +515,7 @@ export async function setTelegramThreadBindingIdleTimeoutBySessionKeyAsync(param
if (!manager) {
return [];
}
const idleTimeoutMs = normalizeDurationMs(params.idleTimeoutMs, 0);
const idleTimeoutMs = resolveNonNegativeIntegerOption(params.idleTimeoutMs, 0);
return manager.updateBySessionKey(params.targetSessionKey, (entry, now) => ({
...entry,
idleTimeoutMs,
@ -570,7 +532,7 @@ export async function setTelegramThreadBindingMaxAgeBySessionKeyAsync(params: {
if (!manager) {
return [];
}
const maxAgeMs = normalizeDurationMs(params.maxAgeMs, 0);
const maxAgeMs = resolveNonNegativeIntegerOption(params.maxAgeMs, 0);
return manager.updateBySessionKey(params.targetSessionKey, (entry, now) => ({
...entry,
maxAgeMs,
@ -607,7 +569,7 @@ export function setTelegramThreadBindingIdleTimeoutBySessionKey(params: {
accountId?: string;
idleTimeoutMs: number;
}): TelegramThreadBindingRecord[] {
const idleTimeoutMs = normalizeDurationMs(params.idleTimeoutMs, 0);
const idleTimeoutMs = resolveNonNegativeIntegerOption(params.idleTimeoutMs, 0);
return updateTelegramBindingsSynchronously({
...params,
update: (entry, now) => ({ ...entry, idleTimeoutMs, lastActivityAt: now }),
@ -620,7 +582,7 @@ export function setTelegramThreadBindingMaxAgeBySessionKey(params: {
accountId?: string;
maxAgeMs: number;
}): TelegramThreadBindingRecord[] {
const maxAgeMs = normalizeDurationMs(params.maxAgeMs, 0);
const maxAgeMs = resolveNonNegativeIntegerOption(params.maxAgeMs, 0);
return updateTelegramBindingsSynchronously({
...params,
update: (entry, now) => ({ ...entry, maxAgeMs, lastActivityAt: now }),

56
extensions/zalo/src/outbound-media.test.ts
View file

@ -1,11 +1,14 @@
import fs from "node:fs";
import path from "node:path";
import type { OpenKeyedStoreOptions } from "openclaw/plugin-sdk/plugin-state-runtime";
// Zalo tests cover outbound media plugin behavior.
import { createPluginStateKeyedStoreForTests } from "openclaw/plugin-sdk/plugin-state-test-runtime";
import {
createPluginStateKeyedStoreForTests,
resetPluginStateStoreForTests,
} from "openclaw/plugin-sdk/plugin-state-test-runtime";
import type { PluginRuntime } from "openclaw/plugin-sdk/runtime-store";
import { closeOpenClawStateDatabaseAsync } from "openclaw/plugin-sdk/sqlite-runtime-testing";
import { resolvePreferredOpenClawTmpDir } from "openclaw/plugin-sdk/temp-path";
import { beforeEach, describe, expect, it, vi } from "vitest";
import { useAutoCleanupTempDirTracker } from "openclaw/plugin-sdk/test-env";
import { afterAll, beforeEach, describe, expect, it, vi } from "vitest";
const loadWebMediaMock = vi.hoisted(() => vi.fn());
@ -22,11 +25,16 @@ import {
} from "./outbound-media.js";
import { setZaloRuntime } from "./runtime.js";
const testStateDirs = useAutoCleanupTempDirTracker((cleanup) => {
afterAll(async () => {
await closeOpenClawStateDatabaseAsync();
resetPluginStateStoreForTests();
cleanup();
});
});
const testStateEnv: NodeJS.ProcessEnv = {
...process.env,
OPENCLAW_STATE_DIR: fs.mkdtempSync(
path.join(resolvePreferredOpenClawTmpDir(), "openclaw-zalo-media-"),
),
OPENCLAW_STATE_DIR: testStateDirs.make("openclaw-zalo-media-", resolvePreferredOpenClawTmpDir()),
};
function openTestStore<T>(options: OpenKeyedStoreOptions) {
@ -47,10 +55,11 @@ function createMockResponse() {
};
}
function installZaloRuntimeForTest(): void {
function installZaloRuntimeForTest(env = testStateEnv): void {
setZaloRuntime({
state: {
openKeyedStore: <T>(options: OpenKeyedStoreOptions) => openTestStore<T>(options),
openKeyedStore: <T>(options: OpenKeyedStoreOptions) =>
createPluginStateKeyedStoreForTests<T>("zalo", { ...options, env }),
},
} as unknown as PluginRuntime);
}
@ -196,6 +205,35 @@ describe("zalo outbound hosted media", () => {
expect(getResponse.res.end).toHaveBeenCalledWith(Buffer.from("image-bytes"));
});
it("uses the current runtime's state after runtime replacement", async () => {
const hostedUrl = new URL(await prepareMedia());
const replacementStateDir = testStateDirs.make(
"openclaw-zalo-media-replacement-",
resolvePreferredOpenClawTmpDir(),
);
try {
installZaloRuntimeForTest({ ...testStateEnv, OPENCLAW_STATE_DIR: replacementStateDir });
const response = createMockResponse();
await tryHandleHostedZaloMediaRequest(
{ method: "HEAD", url: `${hostedUrl.pathname}${hostedUrl.search}` } as never,
response.res as never,
);
expect(response.res.statusCode).toBe(404);
expect(response.res.end).toHaveBeenCalledWith("Not Found");
installZaloRuntimeForTest();
const original = createMockResponse();
await tryHandleHostedZaloMediaRequest(
{ method: "GET", url: `${hostedUrl.pathname}${hostedUrl.search}` } as never,
original.res as never,
);
expect(original.res.statusCode).toBe(200);
expect(original.res.end).toHaveBeenCalledWith(Buffer.from("image-bytes"));
} finally {
installZaloRuntimeForTest();
}
});
it("rejects hosted media preparation when the expiry would exceed a valid Date", async () => {
vi.useFakeTimers();
vi.setSystemTime(new Date(8_640_000_000_000_000));

13
extensions/zalo/src/outbound-media.ts
View file

@ -25,10 +25,14 @@ const ZALO_OUTBOUND_MEDIA_MAX_CHUNK_ROWS =
ZALO_OUTBOUND_MEDIA_MAX_ENTRIES * ZALO_OUTBOUND_MEDIA_CHUNK_ROWS_PER_ENTRY_BUDGET;
let hostedZaloMediaStore: HostedOutboundMediaStore | undefined;
let hostedZaloMediaRuntime: ReturnType<typeof getZaloRuntime> | undefined;
function createHostedZaloMediaStore(): HostedOutboundMediaStore {
function getHostedZaloMediaStore(): HostedOutboundMediaStore {
const runtime = getZaloRuntime();
return createHostedOutboundMediaStore({
if (hostedZaloMediaStore && hostedZaloMediaRuntime === runtime) {
return hostedZaloMediaStore;
}
hostedZaloMediaStore = createHostedOutboundMediaStore({
metadataStore: runtime.state.openKeyedStore<HostedOutboundMediaMetaRecord>({
namespace: ZALO_OUTBOUND_MEDIA_NAMESPACE,
maxEntries: ZALO_OUTBOUND_MEDIA_MAX_ENTRIES + 16,
@ -42,10 +46,7 @@ function createHostedZaloMediaStore(): HostedOutboundMediaStore {
maxChunkRows: ZALO_OUTBOUND_MEDIA_MAX_CHUNK_ROWS,
resolveExpiresAtMs: (ttlMs) => resolveExpiresAtMsFromDurationMs(ttlMs),
});
}
function getHostedZaloMediaStore(): HostedOutboundMediaStore {
hostedZaloMediaStore ??= createHostedZaloMediaStore();
hostedZaloMediaRuntime = runtime;
return hostedZaloMediaStore;
}

32
packages/markdown-core/src/ir-spans.ts
View file

@ -3,6 +3,38 @@ import type {
AssistantTranscriptRoleHeaderKind,
} from "./assistant-transcript-headers.js";
/** A replacement in the original text's UTF-16 coordinates; equal bounds insert text. */
export type MarkdownTextEdit = { start: number; end: number; text: string };
/** Apply disjoint edits; equal-position insertions keep caller order and shift their boundary. */
export function applyMarkdownTextEdits(text: string, edits: readonly MarkdownTextEdit[]) {
let output = "";
let cursor = 0;
const shifts = edits
.toSorted((a, b) => a.start - b.start || a.end - b.end)
.map((edit) => {
output += text.slice(cursor, edit.start) + edit.text;
cursor = edit.end;
return { end: edit.end, shift: output.length - cursor };
});
return {
text: output + text.slice(cursor),
mapOffset: (offset: number): number => {
let low = 0;
let high = shifts.length;
while (low < high) {
const middle = low + Math.floor((high - low) / 2);
if ((shifts[middle]?.end ?? Number.POSITIVE_INFINITY) <= offset) {
low = middle + 1;
} else {
high = middle;
}
}
return offset + (shifts[low - 1]?.shift ?? 0);
},
};
}
export type MarkdownStyle =
| "bold"
| "italic"

39
packages/markdown-core/src/ir.text-edits.test.ts Normal file
View file

@ -0,0 +1,39 @@
import { describe, expect, it } from "vitest";
import { applyMarkdownTextEdits } from "./ir-spans.js";
describe("applyMarkdownTextEdits", () => {
it("projects UTF-16 formatting boundaries through insertion, replacement, and removal", () => {
const result = applyMarkdownTextEdits("😀 abc tail", [
{ start: 7, end: 11, text: "" },
{ start: 0, end: 0, text: "#" },
{ start: 3, end: 6, text: "`abc`" },
]);
expect(result.text).toBe("#😀 `abc` ");
expect([0, 2, 3, 6, 7, 11].map(result.mapOffset)).toEqual([1, 3, 4, 9, 10, 10]);
expect(result.text.slice(result.mapOffset(3), result.mapOffset(6))).toBe("`abc`");
});
it("keeps equal-position insertions in caller order and leaves an empty projection unchanged", () => {
const result = applyMarkdownTextEdits("XY", [
{ start: 1, end: 1, text: "first" },
{ start: 1, end: 1, text: "second" },
]);
expect(result.text).toBe("XfirstsecondY");
expect(result.mapOffset(1)).toBe(12);
const unchanged = applyMarkdownTextEdits("😀XY", []);
expect(unchanged.text).toBe("😀XY");
expect(unchanged.mapOffset(2)).toBe(2);
});
it("applies boundary insertions before replacing the following source range", () => {
const result = applyMarkdownTextEdits("AB", [
{ start: 1, end: 2, text: "XY" },
{ start: 1, end: 1, text: "b" },
]);
expect(result.text).toBe("AbXY");
expect([1, 2].map(result.mapOffset)).toEqual([2, 4]);
});
});

4
scripts/plugin-sdk-surface-report.mts
View file

@ -197,7 +197,7 @@ export function readPluginSdkSurfaceBudgets(env: NodeJS.ProcessEnv = process.env
// +1: owner-approved async coding-tool construction with retained sync compatibility.
// +4: executor controller, binding, context, and resolver.
// +1: required session cleanup failure preserves native ownership before host reset.
3651,
3648,
env,
),
publicFunctionExports: readPluginSdkSurfaceBudgetEnv(
@ -210,7 +210,7 @@ export function readPluginSdkSurfaceBudgets(env: NodeJS.ProcessEnv = process.env
// +1: owner-approved async agent-end preparation with retained sync compatibility.
// +1: owner-approved async coding-tool construction with retained sync compatibility.
// +1: resolve the controller from the current invocation registry.
2114,
2110,
env,
),
publicDeprecatedExports: readPluginSdkSurfaceBudgetEnv(

56
src/channels/turn/delivery-result.test.ts
View file

@ -3,6 +3,7 @@ import { describe, expect, it } from "vitest";
import type { MessageReceipt } from "../message/types.js";
import {
createAcceptedChannelDeliveryResult,
createChannelDeliveryAccumulator,
createChannelDeliveryResultFromReceipt,
createChannelPartialDeliveryError,
isChannelPartialDeliveryError,
@ -13,6 +14,61 @@ import {
resolveChannelTurnDispatchCounts,
} from "./dispatch-result.js";
describe("createChannelDeliveryAccumulator", () => {
it("suppresses an empty batch and leaves failures before acceptance unchanged", () => {
const accepted = createChannelDeliveryAccumulator();
const error = new Error("send rejected");
expect(accepted.result()).toEqual({
visibleReplySent: false,
suppression: { reason: "no_visible_result" },
});
expect(accepted.partialError(error)).toBe(error);
expect(
accepted.partialError(
createChannelPartialDeliveryError(error, {
visibleReplySent: true,
messageIds: ["accepted-attachment"],
}),
),
).toMatchObject({
deliveryResult: { visibleReplySent: true, messageIds: ["accepted-attachment"], content: "" },
});
});
it("keeps immutable snapshots and a nested failure's accepted subset in send order", () => {
const accepted = createChannelDeliveryAccumulator({ kind: "media", replyToId: "parent" });
accepted.add({ messageId: "first" }, "first accepted text");
const snapshot = accepted.result();
accepted.add({ messageId: "attachment" }, "");
const failure = accepted.partialError(
createChannelPartialDeliveryError(new Error("last caption rejected"), {
visibleReplySent: true,
messageIds: ["last-attachment"],
content: "last accepted text",
}),
);
expect(failure).toMatchObject({
deliveryResult: {
visibleReplySent: true,
messageIds: ["first", "attachment", "last-attachment"],
content: "first accepted text\nlast accepted text",
receipt: { primaryPlatformMessageId: "first", replyToId: "parent" },
},
});
expect(snapshot).toMatchObject({
messageIds: ["first"],
content: "first accepted text",
receipt: { raw: [{ messageId: "first" }] },
});
expect(accepted.result()).toMatchObject({
messageIds: ["first", "attachment"],
content: "first accepted text",
});
});
});
describe("createAcceptedChannelDeliveryResult", () => {
it.each([true, false])(
"retains accepted prefixes and nested partial identities (receipt=%s)",

44
src/channels/turn/delivery-result.ts
View file

@ -9,9 +9,53 @@ import type {
ChannelDeliveryOutcome,
ChannelDeliveryResult,
} from "./delivery-outcome.js";
import {
createChannelPartialDeliveryError,
isChannelPartialDeliveryError,
} from "./partial-delivery-error.js";
type ReceiptParams = Parameters<typeof createMessageReceiptFromOutboundResults>[0];
/** Accumulates accepted sends without recording text or identity from failed attempts. */
export function createChannelDeliveryAccumulator(
params: Pick<ReceiptParams, "kind" | "replyToId"> = {},
) {
const results: ReceiptParams["results"][number][] = [];
const contents: string[] = [];
const acceptedResult = (partial?: ChannelDeliveryOutcome) =>
createAcceptedChannelDeliveryResult({
...params,
results: [...results],
...(partial ? { deliveryResults: [partial] } : {}),
content: [...contents, partial?.content].filter(Boolean).join("\n"),
});
return {
get size() {
return results.length;
},
add(result: ReceiptParams["results"][number], content?: string) {
results.push(result);
if (content) {
contents.push(content);
}
},
result() {
return results.length > 0
? acceptedResult()
: {
visibleReplySent: false as const,
suppression: { reason: "no_visible_result" as const },
};
},
partialError(error: unknown): unknown {
const partial = isChannelPartialDeliveryError(error) ? error.deliveryResult : undefined;
return results.length > 0 || partial
? createChannelPartialDeliveryError(error, acceptedResult(partial))
: error;
},
};
}
/** Aggregates caller-confirmed sends, preserving nested receipts before legacy message IDs. */
export function createAcceptedChannelDeliveryResult(
params: Pick<ReceiptParams, "kind" | "replyToId"> & {

107
src/infra/outbound/account-scoped-conversation-bindings.ts
View file

@ -22,6 +22,10 @@ import {
updateCurrentConversationBindingRecord,
} from "./current-conversation-bindings.js";
import type { CurrentConversationBindingTouch } from "./current-conversation-bindings.worker-contract.js";
import {
createAccountScopedBindingAdapter,
projectThreadBindingRecord,
} from "./session-binding-adapter.js";
import { SessionBindingError } from "./session-binding-errors.js";
import {
nativeSessionBindingSelection,
@ -127,28 +131,15 @@ export function createAccountScopedConversationBindingManager<TKind extends stri
idleExpiresAt != null && maxAgeExpiresAt != null
? Math.min(idleExpiresAt, maxAgeExpiresAt)
: (idleExpiresAt ?? maxAgeExpiresAt);
return {
bindingId: `${record.accountId}:${record.conversationId}`,
targetSessionKey: record.targetSessionKey,
targetKind: params.toSessionBindingTargetKind(record.targetKind),
return projectThreadBindingRecord(record, {
conversation: {
channel: params.channel,
accountId: record.accountId,
conversationId: record.conversationId,
},
status: "active",
boundAt: record.boundAt,
expiresAt,
metadata: {
...metadata,
agentId: record.agentId,
label: record.label,
boundBy: record.boundBy,
lastActivityAt: record.lastActivityAt,
idleTimeoutMs,
maxAgeMs,
},
};
targetKind: params.toSessionBindingTargetKind(record.targetKind),
lifecycle: { expiresAt, idleTimeoutMs, maxAgeMs },
metadata: (lifecycleMetadata) => ({ ...metadata, ...lifecycleMetadata }),
});
};
const conversationRef = (conversationId: string) =>
normalizeConversationRef({
@ -308,11 +299,33 @@ export function createAccountScopedConversationBindingManager<TKind extends stri
const sessionBindingAdapter: SessionBindingAdapter &
NativeSessionBindingSelection &
NativeSessionBindingListing = {
channel: params.channel,
accountId,
capabilities: {
placements: ["current"],
},
...createAccountScopedBindingAdapter<SessionBindingRecord>({
channel: params.channel,
accountId,
capabilities: { placements: ["current"] },
project: (record) => record,
listBySessionKey: (targetSessionKey) =>
listCurrentConversationBindingRecordsBySession(targetSessionKey, accountScope),
getByConversation: (ref) =>
resolveCurrentConversationBindingRecord(conversationRef(ref.conversationId)),
touchConversation: manager.touchConversation,
unbindBySessionKey: (targetSessionKey) =>
deleteCurrentConversationBindingRecordsBySession(targetSessionKey.trim(), accountScope),
unbindConversation: (conversationId) =>
updateCurrentConversationBindingRecord(conversationRef(conversationId), () => null)
.previous,
bind: async (input) => {
if (input.conversation.channel !== params.channel || input.placement === "child") {
return null;
}
return bindConversationRecord({
conversationId: input.conversation.conversationId,
targetKind: input.targetKind,
targetSessionKey: input.targetSessionKey,
metadata: input.metadata,
});
},
}),
[nativeSessionBindingSelection]: async (refs) => {
const conversations = refs.map((ref) =>
matchesAccount(ref) ? conversationRef(ref.conversationId) : null,
@ -326,31 +339,12 @@ export function createAccountScopedConversationBindingManager<TKind extends stri
let index = 0;
return conversations.map((ref) => (ref ? (records[index++] ?? null) : null));
},
bind: async (input) => {
if (input.conversation.channel !== params.channel || input.placement === "child") {
return null;
}
return bindConversationRecord({
conversationId: input.conversation.conversationId,
targetKind: input.targetKind,
targetSessionKey: input.targetSessionKey,
metadata: input.metadata,
});
},
listBySession: (targetSessionKey) =>
listCurrentConversationBindingRecordsBySession(targetSessionKey, accountScope),
[nativeSessionBindingListBySession]: (targetSessionKey) =>
listCurrentConversationBindingRecordsBySessionAsync(
targetSessionKey,
accountScope,
assertCurrent,
),
resolveByConversation: (ref) => {
if (ref.channel !== params.channel) {
return null;
}
return resolveCurrentConversationBindingRecord(conversationRef(ref.conversationId));
},
inspectByConversation: (ref) =>
ref.channel === params.channel
? inspectCurrentConversationBindingRecord(conversationRef(ref.conversationId))
@ -389,35 +383,6 @@ export function createAccountScopedConversationBindingManager<TKind extends stri
);
}
},
touch: (bindingId, at) => {
const conversationId = resolveThreadBindingConversationIdFromBindingId({
accountId,
bindingId,
});
if (conversationId) {
manager.touchConversation(conversationId, at);
}
},
unbind: async (input) => {
if (input.targetSessionKey?.trim()) {
return deleteCurrentConversationBindingRecordsBySession(
input.targetSessionKey.trim(),
accountScope,
);
}
const conversationId = resolveThreadBindingConversationIdFromBindingId({
accountId,
bindingId: input.bindingId,
});
if (!conversationId) {
return [];
}
const { previous } = updateCurrentConversationBindingRecord(
conversationRef(conversationId),
() => null,
);
return previous ? [previous] : [];
},
};
registerSessionBindingAdapter(sessionBindingAdapter);

106
src/infra/outbound/session-binding-adapter.ts Normal file
View file

@ -0,0 +1,106 @@
import { resolveThreadBindingConversationIdFromBindingId } from "../../channels/thread-binding-id.js";
import type { SessionBindingAdapter } from "./session-binding-service.js";
import type {
BindingTargetKind,
ConversationRef,
SessionBindingRecord,
} from "./session-binding.types.js";
/** Projects plugin-owned lifecycle facts without normalizing its identity or expiry policy. */
export function projectThreadBindingRecord(
record: {
accountId: string;
targetSessionKey: string;
boundAt: number;
lastActivityAt: number;
agentId?: string;
label?: string;
boundBy?: string;
},
params: {
conversation: Omit<ConversationRef, "accountId">;
bindingId?: string;
targetKind: BindingTargetKind;
lifecycle: { expiresAt?: number; idleTimeoutMs: number; maxAgeMs: number };
metadata?: (lifecycleMetadata: Record<string, unknown>) => Record<string, unknown>;
},
): SessionBindingRecord {
// Keep persisted projection bytes aligned with the worker's conversation capture.
const { channel, ...conversation } = params.conversation;
const metadata = {
agentId: record.agentId,
label: record.label,
boundBy: record.boundBy,
lastActivityAt: record.lastActivityAt,
idleTimeoutMs: params.lifecycle.idleTimeoutMs,
maxAgeMs: params.lifecycle.maxAgeMs,
};
return {
bindingId: params.bindingId ?? `${record.accountId}:${params.conversation.conversationId}`,
targetSessionKey: record.targetSessionKey,
targetKind: params.targetKind,
conversation: { channel, accountId: record.accountId, ...conversation },
status: "active",
boundAt: record.boundAt,
expiresAt: params.lifecycle.expiresAt,
metadata: params.metadata ? params.metadata(metadata) : metadata,
};
}
/** Adapts account-prefixed ids while leaving mutations and their settlement with the manager. */
export function createAccountScopedBindingAdapter<T>(params: {
channel: string;
accountId: string;
capabilities: SessionBindingAdapter["capabilities"];
bind: NonNullable<SessionBindingAdapter["bind"]>;
project: (record: T) => SessionBindingRecord;
listBySessionKey: (targetSessionKey: string) => T[];
getByConversation: (ref: ConversationRef) => T | null | undefined;
touchConversation: (conversationId: string, at?: number) => unknown;
touchConversationAsync?: (conversationId: string, at?: number) => Promise<unknown>;
unbindConversation: (conversationId: string, reason: string) => T | null | Promise<T | null>;
unbindBySessionKey: (targetSessionKey: string, reason: string) => T[] | Promise<T[]>;
}): SessionBindingAdapter {
const touchAsync = params.touchConversationAsync;
const conversationIdFromBinding = (bindingId?: string) =>
resolveThreadBindingConversationIdFromBindingId({ accountId: params.accountId, bindingId });
return {
channel: params.channel,
accountId: params.accountId,
capabilities: params.capabilities,
bind: params.bind,
listBySession: (sessionKey) => params.listBySessionKey(sessionKey).map(params.project),
resolveByConversation: (ref) => {
const record = ref.channel === params.channel ? params.getByConversation(ref) : null;
return record ? params.project(record) : null;
},
touch: (bindingId, at) => {
const conversationId = conversationIdFromBinding(bindingId);
if (conversationId) {
params.touchConversation(conversationId, at);
}
},
...(touchAsync
? {
touchAsync: async (bindingId: string, at?: number) => {
const conversationId = conversationIdFromBinding(bindingId);
if (conversationId) {
await touchAsync(conversationId, at);
}
},
}
: {}),
unbind: async (input) => {
if (input.targetSessionKey?.trim()) {
return (await params.unbindBySessionKey(input.targetSessionKey, input.reason)).map(
params.project,
);
}
const conversationId = conversationIdFromBinding(input.bindingId);
const removed = conversationId
? await params.unbindConversation(conversationId, input.reason)
: null;
return removed ? [params.project(removed)] : [];
},
};
}

13
src/plugin-sdk/allow-from.ts
View file

@ -139,19 +139,6 @@ export type BasicAllowlistResolutionEntry = {
note?: string;
};
/** Clone allowlist resolution entries into a plain serializable shape for UI and docs output. */
export function mapBasicAllowlistResolutionEntries(
entries: BasicAllowlistResolutionEntry[],
): BasicAllowlistResolutionEntry[] {
return entries.map((entry) => ({
input: entry.input,
resolved: entry.resolved,
id: entry.id,
name: entry.name,
note: entry.note,
}));
}
/** Map allowlist inputs sequentially so resolver side effects stay ordered and predictable. */
export async function mapAllowlistResolutionInputs<T>(params: {
/** Ordered allowlist inputs to resolve. */

1
src/plugin-sdk/channel-outbound.ts
View file

@ -216,6 +216,7 @@ export {
deliverWithFinalizableLivePreviewAdapter,
} from "../channels/message/live.js";
export type { LivePreviewDeliveryResult, LivePreviewLifecycle } from "../channels/message/live.js";
export { createChannelDeliveryAccumulator } from "../channels/turn/delivery-result.js";
export {
createMessageReceiptFromOutboundResults,
listMessageReceiptPlatformIds,

1
src/plugin-sdk/computer-use.ts
View file

@ -5,7 +5,6 @@ export {
ComputerUseCapabilityDescriptorSchema,
ScreenSnapshotParamsSchema,
ScreenSnapshotResultSchema,
compileComputerUseValidator,
parseComputerActParamsJSON,
parseScreenSnapshotParamsJSON,
} from "../plugins/computer-use-contract.js";

22
src/plugin-sdk/extension-shared.ts
View file

@ -4,7 +4,6 @@ import { createAmbientNodeProxyAgent, hasAmbientNodeProxyConfigured } from "@ope
import type { z } from "zod";
import { resolveActiveManagedProxyTlsOptions } from "../infra/net/proxy/managed-proxy-undici.js";
import { createDeferredCore } from "../shared/deferred.js";
import { runPassiveAccountLifecycle } from "./channel-lifecycle.core.js";
import { createLoggerBackedRuntime } from "./runtime-logger.internal.js";
export { safeParseJsonWithSchema, safeParseWithSchema } from "../utils/zod-parse.js";
export { buildTimeoutAbortSignal } from "../utils/fetch-timeout.js";
@ -24,10 +23,6 @@ type TrafficStatusSnapshot = {
lastOutboundAt?: number | null;
};
type StoppableMonitor = {
stop: () => void;
};
type RequireOpenAllowFromFn = (params: {
policy?: string;
allowFrom?: Array<string | number>;
@ -75,23 +70,6 @@ export function buildTrafficStatusSummary(snapshot?: TrafficStatusSnapshot | nul
};
}
/**
* Runs a passive monitor until the supplied abort signal fires, then calls `stop()`.
* This adapts simple plugin monitors to the shared passive account lifecycle.
*/
export async function runStoppablePassiveMonitor<TMonitor extends StoppableMonitor>(params: {
abortSignal: AbortSignal;
start: () => Promise<TMonitor>;
}): Promise<void> {
await runPassiveAccountLifecycle({
abortSignal: params.abortSignal,
start: params.start,
stop: async (monitor) => {
monitor.stop();
},
});
}
/**
* Returns the provided runtime or creates a logger-backed fallback for monitor-only paths.
* The fallback cannot exit the process, so command/runtime callers should inject a real runtime.

4
src/plugin-sdk/gateway-runtime.ts
View file

@ -5,10 +5,6 @@ export { isGatewayClientRequestError, isGatewayTransportError } from "../gateway
// they must use the canonical redactor so URL userinfo/tokens never print.
export { redactSensitiveUrlLikeString } from "@openclaw/net-policy/redact-sensitive-url";
export { isLoopbackHost } from "../gateway/net.js";
export async function resolveAdvertisedLanHost(): Promise<string | null> {
const runtime = await import("../infra/advertised-lan-host.js");
return await runtime.resolveAdvertisedLanHostCore();
}
export { resolveHostedPluginSurfaceUrl } from "../gateway/hosted-plugin-surface-url.js";
export type { HostedPluginSurfaceUrlParams } from "../gateway/hosted-plugin-surface-url.js";
export {

15
src/plugin-sdk/json-store.ts
View file

@ -1,6 +1,5 @@
// JSON store helpers provide small atomic persistence primitives for plugin runtime state.
import { pathExists } from "../infra/fs-safe.js";
import { tryReadJson, tryReadJsonSync, writeJson, writeJsonSync } from "../infra/json-files.js";
import { tryReadJsonSync, writeJson, writeJsonSync } from "../infra/json-files.js";
/** Read small JSON blobs synchronously for token/state caches. */
// oxlint-disable-next-line typescript-eslint/no-unnecessary-type-parameters -- public SDK compatibility helper.
@ -11,18 +10,6 @@ export function loadJsonFile<T = unknown>(filePath: string): T | undefined {
/** Persist small JSON blobs synchronously with restrictive permissions. */
export const saveJsonFile = writeJsonSync;
/** Read JSON from disk and fall back cleanly when the file is missing or invalid. */
export async function readJsonFileWithFallback<T>(
filePath: string,
fallback: T,
): Promise<{ value: T; exists: boolean }> {
const parsed = await tryReadJson<T>(filePath);
if (parsed != null) {
return { value: parsed, exists: true };
}
return { value: fallback, exists: await pathExists(filePath) };
}
/** Write JSON with secure file permissions and atomic replacement semantics. */
export async function writeJsonFileAtomically(filePath: string, value: unknown): Promise<void> {
await writeJson(filePath, value, {

1
src/plugin-sdk/provider-auth.ts
View file

@ -42,7 +42,6 @@ export {
export {
ensureApiKeyFromEnvOrPrompt,
ensureApiKeyFromOptionEnvOrPrompt,
normalizeSecretInputModeInput,
promptSecretRefForSetup,
resolveSecretInputModeForEnvSelection,
} from "../plugins/provider-auth-input.js";

4
src/plugin-sdk/text-chunking.ts
View file

@ -34,6 +34,10 @@ export function chunkTextForOutbound(
}
/** Markdown IR parsing and slicing primitives for plugin-owned renderers. */
export {
applyMarkdownTextEdits,
type MarkdownTextEdit,
} from "../../packages/markdown-core/src/ir-spans.js";
export {
chunkMarkdownIR,
markdownToIR,

5
src/plugin-sdk/thread-bindings-session-runtime.ts
View file

@ -14,3 +14,8 @@ export {
type SessionBindingAdapter,
type SessionBindingRecord,
} from "../infra/outbound/session-binding-service.js";
export {
createAccountScopedBindingAdapter,
projectThreadBindingRecord,
} from "../infra/outbound/session-binding-adapter.js";
export type { AccountScopedConversationBindingRecord } from "../infra/outbound/account-scoped-conversation-bindings.js";

11
src/plugins/computer-use-contract.ts
View file

@ -1,5 +1,4 @@
import { type Static, type TProperties, type TSchema, Type } from "typebox";
import { Compile } from "typebox/compile";
import { type Static, type TProperties, Type } from "typebox";
import { lazyCompile } from "../../packages/gateway-protocol/src/protocol-validator.js";
import { closedObject } from "../../packages/gateway-protocol/src/schema/closed-object.js";
@ -417,14 +416,6 @@ export type ScreenSnapshotResult = Static<typeof ScreenSnapshotResultSchema>;
type ComputerUseValidator<Value> = (value: unknown) => value is Value;
/** Compile one Computer Use wire schema into a reusable type-guard validator. */
export function compileComputerUseValidator<const Schema extends TSchema>(
schema: Schema,
): ComputerUseValidator<Static<Schema>> {
const validator = Compile(schema);
return (value: unknown): value is Static<Schema> => validator.Check(value);
}
const validateComputerActParams = lazyCompile(ComputerActParamsSchema);
const validateComputerActResult = lazyCompile(ComputerActResultSchema);
const validateComputerUseCapabilityDescriptor = lazyCompile(ComputerUseCapabilityDescriptorSchema);

11
src/plugins/provider-auth-input.ts
View file

@ -90,17 +90,6 @@ export function normalizeTokenProviderInput(
return normalizeOptionalLowercaseString(tokenProvider);
}
/** Normalizes secret input mode values accepted by provider setup. */
export function normalizeSecretInputModeInput(
secretInputMode: string | null | undefined,
): SecretInputMode | undefined {
const normalized = normalizeOptionalLowercaseString(secretInputMode);
if (normalized === "plaintext" || normalized === "ref") {
return normalized;
}
return undefined;
}
/** Resolves an API key from CLI options first, then environment or prompt fallback. */
export async function ensureApiKeyFromOptionEnvOrPrompt(
params: Parameters<typeof ensureApiKeyFromEnvOrPrompt>[0] & {

46
src/shared/text/strip-markdown.ts
View file

@ -1,6 +1,10 @@
import { findAssistantTranscriptRoleHeaderSpans } from "../../../packages/markdown-core/src/assistant-transcript-headers.js";
import { applyConstructFallbacks } from "../../../packages/markdown-core/src/construct-fallbacks.js";
import type { FormatCapabilityProfile } from "../../../packages/markdown-core/src/format-capabilities.js";
import {
applyMarkdownTextEdits,
type MarkdownTextEdit,
} from "../../../packages/markdown-core/src/ir-spans.js";
import { markdownToIR, type MarkdownIR } from "../../../packages/markdown-core/src/ir.js";
import { stripHtmlFromMarkdown } from "../../../packages/markdown-core/src/strip-html.js";
@ -17,23 +21,15 @@ type StripMarkdownOptions = {
stripHtml?: boolean;
};
type PlainTextInsertion = {
position: number;
text: string;
};
function collectLinkInsertions(
ir: MarkdownIR,
options: StripMarkdownOptions,
): PlainTextInsertion[] {
const insertions: PlainTextInsertion[] = [];
function collectLinkInsertions(ir: MarkdownIR, options: StripMarkdownOptions): MarkdownTextEdit[] {
const insertions: MarkdownTextEdit[] = [];
if ((options.linkStyle ?? "label-and-url") === "label-and-url") {
for (const link of ir.links) {
const href = link.href.trim();
const label = ir.text.slice(link.start, link.end).trim();
const comparableHref = href.startsWith("mailto:") ? href.slice("mailto:".length) : href;
if (href && label && label !== href && label !== comparableHref) {
insertions.push({ position: link.end, text: ` (${href})` });
insertions.push({ start: link.end, end: link.end, text: ` (${href})` });
}
}
}
@ -43,7 +39,7 @@ function collectLinkInsertions(
function collectAssistantTranscriptRoleInsertions(
source: string | MarkdownIR,
options: StripMarkdownOptions,
): PlainTextInsertion[] {
): MarkdownTextEdit[] {
if (options.assistantTranscriptRoleHeaders !== true) {
return [];
}
@ -57,23 +53,7 @@ function collectAssistantTranscriptRoleInsertions(
: (source.annotations ?? []).filter(
(annotation) => annotation.type === "assistant_transcript_role",
);
return spans.map((span) => ({ position: span.start, text: prefix }));
}
function applyPlainTextInsertions(text: string, insertions: PlainTextInsertion[]): string {
if (insertions.length === 0) {
return text;
}
const sorted = insertions.toSorted((a, b) => a.position - b.position);
let output = "";
let cursor = 0;
for (const insertion of sorted) {
const position = Math.max(cursor, Math.min(insertion.position, text.length));
output += text.slice(cursor, position);
output += insertion.text;
cursor = position;
}
return output + text.slice(cursor);
return spans.map((span) => ({ start: span.start, end: span.start, text: prefix }));
}
function cleanSpeechText(text: string): string {
@ -125,13 +105,13 @@ export function stripMarkdown(
? { ...profile, constructs: { ...profile.constructs, linkLabel: "strip" as const } }
: profile;
const projectedIr = effectiveProfile ? applyConstructFallbacks(ir, effectiveProfile) : ir;
const plainText = applyPlainTextInsertions(projectedIr.text, [
const plainText = applyMarkdownTextEdits(projectedIr.text, [
...collectLinkInsertions(projectedIr, options),
...collectAssistantTranscriptRoleInsertions(projectedIr, options),
]).trim();
const projected = applyPlainTextInsertions(
]).text.trim();
const projected = applyMarkdownTextEdits(
plainText,
collectAssistantTranscriptRoleInsertions(plainText, options),
).trim();
).text.trim();
return options.mode === "speech" ? cleanSpeechText(projected) : projected;
}