fix(workflow): make native schema fallback session-aware

Author: Waishnav

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<LocalAgentRunResult> {
     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<string, unknown>;
-      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<string, unknown>;
+        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;
+    }
   }
 }
 

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);

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<LocalAgentRunResult>;
 }
 
-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<RunResult>;
+  run(prompt: string, turnOptions?: TurnOptions): Promise<RunResult>;
 }
 
 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<CodexFactory> {
   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);
+}

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({

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",

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
 // ---------------------------------------------------------------------------