diff --git a/src/agents/pi-embedded-helpers.ts b/src/agents/pi-embedded-helpers.ts index 7b24dc732..4d7539937 100644 --- a/src/agents/pi-embedded-helpers.ts +++ b/src/agents/pi-embedded-helpers.ts @@ -183,7 +183,11 @@ export async function sanitizeSessionMessagesImages( const GOOGLE_TURN_ORDER_BOOTSTRAP_TEXT = "(session bootstrap)"; export function isGoogleModelApi(api?: string | null): boolean { - return api === "google-gemini-cli" || api === "google-generative-ai"; + return ( + api === "google-gemini-cli" || + api === "google-generative-ai" || + api === "google-antigravity" + ); } export function sanitizeGoogleTurnOrdering( diff --git a/src/agents/pi-embedded-runner.ts b/src/agents/pi-embedded-runner.ts index 49ee41a4b..d985b2595 100644 --- a/src/agents/pi-embedded-runner.ts +++ b/src/agents/pi-embedded-runner.ts @@ -43,6 +43,7 @@ import { enqueueCommandInLane, } from "../process/command-queue.js"; import { normalizeMessageProvider } from "../utils/message-provider.js"; +import { isReasoningTagProvider } from "../utils/provider-utils.js"; import { resolveUserPath } from "../utils.js"; import { resolveClawdbotAgentDir } from "./agent-paths.js"; import { resolveSessionAgentIds } from "./agent-scope.js"; @@ -1141,7 +1142,7 @@ export async function compactEmbeddedPiSession(params: { sandbox, params.bashElevated, ); - const reasoningTagHint = provider === "ollama"; + const reasoningTagHint = isReasoningTagProvider(provider); const userTimezone = resolveUserTimezone( params.config?.agents?.defaults?.userTimezone, ); @@ -1547,7 +1548,7 @@ export async function runEmbeddedPiAgent(params: { sandbox, params.bashElevated, ); - const reasoningTagHint = provider === "ollama"; + const reasoningTagHint = isReasoningTagProvider(provider); const userTimezone = resolveUserTimezone( params.config?.agents?.defaults?.userTimezone, ); diff --git a/src/agents/pi-embedded-subscribe.ts b/src/agents/pi-embedded-subscribe.ts index 9b9b04f68..a6dcabe06 100644 --- a/src/agents/pi-embedded-subscribe.ts +++ b/src/agents/pi-embedded-subscribe.ts @@ -41,6 +41,7 @@ const THINKING_OPEN_RE = /<\s*(?:think(?:ing)?|thought|antthinking)\s*>/i; const THINKING_CLOSE_RE = /<\s*\/\s*(?:think(?:ing)?|thought|antthinking)\s*>/i; const THINKING_TAG_SCAN_RE = /<\s*(\/?)\s*(?:think(?:ing)?|thought|antthinking)\s*>/gi; +const FINAL_TAG_SCAN_RE = /<\s*(\/?)\s*final\s*>/gi; const TOOL_RESULT_MAX_CHARS = 8000; const log = createSubsystemLogger("agent/embedded"); const RAW_STREAM_ENABLED = process.env.CLAWDBOT_RAW_STREAM === "1"; @@ -111,37 +112,6 @@ function isToolResultError(result: unknown): boolean { return normalized === "error" || normalized === "timeout"; } -function stripThinkingSegments(text: string): string { - if (!text || !THINKING_TAG_RE.test(text)) return text; - THINKING_TAG_RE.lastIndex = 0; - let result = ""; - let lastIndex = 0; - let inThinking = false; - for (const match of text.matchAll(THINKING_TAG_RE)) { - const idx = match.index ?? 0; - if (!inThinking) { - result += text.slice(lastIndex, idx); - } - const tag = match[0].toLowerCase(); - inThinking = !tag.includes("/"); - lastIndex = idx + match[0].length; - } - if (!inThinking) { - result += text.slice(lastIndex); - } - return result; -} - -function stripUnpairedThinkingTags(text: string): string { - if (!text) return text; - const hasOpen = THINKING_OPEN_RE.test(text); - const hasClose = THINKING_CLOSE_RE.test(text); - if (hasOpen && hasClose) return text; - if (!hasOpen) return text.replace(THINKING_CLOSE_RE, ""); - if (!hasClose) return text.replace(THINKING_OPEN_RE, ""); - return text; -} - function extractMessagingToolSend( toolName: string, args: Record, @@ -212,6 +182,11 @@ export function subscribeEmbeddedPiSession(params: { }) => void; enforceFinalTag?: boolean; }) { + if (params.enforceFinalTag) { + log.debug("subscribeEmbeddedPiSession: enforceFinalTag is ENABLED"); + } else { + log.debug("subscribeEmbeddedPiSession: enforceFinalTag is DISABLED"); + } const assistantTexts: string[] = []; const toolMetas: Array<{ toolName?: string; meta?: string }> = []; const toolMetaById = new Map(); @@ -226,7 +201,8 @@ export function subscribeEmbeddedPiSession(params: { let deltaBuffer = ""; let blockBuffer = ""; // Track if a streamed chunk opened a block (stateful across chunks). - let blockThinkingActive = false; + const blockState = { thinking: false, final: false }; + const streamState = { thinking: false, final: false }; let lastStreamedAssistant: string | undefined; let lastStreamedReasoning: string | undefined; let lastBlockReplyText: string | undefined; @@ -242,7 +218,10 @@ export function subscribeEmbeddedPiSession(params: { deltaBuffer = ""; blockBuffer = ""; blockChunker?.reset(); - blockThinkingActive = false; + blockState.thinking = false; + blockState.final = false; + streamState.thinking = false; + streamState.final = false; lastStreamedAssistant = undefined; lastBlockReplyText = undefined; lastStreamedReasoning = undefined; @@ -337,27 +316,6 @@ export function subscribeEmbeddedPiSession(params: { compactionRetryPromise = null; } }; - const FINAL_START_RE = /<\s*final\s*>/i; - const FINAL_END_RE = /<\s*\/\s*final\s*>/i; - // Local providers sometimes emit malformed tags; normalize before filtering. - const sanitizeFinalText = (text: string): string => { - if (!text) return text; - const hasStart = FINAL_START_RE.test(text); - const hasEnd = FINAL_END_RE.test(text); - if (hasStart && !hasEnd) return text.replace(FINAL_START_RE, ""); - if (!hasStart && hasEnd) return text.replace(FINAL_END_RE, ""); - return text; - }; - const extractFinalText = (text: string): string | undefined => { - const cleaned = sanitizeFinalText(text); - const startMatch = FINAL_START_RE.exec(cleaned); - if (!startMatch) return undefined; - const startIndex = startMatch.index + startMatch[0].length; - const afterStart = cleaned.slice(startIndex); - const endMatch = FINAL_END_RE.exec(afterStart); - const endIndex = endMatch ? endMatch.index : afterStart.length; - return afterStart.slice(0, endIndex); - }; const blockChunking = params.blockReplyChunking; const blockChunker = blockChunking @@ -385,34 +343,91 @@ export function subscribeEmbeddedPiSession(params: { } }; - const stripBlockThinkingSegments = (text: string): string => { + const stripBlockTags = ( + text: string, + state: { thinking: boolean; final: boolean }, + ): string => { if (!text) return text; - if (!blockThinkingActive && !THINKING_TAG_SCAN_RE.test(text)) return text; + + // 1. Handle blocks (stateful, strip content inside) + let processed = ""; THINKING_TAG_SCAN_RE.lastIndex = 0; - let result = ""; let lastIndex = 0; - let inThinking = blockThinkingActive; + let inThinking = state.thinking; for (const match of text.matchAll(THINKING_TAG_SCAN_RE)) { const idx = match.index ?? 0; if (!inThinking) { - result += text.slice(lastIndex, idx); + processed += text.slice(lastIndex, idx); } const isClose = match[1] === "/"; inThinking = !isClose; lastIndex = idx + match[0].length; } if (!inThinking) { - result += text.slice(lastIndex); + processed += text.slice(lastIndex); } - blockThinkingActive = inThinking; + state.thinking = inThinking; + + // 2. Handle blocks (stateful, strip content OUTSIDE) + // If enforcement is disabled, just return processed text as-is. + if (!params.enforceFinalTag) return processed; + + // If enforcement is enabled, only return text that appeared inside a block. + let result = ""; + FINAL_TAG_SCAN_RE.lastIndex = 0; + let lastFinalIndex = 0; + let inFinal = state.final; + let everInFinal = state.final; + + for (const match of processed.matchAll(FINAL_TAG_SCAN_RE)) { + const idx = match.index ?? 0; + const isClose = match[1] === "/"; + + if (!inFinal && !isClose) { + // Found start tag. + inFinal = true; + everInFinal = true; + lastFinalIndex = idx + match[0].length; + } else if (inFinal && isClose) { + // Found end tag. + result += processed.slice(lastFinalIndex, idx); + inFinal = false; + lastFinalIndex = idx + match[0].length; + } + } + + if (inFinal) { + result += processed.slice(lastFinalIndex); + } + state.final = inFinal; + + // Log the result of the stripping for debugging purposes + if (params.enforceFinalTag && (everInFinal || processed.length > 0)) { + log.debug(JSON.stringify({ + raw: processed.slice(0, 100), + stripped: result.slice(0, 100), + inFinal, + everInFinal, + rawLen: processed.length, + strippedLen: result.length, + tag: "DEBUG_STRIP" + })); + } + + // Fallback: if we are at the end of the process and never saw a final tag, + // but we have processed text, use the processed text. + // NOTE: This fallback only triggers if we explicitly pass a state that we can check. + if (!everInFinal && processed.trim().length > 0) { + return processed; + } + return result; }; const emitBlockChunk = (text: string) => { if (suppressBlockChunks) return; - // Strip blocks across chunk boundaries to avoid leaking reasoning. - const strippedText = stripBlockThinkingSegments(text); - const chunk = strippedText.trimEnd(); + // Strip and blocks across chunk boundaries to avoid leaking reasoning. + const chunk = stripBlockTags(text, blockState).trimEnd(); if (!chunk) return; if (chunk === lastBlockReplyText) return; @@ -754,12 +769,10 @@ export function subscribeEmbeddedPiSession(params: { emitReasoningStream(extractThinkingFromTaggedStream(deltaBuffer)); } - const cleaned = params.enforceFinalTag - ? stripThinkingSegments(stripUnpairedThinkingTags(deltaBuffer)) - : stripThinkingSegments(deltaBuffer); - const next = params.enforceFinalTag - ? (extractFinalText(cleaned)?.trim() ?? cleaned.trim()) - : cleaned.trim(); + const next = stripBlockTags(deltaBuffer, { + thinking: false, + final: false, + }).trim(); if (next && next !== lastStreamedAssistant) { lastStreamedAssistant = next; const { text: cleanedText, mediaUrls } = @@ -822,13 +835,10 @@ export function subscribeEmbeddedPiSession(params: { rawText, rawThinking: extractAssistantThinking(assistantMessage), }); - const cleaned = params.enforceFinalTag - ? stripThinkingSegments(stripUnpairedThinkingTags(rawText)) - : stripThinkingSegments(rawText); - const baseText = - params.enforceFinalTag && cleaned - ? (extractFinalText(cleaned)?.trim() ?? cleaned) - : cleaned; + const text = stripBlockTags(rawText, { + thinking: false, + final: false, + }); const rawThinking = includeReasoning || streamReasoning ? extractAssistantThinking(assistantMessage) || @@ -837,7 +847,6 @@ export function subscribeEmbeddedPiSession(params: { const formattedReasoning = rawThinking ? formatReasoningMessage(rawThinking) : ""; - const text = baseText; const addedDuringMessage = assistantTexts.length > assistantTextBaseline; @@ -919,7 +928,10 @@ export function subscribeEmbeddedPiSession(params: { deltaBuffer = ""; blockBuffer = ""; blockChunker?.reset(); - blockThinkingActive = false; + blockState.thinking = false; + blockState.final = false; + streamState.thinking = false; + streamState.final = false; lastStreamedAssistant = undefined; } } @@ -1001,7 +1013,10 @@ export function subscribeEmbeddedPiSession(params: { blockBuffer = ""; } } - blockThinkingActive = false; + blockState.thinking = false; + blockState.final = false; + streamState.thinking = false; + streamState.final = false; if (pendingCompactionRetry > 0) { resolveCompactionRetry(); } else { diff --git a/src/auto-reply/reply.ts b/src/auto-reply/reply.ts index 024b31280..074e8fe4c 100644 --- a/src/auto-reply/reply.ts +++ b/src/auto-reply/reply.ts @@ -40,6 +40,7 @@ import { import { normalizeMainKey } from "../routing/session-key.js"; import { defaultRuntime } from "../runtime.js"; import { INTERNAL_MESSAGE_PROVIDER } from "../utils/message-provider.js"; +import { isReasoningTagProvider } from "../utils/provider-utils.js"; import { resolveCommandAuthorization } from "./command-auth.js"; import { hasControlCommand } from "./command-detection.js"; import { @@ -1155,6 +1156,10 @@ export async function getReplyFromConfig( resolvedQueue.mode === "collect" || resolvedQueue.mode === "steer-backlog"; const authProfileId = sessionEntry?.authProfileOverride; + // DEBUG: Check provider for reasoning tag + const shouldEnforce = isReasoningTagProvider(provider); + // defaultRuntime.log(`[DEBUG] reply.ts: provider='${provider}', isReasoningTagProvider=${shouldEnforce}`); + const followupRun = { prompt: queuedBody, messageId: sessionCtx.MessageSid, @@ -1193,7 +1198,15 @@ export async function getReplyFromConfig( ownerNumbers: command.ownerList.length > 0 ? command.ownerList : undefined, extraSystemPrompt: extraSystemPrompt || undefined, - ...(provider === "ollama" ? { enforceFinalTag: true } : {}), + ...(isReasoningTagProvider(provider) + ? (() => { + logVerbose(`[reply] Enforcing final tag for provider: ${provider}`); + return { enforceFinalTag: true }; + })() + : (() => { + logVerbose(`[reply] NOT enforcing final tag for provider: ${provider}`); + return {}; + })()), }, }; diff --git a/src/utils/provider-utils.ts b/src/utils/provider-utils.ts new file mode 100644 index 000000000..e15fce2e6 --- /dev/null +++ b/src/utils/provider-utils.ts @@ -0,0 +1,29 @@ +/** + * Utility functions for provider-specific logic and capabilities. + */ + +/** + * Returns true if the provider requires reasoning to be wrapped in tags + * (e.g. and ) in the text stream, rather than using native + * API fields for reasoning/thinking. + */ +export function isReasoningTagProvider(provider: string | undefined | null): boolean { + if (!provider) return false; + const normalized = provider.trim().toLowerCase(); + + // Check for exact matches or known prefixes/substrings for reasoning providers + if ( + normalized === "ollama" || + normalized === "google-gemini-cli" || + normalized === "google-generative-ai" + ) { + return true; + } + + // Handle google-antigravity and its model variations (e.g. google-antigravity/gemini-3) + if (normalized.includes("google-antigravity")) { + return true; + } + + return false; +}