github/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-05-07 15:01:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

merge upstream/main

Browse Source
This commit is contained in:
Peter Steinberger
2026-01-06 23:09:01 +01:00
parent 86b56b2308 7aa7fa79d0
commit fec7f37271
365 changed files with 19758 additions and 4684 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -7,6 +7,7 @@ import {
queueEmbeddedPiMessage,
runEmbeddedPiAgent,
} from "../../agents/pi-embedded.js";
import { hasNonzeroUsage } from "../../agents/usage.js";
import {
loadSessionStore,
resolveSessionTranscriptPath,
@@ -32,6 +33,21 @@ import { extractReplyToTag } from "./reply-tags.js";
import { incrementCompactionCount } from "./session-updates.js";
import type { TypingController } from "./typing.js";
const BUN_FETCH_SOCKET_ERROR_RE = /socket connection was closed unexpectedly/i;
const isBunFetchSocketError = (message?: string) =>
Boolean(message && BUN_FETCH_SOCKET_ERROR_RE.test(message));
const formatBunFetchSocketError = (message: string) => {
const trimmed = message.trim();
return [
"⚠️ LLM connection failed. This could be due to server issues, network problems, or context length exceeded (e.g., with local LLMs like LM Studio). Original error:",
"```",
trimmed || "Unknown error",
"```",
].join("\n");
};
export async function runReplyAgent(params: {
commandBody: string;
followupRun: FollowupRun;
@@ -107,6 +123,7 @@ export async function runReplyAgent(params: {
const streamedPayloadKeys = new Set<string>();
const pendingStreamedPayloadKeys = new Set<string>();
const pendingBlockTasks = new Set<Promise<void>>();
const pendingToolTasks = new Set<Promise<void>>();
let didStreamBlockReply = false;
const buildPayloadKey = (payload: ReplyPayload) => {
const text = payload.text?.trim() ?? "";
@@ -188,9 +205,11 @@ export async function runReplyAgent(params: {
runEmbeddedPiAgent({
sessionId: followupRun.run.sessionId,
sessionKey,
surface: sessionCtx.Surface?.trim().toLowerCase() || undefined,
messageProvider:
sessionCtx.Provider?.trim().toLowerCase() || undefined,
sessionFile: followupRun.run.sessionFile,
workspaceDir: followupRun.run.workspaceDir,
agentDir: followupRun.run.agentDir,
config: followupRun.run.config,
skillsSnapshot: followupRun.run.skillsSnapshot,
prompt: commandBody,
@@ -239,7 +258,8 @@ export async function runReplyAgent(params: {
: undefined,
onAgentEvent: (evt) => {
if (evt.stream !== "compaction") return;
const phase = String(evt.data.phase ?? "");
const phase =
typeof evt.data.phase === "string" ? evt.data.phase : "";
const willRetry = Boolean(evt.data.willRetry);
if (phase === "end" && !willRetry) {
autoCompactionCompleted = true;
@@ -310,33 +330,45 @@ export async function runReplyAgent(params: {
: undefined,
shouldEmitToolResult,
onToolResult: opts?.onToolResult
? async (payload) => {
let text = payload.text;
if (!isHeartbeat && text?.includes("HEARTBEAT_OK")) {
const stripped = stripHeartbeatToken(text, {
mode: "message",
? (payload) => {
// `subscribeEmbeddedPiSession` may invoke tool callbacks without awaiting them.
// If a tool callback starts typing after the run finalized, we can end up with
// a typing loop that never sees a matching markRunComplete(). Track and drain.
const task = (async () => {
let text = payload.text;
if (!isHeartbeat && text?.includes("HEARTBEAT_OK")) {
const stripped = stripHeartbeatToken(text, {
mode: "message",
});
if (stripped.didStrip && !didLogHeartbeatStrip) {
didLogHeartbeatStrip = true;
logVerbose(
"Stripped stray HEARTBEAT_OK token from reply",
);
}
if (
stripped.shouldSkip &&
(payload.mediaUrls?.length ?? 0) === 0
) {
return;
}
text = stripped.text;
}
if (!isHeartbeat) {
await typing.startTypingOnText(text);
}
await opts.onToolResult?.({
text,
mediaUrls: payload.mediaUrls,
});
if (stripped.didStrip && !didLogHeartbeatStrip) {
didLogHeartbeatStrip = true;
logVerbose(
"Stripped stray HEARTBEAT_OK token from reply",
);
}
if (
stripped.shouldSkip &&
(payload.mediaUrls?.length ?? 0) === 0
) {
return;
}
text = stripped.text;
}
if (!isHeartbeat) {
await typing.startTypingOnText(text);
}
await opts.onToolResult?.({
text,
mediaUrls: payload.mediaUrls,
});
})()
.catch((err) => {
logVerbose(`tool result delivery failed: ${String(err)}`);
})
.finally(() => {
pendingToolTasks.delete(task);
});
pendingToolTasks.add(task);
}
: undefined,
}),
@@ -408,16 +440,28 @@ export async function runReplyAgent(params: {
}
const payloadArray = runResult.payloads ?? [];
if (payloadArray.length === 0) return finalizeWithFollowup(undefined);
if (pendingBlockTasks.size > 0) {
await Promise.allSettled(pendingBlockTasks);
}
if (pendingToolTasks.size > 0) {
await Promise.allSettled(pendingToolTasks);
}
// Drain any late tool/block deliveries before deciding there's "nothing to send".
// Otherwise, a late typing trigger (e.g. from a tool callback) can outlive the run and
// keep the typing indicator stuck.
if (payloadArray.length === 0) return finalizeWithFollowup(undefined);
const sanitizedPayloads = isHeartbeat
? payloadArray
: payloadArray.flatMap((payload) => {
const text = payload.text;
if (!text || !text.includes("HEARTBEAT_OK")) return [payload];
let text = payload.text;
if (payload.isError && text && isBunFetchSocketError(text)) {
text = formatBunFetchSocketError(text);
}
if (!text || !text.includes("HEARTBEAT_OK"))
return [{ ...payload, text }];
const stripped = stripHeartbeatToken(text, { mode: "message" });
if (stripped.didStrip && !didLogHeartbeatStrip) {
didLogHeartbeatStrip = true;
@@ -485,7 +529,7 @@ export async function runReplyAgent(params: {
sessionEntry?.contextTokens ??
DEFAULT_CONTEXT_TOKENS;
if (usage) {
if (hasNonzeroUsage(usage)) {
const entry = sessionEntry ?? sessionStore[sessionKey];
if (entry) {
const input = usage.input ?? 0;
@@ -552,6 +596,6 @@ export async function runReplyAgent(params: {
finalPayloads.length === 1 ? finalPayloads[0] : finalPayloads,
);
} finally {
typing.cleanup();
typing.markRunComplete();
}
}