diff --git a/config/max-lines-baseline.txt b/config/max-lines-baseline.txt index 62a743e63e9f..e4bf2bb2b02c 100644 --- a/config/max-lines-baseline.txt +++ b/config/max-lines-baseline.txt @@ -382,7 +382,6 @@ src/agents/embedded-agent-runner/transcript-file-state.test.ts src/agents/embedded-agent-subscribe.handlers.lifecycle.test.ts src/agents/embedded-agent-subscribe.handlers.tools.test.ts src/agents/embedded-agent-subscribe.subscribe-embedded-agent-session.subscribeembeddedagentsession.test.ts -src/agents/embedded-agent-subscribe.ts src/agents/failover-error.test.ts src/agents/failover-error.ts src/agents/harness/native-hook-relay.test.ts diff --git a/src/agents/embedded-agent-subscribe.reply-delivery.ts b/src/agents/embedded-agent-subscribe.reply-delivery.ts new file mode 100644 index 000000000000..8a80abe08caf --- /dev/null +++ b/src/agents/embedded-agent-subscribe.reply-delivery.ts @@ -0,0 +1,273 @@ +import { isPromiseLike } from "@openclaw/normalization-core/promise-like"; +import { getReplyPayloadMetadata, setReplyPayloadMetadata } from "../auto-reply/reply-payload.js"; +import { isSilentReplyText, SILENT_REPLY_TOKEN } from "../auto-reply/tokens.js"; +import { emitAgentEvent } from "../infra/agent-events.js"; +import { normalizeTextForComparison } from "./embedded-agent-helpers.js"; +import type { BlockReplyPayload } from "./embedded-agent-payloads.js"; +import { runBestEffortCallback } from "./embedded-agent-subscribe.callback.js"; +import { + consumePendingAssistantReplyDirectivesIntoReply, + consumePendingToolMediaIntoReply, + hasAssistantVisibleReply, + readPendingToolMediaReply, +} from "./embedded-agent-subscribe.handlers.messages.replies.js"; +import type { EmbeddedAgentSubscribeContext } from "./embedded-agent-subscribe.handlers.types.js"; +import type { SubscribeEmbeddedAgentSessionParams } from "./embedded-agent-subscribe.types.js"; + +type ReplyDeliveryParams = { + params: SubscribeEmbeddedAgentSessionParams; + state: EmbeddedAgentSubscribeContext["state"]; + log: EmbeddedAgentSubscribeContext["log"]; +}; + +export function createReplyDelivery({ params, state, log }: ReplyDeliveryParams) { + const assistantTexts = state.assistantTexts; + const pendingBlockReplyTasks = new Set>(); + const pendingPartialReplyTasks = new Set>(); + const shouldAllowSilentTurnText = (text: string | undefined) => + Boolean(text && isSilentReplyText(text, SILENT_REPLY_TOKEN)); + const emitAssistantStreamDataSafely = ( + delivery: EmbeddedAgentSubscribeContext["state"]["deferredAssistantEvents"][number], + ) => { + const { data } = delivery; + emitAgentEvent({ + runId: params.runId, + stream: "assistant", + data, + }); + if (params.onAgentEvent) { + runBestEffortCallback({ + label: "assistant agent event", + log, + callback: () => + params.onAgentEvent?.({ + stream: "assistant", + data, + }), + }); + } + if (delivery.emitPartialReply && params.onPartialReply && state.shouldEmitPartialReplies) { + try { + const maybeTask = params.onPartialReply(data); + if (isPromiseLike(maybeTask)) { + const task = Promise.resolve(maybeTask) + .then(() => undefined) + .catch((error: unknown) => { + log.warn(`assistant partial reply callback failed: ${String(error)}`); + }); + pendingPartialReplyTasks.add(task); + void task.finally(() => { + pendingPartialReplyTasks.delete(task); + }); + } + } catch (error) { + log.warn(`assistant partial reply callback failed: ${String(error)}`); + } + } + }; + const emitAssistantStreamData = ( + data: EmbeddedAgentSubscribeContext["state"]["deferredAssistantEvents"][number]["data"], + options?: { emitPartialReply?: boolean }, + ) => { + const delivery = { data, emitPartialReply: options?.emitPartialReply === true }; + if (state.deferBlockReplyDelivery) { + state.deferredAssistantEvents.push(delivery); + return; + } + emitAssistantStreamDataSafely(delivery); + }; + const flushDeferredAssistantEvents = () => { + if (state.deferredAssistantEvents.length === 0) { + return; + } + const deferred = state.deferredAssistantEvents.splice(0); + for (const delivery of deferred) { + emitAssistantStreamDataSafely(delivery); + } + }; + const clearDeferredAssistantEvents = () => { + state.deferredAssistantEvents.length = 0; + }; + const deferredToolMediaReplies = new WeakSet(); + const emitBlockReplySafely = ( + payload: Parameters>[0], + options?: { assistantMessageIndex?: number }, + ): boolean => { + if (!params.onBlockReply) { + return false; + } + try { + const taggedPayload = + options?.assistantMessageIndex !== undefined + ? setReplyPayloadMetadata(payload, { + assistantMessageIndex: options.assistantMessageIndex, + }) + : payload; + const assistantMessageIndex = + options?.assistantMessageIndex ?? + getReplyPayloadMetadata(taggedPayload)?.assistantMessageIndex; + const context = assistantMessageIndex === undefined ? undefined : { assistantMessageIndex }; + const maybeTask = context + ? params.onBlockReply(taggedPayload, context) + : params.onBlockReply(taggedPayload); + if (!isPromiseLike(maybeTask)) { + return true; + } + const task = Promise.resolve(maybeTask).catch((err: unknown) => { + log.warn(`block reply callback failed: ${String(err)}`); + }); + pendingBlockReplyTasks.add(task); + void task.finally(() => { + pendingBlockReplyTasks.delete(task); + }); + return true; + } catch (err) { + log.warn(`block reply callback failed: ${String(err)}`); + return false; + } + }; + const emitBlockReply = ( + payload: BlockReplyPayload, + options?: { assistantMessageIndex?: number; consumePendingToolMedia?: boolean }, + ) => { + const withAssistantDirectives = consumePendingAssistantReplyDirectivesIntoReply(state, payload); + const consumesPendingToolMedia = + options?.consumePendingToolMedia !== false && readPendingToolMediaReply(state) !== null; + const withToolMedia = + options?.consumePendingToolMedia === false + ? withAssistantDirectives + : consumePendingToolMediaIntoReply(state, withAssistantDirectives); + const assistantTranscriptMediaUrls = Array.from(new Set(payload.mediaUrls ?? [])); + const taggedPayload = + options?.assistantMessageIndex !== undefined + ? setReplyPayloadMetadata(withToolMedia, { + assistantMessageIndex: options.assistantMessageIndex, + ...(assistantTranscriptMediaUrls.length > 0 ? { assistantTranscriptMediaUrls } : {}), + }) + : withToolMedia; + if (state.deferBlockReplyDelivery) { + if (consumesPendingToolMedia) { + deferredToolMediaReplies.add(taggedPayload); + } + state.deferredBlockReplies.push(taggedPayload); + return; + } + const emitted = emitBlockReplySafely(taggedPayload, options); + if (emitted && !taggedPayload.isReasoning && hasAssistantVisibleReply(taggedPayload)) { + state.visibleBlockReplyCount += 1; + if (consumesPendingToolMedia) { + state.hasToolMediaBlockReply = true; + } + } + }; + const flushDeferredBlockReplies = () => { + if (state.deferredBlockReplies.length === 0) { + return; + } + const deferred = state.deferredBlockReplies.splice(0); + for (const payload of deferred) { + const emitted = emitBlockReplySafely(payload); + if (emitted && !payload.isReasoning && hasAssistantVisibleReply(payload)) { + state.visibleBlockReplyCount += 1; + if (deferredToolMediaReplies.has(payload)) { + state.hasToolMediaBlockReply = true; + } + } + } + }; + const clearDeferredBlockReplies = () => { + state.deferredBlockReplies.length = 0; + }; + + const rememberAssistantText = (text: string) => { + state.lastAssistantTextMessageIndex = state.assistantMessageIndex; + state.lastAssistantTextTrimmed = text.trimEnd(); + const normalized = normalizeTextForComparison(text); + state.lastAssistantTextNormalized = normalized.length > 0 ? normalized : undefined; + }; + + const shouldSkipAssistantText = (text: string) => { + if (state.lastAssistantTextMessageIndex !== state.assistantMessageIndex) { + return false; + } + const trimmed = text.trimEnd(); + if (trimmed && trimmed === state.lastAssistantTextTrimmed) { + return true; + } + const normalized = normalizeTextForComparison(text); + if (normalized.length > 0 && normalized === state.lastAssistantTextNormalized) { + return true; + } + return false; + }; + + const pushAssistantText = (text: string) => { + if (!text) { + return; + } + if (params.silentExpected && !shouldAllowSilentTurnText(text)) { + return; + } + if (shouldSkipAssistantText(text)) { + return; + } + assistantTexts.push(text); + rememberAssistantText(text); + }; + + const finalizeAssistantTexts = (args: { + text: string; + addedDuringMessage: boolean; + chunkerHasBuffered: boolean; + }) => { + const { text, addedDuringMessage, chunkerHasBuffered } = args; + + // If we're not streaming block replies, ensure the final payload includes + // the final text even when interim streaming was enabled. + if (state.includeReasoning && text && !params.onBlockReply) { + if (assistantTexts.length > state.assistantTextBaseline) { + assistantTexts.splice( + state.assistantTextBaseline, + assistantTexts.length - state.assistantTextBaseline, + text, + ); + rememberAssistantText(text); + } else { + pushAssistantText(text); + } + state.suppressBlockChunks = true; + } else if (!addedDuringMessage && !chunkerHasBuffered && text) { + // Non-streaming models (no text_delta): ensure assistantTexts gets the final + // text when the chunker has nothing buffered to drain. + pushAssistantText(text); + } + + state.assistantTextBaseline = assistantTexts.length; + }; + + const waitForPendingEvents = async () => { + // Partial presentation stays concurrent with provider events, but terminal + // settlement must observe callbacks launched while the event chain drains. + while (state.pendingEventChain || pendingPartialReplyTasks.size > 0) { + await Promise.allSettled([ + ...(state.pendingEventChain ? [state.pendingEventChain] : []), + ...pendingPartialReplyTasks, + ]); + } + }; + + return { + assistantTexts, + clearDeferredAssistantEvents, + clearDeferredBlockReplies, + emitAssistantStreamData, + emitBlockReply, + finalizeAssistantTexts, + flushDeferredAssistantEvents, + flushDeferredBlockReplies, + pendingBlockReplyTasks, + pushAssistantText, + shouldSkipAssistantText, + waitForPendingEvents, + }; +} diff --git a/src/agents/embedded-agent-subscribe.run-state.ts b/src/agents/embedded-agent-subscribe.run-state.ts new file mode 100644 index 000000000000..ce0e64094578 --- /dev/null +++ b/src/agents/embedded-agent-subscribe.run-state.ts @@ -0,0 +1,152 @@ +import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; +import { createInlineCodeState } from "../../packages/markdown-core/src/code-spans.js"; +import { createEmbeddedRunReplayState } from "./embedded-agent-runner/replay-state.js"; +import type { EmbeddedAgentSubscribeState } from "./embedded-agent-subscribe.handlers.types.js"; +import type { SubscribeEmbeddedAgentSessionParams } from "./embedded-agent-subscribe.types.js"; +import { createThinkingTagStreamState } from "./embedded-agent-utils.js"; +import { mediaUrlsFromGeneratedAttachments } from "./generated-attachments.js"; +import { hasGeneratedMediaCompletionEvent } from "./internal-event-contract.js"; +import type { AgentInternalEvent } from "./internal-events.js"; + +function collectPendingMediaFromInternalEvents( + events: SubscribeEmbeddedAgentSessionParams["internalEvents"], +): { + mediaUrls: string[]; + attachments: NonNullable; + trustByUrl: Map; +} { + if (!events?.length) { + return { mediaUrls: [], attachments: [], trustByUrl: new Map() }; + } + const pending: string[] = []; + const attachments: NonNullable = []; + const indexByUrl = new Map(); + const trustedByUrl = new Map(); + for (const event of events) { + const generatedMediaEvent = hasGeneratedMediaCompletionEvent([event]); + const attachmentByUrl = new Map( + (event.attachments ?? []).flatMap((attachment) => { + const reference = normalizeOptionalString( + attachment.path ?? attachment.url ?? attachment.mediaUrl ?? attachment.filePath, + ); + return reference ? [[reference, attachment] as const] : []; + }), + ); + const mediaUrls = [ + ...(Array.isArray(event.mediaUrls) ? event.mediaUrls : []), + ...mediaUrlsFromGeneratedAttachments(event.attachments), + ]; + for (const mediaUrl of mediaUrls) { + const normalized = normalizeOptionalString(mediaUrl) ?? ""; + if (!normalized) { + continue; + } + const metadata = attachmentByUrl.get(normalized); + const existingIndex = indexByUrl.get(normalized); + if (existingIndex !== undefined) { + trustedByUrl.set(normalized, trustedByUrl.get(normalized) === true || generatedMediaEvent); + if (metadata && Object.keys(attachments[existingIndex] ?? {}).length === 0) { + attachments[existingIndex] = metadata; + } + continue; + } + indexByUrl.set(normalized, pending.length); + trustedByUrl.set(normalized, generatedMediaEvent); + pending.push(normalized); + attachments.push(metadata ?? {}); + } + } + return { mediaUrls: pending, attachments, trustByUrl: trustedByUrl }; +} + +export function createEmbeddedAgentSubscribeState( + params: SubscribeEmbeddedAgentSessionParams, +): EmbeddedAgentSubscribeState { + const reasoningMode = params.reasoningMode ?? "off"; + const canShowReasoning = params.thinkingLevel !== "off"; + const initialPendingToolMedia = collectPendingMediaFromInternalEvents(params.internalEvents); + return { + assistantTexts: [], + toolMetas: [], + acceptedSessionSpawns: [], + toolMetaById: new Map(), + toolSummaryById: new Set(), + liveEditDiffStateById: new Map(), + itemActiveIds: new Set(), + itemStartedCount: 0, + itemCompletedCount: 0, + assistantTurnCount: 0, + lastToolError: undefined, + blockReplyBreak: params.blockReplyBreak ?? "text_end", + reasoningMode, + includeReasoning: reasoningMode === "on" && canShowReasoning, + shouldEmitPartialReplies: !(reasoningMode === "on" && !params.onBlockReply), + streamReasoning: + (params.streamReasoningInNonStreamModes === true + ? reasoningMode !== "on" + : reasoningMode === "stream") && + canShowReasoning && + typeof params.onReasoningStream === "function", + deltaBuffer: "", + thinkingTagStream: createThinkingTagStreamState(), + blockBuffer: "", + // Track if a streamed chunk opened a block (stateful across chunks). + blockState: { thinking: false, final: false, inlineCode: createInlineCodeState() }, + partialBlockState: { thinking: false, final: false, inlineCode: createInlineCodeState() }, + lastStreamedAssistant: undefined, + lastStreamedAssistantCleaned: undefined, + emittedAssistantUpdate: false, + lastStreamedReasoning: undefined, + lastBlockReplyText: undefined, + lastDeliveredBlockReplyText: undefined, + deferBlockReplyDelivery: typeof params.onBeforeTerminalDelivery === "function", + deferredBlockReplies: [], + deferredAssistantEvents: [], + toolExecutionSinceLastBlockReply: false, + reasoningStreamOpen: false, + assistantMessageIndex: 0, + lastAssistantStreamContentIndex: undefined, + lastAssistantStreamItemId: undefined, + lastAssistantTextMessageIndex: -1, + lastAssistantTextNormalized: undefined, + lastAssistantTextTrimmed: undefined, + assistantTextBaseline: 0, + suppressBlockChunks: false, // Avoid late chunk inserts after final text merge. + lastReasoningSent: undefined, + pendingAssistantUsage: undefined, + assistantUsageCommitted: false, + compactionInFlight: false, + lastCompactionTokensAfter: undefined, + pendingCompactionRetry: 0, + compactionRetryResolve: undefined, + compactionRetryReject: undefined, + compactionRetryPromise: null, + unsubscribed: false, + replayState: createEmbeddedRunReplayState(params.initialReplayState), + livenessState: "working", + hadDeterministicSideEffect: false, + pendingEventChain: null, + messagingToolSentTexts: [], + messagingToolSentTextsNormalized: [], + currentSourceMessagingToolSentTextsNormalized: [], + currentSourceMessagingToolHeldPartial: undefined, + messagingToolSentTargets: [], + heartbeatToolResponse: undefined, + messagingToolSentMediaUrls: [], + messagingToolSourceReplyPayloads: [], + messageToolOnlySourceReplyDelivered: false, + pendingMessagingTexts: new Map(), + pendingMessagingTargets: new Map(), + successfulCronAdds: 0, + pendingMessagingMediaUrls: new Map(), + pendingToolMediaUrls: initialPendingToolMedia.mediaUrls, + pendingToolMediaAttachments: initialPendingToolMedia.attachments, + pendingToolMediaTrustByUrl: initialPendingToolMedia.trustByUrl, + pendingToolAudioAsVoice: false, + hasToolMediaBlockReply: false, + visibleBlockReplyCount: 0, + pendingAssistantReplyDirectives: undefined, + deterministicApprovalPromptPending: false, + deterministicApprovalPromptSent: false, + }; +} diff --git a/src/agents/embedded-agent-subscribe.stream-rendering.ts b/src/agents/embedded-agent-subscribe.stream-rendering.ts new file mode 100644 index 000000000000..599002062fc2 --- /dev/null +++ b/src/agents/embedded-agent-subscribe.stream-rendering.ts @@ -0,0 +1,640 @@ +import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; +import type { InlineCodeState } from "../../packages/markdown-core/src/code-spans.js"; +import { + buildCodeSpanIndex, + createInlineCodeState, +} from "../../packages/markdown-core/src/code-spans.js"; +import type { FenceScanState } from "../../packages/markdown-core/src/fences.js"; +import { createStreamingDirectiveAccumulator } from "../auto-reply/reply/streaming-directives.js"; +import { emitAgentEvent } from "../infra/agent-events.js"; +import { findFinalTagMatches } from "../shared/text/final-tags.js"; +import { hasOrphanReasoningCloseBoundary } from "../shared/text/reasoning-tags.js"; +import { + isMessagingToolDuplicateNormalized, + normalizeTextForComparison, +} from "./embedded-agent-helpers.js"; +import { runBestEffortCallback } from "./embedded-agent-subscribe.callback.js"; +import type { EmbeddedAgentSubscribeContext } from "./embedded-agent-subscribe.handlers.types.js"; +import type { SubscribeEmbeddedAgentSessionParams } from "./embedded-agent-subscribe.types.js"; +import { + createThinkingTagStreamState, + stripDowngradedToolCallText, + THINKING_TAG_SCAN_RE, +} from "./embedded-agent-utils.js"; + +const STREAM_STRIPPED_BLOCK_TAG_NAMES = [ + "final", + "think", + "thinking", + "thought", + "antthinking", + "antml:think", + "antml:thinking", + "antml:thought", + "mm:think", + "mm:thinking", + "mm:thought", +] as const; + +function isPotentialTrailingBlockTagFragment(fragment: string): boolean { + if (!fragment.startsWith("<") || fragment.includes(">")) { + return false; + } + const body = fragment.toLowerCase().slice(1).trimStart().replace(/^\//, "").trimStart(); + if (!body) { + return true; + } + const namePart = body.split(/[\s/>]/, 1)[0] ?? ""; + if (!namePart) { + return true; + } + return STREAM_STRIPPED_BLOCK_TAG_NAMES.some((name) => { + return name.startsWith(namePart) || namePart === name; + }); +} + +function splitTrailingBlockTagFragment( + text: string, + isInsideCodeSpan: (index: number) => boolean, +): { text: string; pendingTagFragment?: string } { + const fragmentStart = text.lastIndexOf("<"); + if (fragmentStart === -1 || isInsideCodeSpan(fragmentStart)) { + return { text }; + } + const fragment = text.slice(fragmentStart); + if (!isPotentialTrailingBlockTagFragment(fragment)) { + return { text }; + } + return { + text: text.slice(0, fragmentStart), + pendingTagFragment: fragment, + }; +} + +function splitTrailingFenceFragment( + text: string, + startsAtLineStart: boolean, +): { text: string; pendingFenceFragment?: string } { + const lineStart = text.lastIndexOf("\n") + 1; + const line = text.slice(lineStart); + if ((!startsAtLineStart && lineStart === 0) || !/^(?: {0,3})(?:`+|~+)$/.test(line)) { + return { text }; + } + return { + text: text.slice(0, lineStart), + pendingFenceFragment: line, + }; +} + +type StreamRenderingParams = { + params: SubscribeEmbeddedAgentSessionParams; + state: EmbeddedAgentSubscribeContext["state"]; + log: EmbeddedAgentSubscribeContext["log"]; + blockChunker: EmbeddedAgentSubscribeContext["blockChunker"]; + emitBlockReply: EmbeddedAgentSubscribeContext["emitBlockReply"]; + pendingBlockReplyTasks: Set>; + pushAssistantText: (text: string) => void; + shouldSkipAssistantText: (text: string) => boolean; +}; + +export function createStreamRendering({ + params, + state, + log, + blockChunker, + emitBlockReply, + pendingBlockReplyTasks, + pushAssistantText, + shouldSkipAssistantText, +}: StreamRenderingParams) { + const messagingToolSentTextsNormalized = state.messagingToolSentTextsNormalized; + const messagingToolSourceReplyPayloads = state.messagingToolSourceReplyPayloads; + const replyDirectiveAccumulator = createStreamingDirectiveAccumulator(); + const partialReplyDirectiveAccumulator = createStreamingDirectiveAccumulator(); + + const stripBlockTags = ( + text: string, + stateLocal: { + thinking: boolean; + final: boolean; + inlineCode?: InlineCodeState; + fence?: FenceScanState; + reasoningInlineCode?: InlineCodeState; + reasoningFence?: FenceScanState; + reasoningPendingFenceFragment?: string; + finalInlineCode?: InlineCodeState; + finalFence?: FenceScanState; + pendingFenceFragment?: string; + pendingTagFragment?: string; + }, + options?: { final?: boolean; completeMarkdownChunk?: boolean }, + ): string => { + const input = `${stateLocal.pendingFenceFragment ?? ""}${stateLocal.pendingTagFragment ?? ""}${text}`; + stateLocal.pendingFenceFragment = undefined; + stateLocal.pendingTagFragment = undefined; + if (!input) { + return text; + } + + const { text: fenceInput, pendingFenceFragment } = options?.final + ? { text: input, pendingFenceFragment: undefined } + : options?.completeMarkdownChunk + ? { text: input, pendingFenceFragment: undefined } + : splitTrailingFenceFragment(input, stateLocal.fence?.atLineStart ?? true); + stateLocal.pendingFenceFragment = pendingFenceFragment; + if (!fenceInput) { + return ""; + } + + const inlineStateStart = stateLocal.inlineCode ?? createInlineCodeState(); + const fenceStateStart = stateLocal.fence; + const initialCodeSpans = buildCodeSpanIndex(fenceInput, inlineStateStart, fenceStateStart); + const { text: scanText, pendingTagFragment } = options?.final + ? { text: fenceInput, pendingTagFragment: undefined } + : splitTrailingBlockTagFragment(fenceInput, initialCodeSpans.isInside); + stateLocal.pendingTagFragment = pendingTagFragment; + if (!scanText) { + return ""; + } + const codeSpans = buildCodeSpanIndex(scanText, inlineStateStart, fenceStateStart); + + let processed = ""; + THINKING_TAG_SCAN_RE.lastIndex = 0; + let lastIndex = 0; + let lastCodeIndex = 0; + let inThinking = stateLocal.thinking; + // Hidden reasoning has its own code state: malformed hidden fences must not + // mark later visible text as code, but literal close tags there stay hidden. + let hiddenInlineState: InlineCodeState = stateLocal.reasoningInlineCode + ? { ...stateLocal.reasoningInlineCode } + : createInlineCodeState(); + let hiddenFenceState: FenceScanState | undefined = stateLocal.reasoningFence?.open + ? { + atLineStart: stateLocal.reasoningFence.atLineStart, + open: { ...stateLocal.reasoningFence.open }, + } + : stateLocal.reasoningFence + ? { atLineStart: stateLocal.reasoningFence.atLineStart } + : undefined; + let hiddenPendingFenceFragment = stateLocal.reasoningPendingFenceFragment; + stateLocal.reasoningPendingFenceFragment = undefined; + const advanceHiddenCodeState = (segment: string) => { + const hiddenInput = `${hiddenPendingFenceFragment ?? ""}${segment}`; + hiddenPendingFenceFragment = undefined; + if (!hiddenInput) { + return; + } + const { text: hiddenFenceInput, pendingFenceFragment: pendingFenceFragmentLocal } = + options?.final + ? { text: hiddenInput, pendingFenceFragment: undefined } + : options?.completeMarkdownChunk + ? { text: hiddenInput, pendingFenceFragment: undefined } + : splitTrailingFenceFragment(hiddenInput, hiddenFenceState?.atLineStart ?? true); + hiddenPendingFenceFragment = pendingFenceFragmentLocal; + if (!hiddenFenceInput) { + return; + } + const next = buildCodeSpanIndex(hiddenFenceInput, hiddenInlineState, hiddenFenceState); + hiddenInlineState = next.inlineState; + hiddenFenceState = next.fenceState; + }; + for (const match of scanText.matchAll(THINKING_TAG_SCAN_RE)) { + const idx = match.index ?? 0; + const isClose = match[1] === "/"; + if (inThinking) { + advanceHiddenCodeState(scanText.slice(lastCodeIndex, idx)); + } + const isInsideHiddenCode = + inThinking && (hiddenInlineState.open || Boolean(hiddenFenceState?.open)); + lastCodeIndex = idx + match[0].length; + if ((!inThinking && codeSpans.isInside(idx)) || isInsideHiddenCode) { + if (inThinking) { + advanceHiddenCodeState(match[0]); + } + continue; + } + if (!inThinking) { + if (isClose) { + const afterIndex = idx + match[0].length; + const before = scanText.slice(lastIndex, idx); + const after = scanText.slice(afterIndex); + if (hasOrphanReasoningCloseBoundary({ before, after })) { + processed = ""; + } else { + processed += before; + } + lastIndex = afterIndex; + continue; + } + processed += scanText.slice(lastIndex, idx); + hiddenInlineState = createInlineCodeState(); + hiddenFenceState = undefined; + hiddenPendingFenceFragment = undefined; + } + inThinking = !isClose; + if (!inThinking) { + hiddenInlineState = createInlineCodeState(); + hiddenFenceState = undefined; + hiddenPendingFenceFragment = undefined; + } + lastIndex = idx + match[0].length; + } + if (inThinking) { + advanceHiddenCodeState(scanText.slice(lastCodeIndex)); + } + if (!inThinking) { + processed += scanText.slice(lastIndex); + } + stateLocal.thinking = inThinking; + stateLocal.reasoningInlineCode = inThinking ? hiddenInlineState : undefined; + stateLocal.reasoningFence = inThinking ? hiddenFenceState : undefined; + stateLocal.reasoningPendingFenceFragment = inThinking ? hiddenPendingFenceFragment : undefined; + + // If enforcement is disabled, we still strip the tags themselves to prevent + // hallucinations (e.g. Minimax copying the style) from leaking, but we + // do not enforce buffering/extraction logic. + const finalCodeSpans = buildCodeSpanIndex(processed, inlineStateStart, fenceStateStart); + if (!params.enforceFinalTag) { + stateLocal.inlineCode = finalCodeSpans.inlineState; + stateLocal.fence = finalCodeSpans.fenceState; + return stripFinalTagsOutsideCodeSpans(processed, finalCodeSpans.isInside); + } + + // If enforcement is enabled, only return text that appeared inside a block. + let result = ""; + let lastFinalIndex = 0; + let inFinal = stateLocal.final; + let everInFinal = stateLocal.final; + + for (const match of findFinalTagMatches(processed)) { + const idx = match.index; + if (finalCodeSpans.isInside(idx)) { + continue; + } + const isClose = match.isClose; + const isSelfClosing = match.isSelfClosing; + + if (isSelfClosing) { + if (inFinal) { + result += processed.slice(lastFinalIndex, idx); + inFinal = false; + } else { + inFinal = true; + everInFinal = true; + } + lastFinalIndex = idx + match.text.length; + } else if (!inFinal && !isClose) { + // Found start tag. + inFinal = true; + everInFinal = true; + lastFinalIndex = idx + match.text.length; + } else if (inFinal && isClose) { + // Found end tag. + result += processed.slice(lastFinalIndex, idx); + inFinal = false; + lastFinalIndex = idx + match.text.length; + } + } + + if (inFinal) { + result += processed.slice(lastFinalIndex); + } + stateLocal.final = inFinal; + + // Strict Mode: If enforcing final tags, we MUST NOT return content unless + // we have seen a tag. Otherwise, we leak "thinking out loud" text + // (e.g. "**Locating Manulife**...") that the model emitted without tags. + if (!everInFinal) { + stateLocal.inlineCode = createInlineCodeState(); + stateLocal.fence = finalCodeSpans.fenceState; + stateLocal.finalInlineCode = undefined; + stateLocal.finalFence = undefined; + return ""; + } + + // Hardened Cleanup: Remove any remaining tags that might have been + // missed (e.g. nested tags or hallucinations) to prevent leakage. + const finalResultInlineStateStart = stateLocal.finalInlineCode ?? createInlineCodeState(); + const finalResultFenceStateStart = stateLocal.finalFence; + const resultCodeSpans = buildCodeSpanIndex( + result, + finalResultInlineStateStart, + finalResultFenceStateStart, + ); + stateLocal.inlineCode = finalCodeSpans.inlineState; + stateLocal.fence = finalCodeSpans.fenceState; + stateLocal.finalInlineCode = inFinal ? resultCodeSpans.inlineState : undefined; + stateLocal.finalFence = inFinal ? resultCodeSpans.fenceState : undefined; + return stripFinalTagsOutsideCodeSpans(result, resultCodeSpans.isInside); + }; + + const stripFinalTagsOutsideCodeSpans = (text: string, isInside: (index: number) => boolean) => { + let output = ""; + let lastIndex = 0; + for (const match of findFinalTagMatches(text)) { + const idx = match.index; + if (isInside(idx)) { + continue; + } + output += text.slice(lastIndex, idx); + lastIndex = idx + match.text.length; + } + output += text.slice(lastIndex); + return output; + }; + const hasMessageToolOnlySourceDelivery = () => + params.sourceReplyDeliveryMode === "message_tool_only" && + (state.messageToolOnlySourceReplyDelivered || + params.hasDeliveredMessageToolOnlySourceReply?.() === true || + messagingToolSourceReplyPayloads.length > 0); + + const emitBlockChunk = ( + text: string, + options?: { assistantMessageIndex?: number; final?: boolean; completeMarkdownChunk?: boolean }, + ) => { + if (state.suppressBlockChunks || params.silentExpected) { + return; + } + // Strip and blocks across chunk boundaries to avoid leaking reasoning. + // Also strip downgraded tool call text ([Tool Call: ...], [Historical context: ...], etc.). + const blockReplyText = stripDowngradedToolCallText( + stripBlockTags(text, state.blockState, { + final: options?.final === true, + completeMarkdownChunk: options?.completeMarkdownChunk === true, + }), + ).trimEnd(); + if (!blockReplyText) { + return; + } + if (blockReplyText === state.lastBlockReplyText) { + return; + } + const markBlockReplyTextHandled = () => { + state.lastBlockReplyText = blockReplyText; + state.lastDeliveredBlockReplyText = blockReplyText; + state.toolExecutionSinceLastBlockReply = false; + }; + if (hasMessageToolOnlySourceDelivery()) { + markBlockReplyTextHandled(); + return; + } + let chunk = blockReplyText; + let slicedPrefixReplay = false; + const lastDeliveredBlockReplyText = state.lastDeliveredBlockReplyText; + const blockReplySuffix = lastDeliveredBlockReplyText + ? blockReplyText.slice(lastDeliveredBlockReplyText.length) + : ""; + const prefixReplayCandidate = Boolean( + state.blockReplyBreak === "text_end" && + state.toolExecutionSinceLastBlockReply && + lastDeliveredBlockReplyText && + lastDeliveredBlockReplyText.trimEnd().endsWith(":") && + blockReplyText.length > lastDeliveredBlockReplyText.length && + blockReplyText.startsWith(lastDeliveredBlockReplyText), + ); + if (prefixReplayCandidate && !/^\s/.test(blockReplySuffix)) { + chunk = blockReplySuffix; + slicedPrefixReplay = true; + } + if (!chunk) { + return; + } + + // Only check committed (successful) messaging tool texts - checking pending texts + // is risky because if the tool fails after suppression, the user gets no response + const normalizedChunk = normalizeTextForComparison(chunk); + const normalizedReplaySuffix = prefixReplayCandidate + ? normalizeTextForComparison(blockReplySuffix.trimStart()) + : ""; + const isMessagingDuplicate = + isMessagingToolDuplicateNormalized(normalizedChunk, messagingToolSentTextsNormalized) || + (prefixReplayCandidate && + isMessagingToolDuplicateNormalized( + normalizedReplaySuffix, + messagingToolSentTextsNormalized, + )); + if (isMessagingDuplicate) { + log.debug( + `Skipping block reply - already sent via messaging tool: ${truncateUtf16Safe(chunk, 50)}...`, + ); + if (prefixReplayCandidate) { + markBlockReplyTextHandled(); + } + return; + } + + if (shouldSkipAssistantText(chunk)) { + if (slicedPrefixReplay) { + markBlockReplyTextHandled(); + } + return; + } + + if (!params.onBlockReply) { + pushAssistantText(chunk); + markBlockReplyTextHandled(); + return; + } + const splitResult = replyDirectiveAccumulator.consume(chunk); + if (!splitResult) { + if (slicedPrefixReplay) { + markBlockReplyTextHandled(); + } + return; + } + const { + text: cleanedText, + mediaUrls, + audioAsVoice, + replyToId, + replyToTag, + replyToCurrent, + } = splitResult; + if (!cleanedText && (!mediaUrls || mediaUrls.length === 0) && !audioAsVoice) { + if (slicedPrefixReplay) { + markBlockReplyTextHandled(); + } + return; + } + pushAssistantText(chunk); + emitBlockReply( + { + text: cleanedText, + mediaUrls: mediaUrls?.length ? mediaUrls : undefined, + audioAsVoice, + replyToId, + replyToTag, + replyToCurrent, + }, + { + assistantMessageIndex: options?.assistantMessageIndex ?? state.assistantMessageIndex, + consumePendingToolMedia: + options?.final === true || Boolean(mediaUrls?.length || audioAsVoice), + }, + ); + markBlockReplyTextHandled(); + }; + + const consumeReplyDirectives = (text: string, options?: { final?: boolean }) => + replyDirectiveAccumulator.consume(text, options); + const consumePartialReplyDirectives = (text: string, options?: { final?: boolean }) => + partialReplyDirectiveAccumulator.consume(text, options); + + const flushBlockReplyBuffer = (options?: { + assistantMessageIndex?: number; + final?: boolean; + }): void | Promise => { + if (!params.onBlockReply) { + return; + } + if (blockChunker?.hasBuffered()) { + if (options?.final) { + let pendingChunk: string | undefined; + blockChunker.drain({ + force: true, + emit: (text) => { + if (pendingChunk !== undefined) { + emitBlockChunk(pendingChunk, { + assistantMessageIndex: options.assistantMessageIndex, + completeMarkdownChunk: true, + }); + } + pendingChunk = text; + }, + }); + if (pendingChunk !== undefined) { + emitBlockChunk(pendingChunk, { + assistantMessageIndex: options.assistantMessageIndex, + completeMarkdownChunk: true, + final: true, + }); + } + } else { + blockChunker.drain({ force: true, emit: (text) => emitBlockChunk(text, options) }); + } + blockChunker.reset(); + } else if (state.blockBuffer.length > 0) { + emitBlockChunk(state.blockBuffer, options); + state.blockBuffer = ""; + } + if (options?.final) { + emitBlockChunk("", options); + } + if (pendingBlockReplyTasks.size === 0) { + return; + } + return (async () => { + while (pendingBlockReplyTasks.size > 0) { + await Promise.allSettled(pendingBlockReplyTasks); + } + })(); + }; + + const emitReasoningStream = (text: string) => { + if (params.silentExpected) { + return; + } + const trimmed = text.trim(); + if (!trimmed) { + return; + } + if (trimmed === state.lastStreamedReasoning) { + return; + } + // Compute delta: new text since the last emitted reasoning. + // Guard against non-prefix changes (e.g. trim altering earlier content). + const prior = state.lastStreamedReasoning ?? ""; + const delta = trimmed.startsWith(prior) ? trimmed.slice(prior.length) : trimmed; + state.lastStreamedReasoning = trimmed; + + // Emit-always: the thinking stream always reaches the bus and session + // archive. /reasoning (streamReasoning) gates only the rendering hook + // below; display surfaces (TUI showThinking, webchat isReasoning drops) + // gate presentation on their side. + emitAgentEvent({ + runId: params.runId, + stream: "thinking", + data: { + text: trimmed, + delta, + }, + }); + + // Message-tool-only delivery makes later reasoning private: once the + // user-facing reply has gone out via the message tool, the channel shows + // only what was explicitly sent, so trailing reasoning must stay out of the + // render hook — uniformly, whether the thinking block rode in on a tool call + // or arrived on its own. It still reaches the bus/archive above. + if (state.streamReasoning && !hasMessageToolOnlySourceDelivery() && params.onReasoningStream) { + runBestEffortCallback({ + label: "reasoning stream", + log, + callback: () => + params.onReasoningStream?.({ + text: trimmed, + ...(state.reasoningMode === "stream" ? {} : { requiresReasoningProgressOptIn: true }), + }), + }); + } + }; + + const resetAssistantMessageState = (nextAssistantTextBaseline: number) => { + state.deltaBuffer = ""; + state.thinkingTagStream = createThinkingTagStreamState(); + state.blockBuffer = ""; + blockChunker?.reset(); + replyDirectiveAccumulator.reset(); + partialReplyDirectiveAccumulator.reset(); + state.blockState.thinking = false; + state.blockState.final = false; + state.blockState.inlineCode = createInlineCodeState(); + state.blockState.fence = undefined; + state.blockState.reasoningInlineCode = undefined; + state.blockState.reasoningFence = undefined; + state.blockState.reasoningPendingFenceFragment = undefined; + state.blockState.finalInlineCode = undefined; + state.blockState.finalFence = undefined; + state.blockState.pendingFenceFragment = undefined; + state.blockState.pendingTagFragment = undefined; + state.partialBlockState.thinking = false; + state.partialBlockState.final = false; + state.partialBlockState.inlineCode = createInlineCodeState(); + state.partialBlockState.fence = undefined; + state.partialBlockState.reasoningInlineCode = undefined; + state.partialBlockState.reasoningFence = undefined; + state.partialBlockState.reasoningPendingFenceFragment = undefined; + state.partialBlockState.finalInlineCode = undefined; + state.partialBlockState.finalFence = undefined; + state.partialBlockState.pendingFenceFragment = undefined; + state.partialBlockState.pendingTagFragment = undefined; + state.lastStreamedAssistant = undefined; + state.lastStreamedAssistantCleaned = undefined; + state.currentSourceMessagingToolHeldPartial = undefined; + state.emittedAssistantUpdate = false; + state.lastBlockReplyText = undefined; + state.lastStreamedReasoning = undefined; + state.lastReasoningSent = undefined; + state.reasoningStreamOpen = false; + state.suppressBlockChunks = false; + state.pendingAssistantUsage = undefined; + state.assistantUsageCommitted = false; + state.assistantMessageIndex += 1; + state.lastAssistantStreamContentIndex = undefined; + state.lastAssistantStreamItemId = undefined; + state.lastAssistantTextMessageIndex = -1; + state.lastAssistantTextNormalized = undefined; + state.lastAssistantTextTrimmed = undefined; + state.assistantTextBaseline = nextAssistantTextBaseline; + state.pendingAssistantReplyDirectives = undefined; + }; + + return { + consumePartialReplyDirectives, + consumeReplyDirectives, + emitBlockChunk, + emitReasoningStream, + flushBlockReplyBuffer, + resetAssistantMessageState, + stripBlockTags, + }; +} diff --git a/src/agents/embedded-agent-subscribe.ts b/src/agents/embedded-agent-subscribe.ts index a46b44a680bb..de8298781329 100644 --- a/src/agents/embedded-agent-subscribe.ts +++ b/src/agents/embedded-agent-subscribe.ts @@ -1,48 +1,21 @@ /** * Subscribes to embedded-agent sessions and streams formatted replies/events. */ -import { isPromiseLike } from "@openclaw/normalization-core/promise-like"; -import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; -import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; -import type { InlineCodeState } from "../../packages/markdown-core/src/code-spans.js"; -import { - buildCodeSpanIndex, - createInlineCodeState, -} from "../../packages/markdown-core/src/code-spans.js"; -import type { FenceScanState } from "../../packages/markdown-core/src/fences.js"; -import { getReplyPayloadMetadata, setReplyPayloadMetadata } from "../auto-reply/reply-payload.js"; -import { createStreamingDirectiveAccumulator } from "../auto-reply/reply/streaming-directives.js"; -import { isSilentReplyText, SILENT_REPLY_TOKEN } from "../auto-reply/tokens.js"; import { formatToolAggregate } from "../auto-reply/tool-meta.js"; -import { emitAgentEvent, emitAgentEventIfCurrent } from "../infra/agent-events.js"; +import { emitAgentEventIfCurrent } from "../infra/agent-events.js"; import { recordAgentRunOutputTokens } from "../infra/agent-run-usage.js"; import type { AssistantMessage } from "../llm/types.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; -import { findFinalTagMatches } from "../shared/text/final-tags.js"; -import { hasOrphanReasoningCloseBoundary } from "../shared/text/reasoning-tags.js"; import { parseInlineDirectives } from "../utils/directive-tags.js"; import { isDeliverableMessageChannel, normalizeMessageChannel } from "../utils/message-channel.js"; import { EmbeddedBlockChunker } from "./embedded-agent-block-chunker.js"; -import { - isMessagingToolDuplicateNormalized, - normalizeTextForComparison, -} from "./embedded-agent-helpers.js"; -import type { BlockReplyPayload } from "./embedded-agent-payloads.js"; import { hasCommittedMessagingToolDeliveryEvidence } from "./embedded-agent-runner/delivery-evidence.js"; -import { - createEmbeddedRunReplayState, - mergeEmbeddedRunReplayState, -} from "./embedded-agent-runner/replay-state.js"; +import { mergeEmbeddedRunReplayState } from "./embedded-agent-runner/replay-state.js"; import { consumeEmbeddedToolSendReceipt } from "./embedded-agent-runner/tool-send-receipts.js"; import type { EmbeddedRunLivenessState } from "./embedded-agent-runner/types.js"; import { runBestEffortCallback } from "./embedded-agent-subscribe.callback.js"; import { createEmbeddedAgentSessionEventHandler } from "./embedded-agent-subscribe.handlers.js"; -import { - consumePendingAssistantReplyDirectivesIntoReply, - consumePendingToolMediaIntoReply, - hasAssistantVisibleReply, - readPendingToolMediaReply, -} from "./embedded-agent-subscribe.handlers.messages.replies.js"; +import { readPendingToolMediaReply } from "./embedded-agent-subscribe.handlers.messages.replies.js"; import { cleanupRunToolStartData, handleToolExecutionEnd, @@ -52,37 +25,19 @@ import type { EmbeddedAgentSubscribeContext, EmbeddedAgentSubscribeState, } from "./embedded-agent-subscribe.handlers.types.js"; +import { createReplyDelivery } from "./embedded-agent-subscribe.reply-delivery.js"; +import { createEmbeddedAgentSubscribeState } from "./embedded-agent-subscribe.run-state.js"; +import { createStreamRendering } from "./embedded-agent-subscribe.stream-rendering.js"; import type { SubscribeEmbeddedAgentSessionParams } from "./embedded-agent-subscribe.types.js"; import { extractToolResultMediaArtifact, filterToolResultMediaUrls, } from "./embedded-agent-tool-media.js"; import { buildToolLifecycleErrorResult } from "./embedded-agent-tool-results.js"; -import { - createThinkingTagStreamState, - stripDowngradedToolCallText, - THINKING_TAG_SCAN_RE, -} from "./embedded-agent-utils.js"; -import { mediaUrlsFromGeneratedAttachments } from "./generated-attachments.js"; -import { hasGeneratedMediaCompletionEvent } from "./internal-event-contract.js"; -import type { AgentInternalEvent } from "./internal-events.js"; import type { AgentRunTimeoutPhase } from "./run-timeout-attribution.js"; import type { AgentMessage } from "./runtime/index.js"; import { hasNonzeroUsage, normalizeUsage, type UsageLike } from "./usage.js"; -const STREAM_STRIPPED_BLOCK_TAG_NAMES = [ - "final", - "think", - "thinking", - "thought", - "antthinking", - "antml:think", - "antml:thinking", - "antml:thought", - "mm:think", - "mm:thinking", - "mm:thought", -] as const; const embeddedLog = createSubsystemLogger("agent/embedded"); function resolveEmbeddedAgentSessionLogger(messageChannel?: string) { @@ -93,198 +48,11 @@ function resolveEmbeddedAgentSessionLogger(messageChannel?: string) { return embeddedLog; } -function isPotentialTrailingBlockTagFragment(fragment: string): boolean { - if (!fragment.startsWith("<") || fragment.includes(">")) { - return false; - } - const body = fragment.toLowerCase().slice(1).trimStart().replace(/^\//, "").trimStart(); - if (!body) { - return true; - } - const namePart = body.split(/[\s/>]/, 1)[0] ?? ""; - if (!namePart) { - return true; - } - return STREAM_STRIPPED_BLOCK_TAG_NAMES.some((name) => { - return name.startsWith(namePart) || namePart === name; - }); -} - -function splitTrailingBlockTagFragment( - text: string, - isInsideCodeSpan: (index: number) => boolean, -): { text: string; pendingTagFragment?: string } { - const fragmentStart = text.lastIndexOf("<"); - if (fragmentStart === -1 || isInsideCodeSpan(fragmentStart)) { - return { text }; - } - const fragment = text.slice(fragmentStart); - if (!isPotentialTrailingBlockTagFragment(fragment)) { - return { text }; - } - return { - text: text.slice(0, fragmentStart), - pendingTagFragment: fragment, - }; -} - -function splitTrailingFenceFragment( - text: string, - startsAtLineStart: boolean, -): { text: string; pendingFenceFragment?: string } { - const lineStart = text.lastIndexOf("\n") + 1; - const line = text.slice(lineStart); - if ((!startsAtLineStart && lineStart === 0) || !/^(?: {0,3})(?:`+|~+)$/.test(line)) { - return { text }; - } - return { - text: text.slice(0, lineStart), - pendingFenceFragment: line, - }; -} - -function collectPendingMediaFromInternalEvents( - events: SubscribeEmbeddedAgentSessionParams["internalEvents"], -): { - mediaUrls: string[]; - attachments: NonNullable; - trustByUrl: Map; -} { - if (!events?.length) { - return { mediaUrls: [], attachments: [], trustByUrl: new Map() }; - } - const pending: string[] = []; - const attachments: NonNullable = []; - const indexByUrl = new Map(); - const trustedByUrl = new Map(); - for (const event of events) { - const generatedMediaEvent = hasGeneratedMediaCompletionEvent([event]); - const attachmentByUrl = new Map( - (event.attachments ?? []).flatMap((attachment) => { - const reference = normalizeOptionalString( - attachment.path ?? attachment.url ?? attachment.mediaUrl ?? attachment.filePath, - ); - return reference ? [[reference, attachment] as const] : []; - }), - ); - const mediaUrls = [ - ...(Array.isArray(event.mediaUrls) ? event.mediaUrls : []), - ...mediaUrlsFromGeneratedAttachments(event.attachments), - ]; - for (const mediaUrl of mediaUrls) { - const normalized = normalizeOptionalString(mediaUrl) ?? ""; - if (!normalized) { - continue; - } - const metadata = attachmentByUrl.get(normalized); - const existingIndex = indexByUrl.get(normalized); - if (existingIndex !== undefined) { - trustedByUrl.set(normalized, trustedByUrl.get(normalized) === true || generatedMediaEvent); - if (metadata && Object.keys(attachments[existingIndex] ?? {}).length === 0) { - attachments[existingIndex] = metadata; - } - continue; - } - indexByUrl.set(normalized, pending.length); - trustedByUrl.set(normalized, generatedMediaEvent); - pending.push(normalized); - attachments.push(metadata ?? {}); - } - } - return { mediaUrls: pending, attachments, trustByUrl: trustedByUrl }; -} - export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSessionParams) { const log = resolveEmbeddedAgentSessionLogger(params.messageChannel); - const reasoningMode = params.reasoningMode ?? "off"; - const canShowReasoning = params.thinkingLevel !== "off"; const toolResultFormat = params.toolResultFormat ?? "markdown"; const useMarkdown = toolResultFormat === "markdown"; - const initialPendingToolMedia = collectPendingMediaFromInternalEvents(params.internalEvents); - const state: EmbeddedAgentSubscribeState = { - assistantTexts: [], - toolMetas: [], - acceptedSessionSpawns: [], - toolMetaById: new Map(), - toolSummaryById: new Set(), - liveEditDiffStateById: new Map(), - itemActiveIds: new Set(), - itemStartedCount: 0, - itemCompletedCount: 0, - assistantTurnCount: 0, - lastToolError: undefined, - blockReplyBreak: params.blockReplyBreak ?? "text_end", - reasoningMode, - includeReasoning: reasoningMode === "on" && canShowReasoning, - shouldEmitPartialReplies: !(reasoningMode === "on" && !params.onBlockReply), - streamReasoning: - (params.streamReasoningInNonStreamModes === true - ? reasoningMode !== "on" - : reasoningMode === "stream") && - canShowReasoning && - typeof params.onReasoningStream === "function", - deltaBuffer: "", - thinkingTagStream: createThinkingTagStreamState(), - blockBuffer: "", - // Track if a streamed chunk opened a block (stateful across chunks). - blockState: { thinking: false, final: false, inlineCode: createInlineCodeState() }, - partialBlockState: { thinking: false, final: false, inlineCode: createInlineCodeState() }, - lastStreamedAssistant: undefined, - lastStreamedAssistantCleaned: undefined, - emittedAssistantUpdate: false, - lastStreamedReasoning: undefined, - lastBlockReplyText: undefined, - lastDeliveredBlockReplyText: undefined, - deferBlockReplyDelivery: typeof params.onBeforeTerminalDelivery === "function", - deferredBlockReplies: [], - deferredAssistantEvents: [], - toolExecutionSinceLastBlockReply: false, - reasoningStreamOpen: false, - assistantMessageIndex: 0, - lastAssistantStreamContentIndex: undefined, - lastAssistantStreamItemId: undefined, - lastAssistantTextMessageIndex: -1, - lastAssistantTextNormalized: undefined, - lastAssistantTextTrimmed: undefined, - assistantTextBaseline: 0, - suppressBlockChunks: false, // Avoid late chunk inserts after final text merge. - lastReasoningSent: undefined, - pendingAssistantUsage: undefined, - assistantUsageCommitted: false, - compactionInFlight: false, - lastCompactionTokensAfter: undefined, - pendingCompactionRetry: 0, - compactionRetryResolve: undefined, - compactionRetryReject: undefined, - compactionRetryPromise: null, - unsubscribed: false, - replayState: createEmbeddedRunReplayState(params.initialReplayState), - livenessState: "working", - hadDeterministicSideEffect: false, - pendingEventChain: null, - messagingToolSentTexts: [], - messagingToolSentTextsNormalized: [], - currentSourceMessagingToolSentTextsNormalized: [], - currentSourceMessagingToolHeldPartial: undefined, - messagingToolSentTargets: [], - heartbeatToolResponse: undefined, - messagingToolSentMediaUrls: [], - messagingToolSourceReplyPayloads: [], - messageToolOnlySourceReplyDelivered: false, - pendingMessagingTexts: new Map(), - pendingMessagingTargets: new Map(), - successfulCronAdds: 0, - pendingMessagingMediaUrls: new Map(), - pendingToolMediaUrls: initialPendingToolMedia.mediaUrls, - pendingToolMediaAttachments: initialPendingToolMedia.attachments, - pendingToolMediaTrustByUrl: initialPendingToolMedia.trustByUrl, - pendingToolAudioAsVoice: false, - hasToolMediaBlockReply: false, - visibleBlockReplyCount: 0, - pendingAssistantReplyDirectives: undefined, - deterministicApprovalPromptPending: false, - deterministicApprovalPromptSent: false, - }; + const state: EmbeddedAgentSubscribeState = createEmbeddedAgentSubscribeState(params); const usageTotals = { input: 0, output: 0, @@ -308,280 +76,16 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess const messagingToolSourceReplyPayloads = state.messagingToolSourceReplyPayloads; const pendingMessagingTexts = state.pendingMessagingTexts; const pendingMessagingTargets = state.pendingMessagingTargets; - const pendingBlockReplyTasks = new Set>(); - const pendingPartialReplyTasks = new Set>(); - const replyDirectiveAccumulator = createStreamingDirectiveAccumulator(); - const partialReplyDirectiveAccumulator = createStreamingDirectiveAccumulator(); - const shouldAllowSilentTurnText = (text: string | undefined) => - Boolean(text && isSilentReplyText(text, SILENT_REPLY_TOKEN)); - const emitAssistantStreamDataSafely = ( - delivery: EmbeddedAgentSubscribeContext["state"]["deferredAssistantEvents"][number], - ) => { - const { data } = delivery; - emitAgentEvent({ - runId: params.runId, - stream: "assistant", - data, - }); - if (params.onAgentEvent) { - runBestEffortCallback({ - label: "assistant agent event", - log, - callback: () => - params.onAgentEvent?.({ - stream: "assistant", - data, - }), - }); - } - if (delivery.emitPartialReply && params.onPartialReply && state.shouldEmitPartialReplies) { - try { - const maybeTask = params.onPartialReply(data); - if (isPromiseLike(maybeTask)) { - const task = Promise.resolve(maybeTask) - .then(() => undefined) - .catch((error: unknown) => { - log.warn(`assistant partial reply callback failed: ${String(error)}`); - }); - pendingPartialReplyTasks.add(task); - void task.finally(() => { - pendingPartialReplyTasks.delete(task); - }); - } - } catch (error) { - log.warn(`assistant partial reply callback failed: ${String(error)}`); - } - } - }; - const emitAssistantStreamData = ( - data: EmbeddedAgentSubscribeContext["state"]["deferredAssistantEvents"][number]["data"], - options?: { emitPartialReply?: boolean }, - ) => { - const delivery = { data, emitPartialReply: options?.emitPartialReply === true }; - if (state.deferBlockReplyDelivery) { - state.deferredAssistantEvents.push(delivery); - return; - } - emitAssistantStreamDataSafely(delivery); - }; - const flushDeferredAssistantEvents = () => { - if (state.deferredAssistantEvents.length === 0) { - return; - } - const deferred = state.deferredAssistantEvents.splice(0); - for (const delivery of deferred) { - emitAssistantStreamDataSafely(delivery); - } - }; - const clearDeferredAssistantEvents = () => { - state.deferredAssistantEvents.length = 0; - }; - const deferredToolMediaReplies = new WeakSet(); - const emitBlockReplySafely = ( - payload: Parameters>[0], - options?: { assistantMessageIndex?: number }, - ): boolean => { - if (!params.onBlockReply) { - return false; - } - try { - const taggedPayload = - options?.assistantMessageIndex !== undefined - ? setReplyPayloadMetadata(payload, { - assistantMessageIndex: options.assistantMessageIndex, - }) - : payload; - const assistantMessageIndex = - options?.assistantMessageIndex ?? - getReplyPayloadMetadata(taggedPayload)?.assistantMessageIndex; - const context = assistantMessageIndex === undefined ? undefined : { assistantMessageIndex }; - const maybeTask = context - ? params.onBlockReply(taggedPayload, context) - : params.onBlockReply(taggedPayload); - if (!isPromiseLike(maybeTask)) { - return true; - } - const task = Promise.resolve(maybeTask).catch((err: unknown) => { - log.warn(`block reply callback failed: ${String(err)}`); - }); - pendingBlockReplyTasks.add(task); - void task.finally(() => { - pendingBlockReplyTasks.delete(task); - }); - return true; - } catch (err) { - log.warn(`block reply callback failed: ${String(err)}`); - return false; - } - }; - const emitBlockReply = ( - payload: BlockReplyPayload, - options?: { assistantMessageIndex?: number; consumePendingToolMedia?: boolean }, - ) => { - const withAssistantDirectives = consumePendingAssistantReplyDirectivesIntoReply(state, payload); - const consumesPendingToolMedia = - options?.consumePendingToolMedia !== false && readPendingToolMediaReply(state) !== null; - const withToolMedia = - options?.consumePendingToolMedia === false - ? withAssistantDirectives - : consumePendingToolMediaIntoReply(state, withAssistantDirectives); - const assistantTranscriptMediaUrls = Array.from(new Set(payload.mediaUrls ?? [])); - const taggedPayload = - options?.assistantMessageIndex !== undefined - ? setReplyPayloadMetadata(withToolMedia, { - assistantMessageIndex: options.assistantMessageIndex, - ...(assistantTranscriptMediaUrls.length > 0 ? { assistantTranscriptMediaUrls } : {}), - }) - : withToolMedia; - if (state.deferBlockReplyDelivery) { - if (consumesPendingToolMedia) { - deferredToolMediaReplies.add(taggedPayload); - } - state.deferredBlockReplies.push(taggedPayload); - return; - } - const emitted = emitBlockReplySafely(taggedPayload, options); - if (emitted && !taggedPayload.isReasoning && hasAssistantVisibleReply(taggedPayload)) { - state.visibleBlockReplyCount += 1; - if (consumesPendingToolMedia) { - state.hasToolMediaBlockReply = true; - } - } - }; - const flushDeferredBlockReplies = () => { - if (state.deferredBlockReplies.length === 0) { - return; - } - const deferred = state.deferredBlockReplies.splice(0); - for (const payload of deferred) { - const emitted = emitBlockReplySafely(payload); - if (emitted && !payload.isReasoning && hasAssistantVisibleReply(payload)) { - state.visibleBlockReplyCount += 1; - if (deferredToolMediaReplies.has(payload)) { - state.hasToolMediaBlockReply = true; - } - } - } - }; - const clearDeferredBlockReplies = () => { - state.deferredBlockReplies.length = 0; - }; - - const resetAssistantMessageState = (nextAssistantTextBaseline: number) => { - state.deltaBuffer = ""; - state.thinkingTagStream = createThinkingTagStreamState(); - state.blockBuffer = ""; - blockChunker?.reset(); - replyDirectiveAccumulator.reset(); - partialReplyDirectiveAccumulator.reset(); - state.blockState.thinking = false; - state.blockState.final = false; - state.blockState.inlineCode = createInlineCodeState(); - state.blockState.fence = undefined; - state.blockState.reasoningInlineCode = undefined; - state.blockState.reasoningFence = undefined; - state.blockState.reasoningPendingFenceFragment = undefined; - state.blockState.finalInlineCode = undefined; - state.blockState.finalFence = undefined; - state.blockState.pendingFenceFragment = undefined; - state.blockState.pendingTagFragment = undefined; - state.partialBlockState.thinking = false; - state.partialBlockState.final = false; - state.partialBlockState.inlineCode = createInlineCodeState(); - state.partialBlockState.fence = undefined; - state.partialBlockState.reasoningInlineCode = undefined; - state.partialBlockState.reasoningFence = undefined; - state.partialBlockState.reasoningPendingFenceFragment = undefined; - state.partialBlockState.finalInlineCode = undefined; - state.partialBlockState.finalFence = undefined; - state.partialBlockState.pendingFenceFragment = undefined; - state.partialBlockState.pendingTagFragment = undefined; - state.lastStreamedAssistant = undefined; - state.lastStreamedAssistantCleaned = undefined; - state.currentSourceMessagingToolHeldPartial = undefined; - state.emittedAssistantUpdate = false; - state.lastBlockReplyText = undefined; - state.lastStreamedReasoning = undefined; - state.lastReasoningSent = undefined; - state.reasoningStreamOpen = false; - state.suppressBlockChunks = false; - state.pendingAssistantUsage = undefined; - state.assistantUsageCommitted = false; - state.assistantMessageIndex += 1; - state.lastAssistantStreamContentIndex = undefined; - state.lastAssistantStreamItemId = undefined; - state.lastAssistantTextMessageIndex = -1; - state.lastAssistantTextNormalized = undefined; - state.lastAssistantTextTrimmed = undefined; - state.assistantTextBaseline = nextAssistantTextBaseline; - state.pendingAssistantReplyDirectives = undefined; - }; - - const rememberAssistantText = (text: string) => { - state.lastAssistantTextMessageIndex = state.assistantMessageIndex; - state.lastAssistantTextTrimmed = text.trimEnd(); - const normalized = normalizeTextForComparison(text); - state.lastAssistantTextNormalized = normalized.length > 0 ? normalized : undefined; - }; - - const shouldSkipAssistantText = (text: string) => { - if (state.lastAssistantTextMessageIndex !== state.assistantMessageIndex) { - return false; - } - const trimmed = text.trimEnd(); - if (trimmed && trimmed === state.lastAssistantTextTrimmed) { - return true; - } - const normalized = normalizeTextForComparison(text); - if (normalized.length > 0 && normalized === state.lastAssistantTextNormalized) { - return true; - } - return false; - }; - - const pushAssistantText = (text: string) => { - if (!text) { - return; - } - if (params.silentExpected && !shouldAllowSilentTurnText(text)) { - return; - } - if (shouldSkipAssistantText(text)) { - return; - } - assistantTexts.push(text); - rememberAssistantText(text); - }; - - const finalizeAssistantTexts = (args: { - text: string; - addedDuringMessage: boolean; - chunkerHasBuffered: boolean; - }) => { - const { text, addedDuringMessage, chunkerHasBuffered } = args; - - // If we're not streaming block replies, ensure the final payload includes - // the final text even when interim streaming was enabled. - if (state.includeReasoning && text && !params.onBlockReply) { - if (assistantTexts.length > state.assistantTextBaseline) { - assistantTexts.splice( - state.assistantTextBaseline, - assistantTexts.length - state.assistantTextBaseline, - text, - ); - rememberAssistantText(text); - } else { - pushAssistantText(text); - } - state.suppressBlockChunks = true; - } else if (!addedDuringMessage && !chunkerHasBuffered && text) { - // Non-streaming models (no text_delta): ensure assistantTexts gets the final - // text when the chunker has nothing buffered to drain. - pushAssistantText(text); - } - - state.assistantTextBaseline = assistantTexts.length; - }; + const replyDelivery = createReplyDelivery({ params, state, log }); + const { + clearDeferredAssistantEvents, + clearDeferredBlockReplies, + emitAssistantStreamData, + emitBlockReply, + finalizeAssistantTexts, + flushDeferredAssistantEvents, + flushDeferredBlockReplies, + } = replyDelivery; // ── Messaging tool duplicate detection ────────────────────────────────────── // Track texts sent via messaging tools to suppress duplicate block replies. @@ -874,471 +378,25 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess emitToolResultMessage(toolName, message, result); }; - const stripBlockTags = ( - text: string, - stateLocal: { - thinking: boolean; - final: boolean; - inlineCode?: InlineCodeState; - fence?: FenceScanState; - reasoningInlineCode?: InlineCodeState; - reasoningFence?: FenceScanState; - reasoningPendingFenceFragment?: string; - finalInlineCode?: InlineCodeState; - finalFence?: FenceScanState; - pendingFenceFragment?: string; - pendingTagFragment?: string; - }, - options?: { final?: boolean; completeMarkdownChunk?: boolean }, - ): string => { - const input = `${stateLocal.pendingFenceFragment ?? ""}${stateLocal.pendingTagFragment ?? ""}${text}`; - stateLocal.pendingFenceFragment = undefined; - stateLocal.pendingTagFragment = undefined; - if (!input) { - return text; - } - - const { text: fenceInput, pendingFenceFragment } = options?.final - ? { text: input, pendingFenceFragment: undefined } - : options?.completeMarkdownChunk - ? { text: input, pendingFenceFragment: undefined } - : splitTrailingFenceFragment(input, stateLocal.fence?.atLineStart ?? true); - stateLocal.pendingFenceFragment = pendingFenceFragment; - if (!fenceInput) { - return ""; - } - - const inlineStateStart = stateLocal.inlineCode ?? createInlineCodeState(); - const fenceStateStart = stateLocal.fence; - const initialCodeSpans = buildCodeSpanIndex(fenceInput, inlineStateStart, fenceStateStart); - const { text: scanText, pendingTagFragment } = options?.final - ? { text: fenceInput, pendingTagFragment: undefined } - : splitTrailingBlockTagFragment(fenceInput, initialCodeSpans.isInside); - stateLocal.pendingTagFragment = pendingTagFragment; - if (!scanText) { - return ""; - } - const codeSpans = buildCodeSpanIndex(scanText, inlineStateStart, fenceStateStart); - - let processed = ""; - THINKING_TAG_SCAN_RE.lastIndex = 0; - let lastIndex = 0; - let lastCodeIndex = 0; - let inThinking = stateLocal.thinking; - // Hidden reasoning has its own code state: malformed hidden fences must not - // mark later visible text as code, but literal close tags there stay hidden. - let hiddenInlineState: InlineCodeState = stateLocal.reasoningInlineCode - ? { ...stateLocal.reasoningInlineCode } - : createInlineCodeState(); - let hiddenFenceState: FenceScanState | undefined = stateLocal.reasoningFence?.open - ? { - atLineStart: stateLocal.reasoningFence.atLineStart, - open: { ...stateLocal.reasoningFence.open }, - } - : stateLocal.reasoningFence - ? { atLineStart: stateLocal.reasoningFence.atLineStart } - : undefined; - let hiddenPendingFenceFragment = stateLocal.reasoningPendingFenceFragment; - stateLocal.reasoningPendingFenceFragment = undefined; - const advanceHiddenCodeState = (segment: string) => { - const hiddenInput = `${hiddenPendingFenceFragment ?? ""}${segment}`; - hiddenPendingFenceFragment = undefined; - if (!hiddenInput) { - return; - } - const { text: hiddenFenceInput, pendingFenceFragment: pendingFenceFragmentLocal } = - options?.final - ? { text: hiddenInput, pendingFenceFragment: undefined } - : options?.completeMarkdownChunk - ? { text: hiddenInput, pendingFenceFragment: undefined } - : splitTrailingFenceFragment(hiddenInput, hiddenFenceState?.atLineStart ?? true); - hiddenPendingFenceFragment = pendingFenceFragmentLocal; - if (!hiddenFenceInput) { - return; - } - const next = buildCodeSpanIndex(hiddenFenceInput, hiddenInlineState, hiddenFenceState); - hiddenInlineState = next.inlineState; - hiddenFenceState = next.fenceState; - }; - for (const match of scanText.matchAll(THINKING_TAG_SCAN_RE)) { - const idx = match.index ?? 0; - const isClose = match[1] === "/"; - if (inThinking) { - advanceHiddenCodeState(scanText.slice(lastCodeIndex, idx)); - } - const isInsideHiddenCode = - inThinking && (hiddenInlineState.open || Boolean(hiddenFenceState?.open)); - lastCodeIndex = idx + match[0].length; - if ((!inThinking && codeSpans.isInside(idx)) || isInsideHiddenCode) { - if (inThinking) { - advanceHiddenCodeState(match[0]); - } - continue; - } - if (!inThinking) { - if (isClose) { - const afterIndex = idx + match[0].length; - const before = scanText.slice(lastIndex, idx); - const after = scanText.slice(afterIndex); - if (hasOrphanReasoningCloseBoundary({ before, after })) { - processed = ""; - } else { - processed += before; - } - lastIndex = afterIndex; - continue; - } - processed += scanText.slice(lastIndex, idx); - hiddenInlineState = createInlineCodeState(); - hiddenFenceState = undefined; - hiddenPendingFenceFragment = undefined; - } - inThinking = !isClose; - if (!inThinking) { - hiddenInlineState = createInlineCodeState(); - hiddenFenceState = undefined; - hiddenPendingFenceFragment = undefined; - } - lastIndex = idx + match[0].length; - } - if (inThinking) { - advanceHiddenCodeState(scanText.slice(lastCodeIndex)); - } - if (!inThinking) { - processed += scanText.slice(lastIndex); - } - stateLocal.thinking = inThinking; - stateLocal.reasoningInlineCode = inThinking ? hiddenInlineState : undefined; - stateLocal.reasoningFence = inThinking ? hiddenFenceState : undefined; - stateLocal.reasoningPendingFenceFragment = inThinking ? hiddenPendingFenceFragment : undefined; - - // If enforcement is disabled, we still strip the tags themselves to prevent - // hallucinations (e.g. Minimax copying the style) from leaking, but we - // do not enforce buffering/extraction logic. - const finalCodeSpans = buildCodeSpanIndex(processed, inlineStateStart, fenceStateStart); - if (!params.enforceFinalTag) { - stateLocal.inlineCode = finalCodeSpans.inlineState; - stateLocal.fence = finalCodeSpans.fenceState; - return stripFinalTagsOutsideCodeSpans(processed, finalCodeSpans.isInside); - } - - // If enforcement is enabled, only return text that appeared inside a block. - let result = ""; - let lastFinalIndex = 0; - let inFinal = stateLocal.final; - let everInFinal = stateLocal.final; - - for (const match of findFinalTagMatches(processed)) { - const idx = match.index; - if (finalCodeSpans.isInside(idx)) { - continue; - } - const isClose = match.isClose; - const isSelfClosing = match.isSelfClosing; - - if (isSelfClosing) { - if (inFinal) { - result += processed.slice(lastFinalIndex, idx); - inFinal = false; - } else { - inFinal = true; - everInFinal = true; - } - lastFinalIndex = idx + match.text.length; - } else if (!inFinal && !isClose) { - // Found start tag. - inFinal = true; - everInFinal = true; - lastFinalIndex = idx + match.text.length; - } else if (inFinal && isClose) { - // Found end tag. - result += processed.slice(lastFinalIndex, idx); - inFinal = false; - lastFinalIndex = idx + match.text.length; - } - } - - if (inFinal) { - result += processed.slice(lastFinalIndex); - } - stateLocal.final = inFinal; - - // Strict Mode: If enforcing final tags, we MUST NOT return content unless - // we have seen a tag. Otherwise, we leak "thinking out loud" text - // (e.g. "**Locating Manulife**...") that the model emitted without tags. - if (!everInFinal) { - stateLocal.inlineCode = createInlineCodeState(); - stateLocal.fence = finalCodeSpans.fenceState; - stateLocal.finalInlineCode = undefined; - stateLocal.finalFence = undefined; - return ""; - } - - // Hardened Cleanup: Remove any remaining tags that might have been - // missed (e.g. nested tags or hallucinations) to prevent leakage. - const finalResultInlineStateStart = stateLocal.finalInlineCode ?? createInlineCodeState(); - const finalResultFenceStateStart = stateLocal.finalFence; - const resultCodeSpans = buildCodeSpanIndex( - result, - finalResultInlineStateStart, - finalResultFenceStateStart, - ); - stateLocal.inlineCode = finalCodeSpans.inlineState; - stateLocal.fence = finalCodeSpans.fenceState; - stateLocal.finalInlineCode = inFinal ? resultCodeSpans.inlineState : undefined; - stateLocal.finalFence = inFinal ? resultCodeSpans.fenceState : undefined; - return stripFinalTagsOutsideCodeSpans(result, resultCodeSpans.isInside); - }; - - const stripFinalTagsOutsideCodeSpans = (text: string, isInside: (index: number) => boolean) => { - let output = ""; - let lastIndex = 0; - for (const match of findFinalTagMatches(text)) { - const idx = match.index; - if (isInside(idx)) { - continue; - } - output += text.slice(lastIndex, idx); - lastIndex = idx + match.text.length; - } - output += text.slice(lastIndex); - return output; - }; - const hasMessageToolOnlySourceDelivery = () => - params.sourceReplyDeliveryMode === "message_tool_only" && - (state.messageToolOnlySourceReplyDelivered || - params.hasDeliveredMessageToolOnlySourceReply?.() === true || - messagingToolSourceReplyPayloads.length > 0); - - const emitBlockChunk = ( - text: string, - options?: { assistantMessageIndex?: number; final?: boolean; completeMarkdownChunk?: boolean }, - ) => { - if (state.suppressBlockChunks || params.silentExpected) { - return; - } - // Strip and blocks across chunk boundaries to avoid leaking reasoning. - // Also strip downgraded tool call text ([Tool Call: ...], [Historical context: ...], etc.). - const blockReplyText = stripDowngradedToolCallText( - stripBlockTags(text, state.blockState, { - final: options?.final === true, - completeMarkdownChunk: options?.completeMarkdownChunk === true, - }), - ).trimEnd(); - if (!blockReplyText) { - return; - } - if (blockReplyText === state.lastBlockReplyText) { - return; - } - const markBlockReplyTextHandled = () => { - state.lastBlockReplyText = blockReplyText; - state.lastDeliveredBlockReplyText = blockReplyText; - state.toolExecutionSinceLastBlockReply = false; - }; - if (hasMessageToolOnlySourceDelivery()) { - markBlockReplyTextHandled(); - return; - } - let chunk = blockReplyText; - let slicedPrefixReplay = false; - const lastDeliveredBlockReplyText = state.lastDeliveredBlockReplyText; - const blockReplySuffix = lastDeliveredBlockReplyText - ? blockReplyText.slice(lastDeliveredBlockReplyText.length) - : ""; - const prefixReplayCandidate = Boolean( - state.blockReplyBreak === "text_end" && - state.toolExecutionSinceLastBlockReply && - lastDeliveredBlockReplyText && - lastDeliveredBlockReplyText.trimEnd().endsWith(":") && - blockReplyText.length > lastDeliveredBlockReplyText.length && - blockReplyText.startsWith(lastDeliveredBlockReplyText), - ); - if (prefixReplayCandidate && !/^\s/.test(blockReplySuffix)) { - chunk = blockReplySuffix; - slicedPrefixReplay = true; - } - if (!chunk) { - return; - } - - // Only check committed (successful) messaging tool texts - checking pending texts - // is risky because if the tool fails after suppression, the user gets no response - const normalizedChunk = normalizeTextForComparison(chunk); - const normalizedReplaySuffix = prefixReplayCandidate - ? normalizeTextForComparison(blockReplySuffix.trimStart()) - : ""; - const isMessagingDuplicate = - isMessagingToolDuplicateNormalized(normalizedChunk, messagingToolSentTextsNormalized) || - (prefixReplayCandidate && - isMessagingToolDuplicateNormalized( - normalizedReplaySuffix, - messagingToolSentTextsNormalized, - )); - if (isMessagingDuplicate) { - log.debug( - `Skipping block reply - already sent via messaging tool: ${truncateUtf16Safe(chunk, 50)}...`, - ); - if (prefixReplayCandidate) { - markBlockReplyTextHandled(); - } - return; - } - - if (shouldSkipAssistantText(chunk)) { - if (slicedPrefixReplay) { - markBlockReplyTextHandled(); - } - return; - } - - if (!params.onBlockReply) { - pushAssistantText(chunk); - markBlockReplyTextHandled(); - return; - } - const splitResult = replyDirectiveAccumulator.consume(chunk); - if (!splitResult) { - if (slicedPrefixReplay) { - markBlockReplyTextHandled(); - } - return; - } - const { - text: cleanedText, - mediaUrls, - audioAsVoice, - replyToId, - replyToTag, - replyToCurrent, - } = splitResult; - if (!cleanedText && (!mediaUrls || mediaUrls.length === 0) && !audioAsVoice) { - if (slicedPrefixReplay) { - markBlockReplyTextHandled(); - } - return; - } - pushAssistantText(chunk); - emitBlockReply( - { - text: cleanedText, - mediaUrls: mediaUrls?.length ? mediaUrls : undefined, - audioAsVoice, - replyToId, - replyToTag, - replyToCurrent, - }, - { - assistantMessageIndex: options?.assistantMessageIndex ?? state.assistantMessageIndex, - consumePendingToolMedia: - options?.final === true || Boolean(mediaUrls?.length || audioAsVoice), - }, - ); - markBlockReplyTextHandled(); - }; - - const consumeReplyDirectives = (text: string, options?: { final?: boolean }) => - replyDirectiveAccumulator.consume(text, options); - const consumePartialReplyDirectives = (text: string, options?: { final?: boolean }) => - partialReplyDirectiveAccumulator.consume(text, options); - - const flushBlockReplyBuffer = (options?: { - assistantMessageIndex?: number; - final?: boolean; - }): void | Promise => { - if (!params.onBlockReply) { - return; - } - if (blockChunker?.hasBuffered()) { - if (options?.final) { - let pendingChunk: string | undefined; - blockChunker.drain({ - force: true, - emit: (text) => { - if (pendingChunk !== undefined) { - emitBlockChunk(pendingChunk, { - assistantMessageIndex: options.assistantMessageIndex, - completeMarkdownChunk: true, - }); - } - pendingChunk = text; - }, - }); - if (pendingChunk !== undefined) { - emitBlockChunk(pendingChunk, { - assistantMessageIndex: options.assistantMessageIndex, - completeMarkdownChunk: true, - final: true, - }); - } - } else { - blockChunker.drain({ force: true, emit: (text) => emitBlockChunk(text, options) }); - } - blockChunker.reset(); - } else if (state.blockBuffer.length > 0) { - emitBlockChunk(state.blockBuffer, options); - state.blockBuffer = ""; - } - if (options?.final) { - emitBlockChunk("", options); - } - if (pendingBlockReplyTasks.size === 0) { - return; - } - return (async () => { - while (pendingBlockReplyTasks.size > 0) { - await Promise.allSettled(pendingBlockReplyTasks); - } - })(); - }; - - const emitReasoningStream = (text: string) => { - if (params.silentExpected) { - return; - } - const trimmed = text.trim(); - if (!trimmed) { - return; - } - if (trimmed === state.lastStreamedReasoning) { - return; - } - // Compute delta: new text since the last emitted reasoning. - // Guard against non-prefix changes (e.g. trim altering earlier content). - const prior = state.lastStreamedReasoning ?? ""; - const delta = trimmed.startsWith(prior) ? trimmed.slice(prior.length) : trimmed; - state.lastStreamedReasoning = trimmed; - - // Emit-always: the thinking stream always reaches the bus and session - // archive. /reasoning (streamReasoning) gates only the rendering hook - // below; display surfaces (TUI showThinking, webchat isReasoning drops) - // gate presentation on their side. - emitAgentEvent({ - runId: params.runId, - stream: "thinking", - data: { - text: trimmed, - delta, - }, - }); - - // Message-tool-only delivery makes later reasoning private: once the - // user-facing reply has gone out via the message tool, the channel shows - // only what was explicitly sent, so trailing reasoning must stay out of the - // render hook — uniformly, whether the thinking block rode in on a tool call - // or arrived on its own. It still reaches the bus/archive above. - if (state.streamReasoning && !hasMessageToolOnlySourceDelivery() && params.onReasoningStream) { - runBestEffortCallback({ - label: "reasoning stream", - log, - callback: () => - params.onReasoningStream?.({ - text: trimmed, - ...(state.reasoningMode === "stream" ? {} : { requiresReasoningProgressOptIn: true }), - }), - }); - } - }; + const streamRendering = createStreamRendering({ + params, + state, + log, + blockChunker, + emitBlockReply: replyDelivery.emitBlockReply, + pendingBlockReplyTasks: replyDelivery.pendingBlockReplyTasks, + pushAssistantText: replyDelivery.pushAssistantText, + shouldSkipAssistantText: replyDelivery.shouldSkipAssistantText, + }); + const { + consumePartialReplyDirectives, + consumeReplyDirectives, + emitBlockChunk, + emitReasoningStream, + flushBlockReplyBuffer, + resetAssistantMessageState, + stripBlockTags, + } = streamRendering; const resetForCompactionRetry = () => { state.hadDeterministicSideEffect = @@ -1607,16 +665,7 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess getCompactionCount: () => compactionCount, getLastCompactionTokensAfter: () => state.lastCompactionTokensAfter, getAssistantTurnCount: () => state.assistantTurnCount, - waitForPendingEvents: async () => { - // Partial presentation stays concurrent with provider events, but terminal - // settlement must observe callbacks launched while the event chain drains. - while (state.pendingEventChain || pendingPartialReplyTasks.size > 0) { - await Promise.allSettled([ - ...(state.pendingEventChain ? [state.pendingEventChain] : []), - ...pendingPartialReplyTasks, - ]); - } - }, + waitForPendingEvents: replyDelivery.waitForPendingEvents, getItemLifecycle: () => ({ startedCount: state.itemStartedCount, completedCount: state.itemCompletedCount, @@ -1652,4 +701,3 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess }, }; } -/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ diff --git a/src/gateway/server-methods/chat-send-agent-dispatch.ts b/src/gateway/server-methods/chat-send-agent-dispatch.ts new file mode 100644 index 000000000000..a29d90d7ca5b --- /dev/null +++ b/src/gateway/server-methods/chat-send-agent-dispatch.ts @@ -0,0 +1,507 @@ +// Detached chat.send dispatch owns runtime delivery, post-dispatch persistence, and terminalization. +import { performance } from "node:perf_hooks"; +import { + GATEWAY_CLIENT_CAPS, + hasGatewayClientCap, +} from "../../../packages/gateway-protocol/src/client-info.js"; +import { ErrorCodes, errorShape } from "../../../packages/gateway-protocol/src/index.js"; +import { resolveDefaultAgentId } from "../../agents/agent-scope.js"; +import { resolveProviderIdForAuth } from "../../agents/provider-auth-aliases.js"; +import { dispatchInboundMessageWithProjectedDispatcher } from "../../auto-reply/dispatch.js"; +import type { ReplyMessageInjectionAttempt } from "../../auto-reply/reply/reply-run-registry.js"; +import { measureDiagnosticsTimelineSpan } from "../../infra/diagnostics-timeline.js"; +import { retainGatewayRootWorkAdmissionContinuation } from "../../process/gateway-work-admission.js"; +import { isOperatorUiClient } from "../../utils/message-channel.js"; +import { setGatewayDedupeEntry } from "../agent-turn/agent-job.js"; +import { updateChatRunProvider } from "../chat-abort.js"; +import type { ChatRunTiming } from "../server-chat-state.js"; +import { broadcastChatError, broadcastChatFinal } from "./chat-broadcast.js"; +import type { AdmittedChatSend } from "./chat-send-admission.js"; +import type { prepareChatSendAttachments } from "./chat-send-attachments.js"; +import { resolveWebchatPromptCacheKey } from "./chat-send-background.js"; +import { createChatSendDispatchErrorLifecycle } from "./chat-send-dispatch-errors.js"; +import type { ChatSendExternalAuthorityAdmission } from "./chat-send-external-authority-contract.js"; +import { finalizeAcceptedChatSendMessageInjection } from "./chat-send-message-injection.js"; +import { finalizeChatSendNonAgentReplies } from "./chat-send-nonagent-finalization.js"; +import { + applyChatSendReplyContextFields, + type ChatSendReplyContextFields, +} from "./chat-send-reply-context.js"; +import { createChatSendReplyDispatch } from "./chat-send-reply-dispatch.js"; +import type { NormalizedChatSendRequest } from "./chat-send-request.js"; +import type { PreparedChatSendSession } from "./chat-send-session.js"; +import { finalizeChatSendSourceReplies } from "./chat-send-source-finalization.js"; +import { createChatSendTurnAdoptionLifecycle } from "./chat-send-turn-adoption.js"; +import { applyChatSendManagedMedia, type prepareChatSendUserTurn } from "./chat-send-user-turn.js"; +import { + emitOperatorChatSendServerTiming, + roundedChatSendTimingMs, + type ChatSendServerTimingPhase, +} from "./chat-server-timing.js"; +import type { createGatewayChatUserTurnController } from "./chat-user-turn-recorder.js"; +import { emitSessionsChanged } from "./session-change-event.js"; +import type { GatewayRequestHandlerOptions } from "./types.js"; + +type PreparedChatSendAttachments = Extract< + Awaited>, + { ok: true } +>["value"]; + +type StartChatDispatchParams = { + admissionStartedAt: number; + admission: AdmittedChatSend; + attachments: PreparedChatSendAttachments; + client: GatewayRequestHandlerOptions["client"]; + context: GatewayRequestHandlerOptions["context"]; + cronCreatorAuthority: ReturnType; + externalAuthorityAdmission: ChatSendExternalAuthorityAdmission | undefined; + injection: { + beginCapturedMessageInjection: () => ReplyMessageInjectionAttempt | undefined; + messageInjectionAttempt: ReplyMessageInjectionAttempt | undefined; + preAckReplyContextPromise: Promise | undefined; + replyContextFieldsPromise: Promise | undefined; + }; + request: NormalizedChatSendRequest; + session: PreparedChatSendSession; + terminalizeRestartSafeAdmission: (terminalState: { + retryable: boolean; + status: "failed" | "killed"; + }) => Promise; + timing: { + chatSendAckedAtMs: number; + chatSendTiming: ChatRunTiming | undefined; + }; + turn: ReturnType; + userTurn: ReturnType; +}; + +export function startChatDispatch(params: StartChatDispatchParams): void { + const { + admissionStartedAt, + admission, + attachments, + client, + context, + cronCreatorAuthority, + externalAuthorityAdmission, + injection, + request, + session, + terminalizeRestartSafeAdmission, + timing, + turn, + userTurn, + } = params; + const { imageOrder } = attachments; + const { + activeRunAbort, + admittedSessionId, + chatSendTraceAttributes, + gatewayWorkAdmission, + messageInjectionTarget, + retainGatewayWorkAdmission, + restartSafeAdmission, + setReleaseGatewayRootContinuation, + } = admission; + const { + activeRunScopeKey, + agentId, + backingSessionId, + cfg, + clientRunId, + entry, + expectedLeafEntryId, + expectedRunId, + requestedSessionId, + resolvedSessionModel, + selectedAgent, + sessionKey, + } = session; + const { chatSendReceivedAtMs, clientInfo, p, reconnectResumeRequested, supportsTaskSuggestions } = + request; + const { + accountId, + ctx, + isInternalTextSlashCommandTurn, + pluginBoundMediaPromise, + queuedFollowupOwnerKey, + replyOptionImages, + replyOptionMedia, + } = turn; + const { + persist: persistGatewayUserTurnTranscript, + persistBestEffort: persistGatewayUserTurnTranscriptBestEffort, + recorder: userTurnRecorder, + } = userTurn; + const { beginCapturedMessageInjection, preAckReplyContextPromise, replyContextFieldsPromise } = + injection; + let { messageInjectionAttempt } = injection; + const { chatSendAckedAtMs, chatSendTiming } = timing; + + let agentRunStarted = false; + const replyDispatch = createChatSendReplyDispatch({ + accountId, + isAgentRunStarted: () => agentRunStarted, + logGateway: context.logGateway, + session, + userTurnRecorder, + }); + const queuedFollowup = createChatSendTurnAdoptionLifecycle({ + chatQueuedTurns: context.chatQueuedTurns, + runId: clientRunId, + controller: activeRunAbort.controller, + sessionId: backingSessionId ?? clientRunId, + sessionKey, + agentId: selectedAgent.agentId, + ownerConnId: client?.connId, + ownerDeviceId: client?.connect?.device?.id, + ownerKey: queuedFollowupOwnerKey, + ...(expectedLeafEntryId !== undefined ? { originatingLeafEntryId: expectedLeafEntryId } : {}), + hasCronCreatorAuthority: cronCreatorAuthority !== undefined, + retainWorkAdmission: retainGatewayWorkAdmission, + }); + const dispatchErrorLifecycle = createChatSendDispatchErrorLifecycle({ + admission, + context, + isQueuedFollowupEnqueued: queuedFollowup.isEnqueued, + persistUserTurnTranscript: persistGatewayUserTurnTranscript, + session, + terminalizeRestartSafeAdmission, + userTurnRecorder, + }); + const emitServerTiming = ( + phase: ChatSendServerTimingPhase, + extra?: Record, + dispatchStartedAtMs?: number, + ) => { + emitOperatorChatSendServerTiming({ + context, + client, + phase, + runId: clientRunId, + sessionKey, + agentId, + receivedAtMs: chatSendReceivedAtMs, + ackedAtMs: chatSendAckedAtMs, + dispatchStartedAtMs, + extra, + }); + }; + const dispatchStartedAtMs = performance.now(); + if (chatSendTiming) { + chatSendTiming.dispatchStartedAtMs = dispatchStartedAtMs; + } + emitServerTiming("dispatch-started"); + let firstAssistantServerTimingEmitted = false; + let acceptedMessageInjection = false; + const emitFirstAssistantServerTiming = () => { + if (firstAssistantServerTimingEmitted || chatSendTiming?.firstAssistantEventSent) { + return; + } + firstAssistantServerTimingEmitted = true; + if (chatSendTiming) { + chatSendTiming.firstAssistantEventSent = true; + } + emitServerTiming("first-assistant-event", undefined, dispatchStartedAtMs); + }; + // Reserve the detached dispatch before this request releases its root. Otherwise + // its inherited ALS context becomes retired and rejects queued/session work. + setReleaseGatewayRootContinuation(retainGatewayRootWorkAdmissionContinuation() ?? undefined); + void replyDispatch + .runAgentMediaTranscript(gatewayWorkAdmission, () => + measureDiagnosticsTimelineSpan( + "gateway.chat_send.dispatch_inbound", + async () => { + if (replyContextFieldsPromise && !preAckReplyContextPromise) { + applyChatSendReplyContextFields(ctx, await replyContextFieldsPromise); + messageInjectionAttempt = beginCapturedMessageInjection(); + } + if (messageInjectionAttempt) { + const outcome = await messageInjectionAttempt.outcome; + if (outcome.status === "accepted") { + acceptedMessageInjection = true; + await finalizeAcceptedChatSendMessageInjection({ + context, + ctx, + outcome, + persistUserTurnTranscriptBestEffort: persistGatewayUserTurnTranscriptBestEffort, + session, + startedAt: admissionStartedAt, + target: messageInjectionTarget!, + targetRunId: messageInjectionAttempt.targetRunId, + }); + return { + queuedFinal: false, + counts: { tool: 0, block: 0, final: 0 }, + }; + } + } + applyChatSendManagedMedia(ctx, await pluginBoundMediaPromise); + const dispatchInbound = () => + dispatchInboundMessageWithProjectedDispatcher({ + ctx, + cfg, + dispatcherOptions: replyDispatch.dispatcherOptions, + onSessionMetadataChanges: (changes) => + changes.forEach((change) => emitSessionsChanged(context, change)), + replyOptions: { + runId: clientRunId, + ...(cronCreatorAuthority + ? { cronCreatorAuthorityCapability: cronCreatorAuthority } + : {}), + ...(isOperatorUiClient(clientInfo) + ? { + promptCacheKey: resolveWebchatPromptCacheKey({ + agentId, + provider: resolvedSessionModel.provider, + model: resolvedSessionModel.model, + sessionKey: activeRunScopeKey, + }), + } + : {}), + ...(supportsTaskSuggestions + ? { taskSuggestionDeliveryMode: "gateway" as const } + : {}), + requestedSessionId, + ...(restartSafeAdmission + ? { + expectedExistingSessionId: admittedSessionId, + pinExpectedExistingSession: true, + } + : entry?.sessionId + ? { expectedExistingSessionId: entry.sessionId } + : {}), + resumeRequestedSession: reconnectResumeRequested, + onSessionPrepared: (binding) => { + if (binding.sessionKey === sessionKey) { + userTurn.setAcceptedSessionId(binding.sessionId); + } + }, + abortSignal: activeRunAbort.controller.signal, + // Keep a Gateway-owned cancel identity after this chat.send + // terminalizes while the prompt waits in followup/collect queue. + onFollowupQueueDisposition: (reason) => { + context.logGateway.info("chat queue turn intentionally skipped", { + runId: clientRunId, + sessionKey, + outcome: "skipped", + reason, + }); + }, + turnAdoptionLifecycle: queuedFollowup.lifecycle, + images: replyOptionImages, + imageOrder: imageOrder.length > 0 ? imageOrder : undefined, + media: replyOptionMedia, + thinkingLevelOverride: p.thinking, + fastModeOverride: p.fastMode, + queueModeOverride: p.queueMode, + userTurnTranscriptRecorder: userTurnRecorder, + ...((messageInjectionTarget && !isInternalTextSlashCommandTurn) || + (p.queueMode === "steer" && expectedRunId !== undefined) + ? { messageInjectionAttempted: true as const } + : {}), + ...(restartSafeAdmission ? { suppressNextUserMessagePersistence: true } : {}), + fastModeAutoOnSecondsOverride: p.fastAutoOnSeconds, + onAgentRunStart: (runId) => { + agentRunStarted = replyDispatch.captureAgentTranscriptStart(); + emitServerTiming( + "agent-run-started", + runId !== clientRunId ? { agentRunId: runId } : undefined, + dispatchStartedAtMs, + ); + const connId = typeof client?.connId === "string" ? client.connId : undefined; + const wantsToolEvents = hasGatewayClientCap( + client?.connect?.caps, + GATEWAY_CLIENT_CAPS.TOOL_EVENTS, + ); + if (connId && wantsToolEvents) { + context.registerToolEventRecipient(runId, connId); + // Register for any other active runs *in the same session* so + // late-joining clients (e.g. page refresh mid-response) receive + // in-progress tool events without leaking cross-session data. + const defaultAgentId = resolveDefaultAgentId(cfg); + const selectedGlobalAgentId = + sessionKey === "global" + ? (selectedAgent.agentId ?? defaultAgentId) + : undefined; + for (const [activeRunId, active] of context.chatAbortControllers) { + const activeGlobalAgentId = + active.sessionKey === "global" + ? (active.agentId ?? defaultAgentId) + : undefined; + const sameSelectedGlobalAgent = + sessionKey === "global" && + selectedGlobalAgentId !== undefined && + activeGlobalAgentId === selectedGlobalAgentId; + const sameSession = + active.sessionKey === sessionKey && + (sessionKey !== "global" || sameSelectedGlobalAgent); + if (activeRunId !== runId && sameSession) { + context.registerToolEventRecipient(activeRunId, connId); + } + } + } + }, + onModelSelected: (modelSelection) => { + updateChatRunProvider(context.chatAbortControllers, { + runId: clientRunId, + providerId: modelSelection.provider, + authProviderId: resolveProviderIdForAuth(modelSelection.provider, { + config: cfg, + }), + }); + replyDispatch.onModelSelected(modelSelection); + emitServerTiming( + "model-selected", + { + provider: modelSelection.provider, + model: modelSelection.model, + }, + dispatchStartedAtMs, + ); + }, + }, + }); + const dispatchResult = await (cronCreatorAuthority && externalAuthorityAdmission + ? externalAuthorityAdmission.run( + cronCreatorAuthority, + dispatchInbound, + activeRunAbort.controller.signal, + ) + : dispatchInbound()); + if (dispatchResult.beforeAgentRunBlocked === true) { + userTurnRecorder.markBlocked(); + } + return dispatchResult; + }, + { + phase: "agent-turn", + config: cfg, + attributes: chatSendTraceAttributes, + }, + ), + ) + .then(async () => { + if (acceptedMessageInjection) { + return; + } + emitServerTiming("dispatch-completed", undefined, dispatchStartedAtMs); + const postDispatchStartedAtMs = performance.now(); + await measureDiagnosticsTimelineSpan( + "gateway.chat_send.post_dispatch", + async () => { + const returnedAgentErrorPayloads = agentRunStarted + ? replyDispatch.deliveredReplies + .map((entryInner) => entryInner.payload) + .filter((payload) => payload.isError) + : []; + const returnedAgentErrorMessage = + returnedAgentErrorPayloads + .map((payload) => payload.text?.trim()) + .filter((text): text is string => Boolean(text)) + .join(" | ") || undefined; + if ( + agentRunStarted && + returnedAgentErrorPayloads.length > 0 && + !userTurnRecorder.hasPersisted() && + !userTurnRecorder.isBlocked() + ) { + await persistGatewayUserTurnTranscriptBestEffort(); + } + if ( + agentRunStarted && + returnedAgentErrorPayloads.length === 0 && + !userTurnRecorder.hasPersisted() && + !userTurnRecorder.isBlocked() && + userTurnRecorder.hasRuntimePersistencePending() + ) { + await persistGatewayUserTurnTranscriptBestEffort(); + } + let broadcastedSourceReplyFinal = false; + // Agent runs persist model-visible turns through SessionManager; this dispatcher owns + // live delivery. Mirroring agent finals would duplicate normal assistant turns. The + // non-agent branch has no runtime-owned turn, so it appends one before broadcasting. + if (!agentRunStarted && !queuedFollowup.isEnqueued()) { + await finalizeChatSendNonAgentReplies({ + accountId, + context, + deliveredReplies: replyDispatch.deliveredReplies, + emitFirstAssistantServerTiming, + foldCommandBlocks: isInternalTextSlashCommandTurn, + persistUserTurnTranscript: persistGatewayUserTurnTranscriptBestEffort, + session, + suppressReplies: replyDispatch.hasAppendedWebchatAgentMedia(), + }); + } else { + broadcastedSourceReplyFinal = await finalizeChatSendSourceReplies({ + accountId, + context, + deliveredReplies: replyDispatch.deliveredReplies, + emitFirstAssistantServerTiming, + hasReturnedAgentErrorPayloads: returnedAgentErrorPayloads.length > 0, + session, + }); + } + const shouldBroadcastAgentError = + returnedAgentErrorPayloads.length > 0 && !broadcastedSourceReplyFinal; + if (shouldBroadcastAgentError) { + broadcastChatError({ + context, + runId: clientRunId, + sessionKey, + agentId, + errorMessage: returnedAgentErrorMessage, + }); + } + if (!context.chatRunState.hasAbortMarker(clientRunId)) { + const returnedAgentError = shouldBroadcastAgentError + ? errorShape( + ErrorCodes.UNAVAILABLE, + returnedAgentErrorMessage ?? "agent returned an error payload", + ) + : undefined; + setGatewayDedupeEntry({ + dedupe: context.dedupe, + key: `chat:${clientRunId}`, + entry: { + ts: Date.now(), + ok: !shouldBroadcastAgentError, + payload: shouldBroadcastAgentError + ? { + runId: clientRunId, + status: "error" as const, + summary: returnedAgentErrorMessage ?? "agent returned an error payload", + } + : { runId: clientRunId, status: "ok" as const }, + ...(returnedAgentError ? { error: returnedAgentError } : {}), + }, + }); + } + }, + { + phase: "agent-turn", + config: cfg, + attributes: chatSendTraceAttributes, + }, + ); + emitServerTiming( + "post-dispatch-completed", + { + postDispatchMs: roundedChatSendTimingMs(performance.now() - postDispatchStartedAtMs), + }, + dispatchStartedAtMs, + ); + if (queuedFollowup.isEnqueued() && !context.chatRunState.hasAbortMarker(clientRunId)) { + // Successful queue admission ends this client run. The later + // aggregate/followup owns its own run id. + broadcastChatFinal({ + context, + runId: clientRunId, + sessionKey, + agentId, + }); + } + }) + .catch(dispatchErrorLifecycle.handleError) + .finally(dispatchErrorLifecycle.finalize); +} diff --git a/src/gateway/server-methods/chat-send-handler.ts b/src/gateway/server-methods/chat-send-handler.ts index 0b829d886096..cfc02b237f06 100644 --- a/src/gateway/server-methods/chat-send-handler.ts +++ b/src/gateway/server-methods/chat-send-handler.ts @@ -1,57 +1,28 @@ -// chat.send owns admission, ACK timing, detached dispatch, and terminalization. +// chat.send owns admission, ACK timing, and detached dispatch handoff. import { performance } from "node:perf_hooks"; -import { - GATEWAY_CLIENT_CAPS, - hasGatewayClientCap, -} from "../../../packages/gateway-protocol/src/client-info.js"; -import { ErrorCodes, errorShape } from "../../../packages/gateway-protocol/src/index.js"; -import { resolveDefaultAgentId } from "../../agents/agent-scope.js"; -import { resolveProviderIdForAuth } from "../../agents/provider-auth-aliases.js"; import { createAgentRunRestartAbortError } from "../../agents/run-termination.js"; -import { dispatchInboundMessageWithProjectedDispatcher } from "../../auto-reply/dispatch.js"; import { getAgentEventLifecycleGeneration } from "../../infra/agent-events.js"; -import { - emitDiagnosticsTimelineEvent, - measureDiagnosticsTimelineSpan, -} from "../../infra/diagnostics-timeline.js"; -import { retainGatewayRootWorkAdmissionContinuation } from "../../process/gateway-work-admission.js"; -import { isOperatorUiClient } from "../../utils/message-channel.js"; -import { setGatewayDedupeEntry } from "../agent-turn/agent-job.js"; -import { updateChatRunProvider } from "../chat-abort.js"; +import { emitDiagnosticsTimelineEvent } from "../../infra/diagnostics-timeline.js"; import type { ChatRunTiming } from "../server-chat-state.js"; -import { broadcastChatError, broadcastChatFinal } from "./chat-broadcast.js"; import { terminalizeRestartSafeChatAdmission } from "./chat-restart-recovery.js"; +import { startChatDispatch } from "./chat-send-agent-dispatch.js"; import { prepareChatSendAttachments } from "./chat-send-attachments.js"; -import { - resolveWebchatPromptCacheKey, - scheduleChatDashboardSessionTitle, -} from "./chat-send-background.js"; -import { - createChatSendDispatchErrorLifecycle, - handleChatSendSetupError, -} from "./chat-send-dispatch-errors.js"; +import { scheduleChatDashboardSessionTitle } from "./chat-send-background.js"; +import { handleChatSendSetupError } from "./chat-send-dispatch-errors.js"; import type { ChatSendExternalAuthorityAdmission } from "./chat-send-external-authority-contract.js"; import { createChatSendMessageInjectionStarter, - finalizeAcceptedChatSendMessageInjection, settleChatSendPreAckMessageInjection, } from "./chat-send-message-injection.js"; -import { finalizeChatSendNonAgentReplies } from "./chat-send-nonagent-finalization.js"; import { applyChatSendReplyContextFields } from "./chat-send-reply-context.js"; -import { createChatSendReplyDispatch } from "./chat-send-reply-dispatch.js"; import { prepareAndAdmitChatSend } from "./chat-send-setup.js"; -import { finalizeChatSendSourceReplies } from "./chat-send-source-finalization.js"; -import { createChatSendTurnAdoptionLifecycle } from "./chat-send-turn-adoption.js"; -import { applyChatSendManagedMedia, prepareChatSendUserTurn } from "./chat-send-user-turn.js"; +import { prepareChatSendUserTurn } from "./chat-send-user-turn.js"; import { chatSendAckServerTimingAttributes, - emitOperatorChatSendServerTiming, roundedChatSendTimingMs, shouldIncludeChatSendAckServerTiming, - type ChatSendServerTimingPhase, } from "./chat-server-timing.js"; import { createGatewayChatUserTurnController } from "./chat-user-turn-recorder.js"; -import { emitSessionsChanged } from "./session-change-event.js"; import type { GatewayRequestHandlerOptions } from "./types.js"; export async function handleChatSend( @@ -67,14 +38,8 @@ export async function handleChatSend( return; } const { normalizedRequest, preparedSession, admitted } = setup; - const { - chatSendReceivedAtMs, - clientInfo, - supportsTaskSuggestions, - p, - systemInputProvenance, - reconnectResumeRequested, - } = normalizedRequest.value; + const { chatSendReceivedAtMs, clientInfo, p, systemInputProvenance, reconnectResumeRequested } = + normalizedRequest.value; const { clientRunId, sessionLoadOptions, @@ -85,25 +50,16 @@ export async function handleChatSend( sessionKey, sessionRoutingChanged, selectedAgent, - requestedSessionId, - backingSessionId, agentId, - activeRunScopeKey, - expectedLeafEntryId, - expectedRunId, - resolvedSessionModel, } = preparedSession.value; const { activeRunAbort, admittedSessionId, chatSendTraceAttributes, finishAbortedChatSend, - gatewayWorkAdmission, lifecycleGeneration, messageInjectionTarget, - retainGatewayWorkAdmission, restartSafeAdmission, - setReleaseGatewayRootContinuation, } = admitted.value; const preparedAttachments = await prepareChatSendAttachments({ request: normalizedRequest.value, @@ -166,7 +122,6 @@ export async function handleChatSend( }); const { persist: persistGatewayUserTurnTranscript, - persistBestEffort: persistGatewayUserTurnTranscriptBestEffort, recorder: userTurnRecorder, replyContextFieldsPromise, } = userTurn; @@ -217,15 +172,7 @@ export async function handleChatSend( logGateway: context.logGateway, userTurn, }); - const { - accountId, - ctx, - isInternalTextSlashCommandTurn, - pluginBoundMediaPromise, - queuedFollowupOwnerKey, - replyOptionImages, - replyOptionMedia, - } = preparedUserTurn; + const { ctx, isInternalTextSlashCommandTurn } = preparedUserTurn; const beginCapturedMessageInjection = createChatSendMessageInjectionStarter({ target: messageInjectionTarget, request: normalizedRequest.value, @@ -314,372 +261,30 @@ export async function handleChatSend( sessionLoadOptions, storePath, }); - let agentRunStarted = false; - const replyDispatch = createChatSendReplyDispatch({ - accountId, - isAgentRunStarted: () => agentRunStarted, - logGateway: context.logGateway, - session: preparedSession.value, - userTurnRecorder, - }); - const queuedFollowup = createChatSendTurnAdoptionLifecycle({ - chatQueuedTurns: context.chatQueuedTurns, - runId: clientRunId, - controller: activeRunAbort.controller, - sessionId: backingSessionId ?? clientRunId, - sessionKey, - agentId: selectedAgent.agentId, - ownerConnId: client?.connId, - ownerDeviceId: client?.connect?.device?.id, - ownerKey: queuedFollowupOwnerKey, - ...(expectedLeafEntryId !== undefined ? { originatingLeafEntryId: expectedLeafEntryId } : {}), - hasCronCreatorAuthority: cronCreatorAuthority !== undefined, - retainWorkAdmission: retainGatewayWorkAdmission, - }); - const dispatchErrorLifecycle = createChatSendDispatchErrorLifecycle({ + startChatDispatch({ + admissionStartedAt, admission: admitted.value, + attachments: preparedAttachments.value, + client, context, - isQueuedFollowupEnqueued: queuedFollowup.isEnqueued, - persistUserTurnTranscript: persistGatewayUserTurnTranscript, + cronCreatorAuthority, + externalAuthorityAdmission, + injection: { + beginCapturedMessageInjection, + messageInjectionAttempt, + preAckReplyContextPromise, + replyContextFieldsPromise, + }, + request: normalizedRequest.value, session: preparedSession.value, terminalizeRestartSafeAdmission, - userTurnRecorder, + timing: { + chatSendAckedAtMs, + chatSendTiming, + }, + turn: preparedUserTurn, + userTurn, }); - const emitServerTiming = ( - phase: ChatSendServerTimingPhase, - extra?: Record, - dispatchStartedAtMs?: number, - ) => { - emitOperatorChatSendServerTiming({ - context, - client, - phase, - runId: clientRunId, - sessionKey, - agentId, - receivedAtMs: chatSendReceivedAtMs, - ackedAtMs: chatSendAckedAtMs, - dispatchStartedAtMs, - extra, - }); - }; - const dispatchStartedAtMs = performance.now(); - if (chatSendTiming) { - chatSendTiming.dispatchStartedAtMs = dispatchStartedAtMs; - } - emitServerTiming("dispatch-started"); - let firstAssistantServerTimingEmitted = false; - let acceptedMessageInjection = false; - const emitFirstAssistantServerTiming = () => { - if (firstAssistantServerTimingEmitted || chatSendTiming?.firstAssistantEventSent) { - return; - } - firstAssistantServerTimingEmitted = true; - if (chatSendTiming) { - chatSendTiming.firstAssistantEventSent = true; - } - emitServerTiming("first-assistant-event", undefined, dispatchStartedAtMs); - }; - // Reserve the detached dispatch before this request releases its root. Otherwise - // its inherited ALS context becomes retired and rejects queued/session work. - setReleaseGatewayRootContinuation(retainGatewayRootWorkAdmissionContinuation() ?? undefined); - void replyDispatch - .runAgentMediaTranscript(gatewayWorkAdmission, () => - measureDiagnosticsTimelineSpan( - "gateway.chat_send.dispatch_inbound", - async () => { - if (replyContextFieldsPromise && !preAckReplyContextPromise) { - applyChatSendReplyContextFields(ctx, await replyContextFieldsPromise); - messageInjectionAttempt = beginCapturedMessageInjection(); - } - if (messageInjectionAttempt) { - const outcome = await messageInjectionAttempt.outcome; - if (outcome.status === "accepted") { - acceptedMessageInjection = true; - await finalizeAcceptedChatSendMessageInjection({ - context, - ctx, - outcome, - persistUserTurnTranscriptBestEffort: persistGatewayUserTurnTranscriptBestEffort, - session: preparedSession.value, - startedAt: admissionStartedAt, - target: messageInjectionTarget!, - targetRunId: messageInjectionAttempt.targetRunId, - }); - return { - queuedFinal: false, - counts: { tool: 0, block: 0, final: 0 }, - }; - } - } - applyChatSendManagedMedia(ctx, await pluginBoundMediaPromise); - const dispatchInbound = () => - dispatchInboundMessageWithProjectedDispatcher({ - ctx, - cfg, - dispatcherOptions: replyDispatch.dispatcherOptions, - onSessionMetadataChanges: (changes) => - changes.forEach((change) => emitSessionsChanged(context, change)), - replyOptions: { - runId: clientRunId, - ...(cronCreatorAuthority - ? { cronCreatorAuthorityCapability: cronCreatorAuthority } - : {}), - ...(isOperatorUiClient(clientInfo) - ? { - promptCacheKey: resolveWebchatPromptCacheKey({ - agentId, - provider: resolvedSessionModel.provider, - model: resolvedSessionModel.model, - sessionKey: activeRunScopeKey, - }), - } - : {}), - ...(supportsTaskSuggestions - ? { taskSuggestionDeliveryMode: "gateway" as const } - : {}), - requestedSessionId, - ...(restartSafeAdmission - ? { - expectedExistingSessionId: admittedSessionId, - pinExpectedExistingSession: true, - } - : entry?.sessionId - ? { expectedExistingSessionId: entry.sessionId } - : {}), - resumeRequestedSession: reconnectResumeRequested, - onSessionPrepared: (binding) => { - if (binding.sessionKey === sessionKey) { - userTurn.setAcceptedSessionId(binding.sessionId); - } - }, - abortSignal: activeRunAbort.controller.signal, - // Keep a Gateway-owned cancel identity after this chat.send - // terminalizes while the prompt waits in followup/collect queue. - onFollowupQueueDisposition: (reason) => { - context.logGateway.info("chat queue turn intentionally skipped", { - runId: clientRunId, - sessionKey, - outcome: "skipped", - reason, - }); - }, - turnAdoptionLifecycle: queuedFollowup.lifecycle, - images: replyOptionImages, - imageOrder: imageOrder.length > 0 ? imageOrder : undefined, - media: replyOptionMedia, - thinkingLevelOverride: p.thinking, - fastModeOverride: p.fastMode, - queueModeOverride: p.queueMode, - userTurnTranscriptRecorder: userTurnRecorder, - ...((messageInjectionTarget && !isInternalTextSlashCommandTurn) || - (p.queueMode === "steer" && expectedRunId !== undefined) - ? { messageInjectionAttempted: true as const } - : {}), - ...(restartSafeAdmission ? { suppressNextUserMessagePersistence: true } : {}), - fastModeAutoOnSecondsOverride: p.fastAutoOnSeconds, - onAgentRunStart: (runId) => { - agentRunStarted = replyDispatch.captureAgentTranscriptStart(); - emitServerTiming( - "agent-run-started", - runId !== clientRunId ? { agentRunId: runId } : undefined, - dispatchStartedAtMs, - ); - const connId = typeof client?.connId === "string" ? client.connId : undefined; - const wantsToolEvents = hasGatewayClientCap( - client?.connect?.caps, - GATEWAY_CLIENT_CAPS.TOOL_EVENTS, - ); - if (connId && wantsToolEvents) { - context.registerToolEventRecipient(runId, connId); - // Register for any other active runs *in the same session* so - // late-joining clients (e.g. page refresh mid-response) receive - // in-progress tool events without leaking cross-session data. - const defaultAgentId = resolveDefaultAgentId(cfg); - const selectedGlobalAgentId = - sessionKey === "global" - ? (selectedAgent.agentId ?? defaultAgentId) - : undefined; - for (const [activeRunId, active] of context.chatAbortControllers) { - const activeGlobalAgentId = - active.sessionKey === "global" - ? (active.agentId ?? defaultAgentId) - : undefined; - const sameSelectedGlobalAgent = - sessionKey === "global" && - selectedGlobalAgentId !== undefined && - activeGlobalAgentId === selectedGlobalAgentId; - const sameSession = - active.sessionKey === sessionKey && - (sessionKey !== "global" || sameSelectedGlobalAgent); - if (activeRunId !== runId && sameSession) { - context.registerToolEventRecipient(activeRunId, connId); - } - } - } - }, - onModelSelected: (modelSelection) => { - updateChatRunProvider(context.chatAbortControllers, { - runId: clientRunId, - providerId: modelSelection.provider, - authProviderId: resolveProviderIdForAuth(modelSelection.provider, { - config: cfg, - }), - }); - replyDispatch.onModelSelected(modelSelection); - emitServerTiming( - "model-selected", - { - provider: modelSelection.provider, - model: modelSelection.model, - }, - dispatchStartedAtMs, - ); - }, - }, - }); - const dispatchResult = await (cronCreatorAuthority && externalAuthorityAdmission - ? externalAuthorityAdmission.run( - cronCreatorAuthority, - dispatchInbound, - activeRunAbort.controller.signal, - ) - : dispatchInbound()); - if (dispatchResult.beforeAgentRunBlocked === true) { - userTurnRecorder.markBlocked(); - } - return dispatchResult; - }, - { - phase: "agent-turn", - config: cfg, - attributes: chatSendTraceAttributes, - }, - ), - ) - .then(async () => { - if (acceptedMessageInjection) { - return; - } - emitServerTiming("dispatch-completed", undefined, dispatchStartedAtMs); - const postDispatchStartedAtMs = performance.now(); - await measureDiagnosticsTimelineSpan( - "gateway.chat_send.post_dispatch", - async () => { - const returnedAgentErrorPayloads = agentRunStarted - ? replyDispatch.deliveredReplies - .map((entryInner) => entryInner.payload) - .filter((payload) => payload.isError) - : []; - const returnedAgentErrorMessage = - returnedAgentErrorPayloads - .map((payload) => payload.text?.trim()) - .filter((text): text is string => Boolean(text)) - .join(" | ") || undefined; - if ( - agentRunStarted && - returnedAgentErrorPayloads.length > 0 && - !userTurnRecorder.hasPersisted() && - !userTurnRecorder.isBlocked() - ) { - await persistGatewayUserTurnTranscriptBestEffort(); - } - if ( - agentRunStarted && - returnedAgentErrorPayloads.length === 0 && - !userTurnRecorder.hasPersisted() && - !userTurnRecorder.isBlocked() && - userTurnRecorder.hasRuntimePersistencePending() - ) { - await persistGatewayUserTurnTranscriptBestEffort(); - } - let broadcastedSourceReplyFinal = false; - // Agent runs persist model-visible turns through SessionManager; this dispatcher owns - // live delivery. Mirroring agent finals would duplicate normal assistant turns. The - // non-agent branch has no runtime-owned turn, so it appends one before broadcasting. - if (!agentRunStarted && !queuedFollowup.isEnqueued()) { - await finalizeChatSendNonAgentReplies({ - accountId, - context, - deliveredReplies: replyDispatch.deliveredReplies, - emitFirstAssistantServerTiming, - foldCommandBlocks: isInternalTextSlashCommandTurn, - persistUserTurnTranscript: persistGatewayUserTurnTranscriptBestEffort, - session: preparedSession.value, - suppressReplies: replyDispatch.hasAppendedWebchatAgentMedia(), - }); - } else { - broadcastedSourceReplyFinal = await finalizeChatSendSourceReplies({ - accountId, - context, - deliveredReplies: replyDispatch.deliveredReplies, - emitFirstAssistantServerTiming, - hasReturnedAgentErrorPayloads: returnedAgentErrorPayloads.length > 0, - session: preparedSession.value, - }); - } - const shouldBroadcastAgentError = - returnedAgentErrorPayloads.length > 0 && !broadcastedSourceReplyFinal; - if (shouldBroadcastAgentError) { - broadcastChatError({ - context, - runId: clientRunId, - sessionKey, - agentId, - errorMessage: returnedAgentErrorMessage, - }); - } - if (!context.chatRunState.hasAbortMarker(clientRunId)) { - const returnedAgentError = shouldBroadcastAgentError - ? errorShape( - ErrorCodes.UNAVAILABLE, - returnedAgentErrorMessage ?? "agent returned an error payload", - ) - : undefined; - setGatewayDedupeEntry({ - dedupe: context.dedupe, - key: `chat:${clientRunId}`, - entry: { - ts: Date.now(), - ok: !shouldBroadcastAgentError, - payload: shouldBroadcastAgentError - ? { - runId: clientRunId, - status: "error" as const, - summary: returnedAgentErrorMessage ?? "agent returned an error payload", - } - : { runId: clientRunId, status: "ok" as const }, - ...(returnedAgentError ? { error: returnedAgentError } : {}), - }, - }); - } - }, - { - phase: "agent-turn", - config: cfg, - attributes: chatSendTraceAttributes, - }, - ); - emitServerTiming( - "post-dispatch-completed", - { - postDispatchMs: roundedChatSendTimingMs(performance.now() - postDispatchStartedAtMs), - }, - dispatchStartedAtMs, - ); - if (queuedFollowup.isEnqueued() && !context.chatRunState.hasAbortMarker(clientRunId)) { - // Successful queue admission ends this client run. The later - // aggregate/followup owns its own run id. - broadcastChatFinal({ - context, - runId: clientRunId, - sessionKey, - agentId, - }); - } - }) - .catch(dispatchErrorLifecycle.handleError) - .finally(dispatchErrorLifecycle.finalize); } catch (err) { await handleChatSendSetupError({ admission: admitted.value,