diff --git a/packages/adapters/src/executor-computer-busy-retry.ts b/packages/adapters/src/executor-computer-busy-retry.ts index 5f75aab..64388c8 100644 --- a/packages/adapters/src/executor-computer-busy-retry.ts +++ b/packages/adapters/src/executor-computer-busy-retry.ts @@ -16,7 +16,7 @@ export const COMPUTER_BUSY_EXHAUSTED_ERROR = const COMPUTER_BUSY_RETRY_MARKER = /^computer_busy_retry:(\d+)$/; -export function parseComputerBusyRetryCount(error: string | null | undefined): number { +export function parseComputerBusyRetryCount(error: string | null): number { if (!error) return 0; const match = error.match(COMPUTER_BUSY_RETRY_MARKER); if (!match) return 0; @@ -24,11 +24,11 @@ export function parseComputerBusyRetryCount(error: string | null | undefined): n return Number.isFinite(parsed) && parsed > 0 ? parsed : 0; } -export function isComputerBusyRetryMarker(error: string | null | undefined): boolean { +export function isComputerBusyRetryMarker(error: string | null): boolean { return Boolean(error && COMPUTER_BUSY_RETRY_MARKER.test(error)); } -export function preserveRunErrorForComputerBusyRetry(error: string | null | undefined): string | null { +export function preserveRunErrorForComputerBusyRetry(error: string | null): string | null { return isComputerBusyRetryMarker(error) ? error : null; } diff --git a/packages/adapters/src/executor.ts b/packages/adapters/src/executor.ts index 68761fa..2ee6b69 100644 --- a/packages/adapters/src/executor.ts +++ b/packages/adapters/src/executor.ts @@ -25,7 +25,8 @@ import { routineWakeupJob, runContinueJob, } from "@rakazo/adapter-kit"; -import type { MessageBlock, RunStatus } from "@rakazo/contracts"; +import type { RunStatus } from "@rakazo/contracts"; +import { type MessageBlock, MessageBlock as MessageBlockSchema } from "@rakazo/contracts"; import { ATTACHMENT_MAX_BYTES, BOT_DESCRIPTION_MAX_LENGTH, @@ -100,7 +101,6 @@ import { type ThreadEvents, } from "@rakazo/db"; import { getLogger } from "@rakazo/logging"; -import { parse as parseShellCommand } from "shell-quote"; import { connectAgent, messageConnectedAgent, @@ -139,7 +139,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 +166,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, @@ -263,7 +285,6 @@ import { secretValuesToRedact, serializeModelSecret, } from "./pi-oauth.js"; -import { maxToolCallsPerTurn } from "./pi-model-config.js"; import { assertPlotDataWithinLimits, PLOT_TOOL_GUIDE, @@ -274,7 +295,6 @@ import { searchChartCatalog, } from "./plot-tool.js"; import type { RemoteTransportDependencies } from "./remote-mcp.js"; -import { createRunToolCallBudgetStore } from "./tool-call-budget-store.js"; import { loadReplyContext, messageToAgentHistoryText } from "./reply-context.js"; import { commitConsumedRunSecret, @@ -340,7 +360,6 @@ import { import { createWebProvider } from "./web-provider-factory.js"; import { webFetchFromTool, webSearchFromTool } from "./web-tools.js"; -const modelCredentialLocks = new Map>(); const READ_ONLY_AGENT_TOOLS = new Set([ "computer_observe", "list_files", @@ -358,204 +377,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) { - 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> = { - 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."; @@ -564,37 +386,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 }; - listConnectedPluginSlugs?: (userId: string) => Promise; - /** 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[]; @@ -1222,7 +1013,7 @@ export function createRunExecutor(deps: ExecutorDeps) { }), findDefaultModelCredential(deps.prisma, run), deps.prisma.deploymentSettings.findUnique({ where: { id: "default" } }), - deps.memoryProviders.resolve(run.spaceId), + deps.memoryProviders.resolve({ spaceId: run.spaceId, userId: run.userId }), deps.prisma.taughtSkill.findMany({ where: { botId: run.botId, spaceId: run.spaceId, status: "saved" }, }), @@ -1522,11 +1313,11 @@ export function createRunExecutor(deps: ExecutorDeps) { }, }); if (!source) return undefined; - const parsed = MessageBlock.array().safeParse(source.blocks); + const parsed = MessageBlockSchema.array().safeParse(source.blocks); if (!parsed.success) return undefined; const handoff = inboundHandoffForBot(parsed.data, bot.id); if (!handoff) return undefined; - return renderInboundHandoffWorkshopInstruction(handoff, source.bot?.name); + return renderInboundHandoffWorkshopInstruction(handoff, (source.bot as { name?: string } | null)?.name); })(); const hasMessagingIdentity = deps.messaging ? await deps.messaging.hasIdentity(bot.id) @@ -2183,7 +1974,7 @@ export function createRunExecutor(deps: ExecutorDeps) { }; if (applied?.duplicate) { - const gate = resolveDuplicateEffectGate(applied.effect, name); + const gate = resolveDuplicateEffectGate(applied.effect as any, name); if (gate.action === "return") { if (name === "request_secret") { const replacementSecret = await deps.prisma.secret.findFirst({ @@ -3042,7 +2833,7 @@ export function createRunExecutor(deps: ExecutorDeps) { } const recordedEffect = await recordEffect(deps, run, name, effectKey, args); if (recordedEffect?.duplicate) { - const gate = resolveDuplicateEffectGate(recordedEffect.effect, name); + const gate = resolveDuplicateEffectGate(recordedEffect.effect as any, name); if (gate.action === "execute") { const early = await claimOrReturn("approved"); if (early !== undefined) return early; @@ -3097,7 +2888,7 @@ export function createRunExecutor(deps: ExecutorDeps) { }); } const recordedForAsk = await recordEffect(deps, run, name, effectKey, args); - const missingSecretAction = resolveMissingRunSecretAction(recordedForAsk.effect); + const missingSecretAction = resolveMissingRunSecretAction(recordedForAsk.effect as any); if (missingSecretAction.action === "return") return missingSecretAction.result; const connectionId = args.connectionId ? String(args.connectionId) : undefined; if (connectionId) { @@ -3719,10 +3510,6 @@ export function createRunExecutor(deps: ExecutorDeps) { completion, runSecrets, ), - toolCallBudgetStore: - maxToolCallsPerTurn() > 0 - ? createRunToolCallBudgetStore(deps.prisma, runId) - : undefined, claimSteering: scripted ? undefined : async (seenIds) => { @@ -4175,7 +3962,7 @@ export function createRunExecutor(deps: ExecutorDeps) { const workshopFallback = workshopFallbackAfterBlockedHandoff( proseRecovery.sanitizedText, proseRecovery.blockedUpward, - proseRecovery.proseToolCalls, + proseRecovery.handoffFailed, ); if (proseRecovery.proseToolCalls.length > 0) { assembled = proseRecovery.sanitizedText; @@ -4397,53 +4184,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( - 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, @@ -4480,224 +4220,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; - 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(key: string, fn: () => Promise): Promise { - const previous = modelCredentialLocks.get(key) ?? Promise.resolve(); - let release!: () => void; - const current = previous.then( - () => - new Promise((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 => 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 = []; - - 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"; diff --git a/packages/adapters/src/messaging-delivery.test.ts b/packages/adapters/src/messaging-delivery.test.ts index fb55744..2d06a41 100644 --- a/packages/adapters/src/messaging-delivery.test.ts +++ b/packages/adapters/src/messaging-delivery.test.ts @@ -699,13 +699,9 @@ function createChannelDeps( }); }), createMany: vi.fn(async ({ data }: { data: Array> }) => { - let count = 0; - for (const item of data) { - if (rows.some((row) => row.idempotencyKey === item.idempotencyKey)) continue; + for (const item of data) rows.push({ id: `out-${rows.length + 1}`, status: "pending", ...item }); - count += 1; - } - return { count }; + return { count: data.length }; }), update: vi.fn(async ({ where, data }: { where: { id: string }; data: unknown }) => { const row = rows.find((candidate) => candidate.id === where.id); @@ -754,65 +750,22 @@ function createChannelDeps( } describe("deliverMessagingOutbound channel runs", () => { - it("skips answered ask messages but still mirrors text replies on the same run", async () => { + it("keeps privately answered runs out of the group and peer fan-out on retries", async () => { const deps = createChannelDeps({ messages: [ { id: "ask-1", blocks: [{ kind: "ask", text: "Details?", status: "answered", answer: "Private detail" }], }, - { id: "reply-1", blocks: [{ kind: "text", text: "Done — calendar updated." }] }, + { id: "reply-1", blocks: [{ kind: "text", text: "Reply containing private detail" }] }, ], }); await deliverMessagingOutbound(deps, { runId: "run-1" }, context); - expect(deps.sendToThread).toHaveBeenCalledWith( - { threadId: "sendblue:grp-1", body: "Alice's agent: Done — calendar updated." }, - context, - ); - expect(deps.rows).toEqual([ - expect.objectContaining({ - idempotencyKey: "msg:reply-1", - kind: "group", - body: "Alice's agent: Done — calendar updated.", - }), - ]); - expect(deps.rows.some((row) => String(row.body).includes("Private detail"))).toBe(false); - }); - - it("mirrors completion text after in-app approval on a resumed group run", async () => { - let messages = [ - { id: "ask-1", blocks: [{ kind: "ask", text: "Approve?", status: "pending" }] }, - { id: "interim-1", blocks: [{ kind: "text", text: "Working on it…" }] }, - ]; - const deps = createChannelDeps(); - deps.prisma.message.findMany = vi.fn(async () => messages); await deliverMessagingOutbound(deps, { runId: "run-1" }, context); - expect(deps.sendToThread).toHaveBeenCalledWith( - { threadId: "sendblue:grp-1", body: "Alice's agent: Working on it…" }, - context, - ); - - messages = [ - { - id: "ask-1", - blocks: [{ kind: "ask", text: "Approve?", status: "answered", answer: "yes" }], - }, - { id: "interim-1", blocks: [{ kind: "text", text: "Working on it…" }] }, - { id: "done-1", blocks: [{ kind: "text", text: "All set." }] }, - ]; - deps.sendToThread.mockClear(); - await deliverMessagingOutbound(deps, { runId: "run-1" }, context); - expect(deps.sendToThread).toHaveBeenCalledTimes(1); - expect(deps.sendToThread).toHaveBeenCalledWith( - { threadId: "sendblue:grp-1", body: "Alice's agent: All set." }, - context, - ); - expect(deps.rows).toEqual( - expect.arrayContaining([ - expect.objectContaining({ idempotencyKey: "msg:interim-1" }), - expect.objectContaining({ idempotencyKey: "msg:done-1" }), - ]), - ); + expect(deps.rows).toEqual([]); + expect(deps.sendToThread).not.toHaveBeenCalled(); + expect(deps.contextMessages).toEqual([]); + expect(deps.sendUserMessage).not.toHaveBeenCalled(); }); it("can still publish a group reply alongside an unanswered ask", async () => { diff --git a/packages/adapters/src/pi-model-config.ts b/packages/adapters/src/pi-model-config.ts index fdc69b3..d38702a 100644 --- a/packages/adapters/src/pi-model-config.ts +++ b/packages/adapters/src/pi-model-config.ts @@ -12,13 +12,12 @@ import { builtinModels } from "@earendil-works/pi-ai/providers/all"; import { type Agent, type AgentToolResult } from "@earendil-works/pi-agent-core"; import { type AgentMessage, - type AgentToolExecutionResult, } from "@earendil-works/pi-agent-core"; +import type { AgentToolExecutionResult } from "@rakazo/adapter-kit"; import type { AgentRuntimeEvent, AgentRunRequest, AgentSteeringMessage, - ToolCallBudgetStore, } from "@rakazo/adapter-kit"; import { getLogger } from "@rakazo/logging"; import { DEFAULT_OPENROUTER_MODEL_ID } from "./deployment-model.js"; @@ -67,7 +66,7 @@ export interface ToolCallBudget { inFlight: number; } -// In-memory cache for same-process resume; durable store backs cross-worker continueRun. +// Optional fuse is process-local. continueRun on another worker starts at zero. const toolCallBudgetsByRun = new Map(); /** Optional self-host fuse. Unset, empty, or 0 means unlimited (default). */ @@ -79,70 +78,28 @@ export function maxToolCallsPerTurn(env: NodeJS.ProcessEnv = process.env): numbe return Math.floor(parsed); } -export async function loadToolCallBudget( - runId: string, - store?: ToolCallBudgetStore, -): Promise { - const limit = maxToolCallsPerTurn(); - if (limit <= 0) { - return { count: 0, exceeded: false, limit: 0, inFlight: 0 }; - } - +export function toolCallBudgetFor(runId: string): ToolCallBudget { const existing = toolCallBudgetsByRun.get(runId); if (existing) { existing.inFlight = 0; - existing.limit = limit; + existing.limit = maxToolCallsPerTurn(); return existing; } - - let count = 0; - let exceeded = false; - if (store) { - const snapshot = await store.load(); - if (snapshot) { - count = snapshot.count; - exceeded = snapshot.exceeded; - } - } - - const budget: ToolCallBudget = { count, exceeded, limit, inFlight: 0 }; - toolCallBudgetsByRun.set(runId, budget); + const budget: ToolCallBudget = { + count: 0, + exceeded: false, + limit: maxToolCallsPerTurn(), + inFlight: 0, + }; + if (budget.limit > 0) toolCallBudgetsByRun.set(runId, budget); return budget; } -export async function releaseToolCallBudget( - runId: string, - keepForResume: boolean, - store?: ToolCallBudgetStore, -): Promise { - const budget = toolCallBudgetsByRun.get(runId); - if (keepForResume) { - if (budget && store) { - await store.persist({ count: budget.count, exceeded: budget.exceeded }); - } - return; - } - toolCallBudgetsByRun.delete(runId); - if (store) await store.clear(); +export function releaseToolCallBudget(runId: string, keepForResume: boolean) { + if (!keepForResume) toolCallBudgetsByRun.delete(runId); } -/** @internal Simulates a fresh worker process in unit tests. */ -export function resetToolCallBudgetMemoryForTests(): void { - toolCallBudgetsByRun.clear(); -} - -function persistToolCallBudget( - runId: string | undefined, - budget: ToolCallBudget, - store?: ToolCallBudgetStore, -): void { - if (!runId || !store || budget.limit <= 0) return; - void store.persist({ count: budget.count, exceeded: budget.exceeded }).catch((error) => { - getLogger().warn("tool-call budget persist failed", { runId, error }); - }); -} - -function maybeAbortToolCallBudget(host: { +export function maybeAbortToolCallBudget(host: { toolCallBudget: ToolCallBudget; abortTurn(): void; }): void { @@ -154,8 +111,7 @@ function maybeAbortToolCallBudget(host: { export function beginToolCall(host: { toolCallBudget: ToolCallBudget; abortTurn(): void; - queue: { push(event: { type: string; text: string }): void }; - request?: Pick; + queue: { push(event: AgentRuntimeEvent): void }; }): boolean { const budget = host.toolCallBudget; if (budget.limit <= 0) { @@ -169,7 +125,6 @@ export function beginToolCall(host: { budget.count += 1; if (budget.count <= budget.limit) { budget.inFlight += 1; - persistToolCallBudget(host.request?.runId, budget, host.request?.toolCallBudgetStore); return true; } if (!budget.exceeded) { @@ -179,7 +134,6 @@ export function beginToolCall(host: { text: `Stopped: more than ${budget.limit} tool calls in one turn.`, }); } - persistToolCallBudget(host.request?.runId, budget, host.request?.toolCallBudgetStore); maybeAbortToolCallBudget(host); return false; } @@ -187,10 +141,8 @@ export function beginToolCall(host: { export function endToolCall(host: { toolCallBudget: ToolCallBudget; abortTurn(): void; - request?: Pick; }): void { host.toolCallBudget.inFlight = Math.max(0, host.toolCallBudget.inFlight - 1); - persistToolCallBudget(host.request?.runId, host.toolCallBudget, host.request?.toolCallBudgetStore); maybeAbortToolCallBudget(host); } @@ -200,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, @@ -252,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; @@ -290,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"), @@ -300,7 +252,7 @@ function toPiImages(images: AgentRunRequest["currentTurnImages"]) { // --- Steering message removal --- -function withoutSteeringMessages( +export function withoutSteeringMessages( history: AgentRunRequest["history"], steering: AgentSteeringMessage[], ): AgentRunRequest["history"] { @@ -385,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) => @@ -393,7 +345,7 @@ function imagePartCount(message: AgentMessage): number { ).length; } -function isComputerScreenshotMessage( +export function isComputerScreenshotMessage( message: AgentMessage | undefined, ): message is Extract { if (message?.role !== "toolResult" || !message.content.some((part) => (part as {type?: string}).type === "image")) { @@ -410,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" || @@ -461,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() @@ -514,7 +466,7 @@ export function jsonField(spec: unknown): ReturnType { // --- Result helpers --- -function summarizeToolResult(result: unknown) { +export function summarizeToolResult(result: unknown) { try { const text = JSON.stringify(result); if (!text) return "ok"; @@ -524,7 +476,7 @@ function summarizeToolResult(result: unknown) { } } -function boundAgentToolResult( +export function boundAgentToolResult( result: AgentToolResult, ): AgentToolResult { if (!result || !Array.isArray(result.content)) return result; @@ -534,7 +486,7 @@ function boundAgentToolResult( }; } -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; @@ -554,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]") @@ -574,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 { @@ -589,7 +541,7 @@ function redactActivityUrl(value: unknown): string { } } -function sanitizeError(message: string) { +export function sanitizeError(message: string) { return sanitizeSensitiveText(message); } @@ -610,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) || @@ -618,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; @@ -628,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; diff --git a/packages/adapters/src/pi-runtime.ts b/packages/adapters/src/pi-runtime.ts index a53e1c9..f537483 100644 --- a/packages/adapters/src/pi-runtime.ts +++ b/packages/adapters/src/pi-runtime.ts @@ -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 { - beginToolCall, - endToolCall, - loadToolCallBudget, - releaseToolCallBudget, - toolCallBudgetExceededMessage, - type ToolCallBudget, -} from "./pi-model-config.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 }>(); +// Optional fuse is process-local. continueRun on another worker starts at zero. // 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,28 +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, - 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"; -export { maxToolCallsPerTurn } from "./pi-model-config.js"; - export interface PiAgentRuntimeOptions { /** Directory where Pi JSONL sessions are written. Omit to disable recording. */ sessionRoot?: string; @@ -183,7 +159,7 @@ export class PiAgentRuntime implements AgentRuntime { const toolDefs = request.tools.length ? request.tools : builtinAgentTools; const nestedAgents = new Set(); const completionModel = modelForCompletion(model, request.model.maxTokens); - trackedBudget = await loadToolCallBudget(request.runId, request.toolCallBudgetStore); + trackedBudget = toolCallBudgetFor(request.runId); const host: ToolHost = { queue, request, @@ -436,11 +412,7 @@ export class PiAgentRuntime implements AgentRuntime { if (trackedBudget) { const keepForResume = (signal.aborted || Boolean(resumeHost?.pausePending)) && !trackedBudget.exceeded; - await releaseToolCallBudget( - request.runId, - keepForResume, - request.toolCallBudgetStore, - ); + releaseToolCallBudget(request.runId, keepForResume); } } })(); @@ -457,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 | 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, - 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]!)); @@ -708,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) { - 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, @@ -1265,397 +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 { - 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, -): ReturnType { - // 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)), - ) as unknown as ReturnType; - } - const properties = (schema.properties ?? {}) as Record; - const required = new Set(Array.isArray(schema.required) ? schema.required.map(String) : []); - const fields: Record> = {}; - 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 { - const definition = spec && typeof spec === "object" ? (spec as Record) : {}; - 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(result: AgentToolResult): AgentToolResult { - 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; -} - -interface ToolHost { - queue: EventQueue; - request: AgentRunRequest; - models: Models; - model: Model; - apiKey: string | undefined; - nestedAgents: Set; - subagentGate: { acquire(): Promise; 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 modelForCompletion(model: Model, configuredMaxTokens?: number): Model { - 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((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((resolve) => { - wake = resolve; - }); - } - }, - }; -} - -export function reliableStreamOptions( - model: Pick, "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"; diff --git a/packages/adapters/src/prose-handoff-recovery.test.ts b/packages/adapters/src/prose-handoff-recovery.test.ts index c7fc5d0..6015512 100644 --- a/packages/adapters/src/prose-handoff-recovery.test.ts +++ b/packages/adapters/src/prose-handoff-recovery.test.ts @@ -98,15 +98,13 @@ describe("tryRecoverProseHandoff", () => { describe("workshopFallbackAfterBlockedHandoff", () => { it("returns a workshop line when only blocked handoff prose remains", () => { expect( - workshopFallbackAfterBlockedHandoff("", true, [ - { name: "handoff_to_bot", parameters: { message: "same" } }, - ]), + workshopFallbackAfterBlockedHandoff("", true, false), ).toBe(INBOUND_HANDOFF_WORKSHOP_FALLBACK); }); it("does not override real specialist prose", () => { expect( - workshopFallbackAfterBlockedHandoff("Drei Richtungen: ruhig, mutig, verspielt.", true, []), + workshopFallbackAfterBlockedHandoff("Drei Richtungen: ruhig, mutig, verspielt.", true, false), ).toBeUndefined(); }); });