From 3030d867268658622e27df1a29971fd99a6670e5 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Tue, 21 Jul 2026 22:55:44 +0530 Subject: [PATCH 1/5] feat(agents): pass schema as Codex turn outputSchema Wire LocalAgentRunInput.schema through CodexSdkLocalAgentRuntime to thread.run turn options, and surface parsed structured output. --- src/local-agent-runtime.test.ts | 31 ++++++++++++++++++++++++++++++- src/local-agent-runtime.ts | 23 +++++++++++++++++++++-- 2 files changed, 51 insertions(+), 3 deletions(-) diff --git a/src/local-agent-runtime.test.ts b/src/local-agent-runtime.test.ts index 491623f6..e6bd84be 100644 --- a/src/local-agent-runtime.test.ts +++ b/src/local-agent-runtime.test.ts @@ -13,11 +13,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 +57,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 +84,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 +111,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", diff --git a/src/local-agent-runtime.ts b/src/local-agent-runtime.ts index ebd99feb..39afe078 100644 --- a/src/local-agent-runtime.ts +++ b/src/local-agent-runtime.ts @@ -17,6 +17,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 +26,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 +35,14 @@ export interface LocalAgentRuntime { run(input: LocalAgentRunInput): Promise; } +interface CodexTurnOptions { + outputSchema?: unknown; + signal?: AbortSignal; +} + interface CodexThreadLike { readonly id: string | null; - run(prompt: string): Promise; + run(prompt: string, turnOptions?: CodexTurnOptions): Promise; } interface CodexClientLike { @@ -78,17 +87,27 @@ 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; + const turn = await thread.run(input.prompt, turnOptions); 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, From 6af99da21750ca9cf00a854d4110e9be97ea17e7 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Tue, 21 Jul 2026 22:56:43 +0530 Subject: [PATCH 2/5] feat(agents): Claude native outputFormat + structured_output Pass JSON Schema via outputFormat on query options and prefer structured_output from result messages. OpenCode stays prompt-path only. --- src/local-agent-adapters.test.ts | 40 ++++++++++++++++++++++ src/local-agent-adapters.ts | 58 +++++++++++++++++++++++++++++--- 2 files changed, 94 insertions(+), 4 deletions(-) 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..6f86e48e 100644 --- a/src/local-agent-adapters.ts +++ b/src/local-agent-adapters.ts @@ -70,20 +70,25 @@ class ClaudeLocalAgentAdapter implements LocalAgentAdapter { allowDangerouslySkipPermissions: true, env: claudeCommandEnvironment(process.env), ...(claudeExecutable ? { pathToClaudeCodeExecutable: claudeExecutable } : {}), + ...claudeOutputFormatOptions(input.schema), }, }); 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" && typeof record.result === "string") { - const resultError = claudeResultError(record); - if (resultError) throw new Error(resultError); - finalResponse = record.result; + if (record.type !== "result") continue; + const resultError = claudeResultError(record); + if (resultError) throw new Error(resultError); + const extracted = extractClaudeResultPayload(record); + if (extracted) { + finalResponse = extracted.finalResponse; + structured = extracted.structured; } } @@ -93,10 +98,52 @@ class ClaudeLocalAgentAdapter implements LocalAgentAdapter { providerSessionId, finalResponse, items, + ...(structured !== undefined ? { structured } : {}), }; } } +/** 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, + }, + }; +} + +/** + * 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 { + 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 { const subtype = typeof record.subtype === "string" ? record.subtype : undefined; const isError = record.is_error === true || subtype?.startsWith("error"); @@ -516,6 +563,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; From 7d797e3315d0a1d119502a13b4f541d4d366d0c7 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Tue, 21 Jul 2026 22:57:52 +0530 Subject: [PATCH 3/5] feat(workflow): native-first schema for codex/claude agent() Hardcode NATIVE_SCHEMA_PROVIDERS; attempt 0 uses adapter schema without prompt bloat, then prompt-repair retries with Ajv. Wire schema through runProvider / CLI worker. Document in skill. --- skills/dynamic-workflows/SKILL.md | 1 + src/workflow-api.ts | 18 +++++-- src/workflow-cli.ts | 2 + src/workflow-schema.test.ts | 87 +++++++++++++++++++++++++++++++ src/workflow-schema.ts | 55 ++++++++++++++----- 5 files changed, 146 insertions(+), 17 deletions(-) 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/workflow-api.ts b/src/workflow-api.ts index ea093a0f..c5a0aa00 100644 --- a/src/workflow-api.ts +++ b/src/workflow-api.ts @@ -25,11 +25,15 @@ export interface WorkflowProviderRunInput { 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 +350,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) => + deps.runProvider({ + ...providerBase, + prompt: p, + // Keep schema on adapter for codex/claude native+repair attempts. + 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 +373,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..e2707681 100644 --- a/src/workflow-cli.ts +++ b/src/workflow-cli.ts @@ -329,10 +329,12 @@ export async function runWorkflowWorker( 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-schema.test.ts b/src/workflow-schema.test.ts index 67ad58f4..cd068b5f 100644 --- a/src/workflow-schema.test.ts +++ b/src/workflow-schema.test.ts @@ -3,6 +3,7 @@ import { augmentPromptForSchema, enforceAgentSchema, formatAjvErrors, + NATIVE_SCHEMA_PROVIDERS, } from "./workflow-schema.js"; import { WorkflowEngineError } from "./workflow-api.js"; @@ -21,6 +22,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 +45,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 +62,85 @@ assert.equal( ); } +// Native provider: structured on attempt 0 → single attempt, raw prompt. +{ + const seen: Array<{ prompt: string; mode?: 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 }> = []; + 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 }); + if (opts?.mode === "native") { + return { finalResponse: '{"n":"bad"}', structured: { n: "bad" } }; + } + 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.ok(seen[1]?.prompt.includes("ONLY a JSON")); + assert.deepEqual(retries[0], { attempt: 1, mode: "native" }); +} + +// 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 ?? "missing"); + 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..b76e6f03 100644 --- a/src/workflow-schema.ts +++ b/src/workflow-schema.ts @@ -5,6 +5,9 @@ import type { WorkflowProviderRunResult, WorkflowRunProvider } from "./workflow- 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 +26,25 @@ 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 }, + ) => Promise; + onRetry?: (info: { + attempt: number; + errors: string; + mode: SchemaEnforceMode; + }) => void; maxRetries?: number; } @@ -36,10 +53,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,26 +67,31 @@ 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"; for (let attempt = 0; attempt <= maxRetries; attempt += 1) { + const mode: SchemaEnforceMode = native && attempt === 0 ? "native" : "prompt"; + const prompt = - attempt === 0 - ? basePrompt - : `${basePrompt}\n\nPrevious JSON failed validation:\n${lastErrors}\nReturn only corrected JSON.`; + mode === "native" + ? input.prompt + : attempt === 0 + ? basePrompt + : `${basePrompt}\n\nPrevious JSON failed validation:\n${lastErrors}\nReturn only corrected JSON.`; + + const result = await input.run(prompt, { mode }); - const result = await input.run(prompt); - lastResponse = result.finalResponse; - lastSession = result.providerSessionId ?? lastSession; + const extracted = + result.structured !== undefined + ? result.structured + : tryExtractJson(result.finalResponse); - const extracted = tryExtractJson(result.finalResponse); if (extracted === undefined) { lastErrors = "Response was not valid JSON"; - input.onRetry?.({ attempt: attempt + 1, errors: lastErrors }); + input.onRetry?.({ attempt: attempt + 1, errors: lastErrors, mode }); continue; } @@ -78,11 +102,12 @@ export async function enforceAgentSchema( finalResponse: result.finalResponse, providerSessionId: result.providerSessionId, attempts: attempt + 1, + mode, }; } lastErrors = formatAjvErrors(validate.errors); - input.onRetry?.({ attempt: attempt + 1, errors: lastErrors }); + input.onRetry?.({ attempt: attempt + 1, errors: lastErrors, mode }); } throw new WorkflowEngineError( @@ -122,11 +147,13 @@ export function schemaAwareRunProvider( return enforceAgentSchema({ schema, prompt: base.prompt, + provider: base.provider, onRetry, run: (prompt) => runProvider({ ...base, prompt, + schema, }), }); } From a8b8e5a7efab0e86c33de7e68af1e015bc25d510 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Wed, 22 Jul 2026 00:22:17 +0530 Subject: [PATCH 4/5] fix(workflow): make native schema fallback session-aware --- src/local-agent-adapters.ts | 87 ++++++++++++++------------ src/local-agent-runtime.test.ts | 9 +++ src/local-agent-runtime.ts | 51 ++++++++++++++-- src/workflow-api.ts | 7 ++- src/workflow-cli.ts | 1 + src/workflow-engine.test.ts | 53 ++++++++++++++++ src/workflow-schema.test.ts | 104 ++++++++++++++++++++++++++++++-- src/workflow-schema.ts | 73 ++++++++++++++++------ 8 files changed, 313 insertions(+), 72 deletions(-) diff --git a/src/local-agent-adapters.ts b/src/local-agent-adapters.ts index 6f86e48e..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,47 +61,56 @@ 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 } : {}), - ...claudeOutputFormatOptions(input.schema), - }, - }); + 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 = ""; - 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); - const extracted = extractClaudeResultPayload(record); - if (extracted) { - finalResponse = extracted.finalResponse; - structured = extracted.structured; + 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); + 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 } : {}), - }; + 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; + } } } diff --git a/src/local-agent-runtime.test.ts b/src/local-agent-runtime.test.ts index e6bd84be..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 => ({ @@ -127,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 39afe078..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"; @@ -35,14 +36,42 @@ export interface LocalAgentRuntime { run(input: LocalAgentRunInput): Promise; } -interface CodexTurnOptions { - outputSchema?: unknown; - signal?: AbortSignal; +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, turnOptions?: CodexTurnOptions): Promise; + run(prompt: string, turnOptions?: TurnOptions): Promise; } interface CodexClientLike { @@ -88,7 +117,15 @@ export class CodexSdkLocalAgentRuntime implements LocalAgentRuntime { ? this.codex.resumeThread(input.providerSessionId, options) : this.codex.startThread(options); const turnOptions = input.schema ? { outputSchema: input.schema } : undefined; - const turn = await thread.run(input.prompt, turnOptions); + 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, @@ -120,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 c5a0aa00..6e0eddaf 100644 --- a/src/workflow-api.ts +++ b/src/workflow-api.ts @@ -19,6 +19,7 @@ import { export interface WorkflowProviderRunInput { provider: string; prompt: string; + providerSessionId?: string; model?: string; effort?: string; workspace: string; @@ -351,12 +352,12 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi { schema: agentOpts.schema, prompt, provider, - run: (p) => + run: (p, options) => deps.runProvider({ ...providerBase, prompt: p, - // Keep schema on adapter for codex/claude native+repair attempts. - schema: agentOpts.schema, + providerSessionId: options.providerSessionId, + ...(options.mode === "native" ? { schema: agentOpts.schema } : {}), }), onRetry: ({ attempt, errors, mode }) => { deps.journal.appendEvent({ diff --git a/src/workflow-cli.ts b/src/workflow-cli.ts index e2707681..e1c459c0 100644 --- a/src/workflow-cli.ts +++ b/src/workflow-cli.ts @@ -326,6 +326,7 @@ 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", 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-schema.test.ts b/src/workflow-schema.test.ts index cd068b5f..2084f513 100644 --- a/src/workflow-schema.test.ts +++ b/src/workflow-schema.test.ts @@ -6,6 +6,7 @@ import { 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", { @@ -64,7 +65,7 @@ assert.ok(!NATIVE_SCHEMA_PROVIDERS.has("opencode")); // Native provider: structured on attempt 0 → single attempt, raw prompt. { - const seen: Array<{ prompt: string; mode?: string }> = []; + const seen: Array<{ prompt: string; mode?: string; providerSessionId?: string }> = []; const result = await enforceAgentSchema({ schema: { type: "object", @@ -90,7 +91,11 @@ assert.ok(!NATIVE_SCHEMA_PROVIDERS.has("opencode")); // Native fail then prompt repair. { - const seen: Array<{ prompt: string; mode?: string }> = []; + const seen: Array<{ + prompt: string; + mode?: string; + providerSessionId?: string; + }> = []; const retries: Array<{ attempt: number; mode: string }> = []; const result = await enforceAgentSchema({ schema: { @@ -105,9 +110,17 @@ assert.ok(!NATIVE_SCHEMA_PROVIDERS.has("opencode")); retries.push({ attempt, mode }); }, run: async (prompt, opts) => { - seen.push({ prompt, mode: opts?.mode }); - if (opts?.mode === "native") { - return { finalResponse: '{"n":"bad"}', structured: { n: "bad" } }; + 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 } }; }, @@ -117,10 +130,89 @@ assert.ok(!NATIVE_SCHEMA_PROVIDERS.has("opencode")); 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[] = []; @@ -133,7 +225,7 @@ assert.ok(!NATIVE_SCHEMA_PROVIDERS.has("opencode")); prompt: "give n", provider: "opencode", run: async (prompt, opts) => { - modes.push(opts?.mode ?? "missing"); + modes.push(opts.mode); assert.ok(prompt.includes("ONLY a JSON")); return { finalResponse: '{"n":1}' }; }, diff --git a/src/workflow-schema.ts b/src/workflow-schema.ts index b76e6f03..1d93fe5f 100644 --- a/src/workflow-schema.ts +++ b/src/workflow-schema.ts @@ -2,6 +2,7 @@ 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); @@ -38,7 +39,10 @@ export interface EnforceSchemaInput { provider?: string; run: ( prompt: string, - opts?: { mode: SchemaEnforceMode }, + opts: { + mode: SchemaEnforceMode; + providerSessionId?: string; + }, ) => Promise; onRetry?: (info: { attempt: number; @@ -71,6 +75,7 @@ export async function enforceAgentSchema( const basePrompt = augmentPromptForSchema(input.prompt, input.schema); let lastErrors = "unknown validation error"; + let providerSessionId: string | undefined; for (let attempt = 0; attempt <= maxRetries; attempt += 1) { const mode: SchemaEnforceMode = native && attempt === 0 ? "native" : "prompt"; @@ -82,32 +87,45 @@ export async function enforceAgentSchema( ? basePrompt : `${basePrompt}\n\nPrevious JSON failed validation:\n${lastErrors}\nReturn only corrected JSON.`; - const result = await input.run(prompt, { mode }); - - const extracted = - result.structured !== undefined - ? result.structured - : tryExtractJson(result.finalResponse); + 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; - if (extracted === undefined) { + const candidates = structuredCandidates(result); + if (candidates.length === 0) { lastErrors = "Response was not valid JSON"; - input.onRetry?.({ attempt: attempt + 1, errors: lastErrors, mode }); + 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, - mode, + 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, mode }); + if (attempt < maxRetries) { + input.onRetry?.({ attempt: attempt + 1, errors: lastErrors, mode }); + } } throw new WorkflowEngineError( @@ -149,11 +167,26 @@ export function schemaAwareRunProvider( prompt: base.prompt, provider: base.provider, onRetry, - run: (prompt) => + run: (prompt, options) => runProvider({ ...base, prompt, - schema, + 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; +} From 3343404fdccbdf4e979b8388f7a4dc6708e604db Mon Sep 17 00:00:00 2001 From: Waishnav Date: Wed, 22 Jul 2026 00:22:17 +0530 Subject: [PATCH 5/5] test(workflow): normalize persisted paths on windows --- src/workflow-files.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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,