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

🤖 codex: fix block reply ordering (#503)

Browse Source 
What: serialize block reply sends, make typing non-blocking, add timeout fallback + abort-aware routing, and add regression tests.
Why: prevent out-of-order streamed blocks while keeping final fallback on timeouts.
Tests: ./node_modules/.bin/vitest run src/auto-reply/reply.block-streaming.test.ts src/auto-reply/reply/route-reply.test.ts
Tests: corepack pnpm lint && corepack pnpm build (pass). corepack pnpm test (ran locally; failure observed during run).

Co-authored-by: Josh Palmer <joshp123@users.noreply.github.com>
This commit is contained in:
Josh Palmer
2026-01-08 19:30:24 +01:00
committed by GitHub
parent 7450aed663
commit cc94db458c
7 changed files with 224 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGELOG.md
View File

@@ -2,6 +2,7 @@
## Unreleased
- Auto-reply: preserve block reply ordering with timeout fallback for streaming. (#503) — thanks @joshp123
- WhatsApp: group `/model list` output by provider for scannability. (#456) - thanks @mcinteerj
- Hooks: allow per-hook model overrides for webhook/Gmail runs (e.g. GPT 5 Mini).
- Control UI: logs tab opens at the newest entries (bottom).

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({

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/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[];
};