Migrate a bounded text workflow between OpenAI and Claude

{
  "name": "openai-claude-text-router",
  "private": true,
  "type": "module",
  "engines": {
    "node": "22.22.2"
  },
  "scripts": {
    "check": "node --check src/provider-router.mjs",
    "smoke": "node test/provider-parity.mjs",
    "test:failures": "node test/provider-failures.mjs"
  },
  "dependencies": {
    "@anthropic-ai/sdk": "0.115.0",
    "openai": "7.1.0",
    "zod": "4.4.3"
  }
}

import Anthropic from "@anthropic-ai/sdk";
import OpenAI from "openai";
import { createHash } from "node:crypto";
import { mkdirSync } from "node:fs";
import path from "node:path";
import { DatabaseSync } from "node:sqlite";
import { z } from "zod";

export const CAPABILITY_MATRIX = Object.freeze({
  streaming: "unsupported",
  tools: "unsupported",
  structuredOutput: "unsupported",
  citations: "unsupported",
  multimodal: "unsupported",
  providerHostedState: "unsupported",
  textMessages: "supported",
  applicationIdempotency: "supported"
});

const Message = z.object({
  role: z.enum(["user", "assistant"]),
  content: z.string().min(1).max(32_000)
}).strict();

const Request = z.object({
  tenantId: z.string().regex(/^[a-z0-9_-]{1,80}$/),
  turnId: z.string().regex(/^[a-zA-Z0-9_-]{8,120}$/),
  system: z.string().max(16_000).default(""),
  messages: z.array(Message).min(1).max(100),
  maxOutputTokens: z.number().int().min(1).max(4096).default(1024)
}).strict().superRefine((request, context) => {
  if (request.messages[0]?.role !== "user") {
    context.addIssue({
      code: "custom",
      path: ["messages", 0, "role"],
      message: "conversation_must_start_with_user"
    });
  }
  if (request.messages.at(-1)?.role !== "user") {
    context.addIssue({
      code: "custom",
      path: ["messages", request.messages.length - 1, "role"],
      message: "conversation_must_end_with_user"
    });
  }
  for (let index = 1; index < request.messages.length; index += 1) {
    if (request.messages[index - 1].role === request.messages[index].role) {
      context.addIssue({
        code: "custom",
        path: ["messages", index, "role"],
        message: "conversation_roles_must_alternate"
      });
    }
  }
});

const Result = z.object({
  turnId: z.string(),
  provider: z.enum(["openai", "claude"]),
  providerRequestId: z.string(),
  state: z.literal("completed"),
  text: z.string().max(64_000),
  nativeState: z.string(),
  stopReason: z.string().nullable(),
  usage: z.record(z.string(), z.number().int().nonnegative().nullable())
}).strict();

class ProviderFailure extends Error {
  constructor(code, retryable = false, evidence = {}) {
    super(code);
    this.code = code;
    this.retryable = retryable;
    this.providerRequestId = evidence.providerRequestId ?? null;
    this.nativeState = evidence.nativeState ?? null;
  }
}

export class SqliteTurnStore {
  constructor(filename) {
    this.db = new DatabaseSync(filename);
    this.db.exec([
      "PRAGMA journal_mode=WAL;",
      "CREATE TABLE IF NOT EXISTS turns (",
      "tenant_id TEXT NOT NULL,",
      "turn_id TEXT NOT NULL,",
      "request_hash TEXT NOT NULL,",
      "provider TEXT NOT NULL,",
      "status TEXT NOT NULL CHECK(status IN ('in_progress','completed','failed')),",
      "result_json TEXT,",
      "error_code TEXT,",
      "provider_request_id TEXT,",
      "native_state TEXT,",
      "retryable INTEGER NOT NULL DEFAULT 0,",
      "updated_at TEXT NOT NULL,",
      "PRIMARY KEY (tenant_id, turn_id)",
      ");"
    ].join(" "));
    this.insert = this.db.prepare(
      "INSERT OR IGNORE INTO turns " +
      "(tenant_id,turn_id,request_hash,provider,status,updated_at) " +
      "VALUES (?,?,?,?, 'in_progress', ?)"
    );
    this.select = this.db.prepare(
      "SELECT * FROM turns WHERE tenant_id=? AND turn_id=?"
    );
    this.completeStatement = this.db.prepare(
      "UPDATE turns SET status='completed', result_json=?, error_code=NULL, retryable=0, updated_at=? " +
      "WHERE tenant_id=? AND turn_id=? AND status='in_progress'"
    );
    this.failStatement = this.db.prepare(
      "UPDATE turns SET status='failed', error_code=?, provider_request_id=?, native_state=?, " +
      "retryable=?, updated_at=? " +
      "WHERE tenant_id=? AND turn_id=? AND status='in_progress'"
    );
  }

  claim(request, provider, requestHash) {
    const now = new Date().toISOString();
    const inserted = this.insert.run(
      request.tenantId,
      request.turnId,
      requestHash,
      provider,
      now
    );
    const row = this.select.get(request.tenantId, request.turnId);
    if (row.request_hash !== requestHash) throw new Error("turn_id_payload_conflict");
    if (inserted.changes === 1) return { state: "claimed" };
    if (row.status === "completed") {
      return { state: "completed", result: JSON.parse(row.result_json) };
    }
    return {
      state: "blocked",
      status: row.status,
      errorCode: row.error_code,
      retryable: Boolean(row.retryable)
    };
  }

  complete(request, result) {
    const update = this.completeStatement.run(
      JSON.stringify(result),
      new Date().toISOString(),
      request.tenantId,
      request.turnId
    );
    if (update.changes !== 1) throw new Error("turn_completion_conflict");
  }

  fail(request, failure) {
    this.failStatement.run(
      failure.code,
      failure.providerRequestId,
      failure.nativeState,
      failure.retryable ? 1 : 0,
      new Date().toISOString(),
      request.tenantId,
      request.turnId
    );
  }

  close() {
    this.db.close();
  }
}

function hashRequest(request) {
  return createHash("sha256")
    .update(JSON.stringify({
      system: request.system,
      messages: request.messages,
      maxOutputTokens: request.maxOutputTokens
    }))
    .digest("hex");
}

function requireEnvironment(name, env) {
  const value = env[name];
  if (!value) throw new Error("missing_" + name.toLowerCase());
  return value;
}

export function policyFromEnvironment(env = process.env) {
  const policies = JSON.parse(requireEnvironment("TENANT_PROVIDER_POLICY_JSON", env));
  const openAiAllowed = new Set(
    requireEnvironment("OPENAI_ALLOWED_MODELS", env).split(",").map((item) => item.trim())
  );
  const claudeAllowed = new Set(
    requireEnvironment("CLAUDE_ALLOWED_MODELS", env).split(",").map((item) => item.trim())
  );
  return (tenantId) => {
    const policy = policies[tenantId];
    if (!policy || !["openai", "claude"].includes(policy.provider)) {
      throw new Error("tenant_provider_not_configured");
    }
    const allowed = policy.provider === "openai" ? openAiAllowed : claudeAllowed;
    if (!allowed.has(policy.model)) throw new Error("provider_model_not_allowed");
    return { provider: policy.provider, model: policy.model };
  };
}

function classifyFailure(error) {
  if (error instanceof ProviderFailure) {
    return {
      code: error.code,
      retryable: error.retryable,
      providerRequestId: error.providerRequestId,
      nativeState: error.nativeState
    };
  }
  if (error && error.name === "AbortError") {
    return {
      code: "provider_timeout",
      retryable: true,
      providerRequestId: error.request_id ?? null,
      nativeState: "timeout"
    };
  }
  const status = Number(error && error.status);
  if ([408, 409, 429].includes(status) || status >= 500) {
    return {
      code: "provider_retryable_" + status,
      retryable: true,
      providerRequestId: error.request_id ?? null,
      nativeState: String(status)
    };
  }
  if (status >= 400) {
    return {
      code: "provider_rejected_" + status,
      retryable: false,
      providerRequestId: error.request_id ?? null,
      nativeState: String(status)
    };
  }
  return {
    code: "provider_unknown_failure",
    retryable: false,
    providerRequestId: error?.request_id ?? null,
    nativeState: null
  };
}

function openAiText(response) {
  const parts = [];
  let refused = false;
  for (const item of response.output || []) {
    if (item.type !== "message") continue;
    for (const content of item.content || []) {
      if (content.type === "output_text") parts.push(content.text);
      if (content.type === "refusal") refused = true;
    }
  }
  if (refused) {
    throw new ProviderFailure("openai_refusal", false, {
      providerRequestId: response.id,
      nativeState: response.status
    });
  }
  if (response.status !== "completed") {
    throw new ProviderFailure("openai_" + response.status, false, {
      providerRequestId: response.id,
      nativeState: response.status
    });
  }
  const text = parts.join("");
  if (!text) {
    throw new ProviderFailure("openai_empty_text", false, {
      providerRequestId: response.id,
      nativeState: response.status
    });
  }
  return text;
}

export function createOpenAiAdapter(client) {
  return {
    async generate(request, policy, signal) {
      const response = await client.responses.create({
        model: policy.model,
        instructions: request.system || undefined,
        input: request.messages.map((message) => ({
          role: message.role,
          content: message.content
        })),
        max_output_tokens: request.maxOutputTokens,
        store: false
      }, { signal });
      return {
        turnId: request.turnId,
        provider: "openai",
        providerRequestId: response.id,
        state: "completed",
        text: openAiText(response),
        nativeState: response.status,
        stopReason: null,
        usage: {
          inputTokens: response.usage?.input_tokens ?? null,
          outputTokens: response.usage?.output_tokens ?? null,
          totalTokens: response.usage?.total_tokens ?? null,
          cachedInputTokens:
            response.usage?.input_tokens_details?.cached_tokens ?? null
        }
      };
    }
  };
}

export function createClaudeAdapter(client) {
  return {
    async generate(request, policy, signal) {
      const message = await client.messages.create({
        model: policy.model,
        system: request.system || undefined,
        messages: request.messages,
        max_tokens: request.maxOutputTokens
      }, { signal });
      if (message.stop_reason === "max_tokens") {
        throw new ProviderFailure("claude_max_tokens", false, {
          providerRequestId: message.id,
          nativeState: message.stop_reason
        });
      }
      if (!["end_turn", "stop_sequence"].includes(message.stop_reason)) {
        throw new ProviderFailure("claude_" + String(message.stop_reason), false, {
          providerRequestId: message.id,
          nativeState: message.stop_reason
        });
      }
      if (message.content.some((block) => block.type !== "text")) {
        throw new ProviderFailure("claude_unsupported_content_block", false, {
          providerRequestId: message.id,
          nativeState: message.stop_reason
        });
      }
      const text = message.content.map((block) => block.text).join("");
      if (!text) {
        throw new ProviderFailure("claude_empty_text", false, {
          providerRequestId: message.id,
          nativeState: message.stop_reason
        });
      }
      return {
        turnId: request.turnId,
        provider: "claude",
        providerRequestId: message.id,
        state: "completed",
        text,
        nativeState: "message_completed",
        stopReason: message.stop_reason,
        usage: {
          inputTokens: message.usage.input_tokens,
          outputTokens: message.usage.output_tokens,
          cacheCreationInputTokens: message.usage.cache_creation_input_tokens ?? null,
          cacheReadInputTokens: message.usage.cache_read_input_tokens ?? null
        }
      };
    }
  };
}

export function createRouter({ policyForTenant, adapters, store, timeoutMs = 30_000 }) {
  return {
    async generate(rawRequest) {
      const request = Request.parse(rawRequest);
      const policy = policyForTenant(request.tenantId);
      const adapter = adapters[policy.provider];
      if (!adapter) throw new Error("provider_adapter_missing");
      const requestHash = hashRequest(request);
      const claim = store.claim(request, policy.provider, requestHash);
      if (claim.state === "completed") return Result.parse(claim.result);
      if (claim.state === "blocked") {
        throw new Error(
          claim.status === "failed"
            ? "turn_failed_requires_reconciliation"
            : "turn_already_in_progress"
        );
      }

      try {
        const result = Result.parse(
          await adapter.generate(request, policy, AbortSignal.timeout(timeoutMs))
        );
        store.complete(request, result);
        return result;
      } catch (error) {
        const classified = classifyFailure(error);
        store.fail(request, classified);
        const safeError = new Error(classified.code);
        safeError.retryable = classified.retryable;
        throw safeError;
      }
    }
  };
}

export function createProductionRouter(env = process.env) {
  const openai = new OpenAI({ apiKey: requireEnvironment("OPENAI_API_KEY", env) });
  const claude = new Anthropic({ apiKey: requireEnvironment("ANTHROPIC_API_KEY", env) });
  const databasePath = path.resolve(env.TURN_DATABASE || "data/provider-turns.sqlite");
  mkdirSync(path.dirname(databasePath), { recursive: true, mode: 0o700 });
  const store = new SqliteTurnStore(databasePath);
  return {
    router: createRouter({
      policyForTenant: policyFromEnvironment(env),
      adapters: {
        openai: createOpenAiAdapter(openai),
        claude: createClaudeAdapter(claude)
      },
      store
    }),
    store
  };
}

import assert from "node:assert/strict";
import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import {
  createRouter,
  SqliteTurnStore
} from "../src/provider-router.mjs";

const temp = await fs.mkdtemp(path.join(os.tmpdir(), "provider-parity-"));
const store = new SqliteTurnStore(path.join(temp, "turns.sqlite"));
const calls = { openai: 0, claude: 0 };

const adapters = Object.fromEntries(
  ["openai", "claude"].map((provider) => [provider, {
    async generate(request) {
      calls[provider] += 1;
      return {
        turnId: request.turnId,
        provider,
        providerRequestId: provider + "_request_01",
        state: "completed",
        text: "PARITY_OK",
        nativeState: "completed",
        stopReason: provider === "claude" ? "end_turn" : null,
        usage: { inputTokens: 12, outputTokens: 3 }
      };
    }
  }])
);

const router = createRouter({
  policyForTenant: (tenantId) => ({
    provider: tenantId === "tenant-openai" ? "openai" : "claude",
    model: "fixture-model"
  }),
  adapters,
  store,
  timeoutMs: 1000
});

const base = {
  system: "Return exactly PARITY_OK.",
  messages: [{ role: "user", content: "health check" }],
  maxOutputTokens: 20
};

const first = await router.generate({
  ...base,
  tenantId: "tenant-openai",
  turnId: "turn-openai-001"
});
const duplicate = await router.generate({
  ...base,
  tenantId: "tenant-openai",
  turnId: "turn-openai-001"
});
const second = await router.generate({
  ...base,
  tenantId: "tenant-claude",
  turnId: "turn-claude-001"
});

assert.equal(first.text, "PARITY_OK");
assert.deepEqual(duplicate, first);
assert.equal(second.text, "PARITY_OK");
assert.equal(calls.openai, 1);
assert.equal(calls.claude, 1);

await assert.rejects(
  router.generate({
    ...base,
    messages: [{ role: "user", content: "different payload" }],
    tenantId: "tenant-openai",
    turnId: "turn-openai-001"
  }),
  /turn_id_payload_conflict/
);
await assert.rejects(
  router.generate({
    ...base,
    tenantId: "tenant-openai",
    turnId: "turn-unsupported-001",
    tools: [{ name: "write_file" }]
  })
);

store.close();
await fs.rm(temp, { recursive: true, force: true });
console.log(
  "SMOKE_OK scope=text-only providers=2 duplicate_provider_calls=0 unsupported=blocked"
);