let5see/clawdbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' into commands-list-clean

Browse Source
This commit is contained in:
Luke
2026-01-08 15:35:17 -05:00
committed by GitHub
parent 713eb898f6 01641b34ea
commit a574ae5415
45 changed files with 1587 additions and 387 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
src/auto-reply/chunk.test.ts
View File

@@ -67,6 +67,12 @@ describe("chunkText", () => {
const chunks = chunkText(text, 10);
expect(chunks).toEqual(["Supercalif", "ragilistic", "expialidoc", "ious"]);
});
it("keeps parenthetical phrases together", () => {
const text = "Heads up now (Though now I'm curious)ok";
const chunks = chunkText(text, 35);
expect(chunks).toEqual(["Heads up now", "(Though now I'm curious)ok"]);
});
});
describe("resolveTextChunkLimit", () => {
@@ -184,4 +190,29 @@ describe("chunkMarkdownText", () => {
expect(nonFenceLines.join("\n").trim()).not.toBe("");
}
});
it("keeps parenthetical phrases together", () => {
const text = "Heads up now (Though now I'm curious)ok";
const chunks = chunkMarkdownText(text, 35);
expect(chunks).toEqual(["Heads up now", "(Though now I'm curious)ok"]);
});
it("handles nested parentheses", () => {
const text = "Hello (outer (inner) end) world";
const chunks = chunkMarkdownText(text, 26);
expect(chunks).toEqual(["Hello (outer (inner) end)", "world"]);
});
it("hard-breaks when a parenthetical exceeds the limit", () => {
const text = `(${"a".repeat(80)})`;
const chunks = chunkMarkdownText(text, 20);
expect(chunks[0]?.length).toBe(20);
expect(chunks.join("")).toBe(text);
});
it("ignores unmatched closing parentheses", () => {
const text = "Hello) world (ok)";
const chunks = chunkMarkdownText(text, 12);
expect(chunks).toEqual(["Hello)", "world (ok)"]);
});
});

57
src/auto-reply/chunk.ts
View File

@@ -90,18 +90,27 @@ export function chunkText(text: string, limit: number): string[] {
while (remaining.length > limit) {
const window = remaining.slice(0, limit);
// 1) Prefer a newline break inside the window.
let breakIdx = window.lastIndexOf("\n");
// 1) Prefer a newline break inside the window (outside parentheses).
let lastNewline = -1;
let lastWhitespace = -1;
let depth = 0;
for (let i = 0; i < window.length; i++) {
const char = window[i];
if (char === "(") {
depth += 1;
continue;
}
if (char === ")" && depth > 0) {
depth -= 1;
continue;
}
if (depth !== 0) continue;
if (char === "\n") lastNewline = i;
else if (/\s/.test(char)) lastWhitespace = i;
}
// 2) Otherwise prefer the last whitespace (word boundary) inside the window.
if (breakIdx <= 0) {
for (let i = window.length - 1; i >= 0; i--) {
if (/\s/.test(window[i])) {
breakIdx = i;
break;
}
}
}
let breakIdx = lastNewline > 0 ? lastNewline : lastWhitespace;
// 3) Fallback: hard break exactly at the limit.
if (breakIdx <= 0) breakIdx = limit;
@@ -234,15 +243,27 @@ function pickSafeBreakIndex(
window: string,
spans: ReturnType<typeof parseFenceSpans>,
): number {
let newlineIdx = window.lastIndexOf("\n");
while (newlineIdx > 0) {
if (isSafeFenceBreak(spans, newlineIdx)) return newlineIdx;
newlineIdx = window.lastIndexOf("\n", newlineIdx - 1);
}
for (let i = window.length - 1; i > 0; i--) {
if (/\s/.test(window[i]) && isSafeFenceBreak(spans, i)) return i;
let lastNewline = -1;
let lastWhitespace = -1;
let depth = 0;
for (let i = 0; i < window.length; i++) {
if (!isSafeFenceBreak(spans, i)) continue;
const char = window[i];
if (char === "(") {
depth += 1;
continue;
}
if (char === ")" && depth > 0) {
depth -= 1;
continue;
}
if (depth !== 0) continue;
if (char === "\n") lastNewline = i;
else if (/\s/.test(char)) lastWhitespace = i;
}
if (lastNewline > 0) return lastNewline;
if (lastWhitespace > 0) return lastWhitespace;
return -1;
}

110
src/auto-reply/reply.block-streaming.test.ts
View File

@@ -103,6 +103,61 @@ describe("block streaming", () => {
});
});
it("preserves block reply ordering when typing start is slow", async () => {
await withTempHome(async (home) => {
let releaseTyping: (() => void) | undefined;
const typingGate = new Promise<void>((resolve) => {
releaseTyping = resolve;
});
const onReplyStart = vi.fn(() => typingGate);
const seen: string[] = [];
const onBlockReply = vi.fn(async (payload) => {
seen.push(payload.text ?? "");
});
vi.mocked(runEmbeddedPiAgent).mockImplementation(async (params) => {
void params.onBlockReply?.({ text: "first" });
void params.onBlockReply?.({ text: "second" });
return {
payloads: [{ text: "first" }, { text: "second" }],
meta: {
durationMs: 5,
agentMeta: { sessionId: "s", provider: "p", model: "m" },
},
};
});
const replyPromise = getReplyFromConfig(
{
Body: "ping",
From: "+1004",
To: "+2000",
MessageSid: "msg-125",
Provider: "telegram",
},
{
onReplyStart,
onBlockReply,
},
{
agent: {
model: "anthropic/claude-opus-4-5",
workspace: path.join(home, "clawd"),
},
telegram: { allowFrom: ["*"] },
session: { store: path.join(home, "sessions.json") },
},
);
await waitForCalls(() => onReplyStart.mock.calls.length, 1);
releaseTyping?.();
const res = await replyPromise;
expect(res).toBeUndefined();
expect(seen).toEqual(["first", "second"]);
});
});
it("drops final payloads when block replies streamed", async () => {
await withTempHome(async (home) => {
const onBlockReply = vi.fn().mockResolvedValue(undefined);
@@ -143,4 +198,59 @@ describe("block streaming", () => {
expect(onBlockReply).toHaveBeenCalledTimes(1);
});
});
it("falls back to final payloads when block reply send times out", async () => {
await withTempHome(async (home) => {
let sawAbort = false;
const onBlockReply = vi.fn((_, context) => {
return new Promise<void>((resolve) => {
context?.abortSignal?.addEventListener(
"abort",
() => {
sawAbort = true;
resolve();
},
{ once: true },
);
});
});
vi.mocked(runEmbeddedPiAgent).mockImplementation(async (params) => {
void params.onBlockReply?.({ text: "streamed" });
return {
payloads: [{ text: "final" }],
meta: {
durationMs: 5,
agentMeta: { sessionId: "s", provider: "p", model: "m" },
},
};
});
const replyPromise = getReplyFromConfig(
{
Body: "ping",
From: "+1004",
To: "+2000",
MessageSid: "msg-126",
Provider: "telegram",
},
{
onBlockReply,
blockReplyTimeoutMs: 10,
},
{
agent: {
model: "anthropic/claude-opus-4-5",
workspace: path.join(home, "clawd"),
},
telegram: { allowFrom: ["*"] },
session: { store: path.join(home, "sessions.json") },
},
);
const res = await replyPromise;
expect(res).toMatchObject({ text: "final" });
expect(sawAbort).toBe(true);
});
});
});

73
src/auto-reply/reply/agent-runner.ts
View File

@@ -47,6 +47,7 @@ import type { TypingController } from "./typing.js";
import { createTypingSignaler } from "./typing-mode.js";
const BUN_FETCH_SOCKET_ERROR_RE = /socket connection was closed unexpectedly/i;
const BLOCK_REPLY_SEND_TIMEOUT_MS = 15_000;
const isBunFetchSocketError = (message?: string) =>
Boolean(message && BUN_FETCH_SOCKET_ERROR_RE.test(message));
@@ -61,6 +62,23 @@ const formatBunFetchSocketError = (message: string) => {
].join("\n");
};
const withTimeout = async <T>(
promise: Promise<T>,
timeoutMs: number,
timeoutError: Error,
): Promise<T> => {
if (!timeoutMs || timeoutMs <= 0) return promise;
let timer: NodeJS.Timeout | undefined;
const timeoutPromise = new Promise<never>((_, reject) => {
timer = setTimeout(() => reject(timeoutError), timeoutMs);
});
try {
return await Promise.race([promise, timeoutPromise]);
} finally {
if (timer) clearTimeout(timer);
}
};
export async function runReplyAgent(params: {
commandBody: string;
followupRun: FollowupRun;
@@ -144,7 +162,12 @@ export async function runReplyAgent(params: {
const pendingStreamedPayloadKeys = new Set<string>();
const pendingBlockTasks = new Set<Promise<void>>();
const pendingToolTasks = new Set<Promise<void>>();
let blockReplyChain: Promise<void> = Promise.resolve();
let blockReplyAborted = false;
let didLogBlockReplyAbort = false;
let didStreamBlockReply = false;
const blockReplyTimeoutMs =
opts?.blockReplyTimeoutMs ?? BLOCK_REPLY_SEND_TIMEOUT_MS;
const buildPayloadKey = (payload: ReplyPayload) => {
const text = payload.text?.trim() ?? "";
const mediaList = payload.mediaUrls?.length
@@ -367,16 +390,49 @@ export async function runReplyAgent(params: {
) {
return;
}
if (blockReplyAborted) return;
pendingStreamedPayloadKeys.add(payloadKey);
const task = (async () => {
await typingSignals.signalTextDelta(taggedPayload.text);
await opts.onBlockReply?.(blockPayload);
})()
.then(() => {
void typingSignals
.signalTextDelta(taggedPayload.text)
.catch((err) => {
logVerbose(
`block reply typing signal failed: ${String(err)}`,
);
});
const timeoutError = new Error(
`block reply delivery timed out after ${blockReplyTimeoutMs}ms`,
);
const abortController = new AbortController();
blockReplyChain = blockReplyChain
.then(async () => {
if (blockReplyAborted) return false;
await withTimeout(
opts.onBlockReply?.(blockPayload, {
abortSignal: abortController.signal,
timeoutMs: blockReplyTimeoutMs,
}) ?? Promise.resolve(),
blockReplyTimeoutMs,
timeoutError,
);
return true;
})
.then((didSend) => {
if (!didSend) return;
streamedPayloadKeys.add(payloadKey);
didStreamBlockReply = true;
})
.catch((err) => {
if (err === timeoutError) {
abortController.abort();
blockReplyAborted = true;
if (!didLogBlockReplyAbort) {
didLogBlockReplyAbort = true;
logVerbose(
`block reply delivery timed out after ${blockReplyTimeoutMs}ms; skipping remaining block replies to preserve ordering`,
);
}
return;
}
logVerbose(
`block reply delivery failed: ${String(err)}`,
);
@@ -384,6 +440,7 @@ export async function runReplyAgent(params: {
.finally(() => {
pendingStreamedPayloadKeys.delete(payloadKey);
});
const task = blockReplyChain;
pendingBlockTasks.add(task);
void task.finally(() => pendingBlockTasks.delete(task));
}
@@ -546,10 +603,10 @@ export async function runReplyAgent(params: {
})
.filter(isRenderablePayload);
// Drop final payloads if block streaming is enabled and we already streamed
// block replies. Tool-sent duplicates are filtered below.
// Drop final payloads only when block streaming succeeded end-to-end.
// If streaming aborted (e.g., timeout), fall back to final payloads.
const shouldDropFinalPayloads =
blockStreamingEnabled && didStreamBlockReply;
blockStreamingEnabled && didStreamBlockReply && !blockReplyAborted;
const messagingToolSentTexts = runResult.messagingToolSentTexts ?? [];
const messagingToolSentTargets = runResult.messagingToolSentTargets ?? [];
const suppressMessagingToolReplies = shouldSuppressMessagingToolReplies({

2
src/auto-reply/reply/commands.ts
View File

@@ -456,7 +456,7 @@ export async function handleCommands(params: {
...cfg.agent,
model: {
...cfg.agent?.model,
primary: model,
primary: `${provider}/${model}`,
},
contextTokens,
thinkingDefault: cfg.agent?.thinkingDefault,

13
src/auto-reply/reply/dispatch-from-config.ts
View File

@@ -41,10 +41,14 @@ export async function dispatchReplyFromConfig(params: {
* Note: Only called when shouldRouteToOriginating is true, so
* originatingChannel and originatingTo are guaranteed to be defined.
*/
const sendPayloadAsync = async (payload: ReplyPayload): Promise<void> => {
const sendPayloadAsync = async (
payload: ReplyPayload,
abortSignal?: AbortSignal,
): Promise<void> => {
// TypeScript doesn't narrow these from the shouldRouteToOriginating check,
// but they're guaranteed non-null when this function is called.
if (!originatingChannel || !originatingTo) return;
if (abortSignal?.aborted) return;
const result = await routeReply({
payload,
channel: originatingChannel,
@@ -52,6 +56,7 @@ export async function dispatchReplyFromConfig(params: {
accountId: ctx.AccountId,
threadId: ctx.MessageThreadId,
cfg,
abortSignal,
});
if (!result.ok) {
logVerbose(
@@ -73,10 +78,10 @@ export async function dispatchReplyFromConfig(params: {
dispatcher.sendToolResult(payload);
}
},
onBlockReply: (payload: ReplyPayload) => {
onBlockReply: (payload: ReplyPayload, context) => {
if (shouldRouteToOriginating) {
// Fire-and-forget for streaming block replies when routing.
void sendPayloadAsync(payload);
// Await routed sends so upstream can enforce ordering/timeouts.
return sendPayloadAsync(payload, context?.abortSignal);
} else {
// Synchronous dispatch to preserve callback timing.
dispatcher.sendBlockReply(payload);

16
src/auto-reply/reply/route-reply.test.ts
View File

@@ -31,6 +31,22 @@ vi.mock("../../web/outbound.js", () => ({
const { routeReply } = await import("./route-reply.js");
describe("routeReply", () => {
it("skips sends when abort signal is already aborted", async () => {
mocks.sendMessageSlack.mockClear();
const controller = new AbortController();
controller.abort();
const res = await routeReply({
payload: { text: "hi" },
channel: "slack",
to: "channel:C123",
cfg: {} as never,
abortSignal: controller.signal,
});
expect(res.ok).toBe(false);
expect(res.error).toContain("aborted");
expect(mocks.sendMessageSlack).not.toHaveBeenCalled();
});
it("no-ops on empty payload", async () => {
mocks.sendMessageSlack.mockClear();
const res = await routeReply({

13
src/auto-reply/reply/route-reply.ts
View File

@@ -30,6 +30,8 @@ export type RouteReplyParams = {
threadId?: number;
/** Config for provider-specific settings. */
cfg: ClawdbotConfig;
/** Optional abort signal for cooperative cancellation. */
abortSignal?: AbortSignal;
};
export type RouteReplyResult = {
@@ -52,7 +54,7 @@ export type RouteReplyResult = {
export async function routeReply(
params: RouteReplyParams,
): Promise<RouteReplyResult> {
const { payload, channel, to, accountId, threadId } = params;
const { payload, channel, to, accountId, threadId, abortSignal } = params;
// Debug: `pnpm test src/auto-reply/reply/route-reply.test.ts`
const text = payload.text ?? "";
@@ -72,6 +74,9 @@ export async function routeReply(
text: string;
mediaUrl?: string;
}): Promise<RouteReplyResult> => {
if (abortSignal?.aborted) {
return { ok: false, error: "Reply routing aborted" };
}
const { text, mediaUrl } = params;
switch (channel) {
case "telegram": {
@@ -148,12 +153,18 @@ export async function routeReply(
};
try {
if (abortSignal?.aborted) {
return { ok: false, error: "Reply routing aborted" };
}
if (mediaUrls.length === 0) {
return await sendOne({ text });
}
let last: RouteReplyResult | undefined;
for (let i = 0; i < mediaUrls.length; i++) {
if (abortSignal?.aborted) {
return { ok: false, error: "Reply routing aborted" };
}
const mediaUrl = mediaUrls[i];
const caption = i === 0 ? text : "";
last = await sendOne({ text: caption, mediaUrl });

12
src/auto-reply/status.test.ts
View File

@@ -102,6 +102,18 @@ describe("buildStatusMessage", () => {
expect(text).toContain("🧠 Model: openai/gpt-4.1-mini");
});
it("keeps provider prefix from configured model", () => {
const text = buildStatusMessage({
agent: {
model: "google-antigravity/claude-sonnet-4-5",
},
sessionScope: "per-sender",
queue: { mode: "collect", depth: 0 },
});
expect(text).toContain("🧠 Model: google-antigravity/claude-sonnet-4-5");
});
it("handles missing agent config gracefully", () => {
const text = buildStatusMessage({
agent: {},

12
src/auto-reply/types.ts
View File

@@ -1,14 +1,24 @@
import type { TypingController } from "./reply/typing.js";
export type BlockReplyContext = {
abortSignal?: AbortSignal;
timeoutMs?: number;
};
export type GetReplyOptions = {
onReplyStart?: () => Promise<void> | void;
onTypingController?: (typing: TypingController) => void;
isHeartbeat?: boolean;
onPartialReply?: (payload: ReplyPayload) => Promise<void> | void;
onReasoningStream?: (payload: ReplyPayload) => Promise<void> | void;
onBlockReply?: (payload: ReplyPayload) => Promise<void> | void;
onBlockReply?: (
payload: ReplyPayload,
context?: BlockReplyContext,
) => Promise<void> | void;
onToolResult?: (payload: ReplyPayload) => Promise<void> | void;
disableBlockStreaming?: boolean;
/** Timeout for block reply delivery (ms). */
blockReplyTimeoutMs?: number;
/** If provided, only load these skills for this session (empty = no skills). */
skillFilter?: string[];
};