cynarAI/evOo-local
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

fix(adapters): repair TypeScript errors from cherry-pick merges — 0 TS errors restored

Browse Source 
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
Ingo Wagner 2026-09-22 06:57:55 +00:00
parent cc4a46fb5a
commit 0a8a8319a7
6 changed files with 132 additions and 1260 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
packages/adapters/src/executor-computer-busy-retry.ts
View File

@ -16,7 +16,7 @@ export const COMPUTER_BUSY_EXHAUSTED_ERROR =
const COMPUTER_BUSY_RETRY_MARKER = /^computer_busy_retry:(\d+)$/; 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; if (!error) return 0;
const match = error.match(COMPUTER_BUSY_RETRY_MARKER); const match = error.match(COMPUTER_BUSY_RETRY_MARKER);
if (!match) return 0; if (!match) return 0;
@ -24,11 +24,11 @@ export function parseComputerBusyRetryCount(error: string | null | undefined): n
return Number.isFinite(parsed) && parsed > 0 ? parsed : 0; 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)); 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; return isComputerBusyRetryMarker(error) ? error : null;
} }

554
packages/adapters/src/executor.ts
View File

@ -25,7 +25,8 @@ import {
routineWakeupJob, routineWakeupJob,
runContinueJob, runContinueJob,
} from "@rakazo/adapter-kit"; } 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 { import {
ATTACHMENT_MAX_BYTES, ATTACHMENT_MAX_BYTES,
BOT_DESCRIPTION_MAX_LENGTH, BOT_DESCRIPTION_MAX_LENGTH,
@ -100,7 +101,6 @@ import {
type ThreadEvents, type ThreadEvents,
} from "@rakazo/db"; } from "@rakazo/db";
import { getLogger } from "@rakazo/logging"; import { getLogger } from "@rakazo/logging";
import { parse as parseShellCommand } from "shell-quote";
import { import {
connectAgent, connectAgent,
messageConnectedAgent, messageConnectedAgent,
@ -139,7 +139,20 @@ import {
settleUncertainEffect, settleUncertainEffect,
uncertainEffectResult, uncertainEffectResult,
} from "./approval-effect.js"; } 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 { import {
computerScreenToolResult, computerScreenToolResult,
notifyRun, notifyRun,
@ -153,7 +166,16 @@ import {
persistMessageInTransaction, persistMessageInTransaction,
runNotificationsEnabled, runNotificationsEnabled,
} from "./executor-computer.js"; } 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 { import {
autoReviewTimeoutMs, autoReviewTimeoutMs,
buildAutoReviewPrompt, buildAutoReviewPrompt,
@ -263,7 +285,6 @@ import {
secretValuesToRedact, secretValuesToRedact,
serializeModelSecret, serializeModelSecret,
} from "./pi-oauth.js"; } from "./pi-oauth.js";
import { maxToolCallsPerTurn } from "./pi-model-config.js";
import { import {
assertPlotDataWithinLimits, assertPlotDataWithinLimits,
PLOT_TOOL_GUIDE, PLOT_TOOL_GUIDE,
@ -274,7 +295,6 @@ import {
searchChartCatalog, searchChartCatalog,
} from "./plot-tool.js"; } from "./plot-tool.js";
import type { RemoteTransportDependencies } from "./remote-mcp.js"; import type { RemoteTransportDependencies } from "./remote-mcp.js";
import { createRunToolCallBudgetStore } from "./tool-call-budget-store.js";
import { loadReplyContext, messageToAgentHistoryText } from "./reply-context.js"; import { loadReplyContext, messageToAgentHistoryText } from "./reply-context.js";
import { import {
commitConsumedRunSecret, commitConsumedRunSecret,
@ -340,7 +360,6 @@ import {
import { createWebProvider } from "./web-provider-factory.js"; import { createWebProvider } from "./web-provider-factory.js";
import { webFetchFromTool, webSearchFromTool } from "./web-tools.js"; import { webFetchFromTool, webSearchFromTool } from "./web-tools.js";
const modelCredentialLocks = new Map<string, Promise<void>>();
const READ_ONLY_AGENT_TOOLS = new Set([ const READ_ONLY_AGENT_TOOLS = new Set([
"computer_observe", "computer_observe",
"list_files", "list_files",
@ -358,204 +377,7 @@ const READ_ONLY_AGENT_TOOLS = new Set([
"cloud_agent_status", "cloud_agent_status",
]); ]);
const MAX_MODEL_FILE_BYTES = 250_000; 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)); 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. */ /** Cap the roster so a large Space cannot flood the prompt. */
const BOT_DIRECTORY_LIMIT = 40; const BOT_DIRECTORY_LIMIT = 40;
const MISSING_MODEL_MESSAGE = "Connect a model in Settings before running bots."; 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; 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 { function isAuditableToolResult(value: unknown): value is {
kind: "agent_tool_result"; kind: "agent_tool_result";
content: unknown[]; content: unknown[];
@ -1222,7 +1013,7 @@ export function createRunExecutor(deps: ExecutorDeps) {
}), }),
findDefaultModelCredential(deps.prisma, run), findDefaultModelCredential(deps.prisma, run),
deps.prisma.deploymentSettings.findUnique({ where: { id: "default" } }), 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({ deps.prisma.taughtSkill.findMany({
where: { botId: run.botId, spaceId: run.spaceId, status: "saved" }, where: { botId: run.botId, spaceId: run.spaceId, status: "saved" },
}), }),
@ -1522,11 +1313,11 @@ export function createRunExecutor(deps: ExecutorDeps) {
}, },
}); });
if (!source) return undefined; if (!source) return undefined;
const parsed = MessageBlock.array().safeParse(source.blocks); const parsed = MessageBlockSchema.array().safeParse(source.blocks);
if (!parsed.success) return undefined; if (!parsed.success) return undefined;
const handoff = inboundHandoffForBot(parsed.data, bot.id); const handoff = inboundHandoffForBot(parsed.data, bot.id);
if (!handoff) return undefined; if (!handoff) return undefined;
return renderInboundHandoffWorkshopInstruction(handoff, source.bot?.name); return renderInboundHandoffWorkshopInstruction(handoff, (source.bot as { name?: string } | null)?.name);
})(); })();
const hasMessagingIdentity = deps.messaging const hasMessagingIdentity = deps.messaging
? await deps.messaging.hasIdentity(bot.id) ? await deps.messaging.hasIdentity(bot.id)
@ -2183,7 +1974,7 @@ export function createRunExecutor(deps: ExecutorDeps) {
}; };
if (applied?.duplicate) { if (applied?.duplicate) {
const gate = resolveDuplicateEffectGate(applied.effect, name); const gate = resolveDuplicateEffectGate(applied.effect as any, name);
if (gate.action === "return") { if (gate.action === "return") {
if (name === "request_secret") { if (name === "request_secret") {
const replacementSecret = await deps.prisma.secret.findFirst({ 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); const recordedEffect = await recordEffect(deps, run, name, effectKey, args);
if (recordedEffect?.duplicate) { if (recordedEffect?.duplicate) {
const gate = resolveDuplicateEffectGate(recordedEffect.effect, name); const gate = resolveDuplicateEffectGate(recordedEffect.effect as any, name);
if (gate.action === "execute") { if (gate.action === "execute") {
const early = await claimOrReturn("approved"); const early = await claimOrReturn("approved");
if (early !== undefined) return early; if (early !== undefined) return early;
@ -3097,7 +2888,7 @@ export function createRunExecutor(deps: ExecutorDeps) {
}); });
} }
const recordedForAsk = await recordEffect(deps, run, name, effectKey, args); 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; if (missingSecretAction.action === "return") return missingSecretAction.result;
const connectionId = args.connectionId ? String(args.connectionId) : undefined; const connectionId = args.connectionId ? String(args.connectionId) : undefined;
if (connectionId) { if (connectionId) {
@ -3719,10 +3510,6 @@ export function createRunExecutor(deps: ExecutorDeps) {
completion, completion,
runSecrets, runSecrets,
), ),
toolCallBudgetStore:
maxToolCallsPerTurn() > 0
? createRunToolCallBudgetStore(deps.prisma, runId)
: undefined,
claimSteering: scripted claimSteering: scripted
? undefined ? undefined
: async (seenIds) => { : async (seenIds) => {
@ -4175,7 +3962,7 @@ export function createRunExecutor(deps: ExecutorDeps) {
const workshopFallback = workshopFallbackAfterBlockedHandoff( const workshopFallback = workshopFallbackAfterBlockedHandoff(
proseRecovery.sanitizedText, proseRecovery.sanitizedText,
proseRecovery.blockedUpward, proseRecovery.blockedUpward,
proseRecovery.proseToolCalls, proseRecovery.handoffFailed,
); );
if (proseRecovery.proseToolCalls.length > 0) { if (proseRecovery.proseToolCalls.length > 0) {
assembled = proseRecovery.sanitizedText; 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<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( async function runSandboxCommand(
sandbox: SandboxProvider, sandbox: SandboxProvider,
computer: ComputerRef, computer: ComputerRef,
@ -4480,224 +4220,12 @@ async function runSandboxCommand(
return { stdout, stderr, code }; return { stdout, stderr, code };
} }
/** export { TURN_ATTACHMENT_UNAVAILABLE, STEERING_ATTACHMENT_UNAVAILABLE, missingTurnImagesInstruction, settleSteeringAttachmentLoads } from "./executor-attachments.js";
* The deployment key is a bearer credential for exactly one vendor, so it is handed out export { createRunWorkspaceCheckpoint } from "./executor-checkpoint.js";
* only when the provider that won the resolution above is that vendor. A provider named export { isProtectedComputerLifecycleCommand } from "./executor-shell-guard.js";
* by deployment settings or a bot override gets no key rather than another vendor's. export { type ExecutorDeps } from "./executor-types.js";
*/ export { completionMessageSegments, completionNotificationBody, completionMarksUnread, threadContextForRun, subagentMarksUnread } from "./executor-completion.js";
function deploymentKeyFor(deps: ExecutorDeps, provider: string): string | undefined { export { selectRunConnections, loadCurrentTurnImages } from "./executor-utils.js";
if (!deps.deploymentModelKey) return undefined; export { PAGE_BROWSER_TOOL_NAMES, filterPageBrowserTools, selectBuiltinToolsForRun } from "./executor-tools.js";
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 { recordEffect, completeEffect, uncertainEffectError } from "./executor-effects.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";

63
packages/adapters/src/messaging-delivery.test.ts
View File

@ -699,13 +699,9 @@ function createChannelDeps(
}); });
}), }),
createMany: vi.fn(async ({ data }: { data: Array<Record<string, unknown>> }) => { createMany: vi.fn(async ({ data }: { data: Array<Record<string, unknown>> }) => {
let count = 0; for (const item of data)
for (const item of data) {
if (rows.some((row) => row.idempotencyKey === item.idempotencyKey)) continue;
rows.push({ id: `out-${rows.length + 1}`, status: "pending", ...item }); rows.push({ id: `out-${rows.length + 1}`, status: "pending", ...item });
count += 1; return { count: data.length };
}
return { count };
}), }),
update: vi.fn(async ({ where, data }: { where: { id: string }; data: unknown }) => { update: vi.fn(async ({ where, data }: { where: { id: string }; data: unknown }) => {
const row = rows.find((candidate) => candidate.id === where.id); const row = rows.find((candidate) => candidate.id === where.id);
@ -754,65 +750,22 @@ function createChannelDeps(
} }
describe("deliverMessagingOutbound channel runs", () => { 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({ const deps = createChannelDeps({
messages: [ messages: [
{ {
id: "ask-1", id: "ask-1",
blocks: [{ kind: "ask", text: "Details?", status: "answered", answer: "Private detail" }], 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); 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); await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
expect(deps.sendToThread).toHaveBeenCalledWith( expect(deps.rows).toEqual([]);
{ threadId: "sendblue:grp-1", body: "Alice's agent: Working on it…" }, expect(deps.sendToThread).not.toHaveBeenCalled();
context, expect(deps.contextMessages).toEqual([]);
); expect(deps.sendUserMessage).not.toHaveBeenCalled();
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" }),
]),
);
}); });
it("can still publish a group reply alongside an unanswered ask", async () => { it("can still publish a group reply alongside an unanswered ask", async () => {

112
packages/adapters/src/pi-model-config.ts
View File

@ -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 Agent, type AgentToolResult } from "@earendil-works/pi-agent-core";
import { import {
type AgentMessage, type AgentMessage,
type AgentToolExecutionResult,
} from "@earendil-works/pi-agent-core"; } from "@earendil-works/pi-agent-core";
import type { AgentToolExecutionResult } from "@rakazo/adapter-kit";
import type { import type {
AgentRuntimeEvent, AgentRuntimeEvent,
AgentRunRequest, AgentRunRequest,
AgentSteeringMessage, AgentSteeringMessage,
ToolCallBudgetStore,
} from "@rakazo/adapter-kit"; } from "@rakazo/adapter-kit";
import { getLogger } from "@rakazo/logging"; import { getLogger } from "@rakazo/logging";
import { DEFAULT_OPENROUTER_MODEL_ID } from "./deployment-model.js"; import { DEFAULT_OPENROUTER_MODEL_ID } from "./deployment-model.js";
@ -67,7 +66,7 @@ export interface ToolCallBudget {
inFlight: number; 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<string, ToolCallBudget>(); const toolCallBudgetsByRun = new Map<string, ToolCallBudget>();
/** Optional self-host fuse. Unset, empty, or 0 means unlimited (default). */ /** 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); return Math.floor(parsed);
} }
export async function loadToolCallBudget( export function toolCallBudgetFor(runId: string): ToolCallBudget {
runId: string,
store?: ToolCallBudgetStore,
): Promise<ToolCallBudget> {
const limit = maxToolCallsPerTurn();
if (limit <= 0) {
return { count: 0, exceeded: false, limit: 0, inFlight: 0 };
}
const existing = toolCallBudgetsByRun.get(runId); const existing = toolCallBudgetsByRun.get(runId);
if (existing) { if (existing) {
existing.inFlight = 0; existing.inFlight = 0;
existing.limit = limit; existing.limit = maxToolCallsPerTurn();
return existing; return existing;
} }
const budget: ToolCallBudget = {
let count = 0; count: 0,
let exceeded = false; exceeded: false,
if (store) { limit: maxToolCallsPerTurn(),
const snapshot = await store.load(); inFlight: 0,
if (snapshot) { };
count = snapshot.count; if (budget.limit > 0) toolCallBudgetsByRun.set(runId, budget);
exceeded = snapshot.exceeded;
}
}
const budget: ToolCallBudget = { count, exceeded, limit, inFlight: 0 };
toolCallBudgetsByRun.set(runId, budget);
return budget; return budget;
} }
export async function releaseToolCallBudget( export function releaseToolCallBudget(runId: string, keepForResume: boolean) {
runId: string, if (!keepForResume) toolCallBudgetsByRun.delete(runId);
keepForResume: boolean,
store?: ToolCallBudgetStore,
): Promise<void> {
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();
} }
/** @internal Simulates a fresh worker process in unit tests. */ export function maybeAbortToolCallBudget(host: {
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: {
toolCallBudget: ToolCallBudget; toolCallBudget: ToolCallBudget;
abortTurn(): void; abortTurn(): void;
}): void { }): void {
@ -154,8 +111,7 @@ function maybeAbortToolCallBudget(host: {
export function beginToolCall(host: { export function beginToolCall(host: {
toolCallBudget: ToolCallBudget; toolCallBudget: ToolCallBudget;
abortTurn(): void; abortTurn(): void;
queue: { push(event: { type: string; text: string }): void }; queue: { push(event: AgentRuntimeEvent): void };
request?: Pick<AgentRunRequest, "runId" | "toolCallBudgetStore">;
}): boolean { }): boolean {
const budget = host.toolCallBudget; const budget = host.toolCallBudget;
if (budget.limit <= 0) { if (budget.limit <= 0) {
@ -169,7 +125,6 @@ export function beginToolCall(host: {
budget.count += 1; budget.count += 1;
if (budget.count <= budget.limit) { if (budget.count <= budget.limit) {
budget.inFlight += 1; budget.inFlight += 1;
persistToolCallBudget(host.request?.runId, budget, host.request?.toolCallBudgetStore);
return true; return true;
} }
if (!budget.exceeded) { if (!budget.exceeded) {
@ -179,7 +134,6 @@ export function beginToolCall(host: {
text: `Stopped: more than ${budget.limit} tool calls in one turn.`, text: `Stopped: more than ${budget.limit} tool calls in one turn.`,
}); });
} }
persistToolCallBudget(host.request?.runId, budget, host.request?.toolCallBudgetStore);
maybeAbortToolCallBudget(host); maybeAbortToolCallBudget(host);
return false; return false;
} }
@ -187,10 +141,8 @@ export function beginToolCall(host: {
export function endToolCall(host: { export function endToolCall(host: {
toolCallBudget: ToolCallBudget; toolCallBudget: ToolCallBudget;
abortTurn(): void; abortTurn(): void;
request?: Pick<AgentRunRequest, "runId" | "toolCallBudgetStore">;
}): void { }): void {
host.toolCallBudget.inFlight = Math.max(0, host.toolCallBudget.inFlight - 1); host.toolCallBudget.inFlight = Math.max(0, host.toolCallBudget.inFlight - 1);
persistToolCallBudget(host.request?.runId, host.toolCallBudget, host.request?.toolCallBudgetStore);
maybeAbortToolCallBudget(host); maybeAbortToolCallBudget(host);
} }
@ -200,7 +152,7 @@ export function toolCallBudgetExceededMessage(limit: number): string {
// --- Model resolution --- // --- Model resolution ---
function configuredOpenRouterModel(id: string): Model<"openai-completions"> { export function configuredOpenRouterModel(id: string): Model<"openai-completions"> {
return { return {
id, id,
name: id, name: id,
@ -252,7 +204,7 @@ export function modelsForRequest(
return catalogModels(); return catalogModels();
} }
function resolveRuntimeModel(modelConfig: AgentRunRequest["model"]): { export function resolveRuntimeModel(modelConfig: AgentRunRequest["model"]): {
provider: string; provider: string;
modelId: string; modelId: string;
models: Models; models: Models;
@ -290,7 +242,7 @@ function resolveRuntimeModel(modelConfig: AgentRunRequest["model"]): {
// --- Image helpers --- // --- Image helpers ---
function toPiImages(images: AgentRunRequest["currentTurnImages"]) { export function toPiImages(images: AgentRunRequest["currentTurnImages"]) {
return (images ?? []).map((image) => ({ return (images ?? []).map((image) => ({
type: "image" as const, type: "image" as const,
data: Buffer.from(image.data).toString("base64"), data: Buffer.from(image.data).toString("base64"),
@ -300,7 +252,7 @@ function toPiImages(images: AgentRunRequest["currentTurnImages"]) {
// --- Steering message removal --- // --- Steering message removal ---
function withoutSteeringMessages( export function withoutSteeringMessages(
history: AgentRunRequest["history"], history: AgentRunRequest["history"],
steering: AgentSteeringMessage[], steering: AgentSteeringMessage[],
): AgentRunRequest["history"] { ): AgentRunRequest["history"] {
@ -385,7 +337,7 @@ export function pruneComputerScreenshotContext(
return transformed ?? messages; return transformed ?? messages;
} }
function imagePartCount(message: AgentMessage): number { export function imagePartCount(message: AgentMessage): number {
if (!("content" in message) || !Array.isArray(message.content)) return 0; if (!("content" in message) || !Array.isArray(message.content)) return 0;
return message.content.filter( return message.content.filter(
(part: unknown) => (part: unknown) =>
@ -393,7 +345,7 @@ function imagePartCount(message: AgentMessage): number {
).length; ).length;
} }
function isComputerScreenshotMessage( export function isComputerScreenshotMessage(
message: AgentMessage | undefined, message: AgentMessage | undefined,
): message is Extract<AgentMessage, { role: "toolResult" }> { ): message is Extract<AgentMessage, { role: "toolResult" }> {
if (message?.role !== "toolResult" || !message.content.some((part) => (part as {type?: string}).type === "image")) { if (message?.role !== "toolResult" || !message.content.some((part) => (part as {type?: string}).type === "image")) {
@ -410,7 +362,7 @@ function isComputerScreenshotMessage(
// --- Type guards --- // --- Type guards ---
function isAgentToolExecutionResult(result: unknown): result is AgentToolExecutionResult { export function isAgentToolExecutionResult(result: unknown): result is AgentToolExecutionResult {
if ( if (
!result || !result ||
typeof result !== "object" || typeof result !== "object" ||
@ -461,7 +413,7 @@ export function jsonSchemaParameters(
: Type.Object(fields); : Type.Object(fields);
} }
function enumUnion(values: readonly unknown[]) { export function enumUnion(values: readonly unknown[]) {
const members = values.map((value) => const members = values.map((value) =>
value === null value === null
? Type.Null() ? Type.Null()
@ -514,7 +466,7 @@ export function jsonField(spec: unknown): ReturnType<typeof Type.String> {
// --- Result helpers --- // --- Result helpers ---
function summarizeToolResult(result: unknown) { export function summarizeToolResult(result: unknown) {
try { try {
const text = JSON.stringify(result); const text = JSON.stringify(result);
if (!text) return "ok"; if (!text) return "ok";
@ -524,7 +476,7 @@ function summarizeToolResult(result: unknown) {
} }
} }
function boundAgentToolResult<T>( export function boundAgentToolResult<T>(
result: AgentToolResult<T>, result: AgentToolResult<T>,
): AgentToolResult<T> { ): AgentToolResult<T> {
if (!result || !Array.isArray(result.content)) return result; if (!result || !Array.isArray(result.content)) return result;
@ -534,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 ""; if (!message || typeof message !== "object" || !("content" in message)) return "";
const content = (message as { content?: unknown }).content; const content = (message as { content?: unknown }).content;
if (typeof content === "string") return content; if (typeof content === "string") return content;
@ -554,7 +506,7 @@ function assistantText(message: unknown): string {
// --- Sanitization --- // --- Sanitization ---
function sanitizeSensitiveText(message: string) { export function sanitizeSensitiveText(message: string) {
return message return message
.replace(/sk-or-v1-[a-zA-Z0-9]+/g, "[redacted]") .replace(/sk-or-v1-[a-zA-Z0-9]+/g, "[redacted]")
.replace(/sk-[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(); const raw = String(value ?? "").trim();
if (!raw) return raw; if (!raw) return raw;
try { try {
@ -589,7 +541,7 @@ function redactActivityUrl(value: unknown): string {
} }
} }
function sanitizeError(message: string) { export function sanitizeError(message: string) {
return sanitizeSensitiveText(message); return sanitizeSensitiveText(message);
} }
@ -610,7 +562,7 @@ export function isOpenCodeProvider(provider: string): boolean {
const OPENCODE_SESSION_ERROR = const OPENCODE_SESSION_ERROR =
"OpenCode rejected this chat session. Send the message again."; "OpenCode rejected this chat session. Send the message again.";
function looksLikeOpenCodeSessionError(message: string): boolean { export function looksLikeOpenCodeSessionError(message: string): boolean {
return ( return (
/x-opencode-session/i.test(message) || /x-opencode-session/i.test(message) ||
/session\s*(id|header|required|missing|invalid|expired|stale)/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); const sanitized = sanitizeError(message);
if (isOpenCodeProvider(provider) && looksLikeOpenCodeSessionError(sanitized)) { if (isOpenCodeProvider(provider) && looksLikeOpenCodeSessionError(sanitized)) {
return OPENCODE_SESSION_ERROR; return OPENCODE_SESSION_ERROR;
@ -628,7 +580,7 @@ function sanitizeProviderError(provider: string, message: string): string {
// --- Queue & gate --- // --- Queue & gate ---
interface EventQueue { export interface EventQueue {
push(event: AgentRuntimeEvent): void; push(event: AgentRuntimeEvent): void;
fail(error: Error): void; fail(error: Error): void;
close(): void; close(): void;

651
packages/adapters/src/pi-runtime.ts
View File

@ -5,73 +5,65 @@ import {
type AgentTool, type AgentTool,
type AgentToolResult, type AgentToolResult,
} from "@earendil-works/pi-agent-core"; } from "@earendil-works/pi-agent-core";
import { import { type Api, type Model, Type } from "@earendil-works/pi-ai";
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 { import type {
AdapterContext, AdapterContext,
AgentRunRequest, AgentRunRequest,
AgentRuntime, AgentRuntime,
AgentRuntimeEvent, AgentRuntimeEvent,
AgentSteeringMessage,
AgentToolCompletion, AgentToolCompletion,
AgentToolExecutionResult,
ConnectorTool, ConnectorTool,
} from "@rakazo/adapter-kit"; } from "@rakazo/adapter-kit";
import { getLogger } from "@rakazo/logging"; import { getLogger } from "@rakazo/logging";
import { isToolPauseResult } from "./approval-effect.js"; import { isToolPauseResult } from "./approval-effect.js";
import { builtinAgentTools, DELEGATION_TOOL_NAMES } from "./builtin-tools.js"; import { builtinAgentTools, DELEGATION_TOOL_NAMES } from "./builtin-tools.js";
import { DEFAULT_OPENROUTER_MODEL_ID } from "./deployment-model.js";
import { import {
normalizeOpenAiToolParameters, normalizeOpenAiToolParameters,
openAiToolParametersNeedNormalization, openAiToolParametersNeedNormalization,
} from "./openai-tool-parameters.js"; } from "./openai-tool-parameters.js";
import { PiRuntimeCredentialStore, toOAuthCredential } from "./pi-credentials.js"; import { billedPromptTokens, clipToolResultText } from "./pi-runtime-limits.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 { import {
PiJsonlSessionRecorder, PiJsonlSessionRecorder,
type PiSessionHandle, type PiSessionHandle,
type PiSessionRecorder, type PiSessionRecorder,
} from "./pi-session.js"; } from "./pi-session.js";
import { textContentArg } from "./tool-text.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> }>(); const running = new Map<string, { controller: AbortController; work: Promise<void> }>();
// 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 // 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 // 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. // 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; const MAX_PARALLEL_SUBAGENTS = 4;
// Some OpenAI-compatible models return EOS immediately after a tool result // Some OpenAI-compatible models return EOS immediately after a tool result
// instead of taking another assistant turn. A bounded internal follow-up keeps // 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."; "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 = const TOOL_FINAL_RESPONSE_FALLBACK =
"I completed the tool step but could not produce a final response. Please ask me to continue."; "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 // Pi forwards these names to OpenAI Responses, whose function-name contract is
// ^[a-zA-Z0-9_-]+$ with a maximum length of 64 characters. // ^[a-zA-Z0-9_-]+$ with a maximum length of 64 characters.
const AGENT_TOOL_NAME_PATTERN = /^[a-zA-Z0-9_-]+$/; const AGENT_TOOL_NAME_PATTERN = /^[a-zA-Z0-9_-]+$/;
const MAX_AGENT_TOOL_NAME_LENGTH = 64; const MAX_AGENT_TOOL_NAME_LENGTH = 64;
const FALLBACK_AGENT_TOOL_NAME = "connector_tool"; const FALLBACK_AGENT_TOOL_NAME = "connector_tool";
export { maxToolCallsPerTurn } from "./pi-model-config.js";
export interface PiAgentRuntimeOptions { export interface PiAgentRuntimeOptions {
/** Directory where Pi JSONL sessions are written. Omit to disable recording. */ /** Directory where Pi JSONL sessions are written. Omit to disable recording. */
sessionRoot?: string; sessionRoot?: string;
@ -183,7 +159,7 @@ export class PiAgentRuntime implements AgentRuntime {
const toolDefs = request.tools.length ? request.tools : builtinAgentTools; const toolDefs = request.tools.length ? request.tools : builtinAgentTools;
const nestedAgents = new Set<Agent>(); const nestedAgents = new Set<Agent>();
const completionModel = modelForCompletion(model, request.model.maxTokens); const completionModel = modelForCompletion(model, request.model.maxTokens);
trackedBudget = await loadToolCallBudget(request.runId, request.toolCallBudgetStore); trackedBudget = toolCallBudgetFor(request.runId);
const host: ToolHost = { const host: ToolHost = {
queue, queue,
request, request,
@ -436,11 +412,7 @@ export class PiAgentRuntime implements AgentRuntime {
if (trackedBudget) { if (trackedBudget) {
const keepForResume = const keepForResume =
(signal.aborted || Boolean(resumeHost?.pausePending)) && !trackedBudget.exceeded; (signal.aborted || Boolean(resumeHost?.pausePending)) && !trackedBudget.exceeded;
await releaseToolCallBudget( releaseToolCallBudget(request.runId, keepForResume);
request.runId,
keepForResume,
request.toolCallBudgetStore,
);
} }
} }
})(); })();
@ -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<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[] { function toAgentTools(toolDefs: readonly ConnectorTool[], host: ToolHost): AgentTool[] {
const names = normalizeAgentToolNames(toolDefs); const names = normalizeAgentToolNames(toolDefs);
return toolDefs.map((tool, index) => toAgentTool(tool, host, names[index]!)); 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<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 { function toAgentTool(tool: ConnectorTool, host: ToolHost, exposedName: string): AgentTool {
return { return {
name: exposedName, name: exposedName,
@ -1265,397 +1089,14 @@ function builtinParameters(tool: ConnectorTool) {
return undefined; return undefined;
} }
/** Keep recent visual state while respecting an optional model image budget. */ export {
export function pruneComputerScreenshotContext( conversationSessionId,
messages: AgentMessage[], isOpenCodeProvider,
maxImagesPerPrompt?: number, jsonField,
): AgentMessage[] { jsonSchemaParameters,
const imageLimit = maxToolCallsPerTurn,
maxImagesPerPrompt === undefined modelsForRequest,
? undefined prepareRequestSecretArguments,
: Number.isFinite(maxImagesPerPrompt) pruneComputerScreenshotContext,
? Math.max(0, Math.floor(maxImagesPerPrompt)) reliableStreamOptions,
: DEFAULT_COMPUTER_SCREENSHOTS_TO_KEEP; } from "./pi-model-config.js";
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 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;
}

6
packages/adapters/src/prose-handoff-recovery.test.ts
View File

@ -98,15 +98,13 @@ describe("tryRecoverProseHandoff", () => {
describe("workshopFallbackAfterBlockedHandoff", () => { describe("workshopFallbackAfterBlockedHandoff", () => {
it("returns a workshop line when only blocked handoff prose remains", () => { it("returns a workshop line when only blocked handoff prose remains", () => {
expect( expect(
workshopFallbackAfterBlockedHandoff("", true, [ workshopFallbackAfterBlockedHandoff("", true, false),
{ name: "handoff_to_bot", parameters: { message: "same" } },
]),
).toBe(INBOUND_HANDOFF_WORKSHOP_FALLBACK); ).toBe(INBOUND_HANDOFF_WORKSHOP_FALLBACK);
}); });
it("does not override real specialist prose", () => { it("does not override real specialist prose", () => {
expect( expect(
workshopFallbackAfterBlockedHandoff("Drei Richtungen: ruhig, mutig, verspielt.", true, []), workshopFallbackAfterBlockedHandoff("Drei Richtungen: ruhig, mutig, verspielt.", true, false),
).toBeUndefined(); ).toBeUndefined();
}); });
}); });