refactor(adapters): wire executor and pi-runtime splits via imports
Remove duplicate definitions from executor.ts and pi-runtime.ts that already live in executor-*.ts and pi-model-config.ts. Import extracted symbols with .js extensions and re-export the public executor/pi-runtime surface for existing callers. Export a few pi-model-config helpers required by pi-runtime (including boundAgentToolResult and toolCallBudgetFor) and align beginToolCall queue typing with AgentRuntimeEvent. Co-authored-by: cynarAI <cynarAI@users.noreply.github.com>
This commit is contained in:
parent
9e0850d346
commit
cae230a98f
@ -100,7 +100,6 @@ import {
|
||||
type ThreadEvents,
|
||||
} from "@rakazo/db";
|
||||
import { getLogger } from "@rakazo/logging";
|
||||
import { parse as parseShellCommand } from "shell-quote";
|
||||
import {
|
||||
connectAgent,
|
||||
messageConnectedAgent,
|
||||
@ -139,7 +138,20 @@ import {
|
||||
settleUncertainEffect,
|
||||
uncertainEffectResult,
|
||||
} from "./approval-effect.js";
|
||||
import { recordEffect, completeEffect, uncertainEffectError } from "./executor-effects.js";
|
||||
import {
|
||||
TURN_ATTACHMENT_UNAVAILABLE,
|
||||
STEERING_ATTACHMENT_UNAVAILABLE,
|
||||
missingTurnImagesInstruction,
|
||||
settleSteeringAttachmentLoads,
|
||||
} from "./executor-attachments.js";
|
||||
import { createRunWorkspaceCheckpoint } from "./executor-checkpoint.js";
|
||||
import {
|
||||
completionMessageSegments,
|
||||
completionNotificationBody,
|
||||
completionMarksUnread,
|
||||
threadContextForRun,
|
||||
subagentMarksUnread,
|
||||
} from "./executor-completion.js";
|
||||
import {
|
||||
computerScreenToolResult,
|
||||
notifyRun,
|
||||
@ -153,7 +165,16 @@ import {
|
||||
persistMessageInTransaction,
|
||||
runNotificationsEnabled,
|
||||
} from "./executor-computer.js";
|
||||
|
||||
import { recordEffect, completeEffect, uncertainEffectError } from "./executor-effects.js";
|
||||
import { deploymentKeyFor, resolveModelKey, withModelCredentialLock } from "./executor-model.js";
|
||||
import { isProtectedComputerLifecycleCommand } from "./executor-shell-guard.js";
|
||||
import {
|
||||
PAGE_BROWSER_TOOL_NAMES,
|
||||
filterPageBrowserTools,
|
||||
selectBuiltinToolsForRun,
|
||||
} from "./executor-tools.js";
|
||||
import { type ExecutorDeps } from "./executor-types.js";
|
||||
import { selectRunConnections, loadCurrentTurnImages } from "./executor-utils.js";
|
||||
import {
|
||||
autoReviewTimeoutMs,
|
||||
buildAutoReviewPrompt,
|
||||
@ -338,7 +359,6 @@ import {
|
||||
import { createWebProvider } from "./web-provider-factory.js";
|
||||
import { webFetchFromTool, webSearchFromTool } from "./web-tools.js";
|
||||
|
||||
const modelCredentialLocks = new Map<string, Promise<void>>();
|
||||
const READ_ONLY_AGENT_TOOLS = new Set([
|
||||
"computer_observe",
|
||||
"list_files",
|
||||
@ -356,204 +376,7 @@ const READ_ONLY_AGENT_TOOLS = new Set([
|
||||
"cloud_agent_status",
|
||||
]);
|
||||
const MAX_MODEL_FILE_BYTES = 250_000;
|
||||
const TURN_ATTACHMENT_UNAVAILABLE =
|
||||
"An attachment in this message could not be loaded. Tell the user the attachment was unavailable and do not guess its contents.";
|
||||
const STEERING_ATTACHMENT_UNAVAILABLE = TURN_ATTACHMENT_UNAVAILABLE;
|
||||
const BUILTIN_AGENT_TOOL_NAMES = new Set(builtinAgentTools.map((tool) => tool.name));
|
||||
|
||||
/** Avoid an expensive remote workspace export when a turn never touched the computer. */
|
||||
export function createRunWorkspaceCheckpoint(checkpoint: () => Promise<unknown>) {
|
||||
let dirty = false;
|
||||
return {
|
||||
markDirty() {
|
||||
dirty = true;
|
||||
},
|
||||
markFiles(files: readonly unknown[]) {
|
||||
if (files.length > 0) dirty = true;
|
||||
},
|
||||
async flush() {
|
||||
if (!dirty) return false;
|
||||
dirty = false;
|
||||
try {
|
||||
await checkpoint();
|
||||
return true;
|
||||
} catch (error) {
|
||||
dirty = true;
|
||||
throw error;
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
const SHELL_INTERPRETER_NAMES = /^(?:bash|sh|dash|zsh|ksh|fish)$/;
|
||||
const STATIC_SHELL_EXPANSIONS: Readonly<Record<string, string>> = {
|
||||
HOME: "/home/rakazo",
|
||||
LOGNAME: "rakazo",
|
||||
PATH: "/home/rakazo/.local/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
|
||||
PWD: "/home/rakazo",
|
||||
TMPDIR: "/tmp",
|
||||
USER: "rakazo",
|
||||
WORKSPACE: "/home/rakazo/workspace",
|
||||
XDG_CONFIG_HOME: "/home/rakazo/.config",
|
||||
};
|
||||
const SAFE_SHELL_CONTROL_OPS = new Set([
|
||||
"&&",
|
||||
"||",
|
||||
";",
|
||||
"|",
|
||||
"&",
|
||||
">",
|
||||
"<",
|
||||
">>",
|
||||
">&",
|
||||
"<&",
|
||||
"&>",
|
||||
]);
|
||||
|
||||
function shellCFlagProgram(words: string[], interpreterIndex: number): string | undefined {
|
||||
for (let index = interpreterIndex + 1; index < words.length; index += 1) {
|
||||
const word = words[index] ?? "";
|
||||
if (word.startsWith("--command=")) return word.slice("--command=".length);
|
||||
// bash -c / -lc / -ce and fish --command: the next argument is the program string.
|
||||
if (word === "--command" || /^-[^-]*c/.test(word)) return words[index + 1];
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
function preserveShellCommandBoundaries(command: string): string {
|
||||
let quote: "'" | '"' | undefined;
|
||||
let result = "";
|
||||
for (let index = 0; index < command.length; index += 1) {
|
||||
const character = command[index];
|
||||
const next = command[index + 1];
|
||||
if (character === "\\" && quote !== "'") {
|
||||
if (next === "\n") {
|
||||
index += 1;
|
||||
continue;
|
||||
}
|
||||
// Keep escaped characters intact; an escaped quote is not a boundary.
|
||||
result += character;
|
||||
if (next !== undefined) {
|
||||
result += next;
|
||||
index += 1;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (character === quote) quote = undefined;
|
||||
else if (!quote && (character === "'" || character === '"')) quote = character;
|
||||
result += character === "\n" && !quote ? "\n;" : character;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
function tokenizeProtectedShellCommand(command: string): string[] | "dynamic" {
|
||||
try {
|
||||
// shell-quote treats newlines as whitespace. Preserve command boundaries for
|
||||
// the dot builtin, after folding shell line continuations. Retaining the
|
||||
// newline also preserves comment handling (comments remain fail-closed).
|
||||
const separated = preserveShellCommandBoundaries(command);
|
||||
const parsed = parseShellCommand<{ expansion: string }>(
|
||||
separated,
|
||||
(name) => STATIC_SHELL_EXPANSIONS[name] ?? { expansion: name },
|
||||
{ splitUnquoted: true },
|
||||
);
|
||||
const words: string[] = [];
|
||||
let commandPosition = true;
|
||||
let redirectTarget = false;
|
||||
for (const [index, entry] of parsed.entries()) {
|
||||
if (typeof entry === "string") {
|
||||
// Backtick fragments are not fully tokenized; treat them as dynamic.
|
||||
if (entry.includes("`")) return "dynamic";
|
||||
const word = entry.toLowerCase();
|
||||
// `find .`, `git add .`, and `git -C .` use a path, not the
|
||||
// executable `. script` builtin. Keep the path out of the builtin scan.
|
||||
words.push(word === "." && (!commandPosition || redirectTarget) ? "./" : word);
|
||||
if (redirectTarget) {
|
||||
redirectTarget = false;
|
||||
continue;
|
||||
}
|
||||
const next = parsed[index + 1];
|
||||
if (
|
||||
commandPosition &&
|
||||
/^\d+$/.test(word) &&
|
||||
typeof next === "object" &&
|
||||
"op" in next &&
|
||||
/^[<>]/.test(next.op)
|
||||
) {
|
||||
// A leading file descriptor belongs to a redirect, not the command.
|
||||
} else if (commandPosition && /^(?:then|do|else)$/.test(word)) {
|
||||
commandPosition = true;
|
||||
} else if (commandPosition && (word === "coproc" || word === "function")) return "dynamic";
|
||||
else if (
|
||||
commandPosition &&
|
||||
(/^(?:command|builtin|exec|time|if|elif|while|until|!|\{)$/.test(word) ||
|
||||
word.startsWith("-") ||
|
||||
/^[a-z_][a-z0-9_]*=/.test(word))
|
||||
) {
|
||||
// Shell prefixes and assignments leave the command word pending.
|
||||
} else commandPosition = false;
|
||||
continue;
|
||||
}
|
||||
if ("expansion" in entry) {
|
||||
// Unknown expansions and command substitutions are resolved by bash
|
||||
// after this guard runs, so their eventual value cannot be inspected.
|
||||
return "dynamic";
|
||||
}
|
||||
if ("op" in entry && entry.op === "glob") {
|
||||
words.push(entry.pattern.toLowerCase());
|
||||
continue;
|
||||
}
|
||||
if ("op" in entry && SAFE_SHELL_CONTROL_OPS.has(entry.op)) {
|
||||
if (["&&", "||", ";", "|", "&"].includes(entry.op)) {
|
||||
commandPosition = true;
|
||||
redirectTarget = false;
|
||||
} else redirectTarget = true;
|
||||
continue;
|
||||
}
|
||||
return "dynamic";
|
||||
}
|
||||
return words;
|
||||
} catch {
|
||||
return "dynamic";
|
||||
}
|
||||
}
|
||||
|
||||
export function isProtectedComputerLifecycleCommand(command: string): boolean {
|
||||
const words = tokenizeProtectedShellCommand(command);
|
||||
if (words === "dynamic") return true;
|
||||
|
||||
const commandNames = words.map((word) => word.split("/").at(-1));
|
||||
if (commandNames.some((word) => /^(?:kill|pkill|killall|xkill)$/.test(word ?? ""))) {
|
||||
return true;
|
||||
}
|
||||
// eval/source/. can hide protected commands inside an expansion string that the
|
||||
// outer tokenizer keeps as a single word (e.g. eval "pkill chromium").
|
||||
if (words.includes(".") || commandNames.some((word) => /^(?:eval|source)$/.test(word ?? ""))) {
|
||||
return true;
|
||||
}
|
||||
if (
|
||||
commandNames.some((word) => word === "systemctl" || word === "service") &&
|
||||
words.some((word) => /^(?:stop|restart|kill)$/.test(word))
|
||||
) {
|
||||
return true;
|
||||
}
|
||||
if (
|
||||
words.some((word) =>
|
||||
/(?:\.browser-profiles|--user-data-dir|\/tmp\/\.x11-unix|\/tmp\/\.x\d+-lock)/.test(word),
|
||||
)
|
||||
) {
|
||||
return true;
|
||||
}
|
||||
|
||||
for (let index = 0; index < words.length; index += 1) {
|
||||
const name = words[index]?.split("/").at(-1) ?? "";
|
||||
if (!SHELL_INTERPRETER_NAMES.test(name)) continue;
|
||||
const program = shellCFlagProgram(words, index);
|
||||
if (program && isProtectedComputerLifecycleCommand(program)) return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/** Cap the roster so a large Space cannot flood the prompt. */
|
||||
const BOT_DIRECTORY_LIMIT = 40;
|
||||
const MISSING_MODEL_MESSAGE = "Connect a model in Settings before running bots.";
|
||||
@ -562,37 +385,6 @@ function runtimeFallbackModel(runtime: AgentRuntime) {
|
||||
return runtime.describe().capabilities.scripted ? { provider: "scripted", id: "scripted" } : null;
|
||||
}
|
||||
|
||||
export interface ExecutorDeps {
|
||||
prisma: PrismaClient;
|
||||
events: ThreadEvents;
|
||||
runtime: AgentRuntime;
|
||||
sandbox: SandboxProvider;
|
||||
memory: MemoryStore;
|
||||
memoryProviders: MemoryProviderResolver;
|
||||
home: AgentHomeStore;
|
||||
artifacts?: ArtifactStore;
|
||||
connector?: ConnectorProvider;
|
||||
connectors?: { managed(id: string): ManagedConnectorProvider | undefined };
|
||||
secrets: string[];
|
||||
secretStore: EncryptedSecretStore;
|
||||
deploymentModelKey?: string;
|
||||
dataDir?: string;
|
||||
notifications?: NotificationProvider;
|
||||
jobs: JobPublisher;
|
||||
/** Messaging surface; absent means zero identity queries and no chat prompts. */
|
||||
messaging?: { hasIdentity(botId: string): Promise<boolean> };
|
||||
listConnectedPluginSlugs?: (userId: string) => Promise<string[]>;
|
||||
/** Builtin web_search / web_fetch. Defaults to keyless HTTP when omitted. */
|
||||
web?: WebProvider;
|
||||
/** Page browser (DOM refs) on the bot computer. Defaults to the sandbox live browser when supported. */
|
||||
browser?: BrowserProvider;
|
||||
secretHttp?: RemoteTransportDependencies;
|
||||
/** Remote cloud coding agents. Null/omit means tools stay uninjected. */
|
||||
cloudAgent?: CloudAgentConnection | null;
|
||||
/** Aborted when createApp stop() begins so in-flight continueRun boot waits exit promptly. */
|
||||
shutdownSignal?: AbortSignal;
|
||||
}
|
||||
|
||||
function isAuditableToolResult(value: unknown): value is {
|
||||
kind: "agent_tool_result";
|
||||
content: unknown[];
|
||||
@ -4391,53 +4183,6 @@ export function createRunExecutor(deps: ExecutorDeps) {
|
||||
};
|
||||
}
|
||||
|
||||
export function selectBuiltinToolsForRun(options: {
|
||||
graphicalToolsAllowed: boolean;
|
||||
/** Page browser tools need a graphical computer (Chrome), not model vision. */
|
||||
pageBrowserAllowed?: boolean;
|
||||
groupId: string | null;
|
||||
trigger: string;
|
||||
semanticMemoryEnabled: boolean;
|
||||
cloudAgentEnabled?: boolean;
|
||||
messagingChannelRun: boolean;
|
||||
}) {
|
||||
return selectCloudAgentTools(
|
||||
selectMemoryTools(
|
||||
filterBuiltinToolsForRun(
|
||||
filterBuiltinToolsForThread(
|
||||
filterPageBrowserTools(
|
||||
filterImageReturningComputerTools(builtinAgentTools, options.graphicalToolsAllowed),
|
||||
options.pageBrowserAllowed ?? options.graphicalToolsAllowed,
|
||||
),
|
||||
options.groupId,
|
||||
),
|
||||
options.trigger,
|
||||
),
|
||||
options.semanticMemoryEnabled,
|
||||
),
|
||||
Boolean(options.cloudAgentEnabled),
|
||||
).filter(
|
||||
(tool) =>
|
||||
!options.messagingChannelRun ||
|
||||
(!["remember", "save_memory", "recall_memory", "forget_memory"].includes(tool.name) &&
|
||||
!tool.name.startsWith("scratchpad_")),
|
||||
);
|
||||
}
|
||||
|
||||
export const PAGE_BROWSER_TOOL_NAMES = new Set([
|
||||
"browser_navigate",
|
||||
"browser_snapshot",
|
||||
"browser_act",
|
||||
]);
|
||||
|
||||
export function filterPageBrowserTools<T extends { name: string }>(
|
||||
tools: T[],
|
||||
pageBrowserAllowed: boolean,
|
||||
): T[] {
|
||||
if (pageBrowserAllowed) return tools;
|
||||
return tools.filter((tool) => !PAGE_BROWSER_TOOL_NAMES.has(tool.name));
|
||||
}
|
||||
|
||||
async function runSandboxCommand(
|
||||
sandbox: SandboxProvider,
|
||||
computer: ComputerRef,
|
||||
@ -4474,224 +4219,12 @@ async function runSandboxCommand(
|
||||
return { stdout, stderr, code };
|
||||
}
|
||||
|
||||
/**
|
||||
* The deployment key is a bearer credential for exactly one vendor, so it is handed out
|
||||
* only when the provider that won the resolution above is that vendor. A provider named
|
||||
* by deployment settings or a bot override gets no key rather than another vendor's.
|
||||
*/
|
||||
function deploymentKeyFor(deps: ExecutorDeps, provider: string): string | undefined {
|
||||
if (!deps.deploymentModelKey) return undefined;
|
||||
return provider === resolveDeploymentModel().provider ? deps.deploymentModelKey : undefined;
|
||||
}
|
||||
|
||||
async function resolveModelKey(
|
||||
deps: ExecutorDeps,
|
||||
userId: string,
|
||||
spaceId: string,
|
||||
credential: {
|
||||
secretId: string;
|
||||
provider: string;
|
||||
defaultModel?: string | null;
|
||||
supportsImages?: boolean;
|
||||
} | null,
|
||||
provider: string,
|
||||
modelId: string,
|
||||
registerSecrets?: (values: string[]) => void,
|
||||
): Promise<{
|
||||
apiKey?: string;
|
||||
baseUrl?: string;
|
||||
reasoning?: boolean;
|
||||
maxTokens?: number;
|
||||
contextWindow?: number;
|
||||
thinkingLevel?: AgentRunRequest["model"]["thinkingLevel"];
|
||||
acceptsImages?: boolean;
|
||||
maxImagesPerPrompt?: number;
|
||||
oauth?: AgentModelOAuthCredential;
|
||||
persistOAuth?: (credential: AgentModelOAuthCredential) => Promise<void>;
|
||||
redact: string[];
|
||||
}> {
|
||||
if (credential) {
|
||||
return withModelCredentialLock(credential.secretId, async () => {
|
||||
const row = await deps.prisma.secret.findFirst({
|
||||
where: { id: credential.secretId, userId, spaceId: null },
|
||||
});
|
||||
if (!row) return { apiKey: deploymentKeyFor(deps, provider), redact: [] };
|
||||
const plaintext = deps.secretStore.load(row.ciphertext, row.id);
|
||||
registerSecrets?.(secretValuesToRedact(parseModelSecret(plaintext)));
|
||||
const persist = async (next: string) => {
|
||||
const stored = await deps.secretStore.put(
|
||||
next,
|
||||
{
|
||||
operationId: "cred",
|
||||
traceId: "cred-refresh",
|
||||
spaceId,
|
||||
userId,
|
||||
signal: new AbortController().signal,
|
||||
},
|
||||
row.id,
|
||||
);
|
||||
await deps.prisma.secret.update({
|
||||
where: { id: row.id },
|
||||
data: { ciphertext: stored.ciphertext },
|
||||
});
|
||||
};
|
||||
const resolved = await resolveModelAuth(plaintext, credential.provider, {
|
||||
persist,
|
||||
});
|
||||
const oauth = resolved.secret.kind === "oauth" ? resolved.secret.credential : undefined;
|
||||
const baseUrl =
|
||||
resolved.secret.kind === "openai_compatible" ? resolved.secret.baseUrl : undefined;
|
||||
const acceptsImages =
|
||||
credential.provider === OPENAI_COMPATIBLE_PROVIDER_ID &&
|
||||
resolved.secret.kind === "openai_compatible" &&
|
||||
(modelIdSupportsImages(resolved.secret.visionModelIds, modelId) ||
|
||||
// Legacy secrets have no per-model list, so keep their existing
|
||||
// capability scoped to the model saved in the space preference.
|
||||
(resolved.secret.visionModelIds === undefined &&
|
||||
credential.supportsImages === true &&
|
||||
credential.defaultModel?.trim() === modelId.trim()));
|
||||
return {
|
||||
apiKey: resolved.apiKey,
|
||||
baseUrl,
|
||||
reasoning:
|
||||
resolved.secret.kind === "openai_compatible" ? resolved.secret.reasoning : undefined,
|
||||
maxTokens:
|
||||
resolved.secret.kind === "openai_compatible" ? resolved.secret.maxTokens : undefined,
|
||||
contextWindow:
|
||||
resolved.secret.kind === "openai_compatible" ? resolved.secret.contextWindow : undefined,
|
||||
thinkingLevel:
|
||||
resolved.secret.kind === "openai_compatible" ? resolved.secret.thinkingLevel : undefined,
|
||||
acceptsImages,
|
||||
maxImagesPerPrompt:
|
||||
resolved.secret.kind === "openai_compatible"
|
||||
? resolved.secret.maxImagesPerPrompt
|
||||
: undefined,
|
||||
oauth,
|
||||
persistOAuth: oauth
|
||||
? async (next) => {
|
||||
await withModelCredentialLock(credential.secretId, async () => {
|
||||
const currentRow = await deps.prisma.secret.findFirst({
|
||||
where: { id: credential.secretId, userId, spaceId: null },
|
||||
});
|
||||
if (!currentRow) return;
|
||||
const current = parseModelSecret(
|
||||
deps.secretStore.load(currentRow.ciphertext, currentRow.id),
|
||||
);
|
||||
if (current.kind === "oauth") {
|
||||
const stored = current.credential;
|
||||
if (stored.expires > next.expires) return;
|
||||
if (
|
||||
stored.access === next.access &&
|
||||
stored.refresh === next.refresh &&
|
||||
stored.expires === next.expires
|
||||
) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
await persist(
|
||||
serializeModelSecret({ kind: "oauth", credential: toOAuthCredential(next) }),
|
||||
);
|
||||
});
|
||||
}
|
||||
: undefined,
|
||||
redact: [...secretValuesToRedact(resolved.secret), resolved.apiKey].filter(
|
||||
(value): value is string => Boolean(value),
|
||||
),
|
||||
};
|
||||
});
|
||||
}
|
||||
return { apiKey: deploymentKeyFor(deps, provider), redact: [] };
|
||||
}
|
||||
|
||||
async function withModelCredentialLock<T>(key: string, fn: () => Promise<T>): Promise<T> {
|
||||
const previous = modelCredentialLocks.get(key) ?? Promise.resolve();
|
||||
let release!: () => void;
|
||||
const current = previous.then(
|
||||
() =>
|
||||
new Promise<void>((resolve) => {
|
||||
release = resolve;
|
||||
}),
|
||||
);
|
||||
modelCredentialLocks.set(key, current);
|
||||
await previous;
|
||||
try {
|
||||
return await fn();
|
||||
} finally {
|
||||
release();
|
||||
if (modelCredentialLocks.get(key) === current) modelCredentialLocks.delete(key);
|
||||
}
|
||||
}
|
||||
|
||||
export function selectRunConnections<
|
||||
T extends { connectorId: string; provider: string; status: string },
|
||||
>(rows: T[], connectedComposioProviders: string[]): T[] {
|
||||
const activeKeys = new Set(connectedComposioProviders.map((provider) => `composio:${provider}`));
|
||||
return rows.filter(
|
||||
(row) =>
|
||||
row.status !== "revoked" &&
|
||||
(row.status === "connected" || activeKeys.has(`${row.connectorId}:${row.provider}`)),
|
||||
);
|
||||
}
|
||||
|
||||
export async function loadCurrentTurnImages(
|
||||
deps: ExecutorDeps,
|
||||
blocks: MessageBlock[] | undefined,
|
||||
context: {
|
||||
operationId: string;
|
||||
traceId: string;
|
||||
spaceId: string;
|
||||
userId: string;
|
||||
botId: string;
|
||||
runId: string;
|
||||
signal: AbortSignal;
|
||||
},
|
||||
) {
|
||||
if (!deps.artifacts || !blocks?.length) return undefined;
|
||||
const imageBlocks = blocks.filter(
|
||||
(block): block is Extract<MessageBlock, { kind: "image" }> => block.kind === "image",
|
||||
);
|
||||
if (!imageBlocks.length) return undefined;
|
||||
|
||||
const rows = await deps.prisma.artifact.findMany({
|
||||
where: {
|
||||
id: { in: imageBlocks.map((block) => block.artifactId) },
|
||||
spaceId: context.spaceId,
|
||||
userId: context.userId,
|
||||
},
|
||||
});
|
||||
const byId = new Map(rows.map((row) => [row.id, row]));
|
||||
const images: NonNullable<AgentRunRequest["currentTurnImages"]> = [];
|
||||
|
||||
for (const block of imageBlocks) {
|
||||
const row = byId.get(block.artifactId);
|
||||
if (!row || !isAttachmentImageMimeType(block.mimeType)) continue;
|
||||
try {
|
||||
const bytes = await deps.artifacts.get(row.storageKey, context);
|
||||
images.push({
|
||||
name: block.name,
|
||||
mimeType: block.mimeType as "image/jpeg" | "image/png" | "image/webp" | "image/gif",
|
||||
data: bytes,
|
||||
});
|
||||
} catch (error) {
|
||||
if (context.signal.aborted) throw error;
|
||||
}
|
||||
}
|
||||
|
||||
return images.length ? images : undefined;
|
||||
}
|
||||
|
||||
// Re-export computer/lifecycle utilities from executor-computer.ts
|
||||
export {
|
||||
computerScreenToolResult,
|
||||
notifyRun,
|
||||
renewRunLease,
|
||||
computerRetryDelay,
|
||||
computerRunRequeueData,
|
||||
writeComputerRunRequeue,
|
||||
requeueComputerRun,
|
||||
redactBlocks,
|
||||
publishMessage,
|
||||
runNotificationsEnabled,
|
||||
} from "./executor-computer.js";
|
||||
// Re-export effect utilities from executor-effects.ts
|
||||
export { TURN_ATTACHMENT_UNAVAILABLE, STEERING_ATTACHMENT_UNAVAILABLE, missingTurnImagesInstruction, settleSteeringAttachmentLoads } from "./executor-attachments.js";
|
||||
export { createRunWorkspaceCheckpoint } from "./executor-checkpoint.js";
|
||||
export { isProtectedComputerLifecycleCommand } from "./executor-shell-guard.js";
|
||||
export { type ExecutorDeps } from "./executor-types.js";
|
||||
export { completionMessageSegments, completionNotificationBody, completionMarksUnread, threadContextForRun, subagentMarksUnread } from "./executor-completion.js";
|
||||
export { selectRunConnections, loadCurrentTurnImages } from "./executor-utils.js";
|
||||
export { PAGE_BROWSER_TOOL_NAMES, filterPageBrowserTools, selectBuiltinToolsForRun } from "./executor-tools.js";
|
||||
export { recordEffect, completeEffect, uncertainEffectError } from "./executor-effects.js";
|
||||
export { computerScreenToolResult, notifyRun, renewRunLease, computerRetryDelay, computerRunRequeueData, writeComputerRunRequeue, requeueComputerRun, redactBlocks, publishMessage, persistMessageInTransaction, runNotificationsEnabled } from "./executor-computer.js";
|
||||
|
||||
@ -78,7 +78,7 @@ export function maxToolCallsPerTurn(env: NodeJS.ProcessEnv = process.env): numbe
|
||||
return Math.floor(parsed);
|
||||
}
|
||||
|
||||
function toolCallBudgetFor(runId: string): ToolCallBudget {
|
||||
export function toolCallBudgetFor(runId: string): ToolCallBudget {
|
||||
const existing = toolCallBudgetsByRun.get(runId);
|
||||
if (existing) {
|
||||
existing.inFlight = 0;
|
||||
@ -99,7 +99,7 @@ export function releaseToolCallBudget(runId: string, keepForResume: boolean) {
|
||||
if (!keepForResume) toolCallBudgetsByRun.delete(runId);
|
||||
}
|
||||
|
||||
function maybeAbortToolCallBudget(host: {
|
||||
export function maybeAbortToolCallBudget(host: {
|
||||
toolCallBudget: ToolCallBudget;
|
||||
abortTurn(): void;
|
||||
}): void {
|
||||
@ -111,7 +111,7 @@ function maybeAbortToolCallBudget(host: {
|
||||
export function beginToolCall(host: {
|
||||
toolCallBudget: ToolCallBudget;
|
||||
abortTurn(): void;
|
||||
queue: { push(event: { type: string; text: string }): void };
|
||||
queue: { push(event: AgentRuntimeEvent): void };
|
||||
}): boolean {
|
||||
const budget = host.toolCallBudget;
|
||||
if (budget.limit <= 0) {
|
||||
@ -152,7 +152,7 @@ export function toolCallBudgetExceededMessage(limit: number): string {
|
||||
|
||||
// --- Model resolution ---
|
||||
|
||||
function configuredOpenRouterModel(id: string): Model<"openai-completions"> {
|
||||
export function configuredOpenRouterModel(id: string): Model<"openai-completions"> {
|
||||
return {
|
||||
id,
|
||||
name: id,
|
||||
@ -204,7 +204,7 @@ export function modelsForRequest(
|
||||
return catalogModels();
|
||||
}
|
||||
|
||||
function resolveRuntimeModel(modelConfig: AgentRunRequest["model"]): {
|
||||
export function resolveRuntimeModel(modelConfig: AgentRunRequest["model"]): {
|
||||
provider: string;
|
||||
modelId: string;
|
||||
models: Models;
|
||||
@ -242,7 +242,7 @@ function resolveRuntimeModel(modelConfig: AgentRunRequest["model"]): {
|
||||
|
||||
// --- Image helpers ---
|
||||
|
||||
function toPiImages(images: AgentRunRequest["currentTurnImages"]) {
|
||||
export function toPiImages(images: AgentRunRequest["currentTurnImages"]) {
|
||||
return (images ?? []).map((image) => ({
|
||||
type: "image" as const,
|
||||
data: Buffer.from(image.data).toString("base64"),
|
||||
@ -252,7 +252,7 @@ function toPiImages(images: AgentRunRequest["currentTurnImages"]) {
|
||||
|
||||
// --- Steering message removal ---
|
||||
|
||||
function withoutSteeringMessages(
|
||||
export function withoutSteeringMessages(
|
||||
history: AgentRunRequest["history"],
|
||||
steering: AgentSteeringMessage[],
|
||||
): AgentRunRequest["history"] {
|
||||
@ -337,7 +337,7 @@ export function pruneComputerScreenshotContext(
|
||||
return transformed ?? messages;
|
||||
}
|
||||
|
||||
function imagePartCount(message: AgentMessage): number {
|
||||
export function imagePartCount(message: AgentMessage): number {
|
||||
if (!("content" in message) || !Array.isArray(message.content)) return 0;
|
||||
return message.content.filter(
|
||||
(part: unknown) =>
|
||||
@ -345,7 +345,7 @@ function imagePartCount(message: AgentMessage): number {
|
||||
).length;
|
||||
}
|
||||
|
||||
function isComputerScreenshotMessage(
|
||||
export function isComputerScreenshotMessage(
|
||||
message: AgentMessage | undefined,
|
||||
): message is Extract<AgentMessage, { role: "toolResult" }> {
|
||||
if (message?.role !== "toolResult" || !message.content.some((part) => (part as {type?: string}).type === "image")) {
|
||||
@ -362,7 +362,7 @@ function isComputerScreenshotMessage(
|
||||
|
||||
// --- Type guards ---
|
||||
|
||||
function isAgentToolExecutionResult(result: unknown): result is AgentToolExecutionResult {
|
||||
export function isAgentToolExecutionResult(result: unknown): result is AgentToolExecutionResult {
|
||||
if (
|
||||
!result ||
|
||||
typeof result !== "object" ||
|
||||
@ -413,7 +413,7 @@ export function jsonSchemaParameters(
|
||||
: Type.Object(fields);
|
||||
}
|
||||
|
||||
function enumUnion(values: readonly unknown[]) {
|
||||
export function enumUnion(values: readonly unknown[]) {
|
||||
const members = values.map((value) =>
|
||||
value === null
|
||||
? Type.Null()
|
||||
@ -466,7 +466,7 @@ export function jsonField(spec: unknown): ReturnType<typeof Type.String> {
|
||||
|
||||
// --- Result helpers ---
|
||||
|
||||
function summarizeToolResult(result: unknown) {
|
||||
export function summarizeToolResult(result: unknown) {
|
||||
try {
|
||||
const text = JSON.stringify(result);
|
||||
if (!text) return "ok";
|
||||
@ -476,7 +476,7 @@ function summarizeToolResult(result: unknown) {
|
||||
}
|
||||
}
|
||||
|
||||
function boundAgentToolResult<T>(
|
||||
export function boundAgentToolResult<T>(
|
||||
result: AgentToolResult<T>,
|
||||
): AgentToolResult<T> {
|
||||
if (!result || !Array.isArray(result.content)) return result;
|
||||
@ -486,7 +486,7 @@ function boundAgentToolResult<T>(
|
||||
};
|
||||
}
|
||||
|
||||
function assistantText(message: unknown): string {
|
||||
export function assistantText(message: unknown): string {
|
||||
if (!message || typeof message !== "object" || !("content" in message)) return "";
|
||||
const content = (message as { content?: unknown }).content;
|
||||
if (typeof content === "string") return content;
|
||||
@ -506,7 +506,7 @@ function assistantText(message: unknown): string {
|
||||
|
||||
// --- Sanitization ---
|
||||
|
||||
function sanitizeSensitiveText(message: string) {
|
||||
export function sanitizeSensitiveText(message: string) {
|
||||
return message
|
||||
.replace(/sk-or-v1-[a-zA-Z0-9]+/g, "[redacted]")
|
||||
.replace(/sk-[a-zA-Z0-9-]+/g, "[redacted]")
|
||||
@ -526,7 +526,7 @@ function sanitizeSensitiveText(message: string) {
|
||||
);
|
||||
}
|
||||
|
||||
function redactActivityUrl(value: unknown): string {
|
||||
export function redactActivityUrl(value: unknown): string {
|
||||
const raw = String(value ?? "").trim();
|
||||
if (!raw) return raw;
|
||||
try {
|
||||
@ -541,7 +541,7 @@ function redactActivityUrl(value: unknown): string {
|
||||
}
|
||||
}
|
||||
|
||||
function sanitizeError(message: string) {
|
||||
export function sanitizeError(message: string) {
|
||||
return sanitizeSensitiveText(message);
|
||||
}
|
||||
|
||||
@ -562,7 +562,7 @@ export function isOpenCodeProvider(provider: string): boolean {
|
||||
const OPENCODE_SESSION_ERROR =
|
||||
"OpenCode rejected this chat session. Send the message again.";
|
||||
|
||||
function looksLikeOpenCodeSessionError(message: string): boolean {
|
||||
export function looksLikeOpenCodeSessionError(message: string): boolean {
|
||||
return (
|
||||
/x-opencode-session/i.test(message) ||
|
||||
/session\s*(id|header|required|missing|invalid|expired|stale)/i.test(message) ||
|
||||
@ -570,7 +570,7 @@ function looksLikeOpenCodeSessionError(message: string): boolean {
|
||||
);
|
||||
}
|
||||
|
||||
function sanitizeProviderError(provider: string, message: string): string {
|
||||
export function sanitizeProviderError(provider: string, message: string): string {
|
||||
const sanitized = sanitizeError(message);
|
||||
if (isOpenCodeProvider(provider) && looksLikeOpenCodeSessionError(sanitized)) {
|
||||
return OPENCODE_SESSION_ERROR;
|
||||
@ -580,7 +580,7 @@ function sanitizeProviderError(provider: string, message: string): string {
|
||||
|
||||
// --- Queue & gate ---
|
||||
|
||||
interface EventQueue {
|
||||
export interface EventQueue {
|
||||
push(event: AgentRuntimeEvent): void;
|
||||
fail(error: Error): void;
|
||||
close(): void;
|
||||
|
||||
@ -5,73 +5,65 @@ import {
|
||||
type AgentTool,
|
||||
type AgentToolResult,
|
||||
} from "@earendil-works/pi-agent-core";
|
||||
import {
|
||||
type Api,
|
||||
clampThinkingLevel,
|
||||
type Model,
|
||||
type Models,
|
||||
type ModelThinkingLevel,
|
||||
type SimpleStreamOptions,
|
||||
Type,
|
||||
} from "@earendil-works/pi-ai";
|
||||
import { builtinModels } from "@earendil-works/pi-ai/providers/all";
|
||||
import { type Api, type Model, Type } from "@earendil-works/pi-ai";
|
||||
import type {
|
||||
AdapterContext,
|
||||
AgentRunRequest,
|
||||
AgentRuntime,
|
||||
AgentRuntimeEvent,
|
||||
AgentSteeringMessage,
|
||||
AgentToolCompletion,
|
||||
AgentToolExecutionResult,
|
||||
ConnectorTool,
|
||||
} from "@rakazo/adapter-kit";
|
||||
import { getLogger } from "@rakazo/logging";
|
||||
import { isToolPauseResult } from "./approval-effect.js";
|
||||
import { builtinAgentTools, DELEGATION_TOOL_NAMES } from "./builtin-tools.js";
|
||||
import { DEFAULT_OPENROUTER_MODEL_ID } from "./deployment-model.js";
|
||||
import {
|
||||
normalizeOpenAiToolParameters,
|
||||
openAiToolParametersNeedNormalization,
|
||||
} from "./openai-tool-parameters.js";
|
||||
import { PiRuntimeCredentialStore, toOAuthCredential } from "./pi-credentials.js";
|
||||
import { registerLocalProvider } from "./pi-local-provider.js";
|
||||
import {
|
||||
OPENAI_COMPATIBLE_PROVIDER_ID,
|
||||
registerOpenAiCompatibleCatalog,
|
||||
registerOpenAiCompatibleRuntime,
|
||||
} from "./pi-openai-compatible-provider.js";
|
||||
import {
|
||||
billedPromptTokens,
|
||||
clipToolResultContent,
|
||||
clipToolResultText,
|
||||
MODEL_STREAM_MAX_RETRIES,
|
||||
MODEL_STREAM_TIMEOUT_MS,
|
||||
resolveCompletionMaxTokens,
|
||||
} from "./pi-runtime-limits.js";
|
||||
import { billedPromptTokens, clipToolResultText } from "./pi-runtime-limits.js";
|
||||
import {
|
||||
PiJsonlSessionRecorder,
|
||||
type PiSessionHandle,
|
||||
type PiSessionRecorder,
|
||||
} from "./pi-session.js";
|
||||
import { textContentArg } from "./tool-text.js";
|
||||
import {
|
||||
assistantText,
|
||||
beginToolCall,
|
||||
boundAgentToolResult,
|
||||
conversationSessionId,
|
||||
createGate,
|
||||
createQueue,
|
||||
endToolCall,
|
||||
isAgentToolExecutionResult,
|
||||
jsonSchemaParameters,
|
||||
looksLikeOpenCodeSessionError,
|
||||
modelForCompletion,
|
||||
prepareRequestSecretArguments,
|
||||
pruneComputerScreenshotContext,
|
||||
redactActivityUrl,
|
||||
reliableStreamOptions,
|
||||
releaseToolCallBudget,
|
||||
resolveRuntimeModel,
|
||||
sanitizeError,
|
||||
sanitizeProviderError,
|
||||
sanitizeSensitiveText,
|
||||
summarizeToolResult,
|
||||
thinkingLevelFor,
|
||||
toPiImages,
|
||||
toolCallBudgetExceededMessage,
|
||||
toolCallBudgetFor,
|
||||
withoutSteeringMessages,
|
||||
type ToolCallBudget,
|
||||
type ToolHost,
|
||||
} from "./pi-model-config.js";
|
||||
|
||||
const running = new Map<string, { controller: AbortController; work: Promise<void> }>();
|
||||
interface ToolCallBudget {
|
||||
count: number;
|
||||
exceeded: boolean;
|
||||
limit: number;
|
||||
inFlight: number;
|
||||
}
|
||||
// Optional fuse is process-local. continueRun on another worker starts at zero.
|
||||
const toolCallBudgetsByRun = new Map<string, ToolCallBudget>();
|
||||
// Built on first use, not at module load: entry points call loadRootEnv() after
|
||||
// their imports, and ESM hoists those imports, so module-level env reads here
|
||||
// would run before .env is loaded and miss the local provider entirely.
|
||||
let catalogModelsCache: Models | undefined;
|
||||
function catalogModels(): Models {
|
||||
catalogModelsCache ??= registerOpenAiCompatibleCatalog(registerLocalProvider(builtinModels()));
|
||||
return catalogModelsCache;
|
||||
}
|
||||
const MAX_PARALLEL_SUBAGENTS = 4;
|
||||
// Some OpenAI-compatible models return EOS immediately after a tool result
|
||||
// instead of taking another assistant turn. A bounded internal follow-up keeps
|
||||
@ -81,35 +73,12 @@ const SILENT_TOOL_CONTINUATION_PROMPT =
|
||||
"Continue the original task from the latest tool result. Do not stop after a tool call; use any remaining tools needed, then give the user the final answer.";
|
||||
const TOOL_FINAL_RESPONSE_FALLBACK =
|
||||
"I completed the tool step but could not produce a final response. Please ask me to continue.";
|
||||
const DEFAULT_COMPUTER_SCREENSHOTS_TO_KEEP = 2;
|
||||
// Reasoning-capable models must not start at "off": for OpenRouter, pi-ai maps
|
||||
// that to reasoning.effort "none", which 400s on endpoints that mandate
|
||||
// reasoning (e.g. google/gemini-3.7-flash). Keep a real level when model.reasoning
|
||||
// is set; plain models stay off.
|
||||
const REASONING_MODEL_THINKING_LEVEL: ModelThinkingLevel = "medium";
|
||||
function thinkingLevelFor(
|
||||
model: Model<Api>,
|
||||
preferred?: ModelThinkingLevel | null,
|
||||
): ModelThinkingLevel {
|
||||
if (!model.reasoning) return "off";
|
||||
if (preferred) return clampThinkingLevel(model, preferred);
|
||||
return clampThinkingLevel(model, REASONING_MODEL_THINKING_LEVEL);
|
||||
}
|
||||
// Pi forwards these names to OpenAI Responses, whose function-name contract is
|
||||
// ^[a-zA-Z0-9_-]+$ with a maximum length of 64 characters.
|
||||
const AGENT_TOOL_NAME_PATTERN = /^[a-zA-Z0-9_-]+$/;
|
||||
const MAX_AGENT_TOOL_NAME_LENGTH = 64;
|
||||
const FALLBACK_AGENT_TOOL_NAME = "connector_tool";
|
||||
|
||||
/** Optional self-host fuse. Unset, empty, or 0 means unlimited (default). */
|
||||
export function maxToolCallsPerTurn(env: NodeJS.ProcessEnv = process.env): number {
|
||||
const raw = env.MAX_TOOL_CALLS_PER_TURN?.trim();
|
||||
if (!raw) return 0;
|
||||
const parsed = Number(raw);
|
||||
if (!Number.isFinite(parsed) || parsed <= 0) return 0;
|
||||
return Math.floor(parsed);
|
||||
}
|
||||
|
||||
export interface PiAgentRuntimeOptions {
|
||||
/** Directory where Pi JSONL sessions are written. Omit to disable recording. */
|
||||
sessionRoot?: string;
|
||||
@ -460,108 +429,6 @@ export class PiAgentRuntime implements AgentRuntime {
|
||||
}
|
||||
}
|
||||
|
||||
function toPiImages(images: AgentRunRequest["currentTurnImages"]) {
|
||||
return (images ?? []).map((image) => ({
|
||||
type: "image" as const,
|
||||
data: Buffer.from(image.data).toString("base64"),
|
||||
mimeType: image.mimeType,
|
||||
}));
|
||||
}
|
||||
|
||||
function configuredOpenRouterModel(id: string): Model<"openai-completions"> {
|
||||
// A configured model can intentionally be newer than Pi's static catalog. Keep
|
||||
// pricing conservative, but enable reasoning: unknown OpenRouter endpoints
|
||||
// (e.g. gemini-3.7-flash before the snapshot catches up) often mandate it, and
|
||||
// thinkingLevel "off" becomes effort "none" which those endpoints reject.
|
||||
return {
|
||||
id,
|
||||
name: id,
|
||||
api: "openai-completions",
|
||||
provider: "openrouter",
|
||||
baseUrl: "https://openrouter.ai/api/v1",
|
||||
reasoning: true,
|
||||
input: ["text"],
|
||||
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
|
||||
contextWindow: 16_384,
|
||||
maxTokens: 4_096,
|
||||
};
|
||||
}
|
||||
|
||||
function resolveRuntimeModel(modelConfig: AgentRunRequest["model"]): {
|
||||
provider: string;
|
||||
modelId: string;
|
||||
models: Models;
|
||||
model: Model<Api> | undefined;
|
||||
apiKey: string | undefined;
|
||||
} {
|
||||
const provider = modelConfig.provider === "scripted" ? "openrouter" : modelConfig.provider;
|
||||
const envDefaultModel = process.env.PI_DEFAULT_MODEL?.trim();
|
||||
const envDefaultProvider = process.env.PI_DEFAULT_PROVIDER?.trim() || "openrouter";
|
||||
const modelId =
|
||||
modelConfig.id === "scripted"
|
||||
? envDefaultModel || DEFAULT_OPENROUTER_MODEL_ID
|
||||
: modelConfig.id.trim();
|
||||
const models = modelsForRequest({ model: modelConfig }, provider);
|
||||
let model = models.getModel(provider, modelId);
|
||||
if (!model && provider !== "openrouter" && provider !== OPENAI_COMPATIBLE_PROVIDER_ID) {
|
||||
model = models.getModel("openrouter", modelId);
|
||||
}
|
||||
if (
|
||||
!model &&
|
||||
provider === "openrouter" &&
|
||||
envDefaultProvider === "openrouter" &&
|
||||
modelId === envDefaultModel
|
||||
) {
|
||||
model = configuredOpenRouterModel(modelId);
|
||||
}
|
||||
const apiKey = modelConfig.oauth
|
||||
? undefined
|
||||
: modelConfig.provider === OPENAI_COMPATIBLE_PROVIDER_ID
|
||||
? modelConfig.apiKey || "local"
|
||||
: // Only OpenRouter may fall back to the OpenRouter env key. Handing it to
|
||||
// another provider would ship our key to a vendor it was not issued for.
|
||||
(modelConfig.apiKey ??
|
||||
(provider === "openrouter" ? process.env.OPENROUTER_API_KEY : undefined));
|
||||
return { provider, modelId, models, model, apiKey };
|
||||
}
|
||||
|
||||
export function modelsForRequest(
|
||||
request: Pick<AgentRunRequest, "model">,
|
||||
provider: string,
|
||||
): Models {
|
||||
const oauth = request.model.oauth;
|
||||
if (oauth) {
|
||||
const persist = oauth.persist;
|
||||
return registerOpenAiCompatibleCatalog(
|
||||
registerLocalProvider(
|
||||
builtinModels({
|
||||
credentials: new PiRuntimeCredentialStore(
|
||||
provider,
|
||||
toOAuthCredential(oauth.credential),
|
||||
persist ? (next) => persist(next) : undefined,
|
||||
),
|
||||
}),
|
||||
),
|
||||
);
|
||||
}
|
||||
if (
|
||||
provider === OPENAI_COMPATIBLE_PROVIDER_ID &&
|
||||
request.model.baseUrl &&
|
||||
request.model.id.trim()
|
||||
) {
|
||||
const models = registerOpenAiCompatibleCatalog(registerLocalProvider(builtinModels()));
|
||||
return registerOpenAiCompatibleRuntime(models, {
|
||||
modelId: request.model.id,
|
||||
baseUrl: request.model.baseUrl,
|
||||
reasoning: request.model.reasoning,
|
||||
acceptsImages: request.model.acceptsImages,
|
||||
maxTokens: request.model.maxTokens,
|
||||
contextWindow: request.model.contextWindow,
|
||||
});
|
||||
}
|
||||
return catalogModels();
|
||||
}
|
||||
|
||||
function toAgentTools(toolDefs: readonly ConnectorTool[], host: ToolHost): AgentTool[] {
|
||||
const names = normalizeAgentToolNames(toolDefs);
|
||||
return toolDefs.map((tool, index) => toAgentTool(tool, host, names[index]!));
|
||||
@ -711,52 +578,6 @@ function toHistory(
|
||||
);
|
||||
}
|
||||
|
||||
function withoutSteeringMessages(
|
||||
history: AgentRunRequest["history"],
|
||||
steering: AgentSteeringMessage[],
|
||||
): AgentRunRequest["history"] {
|
||||
if (steering.length === 0) return history;
|
||||
const result = [...history];
|
||||
let beforeIndex = result.length - 1;
|
||||
for (let steeringIndex = steering.length - 1; steeringIndex >= 0; steeringIndex -= 1) {
|
||||
const steeringMessage = steering[steeringIndex];
|
||||
for (let index = beforeIndex; index >= 0; index -= 1) {
|
||||
const message = result[index];
|
||||
if (
|
||||
message?.role !== "user" ||
|
||||
(message.id
|
||||
? message.id !== steeringMessage?.messageId
|
||||
: message.content !== (steeringMessage?.historyText ?? steeringMessage?.text))
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
result.splice(index, 1);
|
||||
beforeIndex = index - 1;
|
||||
break;
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* Normalize `request_secret` arguments.
|
||||
*
|
||||
* `credential` and `replace` must survive: the executor stores a submitted value
|
||||
* only when `credential` is present, and it validates the destination shape
|
||||
* itself. An earlier version of this function listed only label/purpose/
|
||||
* connectionId, so every credential the model supplied was dropped here and the
|
||||
* saved value had nowhere to go.
|
||||
*/
|
||||
export function prepareRequestSecretArguments(raw: Record<string, unknown>) {
|
||||
return {
|
||||
label: String(raw.label ?? "Code"),
|
||||
purpose: String(raw.purpose ?? "otp"),
|
||||
...(raw.connectionId ? { connectionId: String(raw.connectionId) } : {}),
|
||||
...(raw.credential ? { credential: raw.credential } : {}),
|
||||
...(raw.replace === true ? { replace: true } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
function toAgentTool(tool: ConnectorTool, host: ToolHost, exposedName: string): AgentTool {
|
||||
return {
|
||||
name: exposedName,
|
||||
@ -1268,459 +1089,14 @@ function builtinParameters(tool: ConnectorTool) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/** Keep recent visual state while respecting an optional model image budget. */
|
||||
export function pruneComputerScreenshotContext(
|
||||
messages: AgentMessage[],
|
||||
maxImagesPerPrompt?: number,
|
||||
): AgentMessage[] {
|
||||
const imageLimit =
|
||||
maxImagesPerPrompt === undefined
|
||||
? undefined
|
||||
: Number.isFinite(maxImagesPerPrompt)
|
||||
? Math.max(0, Math.floor(maxImagesPerPrompt))
|
||||
: DEFAULT_COMPUTER_SCREENSHOTS_TO_KEEP;
|
||||
let remaining = imageLimit ?? DEFAULT_COMPUTER_SCREENSHOTS_TO_KEEP;
|
||||
if (imageLimit !== undefined) {
|
||||
const nonScreenshotImages = messages.reduce(
|
||||
(count, message) =>
|
||||
isComputerScreenshotMessage(message) ? count : count + imagePartCount(message),
|
||||
0,
|
||||
);
|
||||
if (nonScreenshotImages > imageLimit) {
|
||||
throw new Error(
|
||||
`The configured model image limit is ${imageLimit}, but the prompt contains ${nonScreenshotImages} non-screenshot images.`,
|
||||
);
|
||||
}
|
||||
remaining = imageLimit - nonScreenshotImages;
|
||||
}
|
||||
let transformed: AgentMessage[] | undefined;
|
||||
for (let index = messages.length - 1; index >= 0; index -= 1) {
|
||||
const message = messages[index];
|
||||
if (!isComputerScreenshotMessage(message)) continue;
|
||||
const images = imagePartCount(message);
|
||||
if (remaining >= images) {
|
||||
remaining -= images;
|
||||
continue;
|
||||
}
|
||||
transformed ??= [...messages];
|
||||
transformed[index] = {
|
||||
...message,
|
||||
content: message.content.filter((part) => part.type !== "image"),
|
||||
};
|
||||
}
|
||||
return transformed ?? messages;
|
||||
}
|
||||
|
||||
function imagePartCount(message: AgentMessage): number {
|
||||
if (!("content" in message) || !Array.isArray(message.content)) return 0;
|
||||
return message.content.filter(
|
||||
(part: unknown) =>
|
||||
part !== null && typeof part === "object" && "type" in part && part.type === "image",
|
||||
).length;
|
||||
}
|
||||
|
||||
function isComputerScreenshotMessage(
|
||||
message: AgentMessage | undefined,
|
||||
): message is Extract<AgentMessage, { role: "toolResult" }> {
|
||||
if (message?.role !== "toolResult" || !message.content.some((part) => part.type === "image")) {
|
||||
return false;
|
||||
}
|
||||
const details = message.details;
|
||||
return Boolean(
|
||||
details &&
|
||||
typeof details === "object" &&
|
||||
"frameId" in details &&
|
||||
typeof (details as { frameId?: unknown }).frameId === "string",
|
||||
);
|
||||
}
|
||||
|
||||
function isAgentToolExecutionResult(result: unknown): result is AgentToolExecutionResult {
|
||||
if (
|
||||
!result ||
|
||||
typeof result !== "object" ||
|
||||
(result as { kind?: unknown }).kind !== "agent_tool_result" ||
|
||||
!("content" in result)
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
const content = (result as { content?: unknown }).content;
|
||||
return (
|
||||
Array.isArray(content) &&
|
||||
content.every(
|
||||
(item) =>
|
||||
item &&
|
||||
typeof item === "object" &&
|
||||
((item as { type?: unknown }).type === "text" ||
|
||||
(item as { type?: unknown }).type === "image"),
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
export function jsonSchemaParameters(
|
||||
schema: Record<string, unknown>,
|
||||
): ReturnType<typeof Type.Object> {
|
||||
// Top-level oneOf/anyOf (e.g. request_secret's credential XOR connectionId)
|
||||
// must stay a union. Falling through to properties would drop the exclusivity
|
||||
// and re-expose both destinations as optional siblings.
|
||||
const alternatives = Array.isArray(schema.oneOf)
|
||||
? schema.oneOf
|
||||
: Array.isArray(schema.anyOf)
|
||||
? schema.anyOf
|
||||
: undefined;
|
||||
if (alternatives && alternatives.length > 0 && schema.properties == null) {
|
||||
return Type.Union(
|
||||
alternatives.map((variant) => jsonSchemaParameters(variant as Record<string, unknown>)),
|
||||
) as unknown as ReturnType<typeof Type.Object>;
|
||||
}
|
||||
const properties = (schema.properties ?? {}) as Record<string, unknown>;
|
||||
const required = new Set(Array.isArray(schema.required) ? schema.required.map(String) : []);
|
||||
const fields: Record<string, ReturnType<typeof Type.Optional>> = {};
|
||||
for (const [key, spec] of Object.entries(properties)) {
|
||||
const field = jsonField(spec);
|
||||
fields[key] = (required.has(key) ? field : Type.Optional(field)) as unknown as ReturnType<
|
||||
typeof Type.Optional
|
||||
>;
|
||||
}
|
||||
// Preserve closed objects (e.g. request_secret destination oneOf branches).
|
||||
// Type.Object defaults to open, which would let connectionId+replace match both
|
||||
// anyOf variants after conversion.
|
||||
return schema.additionalProperties === false
|
||||
? Type.Object(fields, { additionalProperties: false })
|
||||
: Type.Object(fields);
|
||||
}
|
||||
|
||||
/** TypeBox only builds literals from primitives; anything else throws while the tool list is
|
||||
* being assembled, which would take down the whole turn. */
|
||||
function enumUnion(values: readonly unknown[]) {
|
||||
const members = values.map((value) =>
|
||||
value === null
|
||||
? Type.Null()
|
||||
: typeof value === "string" || typeof value === "number" || typeof value === "boolean"
|
||||
? Type.Literal(value)
|
||||
: undefined,
|
||||
);
|
||||
return members.every((member) => member !== undefined) ? Type.Union(members) : undefined;
|
||||
}
|
||||
|
||||
export function jsonField(spec: unknown): ReturnType<typeof Type.String> {
|
||||
const definition = spec && typeof spec === "object" ? (spec as Record<string, unknown>) : {};
|
||||
if (Array.isArray(definition.enum) && definition.enum.length > 0) {
|
||||
const union = enumUnion(definition.enum);
|
||||
if (union) return union as never;
|
||||
}
|
||||
// A `const` names the only accepted value. Without this it degraded to a bare
|
||||
// string, so a discriminator like {type: {const: "bearer"}} told the model
|
||||
// nothing about which value to send -- and it guessed, twice.
|
||||
if ("const" in definition) {
|
||||
const literal = enumUnion([definition.const]);
|
||||
if (literal) return literal as never;
|
||||
}
|
||||
// A discriminated union arrives as oneOf/anyOf with no sibling `type`. Without
|
||||
// this branch it fell through to the string default, so a model was told to
|
||||
// send an object-valued field as a bare string -- which is exactly what it did.
|
||||
const variants = Array.isArray(definition.oneOf)
|
||||
? definition.oneOf
|
||||
: Array.isArray(definition.anyOf)
|
||||
? definition.anyOf
|
||||
: undefined;
|
||||
if (variants && variants.length > 0) {
|
||||
return Type.Union(variants.map((variant) => jsonField(variant))) as never;
|
||||
}
|
||||
if (Array.isArray(definition.type) && definition.type.length > 0) {
|
||||
return Type.Union(definition.type.map((type) => jsonField({ ...definition, type }))) as never;
|
||||
}
|
||||
const type = "type" in definition ? String(definition.type) : "string";
|
||||
if (type === "null") return Type.Null() as never;
|
||||
if (type === "number" || type === "integer") return Type.Number() as never;
|
||||
if (type === "boolean") return Type.Boolean() as never;
|
||||
if (type === "array") {
|
||||
const options: {
|
||||
minItems?: number;
|
||||
maxItems?: number;
|
||||
uniqueItems?: boolean;
|
||||
} = {};
|
||||
if (typeof definition.minItems === "number") options.minItems = definition.minItems;
|
||||
if (typeof definition.maxItems === "number") options.maxItems = definition.maxItems;
|
||||
if (definition.uniqueItems === true) options.uniqueItems = true;
|
||||
return Type.Array(jsonField(definition.items), options) as never;
|
||||
}
|
||||
if (type === "object") return jsonSchemaParameters(definition) as never;
|
||||
return Type.String();
|
||||
}
|
||||
|
||||
function summarizeToolResult(result: unknown) {
|
||||
try {
|
||||
const text = JSON.stringify(result);
|
||||
if (!text) return "ok";
|
||||
return clipToolResultText(text);
|
||||
} catch {
|
||||
return "ok";
|
||||
}
|
||||
}
|
||||
|
||||
function boundAgentToolResult<T>(result: AgentToolResult<T>): AgentToolResult<T> {
|
||||
if (!result || !Array.isArray(result.content)) return result;
|
||||
return {
|
||||
...result,
|
||||
content: clipToolResultContent(result.content),
|
||||
};
|
||||
}
|
||||
|
||||
function assistantText(message: unknown): string {
|
||||
if (!message || typeof message !== "object" || !("content" in message)) return "";
|
||||
const content = (message as { content?: unknown }).content;
|
||||
if (typeof content === "string") return content;
|
||||
if (!Array.isArray(content)) return "";
|
||||
return content
|
||||
.map((part) =>
|
||||
part && typeof part === "object" && "type" in part && part.type === "text" && "text" in part
|
||||
? String(part.text)
|
||||
: "",
|
||||
)
|
||||
.join("");
|
||||
}
|
||||
|
||||
function sanitizeSensitiveText(message: string) {
|
||||
return message
|
||||
.replace(/sk-or-v1-[a-zA-Z0-9]+/g, "[redacted]")
|
||||
.replace(/sk-[a-zA-Z0-9-]+/g, "[redacted]")
|
||||
.replace(/Bearer\s+[^\s"',;&]+/gi, "Bearer [redacted]")
|
||||
.replace(/eyJ[a-zA-Z0-9_-]+\.[a-zA-Z0-9_-]+\.[a-zA-Z0-9_-]+/g, "[redacted]")
|
||||
.replace(/COMPOSIO_API_KEY[=:]?\s*\S+/gi, "COMPOSIO_API_KEY=[redacted]")
|
||||
.replace(
|
||||
/((?:api[_-]?key|access[_-]?token|password|secret)\s*[=:]\s*)[^\s"',;&]+/gi,
|
||||
"$1[redacted]",
|
||||
)
|
||||
.replace(/((?:auth|authorization)\s*[=:]\s*)(?!Bearer\b)[^\s"',;&]+/gi, "$1[redacted]");
|
||||
}
|
||||
|
||||
/** Origin + path only for activity chips; drop userinfo, query, and fragment. */
|
||||
function redactActivityUrl(value: unknown): string {
|
||||
const raw = String(value ?? "").trim();
|
||||
if (!raw) return raw;
|
||||
try {
|
||||
const url = new URL(raw);
|
||||
url.username = "";
|
||||
url.password = "";
|
||||
url.search = "";
|
||||
url.hash = "";
|
||||
return url.toString();
|
||||
} catch {
|
||||
// Never echo unparsed input — it may still contain userinfo/secrets.
|
||||
return "[invalid URL]";
|
||||
}
|
||||
}
|
||||
|
||||
function sanitizeError(message: string) {
|
||||
return sanitizeSensitiveText(message);
|
||||
}
|
||||
|
||||
/** Stable OpenCode affinity id for a bot conversation (and optional nested agent). */
|
||||
export function conversationSessionId(threadId: string, botId: string, agentId?: string): string {
|
||||
return agentId ? `${threadId}:${botId}:${agentId}` : `${threadId}:${botId}`;
|
||||
}
|
||||
|
||||
export function isOpenCodeProvider(provider: string): boolean {
|
||||
return provider === "opencode" || provider === "opencode-go";
|
||||
}
|
||||
|
||||
const OPENCODE_SESSION_ERROR = "OpenCode rejected this chat session. Send the message again.";
|
||||
|
||||
function looksLikeOpenCodeSessionError(message: string): boolean {
|
||||
return (
|
||||
/x-opencode-session/i.test(message) ||
|
||||
/session\s*(id|header|required|missing|invalid|expired|stale)/i.test(message) ||
|
||||
/model is unavailable/i.test(message)
|
||||
);
|
||||
}
|
||||
|
||||
function sanitizeProviderError(provider: string, message: string): string {
|
||||
const sanitized = sanitizeError(message);
|
||||
if (isOpenCodeProvider(provider) && looksLikeOpenCodeSessionError(sanitized)) {
|
||||
return OPENCODE_SESSION_ERROR;
|
||||
}
|
||||
return sanitized;
|
||||
}
|
||||
|
||||
interface EventQueue {
|
||||
push(event: AgentRuntimeEvent): void;
|
||||
fail(error: Error): void;
|
||||
close(): void;
|
||||
iterate(): AsyncIterable<AgentRuntimeEvent>;
|
||||
}
|
||||
|
||||
interface ToolHost {
|
||||
queue: EventQueue;
|
||||
request: AgentRunRequest;
|
||||
models: Models;
|
||||
model: Model<Api>;
|
||||
apiKey: string | undefined;
|
||||
nestedAgents: Set<Agent>;
|
||||
subagentGate: { acquire(): Promise<void>; release(): void };
|
||||
toolCallBudget: ToolCallBudget;
|
||||
/** Shared fallback uniqueness when the model omits toolCallId (nested hosts reuse this). */
|
||||
toolCallSeq: { value: number };
|
||||
abortTurn(): void;
|
||||
signal: AbortSignal;
|
||||
depth: number;
|
||||
pausePending: boolean;
|
||||
}
|
||||
|
||||
function toolCallBudgetExceededMessage(limit: number) {
|
||||
return `I stopped after reaching the limit of ${limit} tool calls in this turn. Send another message to continue.`;
|
||||
}
|
||||
|
||||
function toolCallBudgetFor(runId: string): ToolCallBudget {
|
||||
const existing = toolCallBudgetsByRun.get(runId);
|
||||
if (existing) {
|
||||
existing.inFlight = 0;
|
||||
existing.limit = maxToolCallsPerTurn();
|
||||
return existing;
|
||||
}
|
||||
const budget: ToolCallBudget = {
|
||||
count: 0,
|
||||
exceeded: false,
|
||||
limit: maxToolCallsPerTurn(),
|
||||
inFlight: 0,
|
||||
};
|
||||
if (budget.limit > 0) toolCallBudgetsByRun.set(runId, budget);
|
||||
return budget;
|
||||
}
|
||||
|
||||
function releaseToolCallBudget(runId: string, keepForResume: boolean) {
|
||||
if (!keepForResume) toolCallBudgetsByRun.delete(runId);
|
||||
}
|
||||
|
||||
function maybeAbortToolCallBudget(host: ToolHost) {
|
||||
if (host.toolCallBudget.exceeded && host.toolCallBudget.inFlight === 0) {
|
||||
host.abortTurn();
|
||||
}
|
||||
}
|
||||
|
||||
function beginToolCall(host: ToolHost): boolean {
|
||||
const budget = host.toolCallBudget;
|
||||
if (budget.limit <= 0) {
|
||||
budget.count += 1;
|
||||
return true;
|
||||
}
|
||||
if (budget.exceeded) {
|
||||
maybeAbortToolCallBudget(host);
|
||||
return false;
|
||||
}
|
||||
budget.count += 1;
|
||||
if (budget.count <= budget.limit) {
|
||||
budget.inFlight += 1;
|
||||
return true;
|
||||
}
|
||||
if (!budget.exceeded) {
|
||||
budget.exceeded = true;
|
||||
host.queue.push({
|
||||
type: "progress",
|
||||
text: `Stopped: more than ${budget.limit} tool calls in one turn.`,
|
||||
});
|
||||
}
|
||||
maybeAbortToolCallBudget(host);
|
||||
return false;
|
||||
}
|
||||
|
||||
function endToolCall(host: ToolHost) {
|
||||
host.toolCallBudget.inFlight = Math.max(0, host.toolCallBudget.inFlight - 1);
|
||||
maybeAbortToolCallBudget(host);
|
||||
}
|
||||
|
||||
function modelForCompletion(model: Model<Api>, configuredMaxTokens?: number): Model<Api> {
|
||||
const maxTokens = resolveCompletionMaxTokens(model.maxTokens, configuredMaxTokens);
|
||||
if (maxTokens === model.maxTokens) return model;
|
||||
return { ...model, maxTokens };
|
||||
}
|
||||
|
||||
function createGate(max: number) {
|
||||
let active = 0;
|
||||
const waiters: Array<() => void> = [];
|
||||
return {
|
||||
async acquire() {
|
||||
if (active >= max) {
|
||||
await new Promise<void>((resolve) => {
|
||||
waiters.push(resolve);
|
||||
});
|
||||
}
|
||||
active += 1;
|
||||
},
|
||||
release() {
|
||||
active = Math.max(0, active - 1);
|
||||
waiters.shift()?.();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function createQueue(): EventQueue {
|
||||
const items: AgentRuntimeEvent[] = [];
|
||||
let wake: (() => void) | undefined;
|
||||
let closed = false;
|
||||
let failure: Error | undefined;
|
||||
return {
|
||||
push(event) {
|
||||
items.push(event);
|
||||
wake?.();
|
||||
},
|
||||
fail(error) {
|
||||
failure = error;
|
||||
closed = true;
|
||||
wake?.();
|
||||
},
|
||||
close() {
|
||||
closed = true;
|
||||
wake?.();
|
||||
},
|
||||
async *iterate() {
|
||||
while (true) {
|
||||
if (items.length) {
|
||||
yield items.shift()!;
|
||||
continue;
|
||||
}
|
||||
if (failure) throw failure;
|
||||
if (closed) return;
|
||||
await new Promise<void>((resolve) => {
|
||||
wake = resolve;
|
||||
});
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function reliableStreamOptions(
|
||||
model: Pick<Model<Api>, "api" | "provider" | "maxTokens">,
|
||||
options?: SimpleStreamOptions,
|
||||
configuredMaxTokens?: number,
|
||||
): SimpleStreamOptions {
|
||||
let next: SimpleStreamOptions = {
|
||||
...options,
|
||||
timeoutMs: options?.timeoutMs ?? MODEL_STREAM_TIMEOUT_MS,
|
||||
maxRetries: options?.maxRetries ?? MODEL_STREAM_MAX_RETRIES,
|
||||
maxTokens: resolveCompletionMaxTokens(model.maxTokens, configuredMaxTokens, options?.maxTokens),
|
||||
};
|
||||
|
||||
if (model.provider === "openai-codex" || model.api === "openai-codex-responses") {
|
||||
// Pi cannot fall back after a WebSocket has emitted its start event. Long tool
|
||||
// runs then surface abnormal close 1006 as a terminal model error. SSE has
|
||||
// bounded network retries and no long-lived connection between tool turns.
|
||||
next = { ...next, transport: "sse" };
|
||||
}
|
||||
|
||||
// OpenCode Go/Zen require a sticky x-opencode-session header (affinity + some
|
||||
// models 400 without it). Pi 0.85.1 does not attach that header on its own.
|
||||
if (isOpenCodeProvider(model.provider)) {
|
||||
const sessionId = next.sessionId?.trim() || randomUUID();
|
||||
next = {
|
||||
...next,
|
||||
sessionId,
|
||||
headers: {
|
||||
"x-opencode-session": sessionId,
|
||||
"x-opencode-client": "rakazo",
|
||||
...next.headers,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
return next;
|
||||
}
|
||||
export {
|
||||
conversationSessionId,
|
||||
isOpenCodeProvider,
|
||||
jsonField,
|
||||
jsonSchemaParameters,
|
||||
maxToolCallsPerTurn,
|
||||
modelsForRequest,
|
||||
prepareRequestSecretArguments,
|
||||
pruneComputerScreenshotContext,
|
||||
reliableStreamOptions,
|
||||
} from "./pi-model-config.js";
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user