mirror of
https://github.com/openclaw/openclaw.git
synced 2026-05-15 01:50:45 +00:00
1982 lines
65 KiB
TypeScript
1982 lines
65 KiB
TypeScript
import fs from "node:fs/promises";
|
|
import { hasConfiguredModelFallbacks, resolveSessionAgentId } from "../../agents/agent-scope.js";
|
|
import { resolveContextTokensForModel } from "../../agents/context.js";
|
|
import { DEFAULT_CONTEXT_TOKENS } from "../../agents/defaults.js";
|
|
import { resolveModelAuthMode } from "../../agents/model-auth.js";
|
|
import { isCliProvider } from "../../agents/model-selection.js";
|
|
import { queueEmbeddedPiMessage } from "../../agents/pi-embedded-runner/runs.js";
|
|
import { deriveContextPromptTokens, hasNonzeroUsage, normalizeUsage } from "../../agents/usage.js";
|
|
import { enqueueCommitmentExtraction } from "../../commitments/runtime.js";
|
|
import type { OpenClawConfig } from "../../config/config.js";
|
|
import {
|
|
loadSessionStore,
|
|
resolveSessionPluginStatusLines,
|
|
resolveSessionPluginTraceLines,
|
|
type SessionEntry,
|
|
updateSessionStoreEntry,
|
|
} from "../../config/sessions.js";
|
|
import type { TypingMode } from "../../config/types.js";
|
|
import { resolveSessionTranscriptCandidates } from "../../gateway/session-utils.fs.js";
|
|
import { logVerbose } from "../../globals.js";
|
|
import { emitAgentEvent } from "../../infra/agent-events.js";
|
|
import { emitTrustedDiagnosticEvent, isDiagnosticsEnabled } from "../../infra/diagnostic-events.js";
|
|
import {
|
|
createChildDiagnosticTraceContext,
|
|
freezeDiagnosticTraceContext,
|
|
} from "../../infra/diagnostic-trace-context.js";
|
|
import { measureDiagnosticsTimelineSpan } from "../../infra/diagnostics-timeline.js";
|
|
import { enqueueSystemEvent } from "../../infra/system-events.js";
|
|
import { CommandLaneClearedError, GatewayDrainingError } from "../../process/command-queue.js";
|
|
import { resolveSendPolicy } from "../../sessions/send-policy.js";
|
|
import { normalizeOptionalString } from "../../shared/string-coerce.js";
|
|
import {
|
|
estimateUsageCost,
|
|
formatTokenCount,
|
|
resolveModelCostConfig,
|
|
} from "../../utils/usage-format.js";
|
|
import {
|
|
buildFallbackClearedNotice,
|
|
buildFallbackNotice,
|
|
resolveFallbackTransition,
|
|
} from "../fallback-state.js";
|
|
import {
|
|
markReplyPayloadForSourceSuppressionDelivery,
|
|
setReplyPayloadMetadata,
|
|
} from "../reply-payload.js";
|
|
import type { OriginatingChannelType, TemplateContext } from "../templating.js";
|
|
import { resolveResponseUsageMode, type VerboseLevel } from "../thinking.js";
|
|
import { SILENT_REPLY_TOKEN } from "../tokens.js";
|
|
import type { GetReplyOptions, ReplyPayload } from "../types.js";
|
|
import {
|
|
buildKnownAgentRunFailureReplyPayload,
|
|
runAgentTurnWithFallback,
|
|
} from "./agent-runner-execution.js";
|
|
import {
|
|
createShouldEmitToolOutput,
|
|
createShouldEmitToolResult,
|
|
isAudioPayload,
|
|
signalTypingIfNeeded,
|
|
} from "./agent-runner-helpers.js";
|
|
import { runMemoryFlushIfNeeded, runPreflightCompactionIfNeeded } from "./agent-runner-memory.js";
|
|
import { buildReplyPayloads } from "./agent-runner-payloads.js";
|
|
import {
|
|
appendUnscheduledReminderNote,
|
|
hasSessionRelatedCronJobs,
|
|
hasUnbackedReminderCommitment,
|
|
} from "./agent-runner-reminder-guard.js";
|
|
import { resetReplyRunSession } from "./agent-runner-session-reset.js";
|
|
import { appendUsageLine, formatResponseUsageLine } from "./agent-runner-usage-line.js";
|
|
import { resolveQueuedReplyExecutionConfig } from "./agent-runner-utils.js";
|
|
import { createAudioAsVoiceBuffer, createBlockReplyPipeline } from "./block-reply-pipeline.js";
|
|
import { resolveEffectiveBlockStreamingConfig } from "./block-streaming.js";
|
|
import { createFollowupRunner } from "./followup-runner.js";
|
|
import { resolveOriginMessageProvider, resolveOriginMessageTo } from "./origin-routing.js";
|
|
import { drainPendingToolTasks } from "./pending-tool-task-drain.js";
|
|
import { readPostCompactionContext } from "./post-compaction-context.js";
|
|
import { resolveActiveRunQueueAction } from "./queue-policy.js";
|
|
import {
|
|
enqueueFollowupRun,
|
|
refreshQueuedFollowupSession,
|
|
resolvePiSteeringModeForQueueMode,
|
|
scheduleFollowupDrain,
|
|
type FollowupRun,
|
|
type QueueSettings,
|
|
} from "./queue.js";
|
|
import { createReplyMediaContext } from "./reply-media-paths.js";
|
|
import {
|
|
createReplyOperation,
|
|
ReplyRunAlreadyActiveError,
|
|
replyRunRegistry,
|
|
type ReplyOperation,
|
|
} from "./reply-run-registry.js";
|
|
import { createReplyToModeFilterForChannel, resolveReplyToMode } from "./reply-threading.js";
|
|
import { incrementRunCompactionCount, persistRunSessionUsage } from "./session-run-accounting.js";
|
|
import { resolveSourceReplyVisibilityPolicy } from "./source-reply-delivery-mode.js";
|
|
import { createTypingSignaler } from "./typing-mode.js";
|
|
import type { TypingController } from "./typing.js";
|
|
|
|
const BLOCK_REPLY_SEND_TIMEOUT_MS = 15_000;
|
|
|
|
function markBeforeAgentRunBlockedPayloads(payloads: ReplyPayload[]): ReplyPayload[] {
|
|
return payloads.map((payload) =>
|
|
setReplyPayloadMetadata(payload, { beforeAgentRunBlocked: true }),
|
|
);
|
|
}
|
|
|
|
function buildInlinePluginStatusPayload(params: {
|
|
entry: SessionEntry | undefined;
|
|
includeTraceLines: boolean;
|
|
}): ReplyPayload | undefined {
|
|
const statusLines =
|
|
params.entry?.verboseLevel && params.entry.verboseLevel !== "off"
|
|
? resolveSessionPluginStatusLines(params.entry)
|
|
: [];
|
|
const traceLines =
|
|
params.includeTraceLines &&
|
|
(params.entry?.traceLevel === "on" || params.entry?.traceLevel === "raw")
|
|
? resolveSessionPluginTraceLines(params.entry)
|
|
: [];
|
|
const lines = [...statusLines, ...traceLines];
|
|
if (lines.length === 0) {
|
|
return undefined;
|
|
}
|
|
return { text: lines.join("\n") };
|
|
}
|
|
|
|
function formatRawTraceBlock(title: string, value: string | undefined): string {
|
|
const body = value?.trim() ? escapeTraceFence(value) : "<empty>";
|
|
return `🔎 ${title}:\n~~~text\n${body}\n~~~`;
|
|
}
|
|
|
|
function escapeTraceFence(value: string): string {
|
|
return value.replace(/^~~~/gm, "\\~~~");
|
|
}
|
|
|
|
function hasTraceUsageFields(
|
|
usage:
|
|
| {
|
|
input?: number;
|
|
output?: number;
|
|
cacheRead?: number;
|
|
cacheWrite?: number;
|
|
total?: number;
|
|
}
|
|
| undefined,
|
|
): boolean {
|
|
if (!usage) {
|
|
return false;
|
|
}
|
|
return ["input", "output", "cacheRead", "cacheWrite", "total"].some((key) => {
|
|
const value = usage[key as keyof typeof usage];
|
|
return typeof value === "number" && Number.isFinite(value);
|
|
});
|
|
}
|
|
|
|
function formatTraceUsageLine(label: string, value: number | undefined): string {
|
|
return `${label}=${typeof value === "number" && Number.isFinite(value) ? `${value.toLocaleString()} tok (${formatTokenCount(value)})` : "n/a"}`;
|
|
}
|
|
|
|
function formatUsageTraceBlock(
|
|
title: string,
|
|
usage:
|
|
| {
|
|
input?: number;
|
|
output?: number;
|
|
cacheRead?: number;
|
|
cacheWrite?: number;
|
|
total?: number;
|
|
}
|
|
| undefined,
|
|
): string | undefined {
|
|
if (!hasTraceUsageFields(usage)) {
|
|
return undefined;
|
|
}
|
|
return `🔎 ${title}:\n~~~text\n${[
|
|
formatTraceUsageLine("input", usage?.input),
|
|
formatTraceUsageLine("output", usage?.output),
|
|
formatTraceUsageLine("cacheRead", usage?.cacheRead),
|
|
formatTraceUsageLine("cacheWrite", usage?.cacheWrite),
|
|
formatTraceUsageLine("total", usage?.total),
|
|
].join("\n")}\n~~~`;
|
|
}
|
|
|
|
type TraceAttemptView = {
|
|
provider: string;
|
|
model: string;
|
|
result: string;
|
|
reason?: string;
|
|
stage?: string;
|
|
elapsedMs?: number;
|
|
status?: number;
|
|
};
|
|
|
|
type TraceExecutionView = {
|
|
winnerProvider?: string;
|
|
winnerModel?: string;
|
|
attempts?: TraceAttemptView[];
|
|
fallbackUsed?: boolean;
|
|
runner?: "embedded" | "cli";
|
|
};
|
|
|
|
type TracePromptSegmentView = {
|
|
key: string;
|
|
chars: number;
|
|
};
|
|
|
|
type TraceToolSummaryView = {
|
|
calls: number;
|
|
tools: string[];
|
|
failures?: number;
|
|
totalToolTimeMs?: number;
|
|
};
|
|
|
|
type TraceCompletionView = {
|
|
finishReason?: string;
|
|
stopReason?: string;
|
|
refusal?: boolean;
|
|
};
|
|
|
|
type TraceContextManagementView = {
|
|
sessionCompactions?: number;
|
|
lastTurnCompactions?: number;
|
|
preflightCompactionApplied?: boolean;
|
|
postCompactionContextInjected?: boolean;
|
|
};
|
|
|
|
function formatTraceScalar(value: string | number | boolean | undefined): string | undefined {
|
|
if (typeof value === "boolean") {
|
|
return value ? "yes" : "no";
|
|
}
|
|
if (typeof value === "number") {
|
|
return Number.isFinite(value) ? value.toLocaleString() : undefined;
|
|
}
|
|
const trimmed = normalizeOptionalString(value);
|
|
return trimmed ?? undefined;
|
|
}
|
|
|
|
function formatKeyValueTraceBlock(
|
|
title: string,
|
|
fields: Array<[string, string | number | boolean | undefined]>,
|
|
): string | undefined {
|
|
const lines = fields.flatMap(([key, rawValue]) => {
|
|
const value = formatTraceScalar(rawValue);
|
|
return value ? [`${key}=${value}`] : [];
|
|
});
|
|
if (lines.length === 0) {
|
|
return undefined;
|
|
}
|
|
return `🔎 ${title}:\n~~~text\n${lines.join("\n")}\n~~~`;
|
|
}
|
|
|
|
function inferFallbackAttemptResult(attempt: { reason?: string; status?: number }): string {
|
|
if (attempt.reason === "timeout") {
|
|
return "timeout";
|
|
}
|
|
return "candidate_failed";
|
|
}
|
|
|
|
function mergeExecutionTrace(params: {
|
|
fallbackAttempts?: Array<{
|
|
provider: string;
|
|
model: string;
|
|
reason?: string;
|
|
status?: number;
|
|
}>;
|
|
executionTrace?: {
|
|
winnerProvider?: string;
|
|
winnerModel?: string;
|
|
attempts?: TraceAttemptView[];
|
|
fallbackUsed?: boolean;
|
|
runner?: "embedded" | "cli";
|
|
};
|
|
provider?: string;
|
|
model?: string;
|
|
runner: "embedded" | "cli";
|
|
}): TraceExecutionView | undefined {
|
|
const attempts: TraceAttemptView[] = [
|
|
...(params.fallbackAttempts ?? []).map((attempt) =>
|
|
Object.assign(
|
|
{
|
|
provider: attempt.provider,
|
|
model: attempt.model,
|
|
result: inferFallbackAttemptResult(attempt),
|
|
},
|
|
attempt.reason ? { reason: attempt.reason } : {},
|
|
typeof attempt.status === `number` ? { status: attempt.status } : {},
|
|
),
|
|
),
|
|
...(params.executionTrace?.attempts ?? []),
|
|
];
|
|
const winnerProvider =
|
|
params.executionTrace?.winnerProvider ?? normalizeOptionalString(params.provider);
|
|
const winnerModel = params.executionTrace?.winnerModel ?? normalizeOptionalString(params.model);
|
|
if (
|
|
winnerProvider &&
|
|
winnerModel &&
|
|
!attempts.some(
|
|
(attempt) =>
|
|
attempt.provider === winnerProvider &&
|
|
attempt.model === winnerModel &&
|
|
attempt.result === "success",
|
|
)
|
|
) {
|
|
attempts.push({
|
|
provider: winnerProvider,
|
|
model: winnerModel,
|
|
result: "success",
|
|
});
|
|
}
|
|
if (!winnerProvider && !winnerModel && attempts.length === 0) {
|
|
return undefined;
|
|
}
|
|
return {
|
|
winnerProvider,
|
|
winnerModel,
|
|
attempts: attempts.length > 0 ? attempts : undefined,
|
|
fallbackUsed: params.executionTrace?.fallbackUsed ?? attempts.length > 1,
|
|
runner: params.executionTrace?.runner ?? params.runner,
|
|
};
|
|
}
|
|
|
|
function formatExecutionResultTraceBlock(
|
|
executionTrace: TraceExecutionView | undefined,
|
|
): string | undefined {
|
|
if (!executionTrace?.winnerProvider && !executionTrace?.winnerModel) {
|
|
return undefined;
|
|
}
|
|
return formatKeyValueTraceBlock("Execution Result", [
|
|
[
|
|
"winner",
|
|
executionTrace.winnerProvider && executionTrace.winnerModel
|
|
? `${executionTrace.winnerProvider}/${executionTrace.winnerModel}`
|
|
: undefined,
|
|
],
|
|
["fallbackUsed", executionTrace.fallbackUsed],
|
|
["attempts", executionTrace.attempts?.length],
|
|
["runner", executionTrace.runner],
|
|
]);
|
|
}
|
|
|
|
function formatFallbackChainTraceBlock(
|
|
executionTrace: TraceExecutionView | undefined,
|
|
): string | undefined {
|
|
const attempts = executionTrace?.attempts ?? [];
|
|
if (attempts.length <= 1) {
|
|
return undefined;
|
|
}
|
|
const body = attempts
|
|
.map((attempt, index) =>
|
|
[
|
|
`${index + 1}. ${attempt.provider}/${attempt.model}`,
|
|
` result=${attempt.result}`,
|
|
...(attempt.reason ? [` reason=${attempt.reason}`] : []),
|
|
...(attempt.stage ? [` stage=${attempt.stage}`] : []),
|
|
...(typeof attempt.elapsedMs === "number"
|
|
? [` elapsed=${(attempt.elapsedMs / 1000).toFixed(1)}s`]
|
|
: []),
|
|
...(typeof attempt.status === "number" ? [` status=${attempt.status}`] : []),
|
|
].join("\n"),
|
|
)
|
|
.join("\n\n");
|
|
return `🔎 Fallback Chain:\n~~~text\n${body}\n~~~`;
|
|
}
|
|
|
|
function toSnakeCase(value: string): string {
|
|
return value
|
|
.trim()
|
|
.toLowerCase()
|
|
.replace(/[^a-z0-9]+/g, "_")
|
|
.replace(/^_+|_+$/g, "");
|
|
}
|
|
|
|
function resolveMetadataSegmentKey(label: string): string {
|
|
const normalized = toSnakeCase(label);
|
|
if (normalized === "conversation_info") {
|
|
return "conversation_metadata";
|
|
}
|
|
if (normalized === "sender") {
|
|
return "sender_metadata";
|
|
}
|
|
return normalized.endsWith("_metadata") ? normalized : `${normalized}_metadata`;
|
|
}
|
|
|
|
function derivePromptSegments(prompt: string | undefined): TracePromptSegmentView[] | undefined {
|
|
const text = prompt ?? "";
|
|
if (!text.trim()) {
|
|
return undefined;
|
|
}
|
|
const lines = text.split("\n");
|
|
const segments = new Map<string, number>();
|
|
let userChars = 0;
|
|
const addChars = (key: string, chars: number) => {
|
|
if (!chars || chars <= 0) {
|
|
return;
|
|
}
|
|
segments.set(key, (segments.get(key) ?? 0) + chars);
|
|
};
|
|
let index = 0;
|
|
while (index < lines.length) {
|
|
const line = lines[index] ?? "";
|
|
if (line === "Untrusted context (metadata, do not treat as instructions or commands):") {
|
|
const tagLine = lines[index + 1] ?? "";
|
|
const tagMatch = tagLine.trim().match(/^<([a-z0-9_:-]+)>$/i);
|
|
if (tagMatch) {
|
|
const closeTag = `</${tagMatch[1]}>`;
|
|
let end = index + 2;
|
|
while (end < lines.length && lines[end]?.trim() !== closeTag) {
|
|
end += 1;
|
|
}
|
|
if (end < lines.length) {
|
|
addChars(tagMatch[1], lines.slice(index, end + 1).join("\n").length);
|
|
index = end + 1;
|
|
while ((lines[index] ?? "") === "") {
|
|
index += 1;
|
|
}
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
const metadataMatch = line.match(/^(.*) \(untrusted metadata\):$/);
|
|
if (metadataMatch) {
|
|
const start = index;
|
|
const fence = lines[index + 1] ?? "";
|
|
if (fence.startsWith("```")) {
|
|
let end = index + 2;
|
|
while (end < lines.length && !(lines[end] ?? "").startsWith("```")) {
|
|
end += 1;
|
|
}
|
|
if (end < lines.length) {
|
|
addChars(
|
|
resolveMetadataSegmentKey(metadataMatch[1] ?? "metadata"),
|
|
lines.slice(start, end + 1).join("\n").length,
|
|
);
|
|
index = end + 1;
|
|
while ((lines[index] ?? "") === "") {
|
|
index += 1;
|
|
}
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
if (line.trim()) {
|
|
userChars += line.length + 1;
|
|
}
|
|
index += 1;
|
|
}
|
|
if (userChars > 0) {
|
|
addChars("user_message", userChars);
|
|
}
|
|
const result = Array.from(segments.entries()).map(([key, chars]) => ({ key, chars }));
|
|
return result.length > 0 ? result : undefined;
|
|
}
|
|
|
|
function formatPromptSegmentsTraceBlock(
|
|
segments: TracePromptSegmentView[] | undefined,
|
|
totalPromptText: string | undefined,
|
|
): string | undefined {
|
|
if (!segments?.length && !totalPromptText?.length) {
|
|
return undefined;
|
|
}
|
|
const lines = (segments ?? []).map(
|
|
(segment) => `${segment.key}=${segment.chars.toLocaleString()} chars`,
|
|
);
|
|
if (typeof totalPromptText === "string" && totalPromptText.length > 0) {
|
|
lines.push(`totalPromptText=${totalPromptText.length.toLocaleString()} chars`);
|
|
}
|
|
return lines.length > 0 ? `🔎 Prompt Segments:\n~~~text\n${lines.join("\n")}\n~~~` : undefined;
|
|
}
|
|
|
|
function formatToolSummaryTraceBlock(
|
|
toolSummary: TraceToolSummaryView | undefined,
|
|
): string | undefined {
|
|
if (!toolSummary || toolSummary.calls <= 0) {
|
|
return undefined;
|
|
}
|
|
return formatKeyValueTraceBlock("Tool Summary", [
|
|
["calls", toolSummary.calls],
|
|
["tools", toolSummary.tools.length > 0 ? toolSummary.tools.join(", ") : undefined],
|
|
["failures", toolSummary.failures],
|
|
["totalToolTimeMs", toolSummary.totalToolTimeMs],
|
|
]);
|
|
}
|
|
|
|
function formatCompletionTraceBlock(
|
|
completion: TraceCompletionView | undefined,
|
|
): string | undefined {
|
|
if (!completion) {
|
|
return undefined;
|
|
}
|
|
return formatKeyValueTraceBlock("Completion", [
|
|
["finishReason", completion.finishReason],
|
|
["stopReason", completion.stopReason],
|
|
["refusal", completion.refusal],
|
|
]);
|
|
}
|
|
|
|
function formatContextManagementTraceBlock(
|
|
contextManagement: TraceContextManagementView | undefined,
|
|
): string | undefined {
|
|
if (!contextManagement) {
|
|
return undefined;
|
|
}
|
|
return formatKeyValueTraceBlock("Context Management", [
|
|
["sessionCompactions", contextManagement.sessionCompactions],
|
|
["lastTurnCompactions", contextManagement.lastTurnCompactions],
|
|
["preflightCompactionApplied", contextManagement.preflightCompactionApplied],
|
|
["postCompactionContextInjected", contextManagement.postCompactionContextInjected],
|
|
]);
|
|
}
|
|
|
|
async function accumulateSessionUsageFromTranscript(params: {
|
|
sessionId?: string;
|
|
storePath?: string;
|
|
sessionFile?: string;
|
|
}): Promise<
|
|
| {
|
|
input?: number;
|
|
output?: number;
|
|
cacheRead?: number;
|
|
cacheWrite?: number;
|
|
total?: number;
|
|
}
|
|
| undefined
|
|
> {
|
|
const sessionId = normalizeOptionalString(params.sessionId);
|
|
if (!sessionId) {
|
|
return undefined;
|
|
}
|
|
try {
|
|
const candidates = resolveSessionTranscriptCandidates(
|
|
sessionId,
|
|
params.storePath,
|
|
params.sessionFile,
|
|
);
|
|
let transcriptText: string | undefined;
|
|
for (const candidate of candidates) {
|
|
try {
|
|
transcriptText = await fs.readFile(candidate, "utf-8");
|
|
break;
|
|
} catch {
|
|
continue;
|
|
}
|
|
}
|
|
if (!transcriptText) {
|
|
return undefined;
|
|
}
|
|
|
|
let input = 0;
|
|
let output = 0;
|
|
let cacheRead = 0;
|
|
let cacheWrite = 0;
|
|
let sawUsage = false;
|
|
for (const line of transcriptText.split(/\r?\n/)) {
|
|
if (!line.trim()) {
|
|
continue;
|
|
}
|
|
let parsed: { message?: { usage?: unknown } } | undefined;
|
|
try {
|
|
parsed = JSON.parse(line) as { message?: { usage?: unknown } };
|
|
} catch {
|
|
continue;
|
|
}
|
|
const message = parsed?.message;
|
|
if (!message) {
|
|
continue;
|
|
}
|
|
const usage = normalizeUsage(message?.usage as Parameters<typeof normalizeUsage>[0]);
|
|
if (!hasNonzeroUsage(usage)) {
|
|
continue;
|
|
}
|
|
sawUsage = true;
|
|
input += usage.input ?? 0;
|
|
output += usage.output ?? 0;
|
|
cacheRead += usage.cacheRead ?? 0;
|
|
cacheWrite += usage.cacheWrite ?? 0;
|
|
}
|
|
if (!sawUsage) {
|
|
return undefined;
|
|
}
|
|
const total = input + output + cacheRead + cacheWrite;
|
|
return {
|
|
input: input || undefined,
|
|
output: output || undefined,
|
|
cacheRead: cacheRead || undefined,
|
|
cacheWrite: cacheWrite || undefined,
|
|
total: total || undefined,
|
|
};
|
|
} catch {
|
|
return undefined;
|
|
}
|
|
}
|
|
|
|
function formatRequestContextTraceBlock(params: {
|
|
provider?: string;
|
|
model?: string;
|
|
contextLimit?: number;
|
|
promptTokens?: number;
|
|
}): string | undefined {
|
|
const limit = params.contextLimit;
|
|
const used = params.promptTokens;
|
|
if (
|
|
(typeof limit !== "number" || !Number.isFinite(limit) || limit <= 0) &&
|
|
(typeof used !== "number" || !Number.isFinite(used) || used <= 0) &&
|
|
!params.provider &&
|
|
!params.model
|
|
) {
|
|
return undefined;
|
|
}
|
|
const headroom =
|
|
typeof limit === "number" &&
|
|
Number.isFinite(limit) &&
|
|
typeof used === "number" &&
|
|
Number.isFinite(used)
|
|
? Math.max(0, limit - used)
|
|
: undefined;
|
|
const percent =
|
|
typeof limit === "number" &&
|
|
Number.isFinite(limit) &&
|
|
limit > 0 &&
|
|
typeof used === "number" &&
|
|
Number.isFinite(used)
|
|
? Math.round((used / limit) * 100)
|
|
: undefined;
|
|
return `🔎 Context Window (Last Model Request):\n~~~text\n${[
|
|
`provider=${params.provider ?? "n/a"}`,
|
|
`model=${params.model ?? "n/a"}`,
|
|
`used=${typeof used === "number" && Number.isFinite(used) ? `${used.toLocaleString()} tok (${formatTokenCount(used)})` : "n/a"}`,
|
|
`limit=${typeof limit === "number" && Number.isFinite(limit) ? `${limit.toLocaleString()} tok (${formatTokenCount(limit)})` : "n/a"}`,
|
|
`headroom=${typeof headroom === "number" ? `${headroom.toLocaleString()} tok (${formatTokenCount(headroom)})` : "n/a"}`,
|
|
`usage=${typeof percent === "number" ? `${percent}%` : "n/a"}`,
|
|
].join("\n")}\n~~~`;
|
|
}
|
|
|
|
function formatSummaryPromptValue(params: {
|
|
contextLimit?: number;
|
|
promptTokens?: number;
|
|
}): string | undefined {
|
|
const used = params.promptTokens;
|
|
const limit = params.contextLimit;
|
|
if (
|
|
typeof used !== "number" ||
|
|
!Number.isFinite(used) ||
|
|
used <= 0 ||
|
|
typeof limit !== "number" ||
|
|
!Number.isFinite(limit) ||
|
|
limit <= 0
|
|
) {
|
|
return undefined;
|
|
}
|
|
return `${formatTokenCount(used)}/${formatTokenCount(limit)}`;
|
|
}
|
|
|
|
function formatRawTraceSummaryLine(params: {
|
|
executionTrace?: TraceExecutionView;
|
|
completion?: TraceCompletionView;
|
|
contextLimit?: number;
|
|
promptTokens?: number;
|
|
usage?: {
|
|
input?: number;
|
|
output?: number;
|
|
cacheRead?: number;
|
|
cacheWrite?: number;
|
|
total?: number;
|
|
};
|
|
toolSummary?: TraceToolSummaryView;
|
|
contextManagement?: TraceContextManagementView;
|
|
requestShaping?: {
|
|
thinking?: string;
|
|
};
|
|
}): string | undefined {
|
|
const thinking = normalizeOptionalString(params.requestShaping?.thinking);
|
|
const fields = [
|
|
params.executionTrace?.winnerModel
|
|
? `winner=${params.executionTrace.winnerModel}${thinking ? ` 🧠 ${thinking}` : ""}`
|
|
: undefined,
|
|
typeof params.executionTrace?.fallbackUsed === "boolean"
|
|
? `fallback=${params.executionTrace.fallbackUsed ? "yes" : "no"}`
|
|
: undefined,
|
|
typeof params.executionTrace?.attempts?.length === "number"
|
|
? `attempts=${params.executionTrace.attempts.length.toLocaleString()}`
|
|
: undefined,
|
|
params.completion?.stopReason ? `stop=${params.completion.stopReason}` : undefined,
|
|
(() => {
|
|
const prompt = formatSummaryPromptValue({
|
|
contextLimit: params.contextLimit,
|
|
promptTokens: params.promptTokens,
|
|
});
|
|
return prompt ? `prompt=${prompt}` : undefined;
|
|
})(),
|
|
typeof params.usage?.input === "number" && params.usage.input > 0
|
|
? `⬇️ ${formatTokenCount(params.usage.input)}`
|
|
: undefined,
|
|
typeof params.usage?.output === "number" && params.usage.output > 0
|
|
? `⬆️ ${formatTokenCount(params.usage.output)}`
|
|
: undefined,
|
|
typeof params.usage?.cacheRead === "number" && params.usage.cacheRead > 0
|
|
? `♻️ ${formatTokenCount(params.usage.cacheRead)}`
|
|
: undefined,
|
|
typeof params.usage?.cacheWrite === "number" && params.usage.cacheWrite > 0
|
|
? `🆕 ${formatTokenCount(params.usage.cacheWrite)}`
|
|
: undefined,
|
|
typeof params.usage?.total === "number" && params.usage.total > 0
|
|
? `🔢 ${formatTokenCount(params.usage.total)}`
|
|
: undefined,
|
|
typeof params.toolSummary?.calls === "number" && params.toolSummary.calls > 0
|
|
? `tools=${params.toolSummary.calls.toLocaleString()}`
|
|
: undefined,
|
|
typeof params.contextManagement?.lastTurnCompactions === "number" &&
|
|
params.contextManagement.lastTurnCompactions > 0
|
|
? `compactions=${params.contextManagement.lastTurnCompactions.toLocaleString()}`
|
|
: undefined,
|
|
].filter((value): value is string => Boolean(value));
|
|
return fields.length > 0 ? `Summary: ${fields.join(" ")}` : undefined;
|
|
}
|
|
|
|
function buildInlineRawTracePayload(params: {
|
|
entry: SessionEntry | undefined;
|
|
rawUserText?: string;
|
|
rawAssistantText?: string;
|
|
sessionUsage?: {
|
|
input?: number;
|
|
output?: number;
|
|
cacheRead?: number;
|
|
cacheWrite?: number;
|
|
total?: number;
|
|
};
|
|
usage?: {
|
|
input?: number;
|
|
output?: number;
|
|
cacheRead?: number;
|
|
cacheWrite?: number;
|
|
total?: number;
|
|
};
|
|
lastCallUsage?: {
|
|
input?: number;
|
|
output?: number;
|
|
cacheRead?: number;
|
|
cacheWrite?: number;
|
|
total?: number;
|
|
};
|
|
provider?: string;
|
|
model?: string;
|
|
contextLimit?: number;
|
|
promptTokens?: number;
|
|
executionTrace?: TraceExecutionView;
|
|
requestShaping?: {
|
|
authMode?: string;
|
|
thinking?: string;
|
|
reasoning?: string;
|
|
verbose?: string;
|
|
trace?: string;
|
|
fallbackEligible?: boolean;
|
|
blockStreaming?: string;
|
|
};
|
|
promptSegments?: TracePromptSegmentView[];
|
|
toolSummary?: TraceToolSummaryView;
|
|
completion?: TraceCompletionView;
|
|
contextManagement?: TraceContextManagementView;
|
|
}): ReplyPayload | undefined {
|
|
if (params.entry?.traceLevel !== "raw") {
|
|
return undefined;
|
|
}
|
|
const resolvedPromptTokens = deriveContextPromptTokens({
|
|
lastCallUsage: params.lastCallUsage,
|
|
promptTokens: params.promptTokens,
|
|
usage: params.usage,
|
|
});
|
|
const requestContextBlock = formatRequestContextTraceBlock({
|
|
provider: params.provider,
|
|
model: params.model,
|
|
contextLimit: params.contextLimit,
|
|
promptTokens: resolvedPromptTokens,
|
|
});
|
|
const usageBlocks = [
|
|
formatUsageTraceBlock("Usage (Session Total)", params.sessionUsage),
|
|
formatUsageTraceBlock("Usage (Last Turn Total)", params.usage),
|
|
requestContextBlock,
|
|
formatExecutionResultTraceBlock(params.executionTrace),
|
|
formatFallbackChainTraceBlock(params.executionTrace),
|
|
formatKeyValueTraceBlock("Request Shaping", [
|
|
["provider", params.provider],
|
|
["model", params.model],
|
|
["auth", params.requestShaping?.authMode],
|
|
["thinking", params.requestShaping?.thinking],
|
|
["reasoning", params.requestShaping?.reasoning],
|
|
["verbose", params.requestShaping?.verbose],
|
|
["trace", params.requestShaping?.trace],
|
|
["fallbackEligible", params.requestShaping?.fallbackEligible],
|
|
["blockStreaming", params.requestShaping?.blockStreaming],
|
|
]),
|
|
formatPromptSegmentsTraceBlock(params.promptSegments, params.rawUserText),
|
|
formatToolSummaryTraceBlock(params.toolSummary),
|
|
formatCompletionTraceBlock(params.completion),
|
|
formatContextManagementTraceBlock(params.contextManagement),
|
|
].filter((value): value is string => Boolean(value));
|
|
return {
|
|
text: [
|
|
...usageBlocks,
|
|
formatRawTraceBlock("Model Input (User Role)", params.rawUserText),
|
|
formatRawTraceBlock("Model Output (Assistant Role)", params.rawAssistantText),
|
|
formatRawTraceSummaryLine({
|
|
executionTrace: params.executionTrace,
|
|
completion: params.completion,
|
|
contextLimit: params.contextLimit,
|
|
promptTokens: resolvedPromptTokens,
|
|
usage: params.usage,
|
|
toolSummary: params.toolSummary,
|
|
contextManagement: params.contextManagement,
|
|
requestShaping: params.requestShaping,
|
|
}),
|
|
].join("\n\n\n"),
|
|
};
|
|
}
|
|
|
|
function joinCommitmentAssistantText(payloads: ReplyPayload[]): string {
|
|
return payloads
|
|
.filter((payload) => !payload.isError && !payload.isReasoning && !payload.isCompactionNotice)
|
|
.map((payload) => payload.text?.trim())
|
|
.filter((text): text is string => Boolean(text))
|
|
.join("\n")
|
|
.trim();
|
|
}
|
|
|
|
function buildPendingFinalDeliveryText(payloads: ReplyPayload[]): string {
|
|
return payloads
|
|
.filter((payload) => payload.isReasoning !== true)
|
|
.map((payload) => payload.text)
|
|
.filter((text): text is string => Boolean(text))
|
|
.join("\n\n");
|
|
}
|
|
|
|
function enqueueCommitmentExtractionForTurn(params: {
|
|
cfg: OpenClawConfig;
|
|
commandBody: string;
|
|
isHeartbeat: boolean;
|
|
followupRun: FollowupRun;
|
|
sessionCtx: TemplateContext;
|
|
sessionKey?: string;
|
|
replyToChannel?: string;
|
|
payloads: ReplyPayload[];
|
|
runId: string;
|
|
}): void {
|
|
if (params.isHeartbeat) {
|
|
return;
|
|
}
|
|
const userText =
|
|
params.commandBody.trim() ||
|
|
params.sessionCtx.BodyStripped?.trim() ||
|
|
params.sessionCtx.BodyForCommands?.trim() ||
|
|
params.sessionCtx.CommandBody?.trim() ||
|
|
params.sessionCtx.RawBody?.trim() ||
|
|
params.sessionCtx.Body?.trim() ||
|
|
"";
|
|
const assistantText = joinCommitmentAssistantText(params.payloads);
|
|
const sessionKey = params.sessionKey ?? params.followupRun.run.sessionKey;
|
|
const channel =
|
|
params.replyToChannel ??
|
|
params.followupRun.run.messageProvider ??
|
|
params.sessionCtx.Surface ??
|
|
params.sessionCtx.Provider;
|
|
if (!userText || !assistantText || !sessionKey || !channel) {
|
|
return;
|
|
}
|
|
const to = resolveOriginMessageTo({
|
|
originatingTo: params.sessionCtx.OriginatingTo,
|
|
to: params.sessionCtx.To,
|
|
});
|
|
enqueueCommitmentExtraction({
|
|
cfg: params.cfg,
|
|
agentId: params.followupRun.run.agentId,
|
|
sessionKey,
|
|
channel,
|
|
...(params.sessionCtx.AccountId ? { accountId: params.sessionCtx.AccountId } : {}),
|
|
...(to ? { to } : {}),
|
|
...(params.sessionCtx.MessageThreadId !== undefined
|
|
? { threadId: String(params.sessionCtx.MessageThreadId) }
|
|
: {}),
|
|
...(params.followupRun.run.senderId ? { senderId: params.followupRun.run.senderId } : {}),
|
|
userText,
|
|
assistantText,
|
|
...(params.sessionCtx.MessageSidFull || params.sessionCtx.MessageSid
|
|
? { sourceMessageId: params.sessionCtx.MessageSidFull ?? params.sessionCtx.MessageSid }
|
|
: {}),
|
|
sourceRunId: params.runId,
|
|
});
|
|
}
|
|
|
|
function refreshSessionEntryFromStore(params: {
|
|
storePath?: string;
|
|
sessionKey?: string;
|
|
fallbackEntry?: SessionEntry;
|
|
activeSessionStore?: Record<string, SessionEntry>;
|
|
}): SessionEntry | undefined {
|
|
const { storePath, sessionKey, fallbackEntry, activeSessionStore } = params;
|
|
if (!storePath || !sessionKey) {
|
|
return fallbackEntry;
|
|
}
|
|
try {
|
|
const latestStore = loadSessionStore(storePath, { skipCache: true });
|
|
const latestEntry = latestStore?.[sessionKey];
|
|
if (!latestEntry) {
|
|
return fallbackEntry;
|
|
}
|
|
if (activeSessionStore) {
|
|
activeSessionStore[sessionKey] = latestEntry;
|
|
}
|
|
return latestEntry;
|
|
} catch {
|
|
return fallbackEntry;
|
|
}
|
|
}
|
|
|
|
export async function runReplyAgent(params: {
|
|
commandBody: string;
|
|
transcriptCommandBody?: string;
|
|
followupRun: FollowupRun;
|
|
queueKey: string;
|
|
resolvedQueue: QueueSettings;
|
|
shouldSteer: boolean;
|
|
shouldFollowup: boolean;
|
|
isActive: boolean;
|
|
isRunActive?: () => boolean;
|
|
isStreaming: boolean;
|
|
opts?: GetReplyOptions;
|
|
typing: TypingController;
|
|
sessionEntry?: SessionEntry;
|
|
sessionStore?: Record<string, SessionEntry>;
|
|
sessionKey?: string;
|
|
runtimePolicySessionKey?: string;
|
|
storePath?: string;
|
|
defaultModel: string;
|
|
agentCfgContextTokens?: number;
|
|
resolvedVerboseLevel: VerboseLevel;
|
|
toolProgressDetail?: "explain" | "raw";
|
|
isNewSession: boolean;
|
|
blockStreamingEnabled: boolean;
|
|
blockReplyChunking?: {
|
|
minChars: number;
|
|
maxChars: number;
|
|
breakPreference: "paragraph" | "newline" | "sentence";
|
|
flushOnParagraph?: boolean;
|
|
};
|
|
resolvedBlockStreamingBreak: "text_end" | "message_end";
|
|
sessionCtx: TemplateContext;
|
|
shouldInjectGroupIntro: boolean;
|
|
typingMode: TypingMode;
|
|
resetTriggered?: boolean;
|
|
replyThreadingOverride?: TemplateContext["ReplyThreading"];
|
|
replyOperation?: ReplyOperation;
|
|
}): Promise<ReplyPayload | ReplyPayload[] | undefined> {
|
|
const {
|
|
commandBody,
|
|
transcriptCommandBody,
|
|
followupRun,
|
|
queueKey,
|
|
resolvedQueue,
|
|
shouldSteer,
|
|
shouldFollowup,
|
|
isActive,
|
|
isRunActive,
|
|
isStreaming,
|
|
opts,
|
|
typing,
|
|
sessionEntry,
|
|
sessionStore,
|
|
sessionKey,
|
|
runtimePolicySessionKey,
|
|
storePath,
|
|
defaultModel,
|
|
agentCfgContextTokens,
|
|
resolvedVerboseLevel,
|
|
toolProgressDetail,
|
|
isNewSession,
|
|
blockStreamingEnabled,
|
|
blockReplyChunking,
|
|
resolvedBlockStreamingBreak,
|
|
sessionCtx,
|
|
shouldInjectGroupIntro,
|
|
typingMode,
|
|
resetTriggered,
|
|
replyThreadingOverride,
|
|
replyOperation: providedReplyOperation,
|
|
} = params;
|
|
|
|
let activeSessionEntry = sessionEntry;
|
|
const activeSessionStore = sessionStore;
|
|
let activeIsNewSession = isNewSession;
|
|
const effectiveResetTriggered = resetTriggered === true;
|
|
const activeRunQueueMode = effectiveResetTriggered ? "interrupt" : resolvedQueue.mode;
|
|
const effectiveShouldSteer = !effectiveResetTriggered && shouldSteer;
|
|
const effectiveShouldFollowup = !effectiveResetTriggered && shouldFollowup;
|
|
|
|
const isHeartbeat = opts?.isHeartbeat === true;
|
|
const traceAttributes = {
|
|
provider: followupRun.run.provider,
|
|
hasSessionKey: Boolean(sessionKey ?? followupRun.run.sessionKey),
|
|
isHeartbeat,
|
|
queueMode: resolvedQueue.mode,
|
|
isActive,
|
|
blockStreamingEnabled,
|
|
};
|
|
const traceAgentPhase = <T>(name: string, run: () => Promise<T> | T): Promise<T> =>
|
|
measureDiagnosticsTimelineSpan(name, run, {
|
|
phase: "agent-turn",
|
|
config: followupRun.run.config,
|
|
attributes: traceAttributes,
|
|
});
|
|
const typingSignals = createTypingSignaler({
|
|
typing,
|
|
mode: typingMode,
|
|
isHeartbeat,
|
|
});
|
|
|
|
const shouldEmitToolResult = createShouldEmitToolResult({
|
|
sessionKey,
|
|
storePath,
|
|
resolvedVerboseLevel,
|
|
});
|
|
const shouldEmitToolOutput = createShouldEmitToolOutput({
|
|
sessionKey,
|
|
storePath,
|
|
resolvedVerboseLevel,
|
|
});
|
|
|
|
const pendingToolTasks = new Set<Promise<void>>();
|
|
const blockReplyTimeoutMs = opts?.blockReplyTimeoutMs ?? BLOCK_REPLY_SEND_TIMEOUT_MS;
|
|
const touchActiveSessionEntry = async () => {
|
|
if (!activeSessionEntry || !activeSessionStore || !sessionKey) {
|
|
return;
|
|
}
|
|
const updatedAt = Date.now();
|
|
activeSessionEntry.updatedAt = updatedAt;
|
|
activeSessionStore[sessionKey] = activeSessionEntry;
|
|
if (storePath) {
|
|
await updateSessionStoreEntry({
|
|
storePath,
|
|
sessionKey,
|
|
update: async () => ({ updatedAt }),
|
|
});
|
|
}
|
|
};
|
|
|
|
if (effectiveShouldSteer && isStreaming) {
|
|
const steerSessionId =
|
|
(sessionKey ? replyRunRegistry.resolveSessionId(sessionKey) : undefined) ??
|
|
followupRun.run.sessionId;
|
|
const steered = queueEmbeddedPiMessage(steerSessionId, followupRun.prompt, {
|
|
steeringMode: resolvePiSteeringModeForQueueMode(resolvedQueue.mode),
|
|
...(resolvedQueue.debounceMs !== undefined ? { debounceMs: resolvedQueue.debounceMs } : {}),
|
|
});
|
|
if (steered && !effectiveShouldFollowup) {
|
|
await touchActiveSessionEntry();
|
|
typing.cleanup();
|
|
return undefined;
|
|
}
|
|
}
|
|
|
|
const activeRunQueueAction = resolveActiveRunQueueAction({
|
|
isActive,
|
|
isHeartbeat,
|
|
shouldFollowup: effectiveShouldFollowup,
|
|
queueMode: activeRunQueueMode,
|
|
resetTriggered: effectiveResetTriggered,
|
|
});
|
|
|
|
const queuedRunFollowupTurn = createFollowupRunner({
|
|
opts,
|
|
typing,
|
|
typingMode,
|
|
sessionEntry: activeSessionEntry,
|
|
sessionStore: activeSessionStore,
|
|
sessionKey,
|
|
storePath,
|
|
defaultModel,
|
|
agentCfgContextTokens,
|
|
});
|
|
|
|
if (activeRunQueueAction === "drop") {
|
|
typing.cleanup();
|
|
return undefined;
|
|
}
|
|
|
|
if (activeRunQueueAction === "enqueue-followup") {
|
|
enqueueFollowupRun(
|
|
queueKey,
|
|
followupRun,
|
|
resolvedQueue,
|
|
"message-id",
|
|
queuedRunFollowupTurn,
|
|
false,
|
|
);
|
|
// Re-check liveness after enqueue so a stale active snapshot cannot leave
|
|
// the followup queue idle if the original run already finished.
|
|
const queuedBehindActiveRun = isRunActive?.() === true;
|
|
if (!queuedBehindActiveRun) {
|
|
scheduleFollowupDrain(queueKey, queuedRunFollowupTurn);
|
|
}
|
|
await touchActiveSessionEntry();
|
|
if (queuedBehindActiveRun) {
|
|
await typingSignals.signalToolStart();
|
|
} else {
|
|
typing.cleanup();
|
|
}
|
|
return undefined;
|
|
}
|
|
|
|
followupRun.run.config = await resolveQueuedReplyExecutionConfig(followupRun.run.config, {
|
|
originatingChannel: sessionCtx.OriginatingChannel,
|
|
messageProvider: followupRun.run.messageProvider,
|
|
originatingAccountId: followupRun.originatingAccountId,
|
|
agentAccountId: followupRun.run.agentAccountId,
|
|
});
|
|
|
|
const replyToChannel = resolveOriginMessageProvider({
|
|
originatingChannel: sessionCtx.OriginatingChannel,
|
|
provider: sessionCtx.Surface ?? sessionCtx.Provider,
|
|
}) as OriginatingChannelType | undefined;
|
|
const replyToMode = resolveReplyToMode(
|
|
followupRun.run.config,
|
|
replyToChannel,
|
|
sessionCtx.AccountId,
|
|
sessionCtx.ChatType,
|
|
);
|
|
const applyReplyToMode = createReplyToModeFilterForChannel(replyToMode, replyToChannel);
|
|
const cfg = followupRun.run.config;
|
|
const replyMediaContext = createReplyMediaContext({
|
|
cfg,
|
|
sessionKey,
|
|
workspaceDir: followupRun.run.workspaceDir,
|
|
messageProvider: followupRun.run.messageProvider,
|
|
accountId: followupRun.originatingAccountId ?? followupRun.run.agentAccountId,
|
|
groupId: followupRun.run.groupId,
|
|
groupChannel: followupRun.run.groupChannel,
|
|
groupSpace: followupRun.run.groupSpace,
|
|
requesterSenderId: followupRun.run.senderId,
|
|
requesterSenderName: followupRun.run.senderName,
|
|
requesterSenderUsername: followupRun.run.senderUsername,
|
|
requesterSenderE164: followupRun.run.senderE164,
|
|
});
|
|
const blockReplyCoalescing =
|
|
blockStreamingEnabled && opts?.onBlockReply
|
|
? resolveEffectiveBlockStreamingConfig({
|
|
cfg,
|
|
provider: sessionCtx.Provider,
|
|
accountId: sessionCtx.AccountId,
|
|
chunking: blockReplyChunking,
|
|
}).coalescing
|
|
: undefined;
|
|
const blockReplyPipeline =
|
|
blockStreamingEnabled && opts?.onBlockReply
|
|
? createBlockReplyPipeline({
|
|
onBlockReply: opts.onBlockReply,
|
|
timeoutMs: blockReplyTimeoutMs,
|
|
coalescing: blockReplyCoalescing,
|
|
buffer: createAudioAsVoiceBuffer({ isAudioPayload }),
|
|
})
|
|
: null;
|
|
|
|
const replySessionKey = sessionKey ?? followupRun.run.sessionKey;
|
|
let replyOperation: ReplyOperation;
|
|
try {
|
|
replyOperation =
|
|
providedReplyOperation ??
|
|
createReplyOperation({
|
|
sessionId: followupRun.run.sessionId,
|
|
sessionKey: replySessionKey ?? "",
|
|
resetTriggered: effectiveResetTriggered,
|
|
upstreamAbortSignal: opts?.abortSignal,
|
|
});
|
|
} catch (error) {
|
|
if (error instanceof ReplyRunAlreadyActiveError) {
|
|
typing.cleanup();
|
|
return markReplyPayloadForSourceSuppressionDelivery({
|
|
text: "⚠️ Previous run is still shutting down. Please try again in a moment.",
|
|
});
|
|
}
|
|
throw error;
|
|
}
|
|
let runFollowupTurn = queuedRunFollowupTurn;
|
|
let shouldDrainQueuedFollowupsAfterClear = false;
|
|
const returnWithQueuedFollowupDrain = <T>(value: T): T => {
|
|
shouldDrainQueuedFollowupsAfterClear = true;
|
|
return value;
|
|
};
|
|
const drainQueuedFollowupsAfterClear = () => {
|
|
scheduleFollowupDrain(queueKey, runFollowupTurn);
|
|
};
|
|
const prePreflightCompactionCount = activeSessionEntry?.compactionCount ?? 0;
|
|
let preflightCompactionApplied = false;
|
|
|
|
try {
|
|
await typingSignals.signalRunStart();
|
|
|
|
activeSessionEntry = await traceAgentPhase("reply.preflight_compaction", () =>
|
|
runPreflightCompactionIfNeeded({
|
|
cfg,
|
|
followupRun,
|
|
promptForEstimate: followupRun.prompt,
|
|
defaultModel,
|
|
agentCfgContextTokens,
|
|
sessionEntry: activeSessionEntry,
|
|
sessionStore: activeSessionStore,
|
|
sessionKey,
|
|
runtimePolicySessionKey,
|
|
storePath,
|
|
isHeartbeat,
|
|
replyOperation,
|
|
}),
|
|
);
|
|
preflightCompactionApplied =
|
|
(activeSessionEntry?.compactionCount ?? 0) > prePreflightCompactionCount;
|
|
|
|
activeSessionEntry = await traceAgentPhase("reply.memory_flush", () =>
|
|
runMemoryFlushIfNeeded({
|
|
cfg,
|
|
followupRun,
|
|
promptForEstimate: followupRun.prompt,
|
|
sessionCtx,
|
|
opts,
|
|
defaultModel,
|
|
agentCfgContextTokens,
|
|
resolvedVerboseLevel,
|
|
sessionEntry: activeSessionEntry,
|
|
sessionStore: activeSessionStore,
|
|
sessionKey,
|
|
runtimePolicySessionKey,
|
|
storePath,
|
|
isHeartbeat,
|
|
replyOperation,
|
|
}),
|
|
);
|
|
|
|
runFollowupTurn = createFollowupRunner({
|
|
opts,
|
|
typing,
|
|
typingMode,
|
|
sessionEntry: activeSessionEntry,
|
|
sessionStore: activeSessionStore,
|
|
sessionKey,
|
|
storePath,
|
|
defaultModel,
|
|
agentCfgContextTokens,
|
|
});
|
|
|
|
let responseUsageLine: string | undefined;
|
|
type SessionResetOptions = {
|
|
failureLabel: string;
|
|
buildLogMessage: (nextSessionId: string) => string;
|
|
cleanupTranscripts?: boolean;
|
|
};
|
|
const resetSession = async ({
|
|
failureLabel,
|
|
buildLogMessage,
|
|
cleanupTranscripts,
|
|
}: SessionResetOptions): Promise<boolean> =>
|
|
await resetReplyRunSession({
|
|
options: {
|
|
failureLabel,
|
|
buildLogMessage,
|
|
cleanupTranscripts,
|
|
},
|
|
sessionKey,
|
|
queueKey,
|
|
activeSessionEntry,
|
|
activeSessionStore,
|
|
storePath,
|
|
messageThreadId:
|
|
typeof sessionCtx.MessageThreadId === "string" ? sessionCtx.MessageThreadId : undefined,
|
|
followupRun,
|
|
onActiveSessionEntry: (nextEntry) => {
|
|
activeSessionEntry = nextEntry;
|
|
},
|
|
onNewSession: () => {
|
|
activeIsNewSession = true;
|
|
},
|
|
});
|
|
const resetSessionAfterCompactionFailure = async (reason: string): Promise<boolean> =>
|
|
resetSession({
|
|
failureLabel: "compaction failure",
|
|
buildLogMessage: (nextSessionId) =>
|
|
`Auto-compaction failed (${reason}). Restarting session ${sessionKey} -> ${nextSessionId} and retrying.`,
|
|
});
|
|
const resetSessionAfterRoleOrderingConflict = async (reason: string): Promise<boolean> =>
|
|
resetSession({
|
|
failureLabel: "role ordering conflict",
|
|
buildLogMessage: (nextSessionId) =>
|
|
`Role ordering conflict (${reason}). Restarting session ${sessionKey} -> ${nextSessionId}.`,
|
|
cleanupTranscripts: true,
|
|
});
|
|
|
|
replyOperation.setPhase("running");
|
|
const runStartedAt = Date.now();
|
|
const runOutcome = await traceAgentPhase("reply.run_agent_turn", () =>
|
|
runAgentTurnWithFallback({
|
|
commandBody,
|
|
transcriptCommandBody,
|
|
followupRun,
|
|
sessionCtx,
|
|
replyThreading: replyThreadingOverride ?? sessionCtx.ReplyThreading,
|
|
replyOperation,
|
|
opts,
|
|
typingSignals,
|
|
blockReplyPipeline,
|
|
blockStreamingEnabled,
|
|
blockReplyChunking,
|
|
resolvedBlockStreamingBreak,
|
|
applyReplyToMode,
|
|
shouldEmitToolResult,
|
|
shouldEmitToolOutput,
|
|
pendingToolTasks,
|
|
resetSessionAfterCompactionFailure,
|
|
resetSessionAfterRoleOrderingConflict,
|
|
isHeartbeat,
|
|
sessionKey,
|
|
runtimePolicySessionKey,
|
|
getActiveSessionEntry: () => activeSessionEntry,
|
|
activeSessionStore,
|
|
storePath,
|
|
resolvedVerboseLevel,
|
|
toolProgressDetail,
|
|
replyMediaContext,
|
|
}),
|
|
);
|
|
|
|
if (runOutcome.kind === "final") {
|
|
if (!replyOperation.result) {
|
|
replyOperation.fail("run_failed", new Error("reply operation exited with final payload"));
|
|
}
|
|
return returnWithQueuedFollowupDrain(runOutcome.payload);
|
|
}
|
|
|
|
const {
|
|
runId,
|
|
runResult,
|
|
fallbackProvider,
|
|
fallbackModel,
|
|
fallbackAttempts,
|
|
directlySentBlockKeys,
|
|
} = runOutcome;
|
|
let { didLogHeartbeatStrip, autoCompactionCount } = runOutcome;
|
|
|
|
if (
|
|
shouldInjectGroupIntro &&
|
|
activeSessionEntry &&
|
|
activeSessionStore &&
|
|
sessionKey &&
|
|
activeSessionEntry.groupActivationNeedsSystemIntro
|
|
) {
|
|
const updatedAt = Date.now();
|
|
activeSessionEntry.groupActivationNeedsSystemIntro = false;
|
|
activeSessionEntry.updatedAt = updatedAt;
|
|
activeSessionStore[sessionKey] = activeSessionEntry;
|
|
if (storePath) {
|
|
await updateSessionStoreEntry({
|
|
storePath,
|
|
sessionKey,
|
|
update: async () => ({
|
|
groupActivationNeedsSystemIntro: false,
|
|
updatedAt,
|
|
}),
|
|
});
|
|
}
|
|
}
|
|
|
|
const payloadArray = runResult.payloads ?? [];
|
|
|
|
if (blockReplyPipeline) {
|
|
await blockReplyPipeline.flush({ force: true });
|
|
blockReplyPipeline.stop();
|
|
}
|
|
if (pendingToolTasks.size > 0) {
|
|
await drainPendingToolTasks({
|
|
tasks: pendingToolTasks,
|
|
onTimeout: logVerbose,
|
|
});
|
|
}
|
|
|
|
const usage = runResult.meta?.agentMeta?.usage;
|
|
const promptTokens = runResult.meta?.agentMeta?.promptTokens;
|
|
const modelUsed = runResult.meta?.agentMeta?.model ?? fallbackModel ?? defaultModel;
|
|
const providerUsed =
|
|
runResult.meta?.agentMeta?.provider ?? fallbackProvider ?? followupRun.run.provider;
|
|
const verboseEnabled = resolvedVerboseLevel !== "off";
|
|
const selectedProvider = followupRun.run.provider;
|
|
const selectedModel = followupRun.run.model;
|
|
const fallbackStateEntry =
|
|
activeSessionEntry ?? (sessionKey ? activeSessionStore?.[sessionKey] : undefined);
|
|
const fallbackTransition = resolveFallbackTransition({
|
|
selectedProvider,
|
|
selectedModel,
|
|
activeProvider: providerUsed,
|
|
activeModel: modelUsed,
|
|
attempts: fallbackAttempts,
|
|
state: fallbackStateEntry,
|
|
});
|
|
if (fallbackTransition.stateChanged) {
|
|
if (fallbackStateEntry) {
|
|
fallbackStateEntry.fallbackNoticeSelectedModel = fallbackTransition.nextState.selectedModel;
|
|
fallbackStateEntry.fallbackNoticeActiveModel = fallbackTransition.nextState.activeModel;
|
|
fallbackStateEntry.fallbackNoticeReason = fallbackTransition.nextState.reason;
|
|
fallbackStateEntry.updatedAt = Date.now();
|
|
activeSessionEntry = fallbackStateEntry;
|
|
}
|
|
if (sessionKey && fallbackStateEntry && activeSessionStore) {
|
|
activeSessionStore[sessionKey] = fallbackStateEntry;
|
|
}
|
|
if (sessionKey && storePath) {
|
|
await updateSessionStoreEntry({
|
|
storePath,
|
|
sessionKey,
|
|
update: async () => ({
|
|
fallbackNoticeSelectedModel: fallbackTransition.nextState.selectedModel,
|
|
fallbackNoticeActiveModel: fallbackTransition.nextState.activeModel,
|
|
fallbackNoticeReason: fallbackTransition.nextState.reason,
|
|
}),
|
|
});
|
|
}
|
|
}
|
|
const cliSessionId = isCliProvider(providerUsed, cfg)
|
|
? normalizeOptionalString(runResult.meta?.agentMeta?.sessionId)
|
|
: undefined;
|
|
const cliSessionBinding = isCliProvider(providerUsed, cfg)
|
|
? runResult.meta?.agentMeta?.cliSessionBinding
|
|
: undefined;
|
|
const runtimeContextTokens =
|
|
typeof runResult.meta?.agentMeta?.contextTokens === "number" &&
|
|
Number.isFinite(runResult.meta.agentMeta.contextTokens) &&
|
|
runResult.meta.agentMeta.contextTokens > 0
|
|
? Math.floor(runResult.meta.agentMeta.contextTokens)
|
|
: undefined;
|
|
const contextTokensUsed =
|
|
runtimeContextTokens ??
|
|
resolveContextTokensForModel({
|
|
cfg,
|
|
provider: providerUsed,
|
|
model: modelUsed,
|
|
contextTokensOverride: agentCfgContextTokens,
|
|
fallbackContextTokens: activeSessionEntry?.contextTokens ?? DEFAULT_CONTEXT_TOKENS,
|
|
allowAsyncLoad: false,
|
|
}) ??
|
|
DEFAULT_CONTEXT_TOKENS;
|
|
|
|
await persistRunSessionUsage({
|
|
storePath,
|
|
sessionKey,
|
|
cfg,
|
|
usage,
|
|
lastCallUsage: runResult.meta?.agentMeta?.lastCallUsage,
|
|
promptTokens,
|
|
modelUsed,
|
|
providerUsed,
|
|
contextTokensUsed,
|
|
systemPromptReport: runResult.meta?.systemPromptReport,
|
|
cliSessionId,
|
|
cliSessionBinding,
|
|
});
|
|
|
|
// Drain any late tool/block deliveries before deciding there's "nothing to send".
|
|
// Otherwise, a late typing trigger (e.g. from a tool callback) can outlive the run and
|
|
// keep the typing indicator stuck.
|
|
if (payloadArray.length === 0) {
|
|
return returnWithQueuedFollowupDrain(undefined);
|
|
}
|
|
|
|
const currentMessageId = sessionCtx.MessageSidFull ?? sessionCtx.MessageSid;
|
|
const payloadResult = await buildReplyPayloads({
|
|
payloads: payloadArray,
|
|
isHeartbeat,
|
|
didLogHeartbeatStrip,
|
|
silentExpected: followupRun.run.silentExpected,
|
|
blockStreamingEnabled,
|
|
blockReplyPipeline,
|
|
directlySentBlockKeys,
|
|
replyToMode,
|
|
replyToChannel,
|
|
currentMessageId,
|
|
replyThreading: replyThreadingOverride ?? sessionCtx.ReplyThreading,
|
|
messageProvider: followupRun.run.messageProvider,
|
|
messagingToolSentTexts: runResult.messagingToolSentTexts,
|
|
messagingToolSentMediaUrls: runResult.messagingToolSentMediaUrls,
|
|
messagingToolSentTargets: runResult.messagingToolSentTargets,
|
|
originatingChannel: sessionCtx.OriginatingChannel,
|
|
originatingTo: resolveOriginMessageTo({
|
|
originatingTo: sessionCtx.OriginatingTo,
|
|
to: sessionCtx.To,
|
|
}),
|
|
accountId: sessionCtx.AccountId,
|
|
normalizeMediaPaths: replyMediaContext.normalizePayload,
|
|
});
|
|
const { replyPayloads } = payloadResult;
|
|
didLogHeartbeatStrip = payloadResult.didLogHeartbeatStrip;
|
|
|
|
if (replyPayloads.length === 0) {
|
|
return returnWithQueuedFollowupDrain(undefined);
|
|
}
|
|
|
|
const successfulCronAdds = runResult.successfulCronAdds ?? 0;
|
|
const hasReminderCommitment = replyPayloads.some(
|
|
(payload) =>
|
|
!payload.isError &&
|
|
typeof payload.text === "string" &&
|
|
hasUnbackedReminderCommitment(payload.text),
|
|
);
|
|
// Suppress the guard note when an existing cron job (created in a prior
|
|
// turn) already covers the commitment — avoids false positives (#32228).
|
|
const coveredByExistingCron =
|
|
hasReminderCommitment && successfulCronAdds === 0
|
|
? await hasSessionRelatedCronJobs({
|
|
cronStorePath: cfg.cron?.store,
|
|
sessionKey,
|
|
})
|
|
: false;
|
|
const guardedReplyPayloads =
|
|
hasReminderCommitment && successfulCronAdds === 0 && !coveredByExistingCron
|
|
? appendUnscheduledReminderNote(replyPayloads)
|
|
: replyPayloads;
|
|
|
|
enqueueCommitmentExtractionForTurn({
|
|
cfg,
|
|
commandBody,
|
|
isHeartbeat,
|
|
followupRun,
|
|
sessionCtx,
|
|
sessionKey,
|
|
replyToChannel,
|
|
payloads: replyPayloads,
|
|
runId,
|
|
});
|
|
|
|
await signalTypingIfNeeded(guardedReplyPayloads, typingSignals);
|
|
|
|
if (isDiagnosticsEnabled(cfg) && hasNonzeroUsage(usage)) {
|
|
const input = usage.input ?? 0;
|
|
const output = usage.output ?? 0;
|
|
const cacheRead = usage.cacheRead ?? 0;
|
|
const cacheWrite = usage.cacheWrite ?? 0;
|
|
const usagePromptTokens = input + cacheRead + cacheWrite;
|
|
const totalTokens = usage.total ?? usagePromptTokens + output;
|
|
const contextUsedTokens = deriveContextPromptTokens({
|
|
lastCallUsage: runResult.meta?.agentMeta?.lastCallUsage,
|
|
promptTokens,
|
|
usage,
|
|
});
|
|
const costConfig = resolveModelCostConfig({
|
|
provider: providerUsed,
|
|
model: modelUsed,
|
|
config: cfg,
|
|
});
|
|
const costUsd = estimateUsageCost({ usage, cost: costConfig });
|
|
emitTrustedDiagnosticEvent({
|
|
type: "model.usage",
|
|
...(runResult.diagnosticTrace
|
|
? {
|
|
trace: freezeDiagnosticTraceContext(
|
|
createChildDiagnosticTraceContext(runResult.diagnosticTrace),
|
|
),
|
|
}
|
|
: {}),
|
|
sessionKey,
|
|
sessionId: followupRun.run.sessionId,
|
|
channel: replyToChannel,
|
|
agentId: followupRun.run.agentId,
|
|
provider: providerUsed,
|
|
model: modelUsed,
|
|
usage: {
|
|
input,
|
|
output,
|
|
cacheRead,
|
|
cacheWrite,
|
|
promptTokens: usagePromptTokens,
|
|
total: totalTokens,
|
|
},
|
|
lastCallUsage: runResult.meta?.agentMeta?.lastCallUsage,
|
|
context: {
|
|
limit: contextTokensUsed,
|
|
...(contextUsedTokens !== undefined ? { used: contextUsedTokens } : {}),
|
|
},
|
|
costUsd,
|
|
durationMs: Date.now() - runStartedAt,
|
|
});
|
|
}
|
|
|
|
const responseUsageRaw =
|
|
activeSessionEntry?.responseUsage ??
|
|
(sessionKey ? activeSessionStore?.[sessionKey]?.responseUsage : undefined);
|
|
const responseUsageMode = resolveResponseUsageMode(responseUsageRaw);
|
|
if (responseUsageMode !== "off" && hasNonzeroUsage(usage)) {
|
|
const authMode = resolveModelAuthMode(providerUsed, cfg, undefined, {
|
|
workspaceDir: followupRun.run.workspaceDir,
|
|
});
|
|
const showCost = authMode === "api-key";
|
|
const costConfig = showCost
|
|
? resolveModelCostConfig({
|
|
provider: providerUsed,
|
|
model: modelUsed,
|
|
config: cfg,
|
|
})
|
|
: undefined;
|
|
let formatted = formatResponseUsageLine({
|
|
usage,
|
|
showCost,
|
|
costConfig,
|
|
});
|
|
if (formatted && responseUsageMode === "full" && sessionKey) {
|
|
formatted = `${formatted} · session \`${sessionKey}\``;
|
|
}
|
|
if (formatted) {
|
|
responseUsageLine = formatted;
|
|
}
|
|
}
|
|
|
|
if (verboseEnabled) {
|
|
activeSessionEntry = refreshSessionEntryFromStore({
|
|
storePath,
|
|
sessionKey,
|
|
fallbackEntry: activeSessionEntry,
|
|
activeSessionStore,
|
|
});
|
|
}
|
|
|
|
// If verbose is enabled, prepend operational run notices.
|
|
let finalPayloads = guardedReplyPayloads;
|
|
const verboseNotices: ReplyPayload[] = [];
|
|
|
|
if (verboseEnabled && activeIsNewSession) {
|
|
verboseNotices.push({ text: `🧭 New session: ${followupRun.run.sessionId}` });
|
|
}
|
|
|
|
if (fallbackTransition.fallbackTransitioned) {
|
|
emitAgentEvent({
|
|
runId,
|
|
sessionKey,
|
|
stream: "lifecycle",
|
|
data: {
|
|
phase: "fallback",
|
|
selectedProvider,
|
|
selectedModel,
|
|
activeProvider: providerUsed,
|
|
activeModel: modelUsed,
|
|
reasonSummary: fallbackTransition.reasonSummary,
|
|
attemptSummaries: fallbackTransition.attemptSummaries,
|
|
attempts: fallbackAttempts,
|
|
},
|
|
});
|
|
if (verboseEnabled) {
|
|
const fallbackNotice = buildFallbackNotice({
|
|
selectedProvider,
|
|
selectedModel,
|
|
activeProvider: providerUsed,
|
|
activeModel: modelUsed,
|
|
attempts: fallbackAttempts,
|
|
});
|
|
if (fallbackNotice) {
|
|
verboseNotices.push({ text: fallbackNotice });
|
|
}
|
|
}
|
|
}
|
|
if (fallbackTransition.fallbackCleared) {
|
|
emitAgentEvent({
|
|
runId,
|
|
sessionKey,
|
|
stream: "lifecycle",
|
|
data: {
|
|
phase: "fallback_cleared",
|
|
selectedProvider,
|
|
selectedModel,
|
|
activeProvider: providerUsed,
|
|
activeModel: modelUsed,
|
|
previousActiveModel: fallbackTransition.previousState.activeModel,
|
|
},
|
|
});
|
|
if (verboseEnabled) {
|
|
verboseNotices.push({
|
|
text: buildFallbackClearedNotice({
|
|
selectedProvider,
|
|
selectedModel,
|
|
previousActiveModel: fallbackTransition.previousState.activeModel,
|
|
}),
|
|
});
|
|
}
|
|
}
|
|
|
|
if (autoCompactionCount > 0) {
|
|
const previousSessionId = activeSessionEntry?.sessionId ?? followupRun.run.sessionId;
|
|
const count = await incrementRunCompactionCount({
|
|
cfg,
|
|
sessionEntry: activeSessionEntry,
|
|
sessionStore: activeSessionStore,
|
|
sessionKey,
|
|
storePath,
|
|
amount: autoCompactionCount,
|
|
compactionTokensAfter: runResult.meta?.agentMeta?.compactionTokensAfter,
|
|
lastCallUsage: runResult.meta?.agentMeta?.lastCallUsage,
|
|
contextTokensUsed,
|
|
newSessionId: runResult.meta?.agentMeta?.sessionId,
|
|
newSessionFile: runResult.meta?.agentMeta?.sessionFile,
|
|
});
|
|
const refreshedSessionEntry =
|
|
sessionKey && activeSessionStore ? activeSessionStore[sessionKey] : undefined;
|
|
if (refreshedSessionEntry) {
|
|
activeSessionEntry = refreshedSessionEntry;
|
|
refreshQueuedFollowupSession({
|
|
key: queueKey,
|
|
previousSessionId,
|
|
nextSessionId: refreshedSessionEntry.sessionId,
|
|
nextSessionFile: refreshedSessionEntry.sessionFile,
|
|
});
|
|
}
|
|
|
|
// Inject post-compaction workspace context for the next agent turn
|
|
if (sessionKey) {
|
|
const workspaceDir = process.cwd();
|
|
readPostCompactionContext(workspaceDir, {
|
|
cfg,
|
|
agentId: resolveSessionAgentId({ sessionKey, config: cfg }),
|
|
})
|
|
.then((contextContent) => {
|
|
if (contextContent) {
|
|
enqueueSystemEvent(contextContent, { sessionKey, trusted: true });
|
|
}
|
|
})
|
|
.catch(() => {
|
|
// Silent failure — post-compaction context is best-effort
|
|
});
|
|
}
|
|
|
|
if (verboseEnabled) {
|
|
const suffix = typeof count === "number" ? ` (count ${count})` : "";
|
|
verboseNotices.push({ text: `🧹 Auto-compaction complete${suffix}.` });
|
|
}
|
|
}
|
|
const prefixPayloads = [...verboseNotices];
|
|
const isHookBlockedRun = runResult.meta?.error?.kind === "hook_block";
|
|
const rawUserText = isHookBlockedRun
|
|
? runResult.meta?.finalPromptText
|
|
: (runResult.meta?.finalPromptText ??
|
|
sessionCtx.CommandBody ??
|
|
sessionCtx.RawBody ??
|
|
sessionCtx.BodyForAgent ??
|
|
sessionCtx.Body);
|
|
const rawAssistantText = isHookBlockedRun
|
|
? undefined
|
|
: (runResult.meta?.finalAssistantRawText ?? runResult.meta?.finalAssistantVisibleText);
|
|
const traceAuthorized = followupRun.run.traceAuthorized === true;
|
|
const executionTrace = mergeExecutionTrace({
|
|
fallbackAttempts,
|
|
executionTrace: runResult.meta?.executionTrace as TraceExecutionView | undefined,
|
|
provider: providerUsed,
|
|
model: modelUsed,
|
|
runner: isCliProvider(providerUsed, cfg) ? "cli" : "embedded",
|
|
});
|
|
const requestShaping = {
|
|
authMode:
|
|
runResult.meta?.requestShaping?.authMode ??
|
|
(cfg?.models?.providers && providerUsed in cfg.models.providers
|
|
? (resolveModelAuthMode(providerUsed, cfg, undefined, {
|
|
workspaceDir: followupRun.run.workspaceDir,
|
|
}) ?? undefined)
|
|
: undefined),
|
|
thinking:
|
|
runResult.meta?.requestShaping?.thinking ??
|
|
normalizeOptionalString(followupRun.run.thinkLevel),
|
|
reasoning:
|
|
runResult.meta?.requestShaping?.reasoning ??
|
|
normalizeOptionalString(followupRun.run.reasoningLevel),
|
|
verbose:
|
|
runResult.meta?.requestShaping?.verbose ?? normalizeOptionalString(resolvedVerboseLevel),
|
|
trace:
|
|
runResult.meta?.requestShaping?.trace ??
|
|
normalizeOptionalString(activeSessionEntry?.traceLevel),
|
|
fallbackEligible:
|
|
runResult.meta?.requestShaping?.fallbackEligible ??
|
|
hasConfiguredModelFallbacks({
|
|
cfg,
|
|
agentId: followupRun.run.agentId,
|
|
sessionKey: followupRun.run.sessionKey,
|
|
}),
|
|
blockStreaming:
|
|
runResult.meta?.requestShaping?.blockStreaming ??
|
|
normalizeOptionalString(resolvedBlockStreamingBreak),
|
|
};
|
|
const promptSegments =
|
|
(runResult.meta?.promptSegments as TracePromptSegmentView[] | undefined) ??
|
|
derivePromptSegments(rawUserText);
|
|
const toolSummary = runResult.meta?.toolSummary as TraceToolSummaryView | undefined;
|
|
const completion =
|
|
(runResult.meta?.completion as TraceCompletionView | undefined) ??
|
|
(runResult.meta?.stopReason
|
|
? {
|
|
stopReason: runResult.meta.stopReason,
|
|
finishReason: runResult.meta.stopReason,
|
|
...(runResult.meta.stopReason.toLowerCase().includes("refusal")
|
|
? { refusal: true }
|
|
: {}),
|
|
}
|
|
: undefined);
|
|
const contextManagement = {
|
|
...(typeof activeSessionEntry?.compactionCount === "number"
|
|
? { sessionCompactions: activeSessionEntry.compactionCount }
|
|
: {}),
|
|
...(typeof runResult.meta?.contextManagement?.lastTurnCompactions === "number"
|
|
? { lastTurnCompactions: runResult.meta.contextManagement.lastTurnCompactions }
|
|
: typeof runResult.meta?.agentMeta?.compactionCount === "number"
|
|
? { lastTurnCompactions: runResult.meta.agentMeta.compactionCount }
|
|
: {}),
|
|
...(runResult.meta?.contextManagement &&
|
|
typeof runResult.meta.contextManagement.preflightCompactionApplied === "boolean"
|
|
? {
|
|
preflightCompactionApplied: runResult.meta.contextManagement.preflightCompactionApplied,
|
|
}
|
|
: preflightCompactionApplied
|
|
? { preflightCompactionApplied }
|
|
: {}),
|
|
...(runResult.meta?.contextManagement &&
|
|
typeof runResult.meta.contextManagement.postCompactionContextInjected === "boolean"
|
|
? {
|
|
postCompactionContextInjected:
|
|
runResult.meta.contextManagement.postCompactionContextInjected,
|
|
}
|
|
: {}),
|
|
} satisfies TraceContextManagementView;
|
|
const sessionUsage =
|
|
traceAuthorized && activeSessionEntry?.traceLevel === "raw"
|
|
? await accumulateSessionUsageFromTranscript({
|
|
sessionId: runResult.meta?.agentMeta?.sessionId ?? followupRun.run.sessionId,
|
|
storePath,
|
|
sessionFile: followupRun.run.sessionFile,
|
|
})
|
|
: undefined;
|
|
const traceEnabledForSender =
|
|
traceAuthorized &&
|
|
(activeSessionEntry?.traceLevel === "on" || activeSessionEntry?.traceLevel === "raw");
|
|
const shouldAppendTracePayload = verboseEnabled || traceEnabledForSender;
|
|
let trailingPluginStatusPayload: ReplyPayload | undefined;
|
|
if (shouldAppendTracePayload) {
|
|
const pluginStatusPayload = buildInlinePluginStatusPayload({
|
|
entry: activeSessionEntry,
|
|
includeTraceLines: traceEnabledForSender,
|
|
});
|
|
const rawTracePayload =
|
|
traceAuthorized && activeSessionEntry?.traceLevel === "raw"
|
|
? buildInlineRawTracePayload({
|
|
entry: activeSessionEntry,
|
|
rawUserText,
|
|
rawAssistantText,
|
|
sessionUsage,
|
|
usage: runResult.meta?.agentMeta?.usage,
|
|
lastCallUsage: runResult.meta?.agentMeta?.lastCallUsage,
|
|
provider: providerUsed,
|
|
model: modelUsed,
|
|
contextLimit: contextTokensUsed,
|
|
promptTokens,
|
|
executionTrace,
|
|
requestShaping,
|
|
promptSegments,
|
|
toolSummary,
|
|
completion,
|
|
contextManagement,
|
|
})
|
|
: undefined;
|
|
trailingPluginStatusPayload =
|
|
pluginStatusPayload && rawTracePayload
|
|
? { text: `${pluginStatusPayload.text}\n\n${rawTracePayload.text}` }
|
|
: (pluginStatusPayload ?? rawTracePayload);
|
|
}
|
|
if (prefixPayloads.length > 0) {
|
|
finalPayloads = [...prefixPayloads, ...finalPayloads];
|
|
}
|
|
if (trailingPluginStatusPayload) {
|
|
finalPayloads = [...finalPayloads, trailingPluginStatusPayload];
|
|
}
|
|
if (responseUsageLine) {
|
|
finalPayloads = appendUsageLine(finalPayloads, responseUsageLine);
|
|
}
|
|
if (isHookBlockedRun) {
|
|
finalPayloads = markBeforeAgentRunBlockedPayloads(finalPayloads);
|
|
}
|
|
|
|
// Capture only policy-visible final payloads in session store to support
|
|
// durable delivery retries. Hidden reasoning, message-tool-only replies,
|
|
// and sendPolicy-denied replies must not become heartbeat-replayable text.
|
|
if (sessionKey && storePath && finalPayloads.length > 0) {
|
|
const sendPolicy = resolveSendPolicy({
|
|
cfg,
|
|
entry: activeSessionEntry,
|
|
sessionKey: params.runtimePolicySessionKey ?? sessionKey,
|
|
channel:
|
|
sessionCtx.OriginatingChannel ??
|
|
sessionCtx.Surface ??
|
|
sessionCtx.Provider ??
|
|
activeSessionEntry?.channel,
|
|
chatType: activeSessionEntry?.chatType,
|
|
});
|
|
const sourceReplyPolicy = resolveSourceReplyVisibilityPolicy({
|
|
cfg,
|
|
ctx: sessionCtx,
|
|
requested: opts?.sourceReplyDeliveryMode,
|
|
sendPolicy,
|
|
});
|
|
const pendingText = sourceReplyPolicy.suppressDelivery
|
|
? ""
|
|
: buildPendingFinalDeliveryText(finalPayloads);
|
|
if (pendingText) {
|
|
await updateSessionStoreEntry({
|
|
storePath,
|
|
sessionKey,
|
|
update: async () => ({
|
|
pendingFinalDelivery: true,
|
|
pendingFinalDeliveryText: pendingText,
|
|
pendingFinalDeliveryCreatedAt: Date.now(),
|
|
updatedAt: Date.now(),
|
|
}),
|
|
});
|
|
}
|
|
}
|
|
|
|
const result = returnWithQueuedFollowupDrain(
|
|
finalPayloads.length === 1 ? finalPayloads[0] : finalPayloads,
|
|
);
|
|
|
|
return result;
|
|
} catch (error) {
|
|
if (
|
|
replyOperation.result?.kind === "aborted" &&
|
|
replyOperation.result.code === "aborted_for_restart"
|
|
) {
|
|
return returnWithQueuedFollowupDrain(
|
|
markReplyPayloadForSourceSuppressionDelivery({
|
|
text: "⚠️ Gateway is restarting. Please wait a few seconds and try again.",
|
|
}),
|
|
);
|
|
}
|
|
if (replyOperation.result?.kind === "aborted") {
|
|
return returnWithQueuedFollowupDrain({ text: SILENT_REPLY_TOKEN });
|
|
}
|
|
if (error instanceof GatewayDrainingError) {
|
|
replyOperation.fail("gateway_draining", error);
|
|
return returnWithQueuedFollowupDrain(
|
|
markReplyPayloadForSourceSuppressionDelivery({
|
|
text: "⚠️ Gateway is restarting. Please wait a few seconds and try again.",
|
|
}),
|
|
);
|
|
}
|
|
if (error instanceof CommandLaneClearedError) {
|
|
replyOperation.fail("command_lane_cleared", error);
|
|
return returnWithQueuedFollowupDrain(
|
|
markReplyPayloadForSourceSuppressionDelivery({
|
|
text: "⚠️ Gateway is restarting. Please wait a few seconds and try again.",
|
|
}),
|
|
);
|
|
}
|
|
const knownFailurePayload = buildKnownAgentRunFailureReplyPayload({
|
|
err: error,
|
|
sessionCtx,
|
|
resolvedVerboseLevel,
|
|
});
|
|
if (knownFailurePayload) {
|
|
replyOperation.fail("run_failed", error);
|
|
return returnWithQueuedFollowupDrain(knownFailurePayload);
|
|
}
|
|
replyOperation.fail("run_failed", error);
|
|
// Keep the followup queue moving even when an unexpected exception escapes
|
|
// the run path; the caller still receives the original error.
|
|
returnWithQueuedFollowupDrain(undefined);
|
|
throw error;
|
|
} finally {
|
|
if (shouldDrainQueuedFollowupsAfterClear) {
|
|
replyOperation.completeThen(drainQueuedFollowupsAfterClear);
|
|
} else {
|
|
replyOperation.complete();
|
|
}
|
|
blockReplyPipeline?.stop();
|
|
typing.markRunComplete();
|
|
// Safety net: the dispatcher's onIdle callback normally fires
|
|
// markDispatchIdle(), but if the dispatcher exits early, errors,
|
|
// or the reply path doesn't go through it cleanly, the second
|
|
// signal never fires and the typing keepalive loop runs forever.
|
|
// Calling this twice is harmless — cleanup() is guarded by the
|
|
// `active` flag. Same pattern as the followup runner fix (#26881).
|
|
typing.markDispatchIdle();
|
|
}
|
|
}
|