mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-16 18:31:41 +00:00
4289c6c16c
* feat(protocol): add worker live-event frames * feat(gateway): fan out replayable worker live events * feat(gateway): add worker live-event replay receiver * docs(protocol): document worker live events * fix(gateway): preserve deferred worker live events * test(gateway): cover deferred worker event projection * refactor(gateway): revert async projection layer * fix(gateway): retain buffered worker live events * fix(gateway): guard buffered terminal admission * fix(plugin-sdk): trim worker live-event declarations
1615 lines
57 KiB
TypeScript
1615 lines
57 KiB
TypeScript
// Gateway chat runtime projects agent events into chat/session subscriber
|
|
// streams, lifecycle persistence, heartbeat visibility, and live UI updates.
|
|
import { performance } from "node:perf_hooks";
|
|
import type { ChatEvent } from "../../packages/gateway-protocol/src/schema/logs-chat.js";
|
|
import { buildAgentRunTerminalOutcome } from "../agents/agent-run-terminal-outcome.js";
|
|
import { resolveDefaultAgentId } from "../agents/agent-scope.js";
|
|
import { isTimeoutError, resolveFailoverReasonFromError } from "../agents/failover-error.js";
|
|
import { resolveToolSearchCodeDisplayTarget } from "../agents/tool-display-common.js";
|
|
import { readToolValidationErrorSummary } from "../agents/tool-error-summary.js";
|
|
import { DEFAULT_HEARTBEAT_ACK_MAX_CHARS, stripHeartbeatToken } from "../auto-reply/heartbeat.js";
|
|
import { normalizeVerboseLevel } from "../auto-reply/thinking.js";
|
|
import { getRuntimeConfig } from "../config/io.js";
|
|
import {
|
|
type AgentEventPayload,
|
|
type AgentEventRuntimePayload,
|
|
getAgentEventLifecycleGeneration,
|
|
getAgentRunContext,
|
|
getAgentRunContextOwnerStatus,
|
|
} from "../infra/agent-events.js";
|
|
import { formatErrorMessage } from "../infra/errors.js";
|
|
import { resolveHeartbeatVisibility } from "../infra/heartbeat-visibility.js";
|
|
import { logError } from "../logger.js";
|
|
import {
|
|
isAcpSessionKey,
|
|
isSubagentSessionKey,
|
|
parseCronRunScopeSuffix,
|
|
} from "../sessions/session-key-utils.js";
|
|
import { resolveAssistantEventPhase } from "../shared/chat-message-content.js";
|
|
import { setSafeTimeout } from "../utils/timer-delay.js";
|
|
import {
|
|
normalizeLiveAssistantBufferedText,
|
|
projectLiveAssistantBufferedText,
|
|
resolveAssistantLiveChatInput,
|
|
resolveMergedAssistantText,
|
|
shouldSuppressAssistantEventForLiveChat,
|
|
} from "./live-chat-projector.js";
|
|
import { isChatAbortMarkerCurrent } from "./server-chat-state.js";
|
|
import type {
|
|
BufferedAgentEvent,
|
|
ChatRunEntry,
|
|
ChatRunState,
|
|
SessionEventSubscriberRegistry,
|
|
SessionMessageSubscriberRegistry,
|
|
ToolEventRecipientRegistry,
|
|
} from "./server-chat-state.js";
|
|
import { loadGatewaySessionRow } from "./server-chat.load-gateway-session-row.runtime.js";
|
|
import { persistGatewaySessionLifecycleEvent } from "./server-chat.persist-session-lifecycle.runtime.js";
|
|
import {
|
|
deriveGatewaySessionLifecycleProjectionPatch,
|
|
isRestartRecoveryLifecycleEvent,
|
|
isStaleLifecycleEventForSession,
|
|
} from "./session-lifecycle-state.js";
|
|
import { loadSessionEntry } from "./session-utils.js";
|
|
import { formatForLog } from "./ws-log.js";
|
|
|
|
export {
|
|
createChatAbortMarker,
|
|
createChatRunRegistry,
|
|
createChatRunState,
|
|
createSessionEventSubscriberRegistry,
|
|
createSessionMessageSubscriberRegistry,
|
|
createToolEventRecipientRegistry,
|
|
} from "./server-chat-state.js";
|
|
export type {
|
|
ChatAbortMarker,
|
|
ChatRunEntry,
|
|
ChatRunRegistry,
|
|
ChatRunRegistration,
|
|
ChatRunState,
|
|
SessionEventSubscriberRegistry,
|
|
SessionMessageSubscriberRegistry,
|
|
ToolEventRecipientRegistry,
|
|
} from "./server-chat-state.js";
|
|
|
|
function projectToolSearchCodeEventForChannelPayload<T extends { data?: unknown }>(payload: T): T {
|
|
const data = payload.data;
|
|
if (!data || typeof data !== "object") {
|
|
return payload;
|
|
}
|
|
const record = data as Record<string, unknown>;
|
|
if (record.name !== "tool_search_code") {
|
|
return payload;
|
|
}
|
|
const target = resolveToolSearchCodeDisplayTarget(record.args);
|
|
if (!target) {
|
|
return payload;
|
|
}
|
|
const projectedName = target.displayToolName ?? target.toolName;
|
|
if (!projectedName || projectedName === "tool_search_code") {
|
|
return payload;
|
|
}
|
|
|
|
// Channel/node subscribers render from event data, not the richer display
|
|
// helper used by Control UI. Project obvious bridge calls so verbose
|
|
// surfaces name the concrete tool while keeping the bridge identity available.
|
|
const projectedData: Record<string, unknown> = { ...record, name: projectedName };
|
|
if (target.displayArgs) {
|
|
projectedData.args = target.displayArgs;
|
|
} else if (target.detail) {
|
|
projectedData.args = { detail: target.detail };
|
|
}
|
|
if (target.bridgeVerb) {
|
|
projectedData.bridgeToolName = "tool_search_code";
|
|
projectedData.bridgeTargetToolName = target.toolName;
|
|
projectedData.bridgeVerb = target.bridgeVerb;
|
|
}
|
|
return { ...payload, data: projectedData };
|
|
}
|
|
|
|
function resolveHeartbeatAckMaxChars(): number {
|
|
try {
|
|
const cfg = getRuntimeConfig();
|
|
return Math.max(
|
|
0,
|
|
cfg.agents?.defaults?.heartbeat?.ackMaxChars ?? DEFAULT_HEARTBEAT_ACK_MAX_CHARS,
|
|
);
|
|
} catch {
|
|
return DEFAULT_HEARTBEAT_ACK_MAX_CHARS;
|
|
}
|
|
}
|
|
|
|
function resolveHeartbeatContext(runId: string, sourceRunId?: string) {
|
|
const primary = getAgentRunContext(runId);
|
|
if (primary?.isHeartbeat) {
|
|
return primary;
|
|
}
|
|
if (sourceRunId && sourceRunId !== runId) {
|
|
const source = getAgentRunContext(sourceRunId);
|
|
if (source?.isHeartbeat) {
|
|
return source;
|
|
}
|
|
}
|
|
return primary;
|
|
}
|
|
|
|
/**
|
|
* Check if heartbeat ACK/noise should be hidden from interactive chat surfaces.
|
|
*/
|
|
function shouldHideHeartbeatChatOutput(runId: string, sourceRunId?: string): boolean {
|
|
const runContext = resolveHeartbeatContext(runId, sourceRunId);
|
|
if (!runContext?.isHeartbeat) {
|
|
return false;
|
|
}
|
|
|
|
try {
|
|
const cfg = getRuntimeConfig();
|
|
const visibility = resolveHeartbeatVisibility({ cfg, channel: "webchat" });
|
|
return !visibility.showOk;
|
|
} catch {
|
|
// Default to suppressing if we can't load config
|
|
return true;
|
|
}
|
|
}
|
|
|
|
function shouldSuppressHeartbeatToolEvents(runId: string, sourceRunId?: string): boolean {
|
|
return Boolean(resolveHeartbeatContext(runId, sourceRunId)?.isHeartbeat);
|
|
}
|
|
|
|
function shouldMirrorAssistantEventToHiddenSessionMessages(data: unknown): boolean {
|
|
if (!data || typeof data !== "object") {
|
|
return false;
|
|
}
|
|
const record = data as { text?: unknown; delta?: unknown };
|
|
const hasText = typeof record.text === "string" && record.text.length > 0;
|
|
const hasDelta = typeof record.delta === "string" && record.delta.length > 0;
|
|
if (!hasText && !hasDelta) {
|
|
return false;
|
|
}
|
|
return resolveAssistantEventPhase(data) === "commentary";
|
|
}
|
|
|
|
function shouldMirrorAgentEventToHiddenSessionMessages(evt: AgentEventPayload): boolean {
|
|
return evt.stream === "thinking" || evt.stream === "approval" || evt.stream === "lifecycle";
|
|
}
|
|
|
|
function normalizeHeartbeatChatFinalText(params: {
|
|
runId: string;
|
|
sourceRunId?: string;
|
|
text: string;
|
|
}): { suppress: boolean; text: string } {
|
|
if (!shouldHideHeartbeatChatOutput(params.runId, params.sourceRunId)) {
|
|
return { suppress: false, text: params.text };
|
|
}
|
|
|
|
const stripped = stripHeartbeatToken(params.text, {
|
|
mode: "heartbeat",
|
|
maxAckChars: resolveHeartbeatAckMaxChars(),
|
|
});
|
|
if (!stripped.didStrip) {
|
|
return { suppress: false, text: params.text };
|
|
}
|
|
if (stripped.shouldSkip) {
|
|
return { suppress: true, text: "" };
|
|
}
|
|
return { suppress: false, text: stripped.text };
|
|
}
|
|
|
|
/**
|
|
* Keep this aligned with the agent.wait lifecycle-error grace so chat surfaces
|
|
* do not finalize a run before fallback or retry reuses the same runId.
|
|
*/
|
|
const AGENT_LIFECYCLE_ERROR_RETRY_GRACE_MS = 15_000;
|
|
|
|
export type ChatEventBroadcast = (
|
|
event: string,
|
|
payload: unknown,
|
|
opts?: { dropIfSlow?: boolean },
|
|
) => void;
|
|
|
|
export type NodeSendToSession = (sessionKey: string, event: string, payload: unknown) => void;
|
|
|
|
// Derived from ChatErrorEventSchema.errorKind (gateway-protocol); keep set in sync.
|
|
type ChatErrorKind = NonNullable<Extract<ChatEvent, { state: "error" }>["errorKind"]>;
|
|
|
|
const CHAT_ERROR_KINDS = new Set<ChatErrorKind>([
|
|
"refusal",
|
|
"timeout",
|
|
"rate_limit",
|
|
"context_length",
|
|
"unknown",
|
|
]);
|
|
|
|
function buildChatErrorMessage(error: unknown): Record<string, unknown> | undefined {
|
|
const raw = error ? formatForLog(error).trim() : "";
|
|
if (!raw) {
|
|
return undefined;
|
|
}
|
|
const text = raw.startsWith("⚠️") || raw.startsWith("Error:") ? raw : `Error: ${raw}`;
|
|
return {
|
|
role: "assistant",
|
|
content: [{ type: "text", text }],
|
|
timestamp: Date.now(),
|
|
};
|
|
}
|
|
|
|
function readChatErrorKind(value: unknown): ChatErrorKind | undefined {
|
|
return typeof value === "string" && CHAT_ERROR_KINDS.has(value as ChatErrorKind)
|
|
? (value as ChatErrorKind)
|
|
: undefined;
|
|
}
|
|
|
|
// Refusal first (stop-reason fact, not FailoverReason); then canonical failover map.
|
|
export function resolveChatErrorKindFromError(error: unknown): ChatErrorKind | undefined {
|
|
if (error === undefined) {
|
|
return undefined;
|
|
}
|
|
const message = formatErrorMessage(error).toLowerCase();
|
|
if (
|
|
message.includes("refusal") ||
|
|
message.includes("content_filter") ||
|
|
message.includes("sensitive") ||
|
|
message.includes("unhandled stop reason: refusal_policy")
|
|
) {
|
|
return "refusal";
|
|
}
|
|
const reason = resolveFailoverReasonFromError(error);
|
|
if (reason === "rate_limit" || reason === "overloaded") {
|
|
return "rate_limit";
|
|
}
|
|
if (reason === "context_overflow") {
|
|
return "context_length";
|
|
}
|
|
// FailoverReason "timeout" is the retryable-transient bucket and deliberately
|
|
// swallows generic 5xx; only genuinely timeout-shaped errors get the badge.
|
|
return isTimeoutError(error) ? "timeout" : undefined;
|
|
}
|
|
|
|
function excludeConnIds(
|
|
connIds: ReadonlySet<string>,
|
|
excludedConnIds: ReadonlySet<string> | undefined,
|
|
): ReadonlySet<string> {
|
|
if (!excludedConnIds || excludedConnIds.size === 0 || connIds.size === 0) {
|
|
return connIds;
|
|
}
|
|
const filtered = new Set<string>();
|
|
for (const connId of connIds) {
|
|
if (!excludedConnIds.has(connId)) {
|
|
filtered.add(connId);
|
|
}
|
|
}
|
|
return filtered;
|
|
}
|
|
|
|
type BroadcastDelta = { deltaText: string; replace?: true };
|
|
|
|
function resolveBroadcastDelta(params: {
|
|
text: string;
|
|
previousBroadcastText: string | undefined;
|
|
}): BroadcastDelta | undefined {
|
|
if (!params.text) {
|
|
return undefined;
|
|
}
|
|
const previous = params.previousBroadcastText;
|
|
if (previous === undefined) {
|
|
return { deltaText: params.text };
|
|
}
|
|
if (!params.text.startsWith(previous)) {
|
|
return { deltaText: params.text, replace: true };
|
|
}
|
|
const deltaText = params.text.slice(previous.length);
|
|
return deltaText ? { deltaText } : undefined;
|
|
}
|
|
|
|
export type AgentEventHandlerOptions = {
|
|
broadcast: ChatEventBroadcast;
|
|
broadcastToConnIds: (
|
|
event: string,
|
|
payload: unknown,
|
|
connIds: ReadonlySet<string>,
|
|
opts?: { dropIfSlow?: boolean },
|
|
) => void;
|
|
nodeSendToSession: NodeSendToSession;
|
|
agentRunSeq: Map<string, number>;
|
|
chatRunState: ChatRunState;
|
|
resolveSessionKeyForRun: (runId: string) => string | undefined;
|
|
clearAgentRunContext: (
|
|
runId: string,
|
|
lifecycleGeneration?: string,
|
|
contextClaimId?: string,
|
|
) => void;
|
|
toolEventRecipients: ToolEventRecipientRegistry;
|
|
sessionEventSubscribers: SessionEventSubscriberRegistry;
|
|
sessionMessageSubscribers: SessionMessageSubscriberRegistry;
|
|
loadGatewaySessionRowForSnapshot?: typeof loadGatewaySessionRow;
|
|
lifecycleErrorRetryGraceMs?: number;
|
|
isChatSendRunActive?: (runId: string) => boolean;
|
|
clearTrackedActiveRun?: (params: {
|
|
runId: string;
|
|
clientRunId: string;
|
|
sessionKey: string;
|
|
}) => void;
|
|
markTrackedRunTerminalPersisted?: (params: {
|
|
runId: string;
|
|
clientRunId: string;
|
|
sessionKey: string;
|
|
}) => void;
|
|
trackTrackedRunTerminalPersistence?: (params: {
|
|
runId: string;
|
|
clientRunId: string;
|
|
sessionKey: string;
|
|
sessionId?: string;
|
|
observedAt: number;
|
|
persistence: Promise<void>;
|
|
}) => void;
|
|
resolveActiveLifecycleGenerationForRun?: (runId: string) => string | undefined;
|
|
updateRunToolErrorSummary?: (params: {
|
|
runId: string;
|
|
clientRunId: string;
|
|
summary: string | undefined;
|
|
}) => void;
|
|
resolveSessionActiveRunState?: (params: {
|
|
requestedKey: string;
|
|
canonicalKey: string;
|
|
sessionId?: string;
|
|
agentId?: string;
|
|
}) => { active: boolean; runIds: string[] };
|
|
};
|
|
|
|
function roundedChatSendTimingMs(value: number): number {
|
|
return Math.max(0, Math.round(value * 1000) / 1000);
|
|
}
|
|
|
|
export function createAgentEventHandler({
|
|
broadcast,
|
|
broadcastToConnIds,
|
|
nodeSendToSession,
|
|
agentRunSeq,
|
|
chatRunState,
|
|
resolveSessionKeyForRun,
|
|
clearAgentRunContext,
|
|
toolEventRecipients,
|
|
sessionEventSubscribers,
|
|
sessionMessageSubscribers,
|
|
loadGatewaySessionRowForSnapshot = loadGatewaySessionRow,
|
|
lifecycleErrorRetryGraceMs = AGENT_LIFECYCLE_ERROR_RETRY_GRACE_MS,
|
|
isChatSendRunActive = () => false,
|
|
clearTrackedActiveRun,
|
|
markTrackedRunTerminalPersisted,
|
|
trackTrackedRunTerminalPersistence,
|
|
resolveActiveLifecycleGenerationForRun = () => undefined,
|
|
updateRunToolErrorSummary,
|
|
resolveSessionActiveRunState,
|
|
}: AgentEventHandlerOptions): (event: AgentEventPayload) => void {
|
|
const shouldProcessOwnedEvent = (evt: AgentEventRuntimePayload): boolean => {
|
|
const claimId = evt.contextClaimId;
|
|
if (!claimId) {
|
|
return true;
|
|
}
|
|
const lifecycleGeneration = evt.lifecycleGeneration;
|
|
if (!lifecycleGeneration || lifecycleGeneration !== getAgentEventLifecycleGeneration()) {
|
|
return false;
|
|
}
|
|
// A missing claim means detach or supersession revoked this terminal event.
|
|
return getAgentRunContextOwnerStatus(evt.runId, claimId, lifecycleGeneration) === "active";
|
|
};
|
|
const clearRunContextForEvent = (evt: AgentEventRuntimePayload): void => {
|
|
if (evt.contextClaimId) {
|
|
clearAgentRunContext(evt.runId, evt.lifecycleGeneration, evt.contextClaimId);
|
|
return;
|
|
}
|
|
clearAgentRunContext(evt.runId);
|
|
};
|
|
|
|
type TerminalLifecycleOptions = {
|
|
skipChatErrorFinal?: boolean;
|
|
suppressRestartRecoveryProjection?: boolean;
|
|
restartRecoveryState?: { suppress: boolean };
|
|
};
|
|
type PendingTerminalLifecycleError = {
|
|
timer: NodeJS.Timeout;
|
|
event: AgentEventRuntimePayload;
|
|
opts?: TerminalLifecycleOptions;
|
|
};
|
|
|
|
const pendingTerminalLifecycleErrors = new Map<string, PendingTerminalLifecycleError>();
|
|
|
|
type AgentTextThrottleStream = "assistant" | "thinking";
|
|
|
|
const agentTextThrottleKey = (clientRunId: string, stream: AgentTextThrottleStream) =>
|
|
`${clientRunId}:${stream}`;
|
|
|
|
const agentTextThrottleKeys = (clientRunId: string) => [
|
|
clientRunId,
|
|
agentTextThrottleKey(clientRunId, "assistant"),
|
|
agentTextThrottleKey(clientRunId, "thinking"),
|
|
];
|
|
|
|
const clearBufferedChatState = (clientRunId: string) => {
|
|
chatRunState.clearRun(clientRunId);
|
|
};
|
|
|
|
const clearPendingTerminalLifecycleError = (runId: string, lifecycleGeneration?: string) => {
|
|
const pending = pendingTerminalLifecycleErrors.get(runId);
|
|
if (!pending) {
|
|
return;
|
|
}
|
|
if (
|
|
lifecycleGeneration &&
|
|
pending.event.lifecycleGeneration &&
|
|
lifecycleGeneration !== pending.event.lifecycleGeneration
|
|
) {
|
|
return;
|
|
}
|
|
clearTimeout(pending.timer);
|
|
pendingTerminalLifecycleErrors.delete(runId);
|
|
};
|
|
|
|
const resolveRestartRecoveryLifecycleState = (
|
|
sessionKey: string,
|
|
agentId: string | undefined,
|
|
event: AgentEventPayload,
|
|
): { suppress: boolean } => {
|
|
try {
|
|
const { entry } = loadSessionEntry(sessionKey, {
|
|
...(agentId ? { agentId } : {}),
|
|
clone: false,
|
|
});
|
|
return { suppress: isRestartRecoveryLifecycleEvent({ entry, event }) };
|
|
} catch {
|
|
return { suppress: false };
|
|
}
|
|
};
|
|
|
|
// Only subagent/acp keys can carry spawnedBy (mirrors supportsSpawnLineage in
|
|
// sessions-patch.ts). Short-circuit everyone else so high-volume chat streams
|
|
// do not touch the session store. Results are cached per sessionKey because
|
|
// spawnedBy is immutable once set and resolveSpawnedBy sits on the hot event
|
|
// path (delta, flush, final, agent, seq-gap).
|
|
const spawnedByCache = new Map<string, string | null>();
|
|
const resolveSpawnedBy = (sessionKey: string): string | null => {
|
|
if (spawnedByCache.has(sessionKey)) {
|
|
return spawnedByCache.get(sessionKey)!;
|
|
}
|
|
// Non-lineage keys return null without polluting the cache; only
|
|
// subagent/ACP results (positive or null) are worth memoising.
|
|
if (!isSubagentSessionKey(sessionKey) && !isAcpSessionKey(sessionKey)) {
|
|
return null;
|
|
}
|
|
let result: string | null = null;
|
|
try {
|
|
result = loadGatewaySessionRow(sessionKey)?.spawnedBy ?? null;
|
|
} catch {
|
|
// result stays null
|
|
}
|
|
spawnedByCache.set(sessionKey, result);
|
|
return result;
|
|
};
|
|
|
|
const buildSessionEventSnapshot = (
|
|
sessionKey: string,
|
|
evt?: AgentEventPayload,
|
|
agentId?: string,
|
|
includeActiveRunState = false,
|
|
) => {
|
|
const row = loadGatewaySessionRowForSnapshot(sessionKey, agentId ? { agentId } : undefined);
|
|
const omitUnscopedGlobalGoal = sessionKey === "global" && !agentId;
|
|
const lifecyclePatch =
|
|
evt &&
|
|
!isStaleLifecycleEventForSession({
|
|
owningSessionId: evt.sessionId,
|
|
currentSessionId: row?.sessionId,
|
|
})
|
|
? deriveGatewaySessionLifecycleProjectionPatch({
|
|
entry: row
|
|
? {
|
|
updatedAt: row.updatedAt ?? undefined,
|
|
status: row.status,
|
|
startedAt: row.startedAt,
|
|
endedAt: row.endedAt,
|
|
runtimeMs: row.runtimeMs,
|
|
abortedLastRun: row.abortedLastRun,
|
|
}
|
|
: undefined,
|
|
event: evt,
|
|
})
|
|
: {};
|
|
const activeRunState = includeActiveRunState
|
|
? resolveSessionActiveRunState?.({
|
|
requestedKey: sessionKey,
|
|
canonicalKey: row?.key ?? sessionKey,
|
|
...(row?.sessionId ? { sessionId: row.sessionId } : {}),
|
|
...(agentId ? { agentId } : {}),
|
|
})
|
|
: undefined;
|
|
// Agent lifecycle broadcasts merge into cached session rows in the UI.
|
|
// Always replace run identity so a newer start cannot inherit a completed run.
|
|
const activeRunFields = activeRunState
|
|
? { hasActiveRun: activeRunState.active, activeRunIds: activeRunState.runIds }
|
|
: {};
|
|
const session = row ? { ...row, ...lifecyclePatch, ...activeRunFields } : undefined;
|
|
if (session && omitUnscopedGlobalGoal) {
|
|
delete session.goal;
|
|
}
|
|
const snapshotSource = session ?? lifecyclePatch;
|
|
return {
|
|
...(session ? { session } : {}),
|
|
updatedAt: snapshotSource.updatedAt,
|
|
sessionId: row?.sessionId,
|
|
kind: row?.kind,
|
|
channel: row?.channel,
|
|
subject: row?.subject,
|
|
groupChannel: row?.groupChannel,
|
|
space: row?.space,
|
|
chatType: row?.chatType,
|
|
origin: row?.origin,
|
|
spawnedBy: row?.spawnedBy,
|
|
spawnedWorkspaceDir: row?.spawnedWorkspaceDir,
|
|
spawnedCwd: row?.spawnedCwd,
|
|
forkedFromParent: row?.forkedFromParent,
|
|
spawnDepth: row?.spawnDepth,
|
|
subagentRole: row?.subagentRole,
|
|
subagentControlScope: row?.subagentControlScope,
|
|
label: row?.label,
|
|
displayName: row?.displayName,
|
|
deliveryContext: row?.deliveryContext,
|
|
parentSessionKey: row?.parentSessionKey,
|
|
childSessions: row?.childSessions,
|
|
thinkingLevel: row?.thinkingLevel,
|
|
fastMode: row?.fastMode,
|
|
verboseLevel: row?.verboseLevel,
|
|
traceLevel: row?.traceLevel,
|
|
reasoningLevel: row?.reasoningLevel,
|
|
elevatedLevel: row?.elevatedLevel,
|
|
sendPolicy: row?.sendPolicy,
|
|
systemSent: row?.systemSent,
|
|
inputTokens: row?.inputTokens,
|
|
outputTokens: row?.outputTokens,
|
|
lastChannel: row?.lastChannel,
|
|
lastTo: row?.lastTo,
|
|
lastAccountId: row?.lastAccountId,
|
|
lastThreadId: row?.lastThreadId,
|
|
totalTokens: row?.totalTokens,
|
|
totalTokensFresh: row?.totalTokensFresh,
|
|
...(omitUnscopedGlobalGoal ? {} : { goal: row?.goal ?? null }),
|
|
contextTokens: row?.contextTokens,
|
|
estimatedCostUsd: row?.estimatedCostUsd,
|
|
responseUsage: row?.responseUsage,
|
|
// Carry the row-built channel-aware effective mode so the chat snapshot
|
|
// matches the session-event/list projections.
|
|
effectiveResponseUsage: row?.effectiveResponseUsage,
|
|
modelProvider: row?.modelProvider,
|
|
model: row?.model,
|
|
...activeRunFields,
|
|
status: snapshotSource.status,
|
|
startedAt: snapshotSource.startedAt,
|
|
endedAt: snapshotSource.endedAt,
|
|
runtimeMs: snapshotSource.runtimeMs,
|
|
abortedLastRun: snapshotSource.abortedLastRun,
|
|
};
|
|
};
|
|
|
|
const resolveSessionDeliveryKey = (sessionKey: string, agentId?: string) => {
|
|
if (sessionKey !== "global") {
|
|
return sessionKey;
|
|
}
|
|
const scopedAgentId = agentId ?? resolveDefaultAgentId(getRuntimeConfig());
|
|
return `agent:${scopedAgentId}:global`;
|
|
};
|
|
const resolveNodeSessionDeliveryKeys = (sessionKey: string, agentId?: string) => {
|
|
if (sessionKey !== "global") {
|
|
return [sessionKey];
|
|
}
|
|
const defaultAgentId = resolveDefaultAgentId(getRuntimeConfig());
|
|
const scopedAgentId = agentId ?? defaultAgentId;
|
|
const keys = [`agent:${scopedAgentId}:global`];
|
|
if (scopedAgentId === defaultAgentId) {
|
|
keys.push("global");
|
|
}
|
|
return keys;
|
|
};
|
|
const sendNodeSessionPayloadForAgent = (
|
|
sessionKey: string,
|
|
event: string,
|
|
payload: unknown,
|
|
agentId?: string,
|
|
) => {
|
|
for (const deliverySessionKey of resolveNodeSessionDeliveryKeys(sessionKey, agentId)) {
|
|
nodeSendToSession(deliverySessionKey, event, payload);
|
|
}
|
|
};
|
|
|
|
const emitFirstAssistantChatSendTiming = (chatLink: ChatRunEntry | undefined) => {
|
|
const timing = chatLink?.chatSendTiming;
|
|
if (!timing || timing.firstAssistantEventSent) {
|
|
return;
|
|
}
|
|
timing.firstAssistantEventSent = true;
|
|
const nowMs = performance.now();
|
|
broadcastToConnIds(
|
|
"chat.send_timing",
|
|
{
|
|
phase: "first-assistant-event",
|
|
runId: chatLink.clientRunId,
|
|
sessionKey: chatLink.sessionKey,
|
|
...(chatLink.agentId ? { agentId: chatLink.agentId } : {}),
|
|
ackToPhaseMs: roundedChatSendTimingMs(nowMs - timing.ackedAtMs),
|
|
receivedToPhaseMs: roundedChatSendTimingMs(nowMs - timing.receivedAtMs),
|
|
...(timing.dispatchStartedAtMs !== undefined
|
|
? {
|
|
dispatchStartedToPhaseMs: roundedChatSendTimingMs(nowMs - timing.dispatchStartedAtMs),
|
|
}
|
|
: {}),
|
|
},
|
|
new Set([timing.connId]),
|
|
{ dropIfSlow: true },
|
|
);
|
|
};
|
|
|
|
const finalizeLifecycleEvent = (
|
|
evt: AgentEventRuntimePayload,
|
|
opts?: TerminalLifecycleOptions,
|
|
) => {
|
|
if (!shouldProcessOwnedEvent(evt)) {
|
|
return;
|
|
}
|
|
const lifecyclePhase =
|
|
evt.stream === "lifecycle" && typeof evt.data?.phase === "string" ? evt.data.phase : null;
|
|
if (lifecyclePhase !== "end" && lifecyclePhase !== "error") {
|
|
return;
|
|
}
|
|
|
|
const currentRunContext = getAgentRunContext(evt.runId);
|
|
const activeLifecycleGeneration = resolveActiveLifecycleGenerationForRun(evt.runId);
|
|
const currentLifecycleGeneration =
|
|
activeLifecycleGeneration ?? currentRunContext?.lifecycleGeneration;
|
|
|
|
const chatLink = evt.contextClaimId ? undefined : chatRunState.registry.peek(evt.runId);
|
|
const sessionAgentId = chatLink?.agentId ?? evt.agentId;
|
|
const eventSessionKey =
|
|
evt.deliverySessionKey ??
|
|
(typeof evt.sessionKey === "string" && evt.sessionKey.trim() ? evt.sessionKey : undefined);
|
|
const isControlUiVisible =
|
|
evt.controlUiVisible ?? currentRunContext?.isControlUiVisible ?? true;
|
|
const sessionKey =
|
|
chatLink?.sessionKey ?? eventSessionKey ?? resolveSessionKeyForRun(evt.runId);
|
|
const restartRecoverySessionKey = eventSessionKey ?? sessionKey;
|
|
const restartRecoveryAgentId = evt.agentId ?? sessionAgentId;
|
|
const clientRunId = chatLink?.clientRunId ?? evt.runId;
|
|
const eventRunId = chatLink?.clientRunId ?? evt.runId;
|
|
const isAborted =
|
|
isChatAbortMarkerCurrent(chatRunState.abortedRuns.get(clientRunId), chatLink) ||
|
|
isChatAbortMarkerCurrent(chatRunState.abortedRuns.get(evt.runId), chatLink);
|
|
const lifecycleAborted = evt.data?.aborted === true;
|
|
const deliverySessionKey = sessionKey
|
|
? resolveSessionDeliveryKey(sessionKey, sessionAgentId)
|
|
: undefined;
|
|
const restartRecoveryState =
|
|
opts?.restartRecoveryState ??
|
|
(restartRecoverySessionKey
|
|
? resolveRestartRecoveryLifecycleState(
|
|
restartRecoverySessionKey,
|
|
restartRecoveryAgentId,
|
|
evt,
|
|
)
|
|
: undefined);
|
|
const suppressRestartRecoveryProjection =
|
|
opts?.suppressRestartRecoveryProjection === true ||
|
|
Boolean(
|
|
evt.lifecycleGeneration &&
|
|
activeLifecycleGeneration &&
|
|
evt.lifecycleGeneration !== activeLifecycleGeneration,
|
|
) ||
|
|
restartRecoveryState?.suppress === true;
|
|
const isSupersededRestartRecoveryEvent =
|
|
suppressRestartRecoveryProjection &&
|
|
Boolean(
|
|
evt.lifecycleGeneration &&
|
|
currentLifecycleGeneration &&
|
|
evt.lifecycleGeneration !== currentLifecycleGeneration,
|
|
);
|
|
if (isSupersededRestartRecoveryEvent) {
|
|
return;
|
|
}
|
|
|
|
if (
|
|
!suppressRestartRecoveryProjection &&
|
|
sessionKey &&
|
|
(isControlUiVisible ||
|
|
(deliverySessionKey ? sessionMessageSubscribers.get(deliverySessionKey).size > 0 : false))
|
|
) {
|
|
if (!isAborted) {
|
|
const evtStopReason =
|
|
typeof evt.data?.stopReason === "string" ? evt.data.stopReason : undefined;
|
|
const finished = chatLink ? chatRunState.registry.shift(evt.runId) : undefined;
|
|
if (chatLink && !finished) {
|
|
clearRunContextForEvent(evt);
|
|
return;
|
|
}
|
|
|
|
const terminalSessionKey = finished?.sessionKey ?? sessionKey;
|
|
const terminalRunId = finished?.clientRunId ?? eventRunId;
|
|
const terminalAgentId = finished?.agentId ?? sessionAgentId;
|
|
// Some local lifecycle sources only carry the aborted flag. Preserve
|
|
// that terminal state instead of misclassifying the run as a timeout.
|
|
const terminalStopReason = evtStopReason ?? (lifecycleAborted ? "aborted" : undefined);
|
|
const terminalOutcome = buildAgentRunTerminalOutcome({
|
|
status: lifecyclePhase === "error" ? "error" : lifecycleAborted ? "timeout" : "ok",
|
|
error: evt.data?.error,
|
|
stopReason: terminalStopReason,
|
|
livenessState: evt.data?.livenessState,
|
|
timeoutPhase: evt.data?.timeoutPhase,
|
|
providerStarted: evt.data?.providerStarted,
|
|
startedAt: evt.data?.startedAt,
|
|
endedAt: evt.data?.endedAt ?? evt.ts,
|
|
});
|
|
const terminalState =
|
|
terminalOutcome.reason === "completed"
|
|
? "done"
|
|
: terminalOutcome.reason === "cancelled" || terminalOutcome.reason === "aborted"
|
|
? "aborted"
|
|
: "error";
|
|
if (!(opts?.skipChatErrorFinal && terminalState === "error")) {
|
|
emitChatTerminal(
|
|
terminalSessionKey,
|
|
terminalRunId,
|
|
evt.runId,
|
|
evt.seq,
|
|
terminalState,
|
|
terminalOutcome.error ?? evt.data?.error,
|
|
terminalOutcome.stopReason,
|
|
readChatErrorKind(evt.data?.errorKind) ??
|
|
resolveChatErrorKindFromError(evt.data?.error),
|
|
{
|
|
agentId: terminalAgentId,
|
|
controlUiVisible: isControlUiVisible,
|
|
firstAssistantTimingEntry: finished,
|
|
abortErrorMessage: readToolValidationErrorSummary(evt.data?.toolErrorSummary),
|
|
},
|
|
);
|
|
}
|
|
} else {
|
|
clearBufferedChatState(clientRunId);
|
|
if (chatLink) {
|
|
chatRunState.registry.remove(evt.runId, clientRunId, sessionKey);
|
|
}
|
|
}
|
|
}
|
|
|
|
toolEventRecipients.markFinal(evt.runId);
|
|
clearBufferedChatState(clientRunId);
|
|
if (suppressRestartRecoveryProjection && chatLink) {
|
|
chatRunState.registry.remove(evt.runId, clientRunId, sessionKey);
|
|
}
|
|
clearRunContextForEvent(evt);
|
|
agentRunSeq.delete(evt.runId);
|
|
agentRunSeq.delete(clientRunId);
|
|
|
|
if (sessionKey) {
|
|
clearTrackedActiveRun?.({ runId: evt.runId, clientRunId, sessionKey });
|
|
if (!suppressRestartRecoveryProjection) {
|
|
const persistence = persistGatewaySessionLifecycleEvent({
|
|
sessionKey,
|
|
agentId: sessionAgentId,
|
|
event: evt,
|
|
});
|
|
trackTrackedRunTerminalPersistence?.({
|
|
runId: evt.runId,
|
|
clientRunId,
|
|
sessionKey,
|
|
sessionId: evt.sessionId,
|
|
observedAt: evt.ts,
|
|
persistence,
|
|
});
|
|
const broadcastSessionChange = (snapshotEvent?: AgentEventPayload) => {
|
|
if (parseCronRunScopeSuffix(sessionKey).runId) {
|
|
return;
|
|
}
|
|
const sessionEventConnIds = sessionEventSubscribers.getAll();
|
|
if (sessionEventConnIds.size === 0) {
|
|
return;
|
|
}
|
|
broadcastToConnIds(
|
|
"sessions.changed",
|
|
{
|
|
sessionKey,
|
|
...(sessionAgentId ? { agentId: sessionAgentId } : {}),
|
|
phase: lifecyclePhase,
|
|
runId: evt.runId,
|
|
...(eventRunId !== evt.runId ? { clientRunId: eventRunId } : {}),
|
|
ts: evt.ts,
|
|
...buildSessionEventSnapshot(sessionKey, snapshotEvent, sessionAgentId, true),
|
|
},
|
|
sessionEventConnIds,
|
|
{ dropIfSlow: true },
|
|
);
|
|
};
|
|
const markPersisted = () => {
|
|
markTrackedRunTerminalPersisted?.({
|
|
runId: evt.runId,
|
|
clientRunId,
|
|
sessionKey,
|
|
});
|
|
};
|
|
// Terminal writes serialize with restart markers. Reload only after the
|
|
// write so subscribers see the canonical post-race session state.
|
|
void persistence
|
|
.then(() => {
|
|
markPersisted();
|
|
broadcastSessionChange();
|
|
})
|
|
.catch((err: unknown) => {
|
|
logError(
|
|
`gateway: terminal session persistence failed session=${formatForLog(sessionKey)} run=${formatForLog(evt.runId)} error=${formatForLog(err)}`,
|
|
);
|
|
// Persistence recovery remains tracked by the controller entry, but
|
|
// subscribers still need a terminal projection instead of hanging.
|
|
broadcastSessionChange(evt);
|
|
});
|
|
}
|
|
}
|
|
};
|
|
|
|
const scheduleTerminalLifecycleError = (
|
|
evt: AgentEventRuntimePayload,
|
|
opts?: TerminalLifecycleOptions,
|
|
) => {
|
|
clearPendingTerminalLifecycleError(evt.runId);
|
|
const timer = setSafeTimeout(() => {
|
|
const pending = pendingTerminalLifecycleErrors.get(evt.runId);
|
|
if (!pending || pending.timer !== timer) {
|
|
return;
|
|
}
|
|
pendingTerminalLifecycleErrors.delete(evt.runId);
|
|
finalizeLifecycleEvent(pending.event, pending.opts);
|
|
}, lifecycleErrorRetryGraceMs);
|
|
timer.unref?.();
|
|
pendingTerminalLifecycleErrors.set(evt.runId, { timer, event: evt, opts });
|
|
};
|
|
|
|
const emitChatDelta = (
|
|
sessionKey: string,
|
|
agentId: string | undefined,
|
|
clientRunId: string,
|
|
sourceRunId: string,
|
|
seq: number,
|
|
text: string,
|
|
delta?: unknown,
|
|
opts?: { controlUiVisible?: boolean },
|
|
) => {
|
|
const previousRawText = chatRunState.rawBuffers.get(clientRunId) ?? "";
|
|
const mergedRawText = resolveMergedAssistantText({
|
|
previousText: previousRawText,
|
|
nextText: text,
|
|
nextDelta: typeof delta === "string" ? delta : "",
|
|
});
|
|
if (!mergedRawText) {
|
|
return;
|
|
}
|
|
const now = Date.now();
|
|
chatRunState.rawBuffers.set(clientRunId, mergedRawText);
|
|
chatRunState.bufferUpdatedAt.set(clientRunId, now);
|
|
// Sanitize only after merging. Protected blocks and directive tags can span
|
|
// delta frames; cleaning each frame independently can expose their contents.
|
|
const normalizedText = normalizeLiveAssistantBufferedText(mergedRawText);
|
|
const projected = projectLiveAssistantBufferedText(normalizedText);
|
|
const mergedText = projected.text;
|
|
chatRunState.buffers.set(clientRunId, mergedText);
|
|
if (projected.suppress) {
|
|
return;
|
|
}
|
|
if (shouldHideHeartbeatChatOutput(clientRunId, sourceRunId)) {
|
|
return;
|
|
}
|
|
const last = chatRunState.deltaSentAt.get(clientRunId) ?? 0;
|
|
if (now - last < 150) {
|
|
return;
|
|
}
|
|
const broadcastDelta = resolveBroadcastDelta({
|
|
text: mergedText,
|
|
previousBroadcastText: chatRunState.deltaLastBroadcastText.get(clientRunId),
|
|
});
|
|
if (!broadcastDelta) {
|
|
return;
|
|
}
|
|
chatRunState.deltaSentAt.set(clientRunId, now);
|
|
chatRunState.deltaLastBroadcastLen.set(clientRunId, mergedText.length);
|
|
chatRunState.deltaLastBroadcastText.set(clientRunId, mergedText);
|
|
const spawnedBy = resolveSpawnedBy(sessionKey);
|
|
const payload = {
|
|
runId: clientRunId,
|
|
sessionKey,
|
|
...(agentId ? { agentId } : {}),
|
|
...(spawnedBy && { spawnedBy }),
|
|
seq,
|
|
state: "delta" as const,
|
|
deltaText: broadcastDelta.deltaText,
|
|
...(broadcastDelta.replace ? { replace: true as const } : {}),
|
|
message: {
|
|
role: "assistant",
|
|
content: [{ type: "text", text: mergedText }],
|
|
timestamp: now,
|
|
},
|
|
};
|
|
emitFirstAssistantChatSendTiming(chatRunState.registry.peek(sourceRunId));
|
|
sendChatPayload(sessionKey, payload, {
|
|
agentId,
|
|
controlUiVisible: opts?.controlUiVisible ?? true,
|
|
dropIfSlow: true,
|
|
});
|
|
};
|
|
|
|
const resolveBufferedChatTextState = (
|
|
clientRunId: string,
|
|
sourceRunId: string,
|
|
options?: { suppressLeadFragments?: boolean },
|
|
) => {
|
|
const bufferedText = (chatRunState.buffers.get(clientRunId) ?? "").trim();
|
|
const normalizedHeartbeatText = normalizeHeartbeatChatFinalText({
|
|
runId: clientRunId,
|
|
sourceRunId,
|
|
text: bufferedText,
|
|
});
|
|
const projected = projectLiveAssistantBufferedText(normalizedHeartbeatText.text.trim(), {
|
|
suppressLeadFragments: options?.suppressLeadFragments,
|
|
});
|
|
return {
|
|
text: projected.text.trim(),
|
|
shouldSuppressSilent: normalizedHeartbeatText.suppress || projected.suppress,
|
|
};
|
|
};
|
|
|
|
const flushBufferedChatDeltaIfNeeded = (
|
|
sessionKey: string,
|
|
agentId: string | undefined,
|
|
clientRunId: string,
|
|
sourceRunId: string,
|
|
seq: number,
|
|
opts?: { controlUiVisible?: boolean; firstAssistantTimingEntry?: ChatRunEntry },
|
|
) => {
|
|
const { text, shouldSuppressSilent } = resolveBufferedChatTextState(clientRunId, sourceRunId, {
|
|
suppressLeadFragments: true,
|
|
});
|
|
const shouldSuppressHeartbeatStreaming = shouldHideHeartbeatChatOutput(
|
|
clientRunId,
|
|
sourceRunId,
|
|
);
|
|
if (!text || shouldSuppressSilent || shouldSuppressHeartbeatStreaming) {
|
|
return;
|
|
}
|
|
|
|
const now = Date.now();
|
|
const delta = resolveBroadcastDelta({
|
|
text,
|
|
previousBroadcastText: chatRunState.deltaLastBroadcastText.get(clientRunId),
|
|
});
|
|
if (!delta) {
|
|
return;
|
|
}
|
|
const spawnedBy = resolveSpawnedBy(sessionKey);
|
|
const flushPayload = {
|
|
runId: clientRunId,
|
|
sessionKey,
|
|
...(agentId ? { agentId } : {}),
|
|
...(spawnedBy && { spawnedBy }),
|
|
seq,
|
|
state: "delta" as const,
|
|
deltaText: delta.deltaText,
|
|
...(delta.replace ? { replace: true as const } : {}),
|
|
message: {
|
|
role: "assistant",
|
|
content: [{ type: "text", text }],
|
|
timestamp: now,
|
|
},
|
|
};
|
|
emitFirstAssistantChatSendTiming(
|
|
opts?.firstAssistantTimingEntry ?? chatRunState.registry.peek(sourceRunId),
|
|
);
|
|
sendChatPayload(sessionKey, flushPayload, {
|
|
agentId,
|
|
controlUiVisible: opts?.controlUiVisible ?? true,
|
|
dropIfSlow: true,
|
|
});
|
|
chatRunState.deltaLastBroadcastLen.set(clientRunId, text.length);
|
|
chatRunState.deltaLastBroadcastText.set(clientRunId, text);
|
|
chatRunState.deltaSentAt.set(clientRunId, now);
|
|
};
|
|
|
|
const sendChatPayload = (
|
|
sessionKey: string,
|
|
payload: unknown,
|
|
opts?: { agentId?: string; controlUiVisible?: boolean; dropIfSlow?: boolean },
|
|
) => {
|
|
const deliverySessionKey = resolveSessionDeliveryKey(sessionKey, opts?.agentId);
|
|
if (opts?.controlUiVisible ?? true) {
|
|
broadcast("chat", payload, { dropIfSlow: opts?.dropIfSlow });
|
|
sendNodeSessionPayloadForAgent(sessionKey, "chat", payload, opts?.agentId);
|
|
return;
|
|
}
|
|
const recipients = sessionMessageSubscribers.get(deliverySessionKey);
|
|
if (recipients.size > 0) {
|
|
broadcastToConnIds("chat", payload, recipients, { dropIfSlow: opts?.dropIfSlow });
|
|
}
|
|
};
|
|
|
|
const emitChatTerminal = (
|
|
sessionKey: string,
|
|
clientRunId: string,
|
|
sourceRunId: string,
|
|
seq: number,
|
|
jobState: "done" | "error" | "aborted",
|
|
error?: unknown,
|
|
stopReason?: string,
|
|
errorKind?: ChatErrorKind,
|
|
opts?: {
|
|
agentId?: string;
|
|
controlUiVisible?: boolean;
|
|
firstAssistantTimingEntry?: ChatRunEntry;
|
|
abortErrorMessage?: string;
|
|
},
|
|
) => {
|
|
const { text, shouldSuppressSilent } = resolveBufferedChatTextState(clientRunId, sourceRunId, {
|
|
suppressLeadFragments: false,
|
|
});
|
|
// Flush any throttled delta so streaming clients receive the complete text
|
|
// before the final event. The 150 ms throttle in emitChatDelta may have
|
|
// suppressed the most recent chunk, leaving the client with stale text.
|
|
// Only flush if the buffered text differs from the last broadcast to avoid duplicates.
|
|
flushBufferedChatDeltaIfNeeded(sessionKey, opts?.agentId, clientRunId, sourceRunId, seq, opts);
|
|
chatRunState.clearRun(clientRunId);
|
|
const spawnedBy = resolveSpawnedBy(sessionKey);
|
|
if (jobState !== "error") {
|
|
const payload = {
|
|
runId: clientRunId,
|
|
sessionKey,
|
|
...(opts?.agentId ? { agentId: opts.agentId } : {}),
|
|
...(spawnedBy && { spawnedBy }),
|
|
seq,
|
|
state: jobState === "done" ? ("final" as const) : ("aborted" as const),
|
|
...(jobState === "aborted" && opts?.abortErrorMessage
|
|
? { errorMessage: opts.abortErrorMessage }
|
|
: {}),
|
|
...(stopReason && { stopReason }),
|
|
message:
|
|
text && !shouldSuppressSilent
|
|
? {
|
|
role: "assistant",
|
|
content: [{ type: "text", text }],
|
|
timestamp: Date.now(),
|
|
}
|
|
: undefined,
|
|
};
|
|
sendChatPayload(sessionKey, payload, opts);
|
|
return;
|
|
}
|
|
const payload = {
|
|
runId: clientRunId,
|
|
sessionKey,
|
|
...(opts?.agentId ? { agentId: opts.agentId } : {}),
|
|
...(spawnedBy && { spawnedBy }),
|
|
seq,
|
|
state: "error" as const,
|
|
errorMessage: error ? formatForLog(error) : undefined,
|
|
message: buildChatErrorMessage(error),
|
|
...(errorKind && { errorKind }),
|
|
...(stopReason && { stopReason }),
|
|
};
|
|
sendChatPayload(sessionKey, payload, opts);
|
|
};
|
|
|
|
const sendAgentPayload = (
|
|
sessionKey: string | undefined,
|
|
payload: AgentEventPayload & { spawnedBy?: string },
|
|
opts?: { agentId?: string; controlUiVisible?: boolean; dropIfSlow?: boolean },
|
|
) => {
|
|
if (opts?.controlUiVisible ?? true) {
|
|
broadcast("agent", payload);
|
|
if (sessionKey) {
|
|
sendNodeSessionPayloadForAgent(sessionKey, "agent", payload, opts?.agentId);
|
|
}
|
|
return;
|
|
}
|
|
if (!sessionKey) {
|
|
return;
|
|
}
|
|
const recipients = sessionMessageSubscribers.get(
|
|
resolveSessionDeliveryKey(sessionKey, opts?.agentId),
|
|
);
|
|
if (recipients.size > 0) {
|
|
broadcastToConnIds("agent", payload, recipients, { dropIfSlow: opts?.dropIfSlow });
|
|
}
|
|
};
|
|
|
|
const sendNodeAgentPayload = (
|
|
sessionKey: string | undefined,
|
|
payload: AgentEventPayload & { spawnedBy?: string },
|
|
agentId?: string,
|
|
) => {
|
|
if (sessionKey) {
|
|
sendNodeSessionPayloadForAgent(sessionKey, "agent", payload, agentId);
|
|
}
|
|
};
|
|
|
|
const flushBufferedAgentDeltaIfNeeded = (
|
|
clientRunId: string,
|
|
stream?: AgentTextThrottleStream,
|
|
) => {
|
|
const keys = stream
|
|
? [agentTextThrottleKey(clientRunId, stream)]
|
|
: agentTextThrottleKeys(clientRunId);
|
|
const bufferedEntries = keys.flatMap((key) => {
|
|
const buffered = chatRunState.bufferedAgentEvents.get(key);
|
|
if (!buffered) {
|
|
return [];
|
|
}
|
|
return [{ key, buffered }];
|
|
});
|
|
bufferedEntries.sort((a, b) => a.buffered.payload.seq - b.buffered.payload.seq);
|
|
for (const { key, buffered } of bufferedEntries) {
|
|
sendAgentPayload(buffered.sessionKey, buffered.payload, { agentId: buffered.agentId });
|
|
chatRunState.bufferedAgentEvents.delete(key);
|
|
chatRunState.agentDeltaSentAt.set(key, Date.now());
|
|
}
|
|
return bufferedEntries.length > 0;
|
|
};
|
|
|
|
const resolveAgentTextThrottleStream = (
|
|
evt: AgentEventPayload,
|
|
): AgentTextThrottleStream | null => {
|
|
if (evt.stream === "assistant") {
|
|
return "assistant";
|
|
}
|
|
if (evt.stream === "thinking") {
|
|
return "thinking";
|
|
}
|
|
return null;
|
|
};
|
|
|
|
const isAgentTextThrottleEvent = (evt: AgentEventPayload) =>
|
|
resolveAgentTextThrottleStream(evt) !== null && typeof evt.data?.text === "string";
|
|
|
|
const shouldCoalesceAgentTextEvent = (evt: AgentEventPayload) =>
|
|
isAgentTextThrottleEvent(evt) &&
|
|
typeof evt.data.delta === "string" &&
|
|
evt.data.delta.length > 0 &&
|
|
!(Array.isArray(evt.data.mediaUrls) && evt.data.mediaUrls.length > 0) &&
|
|
typeof evt.data.mediaUrl !== "string" &&
|
|
evt.data.replace !== true &&
|
|
(evt.stream !== "assistant" || !shouldSuppressAssistantEventForLiveChat(evt.data));
|
|
|
|
const shouldAdvanceAgentTextThrottle = (evt: AgentEventPayload) =>
|
|
isAgentTextThrottleEvent(evt) &&
|
|
(typeof evt.data.delta === "string" || evt.data.replace === true);
|
|
|
|
const buildBufferedAgentEvent = (
|
|
sessionKey: string | undefined,
|
|
agentId: string | undefined,
|
|
payload: AgentEventPayload & { spawnedBy?: string },
|
|
): BufferedAgentEvent => (sessionKey ? { sessionKey, agentId, payload } : { agentId, payload });
|
|
|
|
const mergeBufferedAgentPayload = (
|
|
previous: BufferedAgentEvent,
|
|
next: BufferedAgentEvent,
|
|
): BufferedAgentEvent => {
|
|
if (previous.payload.stream !== next.payload.stream) {
|
|
return next;
|
|
}
|
|
const previousDelta = previous.payload.data.delta;
|
|
const nextDelta = next.payload.data.delta;
|
|
if (typeof previousDelta !== "string" || typeof nextDelta !== "string") {
|
|
return next;
|
|
}
|
|
return {
|
|
...next,
|
|
payload: {
|
|
...next.payload,
|
|
data: {
|
|
...next.payload.data,
|
|
delta: `${previousDelta}${nextDelta}`,
|
|
},
|
|
},
|
|
};
|
|
};
|
|
|
|
const sendOrBufferAgentTextEvent = (
|
|
clientRunId: string,
|
|
sessionKey: string | undefined,
|
|
agentId: string | undefined,
|
|
payload: AgentEventPayload & { spawnedBy?: string },
|
|
) => {
|
|
const stream = resolveAgentTextThrottleStream(payload);
|
|
if (!stream) {
|
|
sendAgentPayload(sessionKey, payload, { agentId });
|
|
return;
|
|
}
|
|
const now = Date.now();
|
|
const key = agentTextThrottleKey(clientRunId, stream);
|
|
const last = chatRunState.agentDeltaSentAt.get(key);
|
|
if (last !== undefined && now - last < 150) {
|
|
const nextBuffered = buildBufferedAgentEvent(sessionKey, agentId, payload);
|
|
const buffered = chatRunState.bufferedAgentEvents.get(key);
|
|
chatRunState.bufferedAgentEvents.set(
|
|
key,
|
|
buffered ? mergeBufferedAgentPayload(buffered, nextBuffered) : nextBuffered,
|
|
);
|
|
return;
|
|
}
|
|
flushBufferedAgentDeltaIfNeeded(clientRunId);
|
|
sendAgentPayload(sessionKey, payload, { agentId });
|
|
chatRunState.agentDeltaSentAt.set(key, now);
|
|
};
|
|
|
|
const resolveToolVerboseLevel = (runId: string, sessionKey?: string) => {
|
|
const runContext = getAgentRunContext(runId);
|
|
const runVerbose = normalizeVerboseLevel(runContext?.verboseLevel);
|
|
if (!sessionKey) {
|
|
return runVerbose ?? "off";
|
|
}
|
|
try {
|
|
const { cfg, entry } = loadSessionEntry(sessionKey);
|
|
const sessionVerbose = normalizeVerboseLevel(entry?.verboseLevel);
|
|
const sessionUpdatedAt = typeof entry?.updatedAt === "number" ? entry.updatedAt : undefined;
|
|
const sessionChangedAfterRunStarted =
|
|
sessionUpdatedAt !== undefined &&
|
|
runContext?.registeredAt !== undefined &&
|
|
sessionUpdatedAt >= runContext.registeredAt;
|
|
if (sessionVerbose && (!runVerbose || sessionChangedAfterRunStarted)) {
|
|
return sessionVerbose;
|
|
}
|
|
if (runVerbose) {
|
|
return runVerbose;
|
|
}
|
|
const defaultVerbose = normalizeVerboseLevel(cfg.agents?.defaults?.verboseDefault);
|
|
return defaultVerbose ?? "off";
|
|
} catch {
|
|
return runVerbose ?? "off";
|
|
}
|
|
};
|
|
|
|
return (event: AgentEventPayload) => {
|
|
const evt = event as AgentEventRuntimePayload;
|
|
if (!shouldProcessOwnedEvent(evt)) {
|
|
return;
|
|
}
|
|
const lifecyclePhase =
|
|
evt.stream === "lifecycle" && typeof evt.data?.phase === "string" ? evt.data.phase : null;
|
|
|
|
const chatLink = evt.contextClaimId ? undefined : chatRunState.registry.peek(evt.runId);
|
|
const sessionAgentId = chatLink?.agentId ?? evt.agentId;
|
|
const eventSessionKey =
|
|
evt.deliverySessionKey ??
|
|
(typeof evt.sessionKey === "string" && evt.sessionKey.trim() ? evt.sessionKey : undefined);
|
|
const runContext = getAgentRunContext(evt.runId);
|
|
const activeLifecycleGeneration = resolveActiveLifecycleGenerationForRun(evt.runId);
|
|
const isControlUiVisible = evt.controlUiVisible ?? runContext?.isControlUiVisible ?? true;
|
|
const isHeartbeat = runContext?.isHeartbeat;
|
|
const sessionKey =
|
|
chatLink?.sessionKey ?? eventSessionKey ?? resolveSessionKeyForRun(evt.runId);
|
|
const restartRecoverySessionKey = eventSessionKey ?? sessionKey;
|
|
const restartRecoveryAgentId = evt.agentId ?? sessionAgentId;
|
|
const clientRunId = chatLink?.clientRunId ?? evt.runId;
|
|
const eventRunId = chatLink?.clientRunId ?? evt.runId;
|
|
const eventForClients = chatLink ? { ...evt, runId: eventRunId } : evt;
|
|
const isAborted =
|
|
isChatAbortMarkerCurrent(chatRunState.abortedRuns.get(clientRunId), chatLink) ||
|
|
isChatAbortMarkerCurrent(chatRunState.abortedRuns.get(evt.runId), chatLink);
|
|
|
|
const restartRecoveryState = restartRecoverySessionKey
|
|
? resolveRestartRecoveryLifecycleState(restartRecoverySessionKey, restartRecoveryAgentId, evt)
|
|
: undefined;
|
|
const suppressRestartRecoveryLifecycle =
|
|
lifecyclePhase !== null &&
|
|
(Boolean(
|
|
evt.lifecycleGeneration &&
|
|
activeLifecycleGeneration &&
|
|
evt.lifecycleGeneration !== activeLifecycleGeneration,
|
|
) ||
|
|
restartRecoveryState?.suppress === true);
|
|
if (suppressRestartRecoveryLifecycle) {
|
|
clearPendingTerminalLifecycleError(evt.runId, evt.lifecycleGeneration);
|
|
if (lifecyclePhase === "end" || lifecyclePhase === "error") {
|
|
finalizeLifecycleEvent(evt, {
|
|
suppressRestartRecoveryProjection: true,
|
|
restartRecoveryState,
|
|
});
|
|
}
|
|
return;
|
|
}
|
|
if (lifecyclePhase !== null && lifecyclePhase !== "error") {
|
|
clearPendingTerminalLifecycleError(evt.runId);
|
|
}
|
|
|
|
// Include sessionKey so Control UI can filter tool streams per session.
|
|
const spawnedBy = sessionKey ? resolveSpawnedBy(sessionKey) : null;
|
|
const agentPayload = sessionKey
|
|
? {
|
|
...eventForClients,
|
|
sessionKey,
|
|
...(sessionAgentId ? { agentId: sessionAgentId } : {}),
|
|
...(spawnedBy && { spawnedBy }),
|
|
...(isHeartbeat !== undefined && { isHeartbeat }),
|
|
}
|
|
: {
|
|
...eventForClients,
|
|
...(isHeartbeat !== undefined && { isHeartbeat }),
|
|
};
|
|
const hasSessionMessageSubscribers = sessionKey
|
|
? sessionMessageSubscribers.get(resolveSessionDeliveryKey(sessionKey, sessionAgentId)).size >
|
|
0
|
|
: false;
|
|
const last = agentRunSeq.get(evt.runId) ?? 0;
|
|
const isToolEvent = evt.stream === "tool";
|
|
const isItemEvent = evt.stream === "item";
|
|
const toolVerbose = isToolEvent ? resolveToolVerboseLevel(evt.runId, sessionKey) : "off";
|
|
const suppressHeartbeatToolEvents =
|
|
isToolEvent && shouldSuppressHeartbeatToolEvents(clientRunId, evt.runId);
|
|
const shouldCoalesceAgentEvent = shouldCoalesceAgentTextEvent(evt);
|
|
// Channel/node subscribers respect verbose; authenticated Control UI
|
|
// recipients need tool result payloads to render live tool cards.
|
|
const channelToolPayload =
|
|
isToolEvent && toolVerbose !== "full"
|
|
? (() => {
|
|
const data = evt.data ? { ...evt.data } : {};
|
|
delete data.result;
|
|
delete data.partialResult;
|
|
return { ...agentPayload, data };
|
|
})()
|
|
: agentPayload;
|
|
if (last > 0 && evt.seq !== last + 1 && isControlUiVisible) {
|
|
flushBufferedAgentDeltaIfNeeded(clientRunId);
|
|
broadcast("agent", {
|
|
runId: eventRunId,
|
|
stream: "error",
|
|
ts: Date.now(),
|
|
sessionKey,
|
|
...(spawnedBy && { spawnedBy }),
|
|
...(isHeartbeat !== undefined && { isHeartbeat }),
|
|
data: {
|
|
reason: "seq gap",
|
|
expected: last + 1,
|
|
received: evt.seq,
|
|
},
|
|
});
|
|
}
|
|
agentRunSeq.set(evt.runId, evt.seq);
|
|
if (evt.stream === "assistant") {
|
|
updateRunToolErrorSummary?.({ runId: evt.runId, clientRunId, summary: undefined });
|
|
}
|
|
if (isToolEvent) {
|
|
const toolPhase = typeof evt.data?.phase === "string" ? evt.data.phase : "";
|
|
if (toolPhase === "start") {
|
|
updateRunToolErrorSummary?.({ runId: evt.runId, clientRunId, summary: undefined });
|
|
} else if (toolPhase === "result") {
|
|
updateRunToolErrorSummary?.({
|
|
runId: evt.runId,
|
|
clientRunId,
|
|
summary: readToolValidationErrorSummary(evt.data?.toolErrorSummary),
|
|
});
|
|
}
|
|
// Flush pending assistant text before tool-start events so clients can
|
|
// render complete pre-tool text above tool cards (not truncated by delta throttle).
|
|
if (
|
|
toolPhase === "start" &&
|
|
(isControlUiVisible || hasSessionMessageSubscribers) &&
|
|
sessionKey &&
|
|
!isAborted &&
|
|
!suppressHeartbeatToolEvents
|
|
) {
|
|
flushBufferedChatDeltaIfNeeded(
|
|
sessionKey,
|
|
sessionAgentId,
|
|
clientRunId,
|
|
evt.runId,
|
|
evt.seq,
|
|
{
|
|
controlUiVisible: isControlUiVisible,
|
|
},
|
|
);
|
|
flushBufferedAgentDeltaIfNeeded(clientRunId);
|
|
}
|
|
// Always broadcast tool events to registered WS recipients with
|
|
// tool-events capability, regardless of verboseLevel. The verbose
|
|
// setting only controls whether tool details are sent as channel
|
|
// messages to messaging surfaces (Telegram, Discord, etc.).
|
|
const runToolRecipients = toolEventRecipients.get(evt.runId);
|
|
if (
|
|
isControlUiVisible &&
|
|
!suppressHeartbeatToolEvents &&
|
|
runToolRecipients &&
|
|
runToolRecipients.size > 0
|
|
) {
|
|
broadcastToConnIds(
|
|
"agent",
|
|
sessionKey
|
|
? {
|
|
...agentPayload,
|
|
...buildSessionEventSnapshot(sessionKey, undefined, sessionAgentId),
|
|
}
|
|
: agentPayload,
|
|
runToolRecipients,
|
|
);
|
|
}
|
|
if (!isControlUiVisible && sessionKey && !suppressHeartbeatToolEvents) {
|
|
sendAgentPayload(
|
|
sessionKey,
|
|
{ ...agentPayload, ...buildSessionEventSnapshot(sessionKey, undefined, sessionAgentId) },
|
|
{ agentId: sessionAgentId, controlUiVisible: false, dropIfSlow: true },
|
|
);
|
|
}
|
|
// Session subscribers power operator UIs that attach to an existing
|
|
// in-flight session after the run has already started. Those clients do
|
|
// not know the runId in advance, so they cannot register as run-scoped
|
|
// tool recipients. Mirror tool lifecycle onto a session-scoped event so
|
|
// they can render live pending tool cards without polling history.
|
|
if (isControlUiVisible && sessionKey && !suppressHeartbeatToolEvents) {
|
|
const sessionSubscribers = excludeConnIds(
|
|
sessionEventSubscribers.getAll(),
|
|
runToolRecipients,
|
|
);
|
|
if (sessionSubscribers.size > 0) {
|
|
broadcastToConnIds(
|
|
"session.tool",
|
|
{
|
|
...agentPayload,
|
|
...buildSessionEventSnapshot(sessionKey, undefined, sessionAgentId),
|
|
},
|
|
sessionSubscribers,
|
|
{ dropIfSlow: true },
|
|
);
|
|
}
|
|
}
|
|
} else {
|
|
const itemPhase = isItemEvent && typeof evt.data?.phase === "string" ? evt.data.phase : "";
|
|
if (
|
|
itemPhase === "start" &&
|
|
(isControlUiVisible || hasSessionMessageSubscribers) &&
|
|
!isAborted
|
|
) {
|
|
if (sessionKey) {
|
|
flushBufferedChatDeltaIfNeeded(
|
|
sessionKey,
|
|
sessionAgentId,
|
|
clientRunId,
|
|
evt.runId,
|
|
evt.seq,
|
|
{
|
|
controlUiVisible: isControlUiVisible,
|
|
},
|
|
);
|
|
}
|
|
flushBufferedAgentDeltaIfNeeded(clientRunId);
|
|
}
|
|
if (isControlUiVisible) {
|
|
if (shouldCoalesceAgentEvent) {
|
|
sendOrBufferAgentTextEvent(clientRunId, sessionKey, sessionAgentId, agentPayload);
|
|
} else {
|
|
flushBufferedAgentDeltaIfNeeded(clientRunId);
|
|
sendAgentPayload(sessionKey, agentPayload, {
|
|
agentId: sessionAgentId,
|
|
controlUiVisible: isControlUiVisible,
|
|
});
|
|
const textThrottleStream = resolveAgentTextThrottleStream(evt);
|
|
if (textThrottleStream && shouldAdvanceAgentTextThrottle(evt)) {
|
|
chatRunState.agentDeltaSentAt.set(
|
|
agentTextThrottleKey(clientRunId, textThrottleStream),
|
|
Date.now(),
|
|
);
|
|
}
|
|
}
|
|
} else if (
|
|
sessionKey &&
|
|
hasSessionMessageSubscribers &&
|
|
(shouldMirrorAgentEventToHiddenSessionMessages(evt) ||
|
|
(!isAborted &&
|
|
evt.stream === "assistant" &&
|
|
shouldMirrorAssistantEventToHiddenSessionMessages(evt.data)))
|
|
) {
|
|
sendAgentPayload(
|
|
sessionKey,
|
|
{ ...agentPayload, ...buildSessionEventSnapshot(sessionKey, undefined, sessionAgentId) },
|
|
{ agentId: sessionAgentId, controlUiVisible: false, dropIfSlow: true },
|
|
);
|
|
}
|
|
if (!isControlUiVisible && isItemEvent && sessionKey && hasSessionMessageSubscribers) {
|
|
sendAgentPayload(
|
|
sessionKey,
|
|
{ ...agentPayload, ...buildSessionEventSnapshot(sessionKey, undefined, sessionAgentId) },
|
|
{ agentId: sessionAgentId, controlUiVisible: false, dropIfSlow: true },
|
|
);
|
|
}
|
|
}
|
|
|
|
if ((isControlUiVisible || hasSessionMessageSubscribers) && sessionKey) {
|
|
// Send tool events to node/channel subscribers only when verbose is enabled;
|
|
// WS clients already received the event above via broadcastToConnIds.
|
|
if (
|
|
isControlUiVisible &&
|
|
isToolEvent &&
|
|
!suppressHeartbeatToolEvents &&
|
|
toolVerbose !== "off"
|
|
) {
|
|
sendNodeAgentPayload(
|
|
sessionKey,
|
|
projectToolSearchCodeEventForChannelPayload({
|
|
...channelToolPayload,
|
|
...buildSessionEventSnapshot(sessionKey, undefined, sessionAgentId),
|
|
}),
|
|
sessionAgentId,
|
|
);
|
|
}
|
|
const assistantLiveChatInput =
|
|
evt.stream === "assistant" ? resolveAssistantLiveChatInput(evt.data) : undefined;
|
|
if (
|
|
!isAborted &&
|
|
assistantLiveChatInput &&
|
|
!shouldSuppressAssistantEventForLiveChat(evt.data)
|
|
) {
|
|
emitChatDelta(
|
|
sessionKey,
|
|
sessionAgentId,
|
|
clientRunId,
|
|
evt.runId,
|
|
evt.seq,
|
|
assistantLiveChatInput.text,
|
|
assistantLiveChatInput.delta,
|
|
{
|
|
controlUiVisible: isControlUiVisible,
|
|
},
|
|
);
|
|
}
|
|
}
|
|
|
|
if (lifecyclePhase === "error") {
|
|
clearBufferedChatState(clientRunId);
|
|
const skipChatErrorFinal = isChatSendRunActive(evt.runId) && !chatLink;
|
|
const isFallbackExhaustedFailure = evt.data?.fallbackExhaustedFailure === true;
|
|
// Per-attempt provider errors keep the retry grace so fallback can reuse
|
|
// the runId. Once the runner marks fallback as exhausted, clear chat state
|
|
// immediately so webchat sessions do not stay in progress until the timer.
|
|
if (isAborted || isFallbackExhaustedFailure || lifecycleErrorRetryGraceMs <= 0) {
|
|
finalizeLifecycleEvent(evt, { skipChatErrorFinal, restartRecoveryState });
|
|
} else {
|
|
scheduleTerminalLifecycleError(evt, { skipChatErrorFinal, restartRecoveryState });
|
|
}
|
|
return;
|
|
}
|
|
|
|
if (lifecyclePhase === "end") {
|
|
finalizeLifecycleEvent(evt, { restartRecoveryState });
|
|
return;
|
|
}
|
|
|
|
if (sessionKey && lifecyclePhase === "start") {
|
|
void persistGatewaySessionLifecycleEvent({
|
|
sessionKey,
|
|
agentId: sessionAgentId,
|
|
event: evt,
|
|
}).catch((err: unknown) => {
|
|
// Surface the swallowed start-phase persistence failure: a silent write
|
|
// failure drops the run's start marker from restart-recovery accounting
|
|
// with no operator trace, matching the terminal-phase log below.
|
|
logError(
|
|
`gateway: start session persistence failed session=${formatForLog(sessionKey)} run=${formatForLog(evt.runId)} error=${formatForLog(err)}`,
|
|
);
|
|
});
|
|
const sessionEventConnIds = sessionEventSubscribers.getAll();
|
|
if (sessionEventConnIds.size > 0) {
|
|
broadcastToConnIds(
|
|
"sessions.changed",
|
|
{
|
|
sessionKey,
|
|
...(sessionAgentId ? { agentId: sessionAgentId } : {}),
|
|
phase: lifecyclePhase,
|
|
runId: evt.runId,
|
|
...(eventRunId !== evt.runId ? { clientRunId: eventRunId } : {}),
|
|
ts: evt.ts,
|
|
...buildSessionEventSnapshot(sessionKey, evt, sessionAgentId, true),
|
|
},
|
|
sessionEventConnIds,
|
|
{ dropIfSlow: true },
|
|
);
|
|
}
|
|
}
|
|
};
|
|
}
|