mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
refactor: split agent and chat orchestration ownership (#122318)
* refactor(agents): split subscription controllers * refactor(gateway): split chat send dispatch * style(gateway): format chat send handler
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
5b843e1c3f
commit
22b3c2530f
@@ -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
|
||||
|
||||
@@ -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<Promise<void>>();
|
||||
const pendingPartialReplyTasks = new Set<Promise<void>>();
|
||||
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<BlockReplyPayload>();
|
||||
const emitBlockReplySafely = (
|
||||
payload: Parameters<NonNullable<SubscribeEmbeddedAgentSessionParams["onBlockReply"]>>[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<void>(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,
|
||||
};
|
||||
}
|
||||
@@ -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<AgentInternalEvent["attachments"]>;
|
||||
trustByUrl: Map<string, boolean>;
|
||||
} {
|
||||
if (!events?.length) {
|
||||
return { mediaUrls: [], attachments: [], trustByUrl: new Map() };
|
||||
}
|
||||
const pending: string[] = [];
|
||||
const attachments: NonNullable<AgentInternalEvent["attachments"]> = [];
|
||||
const indexByUrl = new Map<string, number>();
|
||||
const trustedByUrl = new Map<string, boolean>();
|
||||
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 <think> 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,
|
||||
};
|
||||
}
|
||||
@@ -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<Promise<void>>;
|
||||
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 <final> 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 <final> start tag.
|
||||
inFinal = true;
|
||||
everInFinal = true;
|
||||
lastFinalIndex = idx + match.text.length;
|
||||
} else if (inFinal && isClose) {
|
||||
// Found </final> 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 <final> tag. Otherwise, we leak "thinking out loud" text
|
||||
// (e.g. "**Locating Manulife**...") that the model emitted without <think> tags.
|
||||
if (!everInFinal) {
|
||||
stateLocal.inlineCode = createInlineCodeState();
|
||||
stateLocal.fence = finalCodeSpans.fenceState;
|
||||
stateLocal.finalInlineCode = undefined;
|
||||
stateLocal.finalFence = undefined;
|
||||
return "";
|
||||
}
|
||||
|
||||
// Hardened Cleanup: Remove any remaining <final> 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 <think> and <final> 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<void> => {
|
||||
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,
|
||||
};
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -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<ReturnType<typeof prepareChatSendAttachments>>,
|
||||
{ ok: true }
|
||||
>["value"];
|
||||
|
||||
type StartChatDispatchParams = {
|
||||
admissionStartedAt: number;
|
||||
admission: AdmittedChatSend;
|
||||
attachments: PreparedChatSendAttachments;
|
||||
client: GatewayRequestHandlerOptions["client"];
|
||||
context: GatewayRequestHandlerOptions["context"];
|
||||
cronCreatorAuthority: ReturnType<ChatSendExternalAuthorityAdmission["resolve"]>;
|
||||
externalAuthorityAdmission: ChatSendExternalAuthorityAdmission | undefined;
|
||||
injection: {
|
||||
beginCapturedMessageInjection: () => ReplyMessageInjectionAttempt | undefined;
|
||||
messageInjectionAttempt: ReplyMessageInjectionAttempt | undefined;
|
||||
preAckReplyContextPromise: Promise<ChatSendReplyContextFields> | undefined;
|
||||
replyContextFieldsPromise: Promise<ChatSendReplyContextFields> | undefined;
|
||||
};
|
||||
request: NormalizedChatSendRequest;
|
||||
session: PreparedChatSendSession;
|
||||
terminalizeRestartSafeAdmission: (terminalState: {
|
||||
retryable: boolean;
|
||||
status: "failed" | "killed";
|
||||
}) => Promise<boolean>;
|
||||
timing: {
|
||||
chatSendAckedAtMs: number;
|
||||
chatSendTiming: ChatRunTiming | undefined;
|
||||
};
|
||||
turn: ReturnType<typeof prepareChatSendUserTurn>;
|
||||
userTurn: ReturnType<typeof createGatewayChatUserTurnController>;
|
||||
};
|
||||
|
||||
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<string, string | number>,
|
||||
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);
|
||||
}
|
||||
@@ -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<string, string | number>,
|
||||
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,
|
||||
|
||||
Reference in New Issue
Block a user