diff --git a/runtime/typescript/packages/anthropic/src/index.ts b/runtime/typescript/packages/anthropic/src/index.ts index 6602e6868..29edebcda 100644 --- a/runtime/typescript/packages/anthropic/src/index.ts +++ b/runtime/typescript/packages/anthropic/src/index.ts @@ -7,8 +7,9 @@ */ export { AnthropicExecutor } from "./executor.js"; -export { AnthropicProcessor, processResponse } from "./processor.js"; +export { AnthropicProcessor, processResponse, processStream } from "./processor.js"; export { buildChatArgs, messageToWire, toolsToWire, outputsToWire } from "./wire.js"; +export { listModels, modelInfoFromWire } from "./models.js"; // Auto-register on import import { registerExecutor, registerProcessor } from "@prompty/core"; diff --git a/runtime/typescript/packages/anthropic/src/models.ts b/runtime/typescript/packages/anthropic/src/models.ts new file mode 100644 index 000000000..1406bd27f --- /dev/null +++ b/runtime/typescript/packages/anthropic/src/models.ts @@ -0,0 +1,75 @@ +/** + * Anthropic model discovery. + * + * @module + */ + +import Anthropic from "@anthropic-ai/sdk"; +import { + ApiKeyConnection, + ModelInfo, + ReferenceConnection, + createModelInfo, + enrichModelInfo, + getConnection, +} from "@prompty/core"; +import type { Connection } from "@prompty/core"; + +interface AnthropicModelsClient { + models: { + list(params?: { limit?: number; after_id?: string }): Promise>; + }; +} + +/** Map one raw Anthropic model response into the canonical generated model. */ +export function modelInfoFromWire(raw: Record): ModelInfo { + return createModelInfo(enrichModelInfo("anthropic", { + id: typeof raw.id === "string" ? raw.id : "", + displayName: typeof raw.display_name === "string" ? raw.display_name : undefined, + ownedBy: "anthropic", + contextWindow: typeof raw.context_length === "number" ? raw.context_length : undefined, + inputModalities: stringArray(raw.input_modalities), + outputModalities: stringArray(raw.output_modalities), + additionalProperties: { ...raw }, + })); +} + +/** List every model available from the Anthropic Models API. */ +export async function listModels(connection: Connection): Promise { + const client = buildClient(connection); + const page = await client.models.list({ limit: 100 }); + const models: ModelInfo[] = []; + + for await (const raw of page) { + models.push(modelInfoFromWire(asRecord(raw))); + } + + return models; +} + +function buildClient(connection: Connection): AnthropicModelsClient { + if (connection instanceof ReferenceConnection) { + return getConnection(connection.name) as AnthropicModelsClient; + } + if (!(connection instanceof ApiKeyConnection)) { + throw new Error( + `Connection kind '${connection.kind}' is not supported by Anthropic listModels. ` + + "Use 'key' for API key auth or 'reference' with registerConnection() for pre-configured clients.", + ); + } + return new Anthropic({ + apiKey: connection.apiKey || process.env.ANTHROPIC_API_KEY, + ...(connection.endpoint ? { baseURL: connection.endpoint } : {}), + }); +} + +function asRecord(value: unknown): Record { + if (typeof value !== "object" || value === null || Array.isArray(value)) { + throw new Error("Anthropic model listing returned a non-object model entry."); + } + return value as Record; +} + +function stringArray(value: unknown): string[] | undefined { + return Array.isArray(value) ? value.map(String) : undefined; +} diff --git a/runtime/typescript/packages/anthropic/src/processor.ts b/runtime/typescript/packages/anthropic/src/processor.ts index 4b95b3890..2d57855bf 100644 --- a/runtime/typescript/packages/anthropic/src/processor.ts +++ b/runtime/typescript/packages/anthropic/src/processor.ts @@ -13,7 +13,16 @@ import type { Prompty } from "@prompty/core"; import type { Processor } from "@prompty/core"; import type { ToolCall } from "@prompty/core"; -import { traceSpan } from "@prompty/core"; +import { + ErrorChunk, + InvocationUsage, + StreamChunk, + TextChunk, + ThinkingChunk, + ToolChunk, + UsageChunk, + traceSpan, +} from "@prompty/core"; import { createStructuredResult } from "@prompty/core"; export class AnthropicProcessor implements Processor { @@ -29,6 +38,10 @@ export class AnthropicProcessor implements Processor { return result; }); } + + processStream(response: AsyncIterable): AsyncIterable { + return processStream(response); + } } /** @@ -39,7 +52,7 @@ export function processResponse(agent: Prompty, response: unknown): unknown { // Streaming response — return content-extracting async generator if (isAsyncIterable(response)) { - return streamGenerator(response); + return legacyStreamGenerator(processStream(response)); } const r = response as Record; @@ -76,53 +89,108 @@ function isAsyncIterable(value: unknown): value is AsyncIterable { * * Tool calls are accumulated and yielded at the end of the stream. */ -async function* streamGenerator( +export async function* processStream( response: AsyncIterable, -): AsyncGenerator { +): AsyncGenerator { const toolCallAcc: Map< number, { id: string; name: string; arguments: string } > = new Map(); + let inputTokens: number | undefined; + let outputTokens: number | undefined; + + try { + for await (const event of response) { + const e = event as Record; + const eventType = e.type as string | undefined; + + if (eventType === "message_start") { + const message = e.message as Record | undefined; + const usage = message?.usage as Record | undefined; + inputTokens = numberValue(usage?.input_tokens) ?? inputTokens; + } else if (eventType === "message_delta") { + const usage = e.usage as Record | undefined; + outputTokens = numberValue(usage?.output_tokens) ?? outputTokens; + } else if (eventType === "content_block_delta") { + const delta = e.delta as Record | undefined; + if (!delta) continue; - for await (const event of response) { - const e = event as Record; - const eventType = e.type as string | undefined; - - if (eventType === "content_block_delta") { - const delta = e.delta as Record | undefined; - if (!delta) continue; - - if (delta.type === "text_delta") { - yield delta.text as string; - } else if (delta.type === "input_json_delta") { - // Accumulate partial JSON for tool arguments - const idx = e.index as number; - const acc = toolCallAcc.get(idx); - if (acc) { - acc.arguments += (delta.partial_json ?? "") as string; + if (delta.type === "text_delta" && typeof delta.text === "string" && delta.text) { + yield new TextChunk({ value: delta.text }); + } else if (delta.type === "thinking_delta" && typeof delta.thinking === "string" && delta.thinking) { + yield new ThinkingChunk({ value: delta.thinking }); + } else if (delta.type === "input_json_delta") { + const idx = typeof e.index === "number" ? e.index : 0; + const acc = toolCallAcc.get(idx); + if (acc && typeof delta.partial_json === "string") { + acc.arguments += delta.partial_json; + } } - } - } else if (eventType === "content_block_start") { - const block = e.content_block as Record | undefined; - if (block?.type === "tool_use") { - const idx = e.index as number; - toolCallAcc.set(idx, { - id: (block.id ?? "") as string, - name: (block.name ?? "") as string, - arguments: "", + } else if (eventType === "content_block_start") { + const block = e.content_block as Record | undefined; + if (block?.type === "tool_use") { + const idx = typeof e.index === "number" ? e.index : 0; + toolCallAcc.set(idx, { + id: stringValue(block.id), + name: stringValue(block.name), + arguments: "", + }); + } + } else if (eventType === "error") { + const error = e.error as Record | undefined; + yield new ErrorChunk({ + message: stringValue(error?.message) || "Anthropic stream failed", }); + return; } } + } catch (error) { + yield new ErrorChunk({ + message: error instanceof Error ? error.message : String(error), + }); + return; } - // Yield accumulated tool calls at the end of the stream const sortedIndices = [...toolCallAcc.keys()].sort((a, b) => a - b); for (const idx of sortedIndices) { const tc = toolCallAcc.get(idx)!; - yield { id: tc.id, name: tc.name, arguments: tc.arguments } as ToolCall; + yield ToolChunk.load({ kind: "tool", toolCall: tc }); + } + if (inputTokens !== undefined || outputTokens !== undefined) { + const input = inputTokens ?? 0; + const output = outputTokens ?? 0; + yield new UsageChunk({ + usage: new InvocationUsage({ + inputTokens: input, + outputTokens: output, + totalTokens: input + output, + }), + }); } } +async function* legacyStreamGenerator( + chunks: AsyncIterable, +): AsyncGenerator { + for await (const chunk of chunks) { + if (chunk instanceof TextChunk) { + yield chunk.value; + } else if (chunk instanceof ToolChunk) { + yield chunk.toolCall; + } else if (chunk instanceof ErrorChunk) { + throw new Error(chunk.message); + } + } +} + +function stringValue(value: unknown): string { + return typeof value === "string" ? value : ""; +} + +function numberValue(value: unknown): number | undefined { + return typeof value === "number" && Number.isFinite(value) ? value : undefined; +} + // --------------------------------------------------------------------------- // Non-streaming response processing // --------------------------------------------------------------------------- diff --git a/runtime/typescript/packages/anthropic/tests/e2e.test.ts b/runtime/typescript/packages/anthropic/tests/e2e.test.ts index 84e78c9d6..7d3bb1735 100644 --- a/runtime/typescript/packages/anthropic/tests/e2e.test.ts +++ b/runtime/typescript/packages/anthropic/tests/e2e.test.ts @@ -17,9 +17,14 @@ import { turn, registerConnection, clearConnections, + ErrorChunk, + TextChunk, + ThinkingChunk, + ToolChunk, + UsageChunk, } from "@prompty/core"; import { AnthropicExecutor } from "../src/executor.js"; -import { AnthropicProcessor, processResponse } from "../src/processor.js"; +import { AnthropicProcessor, processResponse, processStream } from "../src/processor.js"; import { buildChatArgs, messageToWire, toolsToWire, outputsToWire } from "../src/wire.js"; import { registerExecutor, registerProcessor } from "@prompty/core"; import { Message } from "@prompty/core"; @@ -406,6 +411,67 @@ describe("processor", () => { // --------------------------------------------------------------------------- describe("streaming processor", () => { + it("emits canonical text, thinking, tool, usage, and error chunks", async () => { + async function* stream() { + yield { type: "message_start", message: { usage: { input_tokens: 4 } } }; + yield { + type: "content_block_start", + index: 0, + content_block: { type: "tool_use", id: "toolu_1", name: "lookup" }, + }; + yield { + type: "content_block_delta", + index: 1, + delta: { type: "text_delta", text: "Hello" }, + }; + yield { + type: "content_block_delta", + index: 1, + delta: { type: "thinking_delta", thinking: "Considering" }, + }; + yield { + type: "content_block_delta", + index: 0, + delta: { type: "input_json_delta", partial_json: '{"q":"test"}' }, + }; + yield { type: "message_delta", usage: { output_tokens: 3 } }; + } + + const chunks: unknown[] = []; + for await (const chunk of processStream(stream())) chunks.push(chunk); + + expect(chunks[0]).toBeInstanceOf(TextChunk); + expect((chunks[0] as TextChunk).value).toBe("Hello"); + expect(chunks[1]).toBeInstanceOf(ThinkingChunk); + expect((chunks[1] as ThinkingChunk).value).toBe("Considering"); + expect(chunks[2]).toBeInstanceOf(ToolChunk); + expect((chunks[2] as ToolChunk).toolCall).toMatchObject({ + id: "toolu_1", + name: "lookup", + arguments: '{"q":"test"}', + }); + expect(chunks[3]).toBeInstanceOf(UsageChunk); + expect((chunks[3] as UsageChunk).usage).toMatchObject({ + inputTokens: 4, + outputTokens: 3, + totalTokens: 7, + }); + + async function* failed() { + yield { type: "error", error: { message: "overloaded" } }; + yield { + type: "content_block_delta", + index: 0, + delta: { type: "text_delta", text: "must not be emitted" }, + }; + } + const errors: unknown[] = []; + for await (const chunk of processStream(failed())) errors.push(chunk); + expect(errors).toHaveLength(1); + expect(errors[0]).toBeInstanceOf(ErrorChunk); + expect((errors[0] as ErrorChunk).message).toBe("overloaded"); + }); + it("yields text deltas from content_block_delta events", async () => { const events = [ { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } }, diff --git a/runtime/typescript/packages/anthropic/tests/models.test.ts b/runtime/typescript/packages/anthropic/tests/models.test.ts new file mode 100644 index 000000000..d736674a3 --- /dev/null +++ b/runtime/typescript/packages/anthropic/tests/models.test.ts @@ -0,0 +1,67 @@ +import { readFileSync } from "node:fs"; +import { resolve } from "node:path"; + +import { + AnonymousConnection, + ModelInfo, + ReferenceConnection, + clearConnections, + registerConnection, +} from "@prompty/core"; +import { afterEach, describe, expect, it } from "vitest"; + +import { listModels, modelInfoFromWire } from "../src/models.js"; + +interface DiscoveryVector { + name: string; + provider: string; + input: Record; + expected: Record; +} + +const vectorFile = resolve( + import.meta.dirname, + "../../../../../spec/vectors/discovery/discovery_vectors.json", +); +const vectors = ( + JSON.parse(readFileSync(vectorFile, "utf8")) as { vectors: DiscoveryVector[] } +).vectors.filter((vector) => vector.provider === "anthropic"); + +afterEach(() => clearConnections()); + +describe("Anthropic discovery vectors", () => { + for (const vector of vectors) { + it(vector.name, () => { + expect(modelInfoFromWire(vector.input).save()).toEqual(vector.expected); + }); + } +}); + +describe("listModels (Anthropic)", () => { + it("consumes every SDK pagination item", async () => { + registerConnection("anthropic-models", { + models: { + list: async () => ({ + async *[Symbol.asyncIterator]() { + yield { id: "claude-a", display_name: "Claude A", type: "model" }; + yield { id: "claude-b", display_name: "Claude B", type: "model" }; + }, + }), + }, + }); + + const models = await listModels( + new ReferenceConnection({ name: "anthropic-models" }), + ); + + expect(models).toHaveLength(2); + expect(models[0]).toBeInstanceOf(ModelInfo); + expect(models.map((model) => model.id)).toEqual(["claude-a", "claude-b"]); + }); + + it("rejects unsupported connection kinds", async () => { + await expect(listModels(new AnonymousConnection())).rejects.toThrow( + /not supported/, + ); + }); +}); diff --git a/runtime/typescript/packages/core/src/core/discovery.ts b/runtime/typescript/packages/core/src/core/discovery.ts new file mode 100644 index 000000000..2572310a3 --- /dev/null +++ b/runtime/typescript/packages/core/src/core/discovery.ts @@ -0,0 +1,66 @@ +/** + * Shared model-discovery capability enrichment. + * + * Capability data comes from the canonical cross-runtime dataset. Provider + * values are preserved; the dataset only fills fields omitted by the provider. + * + * @module + */ + +import capabilities from "../../../../../../spec/data/model_capabilities.json"; + +import { ModelInfo } from "../model/model/model-info.js"; + +interface CapabilityEntry { + prefix: string; + contextWindow?: number; + inputModalities?: string[]; + outputModalities?: string[]; +} + +interface CapabilityDataset { + providers: Record; +} + +const dataset = capabilities as CapabilityDataset; + +function isTokenBoundary(id: string, prefix: string): boolean { + if (!id.startsWith(prefix)) return false; + if (id.length === prefix.length) return true; + return !/[A-Za-z0-9]/.test(id[prefix.length]); +} + +function findCapabilities(provider: string, id: string): CapabilityEntry | undefined { + return (dataset.providers[provider] ?? []) + .filter((entry) => isTokenBoundary(id, entry.prefix)) + .sort((left, right) => right.prefix.length - left.prefix.length)[0]; +} + +/** + * Fill omitted model capability fields from the shared dataset. + * + * Accepts a partial generated `ModelInfo` initializer so an explicit empty + * modality list remains distinguishable from an omitted field. + */ +export function enrichModelInfo(provider: string, input: Partial): Partial { + const known = findCapabilities(provider, input.id ?? ""); + if (!known) return { ...input }; + + return { + ...input, + contextWindow: input.contextWindow ?? known.contextWindow, + inputModalities: input.inputModalities === undefined ? known.inputModalities : input.inputModalities, + outputModalities: input.outputModalities === undefined ? known.outputModalities : input.outputModalities, + }; +} + +/** + * Construct a generated `ModelInfo` without materializing omitted optional + * collection fields as empty arrays. + */ +export function createModelInfo(input: Partial): ModelInfo { + const info = new ModelInfo(input); + if (input.inputModalities === undefined) delete info.inputModalities; + if (input.outputModalities === undefined) delete info.outputModalities; + return info; +} diff --git a/runtime/typescript/packages/core/src/core/index.ts b/runtime/typescript/packages/core/src/core/index.ts index 8f12035f1..3422d5045 100644 --- a/runtime/typescript/packages/core/src/core/index.ts +++ b/runtime/typescript/packages/core/src/core/index.ts @@ -3,11 +3,13 @@ export * from "./interfaces.js"; export * from "./registry.js"; export * from "./connections.js"; export { load, defaultSaveContext, type LoadOptions } from "./loader.js"; +export { createModelInfo, enrichModelInfo } from "./discovery.js"; export { validateInputs, render, parse, process, + processStream, prepare, run, turn, @@ -51,3 +53,7 @@ export { isStructuredResult, cast, } from "./structured.js"; +export * from "./turn-engine-cancellation.js"; +export * from "./turn-engine-context.js"; +export * from "./turn-engine-ports.js"; +export * from "./turn-engine.js"; diff --git a/runtime/typescript/packages/core/src/core/interfaces.ts b/runtime/typescript/packages/core/src/core/interfaces.ts index 9e36d28b4..9b3afeaa8 100644 --- a/runtime/typescript/packages/core/src/core/interfaces.ts +++ b/runtime/typescript/packages/core/src/core/interfaces.ts @@ -10,6 +10,7 @@ import type { Prompty } from "../model/agent/prompty.js"; import type { Message } from "./types.js"; +import type { StreamChunk } from "../model/events/stream-chunk.js"; // --------------------------------------------------------------------------- // Renderer @@ -83,4 +84,12 @@ export interface Executor { */ export interface Processor { process(agent: Prompty, response: unknown): Promise; + + /** + * Convert raw provider stream items into canonical generated stream chunks. + * Providers that support streaming should implement this method. + */ + processStream?( + response: AsyncIterable, + ): AsyncIterable; } diff --git a/runtime/typescript/packages/core/src/core/loader.ts b/runtime/typescript/packages/core/src/core/loader.ts index 6a0e3b638..54bce001f 100644 --- a/runtime/typescript/packages/core/src/core/loader.ts +++ b/runtime/typescript/packages/core/src/core/loader.ts @@ -11,6 +11,7 @@ import { existsSync, readFileSync, realpathSync } from "node:fs"; import { dirname, extname, isAbsolute, relative, resolve } from "node:path"; import matter from "gray-matter"; +import { parse as parseYaml } from "yaml"; import { LoadContext, SaveContext } from "../model/context.js"; import { Prompty } from "../model/agent/prompty.js"; @@ -216,6 +217,8 @@ function loadFileContent(path: string): unknown { if (ext === ".json") { return JSON.parse(raw); } - // For YAML we return raw string — the loader handles YAML natively + if (ext === ".yaml" || ext === ".yml") { + return parseYaml(raw); + } return raw; } diff --git a/runtime/typescript/packages/core/src/core/pipeline.ts b/runtime/typescript/packages/core/src/core/pipeline.ts index fec4e56da..3634d324d 100644 --- a/runtime/typescript/packages/core/src/core/pipeline.ts +++ b/runtime/typescript/packages/core/src/core/pipeline.ts @@ -32,6 +32,16 @@ */ import { Prompty } from "../model/agent/prompty.js"; +import { ToolCall as ModelToolCall } from "../model/conversation/tool-call.js"; +import { + ErrorChunk, + StreamChunk, + TextChunk, + ThinkingChunk, + ToolChunk, + UsageChunk, +} from "../model/events/stream-chunk.js"; +import type { InvocationUsage } from "../model/model/invocation-usage.js"; import { type ToolCall, Message, @@ -40,7 +50,7 @@ import { text, } from "./types.js"; import { getRenderer, getParser, getExecutor, getProcessor } from "./registry.js"; -import { getLastNonces, clearLastNonces } from "../renderers/common.js"; +import { prepareRenderInputs } from "../renderers/common.js"; import { traceSpan, sanitizeValue } from "../tracing/tracer.js"; import { load } from "./loader.js"; import { dispatchTool, resilientJsonParse } from "./tool-dispatch.js"; @@ -94,21 +104,11 @@ export class ExecuteError extends Error { /** Replace raw nonce strings with readable `{{thread:name}}` in trace output. */ function sanitizeNonces(value: unknown): unknown { - const nonces = getLastNonces(); - if (nonces.size === 0) return value; - - // Build nonce → display name map - const replacements = new Map(); - for (const [name, nonce] of nonces) { - replacements.set(nonce, `[thread: ${name}]`); - } - if (typeof value === "string") { - let result = value; - for (const [nonce, display] of replacements) { - result = result.replaceAll(nonce, display); - } - return result; + return value.replace( + /__PROMPTY_THREAD_[a-f0-9]{8}_(.+?)__/g, + (_nonce, name: string) => `[thread: ${name}]`, + ); } if (Array.isArray(value)) { @@ -239,15 +239,22 @@ function serializeMessages(messages: Message[]): unknown[] { export async function render( agent: Prompty, inputs: Record, +): Promise { + const [renderInputs] = prepareRenderInputs(agent, inputs); + return renderTemplate(agent, agent.instructions ?? "", renderInputs); +} + +async function renderTemplate( + agent: Prompty, + template: string, + inputs: Record, ): Promise { const formatKind = resolveFormatKind(agent); const renderer = getRenderer(formatKind); return traceSpan(renderer.constructor?.name ?? "Renderer", async (emit) => { - const template = agent.instructions ?? ""; - emit("signature", `prompty.renderers.${renderer.constructor?.name ?? "Renderer"}.render`); - emit("inputs", { data: inputs }); + emit("inputs", sanitizeNonces({ data: inputs })); const result = await renderer.render(agent, template, inputs); emit("result", sanitizeNonces(result)); return result; @@ -291,6 +298,49 @@ export async function process( return processor.process(agent, response); } +/** + * Process a raw provider stream into canonical generated stream chunks. + * + * Custom processors that only implement the legacy `process()` streaming + * surface are adapted so existing integrations remain compatible. + */ +export async function processStream( + agent: Prompty, + response: AsyncIterable, +): Promise> { + const provider = resolveProvider(agent); + const processor = getProcessor(provider); + if (processor.processStream) { + return processor.processStream(response); + } + + const legacy = await processor.process(agent, response); + return adaptLegacyStream(legacy); +} + +async function* adaptLegacyStream( + value: unknown, +): AsyncGenerator { + if (!isAsyncIterable(value)) { + if (typeof value === "string") { + yield new TextChunk({ value }); + } else if (isToolCallLike(value)) { + yield new ToolChunk({ toolCall: new ModelToolCall(value) }); + } + return; + } + + for await (const item of value) { + if (item instanceof StreamChunk) { + yield item; + } else if (typeof item === "string") { + yield new TextChunk({ value: item }); + } else if (isToolCallLike(item)) { + yield new ToolChunk({ toolCall: new ModelToolCall(item) }); + } + } +} + // --------------------------------------------------------------------------- // Composite: prepare() = render + parse + thread expansion // --------------------------------------------------------------------------- @@ -313,24 +363,18 @@ export async function prepare( const parserKind = resolveParserKind(agent); const parser = getParser(parserKind); let context: Record | undefined; + const [renderInputs, nonces] = prepareRenderInputs(agent, validatedInputs); if (isStrictMode(agent) && parser.preRender) { const [sanitized, ctx] = parser.preRender(agent.instructions ?? ""); - // Temporarily override instructions for rendering - const originalInstructions = agent.instructions; - agent.instructions = sanitized; context = ctx; - // Render - clearLastNonces(); - const rendered = await render(agent, validatedInputs); - agent.instructions = originalInstructions; + const rendered = await renderTemplate(agent, sanitized, renderInputs); // Parse const messages = await parse(agent, rendered, context); // Thread expansion - const nonces = getLastNonces(); const expanded = expandThreads(messages, nonces, validatedInputs); emit("result", serializeMessages(expanded)); @@ -338,12 +382,10 @@ export async function prepare( } // Non-strict path - clearLastNonces(); - const rendered = await render(agent, validatedInputs); + const rendered = await renderTemplate(agent, agent.instructions ?? "", renderInputs); const messages = await parse(agent, rendered, context); // Thread expansion - const nonces = getLastNonces(); const expanded = expandThreads(messages, nonces, validatedInputs); emit("result", serializeMessages(expanded)); @@ -684,7 +726,8 @@ export async function turn( emit("inputs", sanitizeValue("inputs", inputs)); const tools = options?.tools ?? {}; - const hasTools = Object.keys(tools).length > 0; + const hasTools = + Object.keys(tools).length > 0 || (agent.tools?.length ?? 0) > 0; const onEvent = options?.onEvent; emitEvent(onEvent, "turn_start", { agent: agent.name, @@ -692,6 +735,14 @@ export async function turn( maxIterations: options?.maxIterations ?? DEFAULT_MAX_ITERATIONS, }); + try { + checkCancellation(options?.signal); + } catch (err) { + emitEvent(onEvent, "cancelled", {}); + emitFailedTurnEnd(onEvent, err, 0); + throw err; + } + if (!hasTools) { // Simple mode: prepare → [extensions] → executor → [output guard] → process let messages: Message[]; @@ -752,14 +803,23 @@ export async function turn( }); let response: unknown; try { - response = await executor.execute(agent, messages); + response = await invokeWithRetry( + executor, + agent, + messages, + options?.maxLlmRetries ?? DEFAULT_MAX_LLM_RETRIES, + onEvent, + options?.signal, + ); } catch (err) { emitFailedTurnEnd(onEvent, err, 0); throw err; } - emitEvent(onEvent, "llm_complete", {}); - if (options?.raw) { + if (isAsyncIterable(response)) { + return finalizeSimpleTurnStream(response, messages, onEvent, true); + } + emitEvent(onEvent, "llm_complete", {}); emit("result", response); emitEvent(onEvent, "turn_end", { iterations: 0, status: "success", response }); return response; @@ -772,6 +832,11 @@ export async function turn( throw err; } + if (isAsyncIterable(processed)) { + return finalizeSimpleTurnStream(processed, messages, onEvent, false); + } + emitEvent(onEvent, "llm_complete", {}); + // §13.4 — Output guardrail on final response if (options?.guardrails) { const contentStr = typeof processed === "string" ? processed : JSON.stringify(processed); @@ -885,8 +950,6 @@ export async function turn( emitFailedTurnEnd(onEvent, err, iteration); throw err; } - emitEvent(onEvent, "llm_complete", { iteration }); - // Streaming: consume the stream, extract tool calls from buffered chunks if (isAsyncIterable(response)) { let streamResult: Awaited>; @@ -896,7 +959,11 @@ export async function turn( emitFailedTurnEnd(onEvent, err, iteration, response); throw err; } - const { toolCalls, content } = streamResult; + const { toolCalls, content, usage } = streamResult; + emitEvent(onEvent, "llm_complete", { + iteration, + ...(usage ? { usage: usage.save() } : {}), + }); // §13.4 — Output guardrail if (guardrails && content) { @@ -948,6 +1015,8 @@ export async function turn( continue; } + emitEvent(onEvent, "llm_complete", { iteration }); + // Non-streaming: check raw response for tool calls if (!hasToolCalls(response)) { let finalResult: unknown; @@ -1089,27 +1158,64 @@ async function consumeStream( agent: Prompty, response: unknown, onEvent?: EventCallback, -): Promise<{ toolCalls: ToolCall[]; content: string }> { - const processed = await process(agent, response); +): Promise<{ toolCalls: ToolCall[]; content: string; usage?: InvocationUsage }> { + if (!isAsyncIterable(response)) { + throw new TypeError("Expected an async iterable provider response"); + } + const processed = await processStream(agent, response); const toolCalls: ToolCall[] = []; const textParts: string[] = []; + let usage: InvocationUsage | undefined; + + for await (const item of processed) { + if (item instanceof TextChunk) { + textParts.push(item.value); + emitEvent(onEvent, "token", { token: item.value }); + } else if (item instanceof ThinkingChunk) { + emitEvent(onEvent, "thinking", { thinking: item.value }); + } else if (item instanceof ToolChunk) { + toolCalls.push(item.toolCall); + } else if (item instanceof UsageChunk) { + usage = item.usage; + } else if (item instanceof ErrorChunk) { + throw new Error(item.message); + } + } + + return { toolCalls, content: textParts.join(""), usage }; +} - if (isAsyncIterable(processed)) { - for await (const item of processed) { - if (isToolCallLike(item)) { - toolCalls.push(item); - } else if (typeof item === "string") { - textParts.push(item); +/** + * Preserve the legacy public stream shape while completing turn events only + * after successful stream exhaustion. + */ +async function* finalizeSimpleTurnStream( + stream: AsyncIterable, + messages: Message[], + onEvent: EventCallback | undefined, + raw: boolean, +): AsyncGenerator { + const collected: unknown[] = []; + try { + for await (const item of stream) { + collected.push(item); + if (!raw && typeof item === "string") { emitEvent(onEvent, "token", { token: item }); } + yield item; } - } else if (typeof processed === "string") { - textParts.push(processed); - emitEvent(onEvent, "token", { token: processed }); + } catch (err) { + emitFailedTurnEnd(onEvent, err, 0, collected); + throw err; } - return { toolCalls, content: textParts.join("") }; + emitEvent(onEvent, "llm_complete", {}); + const response = raw ? collected : collected.filter((item) => typeof item === "string").join(""); + if (!raw) { + emitEvent(onEvent, "done", { response, messages }); + } + emitEvent(onEvent, "turn_end", { iterations: 0, status: "success", response }); } diff --git a/runtime/typescript/packages/core/src/core/turn-engine-cancellation.ts b/runtime/typescript/packages/core/src/core/turn-engine-cancellation.ts new file mode 100644 index 000000000..9c7804c1e --- /dev/null +++ b/runtime/typescript/packages/core/src/core/turn-engine-cancellation.ts @@ -0,0 +1,66 @@ +/** + * Cooperative cancellation for canonical turn-engine effect boundaries. + */ + +/** Raised by ports that abort cooperatively while waiting for cancellation. */ +export class TurnCancellationError extends Error { + constructor(message = "Turn execution was cancelled") { + super(message); + this.name = "TurnCancellationError"; + } +} + +/** + * A runtime-native cancellation token that can bridge an AbortSignal. + * + * Cancellation is sticky. Callers can synchronously inspect the token or await + * cancellation without polling. + */ +export class TurnCancellationToken { + readonly #listeners = new Set<() => void>(); + #cancelled = false; + + constructor(signal?: AbortSignal) { + if (signal) { + if (signal.aborted) { + this.#cancelled = true; + } else { + signal.addEventListener("abort", () => this.cancel(), { once: true }); + } + } + } + + static fromAbortSignal(signal: AbortSignal): TurnCancellationToken { + return new TurnCancellationToken(signal); + } + + get isCancellationRequested(): boolean { + return this.#cancelled; + } + + cancel(): void { + if (this.#cancelled) { + return; + } + this.#cancelled = true; + for (const listener of this.#listeners) { + listener(); + } + this.#listeners.clear(); + } + + throwIfCancellationRequested(): void { + if (this.#cancelled) { + throw new TurnCancellationError(); + } + } + + waitForCancellation(): Promise { + if (this.#cancelled) { + return Promise.resolve(); + } + return new Promise((resolve) => { + this.#listeners.add(resolve); + }); + } +} diff --git a/runtime/typescript/packages/core/src/core/turn-engine-context.ts b/runtime/typescript/packages/core/src/core/turn-engine-context.ts new file mode 100644 index 000000000..bbcc160df --- /dev/null +++ b/runtime/typescript/packages/core/src/core/turn-engine-context.ts @@ -0,0 +1,265 @@ +/** + * Ordered context assembly for immutable model-invocation snapshots. + */ + +import { ContextCandidate } from "../model/pipeline/context-candidate.js"; +import { ContextRequest } from "../model/pipeline/context-request.js"; +import { InvocationContextDecision } from "../model/pipeline/invocation-context-decision.js"; +import { ModelInvocationContextSnapshot } from "../model/pipeline/model-invocation-context-snapshot.js"; +import { TurnCancellationToken } from "./turn-engine-cancellation.js"; + +export class TurnContextError extends Error { + constructor(message: string, options?: ErrorOptions) { + super(message, options); + this.name = "TurnContextError"; + } +} + +export interface ContextSource { + readonly name: string; + load( + request: ContextRequest, + cancellation: TurnCancellationToken, + ): Promise; +} + +export interface ContextTransform { + readonly name: string; + apply( + request: ContextRequest, + candidates: readonly ContextCandidate[], + cancellation: TurnCancellationToken, + ): Promise; +} + +export interface ContextPackingStrategy { + readonly name: string; + pack( + request: ContextRequest, + candidates: readonly ContextCandidate[], + cancellation: TurnCancellationToken, + ): Promise; +} + +/** Baseline packer that appends every candidate in deterministic source order. */ +export class AppendContextPackingStrategy implements ContextPackingStrategy { + readonly name = "append"; + + async pack( + request: ContextRequest, + candidates: readonly ContextCandidate[], + ): Promise { + const messages = [...request.messages]; + const decisions: InvocationContextDecision[] = []; + for (const [rank, candidate] of candidates.entries()) { + messages.push(...candidate.messages); + decisions.push( + new InvocationContextDecision({ + candidateId: candidate.id, + disposition: "included", + reason: "included by append strategy", + rank, + metadata: candidate.metadata, + }), + ); + } + + return new ModelInvocationContextSnapshot({ + id: `context:${request.invocationId}`, + sessionId: request.sessionId, + turnId: request.turnId, + invocationId: request.invocationId, + iteration: request.iteration, + messages, + decisions, + stablePrefixMessages: request.stablePrefixMessages, + contextState: request.contextState, + }); + } +} + +/** + * Composes sources, transforms, and packing in registration order. + * + * Every removed candidate is retained as an excluded decision and the returned + * snapshot is deeply frozen to preserve retry identity and immutability. + */ +export class ContextPipeline { + readonly #sources: ContextSource[]; + readonly #transforms: ContextTransform[]; + readonly #packing: ContextPackingStrategy; + + constructor(options: { + sources?: readonly ContextSource[]; + transforms?: readonly ContextTransform[]; + packing?: ContextPackingStrategy; + } = {}) { + this.#sources = [...(options.sources ?? [])]; + this.#transforms = [...(options.transforms ?? [])]; + this.#packing = options.packing ?? new AppendContextPackingStrategy(); + } + + withSource(source: ContextSource): ContextPipeline { + return new ContextPipeline({ + sources: [...this.#sources, source], + transforms: this.#transforms, + packing: this.#packing, + }); + } + + withTransform(transform: ContextTransform): ContextPipeline { + return new ContextPipeline({ + sources: this.#sources, + transforms: [...this.#transforms, transform], + packing: this.#packing, + }); + } + + async prepare( + request: ContextRequest, + cancellation: TurnCancellationToken, + ): Promise { + cancellation.throwIfCancellationRequested(); + let candidates: ContextCandidate[] = []; + for (const source of this.#sources) { + cancellation.throwIfCancellationRequested(); + try { + candidates.push(...(await source.load(request, cancellation))); + } catch (error) { + throw new TurnContextError(`Context source '${source.name}' failed`, { + cause: error, + }); + } + } + assertUniqueCandidates(candidates); + + const excluded: InvocationContextDecision[] = []; + for (const transform of this.#transforms) { + cancellation.throwIfCancellationRequested(); + const before = candidates; + try { + candidates = await transform.apply(request, before, cancellation); + } catch (error) { + throw new TurnContextError( + `Context transform '${transform.name}' failed`, + { cause: error }, + ); + } + assertUniqueCandidates(candidates); + const retained = new Set(candidates.map((candidate) => candidate.id)); + for (const candidate of before) { + if (!retained.has(candidate.id)) { + excluded.push( + new InvocationContextDecision({ + candidateId: candidate.id, + disposition: "excluded", + reason: `excluded by context transform '${transform.name}'`, + metadata: candidate.metadata, + }), + ); + } + } + } + + cancellation.throwIfCancellationRequested(); + let snapshot: ModelInvocationContextSnapshot; + try { + snapshot = await this.#packing.pack(request, candidates, cancellation); + } catch (error) { + throw new TurnContextError( + `Context packing strategy '${this.#packing.name}' failed`, + { cause: error }, + ); + } + + const decided = new Set( + (snapshot.decisions ?? []).map((decision) => decision.candidateId), + ); + for (const candidate of candidates) { + if (!decided.has(candidate.id)) { + excluded.push( + new InvocationContextDecision({ + candidateId: candidate.id, + disposition: "excluded", + reason: `excluded without an explicit decision by packing strategy '${this.#packing.name}'`, + metadata: candidate.metadata, + }), + ); + } + } + snapshot.decisions = [...(snapshot.decisions ?? []), ...excluded]; + validateSnapshot(snapshot, request); + return deepFreeze(snapshot); + } +} + +export function validateSnapshot( + snapshot: ModelInvocationContextSnapshot, + request?: ContextRequest, +): void { + if ( + snapshot.stablePrefixMessages < 0 || + snapshot.stablePrefixMessages > snapshot.messages.length + ) { + throw new TurnContextError( + `Stable prefix contains ${snapshot.stablePrefixMessages} messages but snapshot contains ${snapshot.messages.length}`, + ); + } + + const delegated = snapshot.contextState?.delegatedState ?? []; + if ( + snapshot.contextState?.portability === "portable" && + delegated.length > 0 + ) { + throw new TurnContextError( + "Portable snapshots cannot contain delegated provider state", + ); + } + if ( + snapshot.contextState?.portability === "delegated" && + delegated.length === 0 + ) { + throw new TurnContextError( + "Delegated snapshots must identify provider-held state", + ); + } + + if ( + request && + (snapshot.sessionId !== request.sessionId || + snapshot.turnId !== request.turnId || + snapshot.invocationId !== request.invocationId || + snapshot.iteration !== request.iteration) + ) { + throw new TurnContextError( + `Snapshot identity does not match invocation '${request.invocationId}'`, + ); + } +} + +function assertUniqueCandidates(candidates: readonly ContextCandidate[]): void { + const ids = new Set(); + for (const candidate of candidates) { + if (ids.has(candidate.id)) { + throw new TurnContextError( + `Duplicate context candidate id '${candidate.id}'`, + ); + } + ids.add(candidate.id); + } +} + +function deepFreeze(value: T): T { + if ( + value === null || + (typeof value !== "object" && typeof value !== "function") || + Object.isFrozen(value) + ) { + return value; + } + Object.freeze(value); + for (const child of Object.values(value as Record)) { + deepFreeze(child); + } + return value; +} diff --git a/runtime/typescript/packages/core/src/core/turn-engine-ports.ts b/runtime/typescript/packages/core/src/core/turn-engine-ports.ts new file mode 100644 index 000000000..d69eb9a4c --- /dev/null +++ b/runtime/typescript/packages/core/src/core/turn-engine-ports.ts @@ -0,0 +1,280 @@ +/** + * Runtime-local effect ports for the canonical TypeScript turn engine. + */ + +import { Message } from "../model/conversation/message.js"; +import { TextPart } from "../model/conversation/content-part.js"; +import { EngineCheckpoint } from "../model/pipeline/engine-checkpoint.js"; +import { EngineEvent } from "../model/pipeline/engine-event.js"; +import { EnginePermissionDecision } from "../model/pipeline/engine-permission-decision.js"; +import { FinalOutputPolicyRequest } from "../model/pipeline/final-output-policy-request.js"; +import { FinalOutputPolicyResult } from "../model/pipeline/final-output-policy-result.js"; +import { HostPolicyRequest } from "../model/pipeline/host-policy-request.js"; +import { HostPolicyResult } from "../model/pipeline/host-policy-result.js"; +import { ModelInvocationRequest } from "../model/pipeline/model-invocation-request.js"; +import { ModelInvocationResponse } from "../model/pipeline/model-invocation-response.js"; +import { ModelToolRequest } from "../model/pipeline/model-tool-request.js"; +import { ModelToolResult } from "../model/pipeline/model-tool-result.js"; +import { RetryPolicyRequest } from "../model/pipeline/retry-policy-request.js"; +import { TurnCommit } from "../model/pipeline/turn-commit.js"; +import { + TurnCancellationError, + TurnCancellationToken, +} from "./turn-engine-cancellation.js"; + +/** Runtime port failure classification used by retry and reconciliation logic. */ +export class TurnPortError extends Error { + readonly outcomeUnknown: boolean; + readonly configurationError: boolean; + readonly metadata: Record; + + constructor( + message: string, + options: { + outcomeUnknown?: boolean; + configurationError?: boolean; + metadata?: Record; + cause?: unknown; + } = {}, + ) { + super(message, { cause: options.cause }); + this.name = "TurnPortError"; + this.outcomeUnknown = options.outcomeUnknown ?? false; + this.configurationError = options.configurationError ?? false; + this.metadata = options.metadata ?? {}; + } + + static indeterminate( + message: string, + metadata: Record = {}, + ): TurnPortError { + return new TurnPortError(message, { + outcomeUnknown: true, + metadata, + }); + } + + static configuration(message: string): TurnPortError { + return new TurnPortError(message, { configurationError: true }); + } +} + +/** Typed host-policy rejection committed as a failed turn. */ +export class TurnHostPolicyError extends Error { + readonly errorKind: string; + + constructor(errorKind: string, message: string, options?: ErrorOptions) { + super(message, options); + this.name = "TurnHostPolicyError"; + this.errorKind = errorKind; + } +} + +/** Ephemeral provider output that does not affect semantic event ordering. */ +export type ModelStreamChunk = + | { kind: "text"; value: string } + | { kind: "thinking"; value: string } + | { kind: "provider"; value: unknown }; + +export interface ModelStreamPort { + emit(chunk: ModelStreamChunk): Promise | void; +} + +export interface ModelPort { + invoke( + request: ModelInvocationRequest, + cancellation: TurnCancellationToken, + stream: ModelStreamPort, + ): Promise; +} + +export interface HostPolicyPort { + beforeModel( + request: HostPolicyRequest, + cancellation: TurnCancellationToken, + ): Promise; + + beforeCommit( + request: FinalOutputPolicyRequest, + cancellation: TurnCancellationToken, + ): Promise; +} + +export interface RetryPolicyPort { + backoff( + request: RetryPolicyRequest, + cancellation: TurnCancellationToken, + ): Promise; +} + +/** Converts a completed model/tool batch into provider-valid messages. */ +export interface ConversationPort { + formatToolExchange( + response: ModelInvocationResponse, + results: readonly ModelToolResult[], + ): Message[]; +} + +export interface PermissionPort { + authorize( + request: ModelToolRequest, + cancellation: TurnCancellationToken, + ): Promise; +} + +export interface ToolPort { + execute( + request: ModelToolRequest, + cancellation: TurnCancellationToken, + ): Promise; +} + +/** Atomically persists semantic events and the checkpoint containing them. */ +export interface DurabilityPort { + append(event: EngineEvent): Promise; + + appendWithCheckpoint( + events: readonly EngineEvent[], + checkpoint: EngineCheckpoint, + ): Promise; +} + +export interface PostCommitPort { + afterCommit( + effectId: string, + commit: TurnCommit, + cancellation: TurnCancellationToken, + ): Promise; +} + +export interface Clock { + now(): string; +} + +export interface IdGenerator { + nextId(kind: string): string; +} + +export class AllowAllPermissionPort implements PermissionPort { + async authorize(): Promise { + return new EnginePermissionDecision({ + approved: true, + reason: "allow_all", + }); + } +} + +export class NoopDurabilityPort implements DurabilityPort { + async append(): Promise {} + + async appendWithCheckpoint(): Promise {} +} + +export class NoopPostCommitPort implements PostCommitPort { + async afterCommit(): Promise {} +} + +export class NoopModelStreamPort implements ModelStreamPort { + emit(): void {} +} + +export class NoopHostPolicyPort implements HostPolicyPort { + async beforeModel(request: HostPolicyRequest): Promise { + return new HostPolicyResult({ + messages: request.messages, + stablePrefixMessages: request.stablePrefixMessages, + }); + } + + async beforeCommit( + request: FinalOutputPolicyRequest, + ): Promise { + return new FinalOutputPolicyResult({ output: request.output }); + } +} + +export class NoopRetryPolicyPort implements RetryPolicyPort { + async backoff( + _request: RetryPolicyRequest, + cancellation: TurnCancellationToken, + ): Promise { + if (cancellation.isCancellationRequested) { + throw new TurnCancellationError("Retry backoff was cancelled"); + } + } +} + +export class UnavailableToolPort implements ToolPort { + async execute(request: ModelToolRequest): Promise { + throw TurnPortError.configuration( + `No tool binding is registered for '${request.name}'`, + ); + } +} + +/** + * Provider-neutral conversation formatter preserving assistant content and + * original model-request order. + */ +export class DefaultConversationPort implements ConversationPort { + formatToolExchange( + response: ModelInvocationResponse, + results: readonly ModelToolResult[], + ): Message[] { + const messages = [...(response.assistantMessages ?? [])]; + for (const request of response.toolRequests ?? []) { + const result = results.find((candidate) => candidate.requestId === request.id); + if (!result) { + throw TurnPortError.configuration( + `Tool exchange is missing result '${request.id}'`, + ); + } + messages.push(toolResultMessage(request.id, modelVisibleToolOutput(result))); + } + return messages; + } +} + +export class SystemClock implements Clock { + now(): string { + return new Date().toISOString(); + } +} + +/** Process-local unique identifiers suitable for default non-deterministic runs. */ +export class DefaultIdGenerator implements IdGenerator { + #counter = 0; + + nextId(kind: string): string { + this.#counter += 1; + return `${kind}-${Date.now().toString(36)}-${this.#counter.toString(36)}`; + } +} + +export function modelVisibleToolOutput(result: ModelToolResult): string { + if (typeof result.output === "string") { + return result.output; + } + if (result.output === undefined) { + return ""; + } + return JSON.stringify(result.output); +} + +export function toolResultMessage(requestId: string, value: string): Message { + return new Message({ + role: "tool", + parts: [new TextPart({ value })], + metadata: { tool_call_id: requestId }, + }); +} + +export function normalizePortError(error: unknown): TurnPortError { + if (error instanceof TurnPortError) { + return error; + } + if (error instanceof Error) { + return new TurnPortError(error.message, { cause: error }); + } + return new TurnPortError(String(error)); +} diff --git a/runtime/typescript/packages/core/src/core/turn-engine.ts b/runtime/typescript/packages/core/src/core/turn-engine.ts new file mode 100644 index 000000000..f2d8da7c4 --- /dev/null +++ b/runtime/typescript/packages/core/src/core/turn-engine.ts @@ -0,0 +1,1698 @@ +/** + * Canonical, provider-neutral TypeScript turn state machine. + * + * Generated Typra models are the durable contract. This module owns only + * runtime orchestration state, native effect ports, and recovery errors. + */ + +import { Message } from "../model/conversation/message.js"; +import { ContextRequest } from "../model/pipeline/context-request.js"; +import { DelegatedStateReference } from "../model/pipeline/delegated-state-reference.js"; +import { EngineCheckpoint } from "../model/pipeline/engine-checkpoint.js"; +import { + EngineEvent, + type EngineEventKind, +} from "../model/pipeline/engine-event.js"; +import { EnginePermissionDecision } from "../model/pipeline/engine-permission-decision.js"; +import { FinalOutputPolicyRequest } from "../model/pipeline/final-output-policy-request.js"; +import { HostPolicyRequest } from "../model/pipeline/host-policy-request.js"; +import { InvocationContextState } from "../model/pipeline/invocation-context-state.js"; +import { ModelInvocationContextSnapshot } from "../model/pipeline/model-invocation-context-snapshot.js"; +import { ModelInvocationRequest } from "../model/pipeline/model-invocation-request.js"; +import { ModelInvocationResponse } from "../model/pipeline/model-invocation-response.js"; +import { ModelReconciliationState } from "../model/pipeline/model-reconciliation-state.js"; +import { ModelToolRequest } from "../model/pipeline/model-tool-request.js"; +import { ModelToolResult } from "../model/pipeline/model-tool-result.js"; +import { ResumeContext } from "../model/pipeline/resume-context.js"; +import { RetryPolicyRequest } from "../model/pipeline/retry-policy-request.js"; +import { + TurnCommit, + type EngineTurnStatus, +} from "../model/pipeline/turn-commit.js"; +import { TurnEngineResult } from "../model/pipeline/turn-engine-result.js"; +import { + TurnCancellationError, + TurnCancellationToken, +} from "./turn-engine-cancellation.js"; +import { ContextPipeline, TurnContextError } from "./turn-engine-context.js"; +import { + AllowAllPermissionPort, + type Clock, + type ConversationPort, + DefaultConversationPort, + DefaultIdGenerator, + type DurabilityPort, + type HostPolicyPort, + type IdGenerator, + type ModelPort, + type ModelStreamPort, + NoopDurabilityPort, + NoopHostPolicyPort, + NoopModelStreamPort, + NoopPostCommitPort, + NoopRetryPolicyPort, + normalizePortError, + type PermissionPort, + type PostCommitPort, + type RetryPolicyPort, + SystemClock, + type ToolPort, + TurnHostPolicyError, + TurnPortError, + UnavailableToolPort, + modelVisibleToolOutput, + toolResultMessage, +} from "./turn-engine-ports.js"; + +export class TurnEngineError extends Error { + constructor(message: string, options?: ErrorOptions) { + super(message, options); + this.name = "TurnEngineError"; + } +} + +export class InvalidTurnRequestError extends TurnEngineError { + constructor(message: string) { + super(`Invalid turn request: ${message}`); + this.name = "InvalidTurnRequestError"; + } +} + +export class TurnEnginePortError extends TurnEngineError { + readonly stage: string; + + constructor(stage: string, cause: unknown) { + const source = normalizePortError(cause); + super(`${stage} failed: ${source.message}`, { cause: source }); + this.name = "TurnEnginePortError"; + this.stage = stage; + } +} + +/** + * Atomic persistence failed after an external effect completed. + * + * The checkpoint and completed results are explicit recovery state and must be + * persisted/reconciled rather than treating the operation as retryable. + */ +export class TurnEngineRecoveryRequiredError extends TurnEngineError { + readonly stage: string; + readonly requestId: string; + readonly checkpoint: EngineCheckpoint; + readonly toolResults: readonly ModelToolResult[]; + + constructor(options: { + stage: string; + requestId: string; + checkpoint: EngineCheckpoint; + toolResults: readonly ModelToolResult[]; + cause: unknown; + }) { + const source = normalizePortError(options.cause); + super( + `${options.stage} durability failed after effect '${options.requestId}': ${source.message}`, + { cause: source }, + ); + this.name = "TurnEngineRecoveryRequiredError"; + this.stage = options.stage; + this.requestId = options.requestId; + this.checkpoint = options.checkpoint; + this.toolResults = options.toolResults; + } +} + +export interface TurnEngineRequestInit { + sessionId: string; + turnId: string; + messages: readonly Message[]; + runId?: string; + parentRunId?: string; + delegationDepth?: number; + inputs?: unknown; + maxIterations?: number; + maxModelAttempts?: number; + startIteration?: number; + initialSequence?: number; + stablePrefixMessages?: number; + contextState?: InvocationContextState; + activeInvocationId?: string; + pendingToolRequests?: readonly ModelToolRequest[]; + completedToolResults?: readonly ModelToolResult[]; + completedModelIterations?: number; + reconciliationRequired?: boolean; + modelReconciliation?: ModelReconciliationState; + pendingOutput?: unknown; + finalOutputReady?: boolean; + pendingModelResponse?: ModelInvocationResponse; + policyAppliedForIteration?: boolean; + committedToolResultIds?: readonly string[]; +} + +/** Runtime request state. Durable nested values remain generated model types. */ +export class TurnEngineRequest { + readonly sessionId: string; + readonly turnId: string; + readonly messages: Message[]; + runId: string; + readonly parentRunId?: string; + readonly delegationDepth: number; + readonly inputs?: unknown; + readonly maxIterations: number; + readonly maxModelAttempts: number; + startIteration: number; + readonly initialSequence: number; + readonly stablePrefixMessages: number; + readonly contextState: InvocationContextState; + readonly activeInvocationId?: string; + readonly pendingToolRequests: ModelToolRequest[]; + readonly completedToolResults: ModelToolResult[]; + readonly completedModelIterations: number; + reconciliationRequired: boolean; + readonly modelReconciliation?: ModelReconciliationState; + readonly pendingOutput?: unknown; + readonly finalOutputReady: boolean; + readonly pendingModelResponse?: ModelInvocationResponse; + readonly policyAppliedForIteration: boolean; + readonly committedToolResultIds: string[]; + reconciliationResolution?: ModelToolResult; + modelReconciliationResolution?: ModelInvocationResponse; + + constructor(init: TurnEngineRequestInit) { + this.sessionId = init.sessionId; + this.turnId = init.turnId; + this.messages = [...init.messages]; + this.runId = init.runId ?? ""; + this.parentRunId = init.parentRunId; + this.delegationDepth = init.delegationDepth ?? 0; + this.inputs = init.inputs; + this.maxIterations = init.maxIterations ?? 10; + this.maxModelAttempts = init.maxModelAttempts ?? 3; + this.startIteration = init.startIteration ?? 0; + this.initialSequence = init.initialSequence ?? 0; + this.stablePrefixMessages = + init.stablePrefixMessages ?? init.messages.length; + this.contextState = + init.contextState ?? new InvocationContextState({ portability: "portable" }); + this.activeInvocationId = init.activeInvocationId; + this.pendingToolRequests = [...(init.pendingToolRequests ?? [])]; + this.completedToolResults = [...(init.completedToolResults ?? [])]; + this.completedModelIterations = init.completedModelIterations ?? 0; + this.reconciliationRequired = init.reconciliationRequired ?? false; + this.modelReconciliation = init.modelReconciliation; + this.pendingOutput = init.pendingOutput; + this.finalOutputReady = init.finalOutputReady ?? false; + this.pendingModelResponse = init.pendingModelResponse; + this.policyAppliedForIteration = + init.policyAppliedForIteration ?? false; + this.committedToolResultIds = [...(init.committedToolResultIds ?? [])]; + } + + static fromResume(resume: ResumeContext): TurnEngineRequest { + const checkpoint = resume.checkpoint; + const hasFinishedIteration = + (checkpoint.pendingToolRequests ?? []).length === 0 && + checkpoint.pendingModelResponse === undefined && + !checkpoint.finalOutputReady && + !checkpoint.reconciliationRequired; + const startIteration = checkpoint.resumeSameIteration + ? checkpoint.iteration + : hasFinishedIteration + ? checkpoint.iteration + 1 + : checkpoint.iteration; + + return new TurnEngineRequest({ + sessionId: checkpoint.sessionId, + turnId: checkpoint.turnId, + messages: checkpoint.messages, + runId: checkpoint.runId, + parentRunId: checkpoint.parentRunId, + delegationDepth: checkpoint.delegationDepth, + inputs: checkpoint.inputs, + maxIterations: resume.maxIterations, + maxModelAttempts: + resume.maxModelAttempts > 0 ? resume.maxModelAttempts : 3, + startIteration, + initialSequence: Math.max( + checkpoint.lastSequence, + resume.lastJournalSequence, + ), + stablePrefixMessages: checkpoint.stablePrefixMessages, + contextState: checkpoint.contextState, + activeInvocationId: checkpoint.activeInvocationId, + pendingToolRequests: checkpoint.pendingToolRequests ?? [], + completedToolResults: checkpoint.completedToolResults ?? [], + completedModelIterations: checkpoint.completedModelIterations, + reconciliationRequired: checkpoint.reconciliationRequired, + modelReconciliation: checkpoint.modelReconciliation, + pendingOutput: checkpoint.pendingOutput, + finalOutputReady: checkpoint.finalOutputReady, + pendingModelResponse: checkpoint.pendingModelResponse, + policyAppliedForIteration: checkpoint.policyAppliedForIteration, + committedToolResultIds: checkpointCommittedToolResultIds(checkpoint), + }); + } + + static afterToolReconciliation( + resume: ResumeContext, + resolvedResult: ModelToolResult, + ): TurnEngineRequest { + const checkpoint = cloneCheckpoint(resume.checkpoint); + if (!checkpoint.reconciliationRequired) { + throw new InvalidTurnRequestError( + "checkpoint does not require reconciliation", + ); + } + if (checkpoint.modelReconciliation) { + throw new InvalidTurnRequestError( + "checkpoint requires model reconciliation, not tool reconciliation", + ); + } + if (resolvedResult.outcome === "indeterminate") { + throw new InvalidTurnRequestError( + "resolved tool result must have a determinate outcome", + ); + } + const results = checkpoint.completedToolResults ?? []; + const index = results.findIndex( + (result) => result.requestId === resolvedResult.requestId, + ); + if (index < 0) { + throw new InvalidTurnRequestError( + `checkpoint does not contain indeterminate tool request '${resolvedResult.requestId}'`, + ); + } + if (results[index].outcome !== "indeterminate") { + throw new InvalidTurnRequestError( + `tool request '${resolvedResult.requestId}' is already determinate`, + ); + } + results[index] = resolvedResult; + if (!checkpoint.pendingModelResponse) { + const messageIndex = checkpoint.messages.findIndex( + (message) => + message.metadata["tool_call_id"] === resolvedResult.requestId, + ); + if (messageIndex < 0) { + throw new InvalidTurnRequestError( + `checkpoint is missing the tool result message for '${resolvedResult.requestId}'`, + ); + } + checkpoint.messages[messageIndex] = toolResultMessage( + resolvedResult.requestId, + modelVisibleToolOutput(resolvedResult), + ); + } + checkpoint.reconciliationRequired = false; + const request = TurnEngineRequest.fromResume( + new ResumeContext({ + checkpoint, + maxIterations: resume.maxIterations, + maxModelAttempts: resume.maxModelAttempts, + lastJournalSequence: resume.lastJournalSequence, + metadata: resume.metadata, + }), + ); + request.reconciliationResolution = resolvedResult; + return request; + } + + static afterModelReconciliation( + resume: ResumeContext, + resolvedResponse: ModelInvocationResponse, + ): TurnEngineRequest { + const checkpoint = resume.checkpoint; + if (!checkpoint.reconciliationRequired) { + throw new InvalidTurnRequestError( + "checkpoint does not require reconciliation", + ); + } + const reconciliation = checkpoint.modelReconciliation; + if (!reconciliation) { + throw new InvalidTurnRequestError( + "checkpoint requires tool reconciliation, not model reconciliation", + ); + } + if (checkpoint.activeInvocationId !== reconciliation.invocationId) { + throw new InvalidTurnRequestError( + "model reconciliation identity does not match the active invocation", + ); + } + + const request = TurnEngineRequest.fromResume(resume); + request.startIteration = checkpoint.iteration; + request.reconciliationRequired = false; + request.modelReconciliationResolution = resolvedResponse; + return request; + } +} + +export interface TurnEngineEffects { + model: ModelPort; + stream?: ModelStreamPort; + policy?: HostPolicyPort; + retry?: RetryPolicyPort; + conversation?: ConversationPort; + permission?: PermissionPort; + tools?: ToolPort; + durability?: DurabilityPort; + postCommit?: PostCommitPort; + clock?: Clock; + ids?: IdGenerator; +} + +interface ResolvedTurnEngineEffects { + model: ModelPort; + stream: ModelStreamPort; + policy: HostPolicyPort; + retry: RetryPolicyPort; + conversation: ConversationPort; + permission: PermissionPort; + tools: ToolPort; + durability: DurabilityPort; + postCommit: PostCommitPort; + clock: Clock; + ids: IdGenerator; +} + +/** One canonical orchestration loop for live and deterministic execution. */ +export class TurnEngine { + readonly #context: ContextPipeline; + readonly #effects: ResolvedTurnEngineEffects; + + constructor(context: ContextPipeline, effects: TurnEngineEffects) { + this.#context = context; + this.#effects = { + model: effects.model, + stream: effects.stream ?? new NoopModelStreamPort(), + policy: effects.policy ?? new NoopHostPolicyPort(), + retry: effects.retry ?? new NoopRetryPolicyPort(), + conversation: effects.conversation ?? new DefaultConversationPort(), + permission: effects.permission ?? new AllowAllPermissionPort(), + tools: effects.tools ?? new UnavailableToolPort(), + durability: effects.durability ?? new NoopDurabilityPort(), + postCommit: effects.postCommit ?? new NoopPostCommitPort(), + clock: effects.clock ?? new SystemClock(), + ids: effects.ids ?? new DefaultIdGenerator(), + }; + } + + async resume( + resume: ResumeContext, + cancellation = new TurnCancellationToken(), + ): Promise { + return this.run(TurnEngineRequest.fromResume(resume), cancellation); + } + + async resumeAfterToolReconciliation( + resume: ResumeContext, + resolvedResult: ModelToolResult, + cancellation = new TurnCancellationToken(), + ): Promise { + return this.run( + TurnEngineRequest.afterToolReconciliation(resume, resolvedResult), + cancellation, + ); + } + + async resumeAfterModelReconciliation( + resume: ResumeContext, + resolvedResponse: ModelInvocationResponse, + cancellation = new TurnCancellationToken(), + ): Promise { + return this.run( + TurnEngineRequest.afterModelReconciliation(resume, resolvedResponse), + cancellation, + ); + } + + async run( + request: TurnEngineRequest, + cancellation = new TurnCancellationToken(), + ): Promise { + this.#validateRequest(request); + if (!request.runId) { + request.runId = this.#effects.ids.nextId("run"); + } + const state = new TurnState(request); + await this.#emit(state, "turn_started", { + payload: { + maxIterations: state.maxIterations, + startIteration: state.iteration, + inputs: state.inputs, + }, + }); + + if (state.modelReconciliationResolution) { + const response = state.modelReconciliationResolution; + state.modelReconciliationResolution = undefined; + const reconciliation = state.modelReconciliation; + if (!reconciliation) { + throw new InvalidTurnRequestError( + "model reconciliation response is missing durable reconciliation state", + ); + } + state.reconciliationRequired = false; + state.modelReconciliation = undefined; + try { + state.applyModelResponse(reconciliation.invocationId, response); + } catch (error) { + return this.#commitFailure( + state, + "provider_state_error", + errorMessage(error), + cancellation, + ); + } + await this.#persistModelReconciliation( + state, + reconciliation.invocationId, + reconciliation, + response, + ); + } + + if (state.reconciliationResolution) { + const resolution = state.reconciliationResolution; + state.reconciliationResolution = undefined; + await this.#persistToolReconciliation(state, resolution); + } + + if (state.reconciliationRequired) { + return this.#commitReconciliation( + state, + "effect_outcome_unknown", + "Checkpoint requires explicit effect reconciliation", + cancellation, + ); + } + + if (state.finalOutputReady) { + if (cancellation.isCancellationRequested) { + return this.#commitCancellation(state, cancellation); + } + state.output = state.pendingOutput; + return this.#applyFinalPolicy(state, cancellation); + } + + while (state.iteration < state.maxIterations) { + if (cancellation.isCancellationRequested) { + return this.#commitCancellation(state, cancellation); + } + + if ( + state.pendingToolRequests.length === 0 && + state.pendingModelResponse + ) { + const invocationId = + state.activeInvocationId ?? this.#effects.ids.nextId("invocation"); + let results: ModelToolResult[]; + try { + results = this.#finalizeToolExchange(state); + } catch (error) { + return this.#commitFailure( + state, + "conversation_format_error", + errorMessage(error), + cancellation, + ); + } + await this.#persistToolExchange(state, invocationId, results); + state.activeInvocationId = undefined; + state.iteration += 1; + continue; + } + + if (state.pendingToolRequests.length > 0) { + if (cancellation.isCancellationRequested) { + return this.#commitCancellation(state, cancellation); + } + const invocationId = + state.activeInvocationId ?? this.#effects.ids.nextId("invocation"); + const toolRequest = state.pendingToolRequests.shift()!; + let execution: ToolExecution; + try { + execution = await this.#executeTool( + state, + invocationId, + toolRequest, + cancellation, + ); + } catch (error) { + if ( + error instanceof TurnCancellationError || + cancellation.isCancellationRequested + ) { + return this.#commitCancellation(state, cancellation); + } + if (error instanceof TurnPortError && error.configurationError) { + return this.#commitFailure( + state, + "tool_configuration_error", + error.message, + cancellation, + ); + } + if (error instanceof TurnPortError) { + return this.#commitFailure( + state, + "permission_error", + error.message, + cancellation, + ); + } + throw error; + } + + const outcomeUnknown = execution.result.outcome === "indeterminate"; + state.toolResults.push(execution.result); + if (!state.pendingModelResponse) { + state.messages.push( + toolResultMessage( + toolRequest.id, + modelVisibleToolOutput(execution.result), + ), + ); + } + await this.#persistToolResult( + state, + invocationId, + toolRequest, + execution.executed, + ); + if (outcomeUnknown) { + state.reconciliationRequired = true; + return this.#commitReconciliation( + state, + "effect_outcome_unknown", + "Tool effect outcome is unknown and requires reconciliation", + cancellation, + ); + } + if ( + state.pendingToolRequests.length === 0 && + !state.pendingModelResponse + ) { + state.activeInvocationId = undefined; + state.iteration += 1; + } + continue; + } + + const invocationId = this.#effects.ids.nextId("invocation"); + if (state.policyAppliedForIteration) { + state.policyAppliedForIteration = false; + } else { + let policyResult; + try { + policyResult = await this.#effects.policy.beforeModel( + new HostPolicyRequest({ + sessionId: state.sessionId, + turnId: state.turnId, + iteration: state.iteration, + messages: state.messages, + stablePrefixMessages: state.stablePrefixMessages, + inputs: state.inputs, + }), + cancellation, + ); + } catch (error) { + if (cancellation.isCancellationRequested) { + return this.#commitCancellation(state, cancellation); + } + const policyError = + error instanceof TurnHostPolicyError + ? error + : new TurnHostPolicyError("policy_error", errorMessage(error), { + cause: error, + }); + return this.#commitFailure( + state, + policyError.errorKind, + policyError.message, + cancellation, + ); + } + if (cancellation.isCancellationRequested) { + return this.#commitCancellation(state, cancellation); + } + if ( + policyResult.stablePrefixMessages < 0 || + policyResult.stablePrefixMessages > policyResult.messages.length + ) { + return this.#commitFailure( + state, + "policy_error", + "Host policy stable prefix exceeds rewritten message count", + cancellation, + ); + } + const policyChanged = + !messagesEqual(state.messages, policyResult.messages) || + state.stablePrefixMessages !== policyResult.stablePrefixMessages; + if (policyChanged) { + state.messages = [...policyResult.messages]; + state.stablePrefixMessages = policyResult.stablePrefixMessages; + await this.#persistPolicyUpdate( + state, + invocationId, + policyResult.metadata, + ); + state.policyAppliedForIteration = false; + } + } + + if (cancellation.isCancellationRequested) { + return this.#commitCancellation(state, cancellation); + } + const contextRequest = new ContextRequest({ + sessionId: state.sessionId, + turnId: state.turnId, + invocationId, + iteration: state.iteration, + messages: state.messages, + stablePrefixMessages: Math.min( + state.stablePrefixMessages, + state.messages.length, + ), + contextState: new InvocationContextState({ + portability: state.portability, + delegatedState: state.delegatedState, + }), + inputs: state.inputs, + }); + let snapshot: ModelInvocationContextSnapshot; + try { + snapshot = await this.#context.prepare(contextRequest, cancellation); + } catch (error) { + if ( + error instanceof TurnCancellationError || + cancellation.isCancellationRequested + ) { + return this.#commitCancellation(state, cancellation); + } + return this.#commitFailure( + state, + "context_error", + error instanceof TurnContextError ? error.message : errorMessage(error), + cancellation, + ); + } + await this.#emit(state, "context_prepared", { + invocationId, + iteration: state.iteration, + payload: snapshot.save(), + }); + state.snapshots.push(snapshot); + + if (cancellation.isCancellationRequested) { + return this.#commitCancellation(state, cancellation); + } + + const modelRequest = new ModelInvocationRequest({ context: snapshot }); + state.activeInvocationId = invocationId; + let attempt = 0; + let modelResponse: ModelInvocationResponse | undefined; + while (!modelResponse) { + if (cancellation.isCancellationRequested) { + return this.#commitCancellation(state, cancellation); + } + await this.#emit(state, "model_invocation_started", { + invocationId, + iteration: state.iteration, + payload: { + snapshotId: snapshot.id, + attempt, + messageCount: snapshot.messages.length, + }, + }); + try { + modelResponse = await this.#effects.model.invoke( + modelRequest, + cancellation, + this.#effects.stream, + ); + } catch (error) { + if (cancellation.isCancellationRequested) { + return this.#commitCancellation(state, cancellation); + } + const source = normalizePortError(error); + const failedAttempt = attempt; + attempt += 1; + const exhausted = + source.outcomeUnknown || attempt >= state.maxModelAttempts; + await this.#emit(state, "model_invocation_failed", { + invocationId, + iteration: state.iteration, + payload: { + attempt: failedAttempt, + exhausted, + outcomeUnknown: source.outcomeUnknown, + message: source.message, + }, + }); + if (source.outcomeUnknown) { + state.reconciliationRequired = true; + state.modelReconciliation = new ModelReconciliationState({ + invocationId, + request: modelRequest, + failedAttempt, + message: source.message, + metadata: source.metadata, + }); + await this.#persistModelReconciliationRequired(state, invocationId); + return this.#commitReconciliation( + state, + "model_outcome_unknown", + source.message, + cancellation, + ); + } + if (exhausted) { + return this.#commitFailure( + state, + "model_error", + source.message, + cancellation, + ); + } + try { + await this.#effects.retry.backoff( + new RetryPolicyRequest({ + failedAttempts: attempt, + nextAttempt: attempt + 1, + maxAttempts: state.maxModelAttempts, + reason: source.message, + }), + cancellation, + ); + } catch (retryError) { + if ( + retryError instanceof TurnCancellationError || + cancellation.isCancellationRequested + ) { + return this.#commitCancellation(state, cancellation); + } + return this.#commitFailure( + state, + "retry_policy_error", + errorMessage(retryError), + cancellation, + ); + } + if (cancellation.isCancellationRequested) { + return this.#commitCancellation(state, cancellation); + } + } + } + + state.modelReconciliation = undefined; + state.reconciliationRequired = false; + try { + state.applyModelResponse(invocationId, modelResponse); + } catch (error) { + return this.#commitFailure( + state, + "provider_state_error", + errorMessage(error), + cancellation, + ); + } + await this.#persistModelResponse(state, invocationId, modelResponse); + + if (cancellation.isCancellationRequested) { + return this.#commitCancellation(state, cancellation); + } + if (state.finalOutputReady) { + state.output = state.pendingOutput; + return this.#applyFinalPolicy(state, cancellation); + } + } + + return this.#commitFailure( + state, + "max_iterations", + "Maximum model iterations reached", + cancellation, + ); + } + + #validateRequest(request: TurnEngineRequest): void { + if (!request.sessionId) { + throw new InvalidTurnRequestError("sessionId is required"); + } + if (!request.turnId) { + throw new InvalidTurnRequestError("turnId is required"); + } + if (request.maxModelAttempts <= 0) { + throw new InvalidTurnRequestError( + "maxModelAttempts must be greater than zero", + ); + } + if (request.maxIterations < 0) { + throw new InvalidTurnRequestError( + "maxIterations must not be negative", + ); + } + if (request.startIteration > request.maxIterations) { + throw new InvalidTurnRequestError( + "startIteration must not exceed maxIterations", + ); + } + if ( + request.stablePrefixMessages < 0 || + request.stablePrefixMessages > request.messages.length + ) { + throw new InvalidTurnRequestError( + "stablePrefixMessages exceeds initial message count", + ); + } + if ( + request.contextState.portability === "portable" && + (request.contextState.delegatedState ?? []).length > 0 + ) { + throw new InvalidTurnRequestError( + "portable turns cannot begin with delegated provider state", + ); + } + } + + #finalizeToolExchange(state: TurnState): ModelToolResult[] { + const response = state.pendingModelResponse; + if (!response) { + return []; + } + const requests = response.toolRequests ?? []; + if (requests.length === 0) { + state.pendingModelResponse = undefined; + return []; + } + const results = requests.map((request) => { + const result = state.toolResults.find( + (candidate) => candidate.requestId === request.id, + ); + if (!result) { + throw TurnPortError.configuration( + "Tool exchange is incomplete and cannot be formatted", + ); + } + return result; + }); + const messages = this.#effects.conversation.formatToolExchange( + response, + results, + ); + state.pendingModelResponse = undefined; + state.messages.push(...messages); + return results; + } + + async #executeTool( + state: TurnState, + invocationId: string, + request: ModelToolRequest, + cancellation: TurnCancellationToken, + ): Promise { + cancellation.throwIfCancellationRequested(); + await this.#emit(state, "permission_requested", { + invocationId, + iteration: state.iteration, + payload: { toolRequest: request.save() }, + }); + let decision: EnginePermissionDecision; + try { + decision = await this.#effects.permission.authorize( + request, + cancellation, + ); + } catch (error) { + throw normalizePortError(error); + } + await this.#emitPermissionResolved(state, invocationId, request, decision); + if (!decision.approved) { + const metadata = decision.metadata ?? {}; + return { + executed: false, + result: new ModelToolResult({ + requestId: request.id, + name: request.name, + outcome: "failed", + output: decision.reason ?? "Permission denied", + errorKind: + typeof metadata["errorKind"] === "string" + ? metadata["errorKind"] + : "permission_denied", + metadata, + }), + }; + } + + cancellation.throwIfCancellationRequested(); + await this.#emit(state, "tool_execution_started", { + invocationId, + iteration: state.iteration, + payload: { toolRequest: request.save() }, + }); + cancellation.throwIfCancellationRequested(); + try { + const result = await this.#effects.tools.execute(request, cancellation); + if (result.requestId !== request.id || result.name !== request.name) { + throw TurnPortError.configuration( + `Tool '${request.name}' returned a result for '${result.requestId || ""}'`, + ); + } + return { + executed: true, + result, + }; + } catch (error) { + const source = normalizePortError(error); + if (source.configurationError) { + throw source; + } + return { + executed: true, + result: new ModelToolResult({ + requestId: request.id, + name: request.name, + outcome: source.outcomeUnknown ? "indeterminate" : "failed", + output: source.outcomeUnknown + ? `Tool '${request.name}' outcome is unknown and requires reconciliation: ${source.message}` + : `Tool '${request.name}' failed: ${source.message}`, + errorKind: source.outcomeUnknown + ? "effect_outcome_unknown" + : "tool_error", + metadata: source.metadata, + }), + }; + } + } + + async #persistPolicyUpdate( + state: TurnState, + invocationId: string, + metadata?: Record, + ): Promise { + const sequence = state.sequence + 1; + state.policyAppliedForIteration = true; + const checkpoint = this.#buildCheckpoint(state, sequence, true); + const event = this.#buildEvent(state, sequence, "policy_applied", { + invocationId, + iteration: state.iteration, + payload: { + messages: state.messages.map((message) => message.save()), + stablePrefixMessages: state.stablePrefixMessages, + metadata, + }, + }); + const checkpointEvent = this.#buildCheckpointEvent( + state, + checkpoint, + invocationId, + ); + await this.#appendAtomic( + state, + [event, checkpointEvent], + checkpoint, + "host policy", + invocationId, + ); + } + + async #persistModelResponse( + state: TurnState, + invocationId: string, + response: ModelInvocationResponse, + ): Promise { + const sequence = state.sequence + 1; + const checkpoint = this.#buildCheckpoint(state, sequence, false); + const event = this.#buildEvent( + state, + sequence, + "model_invocation_completed", + { + invocationId, + iteration: state.iteration, + payload: { + hasOutput: response.output !== undefined, + toolRequests: (response.toolRequests ?? []).length, + nextPortability: response.nextContextState?.portability, + delegatedState: response.nextContextState?.delegatedState?.map( + (reference) => reference.save(), + ), + metadata: response.metadata, + }, + }, + ); + const checkpointEvent = this.#buildCheckpointEvent( + state, + checkpoint, + invocationId, + ); + await this.#appendAtomic( + state, + [event, checkpointEvent], + checkpoint, + "model response", + invocationId, + ); + } + + async #persistModelReconciliationRequired( + state: TurnState, + invocationId: string, + ): Promise { + const sequence = state.sequence + 1; + const checkpoint = this.#buildCheckpoint(state, sequence, false); + const event = this.#buildEvent( + state, + sequence, + "model_reconciliation_required", + { + invocationId, + iteration: state.iteration, + payload: state.modelReconciliation?.save(), + }, + ); + const checkpointEvent = this.#buildCheckpointEvent( + state, + checkpoint, + invocationId, + ); + await this.#appendAtomic( + state, + [event, checkpointEvent], + checkpoint, + "model reconciliation", + invocationId, + ); + } + + async #persistModelReconciliation( + state: TurnState, + invocationId: string, + reconciliation: ModelReconciliationState, + response: ModelInvocationResponse, + ): Promise { + const sequence = state.sequence + 1; + const checkpoint = this.#buildCheckpoint(state, sequence, false); + const event = this.#buildEvent( + state, + sequence, + "model_invocation_reconciled", + { + invocationId, + iteration: state.iteration, + payload: { + reconciliation: reconciliation.save(), + hasOutput: response.output !== undefined, + toolRequests: (response.toolRequests ?? []).length, + metadata: response.metadata, + }, + }, + ); + const checkpointEvent = this.#buildCheckpointEvent( + state, + checkpoint, + invocationId, + ); + await this.#appendAtomic( + state, + [event, checkpointEvent], + checkpoint, + "model reconciliation resolution", + invocationId, + ); + } + + async #persistToolResult( + state: TurnState, + invocationId: string, + request: ModelToolRequest, + executed: boolean, + ): Promise { + const result = state.toolResults[state.toolResults.length - 1]; + if (executed) { + const sequence = state.sequence + 1; + const checkpoint = this.#buildCheckpoint(state, sequence, false); + const event = this.#buildEvent( + state, + sequence, + "tool_execution_completed", + { + invocationId, + iteration: state.iteration, + payload: { toolResult: result.save() }, + }, + ); + const checkpointEvent = this.#buildCheckpointEvent( + state, + checkpoint, + invocationId, + ); + await this.#appendAtomic( + state, + [event, checkpointEvent], + checkpoint, + "tool result", + request.id, + ); + return; + } + + // Denial is a committed model-visible result, never an execution-completed + // event. Commit any earlier uncommitted results at the same boundary so + // observable result order remains the original model-request order. + let sequence = state.sequence; + const events: EngineEvent[] = []; + for (const candidate of state.toolResults) { + if (state.committedToolResultIds.has(candidate.requestId)) { + continue; + } + sequence += 1; + events.push( + this.#buildEvent(state, sequence, "tool_result_committed", { + invocationId, + iteration: state.iteration, + payload: { toolResult: candidate.save() }, + }), + ); + state.committedToolResultIds.add(candidate.requestId); + } + const checkpoint = this.#buildCheckpoint(state, sequence, false); + const checkpointEvent = this.#buildCheckpointEvent( + state, + checkpoint, + invocationId, + ); + events.push(checkpointEvent); + await this.#appendAtomic( + state, + events, + checkpoint, + "permission result", + request.id, + ); + } + + async #persistToolExchange( + state: TurnState, + invocationId: string, + results: readonly ModelToolResult[], + ): Promise { + let sequence = state.sequence; + const events: EngineEvent[] = []; + for (const result of results) { + if (state.committedToolResultIds.has(result.requestId)) { + continue; + } + sequence += 1; + events.push( + this.#buildEvent(state, sequence, "tool_result_committed", { + invocationId, + iteration: state.iteration, + payload: { toolResult: result.save() }, + }), + ); + state.committedToolResultIds.add(result.requestId); + } + sequence += 1; + events.push( + this.#buildEvent(state, sequence, "conversation_updated", { + invocationId, + iteration: state.iteration, + payload: { messageCount: state.messages.length }, + }), + ); + const checkpoint = this.#buildCheckpoint(state, sequence, false); + events.push(this.#buildCheckpointEvent(state, checkpoint, invocationId)); + await this.#appendAtomic( + state, + events, + checkpoint, + "tool exchange", + invocationId, + ); + } + + async #persistToolReconciliation( + state: TurnState, + result: ModelToolResult, + ): Promise { + const sequence = state.sequence + 1; + const invocationId = state.activeInvocationId ?? "reconciliation"; + const checkpoint = this.#buildCheckpoint(state, sequence, false); + const event = this.#buildEvent( + state, + sequence, + "tool_result_reconciled", + { + invocationId, + iteration: state.iteration, + payload: { toolResult: result.save() }, + }, + ); + const checkpointEvent = this.#buildCheckpointEvent( + state, + checkpoint, + invocationId, + ); + await this.#appendAtomic( + state, + [event, checkpointEvent], + checkpoint, + "tool reconciliation", + result.requestId, + ); + } + + async #appendAtomic( + state: TurnState, + events: readonly EngineEvent[], + checkpoint: EngineCheckpoint, + stage: string, + requestId: string, + ): Promise { + try { + await this.#effects.durability.appendWithCheckpoint(events, checkpoint); + } catch (error) { + throw new TurnEngineRecoveryRequiredError({ + stage, + requestId, + checkpoint, + toolResults: [...state.toolResults], + cause: error, + }); + } + state.sequence = + events.length > 0 + ? events[events.length - 1].sequence + : checkpoint.lastSequence; + } + + #buildCheckpoint( + state: TurnState, + lastSequence: number, + resumeSameIteration: boolean, + ): EngineCheckpoint { + return new EngineCheckpoint({ + id: this.#effects.ids.nextId("checkpoint"), + sessionId: state.sessionId, + turnId: state.turnId, + runId: state.runId, + parentRunId: state.parentRunId, + delegationDepth: state.delegationDepth, + iteration: state.iteration, + lastSequence, + messages: [...state.messages], + stablePrefixMessages: state.stablePrefixMessages, + inputs: state.inputs, + activeInvocationId: state.activeInvocationId, + pendingToolRequests: [...state.pendingToolRequests], + completedToolResults: [...state.toolResults], + completedModelIterations: state.completedModelIterations, + reconciliationRequired: + state.reconciliationRequired || + state.toolResults.at(-1)?.outcome === "indeterminate", + modelReconciliation: state.modelReconciliation, + pendingOutput: state.pendingOutput, + finalOutputReady: state.finalOutputReady, + pendingModelResponse: state.pendingModelResponse, + resumeSameIteration, + policyAppliedForIteration: state.policyAppliedForIteration, + contextState: new InvocationContextState({ + portability: state.portability, + delegatedState: [...state.delegatedState], + }), + metadata: + state.committedToolResultIds.size === 0 + ? undefined + : { + committedToolResultIds: [...state.committedToolResultIds], + }, + }); + } + + #buildCheckpointEvent( + state: TurnState, + checkpoint: EngineCheckpoint, + invocationId: string, + ): EngineEvent { + return this.#buildEvent( + state, + checkpoint.lastSequence + 1, + "checkpoint_created", + { + invocationId, + iteration: checkpoint.iteration, + payload: { + checkpointId: checkpoint.id, + includedThroughSequence: checkpoint.lastSequence, + }, + }, + ); + } + + async #emitPermissionResolved( + state: TurnState, + invocationId: string, + request: ModelToolRequest, + decision: EnginePermissionDecision, + ): Promise { + await this.#emit(state, "permission_resolved", { + invocationId, + iteration: state.iteration, + payload: { + toolRequestId: request.id, + decision: decision.save(), + }, + }); + } + + async #applyFinalPolicy( + state: TurnState, + cancellation: TurnCancellationToken, + ): Promise { + if (cancellation.isCancellationRequested) { + return this.#commitCancellation(state, cancellation); + } + let result; + try { + result = await this.#effects.policy.beforeCommit( + new FinalOutputPolicyRequest({ + sessionId: state.sessionId, + turnId: state.turnId, + iteration: state.iteration, + messages: state.messages, + output: state.output, + inputs: state.inputs, + }), + cancellation, + ); + } catch (error) { + if (cancellation.isCancellationRequested) { + return this.#commitCancellation(state, cancellation); + } + const policyError = + error instanceof TurnHostPolicyError + ? error + : new TurnHostPolicyError("policy_error", errorMessage(error), { + cause: error, + }); + return this.#commitFailure( + state, + policyError.errorKind, + policyError.message, + cancellation, + ); + } + if (cancellation.isCancellationRequested) { + return this.#commitCancellation(state, cancellation); + } + state.output = result.output; + return this.#commit(state, "success", "turn_committed", cancellation); + } + + #commitCancellation( + state: TurnState, + cancellation: TurnCancellationToken, + ): Promise { + return this.#commit(state, "cancelled", "turn_cancelled", cancellation); + } + + #commitFailure( + state: TurnState, + errorKind: string, + message: string, + cancellation: TurnCancellationToken, + ): Promise { + state.output = { errorKind, message }; + return this.#commit(state, "failed", "turn_failed", cancellation); + } + + #commitReconciliation( + state: TurnState, + errorKind: string, + message: string, + cancellation: TurnCancellationToken, + ): Promise { + state.output = { errorKind, message }; + return this.#commit( + state, + "reconciliation_required", + "turn_reconciliation_required", + cancellation, + ); + } + + async #commit( + state: TurnState, + status: EngineTurnStatus, + kind: EngineEventKind, + cancellation: TurnCancellationToken, + ): Promise { + await this.#emit(state, kind, { + iteration: state.iteration, + payload: { status, output: state.output }, + }); + const commit = new TurnCommit({ + sessionId: state.sessionId, + turnId: state.turnId, + status, + output: state.output, + messages: [...state.messages], + iterations: state.completedModelIterations, + lastSequence: state.sequence, + contextState: new InvocationContextState({ + portability: state.portability, + delegatedState: [...state.delegatedState], + }), + modelReconciliation: state.modelReconciliation, + }); + + let postCommitError: string | undefined; + if (status === "success") { + const effectId = `post_commit:${commit.sessionId.length}:${commit.sessionId}:${commit.turnId.length}:${commit.turnId}`; + try { + await this.#emit(state, "post_commit_started", { + iteration: state.iteration, + payload: { effectId }, + }); + } catch (error) { + postCommitError = `Post-commit effect '${effectId}' was not started because its start event could not be persisted: ${errorMessage(error)}`; + } + if (!postCommitError) { + try { + await this.#effects.postCommit.afterCommit( + effectId, + commit, + cancellation, + ); + try { + await this.#emit(state, "post_commit_completed", { + iteration: state.iteration, + payload: { effectId }, + }); + } catch (error) { + postCommitError = `Post-commit effect '${effectId}' completed, but its completion event could not be persisted: ${errorMessage(error)}`; + } + } catch (error) { + const message = errorMessage(error); + try { + await this.#emit(state, "post_commit_failed", { + iteration: state.iteration, + payload: { effectId, message }, + }); + postCommitError = message; + } catch (eventError) { + postCommitError = `${message}; failure event for post-commit effect '${effectId}' could not be persisted: ${errorMessage(eventError)}`; + } + } + } + } + commit.lastSequence = state.sequence; + return new TurnEngineResult({ + commit, + snapshots: state.snapshots, + toolResults: state.toolResults, + postCommitError, + }); + } + + async #emit( + state: TurnState, + kind: EngineEventKind, + options: EventOptions, + ): Promise { + const sequence = state.sequence + 1; + const event = this.#buildEvent(state, sequence, kind, options); + try { + await this.#effects.durability.append(event); + } catch (error) { + throw new TurnEnginePortError("event journal", error); + } + state.sequence = sequence; + } + + #buildEvent( + state: TurnState, + sequence: number, + kind: EngineEventKind, + options: EventOptions, + ): EngineEvent { + return new EngineEvent({ + sequence, + id: this.#effects.ids.nextId("event"), + timestamp: this.#effects.clock.now(), + sessionId: state.sessionId, + turnId: state.turnId, + runId: state.runId, + parentRunId: state.parentRunId, + delegationDepth: state.delegationDepth, + invocationId: options.invocationId, + iteration: options.iteration, + kind, + payload: options.payload, + }); + } +} + +interface EventOptions { + invocationId?: string; + iteration?: number; + payload?: unknown; +} + +interface ToolExecution { + result: ModelToolResult; + executed: boolean; +} + +class TurnState { + readonly sessionId: string; + readonly turnId: string; + readonly runId: string; + readonly parentRunId?: string; + readonly delegationDepth: number; + messages: Message[]; + readonly inputs?: unknown; + readonly maxIterations: number; + readonly maxModelAttempts: number; + stablePrefixMessages: number; + portability: InvocationContextState["portability"]; + delegatedState: DelegatedStateReference[]; + activeInvocationId?: string; + pendingToolRequests: ModelToolRequest[]; + reconciliationRequired: boolean; + modelReconciliation?: ModelReconciliationState; + completedModelIterations: number; + pendingOutput?: unknown; + finalOutputReady: boolean; + pendingModelResponse?: ModelInvocationResponse; + policyAppliedForIteration: boolean; + reconciliationResolution?: ModelToolResult; + modelReconciliationResolution?: ModelInvocationResponse; + iteration: number; + sequence: number; + output?: unknown; + readonly snapshots: ModelInvocationContextSnapshot[] = []; + readonly toolResults: ModelToolResult[]; + readonly committedToolResultIds: Set; + + constructor(request: TurnEngineRequest) { + this.sessionId = request.sessionId; + this.turnId = request.turnId; + this.runId = request.runId; + this.parentRunId = request.parentRunId; + this.delegationDepth = request.delegationDepth; + this.messages = [...request.messages]; + this.inputs = request.inputs; + this.maxIterations = request.maxIterations; + this.maxModelAttempts = request.maxModelAttempts; + this.stablePrefixMessages = request.stablePrefixMessages; + this.portability = request.contextState.portability; + this.delegatedState = [...(request.contextState.delegatedState ?? [])]; + this.activeInvocationId = request.activeInvocationId; + this.pendingToolRequests = [...request.pendingToolRequests]; + this.reconciliationRequired = request.reconciliationRequired; + this.modelReconciliation = request.modelReconciliation; + this.completedModelIterations = request.completedModelIterations; + this.pendingOutput = request.pendingOutput; + this.finalOutputReady = request.finalOutputReady; + this.pendingModelResponse = request.pendingModelResponse; + this.policyAppliedForIteration = request.policyAppliedForIteration; + this.reconciliationResolution = request.reconciliationResolution; + this.modelReconciliationResolution = + request.modelReconciliationResolution; + this.iteration = request.startIteration; + this.sequence = request.initialSequence; + this.toolResults = [...request.completedToolResults]; + this.committedToolResultIds = new Set(request.committedToolResultIds); + } + + applyModelResponse( + invocationId: string, + response: ModelInvocationResponse, + ): void { + this.completedModelIterations += 1; + const requests = response.toolRequests ?? []; + if (requests.length === 0) { + this.messages.push(...(response.assistantMessages ?? [])); + this.pendingModelResponse = undefined; + } else { + this.pendingModelResponse = response; + } + this.applyProviderState(response); + this.activeInvocationId = invocationId; + this.pendingToolRequests = [...requests]; + this.pendingOutput = response.output; + this.finalOutputReady = requests.length === 0; + } + + private applyProviderState(response: ModelInvocationResponse): void { + if (response.nextContextState) { + this.portability = response.nextContextState.portability; + this.delegatedState = [ + ...(response.nextContextState.delegatedState ?? []), + ]; + } else if (this.portability === "portable") { + this.delegatedState = []; + } + if (this.portability === "portable" && this.delegatedState.length > 0) { + throw new Error( + "Portable provider state cannot retain delegated references", + ); + } + if (this.portability === "delegated" && this.delegatedState.length === 0) { + throw new Error( + "Delegated provider state requires at least one reference", + ); + } + } +} + +function cloneCheckpoint(checkpoint: EngineCheckpoint): EngineCheckpoint { + return EngineCheckpoint.load(checkpoint.save()); +} + +function checkpointCommittedToolResultIds( + checkpoint: EngineCheckpoint, +): string[] { + const value = checkpoint.metadata?.["committedToolResultIds"]; + return Array.isArray(value) + ? value.filter((item): item is string => typeof item === "string") + : []; +} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +function messagesEqual(left: readonly Message[], right: readonly Message[]): boolean { + if (left.length !== right.length) { + return false; + } + return left.every( + (message, index) => + JSON.stringify(message.save()) === JSON.stringify(right[index].save()), + ); +} diff --git a/runtime/typescript/packages/core/src/harness/turn-runner.ts b/runtime/typescript/packages/core/src/harness/turn-runner.ts index a650c1830..9ecf23fb7 100644 --- a/runtime/typescript/packages/core/src/harness/turn-runner.ts +++ b/runtime/typescript/packages/core/src/harness/turn-runner.ts @@ -185,7 +185,7 @@ export class ReferenceTurnRunner { this.recordTurn("permission_completed", turnId, iteration, decision.save()); if (!decision.approved) { - return new HostToolResult({ + const result = new HostToolResult({ requestId: toolRequest.requestId, toolCallId: toolRequest.toolCallId, toolName: toolRequest.toolName, @@ -193,6 +193,8 @@ export class ReferenceTurnRunner { errorKind: "permission_denied", result: { message: decision.reason ?? "Permission denied" }, }); + this.recordTurn("tool_result", turnId, iteration, result.save()); + return result; } this.recordTurn("tool_execution_start", turnId, iteration, toolRequest.save()); diff --git a/runtime/typescript/packages/core/src/index.ts b/runtime/typescript/packages/core/src/index.ts index 74c4da32b..8d3aedd84 100644 --- a/runtime/typescript/packages/core/src/index.ts +++ b/runtime/typescript/packages/core/src/index.ts @@ -58,12 +58,15 @@ export { // Loader load, type LoadOptions, + createModelInfo, + enrichModelInfo, // Pipeline functions validateInputs, render, parse, process, + processStream, prepare, run, turn, @@ -106,6 +109,11 @@ export { cast, } from "./core/index.js"; +export * from "./core/turn-engine-cancellation.js"; +export * from "./core/turn-engine-context.js"; +export * from "./core/turn-engine-ports.js"; +export * from "./core/turn-engine.js"; + // --------------------------------------------------------------------------- // Implementations (core-provided: renderers + parsers only) // --------------------------------------------------------------------------- @@ -197,6 +205,13 @@ export { PermissionDecision, HostToolRequest, HostToolResult, + StreamChunk, + TextChunk, + ThinkingChunk, + ToolChunk, + UsageChunk, + ErrorChunk, + InvocationUsage, type EventJournalWriter, type EventSink, type PermissionResolver, diff --git a/runtime/typescript/packages/core/src/renderers/common.ts b/runtime/typescript/packages/core/src/renderers/common.ts index fb477099c..68822747b 100644 --- a/runtime/typescript/packages/core/src/renderers/common.ts +++ b/runtime/typescript/packages/core/src/renderers/common.ts @@ -8,13 +8,10 @@ * @module */ -import { randomUUID } from "node:crypto"; +import { randomBytes } from "node:crypto"; import type { Prompty } from "../model/agent/prompty.js"; import { RICH_KINDS } from "../core/types.js"; -/** Map of input name → nonce string (set during rendering, read during prepare). */ -let lastNonces: Map = new Map(); - /** * Prepare render inputs: replace thread/image/file/audio values with nonces. * @@ -30,27 +27,15 @@ export function prepareRenderInputs( for (const [name, kind] of Object.entries(richNames)) { if (kind === "thread" || RICH_KINDS.has(kind)) { - const nonce = `__prompty_nonce_${randomUUID().replace(/-/g, "")}__`; + const nonce = `__PROMPTY_THREAD_${randomBytes(4).toString("hex")}_${name}__`; nonces.set(name, nonce); modified[name] = nonce; } } - // Stash for retrieval by prepare() - lastNonces = nonces; return [modified, nonces]; } -/** Retrieve the last nonce mapping set by `prepareRenderInputs`. */ -export function getLastNonces(): Map { - return lastNonces; -} - -/** Clear the stashed nonces. */ -export function clearLastNonces(): void { - lastNonces = new Map(); -} - /** * Get map of `{propertyName: kind}` for inputs with rich kinds * (thread, image, file, audio). diff --git a/runtime/typescript/packages/core/src/renderers/index.ts b/runtime/typescript/packages/core/src/renderers/index.ts index 2d51c32bf..c30b97f3a 100644 --- a/runtime/typescript/packages/core/src/renderers/index.ts +++ b/runtime/typescript/packages/core/src/renderers/index.ts @@ -1,3 +1,3 @@ export { NunjucksRenderer } from "./nunjucks.js"; export { MustacheRenderer } from "./mustache.js"; -export { prepareRenderInputs, getLastNonces, clearLastNonces } from "./common.js"; +export { prepareRenderInputs } from "./common.js"; diff --git a/runtime/typescript/packages/core/src/renderers/mustache.ts b/runtime/typescript/packages/core/src/renderers/mustache.ts index 175d14871..e016f3d6f 100644 --- a/runtime/typescript/packages/core/src/renderers/mustache.ts +++ b/runtime/typescript/packages/core/src/renderers/mustache.ts @@ -7,15 +7,13 @@ import Mustache from "mustache"; import type { Prompty } from "../model/agent/prompty.js"; import type { Renderer } from "../core/interfaces.js"; -import { prepareRenderInputs } from "./common.js"; export class MustacheRenderer implements Renderer { async render( - agent: Prompty, + _agent: Prompty, template: string, inputs: Record, ): Promise { - const [modified] = prepareRenderInputs(agent, inputs); - return Mustache.render(template, modified); + return Mustache.render(template, inputs); } } diff --git a/runtime/typescript/packages/core/src/renderers/nunjucks.ts b/runtime/typescript/packages/core/src/renderers/nunjucks.ts index aabcef453..d3e94abaf 100644 --- a/runtime/typescript/packages/core/src/renderers/nunjucks.ts +++ b/runtime/typescript/packages/core/src/renderers/nunjucks.ts @@ -11,7 +11,6 @@ import nunjucks from "nunjucks"; import type { Prompty } from "../model/agent/prompty.js"; import type { Renderer } from "../core/interfaces.js"; -import { prepareRenderInputs } from "./common.js"; type NunjucksRuntime = { memberLookup: (object: unknown, property: unknown) => unknown; @@ -100,11 +99,10 @@ function renderSafely(template: string, inputs: Record): string export class NunjucksRenderer implements Renderer { async render( - agent: Prompty, + _agent: Prompty, template: string, inputs: Record, ): Promise { - const [modified] = prepareRenderInputs(agent, inputs); - return renderSafely(template, sanitizeInputs(modified)); + return renderSafely(template, sanitizeInputs(inputs)); } } diff --git a/runtime/typescript/packages/core/tests/discovery.test.ts b/runtime/typescript/packages/core/tests/discovery.test.ts new file mode 100644 index 000000000..4de1a05ae --- /dev/null +++ b/runtime/typescript/packages/core/tests/discovery.test.ts @@ -0,0 +1,32 @@ +import { readFileSync } from "node:fs"; +import { resolve } from "node:path"; + +import { describe, expect, it } from "vitest"; + +import { createModelInfo, enrichModelInfo } from "../src/index.js"; + +interface EnrichmentVector { + name: string; + provider: string; + input: Record; + expected: Record; +} + +const vectorFile = resolve( + import.meta.dirname, + "../../../../../spec/vectors/discovery/enrichment_vectors.json", +); +const vectors = ( + JSON.parse(readFileSync(vectorFile, "utf8")) as { vectors: EnrichmentVector[] } +).vectors; + +describe("model capability enrichment vectors", () => { + for (const vector of vectors) { + it(vector.name, () => { + const actual = createModelInfo( + enrichModelInfo(vector.provider, vector.input), + ).save(); + expect(actual).toEqual(vector.expected); + }); + } +}); diff --git a/runtime/typescript/packages/core/tests/loader.test.ts b/runtime/typescript/packages/core/tests/loader.test.ts index 7afa0b0bf..9dfa476c5 100644 --- a/runtime/typescript/packages/core/tests/loader.test.ts +++ b/runtime/typescript/packages/core/tests/loader.test.ts @@ -94,6 +94,41 @@ describe("Loader", () => { expect(agent.description).toBe("shared description"); }); + it.each([ + ["yaml", "metadata.yaml"], + ["yml", "metadata.yml"], + ])("parses ${file:...} %s references as structured data", (_extension, fileName) => { + const root = mkdtempSync(join(tmpdir(), "prompty-loader-")); + const prompt = join(root, "structured-reference.prompty"); + writeFileSync(join(root, fileName), "region: westus\nretries: 3\n", "utf-8"); + writeFileSync( + prompt, + `---\nname: structured-reference\nmetadata: "\${file:${fileName}}"\n---\nHello\n`, + "utf-8", + ); + + const agent = load(prompt); + + expect(agent.metadata).toMatchObject({ region: "westus", retries: 3 }); + }); + + it("preserves JSON and text file reference behavior", () => { + const root = mkdtempSync(join(tmpdir(), "prompty-loader-")); + const prompt = join(root, "references.prompty"); + writeFileSync(join(root, "metadata.json"), '{"region":"eastus","retries":2}', "utf-8"); + writeFileSync(join(root, "description.txt"), "plain text", "utf-8"); + writeFileSync( + prompt, + '---\nname: references\nmetadata: "${file:metadata.json}"\ndescription: "${file:description.txt}"\n---\nHello\n', + "utf-8", + ); + + const agent = load(prompt); + + expect(agent.metadata).toMatchObject({ region: "eastus", retries: 2 }); + expect(agent.description).toBe("plain text"); + }); + it("rejects symlink escapes from the prompt directory", () => { const root = mkdtempSync(join(tmpdir(), "prompty-loader-")); const promptDir = join(root, "prompts"); diff --git a/runtime/typescript/packages/core/tests/pipeline.test.ts b/runtime/typescript/packages/core/tests/pipeline.test.ts index 14a2ca8b8..dc9fd3c99 100644 --- a/runtime/typescript/packages/core/tests/pipeline.test.ts +++ b/runtime/typescript/packages/core/tests/pipeline.test.ts @@ -16,10 +16,11 @@ import { registerProcessor, } from "../src/core/registry.js"; import { Message, text } from "../src/core/types.js"; -import { Prompty } from "@prompty/core"; +import { Prompty, Property, TextChunk } from "@prompty/core"; import type { Renderer, Parser, Executor, Processor } from "../src/core/interfaces.js"; import { NunjucksRenderer } from "../src/renderers/nunjucks.js"; import { PromptyChatParser } from "../src/parsers/prompty.js"; +import { Tracer } from "../src/tracing/tracer.js"; // --------------------------------------------------------------------------- // Mock implementations @@ -142,6 +143,72 @@ describe("Pipeline", () => { const result = await render(agent, { name: "World" }); expect(result).toBe("Hi World"); }); + + it("uses the canonical request-local marker for rich inputs", async () => { + const agent = makeAgent({ instructions: "{{conversation}}" }); + agent.template = { format: { kind: "nunjucks" } } as any; + agent.inputs = [new Property({ name: "conversation", kind: "thread" })]; + registerRenderer("nunjucks", new NunjucksRenderer()); + + const rendered = await render(agent, { + conversation: [{ role: "user", content: "prior message" }], + }); + + expect(rendered).toMatch(/^__PROMPTY_THREAD_[a-f0-9]{8}_conversation__$/); + }); + + it("sanitizes rich-input names with punctuation in trace data", async () => { + const events: [string, unknown][] = []; + Tracer.add("nonce-sanitization", () => (key, value) => { + events.push([key, value]); + }); + const agent = makeAgent({ instructions: "{{conversation-history}}" }); + agent.template = { format: { kind: "mock" } } as any; + agent.inputs = [ + new Property({ name: "conversation-history", kind: "thread" }), + ]; + + try { + await render(agent, { + "conversation-history": [ + { role: "user", content: "prior message" }, + ], + }); + } finally { + Tracer.remove("nonce-sanitization"); + } + + const traceData = JSON.stringify(events); + expect(traceData).toContain("[thread: conversation-history]"); + expect(traceData).not.toContain("__PROMPTY_THREAD_"); + }); + }); + + describe("prepare()", () => { + it("keeps concurrent rich-input mappings and instructions isolated", async () => { + const instructions = "user:\n{{conversation}}"; + const agent = makeAgent({ instructions }); + agent.template = { + format: { kind: "nunjucks" }, + parser: { kind: "prompty" }, + } as any; + agent.inputs = [new Property({ name: "conversation", kind: "thread" })]; + registerRenderer("nunjucks", new NunjucksRenderer()); + registerParser("prompty", new PromptyChatParser()); + + const [first, second] = await Promise.all([ + prepare(agent, { conversation: [{ role: "user", content: "first thread" }] }), + prepare(agent, { conversation: [{ role: "assistant", content: "second thread" }] }), + ]); + + expect(first).toHaveLength(1); + expect(first[0].role).toBe("user"); + expect(first[0].text).toBe("first thread"); + expect(second).toHaveLength(1); + expect(second[0].role).toBe("assistant"); + expect(second[0].text).toBe("second thread"); + expect(agent.instructions).toBe(instructions); + }); }); describe("parse()", () => { @@ -190,6 +257,158 @@ describe("Pipeline", () => { }); describe("turn()", () => { + it("checks cancellation before preparing the prompt", async () => { + const agent = makeAgent(); + agent.template = { format: { kind: "missing" }, parser: { kind: "missing" } } as any; + (agent as any).model = { provider: "mock" }; + const controller = new AbortController(); + controller.abort(); + + await expect( + turn(agent, {}, { signal: controller.signal }), + ).rejects.toThrow("cancelled"); + }); + + it("retries simple turns with the same prepared messages", async () => { + const agent = makeAgent(); + agent.template = { format: { kind: "mock" }, parser: { kind: "mock" } } as any; + (agent as any).model = { provider: "retrying-mock" }; + const requests: Message[][] = []; + + registerExecutor("retrying-mock", { + async execute(_agent, messages) { + requests.push(messages); + if (requests.length === 1) throw new Error("transient"); + return { choices: [{ message: { content: "recovered" } }] }; + }, + formatToolMessages: new MockExecutor().formatToolMessages, + }); + registerProcessor("retrying-mock", new MockProcessor()); + + const result = await turn(agent, {}, { maxLlmRetries: 2 }); + + expect(result).toBe("recovered"); + expect(requests).toHaveLength(2); + expect(requests[0]).toBe(requests[1]); + }); + + it("activates agent mode for tools declared by the prompt", async () => { + const agent = makeAgent(); + agent.template = { format: { kind: "mock" }, parser: { kind: "mock" } } as any; + (agent as any).model = { provider: "asset-tool-mock" }; + agent.tools = [{ name: "echo", kind: "function" }] as any; + let calls = 0; + + registerExecutor("asset-tool-mock", { + async execute() { + calls++; + return calls === 1 + ? { + choices: [{ + message: { + content: null, + tool_calls: [{ + id: "call-asset", + function: { name: "echo", arguments: '{"value":"hello"}' }, + }], + }, + }], + } + : { choices: [{ message: { content: "done" } }] }; + }, + formatToolMessages: new MockExecutor().formatToolMessages, + }); + registerProcessor("asset-tool-mock", new MockProcessor()); + + const result = await turn(agent, {}, { + tools: { echo: (value: unknown) => value }, + }); + + expect(result).toBe("done"); + expect(calls).toBe(2); + }); + + it("returns missing asset tool handlers to the model as failed tool results", async () => { + const agent = makeAgent(); + agent.template = { format: { kind: "mock" }, parser: { kind: "mock" } } as any; + (agent as any).model = { provider: "missing-asset-tool-mock" }; + agent.tools = [{ name: "echo", kind: "function" }] as any; + const requests: Message[][] = []; + + registerExecutor("missing-asset-tool-mock", { + async execute(_agent, messages) { + requests.push(messages); + return requests.length === 1 + ? { + choices: [{ + message: { + content: null, + tool_calls: [{ + id: "call-missing", + function: { name: "echo", arguments: '{"value":"hello"}' }, + }], + }, + }], + } + : { choices: [{ message: { content: "handled failure" } }] }; + }, + formatToolMessages: new MockExecutor().formatToolMessages, + }); + registerProcessor("missing-asset-tool-mock", new MockProcessor()); + + const result = await turn(agent, {}); + + expect(result).toBe("handled failure"); + expect(requests).toHaveLength(2); + expect(requests[1].some((message) => message.text.includes("no callable provided"))).toBe(true); + }); + + it("emits llm_complete only after a simple stream is exhausted", async () => { + const agent = makeAgent(); + agent.template = { format: { kind: "mock" }, parser: { kind: "mock" } } as any; + (agent as any).model = { provider: "streaming-mock" }; + const events: string[] = []; + + registerExecutor("streaming-mock", { + async execute() { + return { + async *[Symbol.asyncIterator]() { + yield { token: "Hello" }; + yield { token: " world" }; + }, + }; + }, + formatToolMessages: new MockExecutor().formatToolMessages, + }); + registerProcessor("streaming-mock", { + async process(_agent, response) { + const source = response as AsyncIterable<{ token: string }>; + return { + async *[Symbol.asyncIterator]() { + for await (const item of source) yield item.token; + }, + }; + }, + async *processStream(response) { + for await (const item of response as AsyncIterable<{ token: string }>) { + yield new TextChunk({ value: item.token }); + } + }, + }); + + const result = await turn(agent, { name: "World" }, { + onEvent: (type) => events.push(type), + }); + expect(events).not.toContain("llm_complete"); + + const chunks: unknown[] = []; + for await (const chunk of result as AsyncIterable) chunks.push(chunk); + + expect(chunks).toEqual(["Hello", " world"]); + expect(events).toContain("llm_complete"); + expect(events.indexOf("llm_complete")).toBeLessThan(events.indexOf("turn_end")); + }); + it("runs a simple agent with no tool calls", async () => { const agent = makeAgent(); agent.template = { format: { kind: "mock" }, parser: { kind: "mock" } } as any; diff --git a/runtime/typescript/packages/core/tests/resilience.test.ts b/runtime/typescript/packages/core/tests/resilience.test.ts index 34fc0aefd..768484ff4 100644 --- a/runtime/typescript/packages/core/tests/resilience.test.ts +++ b/runtime/typescript/packages/core/tests/resilience.test.ts @@ -333,12 +333,15 @@ describe("LLM Call Retry (§9.10)", () => { expect(retryEvent!.data.message).toContain("attempt 2/3"); }); - it("does not retry in simple mode (no tools)", async () => { + it("retries in simple mode (no tools)", async () => { let callCount = 0; const failOnceExecutor: Executor = { async execute(): Promise { callCount++; - throw new Error("API error"); + if (callCount === 1) { + throw new Error("API error"); + } + return { choices: [{ message: { role: "assistant", content: "Recovered" } }] }; }, formatToolMessages() { return []; }, }; @@ -348,11 +351,11 @@ describe("LLM Call Retry (§9.10)", () => { const agent = makeAgent(); - // No tools = simple mode, should NOT retry - await expect( - turn(agent, { name: "Test" }), - ).rejects.toThrow("API error"); - expect(callCount).toBe(1); // Only called once, no retry + const promise = turn(agent, { name: "Test" }, { maxLlmRetries: 2 }); + await vi.runAllTimersAsync(); + + await expect(promise).resolves.toBe("Recovered"); + expect(callCount).toBe(2); }); it("respects maxLlmRetries: 1 (no retries)", async () => { diff --git a/runtime/typescript/packages/core/tests/spec-vectors.test.ts b/runtime/typescript/packages/core/tests/spec-vectors.test.ts index fdc30b66c..7f50819a0 100644 --- a/runtime/typescript/packages/core/tests/spec-vectors.test.ts +++ b/runtime/typescript/packages/core/tests/spec-vectors.test.ts @@ -251,6 +251,28 @@ describe("Spec Vectors: Load", () => { it(`[${vec.name}] ${vec.description}`, testFn); } + + it("requires binding expectations for binding-bearing FunctionTools", () => { + const vector = vectors.find((vec: any) => vec.name === "tools_function_load"); + expect(vector).toBeDefined(); + const expected = structuredClone(vector.expected); + delete expected.tools[0].bindings; + const agent = load(resolve(FIXTURES_DIR, vector.input.fixture)); + + expect(() => validateAgentFields(agent, expected, vector.name)).toThrowError( + /must declare expected bindings/, + ); + }); + + it("accepts equivalent list-form binding expectations", () => { + const vector = vectors.find((vec: any) => vec.name === "tools_function_load"); + expect(vector).toBeDefined(); + const expected = structuredClone(vector.expected); + expected.tools[0].bindings = [{ name: "unit", input: "preferred_unit" }]; + const agent = load(resolve(FIXTURES_DIR, vector.input.fixture)); + + expect(() => validateAgentFields(agent, expected, vector.name)).not.toThrow(); + }); }); /** @@ -334,6 +356,29 @@ function resolveFileRefs(data: any, files: Record): void { } } +function normalizeBindings(bindings: unknown): Array<{ name: string; input: unknown }> { + if (Array.isArray(bindings)) { + return bindings + .map((binding: { name?: unknown; input?: unknown }) => ({ + name: String(binding.name), + input: binding.input, + })) + .sort((left, right) => left.name.localeCompare(right.name)); + } + if (typeof bindings !== "object" || bindings === null) { + return []; + } + return Object.entries(bindings) + .map(([name, value]) => ({ + name, + input: + typeof value === "object" && value !== null && "input" in value + ? (value as { input: unknown }).input + : value, + })) + .sort((left, right) => left.name.localeCompare(right.name)); +} + function validateAgentFields(agent: Prompty, expected: any, vecName: string): void { if (expected.kind !== undefined) { // TS runtime doesn't have a 'kind' field — this is always "prompt" by design @@ -429,17 +474,15 @@ function validateAgentFields(agent: Prompty, expected: any, vecName: string): vo if (ep.enumValues !== undefined) expect(ap.enumValues).toEqual(ep.enumValues); } } + const actualBindings = normalizeBindings((at as FunctionTool).bindings); + if (actualBindings.length > 0) { + expect( + et.bindings, + `[${vecName}] tool '${et.name ?? i}' must declare expected bindings`, + ).toBeDefined(); + } if (et.bindings !== undefined) { - const atBindings = (at as any).bindings as Array<{name: string; input: string}>; - expect(atBindings).toBeDefined(); - expect(atBindings.length).toBeGreaterThan(0); - for (const [bk, bv] of Object.entries(et.bindings as Record)) { - const found = atBindings.find((b: any) => b.name === bk); - expect(found).toBeDefined(); - if (bv.input !== undefined) { - expect(found!.input).toBe(bv.input); - } - } + expect(actualBindings).toEqual(normalizeBindings(et.bindings)); } if (et.serverName !== undefined) { expect((at as any).serverName).toBe(et.serverName); @@ -533,11 +576,7 @@ describe("Spec Vectors: Render", () => { } if (expected.nonce_pattern !== undefined) { - // The spec uses __PROMPTY_THREAD_{hex}_{name}__ but the TS runtime uses - // __prompty_nonce_{uuid_hex}__. Adapt the pattern to match the runtime format. - const adaptedPattern = expected.nonce_pattern - .replace(/__PROMPTY_THREAD_\[a-f0-9\]\{8\}_\w+__/g, "__prompty_nonce_[a-f0-9]{32}__"); - const re = new RegExp(adaptedPattern); + const re = new RegExp(expected.nonce_pattern); expect(rendered).toMatch(re); } }); diff --git a/runtime/typescript/packages/core/tests/turn-engine-harness.ts b/runtime/typescript/packages/core/tests/turn-engine-harness.ts new file mode 100644 index 000000000..6c1770f35 --- /dev/null +++ b/runtime/typescript/packages/core/tests/turn-engine-harness.ts @@ -0,0 +1,177 @@ +import { Message } from "../src/model/conversation/message.js"; +import { EngineCheckpoint } from "../src/model/pipeline/engine-checkpoint.js"; +import { EngineEvent } from "../src/model/pipeline/engine-event.js"; +import { EnginePermissionDecision } from "../src/model/pipeline/engine-permission-decision.js"; +import { InvocationContextState } from "../src/model/pipeline/invocation-context-state.js"; +import { ModelInvocationRequest } from "../src/model/pipeline/model-invocation-request.js"; +import { ModelInvocationResponse } from "../src/model/pipeline/model-invocation-response.js"; +import { ModelToolRequest } from "../src/model/pipeline/model-tool-request.js"; +import { ModelToolResult } from "../src/model/pipeline/model-tool-result.js"; +import type { + Clock, + DurabilityPort, + IdGenerator, + ModelPort, + ModelStreamPort, + PermissionPort, + ToolPort, +} from "../src/core/turn-engine-ports.js"; +import { TurnPortError } from "../src/core/turn-engine-ports.js"; +import type { TurnCancellationToken } from "../src/core/turn-engine-cancellation.js"; + +export type ModelStep = + | ModelInvocationResponse + | Error + | (( + request: ModelInvocationRequest, + cancellation: TurnCancellationToken, + ) => ModelInvocationResponse | Promise); + +export class ScriptedModelPort implements ModelPort { + readonly requests: ModelInvocationRequest[] = []; + readonly steps: ModelStep[]; + + constructor(steps: readonly ModelStep[]) { + this.steps = [...steps]; + } + + async invoke( + request: ModelInvocationRequest, + cancellation: TurnCancellationToken, + _stream: ModelStreamPort, + ): Promise { + this.requests.push(request); + const step = this.steps.shift(); + if (!step) { + throw new TurnPortError("Model script exhausted"); + } + if (step instanceof Error) { + throw step; + } + return typeof step === "function" + ? step(request, cancellation) + : step; + } +} + +export class ScriptedToolPort implements ToolPort { + readonly requests: ModelToolRequest[] = []; + + constructor( + readonly handler: ( + request: ModelToolRequest, + cancellation: TurnCancellationToken, + ) => ModelToolResult | Promise, + ) {} + + async execute( + request: ModelToolRequest, + cancellation: TurnCancellationToken, + ): Promise { + this.requests.push(request); + return this.handler(request, cancellation); + } +} + +export class SelectivePermissionPort implements PermissionPort { + readonly requests: ModelToolRequest[] = []; + + constructor(readonly denied: ReadonlySet = new Set()) {} + + async authorize( + request: ModelToolRequest, + ): Promise { + this.requests.push(request); + const approved = !this.denied.has(request.name); + return new EnginePermissionDecision({ + approved, + reason: approved ? "allowed" : "denied by test", + }); + } +} + +export class RecordingDurabilityPort implements DurabilityPort { + readonly events: EngineEvent[] = []; + readonly checkpoints: EngineCheckpoint[] = []; + readonly attemptedAtomicWrites: { + events: readonly EngineEvent[]; + checkpoint: EngineCheckpoint; + }[] = []; + atomicCalls = 0; + failAtomicAt?: number; + failAppendKind?: EngineEvent["kind"]; + + async append(event: EngineEvent): Promise { + if (event.kind === this.failAppendKind) { + throw new TurnPortError(`append failed for ${event.kind}`); + } + this.events.push(event); + } + + async appendWithCheckpoint( + events: readonly EngineEvent[], + checkpoint: EngineCheckpoint, + ): Promise { + this.atomicCalls += 1; + this.attemptedAtomicWrites.push({ events, checkpoint }); + if (this.atomicCalls === this.failAtomicAt) { + throw new TurnPortError("atomic write failed"); + } + this.events.push(...events); + this.checkpoints.push(checkpoint); + } +} + +export class DeterministicClock implements Clock { + #tick = 0; + + now(): string { + this.#tick += 1; + return `2026-01-01T00:00:${this.#tick.toString().padStart(2, "0")}Z`; + } +} + +export class DeterministicIds implements IdGenerator { + readonly #counts = new Map(); + + nextId(kind: string): string { + const next = (this.#counts.get(kind) ?? 0) + 1; + this.#counts.set(kind, next); + return `${kind}-${next}`; + } +} + +export function response(options: { + output?: unknown; + assistant?: string; + tools?: readonly { + id: string; + name: string; + arguments?: unknown; + }[]; + portability?: InvocationContextState["portability"]; + delegatedState?: InvocationContextState["delegatedState"]; +}): ModelInvocationResponse { + return new ModelInvocationResponse({ + output: options.output, + assistantMessages: + options.assistant === undefined + ? [] + : [Message.assistant(options.assistant)], + toolRequests: (options.tools ?? []).map( + (tool) => + new ModelToolRequest({ + id: tool.id, + name: tool.name, + arguments: tool.arguments, + }), + ), + nextContextState: + options.portability === undefined + ? undefined + : new InvocationContextState({ + portability: options.portability, + delegatedState: options.delegatedState, + }), + }); +} diff --git a/runtime/typescript/packages/core/tests/turn-engine-recovery.test.ts b/runtime/typescript/packages/core/tests/turn-engine-recovery.test.ts new file mode 100644 index 000000000..2d8033188 --- /dev/null +++ b/runtime/typescript/packages/core/tests/turn-engine-recovery.test.ts @@ -0,0 +1,339 @@ +import { describe, expect, it } from "vitest"; + +import { Message } from "../src/model/conversation/message.js"; +import { ModelInvocationResponse } from "../src/model/pipeline/model-invocation-response.js"; +import { ModelToolResult } from "../src/model/pipeline/model-tool-result.js"; +import { ResumeContext } from "../src/model/pipeline/resume-context.js"; +import { ContextPipeline } from "../src/core/turn-engine-context.js"; +import { TurnCancellationToken } from "../src/core/turn-engine-cancellation.js"; +import { + TurnEngine, + TurnEngineRecoveryRequiredError, + TurnEngineRequest, +} from "../src/core/turn-engine.js"; +import { TurnPortError } from "../src/core/turn-engine-ports.js"; +import { + DeterministicClock, + DeterministicIds, + RecordingDurabilityPort, + ScriptedModelPort, + ScriptedToolPort, + response, +} from "./turn-engine-harness.js"; + +function makeEngine(options: { + model: ScriptedModelPort; + tools?: ScriptedToolPort; + durability?: RecordingDurabilityPort; +}): { + engine: TurnEngine; + durability: RecordingDurabilityPort; +} { + const durability = options.durability ?? new RecordingDurabilityPort(); + return { + engine: new TurnEngine(new ContextPipeline(), { + model: options.model, + tools: options.tools, + durability, + clock: new DeterministicClock(), + ids: new DeterministicIds(), + }), + durability, + }; +} + +function request(options: Partial[0]> = {}): TurnEngineRequest { + return new TurnEngineRequest({ + sessionId: "session-recovery", + turnId: "turn-recovery", + messages: [Message.user("run")], + ...options, + }); +} + +describe("turn-engine durability and resume", () => { + it("persists each checkpoint event atomically with its represented state", async () => { + const model = new ScriptedModelPort([ + response({ + tools: [ + { id: "call-a", name: "echo" }, + { id: "call-b", name: "echo" }, + ], + }), + response({ output: "done" }), + ]); + const tools = new ScriptedToolPort( + async (toolRequest) => + new ModelToolResult({ + requestId: toolRequest.id, + name: toolRequest.name, + output: toolRequest.id, + }), + ); + const { engine, durability } = makeEngine({ model, tools }); + + await engine.run(request()); + + expect(durability.attemptedAtomicWrites.length).toBeGreaterThan(0); + for (const write of durability.attemptedAtomicWrites) { + expect(write.events.at(-1)?.kind).toBe("checkpoint_created"); + expect(write.events.at(-1)?.sequence).toBe( + write.checkpoint.lastSequence + 1, + ); + expect(write.checkpoint.stablePrefixMessages).toBe(1); + } + }); + + it("resumes a completed model response without invoking the model again and uses the journal tail", async () => { + const firstDurability = new RecordingDurabilityPort(); + firstDurability.failAtomicAt = 1; + const firstModel = new ScriptedModelPort([response({ output: "saved" })]); + const { engine: firstEngine } = makeEngine({ + model: firstModel, + durability: firstDurability, + }); + + const error = await firstEngine.run(request()).catch((caught) => caught); + expect(error).toBeInstanceOf(TurnEngineRecoveryRequiredError); + const recovery = error as TurnEngineRecoveryRequiredError; + expect(recovery.stage).toBe("model response"); + expect(recovery.checkpoint.finalOutputReady).toBe(true); + expect(firstModel.requests).toHaveLength(1); + + const resumedModel = new ScriptedModelPort([]); + const resumedDurability = new RecordingDurabilityPort(); + const { engine: resumedEngine } = makeEngine({ + model: resumedModel, + durability: resumedDurability, + }); + const result = await resumedEngine.resume( + new ResumeContext({ + checkpoint: recovery.checkpoint, + maxIterations: 10, + maxModelAttempts: 3, + lastJournalSequence: 40, + }), + ); + + expect(result.commit.status).toBe("success"); + expect(result.commit.output).toBe("saved"); + expect(resumedModel.requests).toHaveLength(0); + expect(resumedDurability.events[0].sequence).toBe(41); + }); + + it("resumes a partially committed tool batch at the next uncommitted effect", async () => { + const firstDurability = new RecordingDurabilityPort(); + firstDurability.failAtomicAt = 2; + const model = new ScriptedModelPort([ + response({ + tools: [ + { id: "call-a", name: "echo" }, + { id: "call-b", name: "echo" }, + ], + }), + response({ output: "done" }), + ]); + const executed: string[] = []; + const tools = new ScriptedToolPort(async (toolRequest) => { + executed.push(toolRequest.id); + return new ModelToolResult({ + requestId: toolRequest.id, + name: toolRequest.name, + output: toolRequest.id, + }); + }); + const { engine } = makeEngine({ + model, + tools, + durability: firstDurability, + }); + + const error = await engine.run(request()).catch((caught) => caught); + expect(error).toBeInstanceOf(TurnEngineRecoveryRequiredError); + const recovery = error as TurnEngineRecoveryRequiredError; + expect(recovery.checkpoint.pendingToolRequests?.map((item) => item.id)).toEqual([ + "call-b", + ]); + expect(recovery.checkpoint.completedToolResults?.map((item) => item.requestId)).toEqual([ + "call-a", + ]); + + const resumedDurability = new RecordingDurabilityPort(); + const { engine: resumedEngine } = makeEngine({ + model, + tools, + durability: resumedDurability, + }); + const result = await resumedEngine.resume( + new ResumeContext({ + checkpoint: recovery.checkpoint, + maxIterations: 10, + maxModelAttempts: 3, + lastJournalSequence: 75, + }), + ); + + expect(result.commit.status).toBe("success"); + expect(executed).toEqual(["call-a", "call-b"]); + expect(model.requests).toHaveLength(2); + expect(resumedDurability.events[0].sequence).toBe(76); + expect(result.snapshots?.[0].stablePrefixMessages).toBe(1); + }); +}); + +describe("turn-engine reconciliation", () => { + it("does not retry an indeterminate model and resumes from an explicit resolution", async () => { + const model = new ScriptedModelPort([ + TurnPortError.indeterminate("provider outcome unknown", { + responseId: "resp-1", + }), + ]); + const { engine, durability } = makeEngine({ model }); + + const blocked = await engine.run(request({ maxModelAttempts: 5 })); + expect(blocked.commit.status).toBe("reconciliation_required"); + expect(model.requests).toHaveLength(1); + const checkpoint = durability.checkpoints.find( + (candidate) => candidate.modelReconciliation !== undefined, + ); + expect(checkpoint?.reconciliationRequired).toBe(true); + expect(checkpoint?.modelReconciliation?.failedAttempt).toBe(0); + + const resumedModel = new ScriptedModelPort([]); + const resumedDurability = new RecordingDurabilityPort(); + const { engine: resumedEngine } = makeEngine({ + model: resumedModel, + durability: resumedDurability, + }); + const result = await resumedEngine.resumeAfterModelReconciliation( + new ResumeContext({ + checkpoint: checkpoint!, + maxIterations: 10, + maxModelAttempts: 5, + lastJournalSequence: blocked.commit.lastSequence, + }), + new ModelInvocationResponse({ output: "reconciled" }), + ); + + expect(result.commit.status).toBe("success"); + expect(result.commit.output).toBe("reconciled"); + expect(resumedModel.requests).toHaveLength(0); + expect(resumedDurability.events.map((event) => event.kind)).toContain( + "model_invocation_reconciled", + ); + }); + + it("does not repeat an indeterminate tool after host reconciliation", async () => { + const model = new ScriptedModelPort([ + response({ + tools: [{ id: "call-unknown", name: "write" }], + }), + response({ output: "continued" }), + ]); + const tools = new ScriptedToolPort(async () => { + throw TurnPortError.indeterminate("write may have completed"); + }); + const { engine, durability } = makeEngine({ model, tools }); + + const blocked = await engine.run(request()); + expect(blocked.commit.status).toBe("reconciliation_required"); + expect(tools.requests).toHaveLength(1); + const checkpoint = durability.checkpoints.find( + (candidate) => + candidate.completedToolResults?.at(-1)?.outcome === "indeterminate", + ); + expect(checkpoint?.pendingToolRequests).toHaveLength(0); + + const resumedDurability = new RecordingDurabilityPort(); + const { engine: resumedEngine } = makeEngine({ + model, + tools, + durability: resumedDurability, + }); + const result = await resumedEngine.resumeAfterToolReconciliation( + new ResumeContext({ + checkpoint: checkpoint!, + maxIterations: 10, + maxModelAttempts: 3, + lastJournalSequence: blocked.commit.lastSequence, + }), + new ModelToolResult({ + requestId: "call-unknown", + name: "write", + outcome: "success", + output: "confirmed", + }), + ); + + expect(result.commit.status).toBe("success"); + expect(result.commit.output).toBe("continued"); + expect(tools.requests).toHaveLength(1); + expect(resumedDurability.events.map((event) => event.kind)).toContain( + "tool_result_reconciled", + ); + }); +}); + +describe("turn-engine retry and cancellation boundaries", () => { + it("reuses the same immutable snapshot for equal model attempts", async () => { + const model = new ScriptedModelPort([ + new TurnPortError("transient"), + response({ output: "retried" }), + ]); + const { engine } = makeEngine({ model }); + + const result = await engine.run(request({ maxModelAttempts: 2 })); + + expect(result.commit.status).toBe("success"); + expect(model.requests).toHaveLength(2); + expect(model.requests[0].context).toBe(model.requests[1].context); + expect(Object.isFrozen(model.requests[0].context)).toBe(true); + expect(result.commit.iterations).toBe(1); + }); + + it("applies the same model-attempt budget to later tool-calling rounds", async () => { + const model = new ScriptedModelPort([ + response({ tools: [{ id: "call-a", name: "echo" }] }), + new TurnPortError("second-round transient"), + response({ output: "recovered" }), + ]); + const tools = new ScriptedToolPort( + async (toolRequest) => + new ModelToolResult({ + requestId: toolRequest.id, + name: toolRequest.name, + output: "ok", + }), + ); + const { engine } = makeEngine({ model, tools }); + + const result = await engine.run(request({ maxModelAttempts: 2 })); + + expect(result.commit.status).toBe("success"); + expect(result.commit.output).toBe("recovered"); + expect(result.commit.iterations).toBe(2); + expect(model.requests).toHaveLength(3); + expect(model.requests[1].context).toBe(model.requests[2].context); + }); + + it("persists a completed model response before honoring cancellation at commit", async () => { + const cancellation = new TurnCancellationToken(); + const model = new ScriptedModelPort([ + () => { + cancellation.cancel(); + return response({ output: "must not commit success" }); + }, + ]); + const { engine, durability } = makeEngine({ model }); + + const result = await engine.run(request(), cancellation); + + expect(result.commit.status).toBe("cancelled"); + expect(durability.events.map((event) => event.kind)).toContain( + "model_invocation_completed", + ); + expect(durability.events.map((event) => event.kind)).not.toContain( + "post_commit_started", + ); + }); +}); diff --git a/runtime/typescript/packages/core/tests/turn-engine-vectors.test.ts b/runtime/typescript/packages/core/tests/turn-engine-vectors.test.ts new file mode 100644 index 000000000..c57e7f0f7 --- /dev/null +++ b/runtime/typescript/packages/core/tests/turn-engine-vectors.test.ts @@ -0,0 +1,225 @@ +import { readFileSync } from "node:fs"; +import { join, resolve } from "node:path"; +import { describe, expect, it } from "vitest"; + +import { Message } from "../src/model/conversation/message.js"; +import { DelegatedStateReference } from "../src/model/pipeline/delegated-state-reference.js"; +import { ModelToolResult } from "../src/model/pipeline/model-tool-result.js"; +import { ContextPipeline } from "../src/core/turn-engine-context.js"; +import { TurnCancellationToken } from "../src/core/turn-engine-cancellation.js"; +import { TurnEngine, TurnEngineRequest } from "../src/core/turn-engine.js"; +import { + DeterministicClock, + DeterministicIds, + RecordingDurabilityPort, + ScriptedModelPort, + ScriptedToolPort, + SelectivePermissionPort, + response, +} from "./turn-engine-harness.js"; + +interface VectorMessage { + role: "system" | "user" | "assistant"; + content: string; +} + +interface VectorModelStep { + output?: unknown; + assistant?: string; + tools?: { id: string; name: string; arguments?: unknown }[]; + nextPortability?: "portable" | "delegated" | "opaque"; + delegatedState?: { + provider: string; + kind: string; + id: string; + }[]; +} + +interface TurnVector { + name: string; + cancelBeforeRun?: boolean; + messages: VectorMessage[]; + model: VectorModelStep[]; + toolOutputs?: Record; + denyTools?: string[]; + expected: { + status: "success" | "failed" | "cancelled" | "reconciliation_required"; + output?: unknown; + iterations: number; + snapshots: number; + snapshotStablePrefixes?: number[]; + snapshotPortability?: string[]; + toolResults: number; + toolResultOrder?: string[]; + commitPortability?: string; + delegatedState?: number; + eventKinds?: string[]; + }; +} + +const vectorPath = join( + resolve(import.meta.dirname, "../../../../../spec"), + "vectors", + "engine", + "turn_vectors.json", +); +const vectors = ( + JSON.parse(readFileSync(vectorPath, "utf8")) as { + cases: TurnVector[]; + } +).cases; + +describe("canonical turn-engine shared vectors", () => { + it.each(vectors)("$name", async (vector) => { + const model = new ScriptedModelPort( + vector.model.map((step) => + response({ + output: step.output, + assistant: step.assistant, + tools: step.tools, + portability: step.nextPortability, + delegatedState: step.delegatedState?.map( + (reference) => new DelegatedStateReference(reference), + ), + }), + ), + ); + const tool = new ScriptedToolPort(async (request) => { + return new ModelToolResult({ + requestId: request.id, + name: request.name, + outcome: "success", + output: vector.toolOutputs?.[request.id], + }); + }); + const durability = new RecordingDurabilityPort(); + const engine = new TurnEngine(new ContextPipeline(), { + model, + tools: tool, + permission: new SelectivePermissionPort( + new Set(vector.denyTools ?? []), + ), + durability, + clock: new DeterministicClock(), + ids: new DeterministicIds(), + }); + const cancellation = new TurnCancellationToken(); + if (vector.cancelBeforeRun) { + cancellation.cancel(); + } + + const result = await engine.run( + new TurnEngineRequest({ + sessionId: "session-vector", + turnId: vector.name, + messages: vector.messages.map(toMessage), + }), + cancellation, + ); + + expect(result.commit.status).toBe(vector.expected.status); + if ("output" in vector.expected) { + expect(result.commit.output).toEqual(vector.expected.output); + } + expect(result.commit.iterations).toBe(vector.expected.iterations); + expect(result.snapshots).toHaveLength(vector.expected.snapshots); + expect(result.toolResults).toHaveLength(vector.expected.toolResults); + if (vector.expected.snapshotStablePrefixes) { + expect( + result.snapshots?.map((snapshot) => snapshot.stablePrefixMessages), + ).toEqual(vector.expected.snapshotStablePrefixes); + } + if (vector.expected.snapshotPortability) { + expect( + result.snapshots?.map( + (snapshot) => snapshot.contextState.portability, + ), + ).toEqual(vector.expected.snapshotPortability); + } + if (vector.expected.toolResultOrder) { + expect(result.toolResults?.map((item) => item.requestId)).toEqual( + vector.expected.toolResultOrder, + ); + } + if (vector.expected.commitPortability) { + expect(result.commit.contextState.portability).toBe( + vector.expected.commitPortability, + ); + } + if (vector.expected.delegatedState !== undefined) { + expect(result.commit.contextState.delegatedState).toHaveLength( + vector.expected.delegatedState, + ); + } + if (vector.expected.eventKinds) { + expect(durability.events.map((event) => event.kind)).toEqual( + vector.expected.eventKinds, + ); + } + if ((vector.denyTools ?? []).length > 0) { + expect(durability.events.map((event) => event.kind)).toContain( + "tool_result_committed", + ); + expect(durability.events.map((event) => event.kind)).not.toContain( + "tool_execution_completed", + ); + } + expect(durability.events.map((event) => event.sequence)).toEqual( + durability.events.map((_, index) => index + 1), + ); + }); + + it("journals denied tools as committed results without execution events", async () => { + const durability = new RecordingDurabilityPort(); + const tools = new ScriptedToolPort(async () => { + throw new Error("denied tools must not execute"); + }); + const engine = new TurnEngine(new ContextPipeline(), { + model: new ScriptedModelPort([ + response({ + tools: [{ id: "call-denied", name: "protected", arguments: {} }], + }), + response({ output: "Permission was denied" }), + ]), + tools, + permission: new SelectivePermissionPort(new Set(["protected"])), + durability, + clock: new DeterministicClock(), + ids: new DeterministicIds(), + }); + + const result = await engine.run( + new TurnEngineRequest({ + sessionId: "session-denied", + turnId: "turn-denied", + messages: [Message.user("Read the protected resource")], + }), + ); + + expect(tools.requests).toHaveLength(0); + expect(result.toolResults).toEqual([ + expect.objectContaining({ + requestId: "call-denied", + outcome: "failed", + errorKind: "permission_denied", + }), + ]); + const eventKinds = durability.events.map((event) => event.kind); + expect( + eventKinds.filter((kind) => kind === "tool_result_committed"), + ).toHaveLength(1); + expect(eventKinds).not.toContain("tool_execution_started"); + expect(eventKinds).not.toContain("tool_execution_completed"); + }); +}); + +function toMessage(message: VectorMessage): Message { + switch (message.role) { + case "system": + return Message.system(message.content); + case "assistant": + return Message.assistant(message.content); + case "user": + return Message.user(message.content); + } +} diff --git a/runtime/typescript/packages/foundry/src/azure-models.ts b/runtime/typescript/packages/foundry/src/azure-models.ts index d48f12558..340b1ad4e 100644 --- a/runtime/typescript/packages/foundry/src/azure-models.ts +++ b/runtime/typescript/packages/foundry/src/azure-models.ts @@ -5,7 +5,15 @@ */ import { AzureOpenAI } from "openai"; -import { ModelInfo, ApiKeyConnection, FoundryConnection, ReferenceConnection, getConnection } from "@prompty/core"; +import { + ApiKeyConnection, + FoundryConnection, + ModelInfo, + ReferenceConnection, + createModelInfo, + enrichModelInfo, + getConnection, +} from "@prompty/core"; import type { Connection } from "@prompty/core"; interface FoundryDeployment { @@ -69,15 +77,7 @@ async function listAzureOpenAIModels(client: AzureOpenAI): Promise const models: ModelInfo[] = []; for (const m of page.data) { - const raw = m as unknown as Record; - models.push( - new ModelInfo({ - id: m.id, - ownedBy: m.owned_by, - // Azure may return maxContextLength in capabilities - contextWindow: typeof raw["maxContextLength"] === "number" ? raw["maxContextLength"] : undefined, - }), - ); + models.push(catalogModelToModelInfo(m as unknown as Record)); } return models; @@ -102,19 +102,64 @@ async function listFoundryDeployments( } const data = (await response.json()) as FoundryDeploymentsResponse; - return (data.value ?? []).map((deployment) => { - const capabilities = deployment.properties?.capabilities ?? deployment.properties?.model?.capabilities; - return new ModelInfo({ - id: deployment.name, - displayName: deployment.properties?.model?.name, - ownedBy: deployment.properties?.model?.publisher ?? "azure", - contextWindow: getNumber(capabilities, ["maxContextLength", "contextWindow", "context_length"]) - ?? deployment.properties?.model?.maxContextLength, - inputModalities: getStringArray(capabilities, ["inputModalities", "input_modalities", "supportedInputModalities"]), - outputModalities: getStringArray(capabilities, ["outputModalities", "output_modalities", "supportedOutputModalities"]), - additionalProperties: deployment as unknown as Record, - }); - }); + return (data.value ?? []).map((deployment) => + deploymentToModelInfo(deployment as unknown as Record)); +} + +/** Map one Foundry deployment response into the canonical generated model. */ +export function deploymentToModelInfo(raw: Record): ModelInfo { + const properties = asRecord(raw.properties); + const model = asRecord(properties?.model); + const capabilities = + asRecord(properties?.capabilities) ?? + asRecord(model?.capabilities) ?? + asRecord(raw.capabilities); + + return createModelInfo(enrichModelInfo("foundry", { + id: typeof raw.name === "string" ? raw.name : "", + displayName: + stringValue(raw.modelName) ?? + stringValue(model?.name), + ownedBy: + stringValue(raw.modelPublisher) ?? + stringValue(model?.publisher) ?? + "azure", + contextWindow: + getNumber(capabilities, ["maxContextLength", "contextWindow", "context_length"]) ?? + getNumber(model, ["maxContextLength"]) ?? + getNumber(raw, ["maxContextLength"]), + inputModalities: getStringArray(capabilities, [ + "inputModalities", + "input_modalities", + "supportedInputModalities", + ]), + outputModalities: getStringArray(capabilities, [ + "outputModalities", + "output_modalities", + "supportedOutputModalities", + ]), + additionalProperties: { ...raw }, + })); +} + +/** Map one Azure OpenAI catalog response into the canonical generated model. */ +export function catalogModelToModelInfo(raw: Record): ModelInfo { + return createModelInfo(enrichModelInfo("foundry", { + id: typeof raw.id === "string" ? raw.id : "", + ownedBy: typeof raw.owned_by === "string" ? raw.owned_by : undefined, + contextWindow: getNumber(raw, ["maxContextLength"]), + additionalProperties: { ...raw }, + })); +} + +function asRecord(value: unknown): Record | undefined { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? value as Record + : undefined; +} + +function stringValue(value: unknown): string | undefined { + return typeof value === "string" ? value : undefined; } function getNumber(source: Record | undefined, keys: string[]): number | undefined { diff --git a/runtime/typescript/packages/foundry/src/index.ts b/runtime/typescript/packages/foundry/src/index.ts index e79b14799..375fe9f8a 100644 --- a/runtime/typescript/packages/foundry/src/index.ts +++ b/runtime/typescript/packages/foundry/src/index.ts @@ -12,7 +12,11 @@ export { FoundryExecutor } from "./executor.js"; export { FoundryProcessor } from "./processor.js"; export { AzureExecutor } from "./azure-executor.js"; export { AzureProcessor } from "./azure-processor.js"; -export { listAzureModels } from "./azure-models.js"; +export { + catalogModelToModelInfo, + deploymentToModelInfo, + listAzureModels, +} from "./azure-models.js"; // Auto-register on import import { registerExecutor, registerProcessor } from "@prompty/core"; diff --git a/runtime/typescript/packages/foundry/src/processor.ts b/runtime/typescript/packages/foundry/src/processor.ts index e10efdf26..a975a8837 100644 --- a/runtime/typescript/packages/foundry/src/processor.ts +++ b/runtime/typescript/packages/foundry/src/processor.ts @@ -8,7 +8,8 @@ import type { Prompty } from "@prompty/core"; import type { Processor } from "@prompty/core"; -import { processResponse } from "@prompty/openai"; +import type { StreamChunk } from "@prompty/core"; +import { processResponse, processStream } from "@prompty/openai"; import { traceSpan } from "@prompty/core"; export class FoundryProcessor implements Processor { @@ -21,4 +22,8 @@ export class FoundryProcessor implements Processor { return result; }); } + + processStream(response: AsyncIterable): AsyncIterable { + return processStream(response); + } } diff --git a/runtime/typescript/packages/foundry/tests/discovery-vectors.test.ts b/runtime/typescript/packages/foundry/tests/discovery-vectors.test.ts new file mode 100644 index 000000000..b703d4a84 --- /dev/null +++ b/runtime/typescript/packages/foundry/tests/discovery-vectors.test.ts @@ -0,0 +1,36 @@ +import { readFileSync } from "node:fs"; +import { resolve } from "node:path"; + +import { describe, expect, it } from "vitest"; + +import { + catalogModelToModelInfo, + deploymentToModelInfo, +} from "../src/azure-models.js"; + +interface DiscoveryVector { + name: string; + provider: string; + shape: string; + input: Record; + expected: Record; +} + +const vectorFile = resolve( + import.meta.dirname, + "../../../../../spec/vectors/discovery/discovery_vectors.json", +); +const vectors = ( + JSON.parse(readFileSync(vectorFile, "utf8")) as { vectors: DiscoveryVector[] } +).vectors.filter((vector) => vector.provider === "foundry"); + +describe("Foundry discovery vectors", () => { + for (const vector of vectors) { + it(vector.name, () => { + const actual = vector.shape === "catalog" + ? catalogModelToModelInfo(vector.input) + : deploymentToModelInfo(vector.input); + expect(actual.save()).toEqual(vector.expected); + }); + } +}); diff --git a/runtime/typescript/packages/foundry/tests/models.test.ts b/runtime/typescript/packages/foundry/tests/models.test.ts index 9e4be95e7..46728a9e6 100644 --- a/runtime/typescript/packages/foundry/tests/models.test.ts +++ b/runtime/typescript/packages/foundry/tests/models.test.ts @@ -59,9 +59,8 @@ describe("listAzureModels", () => { it("does not set modalities (Azure API does not return them)", async () => { const models = await listAzureModels(connection); for (const m of models) { - // ModelInfo defaults modalities to [] when not explicitly set - expect(m.inputModalities).toEqual([]); - expect(m.outputModalities).toEqual([]); + expect(m.inputModalities).toBeUndefined(); + expect(m.outputModalities).toBeUndefined(); } }); diff --git a/runtime/typescript/packages/openai/src/index.ts b/runtime/typescript/packages/openai/src/index.ts index fcaf5d61b..2d6c2d4a9 100644 --- a/runtime/typescript/packages/openai/src/index.ts +++ b/runtime/typescript/packages/openai/src/index.ts @@ -7,9 +7,9 @@ */ export { OpenAIExecutor } from "./executor.js"; -export { OpenAIProcessor, processResponse } from "./processor.js"; +export { OpenAIProcessor, processResponse, processStream } from "./processor.js"; export { messageToWire, buildChatArgs, buildEmbeddingArgs, buildImageArgs, buildResponsesArgs } from "./wire.js"; -export { listModels } from "./models.js"; +export { listModels, modelInfoFromWire } from "./models.js"; // Auto-register on import import { registerExecutor, registerProcessor } from "@prompty/core"; diff --git a/runtime/typescript/packages/openai/src/models.ts b/runtime/typescript/packages/openai/src/models.ts index c42fb5eab..ca73e16bb 100644 --- a/runtime/typescript/packages/openai/src/models.ts +++ b/runtime/typescript/packages/openai/src/models.ts @@ -5,21 +5,16 @@ */ import OpenAI from "openai"; -import { ModelInfo, ApiKeyConnection, ReferenceConnection, getConnection } from "@prompty/core"; +import { + ApiKeyConnection, + ModelInfo, + ReferenceConnection, + createModelInfo, + enrichModelInfo, + getConnection, +} from "@prompty/core"; import type { Connection } from "@prompty/core"; -/** Known model metadata for enrichment (context windows and modalities). */ -const KNOWN_MODELS: Record = { - "gpt-4o": { contextWindow: 128_000, inputModalities: ["text", "image"], outputModalities: ["text"] }, - "gpt-4o-mini": { contextWindow: 128_000, inputModalities: ["text", "image"], outputModalities: ["text"] }, - "gpt-4-turbo": { contextWindow: 128_000, inputModalities: ["text", "image"], outputModalities: ["text"] }, - "gpt-4": { contextWindow: 8_192, inputModalities: ["text"], outputModalities: ["text"] }, - "gpt-3.5-turbo": { contextWindow: 16_385, inputModalities: ["text"], outputModalities: ["text"] }, - "text-embedding-3-small": { contextWindow: 8_191, inputModalities: ["text"], outputModalities: [] }, - "text-embedding-3-large": { contextWindow: 8_191, inputModalities: ["text"], outputModalities: [] }, - "dall-e-3": { inputModalities: ["text"], outputModalities: ["image"] }, -}; - /** * List models available from the OpenAI API. * @@ -32,29 +27,19 @@ export async function listModels(connection: Connection): Promise { const models: ModelInfo[] = []; for (const m of page.data) { - const known = findKnownModel(m.id); - models.push( - new ModelInfo({ - id: m.id, - ownedBy: m.owned_by, - contextWindow: known?.contextWindow, - inputModalities: known?.inputModalities, - outputModalities: known?.outputModalities, - }), - ); + models.push(modelInfoFromWire(m as unknown as Record)); } return models; } -/** Match a model id against known models, supporting prefix matching for dated variants. */ -function findKnownModel(id: string): (typeof KNOWN_MODELS)[string] | undefined { - if (KNOWN_MODELS[id]) return KNOWN_MODELS[id]; - // Try prefix match (e.g. "gpt-4o-2024-08-06" → "gpt-4o") - for (const key of Object.keys(KNOWN_MODELS)) { - if (id.startsWith(key + "-")) return KNOWN_MODELS[key]; - } - return undefined; +/** Map one raw OpenAI model response into the canonical generated model. */ +export function modelInfoFromWire(raw: Record): ModelInfo { + return createModelInfo(enrichModelInfo("openai", { + id: typeof raw.id === "string" ? raw.id : "", + ownedBy: typeof raw.owned_by === "string" ? raw.owned_by : undefined, + additionalProperties: { ...raw }, + })); } function buildClient(connection: Connection): OpenAI { diff --git a/runtime/typescript/packages/openai/src/processor.ts b/runtime/typescript/packages/openai/src/processor.ts index 532e7e4c9..136b168d3 100644 --- a/runtime/typescript/packages/openai/src/processor.ts +++ b/runtime/typescript/packages/openai/src/processor.ts @@ -9,7 +9,15 @@ import type { Prompty } from "@prompty/core"; import type { Processor } from "@prompty/core"; import type { ToolCall } from "@prompty/core"; -import { traceSpan } from "@prompty/core"; +import { + ErrorChunk, + InvocationUsage, + StreamChunk, + TextChunk, + ToolChunk, + UsageChunk, + traceSpan, +} from "@prompty/core"; import { createStructuredResult } from "@prompty/core"; export class OpenAIProcessor implements Processor { @@ -25,6 +33,10 @@ export class OpenAIProcessor implements Processor { return result; }); } + + processStream(response: AsyncIterable): AsyncIterable { + return processStream(response); + } } /** @@ -35,7 +47,7 @@ export function processResponse(agent: Prompty, response: unknown): unknown { // Streaming response — return content-extracting async generator if (isAsyncIterable(response)) { - return streamGenerator(response); + return legacyStreamGenerator(processStream(response)); } const r = response as Record; @@ -90,54 +102,157 @@ function isAsyncIterable(value: unknown): value is AsyncIterable { * * Matches the Python `_stream_generator` / `_async_stream_generator`. */ -async function* streamGenerator( +export async function* processStream( response: AsyncIterable, -): AsyncGenerator { +): AsyncGenerator { const toolCallAcc: Map = new Map(); + let usage: InvocationUsage | undefined; + + try { + for await (const chunk of response) { + const c = chunk as Record; + const error = c.error as Record | undefined; + if (error) { + yield new ErrorChunk({ + message: typeof error.message === "string" ? error.message : "OpenAI stream failed", + }); + return; + } - for await (const chunk of response) { - const c = chunk as Record; - const choices = c.choices as Record[] | undefined; - if (!choices || choices.length === 0) continue; + const eventType = c.type as string | undefined; + if (eventType === "response.output_text.delta" && typeof c.delta === "string" && c.delta) { + yield new TextChunk({ value: c.delta }); + } else if ( + (eventType === "response.output_item.added" || eventType === "response.output_item.done") && + isRecord(c.item) && + c.item.type === "function_call" + ) { + const idx = typeof c.output_index === "number" ? c.output_index : toolCallAcc.size; + toolCallAcc.set(idx, { + id: stringValue(c.item.call_id ?? c.item.id), + name: stringValue(c.item.name), + arguments: stringValue(c.item.arguments), + }); + } else if (eventType === "response.function_call_arguments.delta") { + appendResponsesArguments(toolCallAcc, c, false); + } else if (eventType === "response.function_call_arguments.done") { + appendResponsesArguments(toolCallAcc, c, true); + } else if (eventType === "response.completed" && isRecord(c.response)) { + usage = usageFromWire(c.response.usage); + const output = c.response.output; + if (Array.isArray(output)) { + for (const [idx, item] of output.entries()) { + if (isRecord(item) && item.type === "function_call") { + toolCallAcc.set(idx, { + id: stringValue(item.call_id ?? item.id), + name: stringValue(item.name), + arguments: stringValue(item.arguments), + }); + } + } + } + } else if (eventType === "response.refusal.delta" && typeof c.delta === "string" && c.delta) { + yield new ErrorChunk({ message: `Model refused: ${c.delta}` }); + return; + } - const delta = (choices[0] as Record).delta as Record | undefined; - if (!delta) continue; + usage = usageFromWire(c.usage) ?? usage; - // Content - if (delta.content != null) { - yield delta.content as string; - } + const choices = c.choices as Record[] | undefined; + if (choices && choices.length > 0) { + const delta = choices[0].delta as Record | undefined; + if (delta) { + if (typeof delta.content === "string" && delta.content) { + yield new TextChunk({ value: delta.content }); + } - // Tool call deltas — accumulate index-keyed partial chunks - const tcDeltas = delta.tool_calls as Record[] | undefined; - if (tcDeltas) { - for (const tcDelta of tcDeltas) { - const idx = tcDelta.index as number; - if (!toolCallAcc.has(idx)) { - toolCallAcc.set(idx, { id: "", name: "", arguments: "" }); - } - const acc = toolCallAcc.get(idx)!; - if (tcDelta.id) acc.id = tcDelta.id as string; - const fn = tcDelta.function as Record | undefined; - if (fn) { - if (fn.name) acc.name = fn.name as string; - if (fn.arguments) acc.arguments += fn.arguments as string; + const tcDeltas = delta.tool_calls as Record[] | undefined; + if (tcDeltas) { + for (const tcDelta of tcDeltas) { + const idx = typeof tcDelta.index === "number" ? tcDelta.index : 0; + const acc = toolCallAcc.get(idx) ?? { id: "", name: "", arguments: "" }; + if (tcDelta.id) acc.id = stringValue(tcDelta.id); + const fn = tcDelta.function as Record | undefined; + if (fn) { + if (fn.name) acc.name = stringValue(fn.name); + if (fn.arguments) acc.arguments += stringValue(fn.arguments); + } + toolCallAcc.set(idx, acc); + } + } + + if (typeof delta.refusal === "string" && delta.refusal) { + yield new ErrorChunk({ message: `Model refused: ${delta.refusal}` }); + return; + } } } } - - // Refusal - if (delta.refusal != null) { - throw new Error(`Model refused: ${delta.refusal}`); - } + } catch (error) { + yield new ErrorChunk({ + message: error instanceof Error ? error.message : String(error), + }); + return; } - // Yield accumulated tool calls at the end of the stream const sortedIndices = [...toolCallAcc.keys()].sort((a, b) => a - b); for (const idx of sortedIndices) { const tc = toolCallAcc.get(idx)!; - yield { id: tc.id, name: tc.name, arguments: tc.arguments } as ToolCall; + yield ToolChunk.load({ kind: "tool", toolCall: tc }); } + if (usage) { + yield new UsageChunk({ usage }); + } +} + +async function* legacyStreamGenerator( + chunks: AsyncIterable, +): AsyncGenerator { + for await (const chunk of chunks) { + if (chunk instanceof TextChunk) { + yield chunk.value; + } else if (chunk instanceof ToolChunk) { + yield chunk.toolCall; + } else if (chunk instanceof ErrorChunk) { + throw new Error(chunk.message); + } + } +} + +function appendResponsesArguments( + calls: Map, + chunk: Record, + replace: boolean, +): void { + const callId = stringValue(chunk.call_id); + const entry = [...calls.values()].find((call) => call.id === callId); + if (!entry) return; + const value = stringValue(replace ? chunk.arguments : chunk.delta); + entry.arguments = replace ? value : entry.arguments + value; +} + +function usageFromWire(value: unknown): InvocationUsage | undefined { + if (!isRecord(value)) return undefined; + const input = numberValue(value.input_tokens ?? value.prompt_tokens); + const output = numberValue(value.output_tokens ?? value.completion_tokens); + if (input === undefined && output === undefined) return undefined; + return new InvocationUsage({ + inputTokens: input ?? 0, + outputTokens: output ?? 0, + totalTokens: numberValue(value.total_tokens) ?? (input ?? 0) + (output ?? 0), + }); +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null; +} + +function stringValue(value: unknown): string { + return typeof value === "string" ? value : ""; +} + +function numberValue(value: unknown): number | undefined { + return typeof value === "number" && Number.isFinite(value) ? value : undefined; } // --------------------------------------------------------------------------- diff --git a/runtime/typescript/packages/openai/tests/discovery-vectors.test.ts b/runtime/typescript/packages/openai/tests/discovery-vectors.test.ts new file mode 100644 index 000000000..495d0a8ef --- /dev/null +++ b/runtime/typescript/packages/openai/tests/discovery-vectors.test.ts @@ -0,0 +1,29 @@ +import { readFileSync } from "node:fs"; +import { resolve } from "node:path"; + +import { describe, expect, it } from "vitest"; + +import { modelInfoFromWire } from "../src/models.js"; + +interface DiscoveryVector { + name: string; + provider: string; + input: Record; + expected: Record; +} + +const vectorFile = resolve( + import.meta.dirname, + "../../../../../spec/vectors/discovery/discovery_vectors.json", +); +const vectors = ( + JSON.parse(readFileSync(vectorFile, "utf8")) as { vectors: DiscoveryVector[] } +).vectors.filter((vector) => vector.provider === "openai"); + +describe("OpenAI discovery vectors", () => { + for (const vector of vectors) { + it(vector.name, () => { + expect(modelInfoFromWire(vector.input).save()).toEqual(vector.expected); + }); + } +}); diff --git a/runtime/typescript/packages/openai/tests/e2e.test.ts b/runtime/typescript/packages/openai/tests/e2e.test.ts index e6494f6f4..ebadc7029 100644 --- a/runtime/typescript/packages/openai/tests/e2e.test.ts +++ b/runtime/typescript/packages/openai/tests/e2e.test.ts @@ -17,9 +17,13 @@ import { turn, registerConnection, clearConnections, + ErrorChunk, + TextChunk, + ToolChunk, + UsageChunk, } from "@prompty/core"; import { OpenAIExecutor } from "../src/executor.js"; -import { OpenAIProcessor } from "../src/processor.js"; +import { OpenAIProcessor, processStream } from "../src/processor.js"; import { registerExecutor, registerProcessor } from "@prompty/core"; import * as fs from "fs"; import * as path from "path"; @@ -459,6 +463,62 @@ describe("E2E Pipeline", () => { // ========================================================================= describe("streaming", () => { + it("emits canonical text, tool, usage, and error chunks", async () => { + async function* stream() { + yield { choices: [{ delta: { content: "Hello" } }] }; + yield { + choices: [{ + delta: { + tool_calls: [{ + index: 0, + id: "call_1", + function: { name: "lookup", arguments: '{"q":' }, + }], + }, + }], + }; + yield { + choices: [{ + delta: { + tool_calls: [{ index: 0, function: { arguments: '"test"}' } }], + }, + }], + }; + yield { + choices: [{ delta: {} }], + usage: { prompt_tokens: 3, completion_tokens: 2, total_tokens: 5 }, + }; + } + + const chunks: unknown[] = []; + for await (const chunk of processStream(stream())) chunks.push(chunk); + + expect(chunks[0]).toBeInstanceOf(TextChunk); + expect((chunks[0] as TextChunk).value).toBe("Hello"); + expect(chunks[1]).toBeInstanceOf(ToolChunk); + expect((chunks[1] as ToolChunk).toolCall).toMatchObject({ + id: "call_1", + name: "lookup", + arguments: '{"q":"test"}', + }); + expect(chunks[2]).toBeInstanceOf(UsageChunk); + expect((chunks[2] as UsageChunk).usage).toMatchObject({ + inputTokens: 3, + outputTokens: 2, + totalTokens: 5, + }); + + async function* refusal() { + yield { choices: [{ delta: { refusal: "not allowed" } }] }; + yield { choices: [{ delta: { content: "must not be emitted" } }] }; + } + const refused: unknown[] = []; + for await (const chunk of processStream(refusal())) refused.push(chunk); + expect(refused).toHaveLength(1); + expect(refused[0]).toBeInstanceOf(ErrorChunk); + expect((refused[0] as ErrorChunk).message).toContain("not allowed"); + }); + it("returns an async generator that yields content chunks", async () => { // Mock streaming response: an async iterable of chunk objects const chunks = [ diff --git a/runtime/typescript/packages/openai/tests/models.test.ts b/runtime/typescript/packages/openai/tests/models.test.ts index 222edc281..508570432 100644 --- a/runtime/typescript/packages/openai/tests/models.test.ts +++ b/runtime/typescript/packages/openai/tests/models.test.ts @@ -73,9 +73,8 @@ describe("listModels (OpenAI)", () => { const custom = models.find((m) => m.id === "ft:gpt-4o:my-org:custom:abc123")!; expect(custom.ownedBy).toBe("user-org"); expect(custom.contextWindow).toBeUndefined(); - // ModelInfo defaults modalities to [] when not provided - expect(custom.inputModalities).toEqual([]); - expect(custom.outputModalities).toEqual([]); + expect(custom.inputModalities).toBeUndefined(); + expect(custom.outputModalities).toBeUndefined(); }); it("throws for unsupported connection kind", async () => {