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

style: apply prettier formatting

Browse Source
This commit is contained in:
Lauren Rosenberg
2026-01-18 18:21:11 +00:00
parent 576485b0c9
commit 0e94f0c018
1 changed files with 90 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

122
src/auto-reply/reply/agent-runner-execution.ts
View File

@@ -18,7 +18,10 @@ import {
updateSessionStore, updateSessionStore,
} from "../../config/sessions.js"; } from "../../config/sessions.js";
import { logVerbose } from "../../globals.js"; import { logVerbose } from "../../globals.js";
import { emitAgentEvent, registerAgentRunContext } from "../../infra/agent-events.js"; import {
emitAgentEvent,
registerAgentRunContext,
} from "../../infra/agent-events.js";
import { defaultRuntime } from "../../runtime.js"; import { defaultRuntime } from "../../runtime.js";
import { import {
isMarkdownCapableMessageChannel, isMarkdownCapableMessageChannel,
@@ -29,11 +32,20 @@ import type { TemplateContext } from "../templating.js";
import type { VerboseLevel } from "../thinking.js"; import type { VerboseLevel } from "../thinking.js";
import { isSilentReplyText, SILENT_REPLY_TOKEN } from "../tokens.js"; import { isSilentReplyText, SILENT_REPLY_TOKEN } from "../tokens.js";
import type { GetReplyOptions, ReplyPayload } from "../types.js"; import type { GetReplyOptions, ReplyPayload } from "../types.js";
import { buildThreadingToolContext, resolveEnforceFinalTag } from "./agent-runner-utils.js"; import {
import { createBlockReplyPayloadKey, type BlockReplyPipeline } from "./block-reply-pipeline.js"; buildThreadingToolContext,
resolveEnforceFinalTag,
} from "./agent-runner-utils.js";
import {
createBlockReplyPayloadKey,
type BlockReplyPipeline,
} from "./block-reply-pipeline.js";
import type { FollowupRun } from "./queue.js"; import type { FollowupRun } from "./queue.js";
import { parseReplyDirectives } from "./reply-directives.js"; import { parseReplyDirectives } from "./reply-directives.js";
import { applyReplyTagsToPayload, isRenderablePayload } from "./reply-payloads.js"; import {
applyReplyTagsToPayload,
isRenderablePayload,
} from "./reply-payloads.js";
import type { TypingSignaler } from "./typing-mode.js"; import type { TypingSignaler } from "./typing-mode.js";
export type AgentRunLoopResult = export type AgentRunLoopResult =
@@ -96,9 +108,12 @@ export async function runAgentTurnWithFallback(params: {
while (true) { while (true) {
try { try {
const allowPartialStream = !( const allowPartialStream = !(
params.followupRun.run.reasoningLevel === "stream" && params.opts?.onReasoningStream params.followupRun.run.reasoningLevel === "stream" &&
params.opts?.onReasoningStream
); );
const normalizeStreamingText = (payload: ReplyPayload): { text?: string; skip: boolean } => { const normalizeStreamingText = (
payload: ReplyPayload,
): { text?: string; skip: boolean } => {
if (!allowPartialStream) return { skip: true }; if (!allowPartialStream) return { skip: true };
let text = payload.text; let text = payload.text;
if (!params.isHeartbeat && text?.includes("HEARTBEAT_OK")) { if (!params.isHeartbeat && text?.includes("HEARTBEAT_OK")) {
@@ -122,7 +137,9 @@ export async function runAgentTurnWithFallback(params: {
if (!sanitized.trim()) return { skip: true }; if (!sanitized.trim()) return { skip: true };
return { text: sanitized, skip: false }; return { text: sanitized, skip: false };
}; };
const handlePartialForTyping = async (payload: ReplyPayload): Promise<string | undefined> => { const handlePartialForTyping = async (
payload: ReplyPayload,
): Promise<string | undefined> => {
const { text, skip } = normalizeStreamingText(payload); const { text, skip } = normalizeStreamingText(payload);
if (skip || !text) return undefined; if (skip || !text) return undefined;
await params.typingSignals.signalTextDelta(text); await params.typingSignals.signalTextDelta(text);
@@ -157,7 +174,10 @@ export async function runAgentTurnWithFallback(params: {
startedAt, startedAt,
}, },
}); });
const cliSessionId = getCliSessionId(params.getActiveSessionEntry(), provider); const cliSessionId = getCliSessionId(
params.getActiveSessionEntry(),
provider,
);
return runCliAgent({ return runCliAgent({
sessionId: params.followupRun.run.sessionId, sessionId: params.followupRun.run.sessionId,
sessionKey: params.sessionKey, sessionKey: params.sessionKey,
@@ -207,7 +227,8 @@ export async function runAgentTurnWithFallback(params: {
return runEmbeddedPiAgent({ return runEmbeddedPiAgent({
sessionId: params.followupRun.run.sessionId, sessionId: params.followupRun.run.sessionId,
sessionKey: params.sessionKey, sessionKey: params.sessionKey,
messageProvider: params.sessionCtx.Provider?.trim().toLowerCase() || undefined, messageProvider:
params.sessionCtx.Provider?.trim().toLowerCase() || undefined,
agentAccountId: params.sessionCtx.AccountId, agentAccountId: params.sessionCtx.AccountId,
// Provider threading context for tool auto-injection // Provider threading context for tool auto-injection
...buildThreadingToolContext({ ...buildThreadingToolContext({
@@ -223,7 +244,10 @@ export async function runAgentTurnWithFallback(params: {
prompt: params.commandBody, prompt: params.commandBody,
extraSystemPrompt: params.followupRun.run.extraSystemPrompt, extraSystemPrompt: params.followupRun.run.extraSystemPrompt,
ownerNumbers: params.followupRun.run.ownerNumbers, ownerNumbers: params.followupRun.run.ownerNumbers,
enforceFinalTag: resolveEnforceFinalTag(params.followupRun.run, provider), enforceFinalTag: resolveEnforceFinalTag(
params.followupRun.run,
provider,
),
provider, provider,
model, model,
authProfileId, authProfileId,
@@ -240,7 +264,9 @@ export async function runAgentTurnWithFallback(params: {
params.sessionCtx.Provider, params.sessionCtx.Provider,
); );
if (!channel) return "markdown"; if (!channel) return "markdown";
return isMarkdownCapableMessageChannel(channel) ? "markdown" : "plain"; return isMarkdownCapableMessageChannel(channel)
? "markdown"
: "plain";
})(), })(),
bashElevated: params.followupRun.run.bashElevated, bashElevated: params.followupRun.run.bashElevated,
timeoutMs: params.followupRun.run.timeoutMs, timeoutMs: params.followupRun.run.timeoutMs,
@@ -250,7 +276,11 @@ export async function runAgentTurnWithFallback(params: {
onPartialReply: allowPartialStream onPartialReply: allowPartialStream
? async (payload) => { ? async (payload) => {
const textForTyping = await handlePartialForTyping(payload); const textForTyping = await handlePartialForTyping(payload);
if (!params.opts?.onPartialReply || textForTyping === undefined) return; if (
!params.opts?.onPartialReply ||
textForTyping === undefined
)
return;
await params.opts.onPartialReply({ await params.opts.onPartialReply({
text: textForTyping, text: textForTyping,
mediaUrls: payload.mediaUrls, mediaUrls: payload.mediaUrls,
@@ -261,7 +291,8 @@ export async function runAgentTurnWithFallback(params: {
await params.typingSignals.signalMessageStart(); await params.typingSignals.signalMessageStart();
}, },
onReasoningStream: onReasoningStream:
params.typingSignals.shouldStartOnReasoning || params.opts?.onReasoningStream params.typingSignals.shouldStartOnReasoning ||
params.opts?.onReasoningStream
? async (payload) => { ? async (payload) => {
await params.typingSignals.signalReasoningDelta(); await params.typingSignals.signalReasoningDelta();
await params.opts?.onReasoningStream?.({ await params.opts?.onReasoningStream?.({
@@ -274,14 +305,16 @@ export async function runAgentTurnWithFallback(params: {
// Trigger typing when tools start executing. // Trigger typing when tools start executing.
// Must await to ensure typing indicator starts before tool summaries are emitted. // Must await to ensure typing indicator starts before tool summaries are emitted.
if (evt.stream === "tool") { if (evt.stream === "tool") {
const phase = typeof evt.data.phase === "string" ? evt.data.phase : ""; const phase =
typeof evt.data.phase === "string" ? evt.data.phase : "";
if (phase === "start" || phase === "update") { if (phase === "start" || phase === "update") {
await params.typingSignals.signalToolStart(); await params.typingSignals.signalToolStart();
} }
} }
// Track auto-compaction completion // Track auto-compaction completion
if (evt.stream === "compaction") { if (evt.stream === "compaction") {
const phase = typeof evt.data.phase === "string" ? evt.data.phase : ""; const phase =
typeof evt.data.phase === "string" ? evt.data.phase : "";
const willRetry = Boolean(evt.data.willRetry); const willRetry = Boolean(evt.data.willRetry);
if (phase === "end" && !willRetry) { if (phase === "end" && !willRetry) {
autoCompactionCompleted = true; autoCompactionCompleted = true;
@@ -305,14 +338,22 @@ export async function runAgentTurnWithFallback(params: {
params.sessionCtx.MessageSid, params.sessionCtx.MessageSid,
); );
// Let through payloads with audioAsVoice flag even if empty (need to track it) // Let through payloads with audioAsVoice flag even if empty (need to track it)
if (!isRenderablePayload(taggedPayload) && !payload.audioAsVoice) return; if (
const parsed = parseReplyDirectives(taggedPayload.text ?? "", { !isRenderablePayload(taggedPayload) &&
currentMessageId: params.sessionCtx.MessageSid, !payload.audioAsVoice
silentToken: SILENT_REPLY_TOKEN, )
}); return;
const parsed = parseReplyDirectives(
taggedPayload.text ?? "",
{
currentMessageId: params.sessionCtx.MessageSid,
silentToken: SILENT_REPLY_TOKEN,
},
);
const cleaned = parsed.text || undefined; const cleaned = parsed.text || undefined;
const hasRenderableMedia = const hasRenderableMedia =
Boolean(taggedPayload.mediaUrl) || (taggedPayload.mediaUrls?.length ?? 0) > 0; Boolean(taggedPayload.mediaUrl) ||
(taggedPayload.mediaUrls?.length ?? 0) > 0;
// Skip empty payloads unless they have audioAsVoice flag (need to track it) // Skip empty payloads unless they have audioAsVoice flag (need to track it)
if ( if (
!cleaned && !cleaned &&
@@ -326,25 +367,35 @@ export async function runAgentTurnWithFallback(params: {
const blockPayload: ReplyPayload = params.applyReplyToMode({ const blockPayload: ReplyPayload = params.applyReplyToMode({
...taggedPayload, ...taggedPayload,
text: cleaned, text: cleaned,
audioAsVoice: Boolean(parsed.audioAsVoice || payload.audioAsVoice), audioAsVoice: Boolean(
parsed.audioAsVoice || payload.audioAsVoice,
),
replyToId: taggedPayload.replyToId ?? parsed.replyToId, replyToId: taggedPayload.replyToId ?? parsed.replyToId,
replyToTag: taggedPayload.replyToTag || parsed.replyToTag, replyToTag: taggedPayload.replyToTag || parsed.replyToTag,
replyToCurrent: taggedPayload.replyToCurrent || parsed.replyToCurrent, replyToCurrent:
taggedPayload.replyToCurrent || parsed.replyToCurrent,
}); });
void params.typingSignals void params.typingSignals
.signalTextDelta(cleaned ?? taggedPayload.text) .signalTextDelta(cleaned ?? taggedPayload.text)
.catch((err) => { .catch((err) => {
logVerbose(`block reply typing signal failed: ${String(err)}`); logVerbose(
`block reply typing signal failed: ${String(err)}`,
);
}); });
// Use pipeline if available (block streaming enabled), otherwise send directly // Use pipeline if available (block streaming enabled), otherwise send directly
if (params.blockStreamingEnabled && params.blockReplyPipeline) { if (
params.blockStreamingEnabled &&
params.blockReplyPipeline
) {
params.blockReplyPipeline.enqueue(blockPayload); params.blockReplyPipeline.enqueue(blockPayload);
} else { } else {
// Send directly when flushing before tool execution (no streaming). // Send directly when flushing before tool execution (no streaming).
// Track sent key to avoid duplicate in final payloads. // Track sent key to avoid duplicate in final payloads.
directlySentBlockKeys.add(createBlockReplyPayloadKey(blockPayload)); directlySentBlockKeys.add(
createBlockReplyPayloadKey(blockPayload),
);
await params.opts?.onBlockReply?.(blockPayload); await params.opts?.onBlockReply?.(blockPayload);
} }
} }
@@ -405,7 +456,9 @@ export async function runAgentTurnWithFallback(params: {
}; };
} }
if (embeddedError?.kind === "role_ordering") { if (embeddedError?.kind === "role_ordering") {
const didReset = await params.resetSessionAfterRoleOrderingConflict(embeddedError.message); const didReset = await params.resetSessionAfterRoleOrderingConflict(
embeddedError.message,
);
if (didReset) { if (didReset) {
return { return {
kind: "final", kind: "final",
@@ -423,8 +476,10 @@ export async function runAgentTurnWithFallback(params: {
isContextOverflowError(message) || isContextOverflowError(message) ||
/context.*overflow|too large|context window/i.test(message); /context.*overflow|too large|context window/i.test(message);
const isCompactionFailure = isCompactionFailureError(message); const isCompactionFailure = isCompactionFailureError(message);
const isSessionCorruption = /function call turn comes immediately after/i.test(message); const isSessionCorruption =
const isRoleOrderingError = /incorrect role information|roles must alternate/i.test(message); /function call turn comes immediately after/i.test(message);
const isRoleOrderingError =
/incorrect role information|roles must alternate/i.test(message);
if ( if (
isCompactionFailure && isCompactionFailure &&
@@ -440,7 +495,8 @@ export async function runAgentTurnWithFallback(params: {
}; };
} }
if (isRoleOrderingError) { if (isRoleOrderingError) {
const didReset = await params.resetSessionAfterRoleOrderingConflict(message); const didReset =
await params.resetSessionAfterRoleOrderingConflict(message);
if (didReset) { if (didReset) {
return { return {
kind: "final", kind: "final",
@@ -467,7 +523,8 @@ export async function runAgentTurnWithFallback(params: {
try { try {
// Delete transcript file if it exists // Delete transcript file if it exists
if (corruptedSessionId) { if (corruptedSessionId) {
const transcriptPath = resolveSessionTranscriptPath(corruptedSessionId); const transcriptPath =
resolveSessionTranscriptPath(corruptedSessionId);
try { try {
fs.unlinkSync(transcriptPath); fs.unlinkSync(transcriptPath);
} catch { } catch {
@@ -517,6 +574,7 @@ export async function runAgentTurnWithFallback(params: {
fallbackModel, fallbackModel,
didLogHeartbeatStrip, didLogHeartbeatStrip,
autoCompactionCompleted, autoCompactionCompleted,
directlySentBlockKeys: directlySentBlockKeys.size > 0 ? directlySentBlockKeys : undefined, directlySentBlockKeys:
directlySentBlockKeys.size > 0 ? directlySentBlockKeys : undefined,
}; };
} }