diff --git a/skills/dynamic-workflows/SKILL.md b/skills/dynamic-workflows/SKILL.md index e7ab685d..ccb34570 100644 --- a/skills/dynamic-workflows/SKILL.md +++ b/skills/dynamic-workflows/SKILL.md @@ -77,6 +77,7 @@ const out = await agent('Return JSON findings', { }, }) // out is validated object; engine retries ≤2 on invalid JSON +// codex/claude: native structured output first, then prompt repair; others: prompt+Ajv ``` ### Providers diff --git a/src/local-agent-adapters.test.ts b/src/local-agent-adapters.test.ts index 19c7dd87..4438f6cc 100644 --- a/src/local-agent-adapters.test.ts +++ b/src/local-agent-adapters.test.ts @@ -2,7 +2,9 @@ import assert from "node:assert/strict"; import { delimiter } from "node:path"; import { claudeCommandEnvironment, + claudeOutputFormatOptions, createLocalAgentAdapter, + extractClaudeResultPayload, extractOpenCodeFinalResponse, extractPiFinalResponse, extractPiProviderError, @@ -388,3 +390,41 @@ assert.equal( assert.equal(env.PATH, [devspaceBin, "/home/user/.local/bin"].join(delimiter)); } + +{ + assert.deepEqual(claudeOutputFormatOptions(undefined), {}); + const schema = { + type: "object", + properties: { n: { type: "number" } }, + required: ["n"], + }; + assert.deepEqual(claudeOutputFormatOptions(schema), { + outputFormat: { type: "json_schema", schema }, + }); +} + +{ + assert.deepEqual( + extractClaudeResultPayload({ + type: "result", + result: '{"n":1}', + structured_output: { n: 1 }, + }), + { finalResponse: '{"n":1}', structured: { n: 1 } }, + ); + assert.deepEqual( + extractClaudeResultPayload({ + type: "result", + structured_output: { n: 2 }, + }), + { finalResponse: '{"n":2}', structured: { n: 2 } }, + ); + assert.deepEqual( + extractClaudeResultPayload({ + type: "result", + result: "plain text only", + }), + { finalResponse: "plain text only" }, + ); + assert.equal(extractClaudeResultPayload({ type: "assistant" }), undefined); +} diff --git a/src/local-agent-adapters.ts b/src/local-agent-adapters.ts index 68e8b663..fe56e81d 100644 --- a/src/local-agent-adapters.ts +++ b/src/local-agent-adapters.ts @@ -6,6 +6,8 @@ import type { LocalAgentProvider } from "./local-agent-profiles.js"; import { removeDevspaceNodeModulesBinFromPath } from "./local-agent-path.js"; import { createCodexSdkLocalAgentRuntime, + isNativeSchemaUnsupportedFailure, + ProviderSchemaUnsupportedError, type LocalAgentRunInput, type LocalAgentRunResult, } from "./local-agent-runtime.js"; @@ -59,42 +61,98 @@ class ClaudeLocalAgentAdapter implements LocalAgentAdapter { async run(input: LocalAgentRunInput): Promise { const { query } = await import("@anthropic-ai/claude-agent-sdk"); const claudeExecutable = process.env.CLAUDE_COMMAND ?? resolveExecutable("claude"); - const messages = query({ - prompt: input.prompt, - options: { - cwd: input.workspace, - model: input.model, - ...(input.effort ? { thinking: { type: "adaptive" } as const, effort: input.effort as EffortLevel } : {}), - resume: input.providerSessionId, - permissionMode: "bypassPermissions", - allowDangerouslySkipPermissions: true, - env: claudeCommandEnvironment(process.env), - ...(claudeExecutable ? { pathToClaudeCodeExecutable: claudeExecutable } : {}), - }, - }); + try { + const messages = query({ + prompt: input.prompt, + options: { + cwd: input.workspace, + model: input.model, + ...(input.effort + ? { thinking: { type: "adaptive" } as const, effort: input.effort as EffortLevel } + : {}), + resume: input.providerSessionId, + permissionMode: "bypassPermissions", + allowDangerouslySkipPermissions: true, + env: claudeCommandEnvironment(process.env), + ...(claudeExecutable ? { pathToClaudeCodeExecutable: claudeExecutable } : {}), + ...claudeOutputFormatOptions(input.schema), + }, + }); - let providerSessionId = input.providerSessionId ?? null; - let finalResponse = ""; - const items: unknown[] = []; - for await (const message of messages) { - items.push(message); - const record = message as Record; - if (typeof record.session_id === "string") providerSessionId = record.session_id; - if (record.type === "result" && typeof record.result === "string") { + let providerSessionId = input.providerSessionId ?? null; + let finalResponse = ""; + let structured: unknown | undefined; + const items: unknown[] = []; + for await (const message of messages) { + items.push(message); + const record = message as Record; + if (typeof record.session_id === "string") providerSessionId = record.session_id; + if (record.type !== "result") continue; const resultError = claudeResultError(record); if (resultError) throw new Error(resultError); - finalResponse = record.result; + const extracted = extractClaudeResultPayload(record); + if (extracted) { + finalResponse = extracted.finalResponse; + structured = extracted.structured; + } + } + + finalResponse = requireFinalResponse("Claude", finalResponse); + return { + provider: this.provider, + providerSessionId, + finalResponse, + items, + ...(structured !== undefined ? { structured } : {}), + }; + } catch (error) { + if (input.schema && isNativeSchemaUnsupportedFailure(error)) { + throw new ProviderSchemaUnsupportedError(this.provider, error); } + throw error; } + } +} + +/** Build Claude SDK outputFormat when a JSON Schema is requested. */ +export function claudeOutputFormatOptions( + schema: object | undefined, +): { outputFormat: { type: "json_schema"; schema: Record } } | Record { + if (!schema) return {}; + return { + outputFormat: { + type: "json_schema", + schema: schema as Record, + }, + }; +} - finalResponse = requireFinalResponse("Claude", finalResponse); +/** + * Prefer structured_output from a Claude result message; fall back to text result. + * When only structured is present, stringify it for journal finalResponse. + */ +export function extractClaudeResultPayload( + record: Record, +): { finalResponse: string; structured?: unknown } | undefined { + const hasStructured = Object.prototype.hasOwnProperty.call(record, "structured_output"); + const structured = hasStructured ? record.structured_output : undefined; + const text = typeof record.result === "string" ? record.result : undefined; + + if (hasStructured && structured !== undefined) { return { - provider: this.provider, - providerSessionId, - finalResponse, - items, + finalResponse: + text && text.trim() + ? text + : typeof structured === "string" + ? structured + : JSON.stringify(structured), + structured, }; } + if (text !== undefined) { + return { finalResponse: text }; + } + return undefined; } function claudeResultError(record: Record): string | undefined { @@ -516,6 +574,9 @@ async function promptOpencodeSession( sessionId: string, input: LocalAgentRunInput, ): Promise { + // OpenCode SessionPrompt accepts format: { type: 'json_schema', schema }, but + // per-model support is not programmatically discoverable — workflow agent({ schema }) + // keeps the prompt+Ajv path for opencode (no native structured output here). const session = (client as { session: { prompt(parameters?: unknown, options?: unknown): Promise; diff --git a/src/local-agent-runtime.test.ts b/src/local-agent-runtime.test.ts index 491623f6..29b3d69f 100644 --- a/src/local-agent-runtime.test.ts +++ b/src/local-agent-runtime.test.ts @@ -3,6 +3,7 @@ import type { RunResult, ThreadOptions } from "@openai/codex-sdk"; import { CodexSdkLocalAgentRuntime, createCodexSdkLocalAgentRuntime, + isNativeSchemaUnsupportedFailure, } from "./local-agent-runtime.js"; const emptyTurn = (finalResponse: string): RunResult => ({ @@ -13,11 +14,19 @@ const emptyTurn = (finalResponse: string): RunResult => ({ class FakeThread { prompts: string[] = []; + turnOptions: Array<{ outputSchema?: unknown; signal?: AbortSignal } | undefined> = []; constructor(readonly id: string | null) {} - async run(prompt: string): Promise { + async run( + prompt: string, + turnOptions?: { outputSchema?: unknown; signal?: AbortSignal }, + ): Promise { this.prompts.push(prompt); + this.turnOptions.push(turnOptions); + if (turnOptions?.outputSchema) { + return emptyTurn('{"ok":true}'); + } return emptyTurn(`response:${prompt}`); } } @@ -49,7 +58,9 @@ const readOnly = await runtime.run({ assert.equal(readOnly.provider, "codex"); assert.equal(readOnly.providerSessionId, "new-thread"); assert.equal(readOnly.finalResponse, "response:inspect only"); +assert.equal(readOnly.structured, undefined); assert.deepEqual(codex.startThreadInstance.prompts, ["inspect only"]); +assert.deepEqual(codex.startThreadInstance.turnOptions, [undefined]); assert.deepEqual(codex.started[0], { workingDirectory: "/tmp/project", sandboxMode: "read-only", @@ -74,6 +85,24 @@ assert.deepEqual(codex.started[1], { modelReasoningEffort: "high", }); +const schema = { + type: "object", + properties: { ok: { type: "boolean" } }, + required: ["ok"], +} as const; + +const structured = await runtime.run({ + prompt: "return structured", + workspace: "/tmp/project", + schema, +}); + +assert.equal(structured.finalResponse, '{"ok":true}'); +assert.deepEqual(structured.structured, { ok: true }); +assert.deepEqual(codex.startThreadInstance.turnOptions.at(-1), { + outputSchema: schema, +}); + const resumed = await runtime.run({ prompt: "continue", workspace: "/tmp/project", @@ -83,6 +112,7 @@ const resumed = await runtime.run({ assert.equal(resumed.providerSessionId, "resumed-thread"); assert.deepEqual(codex.resumeThreadInstance.prompts, ["continue"]); +assert.deepEqual(codex.resumeThreadInstance.turnOptions, [undefined]); assert.deepEqual(codex.resumed, [ { id: "existing-thread", @@ -98,3 +128,11 @@ assert.deepEqual(codex.resumed, [ const created = await createCodexSdkLocalAgentRuntime(undefined, () => new FakeCodex()); assert.equal(created.provider, "codex"); + +assert.equal( + isNativeSchemaUnsupportedFailure( + new Error("Invalid output schema: keyword is not supported"), + ), + true, +); +assert.equal(isNativeSchemaUnsupportedFailure(new Error("authentication failed")), false); diff --git a/src/local-agent-runtime.ts b/src/local-agent-runtime.ts index ebd99feb..5690e469 100644 --- a/src/local-agent-runtime.ts +++ b/src/local-agent-runtime.ts @@ -5,6 +5,7 @@ import type { RunResult, SandboxMode, ThreadOptions, + TurnOptions, } from "@openai/codex-sdk"; export type LocalAgentWriteMode = "read_only" | "allowed" | "full_access"; @@ -17,6 +18,8 @@ export interface LocalAgentRunInput { model?: string; /** Provider-native effort / reasoning level (was thinking). */ effort?: string; + /** JSON Schema for native structured output (codex/claude). */ + schema?: object; } export interface LocalAgentRunResult { @@ -24,6 +27,8 @@ export interface LocalAgentRunResult { providerSessionId: string | null; finalResponse: string; items: unknown[]; + /** Provider-native structured object when schema was requested. */ + structured?: unknown; } export interface LocalAgentRuntime { @@ -31,9 +36,42 @@ export interface LocalAgentRuntime { run(input: LocalAgentRunInput): Promise; } +export class ProviderSchemaUnsupportedError extends Error { + constructor( + readonly provider: string, + readonly cause: unknown, + ) { + super(`${provider} does not support the requested native output schema: ${errorMessage(cause)}`); + this.name = "ProviderSchemaUnsupportedError"; + } +} + +export function isProviderSchemaUnsupportedError( + error: unknown, +): error is ProviderSchemaUnsupportedError { + return error instanceof ProviderSchemaUnsupportedError; +} + +export function isNativeSchemaUnsupportedFailure(error: unknown): boolean { + const message = errorMessage(error).toLowerCase(); + const mentionsSchema = + /output[ _-]?schema/.test(message) || + /json[ _-]?schema/.test(message) || + /structured[ _-]?output/.test(message) || + /output[ _-]?format/.test(message); + const unsupported = + /not supported/.test(message) || + /unsupported/.test(message) || + /invalid (?:output|json )?schema/.test(message) || + /schema (?:is )?invalid/.test(message) || + /unknown (?:field|parameter|option)/.test(message) || + /not available/.test(message); + return mentionsSchema && unsupported; +} + interface CodexThreadLike { readonly id: string | null; - run(prompt: string): Promise; + run(prompt: string, turnOptions?: TurnOptions): Promise; } interface CodexClientLike { @@ -78,17 +116,35 @@ export class CodexSdkLocalAgentRuntime implements LocalAgentRuntime { const thread = input.providerSessionId ? this.codex.resumeThread(input.providerSessionId, options) : this.codex.startThread(options); - const turn = await thread.run(input.prompt); + const turnOptions = input.schema ? { outputSchema: input.schema } : undefined; + let turn: RunResult; + try { + turn = await thread.run(input.prompt, turnOptions); + } catch (error) { + if (input.schema && isNativeSchemaUnsupportedFailure(error)) { + throw new ProviderSchemaUnsupportedError(this.provider, error); + } + throw error; + } return { provider: this.provider, providerSessionId: thread.id, finalResponse: turn.finalResponse, items: turn.items, + ...(input.schema ? { structured: tryParseJson(turn.finalResponse) } : {}), }; } } +function tryParseJson(text: string): unknown | undefined { + try { + return JSON.parse(text) as unknown; + } catch { + return undefined; + } +} + export async function createCodexSdkLocalAgentRuntime( options?: CodexOptions, codexFactory?: CodexFactory, @@ -101,3 +157,7 @@ async function defaultCodexFactory(): Promise { const module = await import("@openai/codex-sdk"); return (options) => new module.Codex(options) as Codex; } + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} diff --git a/src/workflow-api.ts b/src/workflow-api.ts index ea093a0f..6e0eddaf 100644 --- a/src/workflow-api.ts +++ b/src/workflow-api.ts @@ -19,17 +19,22 @@ import { export interface WorkflowProviderRunInput { provider: string; prompt: string; + providerSessionId?: string; model?: string; effort?: string; workspace: string; signal?: AbortSignal; label?: string; phase?: string; + /** JSON Schema for native structured output (codex/claude). */ + schema?: object; } export interface WorkflowProviderRunResult { finalResponse: string; providerSessionId?: string; + /** Provider-native structured object when schema was requested. */ + structured?: unknown; } export type WorkflowRunProvider = ( @@ -346,14 +351,21 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi { const enforced = await enforceAgentSchema({ schema: agentOpts.schema, prompt, - run: (p) => deps.runProvider({ ...providerBase, prompt: p }), - onRetry: ({ attempt, errors }) => { + provider, + run: (p, options) => + deps.runProvider({ + ...providerBase, + prompt: p, + providerSessionId: options.providerSessionId, + ...(options.mode === "native" ? { schema: agentOpts.schema } : {}), + }), + onRetry: ({ attempt, errors, mode }) => { deps.journal.appendEvent({ runId: deps.runId, type: "schema_retry", phase, label: agentOpts.label, - data: { callIndex: index, attempt, errors }, + data: { callIndex: index, attempt, errors, mode }, }); }, }); @@ -362,6 +374,7 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi { result = { finalResponse: enforced.finalResponse, providerSessionId: enforced.providerSessionId, + structured: enforced.value, }; } else { result = await deps.runProvider(providerBase); diff --git a/src/workflow-cli.ts b/src/workflow-cli.ts index 3a7062db..e1c459c0 100644 --- a/src/workflow-cli.ts +++ b/src/workflow-cli.ts @@ -326,13 +326,16 @@ export async function runWorkflowWorker( const providerResult = await runLocalAgentProvider(input.provider, { prompt: input.prompt, workspace: input.workspace, + providerSessionId: input.providerSessionId, model: input.model, effort: input.effort, writeMode: "allowed", + schema: input.schema, }); return { finalResponse: providerResult.finalResponse, providerSessionId: providerResult.providerSessionId ?? undefined, + structured: providerResult.structured, }; }, resolveNestedSource: async (ref) => { diff --git a/src/workflow-engine.test.ts b/src/workflow-engine.test.ts index ad83aedb..030fb590 100644 --- a/src/workflow-engine.test.ts +++ b/src/workflow-engine.test.ts @@ -279,6 +279,59 @@ import { createStubBudget } from "./workflow-types.js"; await rm(dir, { recursive: true, force: true }); } +// --------------------------------------------------------------------------- +// schema retry: native schema only on first attempt + provider session reuse +// --------------------------------------------------------------------------- +{ + const dir = await mkdtemp(join(tmpdir(), "wf-schema-retry-")); + const store = new WorkflowStore(dir); + const run = store.createRun({ + name: "schema-retry", + source: "inline", + scriptPath: "inline", + scriptHash: "h", + workspaceRoot: dir, + }); + const calls: WorkflowProviderRunInput[] = []; + const api = createWorkflowApi({ + runId: run.id, + journal: store, + meta: { name: "schema-retry", description: "d" }, + args: undefined, + concurrency: 1, + signal: new AbortController().signal, + workspaceRoot: dir, + enabledProviders: ["codex"], + runProvider: async (input) => { + calls.push(input); + if (calls.length === 1) { + return { + finalResponse: '{"n":"bad"}', + structured: { n: "bad" }, + providerSessionId: "sess-1", + }; + } + return { finalResponse: '{"n":2}', providerSessionId: "sess-1" }; + }, + }); + + const out = await api.agent("give n", { + schema: { + type: "object", + properties: { n: { type: "number" } }, + required: ["n"], + }, + }); + assert.deepEqual(out, { n: 2 }); + assert.ok(calls[0]?.schema); + assert.equal(calls[0]?.providerSessionId, undefined); + assert.equal(calls[1]?.schema, undefined); + assert.equal(calls[1]?.providerSessionId, "sess-1"); + + store.close(); + await rm(dir, { recursive: true, force: true }); +} + // --------------------------------------------------------------------------- // executeWorkflow end-to-end with sandbox + nest depth // --------------------------------------------------------------------------- diff --git a/src/workflow-files.test.ts b/src/workflow-files.test.ts index 40fd4e6b..e15c89bf 100644 --- a/src/workflow-files.test.ts +++ b/src/workflow-files.test.ts @@ -32,7 +32,7 @@ import { hashSource } from "./workflow-script.js"; source: "export const meta = { name: 'x', description: 'd' }\nreturn 1\n", preferredName: "demo", }); - assert.match(path, /workflow-scripts\/wfr_test\/demo\.js$/); + assert.match(path.replaceAll("\\", "/"), /workflow-scripts\/wfr_test\/demo\.js$/); const file = await resolveWorkflowScriptFromPathOrName({ file: path, diff --git a/src/workflow-schema.test.ts b/src/workflow-schema.test.ts index 67ad58f4..2084f513 100644 --- a/src/workflow-schema.test.ts +++ b/src/workflow-schema.test.ts @@ -3,8 +3,10 @@ import { augmentPromptForSchema, enforceAgentSchema, formatAjvErrors, + NATIVE_SCHEMA_PROVIDERS, } from "./workflow-schema.js"; import { WorkflowEngineError } from "./workflow-api.js"; +import { ProviderSchemaUnsupportedError } from "./local-agent-runtime.js"; { const prompt = augmentPromptForSchema("find bugs", { @@ -21,6 +23,10 @@ assert.equal( "/n must be number", ); +assert.ok(NATIVE_SCHEMA_PROVIDERS.has("codex")); +assert.ok(NATIVE_SCHEMA_PROVIDERS.has("claude")); +assert.ok(!NATIVE_SCHEMA_PROVIDERS.has("opencode")); + { let attempts = 0; const result = await enforceAgentSchema({ @@ -40,6 +46,7 @@ assert.equal( assert.deepEqual(result.value, { n: 2 }); assert.equal(result.attempts, 2); assert.equal(result.providerSessionId, "sess"); + assert.equal(result.mode, "prompt"); } { @@ -56,4 +63,176 @@ assert.equal( ); } +// Native provider: structured on attempt 0 → single attempt, raw prompt. +{ + const seen: Array<{ prompt: string; mode?: string; providerSessionId?: string }> = []; + const result = await enforceAgentSchema({ + schema: { + type: "object", + properties: { n: { type: "number" } }, + required: ["n"], + additionalProperties: false, + }, + prompt: "give n", + provider: "codex", + run: async (prompt, opts) => { + seen.push({ prompt, mode: opts?.mode }); + return { finalResponse: "noise", structured: { n: 7 } }; + }, + }); + assert.deepEqual(result.value, { n: 7 }); + assert.equal(result.attempts, 1); + assert.equal(result.mode, "native"); + assert.equal(seen.length, 1); + assert.equal(seen[0]?.prompt, "give n"); + assert.equal(seen[0]?.mode, "native"); + assert.ok(!seen[0]?.prompt.includes("ONLY a JSON")); +} + +// Native fail then prompt repair. +{ + const seen: Array<{ + prompt: string; + mode?: string; + providerSessionId?: string; + }> = []; + const retries: Array<{ attempt: number; mode: string }> = []; + const result = await enforceAgentSchema({ + schema: { + type: "object", + properties: { n: { type: "number" } }, + required: ["n"], + additionalProperties: false, + }, + prompt: "give n", + provider: "claude", + onRetry: ({ attempt, mode }) => { + retries.push({ attempt, mode }); + }, + run: async (prompt, opts) => { + seen.push({ + prompt, + mode: opts.mode, + providerSessionId: opts.providerSessionId, + }); + if (opts.mode === "native") { + return { + finalResponse: '{"n":"bad"}', + structured: { n: "bad" }, + providerSessionId: "sess-native", + }; + } + return { finalResponse: '{"n":3}', structured: { n: 3 } }; + }, + }); + assert.deepEqual(result.value, { n: 3 }); + assert.equal(result.attempts, 2); + assert.equal(result.mode, "prompt"); + assert.equal(seen[0]?.mode, "native"); + assert.equal(seen[1]?.mode, "prompt"); + assert.equal(seen[1]?.providerSessionId, "sess-native"); + assert.ok(seen[1]?.prompt.includes("ONLY a JSON")); + assert.deepEqual(retries[0], { attempt: 1, mode: "native" }); +} + +// Native structured strings are parsed when the schema expects a non-string value. +{ + const result = await enforceAgentSchema({ + schema: { + type: "object", + properties: { n: { type: "number" } }, + required: ["n"], + }, + prompt: "give n", + provider: "claude", + run: async () => ({ + finalResponse: '{"n":4}', + structured: '{"n":4}', + }), + }); + assert.deepEqual(result.value, { n: 4 }); +} + +// A classified native-schema capability failure falls back to prompt mode. +{ + const modes: string[] = []; + const result = await enforceAgentSchema({ + schema: { + type: "object", + properties: { n: { type: "number" } }, + required: ["n"], + }, + prompt: "give n", + provider: "codex", + run: async (_prompt, opts) => { + modes.push(opts.mode); + if (opts.mode === "native") { + throw new ProviderSchemaUnsupportedError( + "codex", + new Error("output schema is not supported"), + ); + } + return { finalResponse: '{"n":5}' }; + }, + }); + assert.deepEqual(result.value, { n: 5 }); + assert.deepEqual(modes, ["native", "prompt"]); +} + +// Arbitrary provider failures are not disguised as schema fallback. +{ + let calls = 0; + await assert.rejects( + () => + enforceAgentSchema({ + schema: { type: "object" }, + prompt: "x", + provider: "codex", + run: async () => { + calls += 1; + throw new Error("authentication failed"); + }, + }), + /authentication failed/, + ); + assert.equal(calls, 1); +} + +// Do not report a retry when the retry budget is exhausted. +{ + const retries: number[] = []; + await assert.rejects(() => + enforceAgentSchema({ + schema: { type: "object" }, + prompt: "x", + maxRetries: 0, + onRetry: ({ attempt }) => retries.push(attempt), + run: async () => ({ finalResponse: "not json" }), + }), + ); + assert.deepEqual(retries, []); +} + +// Non-native never gets native mode. +{ + const modes: string[] = []; + const result = await enforceAgentSchema({ + schema: { + type: "object", + properties: { n: { type: "number" } }, + required: ["n"], + }, + prompt: "give n", + provider: "opencode", + run: async (prompt, opts) => { + modes.push(opts.mode); + assert.ok(prompt.includes("ONLY a JSON")); + return { finalResponse: '{"n":1}' }; + }, + }); + assert.deepEqual(result.value, { n: 1 }); + assert.deepEqual(modes, ["prompt"]); + assert.equal(result.mode, "prompt"); +} + console.log("workflow-schema.test.ts: ok"); diff --git a/src/workflow-schema.ts b/src/workflow-schema.ts index 476e97b6..1d93fe5f 100644 --- a/src/workflow-schema.ts +++ b/src/workflow-schema.ts @@ -2,9 +2,13 @@ import { createRequire } from "node:module"; import { WORKFLOW_MAX_SCHEMA_RETRIES } from "./workflow-types.js"; import { tryExtractJson, WorkflowEngineError } from "./workflow-api.js"; import type { WorkflowProviderRunResult, WorkflowRunProvider } from "./workflow-api.js"; +import { isProviderSchemaUnsupportedError } from "./local-agent-runtime.js"; const require = createRequire(import.meta.url); +/** Providers with a real structured-output API (hardcoded — no capability probe). */ +export const NATIVE_SCHEMA_PROVIDERS = new Set(["codex", "claude"]); + type AjvLike = new (opts?: object) => { compile: (schema: object) => ((data: unknown) => boolean) & { errors?: Array<{ instancePath?: string; message?: string }> | null; @@ -23,11 +27,28 @@ function loadAjv(): AjvLike { } } +export type SchemaEnforceMode = "native" | "prompt"; + export interface EnforceSchemaInput { schema: object; prompt: string; - run: (prompt: string) => Promise; - onRetry?: (info: { attempt: number; errors: string }) => void; + /** + * Provider id for native-vs-prompt policy. When in NATIVE_SCHEMA_PROVIDERS, + * attempt 0 uses raw prompt + native structured path; later attempts repair via prompt. + */ + provider?: string; + run: ( + prompt: string, + opts: { + mode: SchemaEnforceMode; + providerSessionId?: string; + }, + ) => Promise; + onRetry?: (info: { + attempt: number; + errors: string; + mode: SchemaEnforceMode; + }) => void; maxRetries?: number; } @@ -36,10 +57,12 @@ export interface EnforceSchemaResult { finalResponse: string; providerSessionId?: string; attempts: number; + mode: SchemaEnforceMode; } /** - * Augment prompt → run → extract JSON → Ajv validate → retry ≤2. + * Native-first for codex/claude; otherwise prompt+extract+Ajv. Always Ajv-validate. + * Retries ≤ WORKFLOW_MAX_SCHEMA_RETRIES after the first attempt. */ export async function enforceAgentSchema( input: EnforceSchemaInput, @@ -48,41 +71,61 @@ export async function enforceAgentSchema( const ajv = new Ajv({ allErrors: true, strict: false }); const validate = ajv.compile(input.schema); const maxRetries = input.maxRetries ?? WORKFLOW_MAX_SCHEMA_RETRIES; + const native = Boolean(input.provider && NATIVE_SCHEMA_PROVIDERS.has(input.provider)); const basePrompt = augmentPromptForSchema(input.prompt, input.schema); - let lastResponse = ""; - let lastSession: string | undefined; let lastErrors = "unknown validation error"; + let providerSessionId: string | undefined; for (let attempt = 0; attempt <= maxRetries; attempt += 1) { - const prompt = - attempt === 0 - ? basePrompt - : `${basePrompt}\n\nPrevious JSON failed validation:\n${lastErrors}\nReturn only corrected JSON.`; + const mode: SchemaEnforceMode = native && attempt === 0 ? "native" : "prompt"; - const result = await input.run(prompt); - lastResponse = result.finalResponse; - lastSession = result.providerSessionId ?? lastSession; + const prompt = + mode === "native" + ? input.prompt + : attempt === 0 + ? basePrompt + : `${basePrompt}\n\nPrevious JSON failed validation:\n${lastErrors}\nReturn only corrected JSON.`; + + let result: WorkflowProviderRunResult; + try { + result = await input.run(prompt, { mode, providerSessionId }); + } catch (error) { + if (mode === "native" && isProviderSchemaUnsupportedError(error) && attempt < maxRetries) { + lastErrors = error.message; + input.onRetry?.({ attempt: attempt + 1, errors: lastErrors, mode }); + continue; + } + throw error; + } + providerSessionId = result.providerSessionId ?? providerSessionId; - const extracted = tryExtractJson(result.finalResponse); - if (extracted === undefined) { + const candidates = structuredCandidates(result); + if (candidates.length === 0) { lastErrors = "Response was not valid JSON"; - input.onRetry?.({ attempt: attempt + 1, errors: lastErrors }); + if (attempt < maxRetries) { + input.onRetry?.({ attempt: attempt + 1, errors: lastErrors, mode }); + } continue; } - const ok = validate(extracted); - if (ok) { - return { - value: extracted, - finalResponse: result.finalResponse, - providerSessionId: result.providerSessionId, - attempts: attempt + 1, + for (const candidate of candidates) { + const ok = validate(candidate); + if (ok) { + return { + value: candidate, + finalResponse: result.finalResponse, + providerSessionId, + attempts: attempt + 1, + mode, + }; }; } lastErrors = formatAjvErrors(validate.errors); - input.onRetry?.({ attempt: attempt + 1, errors: lastErrors }); + if (attempt < maxRetries) { + input.onRetry?.({ attempt: attempt + 1, errors: lastErrors, mode }); + } } throw new WorkflowEngineError( @@ -122,11 +165,28 @@ export function schemaAwareRunProvider( return enforceAgentSchema({ schema, prompt: base.prompt, + provider: base.provider, onRetry, - run: (prompt) => + run: (prompt, options) => runProvider({ ...base, prompt, + providerSessionId: options.providerSessionId, + ...(options.mode === "native" ? { schema } : {}), }), }); } + +function structuredCandidates(result: WorkflowProviderRunResult): unknown[] { + const candidates: unknown[] = []; + if (result.structured !== undefined) { + candidates.push(result.structured); + if (typeof result.structured === "string") { + const parsed = tryExtractJson(result.structured); + if (parsed !== undefined && parsed !== result.structured) candidates.push(parsed); + } + } + const fromText = tryExtractJson(result.finalResponse); + if (fromText !== undefined) candidates.push(fromText); + return candidates; +}