refactor: split hot TypeScript modules and add LOC ratchet (#105894)

* refactor: split high-churn agent and chat modules

* ci: enforce TypeScript LOC ratchet

* ci: anchor LOC baseline updates to merge base

* test: cover LOC ratchet check plans

* chore: refresh TypeScript LOC baseline
This commit is contained in:
Peter Steinberger
2026-07-12 19:57:42 -07:00
committed by GitHub
parent fb48df4a0f
commit c3dbaf4375
26 changed files with 8650 additions and 6626 deletions
+8 -2
View File
@@ -1559,6 +1559,7 @@ jobs:
env:
HISTORICAL_TARGET: ${{ needs.preflight.outputs.compatibility_target }}
FORMAT_CHECK: ${{ needs.preflight.outputs.run_format_check }}
LOC_BASE_SHA: ${{ github.event_name == 'pull_request' && github.event.pull_request.base.sha || (github.event_name == 'push' && github.event.before || '') }}
OPENCLAW_LOCAL_CHECK: "0"
TASK: ${{ matrix.task }}
PR_BASE_SHA: ${{ github.event_name == 'pull_request' && github.event.pull_request.base.sha || '' }}
@@ -1568,12 +1569,17 @@ jobs:
case "$TASK" in
guards)
pnpm check:no-conflict-markers
if [ -n "$LOC_BASE_SHA" ]; then
git fetch --no-tags --depth=1 origin "+${LOC_BASE_SHA}:refs/remotes/origin/loc-base"
pnpm check:loc --base-ref refs/remotes/origin/loc-base
else
pnpm check:loc
fi
pnpm tool-display:check
pnpm check:host-env-policy:swift
pnpm dup:check:coverage
if [ -n "$PR_BASE_SHA" ]; then
git fetch --no-tags --depth=1 origin "+${PR_BASE_SHA}:refs/remotes/origin/pr-base"
node scripts/report-test-temp-creations.mjs --base refs/remotes/origin/pr-base --head HEAD --no-merge-base
node scripts/report-test-temp-creations.mjs --base refs/remotes/origin/loc-base --head HEAD --no-merge-base
fi
pnpm deps:patches:check
pnpm lint:webhook:no-low-level-body-read
+1
View File
@@ -1554,6 +1554,7 @@
"check:host-env-policy:swift": "node scripts/generate-host-env-security-policy-swift.mjs --check",
"check:import-cycles": "node --import tsx scripts/check-import-cycles.ts",
"check:loc": "node --import tsx scripts/check-ts-max-loc.ts --max 500",
"check:loc:update": "node --import tsx scripts/check-ts-max-loc.ts --max 500 --write-baseline",
"check:madge-import-cycles": "node --import tsx scripts/check-madge-import-cycles.ts",
"check:media-download-helpers": "node scripts/check-media-download-helper-roundtrip.mjs",
"check:no-conflict-markers": "node scripts/check-no-conflict-markers.mjs",
+1
View File
@@ -314,6 +314,7 @@ export function createChangedCheckPlan(result, options = {}) {
};
add("conflict markers", ["check:no-conflict-markers"]);
add("TypeScript LOC ratchet", ["check:loc"]);
add("changelog attributions", ["check:changelog-attributions"]);
add("guarded extension wildcard re-exports", ["lint:extensions:no-guarded-wildcard-reexports"]);
add("plugin-sdk wildcard re-exports", ["lint:extensions:no-plugin-sdk-wildcard-reexports"]);
+270 -33
View File
@@ -1,18 +1,28 @@
// Check Ts Max Loc script supports OpenClaw repository automation.
// Enforces the TypeScript file-size ceiling while grandfathering the existing backlog.
import { execFileSync } from "node:child_process";
import { existsSync } from "node:fs";
import { readFile } from "node:fs/promises";
import { readFile, writeFile } from "node:fs/promises";
import { resolve } from "node:path";
import { pathToFileURL } from "node:url";
const DEFAULT_BASELINE_PATH = "scripts/ts-max-loc-baseline.json";
function writeStdoutLine(message: string): void {
process.stdout.write(`${message}\n`);
}
type ParsedArgs = {
export type ParsedArgs = {
baseRef?: string;
baselinePath: string;
maxLines: number;
writeBaseline: boolean;
};
function parseArgs(argv: string[]): ParsedArgs {
export function parseArgs(argv: string[]): ParsedArgs {
let baseRef: string | undefined;
let baselinePath = DEFAULT_BASELINE_PATH;
let maxLines = 500;
let writeBaseline = false;
for (let index = 0; index < argv.length; index++) {
const arg = argv[index];
@@ -28,14 +38,36 @@ function parseArgs(argv: string[]): ParsedArgs {
index++;
continue;
}
if (arg === "--baseline") {
const next = argv[index + 1];
if (!next) {
throw new Error("--baseline requires a path");
}
baselinePath = next;
index++;
continue;
}
if (arg === "--base-ref") {
const next = argv[index + 1];
if (!next || next.startsWith("-") || !/^[A-Za-z0-9_./-]+$/u.test(next)) {
throw new Error("--base-ref requires a git ref");
}
baseRef = next;
index++;
continue;
}
if (arg === "--write-baseline") {
writeBaseline = true;
continue;
}
throw new Error(`Unknown argument: ${arg}`);
}
return { maxLines };
return { baseRef, baselinePath, maxLines, writeBaseline };
}
function gitLsFilesAll(): string[] {
// Include untracked files too so local refactors dont pass by accident.
// Include untracked files too so local refactors do not pass by accident.
const stdout = execFileSync("git", ["ls-files", "--cached", "--others", "--exclude-standard"], {
encoding: "utf8",
});
@@ -45,14 +77,198 @@ function gitLsFilesAll(): string[] {
.filter(Boolean);
}
async function countLines(filePath: string): Promise<number> {
const content = await readFile(filePath, "utf8");
// Count physical lines. Keeps the rule simple + predictable.
return content.split("\n").length;
export function isProductionTypeScriptFile(filePath: string): boolean {
return (
/\.(?:ts|tsx|mts|cts)$/u.test(filePath) &&
!/(^|\/)(test|tests|__tests__|test-helpers?|test-support)(\/|$)|\.(test|spec|suite)\.[cm]?tsx?$|(?:^|[/.-])test-(?:helpers?|support|harness)(?:[/.-]|$)/u.test(
filePath,
)
);
}
async function main(argv = process.argv.slice(2)): Promise<number> {
// Makes `... | head` safe.
export function countPhysicalLines(content: string): number {
if (content.length === 0) {
return 0;
}
const splitCount = content.split("\n").length;
return content.endsWith("\n") ? splitCount - 1 : splitCount;
}
async function countLines(filePath: string): Promise<number> {
const content = await readFile(filePath, "utf8");
return countPhysicalLines(content);
}
type LocResult = {
filePath: string;
lines: number;
};
type LocBaseline = Record<string, number>;
export type LocRatchetViolation = LocResult & {
baselineLines?: number;
reason: "baseline-missing" | "baseline-stale" | "grew";
};
export function findLocRatchetViolations(params: {
baseline: LocBaseline;
maxLines: number;
results: LocResult[];
}): LocRatchetViolation[] {
const currentByPath = new Map(params.results.map((result) => [result.filePath, result.lines]));
const violations: LocRatchetViolation[] = [];
for (const result of params.results) {
const baselineLines = params.baseline[result.filePath];
if (result.lines <= params.maxLines) {
if (baselineLines !== undefined) {
violations.push({ ...result, baselineLines, reason: "baseline-stale" });
}
continue;
}
if (baselineLines === undefined) {
violations.push({ ...result, reason: "baseline-missing" });
} else if (result.lines > baselineLines) {
violations.push({ ...result, baselineLines, reason: "grew" });
} else if (result.lines < baselineLines) {
// Require the baseline to move down with every successful split.
violations.push({ ...result, baselineLines, reason: "baseline-stale" });
}
}
for (const [filePath, baselineLines] of Object.entries(params.baseline)) {
if (!currentByPath.has(filePath)) {
violations.push({ filePath, lines: 0, baselineLines, reason: "baseline-stale" });
}
}
return violations.toSorted(
(left, right) => right.lines - left.lines || left.filePath.localeCompare(right.filePath),
);
}
export function findLocBaselineUpdateViolations(params: {
baseline: LocBaseline;
maxLines: number;
results: LocResult[];
}): LocRatchetViolation[] {
const violations: LocRatchetViolation[] = [];
for (const result of params.results) {
if (result.lines <= params.maxLines) {
continue;
}
const baselineLines = params.baseline[result.filePath];
if (baselineLines === undefined) {
violations.push({ ...result, reason: "baseline-missing" });
} else if (result.lines > baselineLines) {
violations.push({ ...result, baselineLines, reason: "grew" });
}
}
return violations.toSorted(
(left, right) => right.lines - left.lines || left.filePath.localeCompare(right.filePath),
);
}
export function findVersionedBaselineViolations(params: {
baseline: LocBaseline;
baseBaseline: LocBaseline;
}): LocRatchetViolation[] {
const violations: LocRatchetViolation[] = [];
for (const [filePath, lines] of Object.entries(params.baseline)) {
const baselineLines = params.baseBaseline[filePath];
if (baselineLines === undefined) {
violations.push({ filePath, lines, reason: "baseline-missing" });
} else if (lines > baselineLines) {
violations.push({ filePath, lines, baselineLines, reason: "grew" });
}
}
return violations.toSorted(
(left, right) => right.lines - left.lines || left.filePath.localeCompare(right.filePath),
);
}
async function readBaseline(filePath: string): Promise<LocBaseline> {
return parseBaseline(await readFile(filePath, "utf8"), filePath);
}
function parseBaseline(content: string, source: string): LocBaseline {
const parsed = JSON.parse(content) as unknown;
if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) {
throw new Error(`Invalid TypeScript LOC baseline: ${source}`);
}
const baseline: LocBaseline = {};
for (const [entryPath, value] of Object.entries(parsed)) {
if (!Number.isSafeInteger(value) || (value as number) <= 0) {
throw new Error(`Invalid TypeScript LOC baseline entry: ${entryPath}`);
}
baseline[entryPath] = value as number;
}
return baseline;
}
function tryGitOutput(args: string[]): string | undefined {
try {
return execFileSync("git", args, {
encoding: "utf8",
stdio: ["ignore", "pipe", "ignore"],
}).trim();
} catch {
return undefined;
}
}
function resolveComparisonBaseRef(
baselinePath: string,
explicitBaseRef?: string,
): string | undefined {
if (explicitBaseRef) {
return explicitBaseRef;
}
const head = tryGitOutput(["rev-parse", "HEAD"]);
const mergeBase = tryGitOutput(["merge-base", "HEAD", "origin/main"]);
if (mergeBase && mergeBase !== head) {
return mergeBase;
}
const changedBaselinePath = tryGitOutput(["diff", "--name-only", "HEAD", "--", baselinePath]);
if (changedBaselinePath?.split("\n").includes(baselinePath)) {
return "HEAD";
}
return tryGitOutput(["rev-parse", "--verify", "HEAD^"]) ? "HEAD^" : undefined;
}
function readBaselineAtRef(
baseRef: string | undefined,
baselinePath: string,
): LocBaseline | undefined {
if (!baseRef) {
return undefined;
}
if (!tryGitOutput(["rev-parse", "--verify", `${baseRef}^{commit}`])) {
throw new Error(`Invalid TypeScript LOC comparison ref: ${baseRef}`);
}
const content = tryGitOutput(["show", `${baseRef}:${baselinePath}`]);
return content === undefined ? undefined : parseBaseline(content, `${baseRef}:${baselinePath}`);
}
function buildBaseline(results: LocResult[], maxLines: number): LocBaseline {
return Object.fromEntries(
results
.filter((result) => result.lines > maxLines)
.toSorted((left, right) => left.filePath.localeCompare(right.filePath))
.map((result) => [result.filePath, result.lines]),
);
}
function reportViolations(violations: LocRatchetViolation[]): void {
for (const violation of violations) {
writeStdoutLine(
`${violation.lines}\t${violation.baselineLines ?? "-"}\t${violation.reason}\t${violation.filePath}`,
);
}
}
export async function main(argv = process.argv.slice(2)): Promise<number> {
process.stdout.on("error", (error: NodeJS.ErrnoException) => {
if (error.code === "EPIPE") {
process.exit(0);
@@ -60,37 +276,58 @@ async function main(argv = process.argv.slice(2)): Promise<number> {
throw error;
});
const { maxLines } = parseArgs(argv);
const { baseRef, baselinePath, maxLines, writeBaseline } = parseArgs(argv);
const files = gitLsFilesAll()
.filter((filePath) => existsSync(filePath))
.filter((filePath) => filePath.endsWith(".ts") || filePath.endsWith(".tsx"));
.filter(isProductionTypeScriptFile);
const results = await Promise.all(
files.map(async (filePath) => ({ filePath, lines: await countLines(filePath) })),
);
const offenders = results
.filter((result) => result.lines > maxLines)
.toSorted((a, b) => b.lines - a.lines);
if (!offenders.length) {
if (writeBaseline) {
const baseline = await readBaseline(baselinePath);
const comparisonBaseRef = resolveComparisonBaseRef(baselinePath, baseRef);
if (!comparisonBaseRef) {
throw new Error("Unable to resolve a comparison ref for the TypeScript LOC baseline update");
}
const baseBaseline = readBaselineAtRef(comparisonBaseRef, baselinePath);
// A missing baseline at a valid base ref is the one-time initialization path.
const violations = [
...(baseBaseline ? findVersionedBaselineViolations({ baseline, baseBaseline }) : []),
...findLocBaselineUpdateViolations({ baseline, maxLines, results }),
];
reportViolations(violations);
if (violations.length > 0) {
return 1;
}
const updatedBaseline = buildBaseline(results, maxLines);
await writeFile(baselinePath, `${JSON.stringify(updatedBaseline, null, 2)}\n`, "utf8");
writeStdoutLine(`updated ${baselinePath} (${Object.keys(updatedBaseline).length} files)`);
return 0;
}
// Minimal, grep-friendly output.
for (const offender of offenders) {
writeStdoutLine(`${offender.lines}\t${offender.filePath}`);
}
return 1;
const baseline = await readBaseline(baselinePath);
const baseBaseline = readBaselineAtRef(
resolveComparisonBaseRef(baselinePath, baseRef),
baselinePath,
);
const violations = [
...(baseBaseline ? findVersionedBaselineViolations({ baseline, baseBaseline }) : []),
...findLocRatchetViolations({ baseline, maxLines, results }),
];
reportViolations(violations);
return violations.length === 0 ? 0 : 1;
}
try {
const exitCode = await main();
if (exitCode !== 0) {
process.exit(exitCode);
const invokedPath = process.argv[1] ? pathToFileURL(resolve(process.argv[1])).href : undefined;
if (invokedPath === import.meta.url) {
try {
const exitCode = await main();
if (exitCode !== 0) {
process.exit(exitCode);
}
} catch (error) {
process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`);
process.exit(1);
}
} catch (error) {
process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`);
process.exit(1);
}
+1
View File
@@ -87,6 +87,7 @@ export async function main(argv = process.argv.slice(2)) {
parallel: true,
commands: [
{ name: "conflict markers", args: ["check:no-conflict-markers"] },
{ name: "TypeScript LOC ratchet", args: ["check:loc"] },
{ name: "changelog attributions", args: ["check:changelog-attributions"] },
{ name: "database-first legacy-store guard", args: ["check:database-first-legacy-stores"] },
{
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,116 @@
import { formatErrorMessage } from "../../../infra/errors.js";
import { buildTrajectoryArtifacts } from "../../../trajectory/metadata.js";
import {
resolveAttemptTrajectoryTerminal,
resolveTerminalAssistantTexts,
} from "./attempt-trajectory-status.js";
import { resolveFinalAssistantVisibleText } from "./helpers.js";
import type { EmbeddedRunAttemptResult, EmbeddedRunAttemptTrajectoryRecorder } from "./types.js";
type FinalizeEmbeddedAttemptParams = {
result: EmbeddedRunAttemptResult;
trajectoryRecorder?: EmbeddedRunAttemptTrajectoryRecorder | null;
synthesizedPayloadCount: number;
emptyAssistantReplyIsSilent: boolean;
hasTerminalOutput: boolean;
silentExpected?: boolean;
};
/** Classifies the completed attempt and records its terminal trajectory artifacts. */
export function finalizeEmbeddedAttempt(
params: FinalizeEmbeddedAttemptParams,
): EmbeddedRunAttemptResult {
const { result, trajectoryRecorder } = params;
const terminalAssistantTexts = resolveTerminalAssistantTexts({
assistantTexts: result.assistantTexts,
lastAssistantStopReason: result.lastAssistant?.stopReason,
lastAssistantVisibleText: resolveFinalAssistantVisibleText(result.lastAssistant),
});
const terminal = resolveAttemptTrajectoryTerminal({
promptError: result.promptError,
aborted: result.aborted,
externalAbort: result.externalAbort,
timedOut: result.timedOut,
assistantTexts: terminalAssistantTexts,
toolMetas: result.toolMetas,
didSendViaMessagingTool: result.didSendViaMessagingTool,
didSendDeterministicApprovalPrompt: result.didSendDeterministicApprovalPrompt === true,
messagingToolSentTexts: result.messagingToolSentTexts,
messagingToolSentMediaUrls: result.messagingToolSentMediaUrls,
messagingToolSentTargets: result.messagingToolSentTargets,
successfulCronAdds: result.successfulCronAdds ?? 0,
synthesizedPayloadCount: params.synthesizedPayloadCount,
acceptedSessionSpawns: result.acceptedSessionSpawns,
heartbeatToolResponse: result.heartbeatToolResponse,
clientToolCalls: result.clientToolCalls,
yieldDetected: result.yieldDetected,
lastToolError: result.lastToolError,
silentExpected: params.silentExpected,
emptyAssistantReplyIsSilent: params.emptyAssistantReplyIsSilent,
lastAssistantStopReason: result.lastAssistant?.stopReason,
hasTerminalOutput: params.hasTerminalOutput,
});
const promptError = result.promptError ? formatErrorMessage(result.promptError) : undefined;
trajectoryRecorder?.recordEvent("model.completed", {
aborted: result.aborted,
externalAbort: result.externalAbort,
timedOut: result.timedOut,
idleTimedOut: result.idleTimedOut,
timedOutDuringCompaction: result.timedOutDuringCompaction,
timedOutDuringToolExecution: result.timedOutDuringToolExecution,
timedOutByRunBudget: result.timedOutByRunBudget,
promptError,
promptErrorSource: result.promptErrorSource,
terminalError: terminal.terminalError,
usage: result.attemptUsage,
promptCache: result.promptCache,
compactionCount: result.compactionCount,
assistantTexts: result.assistantTexts,
finalPromptText: result.finalPromptText,
messagesSnapshot: result.messagesSnapshot,
});
trajectoryRecorder?.recordEvent(
"trace.artifacts",
buildTrajectoryArtifacts({
status: terminal.status,
aborted: result.aborted,
externalAbort: result.externalAbort,
timedOut: result.timedOut,
idleTimedOut: result.idleTimedOut,
timedOutDuringCompaction: result.timedOutDuringCompaction,
timedOutDuringToolExecution: result.timedOutDuringToolExecution === true,
timedOutByRunBudget: result.timedOutByRunBudget === true,
promptError,
promptErrorSource: result.promptErrorSource,
terminalError: terminal.terminalError,
usage: result.attemptUsage,
promptCache: result.promptCache,
compactionCount: result.compactionCount ?? 0,
assistantTexts: result.assistantTexts,
finalPromptText: result.finalPromptText,
itemLifecycle: result.itemLifecycle,
toolMetas: result.toolMetas,
didSendViaMessagingTool: result.didSendViaMessagingTool,
successfulCronAdds: result.successfulCronAdds ?? 0,
messagingToolSentTexts: result.messagingToolSentTexts,
messagingToolSentMediaUrls: result.messagingToolSentMediaUrls,
messagingToolSentTargets: result.messagingToolSentTargets,
lastToolError: result.lastToolError,
}),
);
trajectoryRecorder?.recordEvent("session.ended", {
status: terminal.status,
aborted: result.aborted,
externalAbort: result.externalAbort,
timedOut: result.timedOut,
idleTimedOut: result.idleTimedOut,
timedOutDuringCompaction: result.timedOutDuringCompaction,
timedOutDuringToolExecution: result.timedOutDuringToolExecution,
timedOutByRunBudget: result.timedOutByRunBudget,
promptError,
terminalError: terminal.terminalError,
});
return result;
}
@@ -0,0 +1,191 @@
/**
* Resolves workspace, sandbox, provider runtime, and phase reporting for an embedded attempt.
*/
import fs from "node:fs/promises";
import { normalizeProviderId } from "@openclaw/model-catalog-core/provider-id";
import { getCurrentPluginMetadataSnapshot } from "../../../plugins/current-plugin-metadata-snapshot.js";
import type { PluginMetadataSnapshot } from "../../../plugins/plugin-metadata-snapshot.types.js";
import {
resolveProviderRuntimePluginHandle,
type ProviderRuntimePluginHandle,
} from "../../../plugins/provider-hook-runtime.js";
import { resolveUserPath } from "../../../utils.js";
import { resolveSessionAgentIds } from "../../agent-scope.js";
import { resolveSandboxContext } from "../../sandbox.js";
import { log } from "../logger.js";
import { mapThinkingLevel, mapThinkingLevelForProvider } from "../utils.js";
import { configureEmbeddedAttemptHttpRuntime } from "./attempt-http-runtime.js";
import {
createEmbeddedRunStageTracker,
formatEmbeddedRunStageSummary,
shouldWarnEmbeddedRunStageSummary,
} from "./attempt-stage-timing.js";
import { resolveAttemptFsWorkspaceOnly } from "./attempt.prompt-helpers.js";
import type { EmbeddedRunAttemptParams } from "./types.js";
function pluginMetadataSnapshotCoversProvider(
snapshot: PluginMetadataSnapshot | undefined,
provider: string,
): snapshot is PluginMetadataSnapshot {
const normalizedProvider = normalizeProviderId(provider);
if (!snapshot || !normalizedProvider) {
return false;
}
return snapshot.manifestRegistry.plugins.some((plugin) => {
const ownsProvider = plugin.providers.some(
(providerId) => normalizeProviderId(providerId) === normalizedProvider,
);
if (ownsProvider) {
return true;
}
const modelCatalogProviderIds = [
...Object.keys(plugin.modelCatalog?.providers ?? {}),
...Object.keys(plugin.modelCatalog?.aliases ?? {}),
];
return modelCatalogProviderIds.some(
(providerId) => normalizeProviderId(providerId) === normalizedProvider,
);
});
}
export async function prepareEmbeddedAttemptSetup(params: EmbeddedRunAttemptParams) {
const resolvedWorkspace = resolveUserPath(params.workspaceDir);
// Ultra is a logical orchestration mode, not a provider effort. Preserve it for
// prompt/status surfaces, then lower only at agent-core and provider boundaries.
const agentCoreThinkingLevel = mapThinkingLevel(params.thinkLevel);
const providerThinkingLevel = mapThinkingLevelForProvider(params.thinkLevel);
const proactiveSubagentOrchestration = params.thinkLevel === "ultra";
configureEmbeddedAttemptHttpRuntime({ timeoutMs: params.timeoutMs });
log.debug(
`embedded run start: runId=${params.runId} sessionId=${params.sessionId} provider=${params.provider} model=${params.modelId} thinking=${params.thinkLevel} messageChannel=${params.messageChannel ?? params.messageProvider ?? "unknown"}`,
);
const prepStages = createEmbeddedRunStageTracker();
const emitPrepStageSummary = (phase: string) => {
const summary = prepStages.snapshot();
const shouldWarn = shouldWarnEmbeddedRunStageSummary(summary);
if (!shouldWarn && !log.isEnabled("trace")) {
return;
}
const message = formatEmbeddedRunStageSummary(
`[trace:embedded-run] prep stages: runId=${params.runId} sessionId=${params.sessionId} phase=${phase}`,
summary,
);
if (shouldWarn) {
log.warn(message);
} else {
log.trace(message);
}
};
const emitCorePluginToolStageSummary = (
phase: string,
summary: ReturnType<typeof prepStages.snapshot>,
) => {
if (summary.stages.length === 0) {
return;
}
const shouldWarn = shouldWarnEmbeddedRunStageSummary(summary, {
totalThresholdMs: 5_000,
stageThresholdMs: 2_000,
});
if (!shouldWarn && !log.isEnabled("trace")) {
return;
}
const message = formatEmbeddedRunStageSummary(
`[trace:embedded-run] core-plugin-tool stages: runId=${params.runId} sessionId=${params.sessionId} phase=${phase}`,
summary,
);
if (shouldWarn) {
log.warn(message);
} else {
log.trace(message);
}
};
await fs.mkdir(resolvedWorkspace, { recursive: true });
const sandboxSessionKey =
params.sandboxSessionKey?.trim() || params.sessionKey?.trim() || params.sessionId;
const sandbox = await resolveSandboxContext({
config: params.config,
execOverrides: params.execOverrides,
sessionKey: sandboxSessionKey,
workspaceDir: resolvedWorkspace,
});
const effectiveWorkspace = sandbox?.enabled
? sandbox.workspaceAccess === "rw"
? resolvedWorkspace
: sandbox.workspaceDir
: resolvedWorkspace;
const requestedCwd = params.cwd ? resolveUserPath(params.cwd) : undefined;
if (sandbox?.enabled && requestedCwd && requestedCwd !== resolvedWorkspace) {
throw new Error(
"cwd override is not supported for sandboxed embedded agent runs; omit cwd or use the agent workspace as cwd",
);
}
const effectiveCwd = sandbox?.enabled ? effectiveWorkspace : (requestedCwd ?? effectiveWorkspace);
await fs.mkdir(effectiveWorkspace, { recursive: true });
let currentPluginMetadataSnapshotResolved = false;
let currentPluginMetadataSnapshot: PluginMetadataSnapshot | undefined;
const getCurrentAttemptPluginMetadataSnapshot = () => {
if (!currentPluginMetadataSnapshotResolved) {
currentPluginMetadataSnapshot = getCurrentPluginMetadataSnapshot({
allowScopedSnapshot: true,
config: params.config,
env: process.env,
workspaceDir: effectiveWorkspace,
});
currentPluginMetadataSnapshotResolved = true;
}
return currentPluginMetadataSnapshot;
};
let providerRuntimeHandle: ProviderRuntimePluginHandle | undefined;
const getProviderRuntimeHandle = () => {
if (providerRuntimeHandle?.plugin) {
return providerRuntimeHandle;
}
const pluginMetadataSnapshot = getCurrentAttemptPluginMetadataSnapshot();
const resolvedHandle = resolveProviderRuntimePluginHandle({
provider: params.provider,
modelId: params.modelId,
config: params.config,
workspaceDir: effectiveWorkspace,
env: process.env,
...(pluginMetadataSnapshotCoversProvider(pluginMetadataSnapshot, params.provider)
? { pluginMetadataSnapshot }
: {}),
});
if (resolvedHandle.plugin) {
providerRuntimeHandle = resolvedHandle;
}
return resolvedHandle;
};
const { sessionAgentId } = resolveSessionAgentIds({
sessionKey: params.sessionKey,
config: params.config,
agentId: params.agentId,
});
const effectiveFsWorkspaceOnly = resolveAttemptFsWorkspaceOnly({
config: params.config,
sessionAgentId,
});
prepStages.mark("workspace-sandbox");
return {
agentCoreThinkingLevel,
effectiveCwd,
effectiveFsWorkspaceOnly,
effectiveWorkspace,
emitCorePluginToolStageSummary,
emitPrepStageSummary,
getCurrentAttemptPluginMetadataSnapshot,
getProviderRuntimeHandle,
prepStages,
proactiveSubagentOrchestration,
providerThinkingLevel,
resolvedWorkspace,
sandbox,
sandboxSessionKey,
sessionAgentId,
};
}
@@ -0,0 +1,192 @@
/**
* Selects and configures the provider transport for one embedded attempt.
*/
import { createCodexNativeWebSearchWrapper } from "../../../llm/providers/stream-wrappers/openai.js";
import type { ProviderRuntimePluginHandle } from "../../../plugins/provider-hook-runtime.js";
import { resolveProviderTextTransforms } from "../../../plugins/provider-runtime.js";
import { wrapStreamFnTextTransforms } from "../../plugin-text-transforms.js";
import { registerProviderStreamForModel } from "../../provider-stream.js";
import type { SandboxContext } from "../../sandbox/types.js";
import type { AgentSession, SettingsManager } from "../../sessions/index.js";
import {
applyExtraParamsToAgent,
resolveAgentTransportOverride,
resolveExplicitSettingsTransport,
resolveExtraParams,
resolvePreparedExtraParams,
} from "../extra-params.js";
import { log } from "../logger.js";
import { resolveCacheRetention } from "../prompt-cache-retention.js";
import {
describeEmbeddedAgentStreamStrategy,
resolveEmbeddedAgentBaseStreamFn,
resolveEmbeddedAgentStreamFn,
} from "../stream-resolution.js";
import type { ProviderThinkLevel } from "../utils.js";
import {
resolveAttemptStreamAuthProfileId,
resolveAttemptToolPolicyMessageProvider,
} from "./attempt.run-decisions.js";
import type { EmbeddedRunAttemptParams } from "./types.js";
export function prepareEmbeddedAttemptTransport(input: {
attempt: EmbeddedRunAttemptParams;
session: AgentSession;
settingsManager: SettingsManager;
providerThinkingLevel: ProviderThinkLevel | undefined;
sessionAgentId: string;
workspaceDir: string;
agentDir: string;
abortSignal: AbortSignal;
getProviderRuntimeHandle: () => ProviderRuntimePluginHandle;
sandboxSessionKey: string;
sandbox?: SandboxContext | null;
codeModeControlsEnabled: boolean;
}) {
const attempt = input.attempt;
const session = input.session;
// Rebuild each turn from the session's original stream base so prior-turn
// wrappers do not pin us to stale provider/API transport behavior.
const defaultSessionStreamFn = resolveEmbeddedAgentBaseStreamFn({
session,
});
const resolvedTransport = resolveExplicitSettingsTransport({
settingsManager: input.settingsManager,
sessionTransport: session.agent.transport,
});
const streamExtraParamsOverride = {
...attempt.streamParams,
fastMode: attempt.fastMode,
};
const preparedRuntimeExtraParams = attempt.runtimePlan?.transport.resolveExtraParams({
extraParamsOverride: streamExtraParamsOverride,
thinkingLevel: input.providerThinkingLevel,
agentId: input.sessionAgentId,
workspaceDir: input.workspaceDir,
model: attempt.model,
resolvedTransport,
});
const resolvedExtraParams = resolveExtraParams({
cfg: attempt.config,
provider: attempt.provider,
modelId: attempt.modelId,
agentId: input.sessionAgentId,
});
const effectiveExtraParams =
preparedRuntimeExtraParams ??
resolvePreparedExtraParams({
cfg: attempt.config,
provider: attempt.provider,
modelId: attempt.modelId,
extraParamsOverride: streamExtraParamsOverride,
thinkingLevel: input.providerThinkingLevel,
agentId: input.sessionAgentId,
agentDir: input.agentDir,
workspaceDir: input.workspaceDir,
resolvedExtraParams,
model: attempt.model,
resolvedTransport,
});
const providerStreamFn = registerProviderStreamForModel({
model: attempt.model,
cfg: attempt.config,
agentDir: input.agentDir,
workspaceDir: input.workspaceDir,
});
const streamStrategy = describeEmbeddedAgentStreamStrategy({
currentStreamFn: defaultSessionStreamFn,
providerStreamFn,
model: attempt.model,
resolvedApiKey: attempt.resolvedApiKey,
});
session.agent.streamFn = resolveEmbeddedAgentStreamFn({
currentStreamFn: defaultSessionStreamFn,
providerStreamFn,
sessionId: attempt.sessionId,
promptCacheKey: attempt.promptCacheKey,
signal: input.abortSignal,
model: attempt.model,
resolvedApiKey: attempt.resolvedApiKey,
authProfileId: resolveAttemptStreamAuthProfileId(attempt),
authStorage: attempt.authStorage,
});
const providerTextTransforms = resolveProviderTextTransforms({
provider: attempt.provider,
config: attempt.config,
workspaceDir: input.workspaceDir,
runtimeHandle: input.getProviderRuntimeHandle(),
});
if (providerTextTransforms?.input?.length) {
session.agent.streamFn = wrapStreamFnTextTransforms({
streamFn: session.agent.streamFn,
input: providerTextTransforms.input,
transformSystemPrompt: false,
});
}
const nativeWebSearchPolicyContext = {
sessionKey: input.sandboxSessionKey,
sandboxToolPolicy: input.sandbox?.tools,
messageProvider: resolveAttemptToolPolicyMessageProvider(attempt),
agentAccountId: attempt.agentAccountId,
groupId: attempt.groupId,
groupChannel: attempt.groupChannel,
groupSpace: attempt.groupSpace,
spawnedBy: attempt.spawnedBy,
senderId: attempt.senderId,
senderName: attempt.senderName,
senderUsername: attempt.senderUsername,
senderE164: attempt.senderE164,
};
applyExtraParamsToAgent(
session.agent,
attempt.config,
attempt.provider,
attempt.modelId,
streamExtraParamsOverride,
input.providerThinkingLevel,
input.sessionAgentId,
input.workspaceDir,
attempt.model,
input.agentDir,
resolvedTransport,
{
preparedExtraParams: effectiveExtraParams,
nativeWebSearchPolicyContext,
},
);
if (input.codeModeControlsEnabled) {
session.agent.streamFn = createCodexNativeWebSearchWrapper(session.agent.streamFn, {
config: attempt.config,
agentDir: input.agentDir,
agentId: input.sessionAgentId,
...nativeWebSearchPolicyContext,
codeModeToolSurfaceEnabled: true,
});
}
const effectivePromptCacheRetention = resolveCacheRetention(
effectiveExtraParams,
attempt.provider,
attempt.model.api,
attempt.modelId,
);
const agentTransportOverride = resolveAgentTransportOverride({
settingsManager: input.settingsManager,
effectiveExtraParams,
});
const effectiveAgentTransport = agentTransportOverride ?? session.agent.transport;
if (agentTransportOverride && session.agent.transport !== agentTransportOverride) {
const previousTransport = session.agent.transport;
log.debug(
`embedded agent transport override: ${previousTransport} -> ${agentTransportOverride} ` +
`(${attempt.provider}/${attempt.modelId})`,
);
}
return {
effectiveAgentTransport,
effectiveExtraParams,
effectivePromptCacheRetention,
providerTextTransforms,
streamStrategy,
};
}
@@ -0,0 +1,350 @@
/**
* Installs replay, tool-call, timeout, and diagnostic guards around an embedded stream.
*/
import { resolveDiagnosticModelContentCapturePolicy } from "../../../infra/diagnostic-llm-content.js";
import type { DiagnosticTraceContext } from "../../../infra/diagnostic-trace-context.js";
import { resolveToolCallArgumentsEncoding } from "../../../plugins/provider-model-compat.js";
import type { resolveProviderTextTransforms } from "../../../plugins/provider-runtime.js";
import { createAnthropicPayloadLogger } from "../../anthropic-payload-log.js";
import { createCacheTrace } from "../../cache-trace.js";
import { wrapStreamFnTextTransforms } from "../../plugin-text-transforms.js";
import type { AgentSession, SessionManager } from "../../sessions/index.js";
import { resolveAgentTimeoutMs } from "../../timeout.js";
import type { TranscriptPolicy } from "../../transcript-policy.js";
import { shouldAllowProviderOwnedThinkingReplay } from "../../transcript-policy.js";
import { log } from "../logger.js";
import { collectPromptCacheToolNames } from "../prompt-cache-observability.js";
import { repairRejectedThinkingReplayInSessionManager } from "../thinking-replay-repair.js";
import {
dropReasoningFromHistory,
dropThinkingBlocks,
wrapAnthropicStreamWithRecovery,
} from "../thinking.js";
import { wrapStreamFnWithDiagnosticModelCallEvents } from "./attempt.model-diagnostic-events.js";
import { resolveUnknownToolGuardThreshold } from "./attempt.run-decisions.js";
import type { createEmbeddedAttemptSessionLockController } from "./attempt.session-lock.js";
import { createYieldAbortedResponse } from "./attempt.sessions-yield.js";
import { wrapStreamFnHandleSensitiveStopReason } from "./attempt.stop-reason-recovery.js";
import {
shouldRepairMalformedToolCallArguments,
wrapStreamFnDecodeXaiToolCallArguments,
wrapStreamFnRepairMalformedToolCallArguments,
} from "./attempt.tool-call-argument-repair.js";
import {
sanitizeOpenAIResponsesReplayForStream,
sanitizeReplayToolCallIdsForStream,
shouldApplyReplayToolCallIdSanitizer,
wrapStreamFnPromoteStandaloneTextToolCalls,
wrapStreamFnSanitizeMalformedToolCalls,
wrapStreamFnTrimToolCallNames,
} from "./attempt.tool-call-normalization.js";
import {
resolveLlmFirstEventTimeoutMs,
resolveLlmIdleTimeoutMs,
streamWithIdleTimeout,
} from "./llm-idle-timeout.js";
import { wrapStreamFnWithMessageTransform } from "./message-transform-stream-wrapper.js";
import type { EmbeddedRunAttemptParams } from "./types.js";
type CacheTrace = ReturnType<typeof createCacheTrace>;
type AnthropicPayloadLogger = ReturnType<typeof createAnthropicPayloadLogger>;
type AttemptSessionLockController = Awaited<
ReturnType<typeof createEmbeddedAttemptSessionLockController>
>;
export function installEmbeddedAttemptStreamGuards(input: {
attempt: EmbeddedRunAttemptParams;
session: AgentSession;
sessionAgentId: string;
cacheTrace: CacheTrace;
allCustomTools: Array<{ name?: string }>;
systemPromptText: string;
transcriptPolicy: TranscriptPolicy;
sessionManager: SessionManager | undefined;
sessionLockController: AttemptSessionLockController;
isOpenAIResponsesApi: boolean;
replayAllowedToolNames: Set<string>;
liveAllowedToolNames: Set<string>;
isYieldDetected: () => boolean;
clientToolLoopDetection: ReturnType<
typeof import("../../agent-tools.js").resolveToolLoopDetectionConfig
>;
anthropicPayloadLogger: AnthropicPayloadLogger;
onRejectedThinkingReplayRepaired: () => void;
onIdleTimeout: (error: Error) => void;
effectiveAgentTransport: AgentSession["agent"]["transport"];
providerTextTransforms: ReturnType<typeof resolveProviderTextTransforms>;
abortSignal: AbortSignal;
runTrace: DiagnosticTraceContext;
}) {
const attempt = input.attempt;
const session = input.session;
const cacheObservabilityEnabled = Boolean(input.cacheTrace) || log.isEnabled("debug");
const promptCacheToolNames = collectPromptCacheToolNames(
input.allCustomTools as Array<{ name?: string }>,
);
if (input.cacheTrace) {
input.cacheTrace.recordStage("session:loaded", {
messages: session.messages,
system: input.systemPromptText,
note: "after session create",
});
session.agent.streamFn = input.cacheTrace.wrapStreamFn(session.agent.streamFn);
}
// Anthropic Claude endpoints can reject replayed `thinking` blocks on
// any follow-up provider call, including tool continuations. Sanitize
// outbound messages where policy allows rewriting; otherwise preserve
// latest thinking and let the recovery wrapper retry once without it.
if (
input.transcriptPolicy.dropThinkingBlocks ||
input.transcriptPolicy.dropReasoningFromHistory
) {
session.agent.streamFn = wrapStreamFnWithMessageTransform(
session.agent.streamFn,
(messages) => {
const reasoningSanitized = input.transcriptPolicy.dropReasoningFromHistory
? dropReasoningFromHistory(messages)
: messages;
return input.transcriptPolicy.dropThinkingBlocks
? dropThinkingBlocks(reasoningSanitized)
: reasoningSanitized;
},
);
}
if (
input.transcriptPolicy.preserveSignatures ||
input.transcriptPolicy.dropThinkingBlocks ||
input.transcriptPolicy.dropReasoningFromHistory
) {
session.agent.streamFn = wrapAnthropicStreamWithRecovery(session.agent.streamFn, {
id: session.sessionId,
onRecoveredAnthropicThinking: () => {
if (!input.sessionManager) {
log.warn(
`[session-recovery] unable to repair rejected thinking replay: session manager unavailable sessionId=${session.sessionId}`,
);
return;
}
const repair = repairRejectedThinkingReplayInSessionManager({
sessionManager: input.sessionManager,
sessionFile: attempt.sessionFile,
sessionId: attempt.sessionId,
sessionKey: attempt.sessionKey,
agentId: input.sessionAgentId,
});
if (repair.repaired) {
input.onRejectedThinkingReplayRepaired();
input.sessionLockController.refreshAfterOwnedSessionWrite();
return;
}
log.warn(
`[session-recovery] rejected thinking replay retry succeeded but transcript repair made no changes: ` +
`sessionId=${session.sessionId} reason=${repair.reason ?? "unknown"}`,
);
},
});
}
// Mistral (and other strict providers) reject tool call IDs that don't match their
// format requirements (e.g. [a-zA-Z0-9]{9}). sanitizeSessionHistory only processes
// historical messages at attempt start, but the agent loop's internal tool call →
// tool result cycles bypass that path. Wrap streamFn so every outbound request
// sees sanitized tool call IDs.
const replayToolCallIdSanitizerDecision = {
sanitizeToolCallIds: input.transcriptPolicy.sanitizeToolCallIds,
toolCallIdMode: input.transcriptPolicy.toolCallIdMode,
isOpenAIResponsesApi: input.isOpenAIResponsesApi,
};
if (shouldApplyReplayToolCallIdSanitizer(replayToolCallIdSanitizerDecision)) {
const mode = replayToolCallIdSanitizerDecision.toolCallIdMode;
session.agent.streamFn = wrapStreamFnWithMessageTransform(
session.agent.streamFn,
(messages, model) =>
sanitizeReplayToolCallIdsForStream({
messages,
mode,
allowedToolNames: input.replayAllowedToolNames,
preserveNativeAnthropicToolUseIds:
input.transcriptPolicy.preserveNativeAnthropicToolUseIds,
duplicateToolCallIdStyle: input.transcriptPolicy.duplicateToolCallIdStyle,
preserveReplaySafeThinkingToolCallIds: shouldAllowProviderOwnedThinkingReplay({
modelApi: (model as { api?: unknown })?.api as string | null | undefined,
provider: attempt.provider,
policy: input.transcriptPolicy,
}),
repairToolUseResultPairing: input.transcriptPolicy.repairToolUseResultPairing,
}),
);
}
if (input.isOpenAIResponsesApi) {
session.agent.streamFn = wrapStreamFnWithMessageTransform(session.agent.streamFn, (messages) =>
sanitizeOpenAIResponsesReplayForStream(messages),
);
}
const innerStreamFn = session.agent.streamFn;
session.agent.streamFn = (model, context, options) => {
const signal = input.abortSignal as AbortSignal & { reason?: unknown };
if (input.isYieldDetected() && signal.aborted && signal.reason === "sessions_yield") {
return createYieldAbortedResponse(model) as unknown as Awaited<
ReturnType<typeof innerStreamFn>
>;
}
return innerStreamFn(model, context, options);
};
// Some models emit tool names with surrounding whitespace (e.g. " read ").
// agent runtime dispatches tool calls with exact string matching, so normalize
// names on the live response stream before tool execution.
session.agent.streamFn = wrapStreamFnSanitizeMalformedToolCalls(
session.agent.streamFn,
input.replayAllowedToolNames,
input.transcriptPolicy,
attempt.provider,
);
session.agent.streamFn = wrapStreamFnPromoteStandaloneTextToolCalls(
session.agent.streamFn,
input.liveAllowedToolNames,
);
session.agent.streamFn = wrapStreamFnTrimToolCallNames(
session.agent.streamFn,
input.liveAllowedToolNames,
{
unknownToolThreshold: resolveUnknownToolGuardThreshold(input.clientToolLoopDetection),
},
);
if (
shouldRepairMalformedToolCallArguments({
provider: attempt.provider,
modelApi: attempt.model.api,
})
) {
session.agent.streamFn = wrapStreamFnRepairMalformedToolCallArguments(session.agent.streamFn);
}
if (resolveToolCallArgumentsEncoding(attempt.model) === "html-entities") {
session.agent.streamFn = wrapStreamFnDecodeXaiToolCallArguments(session.agent.streamFn);
}
// Tool-call repair can replace structured arguments from fragmented deltas.
// Restore provider-masked text afterward so executable args stay canonical.
if (input.providerTextTransforms?.output?.length) {
session.agent.streamFn = wrapStreamFnTextTransforms({
streamFn: session.agent.streamFn,
output: input.providerTextTransforms.output,
});
}
if (input.anthropicPayloadLogger) {
session.agent.streamFn = input.anthropicPayloadLogger.wrapStreamFn(session.agent.streamFn);
}
// Anthropic-compatible providers can add new stop reasons before shared model runtime maps them.
// Recover the known "sensitive" stop reason here so a model refusal does not
// bubble out as an uncaught runner error and stall channel polling.
session.agent.streamFn = wrapStreamFnHandleSensitiveStopReason(session.agent.streamFn);
// Wrap stream with idle timeout detection.
//
// Prefer the caller's explicit `runTimeoutOverrideMs` when provided —
// it carries the "this run was launched with a deliberate per-run
// timeout" signal without losing it when the value numerically equals
// `agents.defaults.timeoutSeconds`. Fall back to the value-equality
// heuristic for callers that haven't been migrated to plumb the flag.
const configuredRunTimeoutMs = resolveAgentTimeoutMs({
cfg: attempt.config,
});
const resolvedRunTimeoutMs =
attempt.runTimeoutOverrideMs ??
(attempt.timeoutMs !== configuredRunTimeoutMs ? attempt.timeoutMs : undefined);
const idleTimeoutMs = resolveLlmIdleTimeoutMs({
cfg: attempt.config,
trigger: attempt.trigger,
runTimeoutMs: resolvedRunTimeoutMs,
modelRequestTimeoutMs: (attempt.model as { requestTimeoutMs?: number }).requestTimeoutMs,
model: {
baseUrl: attempt.model.baseUrl,
id: attempt.modelId,
provider: attempt.provider,
},
});
const firstEventTimeoutMs = resolveLlmFirstEventTimeoutMs({
cfg: attempt.config,
runTimeoutMs: resolvedRunTimeoutMs,
modelRequestTimeoutMs: (attempt.model as { requestTimeoutMs?: number }).requestTimeoutMs,
model: {
baseUrl: attempt.model.baseUrl,
id: attempt.modelId,
provider: attempt.provider,
},
});
if (idleTimeoutMs > 0) {
session.agent.streamFn = streamWithIdleTimeout(
session.agent.streamFn,
idleTimeoutMs,
(error) => input.onIdleTimeout(error),
{ runId: attempt.runId },
);
} else if (firstEventTimeoutMs > 0) {
// Local providers opt out of gap policing, but the transport first-event
// guard only arms after stream creation. A request whose headers never
// arrive would otherwise wedge until the run budget with no watchdog.
session.agent.streamFn = streamWithIdleTimeout(
session.agent.streamFn,
firstEventTimeoutMs,
(error) => input.onIdleTimeout(error),
{ runId: attempt.runId, scope: "creation-only" },
);
}
if (firstEventTimeoutMs > 0) {
const baseStreamFn = session.agent.streamFn;
session.agent.streamFn = (model, context, options) => {
type FirstEventStreamOptions = {
firstEventTimeoutMs?: number;
onFirstEventTimeout?: (error: Error) => void;
};
const optionsWithFirstEvent = options as FirstEventStreamOptions | undefined;
return baseStreamFn(model, context, {
...options,
firstEventTimeoutMs: optionsWithFirstEvent?.firstEventTimeoutMs ?? firstEventTimeoutMs,
onFirstEventTimeout: optionsWithFirstEvent?.onFirstEventTimeout ?? input.onIdleTimeout,
} as typeof options);
};
}
let diagnosticModelCallSeq = 0;
session.agent.streamFn = wrapStreamFnWithDiagnosticModelCallEvents(session.agent.streamFn, {
runId: attempt.runId,
...(attempt.sessionKey && { sessionKey: attempt.sessionKey }),
...(attempt.sessionId && { sessionId: attempt.sessionId }),
provider: attempt.provider,
model: attempt.modelId,
api: attempt.model.api,
transport: input.effectiveAgentTransport,
...(attempt.contextWindowInfo?.tokens
? { contextTokenBudget: attempt.contextWindowInfo.tokens }
: {}),
...(attempt.contextWindowInfo?.source
? { contextWindowSource: attempt.contextWindowInfo.source }
: {}),
...(attempt.contextWindowInfo?.referenceTokens
? { contextWindowReferenceTokens: attempt.contextWindowInfo.referenceTokens }
: {}),
trace: input.runTrace,
contentCapture: resolveDiagnosticModelContentCapturePolicy(attempt.config),
nextCallId: () => `${attempt.runId}:model:${(diagnosticModelCallSeq += 1)}`,
onStarted: () => {
attempt.onExecutionPhase?.({
phase: "model_call_started",
provider: attempt.provider,
model: attempt.modelId,
firstModelCallStarted: true,
});
},
});
return {
cacheObservabilityEnabled,
promptCacheToolNames,
};
}
+74 -722
View File
@@ -5,7 +5,6 @@ import fs from "node:fs/promises";
import os from "node:os";
import { ensureSystemPromptCacheBoundary } from "@openclaw/ai/internal/shared";
import { MAX_IMAGE_BYTES } from "@openclaw/media-core/constants";
import { normalizeProviderId } from "@openclaw/model-catalog-core/provider-id";
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
import { isAcpRuntimeSpawnAvailable } from "../../../acp/runtime/availability.js";
import { buildHierarchyReinforcementMessage } from "../../../auto-reply/handoff-summarizer.js";
@@ -45,7 +44,6 @@ import {
emitTrustedDiagnosticEvent,
emitTrustedDiagnosticEventWithPrivateData,
} from "../../../infra/diagnostic-events.js";
import { resolveDiagnosticModelContentCapturePolicy } from "../../../infra/diagnostic-llm-content.js";
import {
createChildDiagnosticTraceContext,
createDiagnosticTraceContext,
@@ -57,28 +55,17 @@ import { formatErrorMessage, toErrorObject } from "../../../infra/errors.js";
import { resolveHeartbeatSummaryForAgent } from "../../../infra/heartbeat-summary.js";
import { getMachineDisplayName } from "../../../infra/machine-name.js";
import { resolveRuntimeOsLabel } from "../../../infra/os-summary.js";
import { createCodexNativeWebSearchWrapper } from "../../../llm/providers/stream-wrappers/openai.js";
import type { AssistantMessage, UserMessage } from "../../../llm/types.js";
import { listRegisteredPluginAgentPromptGuidance } from "../../../plugins/command-registry-state.js";
import { getCurrentPluginMetadataSnapshot } from "../../../plugins/current-plugin-metadata-snapshot.js";
import {
buildAgentHookContextChannelFields,
buildAgentHookContextIdentityFields,
} from "../../../plugins/hook-agent-context.js";
import { resolveBlockMessage } from "../../../plugins/hook-decision-types.js";
import { getGlobalHookRunner } from "../../../plugins/hook-runner-global.js";
import type { PluginMetadataSnapshot } from "../../../plugins/plugin-metadata-snapshot.types.js";
import {
resolveProviderRuntimePluginHandle,
type ProviderRuntimePluginHandle,
} from "../../../plugins/provider-hook-runtime.js";
import {
extractModelCompat,
resolveToolCallArgumentsEncoding,
} from "../../../plugins/provider-model-compat.js";
import { extractModelCompat } from "../../../plugins/provider-model-compat.js";
import {
resolveProviderSystemPromptContribution,
resolveProviderTextTransforms,
transformProviderSystemPrompt,
} from "../../../plugins/provider-runtime.js";
import { copyPluginToolMeta, getPluginToolMeta } from "../../../plugins/tools.js";
@@ -91,15 +78,11 @@ import {
applySkillEnvOverrides,
applySkillEnvOverridesFromSnapshot,
} from "../../../skills/runtime/env-overrides.js";
import {
buildTrajectoryArtifacts,
buildTrajectoryRunMetadata,
} from "../../../trajectory/metadata.js";
import { buildTrajectoryRunMetadata } from "../../../trajectory/metadata.js";
import {
createTrajectoryRuntimeRecorder,
toTrajectoryToolDefinitions,
} from "../../../trajectory/runtime.js";
import { resolveUserPath } from "../../../utils.js";
import { normalizeMessageChannel } from "../../../utils/message-channel.js";
import { isReasoningTagProvider } from "../../../utils/provider-utils.js";
import { createBundleLspToolRuntime } from "../../agent-bundle-lsp-runtime.js";
@@ -201,10 +184,8 @@ import {
import { resolveModelAuthMode } from "../../model-auth.js";
import { resolveDefaultModelForAgent } from "../../model-selection.js";
import { supportsModelTools } from "../../model-tool-support.js";
import { wrapStreamFnTextTransforms } from "../../plugin-text-transforms.js";
import { resolveAgentPromptSurfaceForSessionKey } from "../../prompt-surface.js";
import { describeProviderRequestRoutingSummary } from "../../provider-attribution.js";
import { registerProviderStreamForModel } from "../../provider-stream.js";
import {
AGENT_RUN_RESTART_ABORT_STOP_REASON,
createAgentRunRestartAbortError,
@@ -216,7 +197,6 @@ import {
normalizeAgentRuntimeTools,
} from "../../runtime-plan/tools.js";
import type { AgentMessage } from "../../runtime/index.js";
import { resolveSandboxContext } from "../../sandbox.js";
import { resolveSandboxRuntimeStatus } from "../../sandbox/runtime-status.js";
import {
invalidateSessionFileRepairCache,
@@ -246,7 +226,6 @@ import {
appendModelIdentitySystemPrompt,
buildModelIdentityPromptLine,
} from "../../system-prompt.js";
import { resolveAgentTimeoutMs } from "../../timeout.js";
import {
buildEmptyExplicitToolAllowlistError,
collectExplicitToolAllowlistSources,
@@ -285,7 +264,6 @@ import {
replaceWithEffectiveCronCreatorToolAllowlist,
type CronCreatorToolAllowlistEntry,
} from "../../tools/cron-tool.js";
import { shouldAllowProviderOwnedThinkingReplay } from "../../transcript-policy.js";
import { normalizeUsage, type NormalizedUsage } from "../../usage.js";
import {
DEFAULT_BOOTSTRAP_FILENAME,
@@ -302,25 +280,16 @@ import {
import { runContextEngineMaintenance } from "../context-engine-maintenance.js";
import { applyFinalEffectiveToolPolicy } from "../effective-tool-policy.js";
import { buildEmbeddedExtensionFactories } from "../extensions.js";
import {
applyExtraParamsToAgent,
resolveAgentTransportOverride,
resolveExplicitSettingsTransport,
resolveExtraParams,
resolvePreparedExtraParams,
} from "../extra-params.js";
import { prepareGooglePromptCacheStreamFn } from "../google-prompt-cache.js";
import { getHistoryLimitFromSessionKey, limitHistoryTurns } from "../history.js";
import { log } from "../logger.js";
import { buildEmbeddedMessageActionDiscoveryInput } from "../message-action-discovery-input.js";
import {
collectPromptCacheToolNames,
beginPromptCacheObservation,
completePromptCacheObservation,
type PromptCacheBreak,
type PromptCacheChange,
} from "../prompt-cache-observability.js";
import { resolveCacheRetention } from "../prompt-cache-retention.js";
import {
normalizeAssistantReplayContent,
sanitizeSessionHistory,
@@ -351,19 +320,12 @@ import {
markSessionUserTurnsSent,
} from "../session-prompt-state.js";
import {
describeEmbeddedAgentStreamStrategy,
resetEmbeddedAgentBaseStreamFnCacheForTest,
resolveEmbeddedAgentApiKey,
resolveEmbeddedAgentBaseStreamFn,
resolveEmbeddedAgentStreamFn,
} from "../stream-resolution.js";
import { applySystemPromptToSession } from "../system-prompt.js";
import { repairRejectedThinkingReplayInSessionManager } from "../thinking-replay-repair.js";
import {
dropReasoningFromHistory,
dropThinkingBlocks,
wrapAnthropicStreamWithRecovery,
} from "../thinking.js";
import {
collectCoreBuiltinToolNames,
collectRegisteredToolNames,
@@ -381,17 +343,15 @@ import {
truncateOversizedToolResultsInSessionManager,
} from "../tool-result-truncation.js";
import { splitSdkTools } from "../tool-split.js";
import { mapThinkingLevel, mapThinkingLevelForProvider } from "../utils.js";
import { flushPendingToolResultsAfterIdle } from "../wait-for-idle-before-flush.js";
import { abortable as abortableWithSignal } from "./abortable.js";
import { releaseEmbeddedAttemptSessionLockForAbort } from "./attempt-abort.js";
import { configureEmbeddedAttemptHttpRuntime } from "./attempt-http-runtime.js";
import { finalizeEmbeddedAttempt } from "./attempt-finalize.js";
import { createEmbeddedAgentSessionWithResourceLoader } from "./attempt-session.js";
import {
createEmbeddedRunStageTracker,
formatEmbeddedRunStageSummary,
shouldWarnEmbeddedRunStageSummary,
} from "./attempt-stage-timing.js";
import { prepareEmbeddedAttemptSetup } from "./attempt-setup.js";
import { createEmbeddedRunStageTracker } from "./attempt-stage-timing.js";
import { prepareEmbeddedAttemptTransport } from "./attempt-stream-transport.js";
import { installEmbeddedAttemptStreamGuards } from "./attempt-stream.js";
import { buildAttemptSystemPrompt } from "./attempt-system-prompt.js";
import {
applyEmbeddedAttemptToolsAllow,
@@ -401,10 +361,6 @@ import {
shouldCreateBundleMcpRuntimeForAttempt,
} from "./attempt-tool-construction-plan.js";
import { flushEmbeddedAttemptTrajectoryRecorder } from "./attempt-trajectory-flush-cleanup.js";
import {
resolveAttemptTrajectoryTerminal,
resolveTerminalAssistantTexts,
} from "./attempt-trajectory-status.js";
import {
requiresCompletionRequiredAsyncTaskWait,
shouldWaitForCompletionRequiredAsyncTasks,
@@ -430,12 +386,10 @@ import {
normalizeMessagesForCurrentPromptBoundary,
normalizeMessagesForLlmBoundary,
} from "./attempt.llm-boundary.js";
import { wrapStreamFnWithDiagnosticModelCallEvents } from "./attempt.model-diagnostic-events.js";
import {
buildAfterTurnRuntimeContext,
buildAfterTurnRuntimeContextFromUsage,
prependSystemPromptAddition,
resolveAttemptFsWorkspaceOnly,
resolveAttemptMediaTaskSystemPromptAddition,
resolvePromptBuildHookResult,
resolvePromptModeForSession,
@@ -445,10 +399,8 @@ import {
} from "./attempt.prompt-helpers.js";
import { steerActiveSessionWithOptionalDeliveryWait } from "./attempt.queue-message.js";
import {
resolveAttemptStreamAuthProfileId,
resolveAttemptToolPolicyMessageProvider,
resolveEmbeddedAttemptSessionWriteLockOptions,
resolveUnknownToolGuardThreshold,
shouldRunLlmOutputHooksForAttempt,
} from "./attempt.run-decisions.js";
import {
@@ -459,13 +411,11 @@ import {
installPromptSubmissionLockRelease,
} from "./attempt.session-lock.js";
import {
createYieldAbortedResponse,
persistSessionsYieldContextMessage,
queueSessionsYieldInterruptMessage,
stripSessionsYieldArtifacts,
waitForSessionsYieldAbortSettle,
} from "./attempt.sessions-yield.js";
import { wrapStreamFnHandleSensitiveStopReason } from "./attempt.stop-reason-recovery.js";
import {
buildEmbeddedSubscriptionParams,
cleanupEmbeddedAttemptResources,
@@ -476,19 +426,6 @@ import {
resolveAttemptSpawnWorkspaceDir,
shouldPersistCompletedBootstrapTurn,
} from "./attempt.thread-helpers.js";
import {
shouldRepairMalformedToolCallArguments,
wrapStreamFnDecodeXaiToolCallArguments,
wrapStreamFnRepairMalformedToolCallArguments,
} from "./attempt.tool-call-argument-repair.js";
import {
sanitizeOpenAIResponsesReplayForStream,
sanitizeReplayToolCallIdsForStream,
shouldApplyReplayToolCallIdSanitizer,
wrapStreamFnPromoteStandaloneTextToolCalls,
wrapStreamFnSanitizeMalformedToolCalls,
wrapStreamFnTrimToolCallNames,
} from "./attempt.tool-call-normalization.js";
import { buildEmbeddedAttemptToolRunContext } from "./attempt.tool-run-context.js";
import {
buildToolSearchRunPlan,
@@ -522,11 +459,6 @@ import {
resolveSilentToolResultReplyPayload,
shouldTreatEmptyAssistantReplyAsSilent,
} from "./incomplete-turn.js";
import {
resolveLlmFirstEventTimeoutMs,
resolveLlmIdleTimeoutMs,
streamWithIdleTimeout,
} from "./llm-idle-timeout.js";
import {
resolveMessageMergeStrategy,
type MessageMergeStrategy,
@@ -595,31 +527,6 @@ export {
const MAX_BTW_SNAPSHOT_MESSAGES = 100;
const aggregateToolResultPressureWarnings = new Set<string>();
function pluginMetadataSnapshotCoversProvider(
snapshot: PluginMetadataSnapshot | undefined,
provider: string,
): snapshot is PluginMetadataSnapshot {
const normalizedProvider = normalizeProviderId(provider);
if (!snapshot || !normalizedProvider) {
return false;
}
return snapshot.manifestRegistry.plugins.some((plugin) => {
const ownsProvider = plugin.providers.some(
(providerId) => normalizeProviderId(providerId) === normalizedProvider,
);
if (ownsProvider) {
return true;
}
const modelCatalogProviderIds = [
...Object.keys(plugin.modelCatalog?.providers ?? {}),
...Object.keys(plugin.modelCatalog?.aliases ?? {}),
];
return modelCatalogProviderIds.some(
(providerId) => normalizeProviderId(providerId) === normalizedProvider,
);
});
}
function summarizeMessagePayload(msg: AgentMessage): { textChars: number; imageBlocks: number } {
const content = (msg as { content?: unknown }).content;
if (typeof content === "string") {
@@ -1096,128 +1003,24 @@ async function resolveExistingAttemptTranscriptState(params: {
export async function runEmbeddedAttempt(
params: EmbeddedRunAttemptParams,
): Promise<EmbeddedRunAttemptResult> {
const resolvedWorkspace = resolveUserPath(params.workspaceDir);
const runAbortController = new AbortController();
// Ultra is a logical orchestration mode, not a provider effort. Preserve it for
// prompt/status surfaces, then lower only at agent-core and provider boundaries.
const agentCoreThinkingLevel = mapThinkingLevel(params.thinkLevel);
const providerThinkingLevel = mapThinkingLevelForProvider(params.thinkLevel);
const proactiveSubagentOrchestration = params.thinkLevel === "ultra";
configureEmbeddedAttemptHttpRuntime({ timeoutMs: params.timeoutMs });
log.debug(
`embedded run start: runId=${params.runId} sessionId=${params.sessionId} provider=${params.provider} model=${params.modelId} thinking=${params.thinkLevel} messageChannel=${params.messageChannel ?? params.messageProvider ?? "unknown"}`,
);
const prepStages = createEmbeddedRunStageTracker();
const emitPrepStageSummary = (phase: string) => {
const summary = prepStages.snapshot();
const shouldWarn = shouldWarnEmbeddedRunStageSummary(summary);
if (!shouldWarn && !log.isEnabled("trace")) {
return;
}
const message = formatEmbeddedRunStageSummary(
`[trace:embedded-run] prep stages: runId=${params.runId} sessionId=${params.sessionId} phase=${phase}`,
summary,
);
if (shouldWarn) {
log.warn(message);
} else {
log.trace(message);
}
};
const emitCorePluginToolStageSummary = (
phase: string,
summary: ReturnType<typeof prepStages.snapshot>,
) => {
if (summary.stages.length === 0) {
return;
}
const shouldWarn = shouldWarnEmbeddedRunStageSummary(summary, {
totalThresholdMs: 5_000,
stageThresholdMs: 2_000,
});
if (!shouldWarn && !log.isEnabled("trace")) {
return;
}
const message = formatEmbeddedRunStageSummary(
`[trace:embedded-run] core-plugin-tool stages: runId=${params.runId} sessionId=${params.sessionId} phase=${phase}`,
summary,
);
if (shouldWarn) {
log.warn(message);
} else {
log.trace(message);
}
};
await fs.mkdir(resolvedWorkspace, { recursive: true });
const sandboxSessionKey =
params.sandboxSessionKey?.trim() || params.sessionKey?.trim() || params.sessionId;
const sandbox = await resolveSandboxContext({
config: params.config,
execOverrides: params.execOverrides,
sessionKey: sandboxSessionKey,
workspaceDir: resolvedWorkspace,
});
const effectiveWorkspace = sandbox?.enabled
? sandbox.workspaceAccess === "rw"
? resolvedWorkspace
: sandbox.workspaceDir
: resolvedWorkspace;
const requestedCwd = params.cwd ? resolveUserPath(params.cwd) : undefined;
if (sandbox?.enabled && requestedCwd && requestedCwd !== resolvedWorkspace) {
throw new Error(
"cwd override is not supported for sandboxed embedded agent runs; omit cwd or use the agent workspace as cwd",
);
}
const effectiveCwd = sandbox?.enabled ? effectiveWorkspace : (requestedCwd ?? effectiveWorkspace);
await fs.mkdir(effectiveWorkspace, { recursive: true });
let currentPluginMetadataSnapshotResolved = false;
let currentPluginMetadataSnapshot: PluginMetadataSnapshot | undefined;
const getCurrentAttemptPluginMetadataSnapshot = () => {
if (!currentPluginMetadataSnapshotResolved) {
currentPluginMetadataSnapshot = getCurrentPluginMetadataSnapshot({
allowScopedSnapshot: true,
config: params.config,
env: process.env,
workspaceDir: effectiveWorkspace,
});
currentPluginMetadataSnapshotResolved = true;
}
return currentPluginMetadataSnapshot;
};
let providerRuntimeHandle: ProviderRuntimePluginHandle | undefined;
const getProviderRuntimeHandle = () => {
if (providerRuntimeHandle?.plugin) {
return providerRuntimeHandle;
}
const pluginMetadataSnapshot = getCurrentAttemptPluginMetadataSnapshot();
const resolvedHandle = resolveProviderRuntimePluginHandle({
provider: params.provider,
modelId: params.modelId,
config: params.config,
workspaceDir: effectiveWorkspace,
env: process.env,
...(pluginMetadataSnapshotCoversProvider(pluginMetadataSnapshot, params.provider)
? { pluginMetadataSnapshot }
: {}),
});
if (resolvedHandle.plugin) {
providerRuntimeHandle = resolvedHandle;
}
return resolvedHandle;
};
const { sessionAgentId } = resolveSessionAgentIds({
sessionKey: params.sessionKey,
config: params.config,
agentId: params.agentId,
});
const effectiveFsWorkspaceOnly = resolveAttemptFsWorkspaceOnly({
config: params.config,
const {
agentCoreThinkingLevel,
effectiveCwd,
effectiveFsWorkspaceOnly,
effectiveWorkspace,
emitCorePluginToolStageSummary,
emitPrepStageSummary,
getCurrentAttemptPluginMetadataSnapshot,
getProviderRuntimeHandle,
prepStages,
proactiveSubagentOrchestration,
providerThinkingLevel,
resolvedWorkspace,
sandbox,
sandboxSessionKey,
sessionAgentId,
});
prepStages.mark("workspace-sandbox");
} = await prepareEmbeddedAttemptSetup(params);
let restoreSkillEnv: (() => void) | undefined;
let aborted = Boolean(params.abortSignal?.aborted);
@@ -3296,427 +3099,56 @@ export async function runEmbeddedAttempt(
}),
);
// Rebuild each turn from the session's original stream base so prior-turn
// wrappers do not pin us to stale provider/API transport behavior.
const defaultSessionStreamFn = resolveEmbeddedAgentBaseStreamFn({
const {
effectiveAgentTransport,
effectiveExtraParams,
effectivePromptCacheRetention,
providerTextTransforms,
streamStrategy,
} = prepareEmbeddedAttemptTransport({
attempt: params,
session: activeSession,
});
const resolvedTransport = resolveExplicitSettingsTransport({
settingsManager,
sessionTransport: activeSession.agent.transport,
});
const streamExtraParamsOverride = {
...params.streamParams,
fastMode: params.fastMode,
};
const preparedRuntimeExtraParams = params.runtimePlan?.transport.resolveExtraParams({
extraParamsOverride: streamExtraParamsOverride,
thinkingLevel: providerThinkingLevel,
agentId: sessionAgentId,
workspaceDir: effectiveWorkspace,
model: params.model,
resolvedTransport,
});
const resolvedExtraParams = resolveExtraParams({
cfg: params.config,
provider: params.provider,
modelId: params.modelId,
agentId: sessionAgentId,
});
const effectiveExtraParams =
preparedRuntimeExtraParams ??
resolvePreparedExtraParams({
cfg: params.config,
provider: params.provider,
modelId: params.modelId,
extraParamsOverride: streamExtraParamsOverride,
thinkingLevel: providerThinkingLevel,
agentId: sessionAgentId,
agentDir,
workspaceDir: effectiveWorkspace,
resolvedExtraParams,
model: params.model,
resolvedTransport,
});
const providerStreamFn = registerProviderStreamForModel({
model: params.model,
cfg: params.config,
agentDir,
workspaceDir: effectiveWorkspace,
});
const streamStrategy = describeEmbeddedAgentStreamStrategy({
currentStreamFn: defaultSessionStreamFn,
providerStreamFn,
model: params.model,
resolvedApiKey: params.resolvedApiKey,
});
activeSession.agent.streamFn = resolveEmbeddedAgentStreamFn({
currentStreamFn: defaultSessionStreamFn,
providerStreamFn,
sessionId: params.sessionId,
promptCacheKey: params.promptCacheKey,
signal: runAbortController.signal,
model: params.model,
resolvedApiKey: params.resolvedApiKey,
authProfileId: resolveAttemptStreamAuthProfileId(params),
authStorage: params.authStorage,
});
const providerTextTransforms = resolveProviderTextTransforms({
provider: params.provider,
config: params.config,
workspaceDir: effectiveWorkspace,
runtimeHandle: getProviderRuntimeHandle(),
});
if (providerTextTransforms?.input?.length) {
activeSession.agent.streamFn = wrapStreamFnTextTransforms({
streamFn: activeSession.agent.streamFn,
input: providerTextTransforms.input,
transformSystemPrompt: false,
});
}
const nativeWebSearchPolicyContext = {
sessionKey: sandboxSessionKey,
sandboxToolPolicy: sandbox?.tools,
messageProvider: resolveAttemptToolPolicyMessageProvider(params),
agentAccountId: params.agentAccountId,
groupId: params.groupId,
groupChannel: params.groupChannel,
groupSpace: params.groupSpace,
spawnedBy: params.spawnedBy,
senderId: params.senderId,
senderName: params.senderName,
senderUsername: params.senderUsername,
senderE164: params.senderE164,
};
applyExtraParamsToAgent(
activeSession.agent,
params.config,
params.provider,
params.modelId,
streamExtraParamsOverride,
providerThinkingLevel,
sessionAgentId,
effectiveWorkspace,
params.model,
workspaceDir: effectiveWorkspace,
agentDir,
resolvedTransport,
{
preparedExtraParams: effectiveExtraParams,
nativeWebSearchPolicyContext,
},
);
if (codeModeControlsEnabledForRun) {
activeSession.agent.streamFn = createCodexNativeWebSearchWrapper(
activeSession.agent.streamFn,
{
config: params.config,
agentDir,
agentId: sessionAgentId,
...nativeWebSearchPolicyContext,
codeModeToolSurfaceEnabled: true,
},
);
}
const effectivePromptCacheRetention = resolveCacheRetention(
effectiveExtraParams,
params.provider,
params.model.api,
params.modelId,
);
const agentTransportOverride = resolveAgentTransportOverride({
settingsManager,
effectiveExtraParams,
abortSignal: runAbortController.signal,
getProviderRuntimeHandle,
sandboxSessionKey,
sandbox,
codeModeControlsEnabled: codeModeControlsEnabledForRun,
});
const effectiveAgentTransport = agentTransportOverride ?? activeSession.agent.transport;
if (agentTransportOverride && activeSession.agent.transport !== agentTransportOverride) {
const previousTransport = activeSession.agent.transport;
log.debug(
`embedded agent transport override: ${previousTransport} -> ${agentTransportOverride} ` +
`(${params.provider}/${params.modelId})`,
);
}
const { cacheObservabilityEnabled, promptCacheToolNames } =
installEmbeddedAttemptStreamGuards({
attempt: params,
session: activeSession,
sessionAgentId,
cacheTrace,
allCustomTools,
systemPromptText,
transcriptPolicy,
sessionManager,
sessionLockController,
isOpenAIResponsesApi,
replayAllowedToolNames,
liveAllowedToolNames,
isYieldDetected: () => yieldDetected,
clientToolLoopDetection,
anthropicPayloadLogger,
onRejectedThinkingReplayRepaired: () => {
repairedRejectedThinkingReplay = true;
},
onIdleTimeout: (error) => idleTimeoutTrigger?.(error),
effectiveAgentTransport,
providerTextTransforms,
abortSignal: runAbortController.signal,
runTrace,
});
prepStages.mark("stream-setup");
emitPrepStageSummary("stream-ready");
const cacheObservabilityEnabled = Boolean(cacheTrace) || log.isEnabled("debug");
const promptCacheToolNames = collectPromptCacheToolNames(
allCustomTools as Array<{ name?: string }>,
);
let promptCacheChangesForTurn: PromptCacheChange[] | null = null;
if (cacheTrace) {
cacheTrace.recordStage("session:loaded", {
messages: activeSession.messages,
system: systemPromptText,
note: "after session create",
});
activeSession.agent.streamFn = cacheTrace.wrapStreamFn(activeSession.agent.streamFn);
}
// Anthropic Claude endpoints can reject replayed `thinking` blocks on
// any follow-up provider call, including tool continuations. Sanitize
// outbound messages where policy allows rewriting; otherwise preserve
// latest thinking and let the recovery wrapper retry once without it.
if (transcriptPolicy.dropThinkingBlocks || transcriptPolicy.dropReasoningFromHistory) {
activeSession.agent.streamFn = wrapStreamFnWithMessageTransform(
activeSession.agent.streamFn,
(messages) => {
const reasoningSanitized = transcriptPolicy.dropReasoningFromHistory
? dropReasoningFromHistory(messages)
: messages;
return transcriptPolicy.dropThinkingBlocks
? dropThinkingBlocks(reasoningSanitized)
: reasoningSanitized;
},
);
}
if (
transcriptPolicy.preserveSignatures ||
transcriptPolicy.dropThinkingBlocks ||
transcriptPolicy.dropReasoningFromHistory
) {
activeSession.agent.streamFn = wrapAnthropicStreamWithRecovery(
activeSession.agent.streamFn,
{
id: activeSession.sessionId,
onRecoveredAnthropicThinking: () => {
if (!sessionManager) {
log.warn(
`[session-recovery] unable to repair rejected thinking replay: session manager unavailable sessionId=${activeSession.sessionId}`,
);
return;
}
const repair = repairRejectedThinkingReplayInSessionManager({
sessionManager,
sessionFile: params.sessionFile,
sessionId: params.sessionId,
sessionKey: params.sessionKey,
agentId: sessionAgentId,
});
if (repair.repaired) {
repairedRejectedThinkingReplay = true;
sessionLockController.refreshAfterOwnedSessionWrite();
return;
}
log.warn(
`[session-recovery] rejected thinking replay retry succeeded but transcript repair made no changes: ` +
`sessionId=${activeSession.sessionId} reason=${repair.reason ?? "unknown"}`,
);
},
},
);
}
// Mistral (and other strict providers) reject tool call IDs that don't match their
// format requirements (e.g. [a-zA-Z0-9]{9}). sanitizeSessionHistory only processes
// historical messages at attempt start, but the agent loop's internal tool call →
// tool result cycles bypass that path. Wrap streamFn so every outbound request
// sees sanitized tool call IDs.
const replayToolCallIdSanitizerDecision = {
sanitizeToolCallIds: transcriptPolicy.sanitizeToolCallIds,
toolCallIdMode: transcriptPolicy.toolCallIdMode,
isOpenAIResponsesApi,
};
if (shouldApplyReplayToolCallIdSanitizer(replayToolCallIdSanitizerDecision)) {
const mode = replayToolCallIdSanitizerDecision.toolCallIdMode;
activeSession.agent.streamFn = wrapStreamFnWithMessageTransform(
activeSession.agent.streamFn,
(messages, model) =>
sanitizeReplayToolCallIdsForStream({
messages,
mode,
allowedToolNames: replayAllowedToolNames,
preserveNativeAnthropicToolUseIds: transcriptPolicy.preserveNativeAnthropicToolUseIds,
duplicateToolCallIdStyle: transcriptPolicy.duplicateToolCallIdStyle,
preserveReplaySafeThinkingToolCallIds: shouldAllowProviderOwnedThinkingReplay({
modelApi: (model as { api?: unknown })?.api as string | null | undefined,
provider: params.provider,
policy: transcriptPolicy,
}),
repairToolUseResultPairing: transcriptPolicy.repairToolUseResultPairing,
}),
);
}
if (isOpenAIResponsesApi) {
activeSession.agent.streamFn = wrapStreamFnWithMessageTransform(
activeSession.agent.streamFn,
(messages) => sanitizeOpenAIResponsesReplayForStream(messages),
);
}
const innerStreamFn = activeSession.agent.streamFn;
activeSession.agent.streamFn = (model, context, options) => {
const signal = runAbortController.signal as AbortSignal & { reason?: unknown };
if (yieldDetected && signal.aborted && signal.reason === "sessions_yield") {
return createYieldAbortedResponse(model) as unknown as Awaited<
ReturnType<typeof innerStreamFn>
>;
}
return innerStreamFn(model, context, options);
};
// Some models emit tool names with surrounding whitespace (e.g. " read ").
// agent runtime dispatches tool calls with exact string matching, so normalize
// names on the live response stream before tool execution.
activeSession.agent.streamFn = wrapStreamFnSanitizeMalformedToolCalls(
activeSession.agent.streamFn,
replayAllowedToolNames,
transcriptPolicy,
params.provider,
);
activeSession.agent.streamFn = wrapStreamFnPromoteStandaloneTextToolCalls(
activeSession.agent.streamFn,
liveAllowedToolNames,
);
activeSession.agent.streamFn = wrapStreamFnTrimToolCallNames(
activeSession.agent.streamFn,
liveAllowedToolNames,
{
unknownToolThreshold: resolveUnknownToolGuardThreshold(clientToolLoopDetection),
},
);
if (
shouldRepairMalformedToolCallArguments({
provider: params.provider,
modelApi: params.model.api,
})
) {
activeSession.agent.streamFn = wrapStreamFnRepairMalformedToolCallArguments(
activeSession.agent.streamFn,
);
}
if (resolveToolCallArgumentsEncoding(params.model) === "html-entities") {
activeSession.agent.streamFn = wrapStreamFnDecodeXaiToolCallArguments(
activeSession.agent.streamFn,
);
}
// Tool-call repair can replace structured arguments from fragmented deltas.
// Restore provider-masked text afterward so executable args stay canonical.
if (providerTextTransforms?.output?.length) {
activeSession.agent.streamFn = wrapStreamFnTextTransforms({
streamFn: activeSession.agent.streamFn,
output: providerTextTransforms.output,
});
}
if (anthropicPayloadLogger) {
activeSession.agent.streamFn = anthropicPayloadLogger.wrapStreamFn(
activeSession.agent.streamFn,
);
}
// Anthropic-compatible providers can add new stop reasons before shared model runtime maps them.
// Recover the known "sensitive" stop reason here so a model refusal does not
// bubble out as an uncaught runner error and stall channel polling.
activeSession.agent.streamFn = wrapStreamFnHandleSensitiveStopReason(
activeSession.agent.streamFn,
);
// Wrap stream with idle timeout detection.
//
// Prefer the caller's explicit `runTimeoutOverrideMs` when provided —
// it carries the "this run was launched with a deliberate per-run
// timeout" signal without losing it when the value numerically equals
// `agents.defaults.timeoutSeconds`. Fall back to the value-equality
// heuristic for callers that haven't been migrated to plumb the flag.
const configuredRunTimeoutMs = resolveAgentTimeoutMs({
cfg: params.config,
});
const resolvedRunTimeoutMs =
params.runTimeoutOverrideMs ??
(params.timeoutMs !== configuredRunTimeoutMs ? params.timeoutMs : undefined);
const idleTimeoutMs = resolveLlmIdleTimeoutMs({
cfg: params.config,
trigger: params.trigger,
runTimeoutMs: resolvedRunTimeoutMs,
modelRequestTimeoutMs: (params.model as { requestTimeoutMs?: number }).requestTimeoutMs,
model: {
baseUrl: params.model.baseUrl,
id: params.modelId,
provider: params.provider,
},
});
const firstEventTimeoutMs = resolveLlmFirstEventTimeoutMs({
cfg: params.config,
runTimeoutMs: resolvedRunTimeoutMs,
modelRequestTimeoutMs: (params.model as { requestTimeoutMs?: number }).requestTimeoutMs,
model: {
baseUrl: params.model.baseUrl,
id: params.modelId,
provider: params.provider,
},
});
if (idleTimeoutMs > 0) {
activeSession.agent.streamFn = streamWithIdleTimeout(
activeSession.agent.streamFn,
idleTimeoutMs,
(error) => idleTimeoutTrigger?.(error),
{ runId: params.runId },
);
} else if (firstEventTimeoutMs > 0) {
// Local providers opt out of gap policing, but the transport first-event
// guard only arms after stream creation. A request whose headers never
// arrive would otherwise wedge until the run budget with no watchdog.
activeSession.agent.streamFn = streamWithIdleTimeout(
activeSession.agent.streamFn,
firstEventTimeoutMs,
(error) => idleTimeoutTrigger?.(error),
{ runId: params.runId, scope: "creation-only" },
);
}
if (firstEventTimeoutMs > 0) {
const baseStreamFn = activeSession.agent.streamFn;
activeSession.agent.streamFn = (model, context, options) => {
type FirstEventStreamOptions = {
firstEventTimeoutMs?: number;
onFirstEventTimeout?: (error: Error) => void;
};
const optionsWithFirstEvent = options as FirstEventStreamOptions | undefined;
return baseStreamFn(model, context, {
...options,
firstEventTimeoutMs: optionsWithFirstEvent?.firstEventTimeoutMs ?? firstEventTimeoutMs,
onFirstEventTimeout: optionsWithFirstEvent?.onFirstEventTimeout ?? idleTimeoutTrigger,
} as typeof options);
};
}
let diagnosticModelCallSeq = 0;
activeSession.agent.streamFn = wrapStreamFnWithDiagnosticModelCallEvents(
activeSession.agent.streamFn,
{
runId: params.runId,
...(params.sessionKey && { sessionKey: params.sessionKey }),
...(params.sessionId && { sessionId: params.sessionId }),
provider: params.provider,
model: params.modelId,
api: params.model.api,
transport: effectiveAgentTransport,
...(params.contextWindowInfo?.tokens
? { contextTokenBudget: params.contextWindowInfo.tokens }
: {}),
...(params.contextWindowInfo?.source
? { contextWindowSource: params.contextWindowInfo.source }
: {}),
...(params.contextWindowInfo?.referenceTokens
? { contextWindowReferenceTokens: params.contextWindowInfo.referenceTokens }
: {}),
trace: runTrace,
contentCapture: resolveDiagnosticModelContentCapturePolicy(params.config),
nextCallId: () => `${params.runId}:model:${(diagnosticModelCallSeq += 1)}`,
onStarted: () => {
params.onExecutionPhase?.({
phase: "model_call_started",
provider: params.provider,
model: params.modelId,
firstModelCallStarted: true,
});
},
},
);
try {
if (isRawModelRun) {
activeSession.agent.reset();
@@ -6262,97 +5694,7 @@ export async function runEmbeddedAttempt(
timedOutDuringCompaction,
},
});
const terminalAssistantTexts = resolveTerminalAssistantTexts({
assistantTexts,
lastAssistantStopReason: lastAssistant?.stopReason,
lastAssistantVisibleText: resolveFinalAssistantVisibleText(lastAssistant),
});
const attemptTrajectoryTerminal = resolveAttemptTrajectoryTerminal({
promptError,
aborted,
externalAbort,
timedOut,
assistantTexts: terminalAssistantTexts,
toolMetas: toolMetasNormalized,
didSendViaMessagingTool: didSendViaMessagingTool(),
didSendDeterministicApprovalPrompt: didSendDeterministicApprovalPromptNow,
messagingToolSentTexts: getMessagingToolSentTexts(),
messagingToolSentMediaUrls: getMessagingToolSentMediaUrls(),
messagingToolSentTargets: getMessagingToolSentTargets(),
successfulCronAdds: getSuccessfulCronAdds(),
synthesizedPayloadCount,
acceptedSessionSpawns,
heartbeatToolResponse,
clientToolCalls: completedClientToolCalls,
yieldDetected,
lastToolError,
silentExpected: params.silentExpected,
emptyAssistantReplyIsSilent,
lastAssistantStopReason: lastAssistant?.stopReason,
hasTerminalOutput,
});
trajectoryRecorder?.recordEvent("model.completed", {
aborted,
externalAbort,
timedOut,
idleTimedOut,
timedOutDuringCompaction,
timedOutDuringToolExecution,
timedOutByRunBudget,
promptError: promptError ? formatErrorMessage(promptError) : undefined,
promptErrorSource,
terminalError: attemptTrajectoryTerminal.terminalError,
usage: attemptUsage,
promptCache,
compactionCount: getCompactionCount(),
assistantTexts,
finalPromptText,
messagesSnapshot,
});
trajectoryRecorder?.recordEvent(
"trace.artifacts",
buildTrajectoryArtifacts({
status: attemptTrajectoryTerminal.status,
aborted,
externalAbort,
timedOut,
idleTimedOut,
timedOutDuringCompaction,
timedOutDuringToolExecution,
timedOutByRunBudget,
promptError: promptError ? formatErrorMessage(promptError) : undefined,
promptErrorSource,
terminalError: attemptTrajectoryTerminal.terminalError,
usage: attemptUsage,
promptCache,
compactionCount: getCompactionCount(),
assistantTexts,
finalPromptText,
itemLifecycle: getItemLifecycle(),
toolMetas: toolMetasNormalized,
didSendViaMessagingTool: didSendViaMessagingTool(),
successfulCronAdds: getSuccessfulCronAdds(),
messagingToolSentTexts: getMessagingToolSentTexts(),
messagingToolSentMediaUrls: getMessagingToolSentMediaUrls(),
messagingToolSentTargets: getMessagingToolSentTargets(),
lastToolError,
}),
);
trajectoryRecorder?.recordEvent("session.ended", {
status: attemptTrajectoryTerminal.status,
aborted,
externalAbort,
timedOut,
idleTimedOut,
timedOutDuringCompaction,
timedOutDuringToolExecution,
timedOutByRunBudget,
promptError: promptError ? formatErrorMessage(promptError) : undefined,
terminalError: attemptTrajectoryTerminal.terminalError,
});
trajectoryEndRecorded = true;
return {
const result: EmbeddedRunAttemptResult = {
replayMetadata,
currentAttemptReplayMetadata,
itemLifecycle: getItemLifecycle(),
@@ -6410,6 +5752,16 @@ export async function runEmbeddedAttempt(
clientToolCalls: completedClientToolCalls.length > 0 ? completedClientToolCalls : undefined,
yieldDetected: yieldDetected || undefined,
};
const finalizedResult = finalizeEmbeddedAttempt({
result,
trajectoryRecorder,
synthesizedPayloadCount,
emptyAssistantReplyIsSilent,
hasTerminalOutput,
silentExpected: params.silentExpected,
});
trajectoryEndRecorded = true;
return finalizedResult;
} finally {
if (trajectoryRecorder && !trajectoryEndRecorded) {
trajectoryRecorder.recordEvent("session.ended", {
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+315
View File
@@ -0,0 +1,315 @@
import {
findOpenAIStrictToolProjectionDiagnostics,
resolveOpenAIProjectedToolsStrictToolFlag,
type OpenAIToolProjection,
} from "@openclaw/ai/internal/openai";
import { isRecord } from "@openclaw/normalization-core/record-coerce";
import { normalizeLowercaseStringOrEmpty } from "@openclaw/normalization-core/string-coerce";
import { sha256Hex } from "../infra/crypto-digest.js";
import type { Context, Model } from "../llm/types.js";
import { isCodeModeModelVisibleToolName } from "./code-mode-control-tools.js";
import { buildCopilotDynamicHeaders, hasCopilotVisionInput } from "./copilot-dynamic-headers.js";
import { detectOpenAICompletionsCompat } from "./openai-completions-compat.js";
import { resolveOpenAIReasoningEffortMap } from "./openai-reasoning-compat.js";
import { log, type OpenAIModeModel } from "./openai-transport-shared.js";
import { resolveProviderRequestPolicyConfig } from "./provider-request-config.js";
import { resolveModelRequestTimeoutMs } from "./provider-transport-fetch.js";
const MAX_OPENAI_STRICT_TOOL_DOWNGRADE_DIAGNOSTIC_KEYS = 256;
const OPENAI_CODEX_RESPONSES_PROVIDERS = new Set(["openai"]);
const loggedOpenAIStrictToolDowngradeDiagnosticKeys = new Set<string>();
function readToolPayloadField(record: Record<string, unknown>, field: string): unknown {
try {
return record[field];
} catch {
return undefined;
}
}
function transportPayloadToolName(tool: unknown): string | undefined {
if (!isRecord(tool)) {
return undefined;
}
const name = readToolPayloadField(tool, "name");
if (typeof name === "string") {
return name;
}
const fn = readToolPayloadField(tool, "function");
if (!isRecord(fn)) {
return undefined;
}
const fnName = readToolPayloadField(fn, "name");
return typeof fnName === "string" ? fnName : undefined;
}
export function enforceCodeModeResponsesToolSurface(payload: unknown): void {
if (!isRecord(payload) || !Array.isArray(payload.tools)) {
return;
}
payload.tools = payload.tools.filter((tool) => {
const name = transportPayloadToolName(tool);
return typeof name === "string" && isCodeModeModelVisibleToolName(name);
});
}
export function assertCodeModeResponsesToolSurface(payload: unknown): void {
if (!isRecord(payload) || !Array.isArray(payload.tools)) {
throw new Error("Code mode payload tool surface violation: expected exec,wait; got no tools");
}
const names = payload.tools
.map(transportPayloadToolName)
.filter((name): name is string => typeof name === "string" && name.length > 0)
.toSorted((left, right) => left.localeCompare(right));
if (
names.length >= 2 &&
new Set(names).size === names.length &&
names.filter((name) => name === "exec").length === 1 &&
names.filter((name) => name === "wait").length === 1 &&
names.every(isCodeModeModelVisibleToolName)
) {
return;
}
throw new Error(
`Code mode payload tool surface violation: expected exec,wait plus direct-only tools; got ${
names.length > 0 ? names.join(",") : "none"
}`,
);
}
function buildOpenAIStrictToolDowngradeDiagnosticKey(
diagnostics: ReturnType<typeof findOpenAIStrictToolProjectionDiagnostics>,
context: { transport: "responses" | "completions"; model: OpenAIModeModel },
): string {
return sha256Hex(
JSON.stringify({
transport: context.transport,
provider: context.model.provider ?? null,
model: context.model.id ?? null,
diagnostics: diagnostics.map((entry) => ({
toolIndex: entry.toolIndex,
toolName: entry.toolName ?? null,
violations: entry.violations,
})),
}),
);
}
function shouldLogOpenAIStrictToolDowngradeDiagnostic(
diagnostics: ReturnType<typeof findOpenAIStrictToolProjectionDiagnostics>,
context: { transport: "responses" | "completions"; model: OpenAIModeModel },
): boolean {
const key = buildOpenAIStrictToolDowngradeDiagnosticKey(diagnostics, context);
if (loggedOpenAIStrictToolDowngradeDiagnosticKeys.has(key)) {
return false;
}
if (
loggedOpenAIStrictToolDowngradeDiagnosticKeys.size >=
MAX_OPENAI_STRICT_TOOL_DOWNGRADE_DIAGNOSTIC_KEYS
) {
loggedOpenAIStrictToolDowngradeDiagnosticKeys.clear();
}
loggedOpenAIStrictToolDowngradeDiagnosticKeys.add(key);
return true;
}
export function resolveOpenAIStrictToolFlagWithDiagnostics(
projection: OpenAIToolProjection,
strictSetting: boolean | null | undefined,
context: { transport: "responses" | "completions"; model: OpenAIModeModel },
): boolean | undefined {
const strict = resolveOpenAIProjectedToolsStrictToolFlag(projection, strictSetting);
if (strictSetting === true && strict === false && log.isEnabled("debug", "any")) {
const diagnostics = findOpenAIStrictToolProjectionDiagnostics(projection);
if (!shouldLogOpenAIStrictToolDowngradeDiagnostic(diagnostics, context)) {
return strict;
}
const sample = diagnostics.slice(0, 5).map((entry) => ({
tool: entry.toolName ?? `tool[${entry.toolIndex}]`,
violations: entry.violations.slice(0, 8),
}));
log.debug(
`OpenAI ${context.transport} tool schema strict mode downgraded to strict=false for ` +
`${context.model.provider ?? "unknown"}/${context.model.id ?? "unknown"} ` +
`because ${diagnostics.length} tool schema(s) are not strict-compatible`,
{
transport: context.transport,
provider: context.model.provider,
model: context.model.id,
incompatibleToolCount: diagnostics.length,
sample,
},
);
}
return strict;
}
export function isOpenAICodexResponsesModel(model: Model): boolean {
return (
OPENAI_CODEX_RESPONSES_PROVIDERS.has(model.provider) &&
(model.api === "openai-chatgpt-responses" ||
model.api === "openclaw-openai-responses-transport")
);
}
function isNativeOpenAICodexResponsesBaseUrl(baseUrl?: string): boolean {
const trimmed = typeof baseUrl === "string" ? baseUrl.trim() : "";
if (!trimmed) {
return false;
}
try {
const url = new URL(trimmed);
if (url.protocol !== "http:" && url.protocol !== "https:") {
return false;
}
if (url.hostname.toLowerCase() !== "chatgpt.com") {
return false;
}
const pathname = url.pathname.replace(/\/+$/u, "").toLowerCase();
return [
"/backend-api",
"/backend-api/v1",
"/backend-api/codex",
"/backend-api/codex/v1",
].includes(pathname);
} catch {
return false;
}
}
export function usesNativeOpenAICodexResponsesBackend(model: Model): boolean {
return isOpenAICodexResponsesModel(model) && isNativeOpenAICodexResponsesBaseUrl(model.baseUrl);
}
export function buildOpenAIClientHeaders(
model: Model,
context: Context,
optionHeaders?: Record<string, string>,
turnHeaders?: Record<string, string>,
sessionId?: string,
): Record<string, string> {
const providerHeaders = { ...model.headers };
if (model.provider === "github-copilot") {
Object.assign(
providerHeaders,
buildCopilotDynamicHeaders({
messages: context.messages,
hasImages: hasCopilotVisionInput(context.messages),
}),
);
}
const callerHeaders = { ...optionHeaders, ...turnHeaders };
const headers = resolveProviderRequestPolicyConfig({
provider: model.provider,
api: model.api,
baseUrl: model.baseUrl,
capability: "llm",
transport: "stream",
providerHeaders,
callerHeaders: Object.keys(callerHeaders).length > 0 ? callerHeaders : undefined,
precedence: "caller-wins",
}).headers;
const resolvedHeaders = headers ?? {};
// Preserve ChatGPT Responses session affinity; the native backend accepts this spelling.
if (
sessionId &&
!Object.keys(resolvedHeaders).some(
(key) => normalizeLowercaseStringOrEmpty(key) === "session_id",
) &&
usesNativeOpenAICodexResponsesBackend(model)
) {
resolvedHeaders.session_id = sessionId;
}
return resolvedHeaders;
}
function resolveOpenAISdkTimeoutMs(model: Model): number | undefined {
return resolveModelRequestTimeoutMs(model, undefined);
}
export function buildOpenAISdkClientOptions(model: Model): { timeout?: number } {
const timeout = resolveOpenAISdkTimeoutMs(model);
return timeout === undefined ? {} : { timeout };
}
export function buildOpenAISdkRequestOptions(
model: Model,
signal?: AbortSignal,
options?: { stream?: boolean },
): { signal?: AbortSignal; timeout?: number; headers?: Record<string, string> } | undefined {
const timeout = resolveOpenAISdkTimeoutMs(model);
const headers =
options?.stream === true && usesNativeOpenAICodexResponsesBackend(model)
? { Accept: "text/event-stream" }
: undefined;
if (timeout === undefined && !signal && !headers) {
return undefined;
}
return {
...(headers ? { headers } : {}),
...(signal ? { signal } : {}),
...(timeout !== undefined ? { timeout } : {}),
};
}
function detectCompat(model: OpenAIModeModel) {
const { defaults } = detectOpenAICompletionsCompat(model);
return {
supportsStore: defaults.supportsStore,
supportsDeveloperRole: defaults.supportsDeveloperRole,
supportsReasoningEffort: defaults.supportsReasoningEffort,
reasoningEffortMap: {},
supportsUsageInStreaming: defaults.supportsUsageInStreaming,
maxTokensField: defaults.maxTokensField,
requiresToolResultName: false,
requiresAssistantAfterToolResult: false,
requiresThinkingAsText: false,
thinkingFormat: defaults.thinkingFormat,
visibleReasoningDetailTypes: defaults.visibleReasoningDetailTypes,
openRouterRouting: {},
vercelGatewayRouting: {},
supportsStrictMode: defaults.supportsStrictMode,
requiresReasoningContentOnAssistantMessages:
defaults.requiresReasoningContentOnAssistantMessages,
requiresNonEmptyUserOrAssistantMessage: defaults.requiresNonEmptyUserOrAssistantMessage,
};
}
export function getCompat(model: OpenAIModeModel) {
const detected = detectCompat(model);
const compat = model.compat ?? {};
const supportsStore =
typeof compat.supportsStore === "boolean" ? compat.supportsStore : detected.supportsStore;
const supportsReasoningEffort =
typeof compat.supportsReasoningEffort === "boolean"
? compat.supportsReasoningEffort
: detected.supportsReasoningEffort;
return {
supportsStore,
supportsDeveloperRole: compat.supportsDeveloperRole ?? detected.supportsDeveloperRole,
supportsReasoningEffort,
reasoningEffortMap: resolveOpenAIReasoningEffortMap(model, detected.reasoningEffortMap),
supportsUsageInStreaming: compat.supportsUsageInStreaming ?? detected.supportsUsageInStreaming,
maxTokensField: (compat.maxTokensField as string | undefined) ?? detected.maxTokensField,
requiresToolResultName: compat.requiresToolResultName ?? detected.requiresToolResultName,
requiresAssistantAfterToolResult:
compat.requiresAssistantAfterToolResult ?? detected.requiresAssistantAfterToolResult,
requiresThinkingAsText: compat.requiresThinkingAsText ?? detected.requiresThinkingAsText,
thinkingFormat: compat.thinkingFormat ?? detected.thinkingFormat,
openRouterRouting: (compat.openRouterRouting as Record<string, unknown> | undefined) ?? {},
vercelGatewayRouting:
(compat.vercelGatewayRouting as Record<string, unknown> | undefined) ??
detected.vercelGatewayRouting,
supportsStrictMode: compat.supportsStrictMode ?? detected.supportsStrictMode,
supportsPromptCacheKey: compat.supportsPromptCacheKey === true,
supportsLongCacheRetention: compat.supportsLongCacheRetention !== false,
requiresStringContent: compat.requiresStringContent ?? false,
strictMessageKeys: compat.strictMessageKeys === true,
visibleReasoningDetailTypes:
compat.visibleReasoningDetailTypes ?? detected.visibleReasoningDetailTypes,
requiresReasoningContentOnAssistantMessages:
compat.requiresReasoningContentOnAssistantMessages ??
detected.requiresReasoningContentOnAssistantMessages,
requiresNonEmptyUserOrAssistantMessage: detected.requiresNonEmptyUserOrAssistantMessage,
};
}
+155
View File
@@ -0,0 +1,155 @@
/** Shared options, usage shape, cache identity, ordering, and stream scheduling for OpenAI APIs. */
import {
clampOpenAIPromptCacheKey,
type OpenAICompletionsToolChoice,
type OpenAIReasoningEffort,
} from "@openclaw/ai/internal/openai";
import type { ModelCompatConfig } from "../config/types.models.js";
import type { Api, Model, Usage } from "../llm/types.js";
import { createSubsystemLogger } from "../logging/subsystem.js";
const MODEL_STREAM_COOPERATIVE_YIELD_INTERVAL_MS = 12;
const MODEL_STREAM_COOPERATIVE_YIELD_MAX_EVENTS = 64;
export const GEMINI_THOUGHT_SIGNATURE_VALIDATOR_SKIP = "skip_thought_signature_validator";
export const log = createSubsystemLogger("openai-transport");
export type BaseOpenAIStreamOptions = {
temperature?: number;
topP?: number;
maxTokens?: number;
stop?: string[];
signal?: AbortSignal;
apiKey?: string;
cacheRetention?: "none" | "short" | "long";
sessionId?: string;
promptCacheKey?: string;
authProfileId?: string;
onPayload?: (payload: unknown, model: Model) => unknown;
headers?: Record<string, string>;
firstEventTimeoutMs?: number;
onFirstEventTimeout?: (reason: Error) => void;
openclawCodeModeToolSurface?: boolean;
responseFormat?: Record<string, unknown>;
frequencyPenalty?: number;
presencePenalty?: number;
seed?: number;
};
export type OpenAICompletionsOptions = BaseOpenAIStreamOptions & {
toolChoice?: OpenAICompletionsToolChoice;
reasoning?: OpenAIReasoningEffort;
reasoningEffort?: OpenAIReasoningEffort;
};
type OpenAIModeCompatInput = Omit<ModelCompatConfig, "thinkingFormat"> & {
thinkingFormat?: string;
};
export type OpenAIModeModel = Omit<Model, "compat"> & {
compat?: OpenAIModeCompatInput | null;
};
export type MutableAssistantOutput = {
role: "assistant";
content: Array<Record<string, unknown>>;
api: Api;
provider: string;
model: string;
usage: {
input: number;
output: number;
cacheRead: number;
cacheWrite: number;
reasoningTokens?: number;
totalTokens: number;
cost: Usage["cost"];
};
stopReason: string;
timestamp: number;
responseId?: string;
errorMessage?: string;
errorCode?: string;
errorType?: string;
errorBody?: string;
};
type ModelStreamCooperativeScheduler = {
afterEvent: () => Promise<void>;
};
export function throwIfModelStreamAborted(signal?: AbortSignal): void {
if (signal?.aborted) {
throw new Error("Request was aborted");
}
}
export function createModelStreamCooperativeScheduler(
signal?: AbortSignal,
): ModelStreamCooperativeScheduler {
let lastYieldedAt = Date.now();
let eventsSinceYield = 0;
return {
async afterEvent() {
throwIfModelStreamAborted(signal);
eventsSinceYield += 1;
const now = Date.now();
if (
eventsSinceYield < MODEL_STREAM_COOPERATIVE_YIELD_MAX_EVENTS &&
now - lastYieldedAt < MODEL_STREAM_COOPERATIVE_YIELD_INTERVAL_MS
) {
return;
}
eventsSinceYield = 0;
lastYieldedAt = now;
await new Promise<void>((resolve) => {
setTimeout(resolve, 0);
});
throwIfModelStreamAborted(signal);
},
};
}
export function resolveCacheRetention(
cacheRetention: string | undefined,
): "short" | "long" | "none" {
if (cacheRetention === "short" || cacheRetention === "long" || cacheRetention === "none") {
return cacheRetention;
}
if (typeof process !== "undefined" && process.env.OPENCLAW_CACHE_RETENTION === "long") {
return "long";
}
return "short";
}
export function resolvePromptCacheKey(
options: Pick<BaseOpenAIStreamOptions, "promptCacheKey" | "sessionId"> | undefined,
cacheRetention: "short" | "long" | "none",
): string | undefined {
if (cacheRetention === "none") {
return undefined;
}
return clampOpenAIPromptCacheKey(options?.promptCacheKey ?? options?.sessionId);
}
function compareTransportToolText(left: string | undefined, right: string | undefined): number {
const leftText = left ?? "";
const rightText = right ?? "";
if (leftText < rightText) {
return -1;
}
if (leftText > rightText) {
return 1;
}
return 0;
}
export function sortTransportToolsByName<T extends { name?: string; description?: string }>(
tools: readonly T[],
): T[] {
return tools.toSorted(
(left, right) =>
compareTransportToolText(left.name, right.name) ||
compareTransportToolText(left.description, right.description),
);
}
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,321 @@
import { expectDefined } from "@openclaw/normalization-core";
import {
readPairingQrReplyChannelData,
type ReplyPayload,
} from "../../auto-reply/reply-payload.js";
import { normalizeReplyPayloadsForDelivery } from "../../infra/outbound/payloads.js";
import { renderQrPngDataUrl } from "../../media/qr-image.js";
import { renderQrTerminal } from "../../media/qr-terminal.js";
import { stripInlineDirectiveTagsForDisplay } from "../../utils/directive-tags.js";
import { stripEnvelopeFromMessage } from "../chat-sanitize.js";
import {
cleanupManagedOutgoingImageRecords,
createManagedOutgoingImageBlocks,
} from "../managed-image-attachments.js";
import { formatForLog } from "../ws-log.js";
import { buildWebchatAudioContentBlocksFromReplyPayloads } from "./chat-webchat-media.js";
import type { GatewayRequestContext } from "./types.js";
const MANAGED_OUTGOING_IMAGE_PATH_PREFIX = "/api/chat/media/outgoing/";
const chatHistoryManagedImageCleanupState = new Map<string, Promise<void>>();
export type AssistantDisplayContentBlock = Record<string, unknown>;
export function isMediaBearingPayload(payload: ReplyPayload): boolean {
if (payload.isReasoning === true) {
return false;
}
if (payload.mediaUrl?.trim()) {
return true;
}
return Boolean(payload.mediaUrls?.some((url) => url.trim()));
}
export function hasSensitiveMediaPayload(payloads: ReplyPayload[]): boolean {
return payloads.some(
(payload) =>
payload.sensitiveMedia === true &&
(isMediaBearingPayload(payload) || Boolean(readPairingQrReplyChannelData(payload))),
);
}
async function buildPairingQrAssistantContentBlock(
payload: ReplyPayload,
): Promise<AssistantDisplayContentBlock | undefined> {
const qr = readPairingQrReplyChannelData(payload);
if (!qr) {
return undefined;
}
const [imageUrl, terminalText] = await Promise.all([
renderQrPngDataUrl(qr.setupCode),
renderQrTerminal(qr.setupCode, { small: true }),
]);
return {
type: "openclaw_pairing_qr",
image_url: imageUrl,
terminalText,
alt: "OpenClaw pairing QR code",
expiresAtMs: qr.expiresAtMs,
sensitive: true,
};
}
export function sanitizeAssistantDisplayText(value?: string | null): string | undefined {
if (!value) {
return undefined;
}
const withoutEnvelope = stripEnvelopeFromMessage(value);
const normalized = typeof withoutEnvelope === "string" ? withoutEnvelope : value;
const stripped = stripInlineDirectiveTagsForDisplay(normalized).text.trim();
return stripped || undefined;
}
export function extractAssistantDisplayTextFromContent(
content?: readonly AssistantDisplayContentBlock[] | null,
): string | undefined {
if (!Array.isArray(content) || content.length === 0) {
return undefined;
}
const parts = content
.map((block) => {
if (block?.type !== "text" || typeof block.text !== "string") {
return "";
}
return block.text.trim();
})
.filter(Boolean);
return parts.length > 0 ? parts.join("\n\n") : undefined;
}
export async function buildAssistantDisplayContentFromReplyPayloads(params: {
sessionKey: string;
agentId?: string;
payloads: ReplyPayload[];
managedImageLocalRoots?: Parameters<typeof createManagedOutgoingImageBlocks>[0]["localRoots"];
includeSensitiveMedia?: boolean;
includeSensitiveDisplay?: boolean;
onLocalAudioAccessDenied?: (message: string) => void;
onManagedImagePrepareError?: (message: string) => void;
onSensitiveDisplayPrepareError?: (message: string) => void;
}): Promise<AssistantDisplayContentBlock[] | undefined> {
const rawTextPayloadCount = params.payloads.filter(
(payload) =>
payload.isReasoning !== true &&
typeof payload.text === "string" &&
payload.text.trim().length > 0,
).length;
const normalized = normalizeReplyPayloadsForDelivery(params.payloads);
if (normalized.length === 0) {
return rawTextPayloadCount > 0 ? [{ type: "text", text: "" }] : undefined;
}
const content: AssistantDisplayContentBlock[] = [];
let strippedTextPayloadCount = 0;
for (const payload of normalized) {
const text = sanitizeAssistantDisplayText(payload.text);
if (text) {
content.push({ type: "text", text });
} else if (typeof payload.text === "string" && payload.text.trim().length > 0) {
strippedTextPayloadCount += 1;
}
if (params.includeSensitiveDisplay === true) {
try {
const pairingQrBlock = await buildPairingQrAssistantContentBlock(payload);
if (pairingQrBlock) {
content.push(pairingQrBlock);
}
} catch (err) {
params.onSensitiveDisplayPrepareError?.(formatForLog(err));
}
}
if (params.includeSensitiveMedia === false && payload.sensitiveMedia === true) {
continue;
}
const audioBlocks = await buildWebchatAudioContentBlocksFromReplyPayloads([payload], {
localRoots: Array.isArray(params.managedImageLocalRoots)
? params.managedImageLocalRoots
: undefined,
onLocalAudioAccessDenied: (err) => {
params.onLocalAudioAccessDenied?.(formatForLog(err));
},
});
content.push(...audioBlocks);
const mediaUrls = Array.from(
new Set([
...(Array.isArray(payload.mediaUrls) ? payload.mediaUrls : []),
...(typeof payload.mediaUrl === "string" ? [payload.mediaUrl] : []),
]),
);
const imageBlocks = await createManagedOutgoingImageBlocks({
sessionKey: params.sessionKey,
...(params.sessionKey === "global" && params.agentId ? { agentId: params.agentId } : {}),
mediaUrls,
localRoots: params.managedImageLocalRoots,
continueOnPrepareError: true,
onPrepareError: (error) => {
params.onManagedImagePrepareError?.(error.message);
},
});
if (imageBlocks.length > 0) {
content.push(...imageBlocks);
}
}
if (content.length > 0) {
return content;
}
return strippedTextPayloadCount > 0 ? [{ type: "text", text: "" }] : undefined;
}
export function replaceAssistantContentTextBlocks(
content: readonly AssistantDisplayContentBlock[] | undefined,
transcriptMediaMessage: { content: Array<Record<string, unknown>> } | null,
): AssistantDisplayContentBlock[] | undefined {
const transcriptTextBlocks = (transcriptMediaMessage?.content ?? []).filter(
(block): block is AssistantDisplayContentBlock =>
Boolean(block) &&
typeof block === "object" &&
block.type === "text" &&
typeof block.text === "string",
);
if (transcriptTextBlocks.length === 0) {
return content ? [...content] : undefined;
}
if (!content || content.length === 0) {
return [...transcriptTextBlocks];
}
const merged: AssistantDisplayContentBlock[] = [];
let transcriptTextIndex = 0;
for (const block of content) {
if (
block?.type === "text" &&
typeof block.text === "string" &&
transcriptTextIndex < transcriptTextBlocks.length
) {
merged.push(
expectDefined(
transcriptTextBlocks[transcriptTextIndex++],
"transcript text blocks entry at transcript text index++",
),
);
continue;
}
merged.push(block);
}
if (transcriptTextIndex < transcriptTextBlocks.length) {
merged.unshift(...transcriptTextBlocks.slice(transcriptTextIndex));
}
return merged;
}
function isManagedOutgoingImageUrl(value: unknown): boolean {
if (typeof value !== "string" || !value.trim()) {
return false;
}
try {
const parsed = new URL(value, "http://localhost");
return parsed.pathname.startsWith(MANAGED_OUTGOING_IMAGE_PATH_PREFIX);
} catch {
return false;
}
}
export function stripManagedOutgoingAssistantContentBlocks(
content: readonly AssistantDisplayContentBlock[] | undefined,
): AssistantDisplayContentBlock[] | undefined {
if (!content || content.length === 0) {
return undefined;
}
const filtered = content.filter((block) => {
if (block?.type !== "image") {
return true;
}
return !(isManagedOutgoingImageUrl(block.url) || isManagedOutgoingImageUrl(block.openUrl));
});
return filtered.length > 0 ? filtered : undefined;
}
export function extractAssistantDisplayText(
content: readonly AssistantDisplayContentBlock[] | undefined,
): string | undefined {
if (!content || content.length === 0) {
return undefined;
}
const text = content
.map((block) => (block?.type === "text" && typeof block.text === "string" ? block.text : ""))
.filter(Boolean)
.join("\n\n")
.trim();
return text || undefined;
}
export function hasAssistantDisplayMediaContent(
content: readonly AssistantDisplayContentBlock[] | undefined,
): boolean {
return Boolean(content?.some((block) => block?.type !== "text"));
}
export function hasVisibleAssistantFinalMessage(
message: Record<string, unknown> | undefined,
): boolean {
if (!message) {
return false;
}
if (typeof message.text === "string" && message.text.trim()) {
return true;
}
const content = Array.isArray(message.content) ? message.content : [];
return content.some((block) => {
if (!block || typeof block !== "object") {
return false;
}
const record = block as Record<string, unknown>;
if (record.type === "text") {
return typeof record.text === "string" && record.text.trim().length > 0;
}
return true;
});
}
export function hasManagedOutgoingAssistantContent(
content: readonly AssistantDisplayContentBlock[] | undefined,
): boolean {
return Boolean(
content?.some(
(block) =>
block?.type === "image" &&
(isManagedOutgoingImageUrl(block.url) || isManagedOutgoingImageUrl(block.openUrl)),
),
);
}
export function scheduleChatHistoryManagedImageCleanup(params: {
sessionKey: string;
agentId?: string;
context: Pick<GatewayRequestContext, "logGateway">;
}) {
const cleanupKey =
params.sessionKey === "global" && params.agentId
? `agent:${params.agentId}:global`
: params.sessionKey;
if (chatHistoryManagedImageCleanupState.has(cleanupKey)) {
return;
}
const pending = cleanupManagedOutgoingImageRecords({
sessionKey: params.sessionKey,
...(params.sessionKey === "global" && params.agentId ? { agentId: params.agentId } : {}),
})
.then(() => undefined)
.catch((error: unknown) => {
params.context.logGateway.debug(
`chat.history managed image cleanup skipped sessionKey=${JSON.stringify(params.sessionKey)} error=${formatForLog(error)}`,
);
})
.finally(() => {
if (chatHistoryManagedImageCleanupState.get(cleanupKey) === pending) {
chatHistoryManagedImageCleanupState.delete(cleanupKey);
}
});
chatHistoryManagedImageCleanupState.set(cleanupKey, pending);
}
@@ -1,7 +1,7 @@
// Covers the chat.history final byte-budget fallback, including the sentinel
// that prevents an empty (blank) transcript from being returned to the dashboard.
import { describe, expect, it } from "vitest";
import { enforceChatHistoryFinalBudget } from "./chat.js";
import { enforceChatHistoryFinalBudget } from "./chat-history-budget.js";
type DisplayMessage = {
role?: string;
@@ -0,0 +1,129 @@
import { jsonUtf8Bytes } from "../../infra/json-utf8-bytes.js";
import { logLargePayload } from "../../logging/diagnostic-payload.js";
export const CHAT_HISTORY_MAX_SINGLE_MESSAGE_BYTES = 128 * 1024;
const CHAT_HISTORY_OVERSIZED_PLACEHOLDER = "[chat.history omitted: message too large]";
const CHAT_HISTORY_UNAVAILABLE_SENTINEL =
"[chat.history unavailable: transcript too large to display; the full history is preserved on disk]";
let chatHistoryOmittedEmitCount = 0;
function buildChatHistoryUnavailableSentinel(): Record<string, unknown> {
return {
role: "assistant",
timestamp: Date.now(),
content: [{ type: "text", text: CHAT_HISTORY_UNAVAILABLE_SENTINEL }],
};
}
function buildOversizedHistoryPlaceholder(message?: unknown): Record<string, unknown> {
const role =
message &&
typeof message === "object" &&
typeof (message as { role?: unknown }).role === "string"
? (message as { role: string }).role
: "assistant";
const timestamp =
message &&
typeof message === "object" &&
typeof (message as { timestamp?: unknown }).timestamp === "number"
? (message as { timestamp: number }).timestamp
: Date.now();
const rawMetadata =
message && typeof message === "object"
? (message as Record<string, unknown>)["__openclaw"]
: undefined;
const metadata =
rawMetadata && typeof rawMetadata === "object" && !Array.isArray(rawMetadata)
? (rawMetadata as Record<string, unknown>)
: {};
const metadataId = typeof metadata.id === "string" ? metadata.id : undefined;
const metadataSeq = typeof metadata.seq === "number" ? metadata.seq : undefined;
const metadataIdempotencyKey =
typeof metadata.idempotencyKey === "string" ? metadata.idempotencyKey : undefined;
return {
role,
timestamp,
content: [{ type: "text", text: CHAT_HISTORY_OVERSIZED_PLACEHOLDER }],
__openclaw: {
...(metadataId ? { id: metadataId } : {}),
...(metadataSeq !== undefined ? { seq: metadataSeq } : {}),
...(metadataIdempotencyKey ? { idempotencyKey: metadataIdempotencyKey } : {}),
truncated: true,
reason: "oversized",
},
};
}
export function replaceOversizedChatHistoryMessages(params: {
messages: unknown[];
maxSingleMessageBytes: number;
}): { messages: unknown[]; replacedCount: number } {
const { messages, maxSingleMessageBytes } = params;
if (messages.length === 0) {
return { messages, replacedCount: 0 };
}
let replacedCount = 0;
const next = messages.map((message) => {
if (jsonUtf8Bytes(message) <= maxSingleMessageBytes) {
return message;
}
replacedCount += 1;
return buildOversizedHistoryPlaceholder(message);
});
return { messages: replacedCount > 0 ? next : messages, replacedCount };
}
// Preserve a visible terminal record when the complete projected history cannot fit.
export function enforceChatHistoryFinalBudget(params: { messages: unknown[]; maxBytes: number }): {
messages: unknown[];
} {
const { messages, maxBytes } = params;
if (messages.length === 0) {
return { messages };
}
if (jsonUtf8Bytes(messages) <= maxBytes) {
return { messages };
}
const last = messages.at(-1);
if (last && jsonUtf8Bytes([last]) <= maxBytes) {
return { messages: [last] };
}
const placeholder = buildOversizedHistoryPlaceholder(last);
if (jsonUtf8Bytes([placeholder]) <= maxBytes) {
return { messages: [placeholder] };
}
return { messages: [buildChatHistoryUnavailableSentinel()] };
}
export function reportOmittedChatHistory(params: {
originalMessages: unknown[];
finalMessages: unknown[];
normalizedBytes: number;
maxHistoryBytes: number;
logDebug: (message: string) => void;
}): number {
const { originalMessages, finalMessages, normalizedBytes, maxHistoryBytes, logDebug } = params;
const survivors = new Set(finalMessages);
let omittedCount = 0;
for (const message of originalMessages) {
if (!survivors.has(message)) {
omittedCount += 1;
}
}
if (omittedCount === 0) {
return 0;
}
chatHistoryOmittedEmitCount += omittedCount;
logLargePayload({
surface: "gateway.chat.history",
action: "truncated",
bytes: normalizedBytes,
limitBytes: maxHistoryBytes,
count: omittedCount,
reason: "chat_history_budget",
});
logDebug(
`chat.history omitted oversized payloads count=${omittedCount} total=${chatHistoryOmittedEmitCount}`,
);
return omittedCount;
}
@@ -13,7 +13,7 @@ import {
enforceChatHistoryFinalBudget,
replaceOversizedChatHistoryMessages,
reportOmittedChatHistory,
} from "./chat.js";
} from "./chat-history-budget.js";
type Captured = DiagnosticPayloadLargeEvent[];
@@ -0,0 +1,327 @@
import {
GATEWAY_CLIENT_MODES,
GATEWAY_CLIENT_NAMES,
} from "../../../packages/gateway-protocol/src/client-info.js";
import { CHAT_SEND_SESSION_KEY_MAX_LENGTH } from "../../../packages/gateway-protocol/src/schema.js";
import { listAgentIds } from "../../agents/agent-scope.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import { getSessionBindingService } from "../../infra/outbound/session-binding-service.js";
import type { ChannelRouteRef } from "../../plugin-sdk/channel-route.js";
import { isPluginOwnedSessionBindingRecord } from "../../plugins/conversation-binding.js";
import { normalizeAgentId, scopeLegacySessionKeyToAgent } from "../../routing/session-key.js";
import { parseAgentSessionKey } from "../../sessions/session-key-utils.js";
import { deliveryContextFromSession } from "../../utils/delivery-context.shared.js";
import {
INTERNAL_MESSAGE_CHANNEL,
isGatewayCliClient,
isWebchatClient,
normalizeMessageChannel,
} from "../../utils/message-channel.js";
import { sanitizeChatSendMessageInput } from "../chat-input-sanitize.js";
import { ADMIN_SCOPE } from "../method-scopes.js";
import { resolveSessionStoreKey } from "../session-utils.js";
import type { GatewayRequestHandlerOptions } from "./types.js";
const CHANNEL_AGNOSTIC_SESSION_SCOPES = new Set([
"main",
"direct",
"dm",
"group",
"channel",
"cron",
"run",
"subagent",
"acp",
"thread",
"topic",
]);
const CHANNEL_SCOPED_SESSION_SHAPES = new Set(["direct", "dm", "group", "channel"]);
export type ChatSendDeliveryEntry = {
route?: ChannelRouteRef;
deliveryContext?: {
channel?: string;
to?: string;
accountId?: string;
threadId?: string | number;
};
origin?: {
provider?: string;
accountId?: string;
threadId?: string | number;
};
lastChannel?: string;
lastTo?: string;
lastAccountId?: string;
lastThreadId?: string | number;
};
export type ChatSendOriginatingRoute = {
originatingChannel: string;
originatingTo?: string;
accountId?: string;
messageThreadId?: string | number;
explicitDeliverRoute: boolean;
};
export type ChatSendExplicitOrigin = {
originatingChannel?: string;
originatingTo?: string;
accountId?: string;
messageThreadId?: string;
};
function normalizeOptionalText(value?: string | null): string | undefined {
const trimmed = value?.trim();
return trimmed ? trimmed : undefined;
}
export function validateChatSelectedAgent(params: {
cfg: OpenClawConfig;
requestedSessionKey: string;
agentId?: string;
}): { ok: true; agentId?: string } | { ok: false; error: string } {
const agentId = params.agentId ? normalizeAgentId(params.agentId) : undefined;
if (!agentId) {
return { ok: true };
}
if (!listAgentIds(params.cfg).includes(agentId)) {
return { ok: false, error: `Unknown agent id "${params.agentId}"` };
}
const requestedSessionKey = params.requestedSessionKey.trim();
const parsed = parseAgentSessionKey(requestedSessionKey);
if (parsed && normalizeAgentId(parsed.agentId) !== agentId) {
return {
ok: false,
error: `agentId "${params.agentId}" does not match session key "${params.requestedSessionKey}"`,
};
}
if (requestedSessionKey.toLowerCase() === "global") {
return { ok: true, agentId };
}
if (resolveSessionStoreKey({ cfg: params.cfg, sessionKey: requestedSessionKey }) === "global") {
return { ok: true, agentId };
}
if (!parsed || normalizeAgentId(parsed.agentId) !== agentId) {
return {
ok: false,
error: `agentId "${params.agentId}" does not match session key "${params.requestedSessionKey}"`,
};
}
return { ok: true, agentId };
}
export function resolveRequestedChatAgentId(params: {
cfg?: OpenClawConfig;
requestedSessionKey: string;
agentId?: string;
}): string | undefined {
const explicitAgentId = normalizeOptionalText(params.agentId);
if (explicitAgentId) {
return normalizeAgentId(explicitAgentId);
}
if (!params.cfg) {
return undefined;
}
const parsed = parseAgentSessionKey(params.requestedSessionKey.trim());
if (
!parsed?.agentId ||
resolveSessionStoreKey({ cfg: params.cfg, sessionKey: params.requestedSessionKey }) !== "global"
) {
return undefined;
}
return normalizeAgentId(parsed.agentId);
}
export function resolveChatSendActiveScopeKey(params: {
sessionKey: string;
agentId?: string;
mainKey?: string;
}): string {
if (params.sessionKey !== "global" || !params.agentId) {
return params.sessionKey;
}
return (
scopeLegacySessionKeyToAgent({
agentId: params.agentId,
sessionKey: params.sessionKey,
mainKey: params.mainKey,
}) ?? params.sessionKey
);
}
export function resolveChatSendOriginatingRoute(params: {
client?: { mode?: string | null; id?: string | null } | null;
deliver?: boolean;
entry?: ChatSendDeliveryEntry;
explicitOrigin?: ChatSendExplicitOrigin;
hasConnectedClient?: boolean;
mainKey?: string;
sessionKey: string;
}): ChatSendOriginatingRoute {
if (params.explicitOrigin?.originatingChannel && params.explicitOrigin.originatingTo) {
return {
originatingChannel: params.explicitOrigin.originatingChannel,
originatingTo: params.explicitOrigin.originatingTo,
...(params.explicitOrigin.accountId ? { accountId: params.explicitOrigin.accountId } : {}),
...(params.explicitOrigin.messageThreadId
? { messageThreadId: params.explicitOrigin.messageThreadId }
: {}),
explicitDeliverRoute: params.deliver === true,
};
}
if (params.deliver !== true) {
return { originatingChannel: INTERNAL_MESSAGE_CHANNEL, explicitDeliverRoute: false };
}
const sessionDeliveryContext = deliveryContextFromSession(params.entry);
const routeChannelCandidate = normalizeMessageChannel(
sessionDeliveryContext?.channel ?? params.entry?.lastChannel ?? params.entry?.origin?.provider,
);
const routeToCandidate = sessionDeliveryContext?.to ?? params.entry?.lastTo;
const routeAccountIdCandidate =
sessionDeliveryContext?.accountId ??
params.entry?.lastAccountId ??
params.entry?.origin?.accountId ??
undefined;
const routeThreadIdCandidate =
sessionDeliveryContext?.threadId ??
params.entry?.lastThreadId ??
params.entry?.origin?.threadId;
if (params.sessionKey.length > CHAT_SEND_SESSION_KEY_MAX_LENGTH) {
return { originatingChannel: INTERNAL_MESSAGE_CHANNEL, explicitDeliverRoute: false };
}
const parsedSessionKey = parseAgentSessionKey(params.sessionKey);
const sessionScopeParts = (parsedSessionKey?.rest ?? params.sessionKey)
.split(":", 3)
.filter(Boolean);
const sessionScopeHead = sessionScopeParts[0];
const sessionChannelHint = normalizeMessageChannel(sessionScopeHead);
const normalizedSessionScopeHead = (sessionScopeHead ?? "").trim().toLowerCase();
const sessionPeerShapeCandidates = [sessionScopeParts[1], sessionScopeParts[2]]
.map((part) => (part ?? "").trim().toLowerCase())
.filter(Boolean);
const isChannelAgnosticSessionScope = CHANNEL_AGNOSTIC_SESSION_SCOPES.has(
normalizedSessionScopeHead,
);
const isChannelScopedSession = sessionPeerShapeCandidates.some((part) =>
CHANNEL_SCOPED_SESSION_SHAPES.has(part),
);
const hasLegacyChannelPeerShape =
!isChannelScopedSession &&
typeof sessionScopeParts[1] === "string" &&
sessionChannelHint === routeChannelCandidate;
const isFromWebchatClient = isWebchatClient(params.client);
const isFromGatewayCliClient = isGatewayCliClient(params.client);
const hasClientMetadata =
(typeof params.client?.mode === "string" && params.client.mode.trim().length > 0) ||
(typeof params.client?.id === "string" && params.client.id.trim().length > 0);
const configuredMainKey = (params.mainKey ?? "main").trim().toLowerCase();
const isConfiguredMainSessionScope =
normalizedSessionScopeHead.length > 0 && normalizedSessionScopeHead === configuredMainKey;
const canInheritConfiguredMainRoute =
isConfiguredMainSessionScope &&
params.hasConnectedClient &&
(isFromGatewayCliClient || !hasClientMetadata);
// Webchat never inherits external delivery. Main-session inheritance is CLI-only
// unless an old caller omitted client metadata entirely.
const canInheritDeliverableRoute = Boolean(
!isFromWebchatClient &&
sessionChannelHint &&
sessionChannelHint !== INTERNAL_MESSAGE_CHANNEL &&
((!isChannelAgnosticSessionScope && (isChannelScopedSession || hasLegacyChannelPeerShape)) ||
canInheritConfiguredMainRoute),
);
const hasDeliverableRoute =
canInheritDeliverableRoute &&
routeChannelCandidate &&
routeChannelCandidate !== INTERNAL_MESSAGE_CHANNEL &&
typeof routeToCandidate === "string" &&
routeToCandidate.trim().length > 0;
if (!hasDeliverableRoute) {
return { originatingChannel: INTERNAL_MESSAGE_CHANNEL, explicitDeliverRoute: false };
}
return {
originatingChannel: routeChannelCandidate,
originatingTo: routeToCandidate,
accountId: routeAccountIdCandidate,
messageThreadId: routeThreadIdCandidate,
explicitDeliverRoute: true,
};
}
function isAcpSessionKey(sessionKey: string | undefined): boolean {
return Boolean(sessionKey?.split(":").includes("acp"));
}
export function explicitOriginTargetsAcpSession(
origin: ChatSendExplicitOrigin | undefined,
): boolean {
if (!origin?.originatingChannel || !origin.originatingTo || !origin.accountId) {
return false;
}
const channel = normalizeMessageChannel(origin.originatingChannel);
if (!channel || channel === INTERNAL_MESSAGE_CHANNEL) {
return false;
}
const binding = getSessionBindingService().resolveByConversation({
channel,
accountId: origin.accountId,
conversationId: origin.originatingTo,
});
return isAcpSessionKey(binding?.targetSessionKey);
}
export function explicitOriginTargetsPluginBinding(
origin: ChatSendExplicitOrigin | undefined,
): boolean {
if (!origin?.originatingChannel || !origin.originatingTo || !origin.accountId) {
return false;
}
const channel = normalizeMessageChannel(origin.originatingChannel);
if (!channel || channel === INTERNAL_MESSAGE_CHANNEL) {
return false;
}
const binding = getSessionBindingService().resolveByConversation({
channel,
accountId: origin.accountId,
conversationId: origin.originatingTo,
});
return isPluginOwnedSessionBindingRecord(binding);
}
export function normalizeOptionalChatSystemReceipt(
value: unknown,
): { ok: true; receipt?: string } | { ok: false; error: string } {
if (value == null) {
return { ok: true };
}
if (typeof value !== "string") {
return { ok: false, error: "systemProvenanceReceipt must be a string" };
}
const sanitized = sanitizeChatSendMessageInput(value);
if (!sanitized.ok) {
return sanitized;
}
const receipt = sanitized.message.trim();
return { ok: true, receipt: receipt || undefined };
}
export function isAcpBridgeClient(client: GatewayRequestHandlerOptions["client"]): boolean {
const info = client?.connect?.client;
return (
info?.id === GATEWAY_CLIENT_NAMES.CLI &&
info?.mode === GATEWAY_CLIENT_MODES.CLI &&
info?.displayName === "ACP" &&
info?.version === "acp"
);
}
export function hasGatewayAdminScope(client: GatewayRequestHandlerOptions["client"]): boolean {
const scopes = Array.isArray(client?.connect?.scopes) ? client.connect.scopes : [];
return scopes.includes(ADMIN_SCOPE);
}
@@ -0,0 +1,103 @@
import { isOperatorUiClient } from "../../utils/message-channel.js";
import type { GatewayClient, GatewayRequestContext } from "./types.js";
export type ChatSendAckServerTiming = {
receivedToAckMs: number;
loadSessionMs: number;
prepareAttachmentsMs?: number;
};
export type ChatSendServerTimingPhase =
| "dispatch-started"
| "model-selected"
| "agent-run-started"
| "first-assistant-event"
| "dispatch-completed"
| "post-dispatch-completed";
export function roundedChatSendTimingMs(value: number): number {
return Math.max(0, Math.round(value * 1000) / 1000);
}
export function chatSendAckServerTimingAttributes(
timing: ChatSendAckServerTiming | undefined,
): Record<string, number> {
if (!timing) {
return {};
}
return {
serverReceivedToAckMs: timing.receivedToAckMs,
serverLoadSessionMs: timing.loadSessionMs,
...(timing.prepareAttachmentsMs !== undefined
? { serverPrepareAttachmentsMs: timing.prepareAttachmentsMs }
: {}),
};
}
export function shouldIncludeChatSendAckServerTiming(client?: {
id?: string | null;
mode?: string | null;
}): boolean {
return isOperatorUiClient(client);
}
const CONTROL_UI_RECONNECT_RESUME_PARAM = "__controlUiReconnectResume";
export function resolveControlUiReconnectResumeParams(
params: unknown,
clientInfo?: { id?: string | null; mode?: string | null },
): { params: unknown; resumeRequested: boolean } {
if (!params || typeof params !== "object" || Array.isArray(params)) {
return { params, resumeRequested: false };
}
const record = params as Record<string, unknown>;
const resumeRequested =
record[CONTROL_UI_RECONNECT_RESUME_PARAM] === true && isOperatorUiClient(clientInfo);
if (!resumeRequested) {
return { params, resumeRequested: false };
}
const validatedParams = { ...record };
delete validatedParams[CONTROL_UI_RECONNECT_RESUME_PARAM];
return { params: validatedParams, resumeRequested: true };
}
export function emitOperatorChatSendServerTiming(params: {
context: Pick<GatewayRequestContext, "broadcastToConnIds">;
client?: GatewayClient | null;
phase: ChatSendServerTimingPhase;
runId: string;
sessionKey: string;
agentId?: string;
receivedAtMs: number;
ackedAtMs: number;
dispatchStartedAtMs?: number;
extra?: Record<string, string | number>;
}) {
const connId =
typeof params.client?.connId === "string" && params.client.connId.trim()
? params.client.connId.trim()
: undefined;
if (!connId || !isOperatorUiClient(params.client?.connect?.client)) {
return;
}
const nowMs = performance.now();
params.context.broadcastToConnIds(
"chat.send_timing",
{
phase: params.phase,
runId: params.runId,
sessionKey: params.sessionKey,
...(params.agentId ? { agentId: params.agentId } : {}),
ackToPhaseMs: roundedChatSendTimingMs(nowMs - params.ackedAtMs),
receivedToPhaseMs: roundedChatSendTimingMs(nowMs - params.receivedAtMs),
...(params.dispatchStartedAtMs !== undefined
? {
dispatchStartedToPhaseMs: roundedChatSendTimingMs(nowMs - params.dispatchStartedAtMs),
}
: {}),
...params.extra,
},
new Set([connId]),
{ dropIfSlow: true },
);
}
@@ -0,0 +1,60 @@
import { createHash } from "node:crypto";
import {
buildTtsSupplementMediaPayload,
getReplyPayloadTtsSupplement,
isReplyPayloadTtsSupplement,
} from "openclaw/plugin-sdk/reply-payload";
import type { ReplyPayload } from "../../auto-reply/reply-payload.js";
import { projectChatDisplayMessage } from "../chat-display-projection.js";
import {
extractAssistantDisplayTextFromContent,
type AssistantDisplayContentBlock,
} from "./chat-assistant-content.js";
import type { GatewayInjectedTtsSupplementMarker } from "./chat-transcript-inject.js";
export function stripVisibleTextFromTtsSupplement(payload: ReplyPayload): ReplyPayload {
return isReplyPayloadTtsSupplement(payload) ? buildTtsSupplementMediaPayload(payload) : payload;
}
function resolveTtsSupplementMarkerText(text: string): string {
const trimmed = text.trim();
const projected = projectChatDisplayMessage(
{
role: "assistant",
content: [{ type: "text", text: trimmed }],
},
{ maxChars: Number.MAX_SAFE_INTEGER },
);
const projectedContent = Array.isArray(projected?.content)
? (projected.content as AssistantDisplayContentBlock[])
: undefined;
return (
extractAssistantDisplayTextFromContent(projectedContent) ??
(typeof projected?.text === "string" ? projected.text.trim() : undefined) ??
trimmed
);
}
export function buildTtsSupplementTranscriptMarker(
payload: ReplyPayload,
): GatewayInjectedTtsSupplementMarker | undefined {
const supplement = getReplyPayloadTtsSupplement(payload);
if (!supplement) {
return undefined;
}
const visibleText = resolveTtsSupplementMarkerText(
payload.text?.trim() || supplement.spokenText.trim(),
);
return {
textSha256: createHash("sha256").update(visibleText).digest("hex"),
};
}
export function buildMediaOnlyTtsSupplementTranscriptMarker(
payload: ReplyPayload,
): GatewayInjectedTtsSupplementMarker | undefined {
if (payload.text?.trim()) {
return undefined;
}
return buildTtsSupplementTranscriptMarker(payload);
}
File diff suppressed because it is too large Load Diff
+4
View File
@@ -1276,6 +1276,7 @@ describe("scripts/changed-lanes", () => {
});
expect(plan.commands.map((command) => command.name)).toEqual([
"conflict markers",
"TypeScript LOC ratchet",
"changelog attributions",
"guarded extension wildcard re-exports",
"plugin-sdk wildcard re-exports",
@@ -1560,6 +1561,7 @@ describe("scripts/changed-lanes", () => {
});
expect(plan.commands.map((command) => command.args[0])).toEqual([
"check:no-conflict-markers",
"check:loc",
"check:changelog-attributions",
"lint:extensions:no-guarded-wildcard-reexports",
"lint:extensions:no-plugin-sdk-wildcard-reexports",
@@ -2130,6 +2132,7 @@ describe("scripts/changed-lanes", () => {
});
expect(plan.commands).toEqual([
{ name: "conflict markers", args: ["check:no-conflict-markers"] },
{ name: "TypeScript LOC ratchet", args: ["check:loc"] },
{ name: "changelog attributions", args: ["check:changelog-attributions"] },
{
name: "guarded extension wildcard re-exports",
@@ -2152,6 +2155,7 @@ describe("scripts/changed-lanes", () => {
expect(result.docsOnly).toBe(true);
expect(plan.commands).toEqual([
{ name: "conflict markers", args: ["check:no-conflict-markers"] },
{ name: "TypeScript LOC ratchet", args: ["check:loc"] },
{ name: "changelog attributions", args: ["check:changelog-attributions"] },
{
name: "guarded extension wildcard re-exports",
+124
View File
@@ -1,6 +1,14 @@
// Check Ts Max Loc tests cover CLI argument validation before repository scans.
import { spawnSync } from "node:child_process";
import { describe, expect, it } from "vitest";
import {
countPhysicalLines,
findLocBaselineUpdateViolations,
findLocRatchetViolations,
findVersionedBaselineViolations,
isProductionTypeScriptFile,
parseArgs,
} from "../../scripts/check-ts-max-loc.js";
function runCheckTsMaxLoc(args: string[]) {
return spawnSync(process.execPath, ["--import", "tsx", "scripts/check-ts-max-loc.ts", ...args], {
@@ -27,4 +35,120 @@ describe("scripts/check-ts-max-loc", () => {
expect(result.stderr).toBe("--max requires a positive integer\n");
}
});
it("parses a safe comparison base ref", () => {
expect(parseArgs(["--base-ref", "refs/remotes/origin/pr-base"])).toMatchObject({
baseRef: "refs/remotes/origin/pr-base",
});
expect(() => parseArgs(["--base-ref", "main^{tree}"])).toThrow("--base-ref requires a git ref");
});
it("fails closed when a comparison ref does not exist", () => {
const result = runCheckTsMaxLoc(["--base-ref", "refs/heads/__loc-ratchet-missing__"]);
expect(result.status).toBe(1);
expect(result.stdout).toBe("");
expect(result.stderr).toBe(
"Invalid TypeScript LOC comparison ref: refs/heads/__loc-ratchet-missing__\n",
);
});
it("grandfathers exact legacy sizes and rejects growth or stale baselines", () => {
const violations = findLocRatchetViolations({
maxLines: 500,
baseline: {
"src/grew.ts": 700,
"src/shrank.ts": 700,
"src/now-small.ts": 700,
"src/removed.ts": 700,
"src/unchanged.ts": 700,
},
results: [
{ filePath: "src/grew.ts", lines: 701 },
{ filePath: "src/new.ts", lines: 501 },
{ filePath: "src/now-small.ts", lines: 500 },
{ filePath: "src/shrank.ts", lines: 699 },
{ filePath: "src/unchanged.ts", lines: 700 },
],
});
expect(violations).toEqual([
{ filePath: "src/grew.ts", lines: 701, baselineLines: 700, reason: "grew" },
{ filePath: "src/shrank.ts", lines: 699, baselineLines: 700, reason: "baseline-stale" },
{ filePath: "src/new.ts", lines: 501, reason: "baseline-missing" },
{
filePath: "src/now-small.ts",
lines: 500,
baselineLines: 700,
reason: "baseline-stale",
},
{ filePath: "src/removed.ts", lines: 0, baselineLines: 700, reason: "baseline-stale" },
]);
});
it("counts physical lines without treating a terminal newline as another line", () => {
expect(countPhysicalLines("")).toBe(0);
expect(countPhysicalLines("one")).toBe(1);
expect(countPhysicalLines("one\n")).toBe(1);
expect(countPhysicalLines("one\ntwo\n")).toBe(2);
});
it("excludes repository test and test-support naming conventions", () => {
expect(isProductionTypeScriptFile("src/runtime.ts")).toBe(true);
expect(isProductionTypeScriptFile("src/runtime.mts")).toBe(true);
expect(isProductionTypeScriptFile("src/runtime.cts")).toBe(true);
for (const filePath of [
"src/runtime.test.ts",
"src/runtime.spec.tsx",
"src/runtime.suite.ts",
"src/runtime.test-harness.ts",
"src/runtime.test-support.ts",
"src/runtime-test-helpers.ts",
"src/test-helpers/runtime.ts",
"test/runtime.ts",
]) {
expect(isProductionTypeScriptFile(filePath), filePath).toBe(false);
}
});
it("allows baseline updates only for decreases and removals", () => {
const violations = findLocBaselineUpdateViolations({
maxLines: 500,
baseline: {
"src/grew.ts": 700,
"src/shrank.ts": 700,
"src/removed.ts": 700,
},
results: [
{ filePath: "src/grew.ts", lines: 701 },
{ filePath: "src/shrank.ts", lines: 650 },
{ filePath: "src/new.ts", lines: 501 },
],
});
expect(violations).toEqual([
{ filePath: "src/grew.ts", lines: 701, baselineLines: 700, reason: "grew" },
{ filePath: "src/new.ts", lines: 501, reason: "baseline-missing" },
]);
});
it("rejects versioned baseline additions and increases", () => {
const violations = findVersionedBaselineViolations({
baseBaseline: {
"src/grew.ts": 700,
"src/shrank.ts": 700,
"src/removed.ts": 700,
},
baseline: {
"src/grew.ts": 701,
"src/shrank.ts": 650,
"src/new.ts": 501,
},
});
expect(violations).toEqual([
{ filePath: "src/grew.ts", lines: 701, baselineLines: 700, reason: "grew" },
{ filePath: "src/new.ts", lines: 501, reason: "baseline-missing" },
]);
});
});