From 6c8dd1cc90a3d080e0e03df9e023642972660242 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Elberte=20Pl=C3=ADnio?= Date: Sun, 19 Jul 2026 17:40:37 -0300 Subject: [PATCH] refactor(edge,router): claim router work durably before provider invocation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Issue #37 (CAND-4): operator-router's idempotency only became durable AFTER provider work. The handler checked `credit_ledger` for a completed route, invoked the OpenAI-compatible provider, and only then wrote a debit row keyed by the idempotency key — two concurrent requests for the same key could both find nothing completed, both call the provider, and the losing response was simply discarded after the winner's debit landed (wasted provider spend with no compensating record). New `router_attempts` table and three service-role-only RPCs (`router_attempt_claim` / `router_attempt_complete` / `router_attempt_fail`, supabase/migrations/20260722000000_router_attempt_claim.sql) move the durable decision in front of the provider call: at most one caller ever owns "go call the provider" for a given (user_id, idempotency_key) at a time. A claim that is neither completed nor explicitly failed is recoverable once its lease elapses. This is additive only: existing `credit_ledger` rows written before this migration remain authoritative for old completed routes (`findDebitedRouteResult`), independent of whether a `router_attempts` row exists for that key. `createOperatorRouterHandler` now claims before invoking the provider, completes the claim with the provider outcome, then debits — releasing the claim immediately on a definitive provider failure, and skipping straight to a debit retry (no re-invocation) when a claim is found already completed. - `packages/edge-shared/src/index.ts`: `claimRouteAttempt`, `completeRouteAttempt`, `failRouteAttempt`, `findDebitedRouteResult` own the RPC/route-metadata encode-decode boundary; `debitCredits` is unchanged. `createOperatorRouterHandler` takes `serviceSupabase` instead of separate `findCompletedRoute`/`debit` callbacks, and returns 409 `attempt_in_progress` while another claim is live. - `supabase/functions/operator-router/index.ts`: wires the caller's service-role client straight through; drops the adapter-owned `findCompletedRoute` query and `debit` callback. - `packages/edge-shared/test/edge-shared.test.ts`: replaces the `findCompletedRoute`/`debit` fakes with a `FakeRouterServiceSupabase` reimplementing the claim/complete/fail/debit RPC contract, covering the full claim -> provider -> complete -> debit choreography, live claims, lease recovery, and debit-retry-without-re-invoking-the- provider. - `supabase/tests/database/router_attempt_claim.test.sql`: pgTAP coverage for claim/in_progress/completed outcomes, idempotent completion replay, failure recovery, lease-based recovery, per-user isolation, privilege grants, and input validation. - `packages/edge-shared/test/router-attempt.contract.test.ts` (+ `router-fixtures.ts`, `postgres-router-client.ts`): a local-Postgres contract lane, following #47/#50's convention, proving the durable claim/complete/fail/debit functions against real Postgres, including two genuinely concurrent claims for the same key resolving to exactly one claim, and a full `createOperatorRouterHandler` run proving the provider is invoked and the ledger is debited exactly once across two concurrent same-key requests. Wired into `bun run test:supabase`. - Bumps `@pickforge/*` 0.11.0 -> 0.12.0 and re-pins every Deno function's `npm:` import, per the existing publish-then-pin convention. Validated: `bun run typecheck`; `bun run test` (366/366); `bun run build`; `supabase db reset` (all migrations apply cleanly); `bun run test:supabase` against real local Postgres — pgTAP (178 tests, including the new router_attempt_claim lane), welcome-credits concurrency, the CAND-1 checkout-lifecycle contract lane (20/20) and CAND-3 sync LWW contract lane (8/8) stay green, and the new router attempt contract lane (8/8) all pass; `supabase db advisors --local --type security --fail-on warn` clean. `deno check` against the real npm registry isn't possible pre-publish (same documented gap as #48: published `@pickforge/*` is still behind main) — confirmed the pre-existing gap is unchanged (this diff's remaining `deno check` errors are a strict subset of main's) and validated the adapter's typing directly against local `src` via `tsc` instead. --- package.json | 2 +- packages/auth/package.json | 2 +- packages/billing/package.json | 2 +- packages/brand/package.json | 2 +- packages/edge-shared/README.md | 4 +- packages/edge-shared/package.json | 2 +- packages/edge-shared/src/index.ts | 322 ++++++++++++- packages/edge-shared/test/edge-shared.test.ts | 454 +++++++++++++++--- .../test/postgres-router-client.ts | 181 +++++++ .../test/router-attempt.contract.test.ts | 288 +++++++++++ packages/edge-shared/test/router-fixtures.ts | 113 +++++ packages/flags/package.json | 2 +- packages/sync/package.json | 2 +- packages/tauri-release/package.json | 2 +- .../create-credit-checkout/deno.json | 4 +- supabase/functions/delete-account/deno.json | 2 +- .../functions/export-account-data/deno.json | 2 +- supabase/functions/operator-router/deno.json | 4 +- supabase/functions/operator-router/index.ts | 32 +- .../20260722000000_router_attempt_claim.sql | 234 +++++++++ .../database/router_attempt_claim.test.sql | 184 +++++++ 21 files changed, 1710 insertions(+), 130 deletions(-) create mode 100644 packages/edge-shared/test/postgres-router-client.ts create mode 100644 packages/edge-shared/test/router-attempt.contract.test.ts create mode 100644 packages/edge-shared/test/router-fixtures.ts create mode 100644 supabase/migrations/20260722000000_router_attempt_claim.sql create mode 100644 supabase/tests/database/router_attempt_claim.test.sql diff --git a/package.json b/package.json index 1168234..a51d6bc 100644 --- a/package.json +++ b/package.json @@ -10,7 +10,7 @@ "scripts": { "build": "bun run --cwd packages/tauri-release build && bun run --cwd packages/auth build && bun run --cwd packages/brand build && bun run --cwd packages/flags build && bun run --cwd packages/billing build && bun run --cwd packages/edge-shared build && bun run --cwd packages/sync build", "test": "vitest run", - "test:supabase": "supabase test db supabase/tests/database --local && bun run supabase/tests/welcome-credits-concurrency.ts && bun test packages/billing/test/checkout-lifecycle.contract.test.ts && bun test packages/sync/test/lww.contract.test.ts", + "test:supabase": "supabase test db supabase/tests/database --local && bun run supabase/tests/welcome-credits-concurrency.ts && bun test packages/billing/test/checkout-lifecycle.contract.test.ts && bun test packages/sync/test/lww.contract.test.ts && bun test packages/edge-shared/test/router-attempt.contract.test.ts", "test:coverage": "vitest run --coverage", "typecheck": "tsc -p tsconfig.json --noEmit", "check": "bun run typecheck && bun run test && bun run test:coverage && bun run build" diff --git a/packages/auth/package.json b/packages/auth/package.json index e5cf176..42dd465 100644 --- a/packages/auth/package.json +++ b/packages/auth/package.json @@ -1,6 +1,6 @@ { "name": "@pickforge/auth", - "version": "0.11.0", + "version": "0.12.0", "description": "UI-free Supabase Auth and entitlements wrapper for Pickforge apps.", "license": "MIT", "repository": { diff --git a/packages/billing/package.json b/packages/billing/package.json index cadb674..ee5297e 100644 --- a/packages/billing/package.json +++ b/packages/billing/package.json @@ -1,6 +1,6 @@ { "name": "@pickforge/billing", - "version": "0.11.0", + "version": "0.12.0", "description": "UI-free Stripe billing and credit-ledger helpers for Pickforge apps.", "license": "MIT", "repository": { diff --git a/packages/brand/package.json b/packages/brand/package.json index 105428c..4f66bd8 100644 --- a/packages/brand/package.json +++ b/packages/brand/package.json @@ -1,6 +1,6 @@ { "name": "@pickforge/brand", - "version": "0.11.0", + "version": "0.12.0", "description": "Pickforge CSS tokens, fonts, reset, and primitives.", "license": "MIT", "repository": { diff --git a/packages/edge-shared/README.md b/packages/edge-shared/README.md index bb31059..c77ef9f 100644 --- a/packages/edge-shared/README.md +++ b/packages/edge-shared/README.md @@ -4,9 +4,9 @@ Deno-compatible helpers for Pickforge Edge Functions. Clients are injected struc `assertRouteRequest` accepts only the hosted-router wire shape: `commandText` plus optional `context.projectName`, `context.chatNames`, and `context.widgetLabels`. Command text is user-authored and passed through unchanged; attached context is guarded against paths, host identifiers, tailnet IPs, and long serial-like tokens with `boundary_violation`. -`createOperatorRouterHandler` composes bearer-token authentication, request boundary validation, idempotent credit debiting, and an injected chat completion. A replayed idempotency key returns its stored proposal without another model call. +`createOperatorRouterHandler` composes bearer-token authentication, request boundary validation, a durable pre-provider claim, idempotent credit debiting, and an injected chat completion. Router work is durably claimed (`claimRouteAttempt`) BEFORE the provider is invoked, so at most one caller ever calls the provider and debits for a given idempotency key; `completeRouteAttempt`/`failRouteAttempt` record the outcome, and a stale (lease-expired) claim is recoverable by a later caller. A replayed, already-debited idempotency key returns its stored proposal (`findDebitedRouteResult`, honoring rows written before the durable claim table existed) without another model call. -Hosted routing is limited to 10 uncached attempts per user in each 10-second window. Its responses are `200` with a proposal, `402` for insufficient credits, `429` for `rate_limited`, boundary/auth `4xx`, and `5xx` for server or provider failures. +Hosted routing is limited to 10 uncached attempts per user in each 10-second window. Its responses are `200` with a proposal, `402` for insufficient credits, `409` for `attempt_in_progress` (another live claim owns this key right now; retry), `429` for `rate_limited`, boundary/auth `4xx`, and `5xx` for server or provider failures. ## Usage diff --git a/packages/edge-shared/package.json b/packages/edge-shared/package.json index bfed844..e9904c0 100644 --- a/packages/edge-shared/package.json +++ b/packages/edge-shared/package.json @@ -1,6 +1,6 @@ { "name": "@pickforge/edge-shared", - "version": "0.11.0", + "version": "0.12.0", "description": "Deno-compatible shared helpers for Pickforge Edge Functions.", "license": "MIT", "repository": { diff --git a/packages/edge-shared/src/index.ts b/packages/edge-shared/src/index.ts index 23ebcf5..ceb2e64 100644 --- a/packages/edge-shared/src/index.ts +++ b/packages/edge-shared/src/index.ts @@ -164,10 +164,9 @@ export interface ChatCompletionResult { export interface CreateOperatorRouterHandlerOptions { supabase: Pick; - findCompletedRoute(userId: string, idempotencyKey: string): Promise; + serviceSupabase: Pick; getCreditBalance(userId: string): Promise; consumeRateLimit(userId: string): Promise; - debit(options: Omit): Promise; chatComplete(options: { model: string; apiKey: string; @@ -180,6 +179,8 @@ export interface CreateOperatorRouterHandlerOptions { baseUrl: string; creditCostCents: number; systemPrompt?: string; + /** How long a claim may go unconfirmed before another caller may recover it. Defaults to 30 seconds. */ + attemptLeaseSeconds?: number; } export interface AccountAdminClientLike { @@ -307,6 +308,21 @@ export interface StoredRouteProposal { usage: ChatCompletionUsage; } +/** + * Outcome of durably claiming a router attempt (`router_attempt_claim`) + * BEFORE any provider invocation happens: + * - `claimed`: this caller now owns provider invocation for the key. + * - `in_progress`: another live claim owns it right now; the caller must not + * invoke the provider or debit, and should tell the client to retry. + * - `completed`: the provider already ran for this key (most likely a prior + * claim owner whose debit failed); the caller should retry only the debit + * step with `result` instead of re-invoking the provider. + */ +export type RouteAttemptClaim = + | { outcome: "claimed" } + | { outcome: "in_progress" } + | { outcome: "completed"; result: StoredRouteProposal }; + interface EntitlementRow { value?: EdgeSharedJson; expires_at: string | null; @@ -429,6 +445,168 @@ export async function debitCredits({ throw invalidRpcResult(); } +/** + * Reads the confirmed-paid result for a router idempotency key from the + * credit ledger. A ledger row existing here means the provider ran and the + * debit committed — this is the money-path source of truth for "has this + * key already been charged", independent of whether a `router_attempts` + * claim row still exists (rows written before that table existed only ever + * have a ledger row, and this keeps honoring them). + */ +export async function findDebitedRouteResult({ + supabase, + userId, + idempotencyKey, +}: { + supabase: Pick; + userId: string; + idempotencyKey: string; +}): Promise { + const validUserId = validateNonEmptyString(userId, "userId", "invalid_string"); + const validIdempotencyKey = validateNonEmptyString( + idempotencyKey, + "idempotencyKey", + "invalid_idempotency_key", + ); + const { data, error } = await supabase + .from<{ metadata: unknown }>("credit_ledger") + .select("metadata") + .eq("user_id", validUserId) + .eq("idempotency_key", validIdempotencyKey) + .maybeSingle(); + if (error !== null) { + throw databaseError("Failed to read stored router result", error); + } + if (data === null) { + return null; + } + + return decodeLegacyLedgerRoute(data.metadata); +} + +/** + * Durably claims a router attempt BEFORE any provider invocation, so at most + * one caller ever owns "go call the provider" for a given (userId, + * idempotencyKey) at a time. See + * supabase/migrations/20260722000000_router_attempt_claim.sql for the + * durable decision this wraps, including in-progress recovery via + * `leaseSeconds`. + */ +export async function claimRouteAttempt({ + supabase, + userId, + idempotencyKey, + leaseSeconds = 30, +}: { + supabase: Pick; + userId: string; + idempotencyKey: string; + leaseSeconds?: number; +}): Promise { + const validUserId = validateNonEmptyString(userId, "userId", "invalid_string"); + const validIdempotencyKey = validateNonEmptyString( + idempotencyKey, + "idempotencyKey", + "invalid_idempotency_key", + ); + const validLeaseSeconds = validatePositiveInteger(leaseSeconds, "leaseSeconds"); + + const { data, error } = await supabase.rpc>("router_attempt_claim", { + target_user: validUserId, + idem_key: validIdempotencyKey, + lease_seconds: validLeaseSeconds, + }); + if (error !== null) { + throw databaseError("Failed to claim router attempt", error); + } + if (!isRecord(data)) { + throw invalidRpcResult(); + } + if (data.outcome === "claimed") { + return { outcome: "claimed" }; + } + if (data.outcome === "in_progress") { + return { outcome: "in_progress" }; + } + if (data.outcome === "completed") { + return { outcome: "completed", result: decodeRouteAttemptResult(data) }; + } + + throw invalidRpcResult(); +} + +/** + * Records a claimed attempt's provider outcome, independent of whether the + * following debit succeeds — so a retry that lost the debit can skip + * straight to a debit retry with the same result instead of re-invoking the + * provider. Idempotent: completing an already-completed attempt returns the + * stored outcome instead of erroring. + */ +export async function completeRouteAttempt({ + supabase, + userId, + idempotencyKey, + result, +}: { + supabase: Pick; + userId: string; + idempotencyKey: string; + result: StoredRouteProposal; +}): Promise { + const validUserId = validateNonEmptyString(userId, "userId", "invalid_string"); + const validIdempotencyKey = validateNonEmptyString( + idempotencyKey, + "idempotencyKey", + "invalid_idempotency_key", + ); + const validProposalJson = validateNonEmptyString(result.proposalJson, "proposalJson", "invalid_string"); + + const { data, error } = await supabase.rpc>("router_attempt_complete", { + target_user: validUserId, + idem_key: validIdempotencyKey, + new_proposal_json: validProposalJson, + new_usage_input: result.usage.input, + new_usage_output: result.usage.output, + }); + if (error !== null) { + throw databaseError("Failed to record router attempt outcome", error); + } + if (!isRecord(data) || data.outcome !== "completed") { + throw invalidRpcResult(); + } + + return decodeRouteAttemptResult(data); +} + +/** + * Releases a claim immediately after a definitive provider failure, so a + * retry does not have to wait out the claim's lease to try again. + */ +export async function failRouteAttempt({ + supabase, + userId, + idempotencyKey, +}: { + supabase: Pick; + userId: string; + idempotencyKey: string; +}): Promise { + const validUserId = validateNonEmptyString(userId, "userId", "invalid_string"); + const validIdempotencyKey = validateNonEmptyString( + idempotencyKey, + "idempotencyKey", + "invalid_idempotency_key", + ); + + const { error } = await supabase.rpc("router_attempt_fail", { + target_user: validUserId, + idem_key: validIdempotencyKey, + }); + if (error !== null) { + throw databaseError("Failed to mark router attempt failed", error); + } +} + export function newIdempotencyKey(scope: string, uniquePart: string): string { return `${validateNonEmptyString(scope, "scope", "invalid_idempotency_key")}:${validateNonEmptyString( uniquePart, @@ -475,16 +653,16 @@ export function assertRouteRequest(body: unknown): RouteRequest { export function createOperatorRouterHandler({ supabase, - findCompletedRoute, + serviceSupabase, getCreditBalance, consumeRateLimit, - debit, chatComplete, model, apiKey, baseUrl, creditCostCents, systemPrompt = operatorRouterSystemPrompt, + attemptLeaseSeconds = 30, }: CreateOperatorRouterHandlerOptions): (req: Request) => Promise { const validModel = validateNonEmptyString(model, "model", "invalid_string"); const validApiKey = validateNonEmptyString(apiKey, "apiKey", "invalid_string"); @@ -501,9 +679,14 @@ export function createOperatorRouterHandler({ "invalid_idempotency_key", ); const scopedIdempotencyKey = newIdempotencyKey("router", `${userId}:${idempotencyKey}`); - const completedRoute = await findCompletedRoute(userId, scopedIdempotencyKey); - if (completedRoute !== null) { - return routeResponse(completedRoute, validCreditCostCents); + + const debitedRoute = await findDebitedRouteResult({ + supabase: serviceSupabase, + userId, + idempotencyKey: scopedIdempotencyKey, + }); + if (debitedRoute !== null) { + return routeResponse(debitedRoute, validCreditCostCents); } if (!(await consumeRateLimit(userId))) { return jsonResponse(429, { error: "rate_limited" }); @@ -516,36 +699,87 @@ export function createOperatorRouterHandler({ throw new EdgeSharedError("insufficient_credits", "Not enough credits", { balance }); } - const completion = await chatComplete({ - model: validModel, - apiKey: validApiKey, - baseUrl: validBaseUrl, - systemPrompt, - userPrompt: JSON.stringify(routeRequest), + // Durably claim BEFORE any provider invocation: at most one caller ever + // owns "go call the provider" for this key at a time. + const claim = await claimRouteAttempt({ + supabase: serviceSupabase, + userId, + idempotencyKey: scopedIdempotencyKey, + leaseSeconds: attemptLeaseSeconds, }); - const route = { - proposalJson: completion.text, - usage: { input: completion.usage.input, output: completion.usage.output }, - }; + if (claim.outcome === "in_progress") { + return jsonResponse(409, { error: "attempt_in_progress" }); + } + + let route: StoredRouteProposal; + if (claim.outcome === "completed") { + // A prior claim owner already ran the provider for this key (most + // likely this caller's own retry after a debit failure): skip + // straight to a debit retry with the stored result. + route = claim.result; + } else { + let completion: ChatCompletionResult; + try { + completion = await chatComplete({ + model: validModel, + apiKey: validApiKey, + baseUrl: validBaseUrl, + systemPrompt, + userPrompt: JSON.stringify(routeRequest), + }); + } catch (error) { + // A definitive provider failure releases the claim immediately so a + // retry does not have to wait out the lease. + await failRouteAttempt({ + supabase: serviceSupabase, + userId, + idempotencyKey: scopedIdempotencyKey, + }).catch(() => { + // The original provider error remains the actionable failure; the + // lease will still make the claim recoverable if this also fails. + }); + throw error; + } + + route = await completeRouteAttempt({ + supabase: serviceSupabase, + userId, + idempotencyKey: scopedIdempotencyKey, + result: { + proposalJson: completion.text, + usage: { input: completion.usage.input, output: completion.usage.output }, + }, + }); + } + let debitResult: DebitCreditsResult; try { - debitResult = await debit({ + debitResult = await debitCredits({ + supabase: serviceSupabase, userId, amountCents: validCreditCostCents, reason: "Operator routing", idempotencyKey: scopedIdempotencyKey, - metadata: route, + metadata: { proposalJson: route.proposalJson, usage: { input: route.usage.input, output: route.usage.output } }, }); } catch (error) { - const storedRoute = await findCompletedRoute(userId, scopedIdempotencyKey); - if (storedRoute !== null) { - return routeResponse(storedRoute, validCreditCostCents); + const recovered = await findDebitedRouteResult({ + supabase: serviceSupabase, + userId, + idempotencyKey: scopedIdempotencyKey, + }); + if (recovered !== null) { + return routeResponse(recovered, validCreditCostCents); } throw error; } if (debitResult.duplicate) { - const storedRoute = await findCompletedRoute(userId, scopedIdempotencyKey); + const storedRoute = await findDebitedRouteResult({ + supabase: serviceSupabase, + userId, + idempotencyKey: scopedIdempotencyKey, + }); if (storedRoute === null) { throw new EdgeSharedError("invalid_rpc_result", "Stored router result is missing"); } @@ -1024,6 +1258,40 @@ function routeResponse(route: StoredRouteProposal | ChatCompletionResult, costCe return jsonResponse(200, { proposalJson, usage: route.usage, costCents }); } +function decodeRouteAttemptResult(value: Record): StoredRouteProposal { + const proposalJson = value.proposal_json; + const usageInput = value.usage_input; + const usageOutput = value.usage_output; + if ( + typeof proposalJson !== "string" || + !Number.isSafeInteger(usageInput) || + !Number.isSafeInteger(usageOutput) + ) { + throw invalidRpcResult(); + } + + return { proposalJson, usage: { input: usageInput as number, output: usageOutput as number } }; +} + +function decodeLegacyLedgerRoute(metadata: unknown): StoredRouteProposal | null { + if (!isRecord(metadata)) { + return null; + } + + const proposalJson = metadata.proposalJson; + const usage = isRecord(metadata.usage) ? metadata.usage : undefined; + if ( + usage === undefined || + typeof proposalJson !== "string" || + !Number.isSafeInteger(usage.input) || + !Number.isSafeInteger(usage.output) + ) { + return null; + } + + return { proposalJson, usage: { input: usage.input as number, output: usage.output as number } }; +} + async function readRouteRequest(req: Request): Promise { try { return assertRouteRequest(await req.json()); @@ -1052,9 +1320,13 @@ function routerErrorStatus(code: EdgeSharedErrorCode): number { return code === "database_error" ? 500 : 400; } -function validatePositiveInteger(value: unknown, field: string): number { +function validatePositiveInteger( + value: unknown, + field: string, + code: EdgeSharedErrorCode = "invalid_debit_amount", +): number { if (typeof value !== "number" || !Number.isSafeInteger(value) || value <= 0) { - throw new EdgeSharedError("invalid_debit_amount", `${field} must be a positive integer`); + throw new EdgeSharedError(code, `${field} must be a positive integer`); } return value; diff --git a/packages/edge-shared/test/edge-shared.test.ts b/packages/edge-shared/test/edge-shared.test.ts index 74dfb88..02a14c7 100644 --- a/packages/edge-shared/test/edge-shared.test.ts +++ b/packages/edge-shared/test/edge-shared.test.ts @@ -1216,19 +1216,18 @@ describe("@pickforge/edge-shared", () => { }); }); - it("routes a valid command then debits with a namespaced key", async () => { + it("routes a valid command then claims, completes, and debits with a namespaced key", async () => { const supabase = new MemorySupabase(); - const debit = vi.fn(async () => ({ duplicate: false as const, balance: 98 })); + const serviceSupabase = new FakeRouterServiceSupabase({ [USER_ID]: 100 }); const chatComplete = vi.fn(async () => ({ text: "{\"action\":{\"action\":\"openProject\"},\"confidence\":0.9}", usage: { input: 42, output: 13 }, })); const handler = createOperatorRouterHandler({ supabase, - findCompletedRoute: vi.fn(async () => null), + serviceSupabase, consumeRateLimit: vi.fn(async () => true), getCreditBalance: vi.fn(async () => 100), - debit, chatComplete, model: "gpt-5.4-mini", apiKey: "key", @@ -1250,28 +1249,34 @@ describe("@pickforge/edge-shared", () => { usage: { input: 42, output: 13 }, costCents: 2, }); - expect(debit).toHaveBeenCalledWith({ - userId: USER_ID, - amountCents: 2, - reason: "Operator routing", - idempotencyKey: `router:${USER_ID}:router:attempt-1`, - metadata: { - proposalJson: "{\"action\":{\"action\":\"openProject\"},\"confidence\":0.9}", - usage: { input: 42, output: 13 }, + expect(chatComplete).toHaveBeenCalledOnce(); + const scopedKey = `router:${USER_ID}:router:attempt-1`; + expect(serviceSupabase.rpcCalls.map((call) => call.fn)).toEqual([ + "router_attempt_claim", + "router_attempt_complete", + "debit_credits", + ]); + expect(serviceSupabase.rpcCalls[2]).toMatchObject({ + args: { + target_user: USER_ID, + idem_key: scopedKey, + reason: "Operator routing", + usage_metadata: { + proposalJson: "{\"action\":{\"action\":\"openProject\"},\"confidence\":0.9}", + usage: { input: 42, output: 13 }, + }, }, }); - expect(chatComplete).toHaveBeenCalledOnce(); }); - it("returns insufficient credits without calling the model", async () => { - const supabase = new MemorySupabase(); + it("returns insufficient credits without claiming or calling the model", async () => { + const serviceSupabase = new FakeRouterServiceSupabase(); const chatComplete = vi.fn(); const handler = createOperatorRouterHandler({ - supabase, - findCompletedRoute: vi.fn(async () => null), + supabase: new MemorySupabase(), + serviceSupabase, consumeRateLimit: vi.fn(async () => true), getCreditBalance: vi.fn(async () => 1), - debit: vi.fn(), chatComplete, model: "gpt-5.4-mini", apiKey: "key", @@ -1283,18 +1288,21 @@ describe("@pickforge/edge-shared", () => { expect(response.status).toBe(402); await expect(response.json()).resolves.toEqual({ error: "insufficient_credits", balance: 1 }); expect(chatComplete).not.toHaveBeenCalled(); + expect(serviceSupabase.rpcCalls).toHaveLength(0); }); - it("returns a stored proposal without calling the model for a replay", async () => { - const supabase = new MemorySupabase(); - const debit = vi.fn(async () => ({ duplicate: false as const, balance: 98 })); + it("returns a stored proposal without calling the model for a replay of an already-debited (legacy) route", async () => { + const serviceSupabase = new FakeRouterServiceSupabase({ [USER_ID]: 100 }); + serviceSupabase.seedLegacyLedgerRoute(USER_ID, `router:${USER_ID}:router:attempt-1`, { + proposalJson: "{\"stored\":true}", + usage: { input: 3, output: 2 }, + }); const chatComplete = vi.fn(async () => ({ text: "{}", usage: { input: 1, output: 1 } })); const handler = createOperatorRouterHandler({ - supabase, - findCompletedRoute: vi.fn(async () => ({ proposalJson: "{\"stored\":true}", usage: { input: 3, output: 2 } })), + supabase: new MemorySupabase(), + serviceSupabase, consumeRateLimit: vi.fn(async () => true), getCreditBalance: vi.fn(async () => 100), - debit, chatComplete, model: "gpt-5.4-mini", apiKey: "key", @@ -1310,19 +1318,17 @@ describe("@pickforge/edge-shared", () => { costCents: 2, }); expect(chatComplete).not.toHaveBeenCalled(); - expect(debit).not.toHaveBeenCalled(); + expect(serviceSupabase.rpcCalls).toHaveLength(0); }); - it("rejects forbidden attached context before debiting or calling the model", async () => { - const supabase = new MemorySupabase(); - const debit = vi.fn(); + it("rejects forbidden attached context before claiming, debiting, or calling the model", async () => { + const serviceSupabase = new FakeRouterServiceSupabase({ [USER_ID]: 100 }); const chatComplete = vi.fn(); const handler = createOperatorRouterHandler({ - supabase, - findCompletedRoute: vi.fn(async () => null), + supabase: new MemorySupabase(), + serviceSupabase, consumeRateLimit: vi.fn(async () => true), getCreditBalance: vi.fn(async () => 100), - debit, chatComplete, model: "gpt-5.4-mini", apiKey: "key", @@ -1335,23 +1341,21 @@ describe("@pickforge/edge-shared", () => { ); expect(response.status).toBe(400); await expect(response.json()).resolves.toEqual({ error: "boundary_violation" }); - expect(debit).not.toHaveBeenCalled(); expect(chatComplete).not.toHaveBeenCalled(); + expect(serviceSupabase.rpcCalls).toHaveLength(0); }); - it("does not debit a failed route and allows a retry", async () => { - const supabase = new MemorySupabase(); - const debit = vi.fn(async () => ({ duplicate: false as const, balance: 98 })); + it("does not debit a failed provider attempt and lets an immediate retry reclaim it", async () => { + const serviceSupabase = new FakeRouterServiceSupabase({ [USER_ID]: 100 }); const chatComplete = vi .fn() .mockRejectedValueOnce(new Error("provider unavailable")) .mockResolvedValueOnce({ text: "{}", usage: { input: 1, output: 1 } }); const handler = createOperatorRouterHandler({ - supabase, - findCompletedRoute: vi.fn(async () => null), + supabase: new MemorySupabase(), + serviceSupabase, consumeRateLimit: vi.fn(async () => true), getCreditBalance: vi.fn(async () => 100), - debit, chatComplete, model: "gpt-5.4-mini", apiKey: "key", @@ -1360,20 +1364,123 @@ describe("@pickforge/edge-shared", () => { }); await expect(handler(routeRequest("open project Billing"))).resolves.toMatchObject({ status: 500 }); - expect(debit).not.toHaveBeenCalled(); + expect(serviceSupabase.rpcCalls.map((call) => call.fn)).toEqual([ + "router_attempt_claim", + "router_attempt_fail", + ]); await expect(handler(routeRequest("open project Billing"))).resolves.toMatchObject({ status: 200 }); - expect(debit).toHaveBeenCalledOnce(); + expect(serviceSupabase.rpcCalls.map((call) => call.fn)).toEqual([ + "router_attempt_claim", + "router_attempt_fail", + "router_attempt_claim", + "router_attempt_complete", + "debit_credits", + ]); + expect(chatComplete).toHaveBeenCalledTimes(2); + }); + + it("returns attempt_in_progress without invoking the provider or debiting while a claim is live", async () => { + const serviceSupabase = new FakeRouterServiceSupabase({ [USER_ID]: 100 }); + serviceSupabase.seedClaim(USER_ID, `router:${USER_ID}:router:attempt-1`); + const chatComplete = vi.fn(); + const handler = createOperatorRouterHandler({ + supabase: new MemorySupabase(), + serviceSupabase, + consumeRateLimit: vi.fn(async () => true), + getCreditBalance: vi.fn(async () => 100), + chatComplete, + model: "gpt-5.4-mini", + apiKey: "key", + baseUrl: "https://api.openai.com/v1", + creditCostCents: 2, + }); + + const response = await handler(routeRequest("open project Billing")); + expect(response.status).toBe(409); + await expect(response.json()).resolves.toEqual({ error: "attempt_in_progress" }); + expect(chatComplete).not.toHaveBeenCalled(); + expect(serviceSupabase.rpcCalls.map((call) => call.fn)).toEqual(["router_attempt_claim"]); + }); + + it("recovers a stale (lease-expired) claim and invokes the provider exactly once", async () => { + const serviceSupabase = new FakeRouterServiceSupabase({ [USER_ID]: 100 }); + serviceSupabase.seedClaim(USER_ID, `router:${USER_ID}:router:attempt-1`, 31_000); + const chatComplete = vi.fn(async () => ({ text: "{\"recovered\":true}", usage: { input: 1, output: 1 } })); + const handler = createOperatorRouterHandler({ + supabase: new MemorySupabase(), + serviceSupabase, + consumeRateLimit: vi.fn(async () => true), + getCreditBalance: vi.fn(async () => 100), + chatComplete, + model: "gpt-5.4-mini", + apiKey: "key", + baseUrl: "https://api.openai.com/v1", + creditCostCents: 2, + attemptLeaseSeconds: 30, + }); + + const response = await handler(routeRequest("open project Billing")); + expect(response.status).toBe(200); + await expect(response.json()).resolves.toEqual({ + proposalJson: "{\"recovered\":true}", + usage: { input: 1, output: 1 }, + costCents: 2, + }); + expect(chatComplete).toHaveBeenCalledOnce(); + }); + + it("retries only the debit, without re-invoking the provider, after a debit fails post-completion", async () => { + const serviceSupabase = new FakeRouterServiceSupabase({ [USER_ID]: 100 }); + serviceSupabase.failNextDebitWithoutCommitting(); + const chatComplete = vi.fn(async () => ({ text: "{\"once\":true}", usage: { input: 4, output: 6 } })); + const handler = createOperatorRouterHandler({ + supabase: new MemorySupabase(), + serviceSupabase, + consumeRateLimit: vi.fn(async () => true), + getCreditBalance: vi.fn(async () => 100), + chatComplete, + model: "gpt-5.4-mini", + apiKey: "key", + baseUrl: "https://api.openai.com/v1", + creditCostCents: 2, + }); + + await expect(handler(routeRequest("open project Billing"))).resolves.toMatchObject({ status: 500 }); + expect(chatComplete).toHaveBeenCalledOnce(); + expect(serviceSupabase.rpcCalls.map((call) => call.fn)).toEqual([ + "router_attempt_claim", + "router_attempt_complete", + "debit_credits", + ]); + + const response = await handler(routeRequest("open project Billing")); + expect(response.status).toBe(200); + await expect(response.json()).resolves.toEqual({ + proposalJson: "{\"once\":true}", + usage: { input: 4, output: 6 }, + costCents: 2, + }); + // The provider was invoked only once across both calls: the retry's claim + // finds the attempt already completed and skips straight to the debit. + expect(chatComplete).toHaveBeenCalledOnce(); + expect(serviceSupabase.rpcCalls.map((call) => call.fn)).toEqual([ + "router_attempt_claim", + "router_attempt_complete", + "debit_credits", + "router_attempt_claim", + "debit_credits", + ]); }); it("returns the stored result after a concurrent duplicate debit", async () => { - const stored = { proposalJson: "{\"stored\":true}", usage: { input: 5, output: 3 } }; - const findCompletedRoute = vi.fn().mockResolvedValueOnce(null).mockResolvedValueOnce(stored); + const serviceSupabase = new FakeRouterServiceSupabase({ [USER_ID]: 100 }); + const winner = { proposalJson: "{\"stored\":true}", usage: { input: 5, output: 3 } }; + serviceSupabase.forceDuplicateDebitOnce(winner); const handler = createOperatorRouterHandler({ supabase: new MemorySupabase(), - findCompletedRoute, + serviceSupabase, consumeRateLimit: vi.fn(async () => true), getCreditBalance: vi.fn(async () => 100), - debit: vi.fn(async () => ({ duplicate: true as const })), chatComplete: vi.fn(async () => ({ text: "{\"fresh\":true}", usage: { input: 1, output: 1 } })), model: "gpt-5.4-mini", apiKey: "key", @@ -1382,21 +1489,20 @@ describe("@pickforge/edge-shared", () => { }); const response = await handler(routeRequest("open project Billing")); - await expect(response.json()).resolves.toEqual({ ...stored, costCents: 2 }); - expect(findCompletedRoute).toHaveBeenCalledTimes(2); + await expect(response.json()).resolves.toEqual({ ...winner, costCents: 2 }); }); it("returns the stored proposal when a committed debit loses its response", async () => { - const stored = { proposalJson: "{\"stored\":true}", usage: { input: 5, output: 3 } }; + const serviceSupabase = new FakeRouterServiceSupabase({ [USER_ID]: 100 }); + const scopedKey = `router:${USER_ID}:router:attempt-1`; + serviceSupabase.loseNextDebitResponseAfterCommitting(); + const handler = createOperatorRouterHandler({ supabase: new MemorySupabase(), - findCompletedRoute: vi.fn().mockResolvedValueOnce(null).mockResolvedValueOnce(stored), + serviceSupabase, consumeRateLimit: vi.fn(async () => true), getCreditBalance: vi.fn(async () => 100), - debit: vi.fn(async () => { - throw new Error("response lost"); - }), - chatComplete: vi.fn(async () => ({ text: "{\"fresh\":true}", usage: { input: 1, output: 1 } })), + chatComplete: vi.fn(async () => ({ text: "{\"stored\":true}", usage: { input: 5, output: 3 } })), model: "gpt-5.4-mini", apiKey: "key", baseUrl: "https://api.openai.com/v1", @@ -1405,7 +1511,12 @@ describe("@pickforge/edge-shared", () => { const response = await handler(routeRequest("open project Billing")); expect(response.status).toBe(200); - await expect(response.json()).resolves.toEqual({ ...stored, costCents: 2 }); + await expect(response.json()).resolves.toEqual({ + proposalJson: "{\"stored\":true}", + usage: { input: 5, output: 3 }, + costCents: 2, + }); + expect(serviceSupabase.readLedgerRoute(USER_ID, scopedKey)).not.toBeNull(); }); it.each([ @@ -1441,15 +1552,14 @@ describe("@pickforge/edge-shared", () => { }); it("returns rate_limited after ten routed attempts", async () => { - const debit = vi.fn(async () => ({ duplicate: false as const, balance: 98 })); + const serviceSupabase = new FakeRouterServiceSupabase({ [USER_ID]: 100 }); const chatComplete = vi.fn(async () => ({ text: "{}", usage: { input: 1, output: 1 } })); let attempts = 0; const handler = createOperatorRouterHandler({ supabase: new MemorySupabase(), - findCompletedRoute: vi.fn(async () => null), + serviceSupabase, consumeRateLimit: vi.fn(async () => ++attempts <= 10), getCreditBalance: vi.fn(async () => 100), - debit, chatComplete, model: "gpt-5.4-mini", apiKey: "key", @@ -1466,17 +1576,17 @@ describe("@pickforge/edge-shared", () => { const response = await handler(routeRequest("open project Billing", undefined, "attempt-10")); expect(response.status).toBe(429); await expect(response.json()).resolves.toEqual({ error: "rate_limited" }); - expect(debit).toHaveBeenCalledTimes(10); expect(chatComplete).toHaveBeenCalledTimes(10); + expect(serviceSupabase.rpcCalls.filter((call) => call.fn === "debit_credits")).toHaveLength(10); }); it("maps invalid JSON to a boundary violation", async () => { + const serviceSupabase = new FakeRouterServiceSupabase({ [USER_ID]: 100 }); const handler = createOperatorRouterHandler({ supabase: new MemorySupabase(), - findCompletedRoute: vi.fn(async () => null), + serviceSupabase, consumeRateLimit: vi.fn(async () => true), getCreditBalance: vi.fn(async () => 100), - debit: vi.fn(), chatComplete: vi.fn(), model: "gpt-5.4-mini", apiKey: "key", @@ -1493,6 +1603,7 @@ describe("@pickforge/edge-shared", () => { ); expect(response.status).toBe(400); await expect(response.json()).resolves.toEqual({ error: "boundary_violation" }); + expect(serviceSupabase.rpcCalls).toHaveLength(0); }); }); @@ -1889,6 +2000,231 @@ class MemoryQuery implements SupabaseQueryBuilderLike { } } +interface FakeRouterAttemptRow { + status: "claimed" | "completed" | "failed"; + proposalJson?: string; + usageInput?: number; + usageOutput?: number; + claimedAtMs: number; +} + +/** + * Reimplements the router_attempt_claim / router_attempt_complete / + * router_attempt_fail / debit_credits RPC contracts (and a minimal + * credit_ledger table) in memory, exactly like `scopedDebitSupabase` already + * does for `debit_credits` alone. The real durable decisions are proven + * separately against Postgres by + * packages/edge-shared/test/router-attempt.contract.test.ts; this fake only + * needs to honor the same contract so createOperatorRouterHandler's own + * choreography (claim -> provider -> complete -> debit, and every recovery + * path) can be exercised fast and deterministically. + */ +class FakeRouterServiceSupabase implements Pick { + readonly rpcCalls: Array<{ fn: string; args: Record }> = []; + nowMs = 0; + private forcedDuplicateOnce: StoredRouteProposalLike | null = null; + private failNextDebitWithoutWriting = false; + private loseNextDebitResponseAfterCommit = false; + private readonly attempts = new Map(); + private readonly ledger = new Map(); + private readonly balances: Map; + + constructor(initialBalances: Record = {}) { + this.balances = new Map(Object.entries(initialBalances)); + } + + seedLegacyLedgerRoute(userId: string, idempotencyKey: string, route: StoredRouteProposalLike): void { + this.ledger.set(`${userId}:${idempotencyKey}`, route); + } + + seedClaim(userId: string, idempotencyKey: string, ageMs = 0): void { + this.attempts.set(`${userId}:${idempotencyKey}`, { status: "claimed", claimedAtMs: this.nowMs - ageMs }); + } + + forceDuplicateDebitOnce(route: StoredRouteProposalLike): void { + this.forcedDuplicateOnce = route; + } + + failNextDebitWithoutCommitting(): void { + this.failNextDebitWithoutWriting = true; + } + + loseNextDebitResponseAfterCommitting(): void { + this.loseNextDebitResponseAfterCommit = true; + } + + from(table: string): SupabaseQueryBuilderLike { + if (table !== "credit_ledger") { + throw new Error(`Unknown table: ${table}`); + } + + return new FakeLedgerQuery(this); + } + + async rpc(fn: string, args: Record = {}): Promise> { + this.rpcCalls.push({ fn, args }); + if (fn === "router_attempt_claim") return this.claim(args) as SupabaseQueryResult; + if (fn === "router_attempt_complete") return this.complete(args) as SupabaseQueryResult; + if (fn === "router_attempt_fail") return this.fail(args) as SupabaseQueryResult; + if (fn === "debit_credits") return this.debit(args) as SupabaseQueryResult; + + return { data: null, error: { message: `Unknown rpc: ${fn}` } }; + } + + readLedgerRoute(userId: string, idempotencyKey: string): unknown { + return this.ledger.get(`${userId}:${idempotencyKey}`) ?? null; + } + + private claim(args: Record): SupabaseQueryResult { + const key = attemptKey(args); + const leaseMs = Number(args.lease_seconds ?? 30) * 1000; + const existing = this.attempts.get(key); + if (existing === undefined) { + this.attempts.set(key, { status: "claimed", claimedAtMs: this.nowMs }); + return { data: { outcome: "claimed" }, error: null }; + } + if (existing.status === "completed") { + return { data: completedPayload(existing), error: null }; + } + if (existing.status === "failed" || this.nowMs - existing.claimedAtMs >= leaseMs) { + existing.status = "claimed"; + existing.claimedAtMs = this.nowMs; + existing.proposalJson = undefined; + existing.usageInput = undefined; + existing.usageOutput = undefined; + return { data: { outcome: "claimed" }, error: null }; + } + + return { data: { outcome: "in_progress" }, error: null }; + } + + private complete(args: Record): SupabaseQueryResult { + const key = attemptKey(args); + const existing = this.attempts.get(key); + if (existing === undefined) { + return { data: null, error: { message: "router attempt is not claimed" } }; + } + if (existing.status !== "completed") { + existing.status = "completed"; + existing.proposalJson = String(args.new_proposal_json); + existing.usageInput = Number(args.new_usage_input); + existing.usageOutput = Number(args.new_usage_output); + } + + return { data: completedPayload(existing), error: null }; + } + + private fail(args: Record): SupabaseQueryResult { + const key = attemptKey(args); + const existing = this.attempts.get(key); + if (existing !== undefined && existing.status === "claimed") { + existing.status = "failed"; + } + + return { data: null, error: null }; + } + + private debit(args: Record): SupabaseQueryResult { + const key = attemptKey({ target_user: args.target_user, idem_key: args.idem_key }); + if (this.forcedDuplicateOnce !== null) { + const route = this.forcedDuplicateOnce; + this.forcedDuplicateOnce = null; + if (!this.ledger.has(key)) { + this.ledger.set(key, route); + } + return { data: { status: "duplicate" }, error: null }; + } + if (this.failNextDebitWithoutWriting) { + this.failNextDebitWithoutWriting = false; + return { data: null, error: { message: "database unavailable" } }; + } + if (this.ledger.has(key)) { + return { data: { status: "duplicate" }, error: null }; + } + + const userId = String(args.target_user); + const debitCents = Number(args.debit_cents); + const balance = this.balances.get(userId) ?? 0; + if (balance < debitCents) { + return { data: { status: "insufficient", balance }, error: null }; + } + + const nextBalance = balance - debitCents; + this.balances.set(userId, nextBalance); + this.ledger.set(key, args.usage_metadata); + + if (this.loseNextDebitResponseAfterCommit) { + this.loseNextDebitResponseAfterCommit = false; + return { data: null, error: { message: "response lost" } }; + } + + return { data: { status: "ok", balance: nextBalance }, error: null }; + } +} + +interface StoredRouteProposalLike { + proposalJson: string; + usage: { input: number; output: number }; +} + +function attemptKey(args: Record): string { + return `${String(args.target_user)}:${String(args.idem_key)}`; +} + +function completedPayload(row: FakeRouterAttemptRow): Record { + return { + outcome: "completed", + proposal_json: row.proposalJson, + usage_input: row.usageInput, + usage_output: row.usageOutput, + }; +} + +class FakeLedgerQuery implements SupabaseQueryBuilderLike { + private readonly filters: Record = {}; + + constructor(private readonly supabase: FakeRouterServiceSupabase) {} + + select(): SupabaseQueryBuilderLike { + return this; + } + + eq(column: string, value: unknown): SupabaseQueryBuilderLike { + this.filters[column] = value; + return this; + } + + is(column: string, value: unknown): SupabaseQueryBuilderLike { + return this.eq(column, value); + } + + order(): SupabaseQueryBuilderLike { + return this; + } + + range(): SupabaseQueryBuilderLike { + return this; + } + + async maybeSingle(): Promise> { + const userId = String(this.filters.user_id); + const idempotencyKey = String(this.filters.idempotency_key); + const metadata = this.supabase.readLedgerRoute(userId, idempotencyKey); + + return { data: metadata === null ? null : ({ metadata } as T), error: null }; + } + + then, TResult2 = never>( + onfulfilled?: ((value: SupabaseQueryResult) => TResult1 | PromiseLike) | null, + onrejected?: ((reason: unknown) => TResult2 | PromiseLike) | null, + ): PromiseLike { + return this.maybeSingle().then( + onfulfilled as (value: SupabaseQueryResult) => TResult1 | PromiseLike, + onrejected, + ); + } +} + function scopedDebitSupabase(initialBalances: Record): Pick { const balances = new Map(Object.entries(initialBalances)); const debits = new Set(); diff --git a/packages/edge-shared/test/postgres-router-client.ts b/packages/edge-shared/test/postgres-router-client.ts new file mode 100644 index 0000000..1b6d117 --- /dev/null +++ b/packages/edge-shared/test/postgres-router-client.ts @@ -0,0 +1,181 @@ +// Local-Postgres contract-lane adapter. Implements the same `SupabaseClientLike` +// boundary `claimRouteAttempt` / `completeRouteAttempt` / `failRouteAttempt` / +// `findDebitedRouteResult` / `debitCredits` are written against, but backed by +// a real `bun:sql` connection instead of the in-memory fake used by the fast +// unit tests. No claim/complete/debit decision vocabulary is reimplemented +// here: every decision is made by the durable SQL functions in +// supabase/migrations, exactly as production (the service-role client) does. +import type { SQL } from "bun"; +import type { + SupabaseClientLike, + SupabaseErrorLike, + SupabaseQueryBuilderLike, + SupabaseQueryResult, +} from "../src/index.js"; + +// The subset of RPCs whose SQL return type is jsonb: bun's postgres driver +// hands jsonb columns back as text, so these need JSON.parse. +const JSONB_RPCS = new Set(["router_attempt_claim", "router_attempt_complete", "debit_credits"]); + +const KNOWN_TABLES = ["credit_ledger"] as const; +type KnownTable = (typeof KNOWN_TABLES)[number]; + +export function createPostgresRouterClient(sql: SQL): Pick { + return { + from(table: string) { + if (!isKnownTable(table)) { + throw new Error(`Unknown table: ${table}`); + } + return new PostgresLedgerQuery(sql, table); + }, + rpc(fn: string, args: Record = {}) { + return callRpc(sql, fn, args) as Promise>; + }, + }; +} + +/** + * Production RPC calls each run as their own independent PostgREST + * request/transaction, so a caught error (e.g. a rejected claim) on one call + * never affects another. The contract lane instead nests calls inside one + * outer per-test transaction (for cheap rollback-based cleanup), so each call + * runs in its own SAVEPOINT: an error rolls back only that call, keeping the + * outer transaction usable, exactly like independent requests would behave + * in production. + */ +async function withIsolatedQuery(sql: SQL, run: (handle: SQL) => Promise): Promise { + const transactional = sql as SQL & { + savepoint?: (fn: (sp: SQL) => Promise) => Promise; + }; + if (typeof transactional.savepoint === "function") { + return transactional.savepoint((sp) => run(sp)); + } + return run(sql); +} + +async function callRpc( + sql: SQL, + fn: string, + args: Record, +): Promise> { + const keys = Object.keys(args); + const placeholders = keys.map((key, index) => `${key} => $${index + 1}`).join(", "); + // Bun's `postgres` driver double-encodes JSON.stringify'd object params + // bound to a jsonb-typed function argument (it JSON-serializes them again + // client-side), so pass objects (like usage_metadata) through as-is and let + // the driver encode them. + const values = keys.map((key) => args[key]); + const text = `select public.${fn}(${placeholders}) as result`; + + try { + const rows = await withIsolatedQuery(sql, (handle) => handle.unsafe(text, values)); + const raw = rows[0]?.result; + if (JSONB_RPCS.has(fn)) { + return { data: typeof raw === "string" ? JSON.parse(raw) : (raw ?? null), error: null }; + } + return { data: raw === "" ? null : raw, error: null }; + } catch (error) { + return { data: null, error: toSupabaseError(error) }; + } +} + +function toSupabaseError(error: unknown): SupabaseErrorLike { + if (error !== null && typeof error === "object" && "message" in error) { + const pgError = error as { message: unknown; code?: unknown; errno?: unknown }; + const code = + typeof pgError.errno === "string" + ? pgError.errno + : typeof pgError.code === "string" + ? pgError.code + : undefined; + return { + code, + message: typeof pgError.message === "string" ? pgError.message : String(pgError.message), + }; + } + return { message: String(error) }; +} + +class PostgresLedgerQuery implements SupabaseQueryBuilderLike { + private columns = "*"; + private readonly filters: Array<{ column: string; value: unknown }> = []; + + constructor( + private readonly sql: SQL, + private readonly table: KnownTable, + ) {} + + select(columns = "*"): SupabaseQueryBuilderLike { + this.columns = columns; + return this; + } + + eq(column: string, value: unknown): SupabaseQueryBuilderLike { + this.filters.push({ column, value }); + return this; + } + + is(column: string, value: null): SupabaseQueryBuilderLike { + return this.eq(column, value); + } + + order(): SupabaseQueryBuilderLike { + return this; + } + + range(): SupabaseQueryBuilderLike { + return this; + } + + async maybeSingle(): Promise> { + const result = await this.execute(); + const rows = Array.isArray(result.data) ? result.data : result.data === null ? [] : [result.data]; + return { data: (rows[0] ?? null) as T | null, error: result.error }; + } + + then, TResult2 = never>( + onfulfilled?: ((value: SupabaseQueryResult) => TResult1 | PromiseLike) | null, + onrejected?: ((reason: unknown) => TResult2 | PromiseLike) | null, + ): PromiseLike { + return this.execute().then(onfulfilled, onrejected); + } + + private async execute(): Promise> { + try { + return await withIsolatedQuery(this.sql, (handle) => this.runSelect(handle)); + } catch (error) { + return { data: null, error: toSupabaseError(error) }; + } + } + + private async runSelect(handle: SQL): Promise> { + const params: unknown[] = []; + const where = this.filters + .map((filter) => { + params.push(filter.value); + return `${filter.column} = $${params.length}`; + }) + .join(" and "); + let text = `select ${this.columns} from public.${this.table}`; + if (where.length > 0) { + text += ` where ${where}`; + } + const rows = await handle.unsafe(text, params); + return { data: rows.map(parseJsonbMetadata) as T, error: null }; + } +} + +// Bun's postgres driver decodes a jsonb column fetched via a plain SELECT as +// text (unlike an insert/update `returning`, which the driver does parse), +// but `SupabaseClientLike` consumers expect the already-decoded value, same +// as a real supabase-js/PostgREST client returns. +function parseJsonbMetadata(row: Record): Record { + if (typeof row.metadata !== "string") { + return row; + } + return { ...row, metadata: JSON.parse(row.metadata) }; +} + +function isKnownTable(table: string): table is KnownTable { + return (KNOWN_TABLES as readonly string[]).includes(table); +} diff --git a/packages/edge-shared/test/router-attempt.contract.test.ts b/packages/edge-shared/test/router-attempt.contract.test.ts new file mode 100644 index 0000000..2b4dafd --- /dev/null +++ b/packages/edge-shared/test/router-attempt.contract.test.ts @@ -0,0 +1,288 @@ +// Operator-router durable claim contract lane (issue #37, CAND-4). +// +// Exercises the production `claimRouteAttempt` / `completeRouteAttempt` / +// `failRouteAttempt` / `findDebitedRouteResult` / `debitCredits` functions, +// and the full `createOperatorRouterHandler` composition built from them, +// against the REAL durable `router_attempt_claim` / `router_attempt_complete` +// / `router_attempt_fail` SQL functions in +// supabase/migrations/20260722000000_router_attempt_claim.sql, running on a +// local Supabase Postgres. The claim/complete/fail decision lives entirely +// behind the `SupabaseClientLike` boundary — this file never reimplements +// that policy, it only asserts on outcomes, including genuine concurrent +// races between independently committing Postgres transactions. +// +// Prerequisite: a local Supabase Postgres via `supabase start` (requires +// Docker). When it is not reachable, every test below is skipped with a +// console message instead of failing the run. CI's `database` job starts +// Supabase and runs this lane for real via `bun run test:supabase`. +import { afterAll, beforeAll, describe, expect, it } from "bun:test"; +import type { SQL } from "bun"; +import { + claimRouteAttempt, + completeRouteAttempt, + createOperatorRouterHandler, + debitCredits, + failRouteAttempt, + findDebitedRouteResult, +} from "../src/index.js"; +import { createPostgresRouterClient } from "./postgres-router-client.js"; +import { + cleanupRouterFixtures, + connectToLocalDatabase, + DEFAULT_DATABASE_URL, + grantCredits, + insertAuthUser, + resolveLocalDatabaseUrl, + uniqueId, + withRollback, +} from "./router-fixtures.js"; + +const sql = await connectToLocalDatabase(); + +if (sql === null) { + console.log( + "[edge-shared contract lane] Skipping router-attempt Postgres contract tests: " + + `no local Supabase Postgres reachable at ${resolveLocalDatabaseUrl() ?? process.env.SUPABASE_DB_URL ?? DEFAULT_DATABASE_URL}. ` + + "Run `supabase start` (requires Docker) to exercise this lane locally; " + + "CI's `database` job runs it automatically via `bun run test:supabase`.", + ); +} + +describe.skipIf(sql === null)("@pickforge/edge-shared router attempt claim contract (local Postgres)", () => { + const db = sql as SQL; + let welcomeCreditsCampaignWasEnabled = true; + + beforeAll(async () => { + // The launch welcome-credit campaign grants an extra 100 cents to every + // new auth user while under its cap. That is orthogonal to router claim + // policy, so disable it for this lane's duration rather than coupling + // credit-balance assertions to however many users other test runs have + // already created. + const [campaign] = await db.unsafe( + "select enabled from welcome_credits_private.campaigns where campaign_key = 'launch_welcome_first_50'", + ); + welcomeCreditsCampaignWasEnabled = Boolean(campaign?.enabled); + await db.unsafe( + "update welcome_credits_private.campaigns set enabled = false where campaign_key = 'launch_welcome_first_50'", + ); + }); + + afterAll(async () => { + await db.unsafe("update welcome_credits_private.campaigns set enabled = $1 where campaign_key = 'launch_welcome_first_50'", [ + welcomeCreditsCampaignWasEnabled, + ]); + await db.close(); + }); + + async function createUser(tx: SQL, label: string): Promise { + const userId = crypto.randomUUID(); + await insertAuthUser(tx, { id: userId, email: `${uniqueId(label)}@example.invalid` }); + return userId; + } + + it("durably claims an absent key, and a live claim is reported in_progress, never re-claimed", async () => { + await withRollback(db, async (tx) => { + const userId = await createUser(tx, "claim-live"); + const client = createPostgresRouterClient(tx); + const key = uniqueId("router:attempt"); + + await expect(claimRouteAttempt({ supabase: client, userId, idempotencyKey: key })).resolves.toEqual({ + outcome: "claimed", + }); + await expect(claimRouteAttempt({ supabase: client, userId, idempotencyKey: key })).resolves.toEqual({ + outcome: "in_progress", + }); + }); + }); + + it("records a provider outcome, and a later claim returns it instead of reclaiming", async () => { + await withRollback(db, async (tx) => { + const userId = await createUser(tx, "claim-complete"); + const client = createPostgresRouterClient(tx); + const key = uniqueId("router:attempt"); + const result = { proposalJson: "{\"action\":\"openProject\"}", usage: { input: 42, output: 13 } }; + + await claimRouteAttempt({ supabase: client, userId, idempotencyKey: key }); + await expect( + completeRouteAttempt({ supabase: client, userId, idempotencyKey: key, result }), + ).resolves.toEqual(result); + + // A later caller (e.g. the original caller's own retry after a debit + // failure) sees the completed outcome and can skip straight to a debit + // retry, instead of re-invoking the provider. + await expect(claimRouteAttempt({ supabase: client, userId, idempotencyKey: key })).resolves.toEqual({ + outcome: "completed", + result, + }); + }); + }); + + it("releases a claim on a definitive provider failure so a retry can reclaim immediately", async () => { + await withRollback(db, async (tx) => { + const userId = await createUser(tx, "claim-fail"); + const client = createPostgresRouterClient(tx); + const key = uniqueId("router:attempt"); + + await claimRouteAttempt({ supabase: client, userId, idempotencyKey: key }); + await failRouteAttempt({ supabase: client, userId, idempotencyKey: key }); + + await expect(claimRouteAttempt({ supabase: client, userId, idempotencyKey: key })).resolves.toEqual({ + outcome: "claimed", + }); + }); + }); + + it("recovers a stale (lease-expired) claim without waiting out a fresh lease", async () => { + await withRollback(db, async (tx) => { + const userId = await createUser(tx, "claim-stale"); + const client = createPostgresRouterClient(tx); + const key = uniqueId("router:attempt"); + + await claimRouteAttempt({ supabase: client, userId, idempotencyKey: key, leaseSeconds: 30 }); + await tx.unsafe( + `update public.router_attempts set claimed_at = now() - interval '1 hour' + where user_id = $1 and idempotency_key = $2`, + [userId, key], + ); + + await expect( + claimRouteAttempt({ supabase: client, userId, idempotencyKey: key, leaseSeconds: 30 }), + ).resolves.toEqual({ outcome: "claimed" }); + }); + }); + + it("couples debit to the completed route, and honors it as a legacy ledger replay", async () => { + await withRollback(db, async (tx) => { + const userId = await createUser(tx, "claim-debit"); + await grantCredits(tx, userId, 100); + const client = createPostgresRouterClient(tx); + const key = uniqueId("router:attempt"); + const result = { proposalJson: "{\"action\":\"openProject\"}", usage: { input: 8, output: 5 } }; + + await claimRouteAttempt({ supabase: client, userId, idempotencyKey: key }); + await completeRouteAttempt({ supabase: client, userId, idempotencyKey: key, result }); + + await expect(findDebitedRouteResult({ supabase: client, userId, idempotencyKey: key })).resolves.toBeNull(); + + await expect( + debitCredits({ + supabase: client, + userId, + amountCents: 2, + reason: "Operator routing", + idempotencyKey: key, + metadata: { proposalJson: result.proposalJson, usage: result.usage }, + }), + ).resolves.toEqual({ duplicate: false, balance: 98 }); + + // The debited ledger row is the money-path source of truth for "has + // this key already been charged" — findDebitedRouteResult reads it + // back the same way it would honor a row written before the durable + // claim table existed. + await expect(findDebitedRouteResult({ supabase: client, userId, idempotencyKey: key })).resolves.toEqual( + result, + ); + }); + }); + + it("honors a pre-existing (legacy) debited ledger row with no router_attempts claim at all", async () => { + await withRollback(db, async (tx) => { + const userId = await createUser(tx, "legacy-ledger"); + const client = createPostgresRouterClient(tx); + const key = uniqueId("router:attempt"); + await tx.unsafe( + `insert into public.credit_ledger (user_id, amount_cents, kind, description, idempotency_key, metadata) + values ($1, -2, 'usage', 'Operator routing', $2, $3::jsonb)`, + [ + userId, + key, + JSON.stringify({ proposalJson: "{\"legacy\":true}", usage: { input: 1, output: 1 } }), + ], + ); + + await expect(findDebitedRouteResult({ supabase: client, userId, idempotencyKey: key })).resolves.toEqual({ + proposalJson: "{\"legacy\":true}", + usage: { input: 1, output: 1 }, + }); + }); + }); + + it("resolves two genuinely concurrent claims for the same key to exactly one claim and one in_progress", async () => { + const userId = crypto.randomUUID(); + await insertAuthUser(db, { id: userId, email: `${uniqueId("concurrent-claim")}@example.invalid` }); + + try { + const key = uniqueId("router:attempt"); + const [first, second] = await Promise.all([ + db.begin(async (tx) => claimRouteAttempt({ supabase: createPostgresRouterClient(tx), userId, idempotencyKey: key })), + db.begin(async (tx) => claimRouteAttempt({ supabase: createPostgresRouterClient(tx), userId, idempotencyKey: key })), + ]); + + const outcomes = [first.outcome, second.outcome].sort(); + expect(outcomes).toEqual(["claimed", "in_progress"]); + + const rows = await db.unsafe( + "select status from public.router_attempts where user_id = $1 and idempotency_key = $2", + [userId, key], + ); + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ status: "claimed" }); + } finally { + await cleanupRouterFixtures(db, { userIds: [userId] }); + } + }); + + it("invokes the provider and debits exactly once across two genuinely concurrent same-key requests", async () => { + const userId = crypto.randomUUID(); + await insertAuthUser(db, { id: userId, email: `${uniqueId("concurrent-handler")}@example.invalid` }); + await grantCredits(db, userId, 100); + + try { + const serviceSupabase = createPostgresRouterClient(db); + let providerCallCount = 0; + const handler = createOperatorRouterHandler({ + supabase: { + auth: { getUser: async () => ({ data: { user: { id: userId } }, error: null }) }, + }, + serviceSupabase, + getCreditBalance: async () => 100, + consumeRateLimit: async () => true, + chatComplete: async () => { + providerCallCount += 1; + // Give both concurrent requests a chance to reach the provider + // step before either resolves, so a real race is exercised even + // though only one of them should ever get here. + await new Promise((resolve) => setTimeout(resolve, 20)); + return { text: "{\"action\":{\"action\":\"openProject\"}}", usage: { input: 4, output: 6 } }; + }, + model: "gpt-5.4-mini", + apiKey: "key", + baseUrl: "https://api.openai.com/v1", + creditCostCents: 2, + }); + + const request = () => + new Request("https://edge.test", { + method: "POST", + headers: { Authorization: "Bearer token", "x-idempotency-key": "concurrent-attempt" }, + body: JSON.stringify({ commandText: "open project Billing" }), + }); + + const [first, second] = await Promise.all([handler(request()), handler(request())]); + const statuses = [first.status, second.status].sort(); + // The winner gets 200; the loser sees the live claim and is told to + // retry (409), never a second provider call or a second debit. + expect(statuses).toEqual([200, 409]); + expect(providerCallCount).toBe(1); + + const ledgerRows = await db.unsafe( + "select amount_cents from public.credit_ledger where user_id = $1 and kind = 'usage'", + [userId], + ); + expect(ledgerRows).toHaveLength(1); + expect(ledgerRows[0]).toMatchObject({ amount_cents: -2 }); + } finally { + await cleanupRouterFixtures(db, { userIds: [userId] }); + } + }); +}); diff --git a/packages/edge-shared/test/router-fixtures.ts b/packages/edge-shared/test/router-fixtures.ts new file mode 100644 index 0000000..e524ee0 --- /dev/null +++ b/packages/edge-shared/test/router-fixtures.ts @@ -0,0 +1,113 @@ +// Local-Postgres contract-lane fixtures. Mirrors the `SUPABASE_DB_URL` / +// localhost-only guard used by `packages/billing/test/lifecycle-fixtures.ts` +// and `packages/sync/test/lww-fixtures.ts` so this lane can never +// accidentally target a non-disposable database, and gracefully reports why +// it is skipped when no local Supabase is running. +import { SQL } from "bun"; + +export const DEFAULT_DATABASE_URL = "postgresql://postgres:postgres@127.0.0.1:54322/postgres"; + +export function resolveLocalDatabaseUrl(): string | null { + const raw = process.env.SUPABASE_DB_URL ?? DEFAULT_DATABASE_URL; + let parsed: URL; + try { + parsed = new URL(raw); + } catch { + return null; + } + const isDisposableLocalDatabase = + ["127.0.0.1", "localhost", "[::1]"].includes(parsed.hostname) && + parsed.port === "54322" && + parsed.pathname === "/postgres" && + parsed.username === "postgres"; + return isDisposableLocalDatabase ? raw : null; +} + +/** + * Connects to the local Supabase Postgres started by `supabase start`. Returns + * `null` (never throws) when the database is unreachable so the contract lane + * can skip cleanly instead of failing CI runs that have no local Postgres. + */ +export async function connectToLocalDatabase(): Promise { + const databaseUrl = resolveLocalDatabaseUrl(); + if (databaseUrl === null) { + return null; + } + + const sql = new SQL(databaseUrl, { max: 10 }); + try { + await sql.unsafe("select 1"); + } catch { + await sql.close().catch(() => {}); + return null; + } + return sql; +} + +export function uniqueId(prefix: string): string { + return `${prefix}_${crypto.randomUUID().replaceAll("-", "")}`; +} + +export async function insertAuthUser( + sql: Pick, + user: { id: string; email: string }, +): Promise { + await sql.unsafe( + `insert into auth.users ( + id, aud, role, email, raw_app_meta_data, raw_user_meta_data, is_anonymous, created_at, updated_at + ) values ($1, 'authenticated', 'authenticated', $2, '{}'::jsonb, '{}'::jsonb, false, now(), now())`, + [user.id, user.email], + ); +} + +/** + * Grants `amountCents` directly on the ledger so a test user has a known, + * deterministic balance to debit from — independent of however many launch + * welcome credits other test runs have already issued. + */ +export async function grantCredits(sql: Pick, userId: string, amountCents: number): Promise { + await sql.unsafe( + `insert into public.credit_ledger (user_id, amount_cents, kind, description, idempotency_key) + values ($1, $2, 'grant', 'contract-lane fixture credit', $3)`, + [userId, amountCents, uniqueId("fixture-grant")], + ); +} + +/** + * Cleans up rows the contract lane cannot roll back transactionally: used + * only by the true-concurrency tests, which need two independent + * connections/transactions racing for real, so they commit instead of + * rolling back. Deleting the auth user cascades its router_attempts and + * credit_ledger rows. + */ +export async function cleanupRouterFixtures( + sql: Pick, + fixtures: { userIds: string[] }, +): Promise { + if (fixtures.userIds.length === 0) { + return; + } + await sql.unsafe("delete from auth.users where id = any($1::uuid[])", [ + sql.array(fixtures.userIds, "uuid"), + ]); +} + +/** + * Runs `run` inside a transaction that always rolls back, so each contract + * test is fully isolated without needing bespoke per-table cleanup. Only the + * true-concurrency tests opt out of this helper, since they need two + * independently committing connections. + */ +export async function withRollback(sql: SQL, run: (tx: SQL) => Promise): Promise { + const rollbackSentinel = Symbol("contract-lane-rollback"); + try { + await sql.begin(async (tx) => { + await run(tx); + throw rollbackSentinel; + }); + } catch (error) { + if (error !== rollbackSentinel) { + throw error; + } + } +} diff --git a/packages/flags/package.json b/packages/flags/package.json index 15b6935..8321572 100644 --- a/packages/flags/package.json +++ b/packages/flags/package.json @@ -1,6 +1,6 @@ { "name": "@pickforge/flags", - "version": "0.11.0", + "version": "0.12.0", "description": "UI-free feature-flag registry for release gating in Pickforge apps.", "license": "MIT", "repository": { diff --git a/packages/sync/package.json b/packages/sync/package.json index f1d2662..740c491 100644 --- a/packages/sync/package.json +++ b/packages/sync/package.json @@ -1,6 +1,6 @@ { "name": "@pickforge/sync", - "version": "0.11.0", + "version": "0.12.0", "description": "UI-free settings sync helpers for Pickforge apps.", "license": "MIT", "repository": { diff --git a/packages/tauri-release/package.json b/packages/tauri-release/package.json index 7a28a19..489bc54 100644 --- a/packages/tauri-release/package.json +++ b/packages/tauri-release/package.json @@ -1,6 +1,6 @@ { "name": "@pickforge/tauri-release", - "version": "0.11.0", + "version": "0.12.0", "description": "Signed Tauri release and updater-feed automation for Pickforge apps.", "license": "MIT", "repository": { diff --git a/supabase/functions/create-credit-checkout/deno.json b/supabase/functions/create-credit-checkout/deno.json index 36cf97a..1b6cf6c 100644 --- a/supabase/functions/create-credit-checkout/deno.json +++ b/supabase/functions/create-credit-checkout/deno.json @@ -2,7 +2,7 @@ "imports": { "stripe": "npm:stripe@19.1.0", "@supabase/supabase-js": "npm:@supabase/supabase-js@2.110.0", - "@pickforge/billing": "npm:@pickforge/billing@0.11.0", - "@pickforge/edge-shared": "npm:@pickforge/edge-shared@0.11.0" + "@pickforge/billing": "npm:@pickforge/billing@0.12.0", + "@pickforge/edge-shared": "npm:@pickforge/edge-shared@0.12.0" } } diff --git a/supabase/functions/delete-account/deno.json b/supabase/functions/delete-account/deno.json index 3cdd028..16ee2f6 100644 --- a/supabase/functions/delete-account/deno.json +++ b/supabase/functions/delete-account/deno.json @@ -2,6 +2,6 @@ "imports": { "stripe": "npm:stripe@19.1.0", "@supabase/supabase-js": "npm:@supabase/supabase-js@2.110.0", - "@pickforge/edge-shared": "npm:@pickforge/edge-shared@0.11.0" + "@pickforge/edge-shared": "npm:@pickforge/edge-shared@0.12.0" } } diff --git a/supabase/functions/export-account-data/deno.json b/supabase/functions/export-account-data/deno.json index 8471f20..d904541 100644 --- a/supabase/functions/export-account-data/deno.json +++ b/supabase/functions/export-account-data/deno.json @@ -1,6 +1,6 @@ { "imports": { "@supabase/supabase-js": "npm:@supabase/supabase-js@2.110.0", - "@pickforge/edge-shared": "npm:@pickforge/edge-shared@0.11.0" + "@pickforge/edge-shared": "npm:@pickforge/edge-shared@0.12.0" } } diff --git a/supabase/functions/operator-router/deno.json b/supabase/functions/operator-router/deno.json index d97951e..6c5d956 100644 --- a/supabase/functions/operator-router/deno.json +++ b/supabase/functions/operator-router/deno.json @@ -1,7 +1,7 @@ { "imports": { "@supabase/supabase-js": "npm:@supabase/supabase-js@2.110.0", - "@pickforge/billing": "npm:@pickforge/billing@0.11.0", - "@pickforge/edge-shared": "npm:@pickforge/edge-shared@0.11.0" + "@pickforge/billing": "npm:@pickforge/billing@0.12.0", + "@pickforge/edge-shared": "npm:@pickforge/edge-shared@0.12.0" } } diff --git a/supabase/functions/operator-router/index.ts b/supabase/functions/operator-router/index.ts index 569dc85..f2028b0 100644 --- a/supabase/functions/operator-router/index.ts +++ b/supabase/functions/operator-router/index.ts @@ -6,7 +6,6 @@ import { createRequiredEnv, corsHeaders, corsPreflightResponse, - debitCredits, EdgeSharedError, jsonResponse, withCors, @@ -35,47 +34,20 @@ Deno.serve(async (req) => { const handler = createOperatorRouterHandler({ supabase: createCallerSupabase(req), - findCompletedRoute, + serviceSupabase, getCreditBalance: (userId) => getCreditBalanceCents({ supabase: serviceSupabase, userId }), consumeRateLimit, - debit: (options) => debitCredits({ supabase: serviceSupabase, ...options }), chatComplete, model: Deno.env.get("OPENAI_ROUTER_MODEL") ?? "gpt-5.4-mini", apiKey, baseUrl: Deno.env.get("OPENAI_BASE_URL") ?? "https://api.openai.com/v1", creditCostCents: readPositiveIntegerEnv("ROUTER_CREDIT_COST_CENTS", 2), + attemptLeaseSeconds: readPositiveIntegerEnv("ROUTER_ATTEMPT_LEASE_SECONDS", 30), }); return withCors(handler(req)); }); -async function findCompletedRoute(userId: string, idempotencyKey: string) { - const { data, error } = await serviceSupabase - .from<{ metadata: unknown }>("credit_ledger") - .select("metadata") - .eq("user_id", userId) - .eq("idempotency_key", idempotencyKey) - .maybeSingle(); - if (error !== null) { - throw new EdgeSharedError("database_error", "Failed to read stored router result", { cause: error }); - } - if (data === null) { - return null; - } - - const proposalJson = (data.metadata as { proposalJson?: unknown }).proposalJson; - const usage = (data.metadata as { usage?: { input?: unknown; output?: unknown } }).usage; - if ( - typeof proposalJson !== "string" || - !Number.isSafeInteger(usage?.input) || - !Number.isSafeInteger(usage?.output) - ) { - return null; - } - - return { proposalJson, usage: { input: usage.input, output: usage.output } }; -} - async function consumeRateLimit(userId: string): Promise { const { data, error } = await serviceSupabase.rpc("consume_router_rate_limit", { target_user: userId, diff --git a/supabase/migrations/20260722000000_router_attempt_claim.sql b/supabase/migrations/20260722000000_router_attempt_claim.sql new file mode 100644 index 0000000..a062786 --- /dev/null +++ b/supabase/migrations/20260722000000_router_attempt_claim.sql @@ -0,0 +1,234 @@ +-- Issue #37 (CAND-4): operator-router's idempotency only became durable +-- AFTER provider work. The handler checked `credit_ledger` for a completed +-- route, invoked the OpenAI-compatible provider, and only then wrote a +-- debit row keyed by the idempotency key. Two concurrent requests for the +-- same key could both find nothing completed, both call the provider, and +-- the losing response was simply discarded after the winner's debit landed +-- — wasted provider spend with no compensating record. +-- +-- `router_attempts` and its three RPCs move the durable decision in front of +-- the provider call: `router_attempt_claim` durably claims the (user_id, +-- idempotency_key) attempt before any provider invocation happens, so at +-- most one caller ever owns "go call the provider" for a given key at a +-- time. `router_attempt_complete` records the provider outcome once it is +-- known (independent of whether the following debit succeeds), and +-- `router_attempt_fail` releases a claim immediately after a definitive +-- provider failure so a retry does not have to wait out the lease. +-- +-- A claim that is neither completed nor explicitly failed (e.g. the edge +-- function crashed mid-flight) is recoverable once its lease +-- (`lease_seconds`, caller-supplied) has elapsed — a stale claim is exactly +-- the "in-progress recovery" case: nothing else can prove the original +-- attempt is dead, so a bounded lease is the only way to make forward +-- progress possible again. +-- +-- This is additive only (new table, new functions): existing +-- `credit_ledger` rows written by `debit_credits` before this migration +-- remain the authoritative record for old completed routes, and the +-- application layer (edge-shared) keeps honoring them — this migration does +-- not touch `credit_ledger` or `debit_credits`. +create table public.router_attempts ( + user_id uuid not null references auth.users(id) on delete cascade, + idempotency_key text not null, + status text not null check (status in ('claimed', 'completed', 'failed')), + proposal_json text, + usage_input integer, + usage_output integer, + claimed_at timestamptz not null default now(), + completed_at timestamptz, + primary key (user_id, idempotency_key), + constraint router_attempts_completed_has_result check ( + status <> 'completed' + or ( + proposal_json is not null + and usage_input is not null + and usage_output is not null + and completed_at is not null + ) + ) +); + +alter table public.router_attempts enable row level security; + +-- Server-only idempotency table: RLS is enabled intentionally with no +-- policies, and no client role gets any direct table privilege at all — the +-- three RPCs below (service_role only) are the only way in. +revoke all on public.router_attempts from anon, authenticated; + +-- Durably claims (user_id, idempotency_key) before any provider invocation. +-- Returns one of: +-- {"outcome": "claimed"} caller now owns provider invocation for this key. +-- {"outcome": "in_progress"} another live claim owns it right now; caller +-- must not invoke the provider or debit. +-- {"outcome": "completed", "proposal_json": ..., "usage_input": ..., +-- "usage_output": ...} the provider already ran for this key; caller +-- should skip straight to the debit step with +-- the returned result instead of re-invoking +-- the provider. +create function public.router_attempt_claim( + target_user uuid, + idem_key text, + lease_seconds integer default 30 +) +returns jsonb +language plpgsql +security invoker +set search_path = public +as $$ +declare + current_row public.router_attempts%rowtype; +begin + if idem_key is null or btrim(idem_key) = '' then + raise exception 'idem_key must be a non-empty string' + using errcode = '22023'; + end if; + + if lease_seconds is null or lease_seconds <= 0 then + raise exception 'lease_seconds must be a positive integer' + using errcode = '22023'; + end if; + + insert into public.router_attempts (user_id, idempotency_key, status, claimed_at) + values (target_user, idem_key, 'claimed', now()) + on conflict (user_id, idempotency_key) do nothing + returning * into current_row; + + if found then + return jsonb_build_object('outcome', 'claimed'); + end if; + + -- Lost the insert race (or a prior attempt already exists): lock the row + -- and decide what a caller may do with it. + select * + into current_row + from public.router_attempts + where user_id = target_user + and idempotency_key = idem_key + for update; + + if current_row.status = 'completed' then + return jsonb_build_object( + 'outcome', 'completed', + 'proposal_json', current_row.proposal_json, + 'usage_input', current_row.usage_input, + 'usage_output', current_row.usage_output + ); + end if; + + if current_row.status = 'failed' + or current_row.claimed_at <= now() - make_interval(secs => lease_seconds) + then + update public.router_attempts + set status = 'claimed', + claimed_at = now(), + proposal_json = null, + usage_input = null, + usage_output = null, + completed_at = null + where user_id = target_user + and idempotency_key = idem_key; + + return jsonb_build_object('outcome', 'claimed'); + end if; + + return jsonb_build_object('outcome', 'in_progress'); +end; +$$; + +-- Records a provider outcome for a claimed attempt. Idempotent: replaying a +-- completion for an already-completed attempt (e.g. a retried caller that +-- still holds a reference to its own claim) returns the stored outcome +-- instead of erroring. +create function public.router_attempt_complete( + target_user uuid, + idem_key text, + new_proposal_json text, + new_usage_input integer, + new_usage_output integer +) +returns jsonb +language plpgsql +security invoker +set search_path = public +as $$ +declare + current_row public.router_attempts%rowtype; +begin + if new_proposal_json is null or btrim(new_proposal_json) = '' then + raise exception 'new_proposal_json must be a non-empty string' + using errcode = '22023'; + end if; + + if new_usage_input is null or new_usage_input < 0 or new_usage_output is null or new_usage_output < 0 then + raise exception 'new_usage_input and new_usage_output must be non-negative integers' + using errcode = '22023'; + end if; + + update public.router_attempts + set status = 'completed', + proposal_json = new_proposal_json, + usage_input = new_usage_input, + usage_output = new_usage_output, + completed_at = now() + where user_id = target_user + and idempotency_key = idem_key + and status = 'claimed' + returning * into current_row; + + if found then + return jsonb_build_object( + 'outcome', 'completed', + 'proposal_json', current_row.proposal_json, + 'usage_input', current_row.usage_input, + 'usage_output', current_row.usage_output + ); + end if; + + select * + into current_row + from public.router_attempts + where user_id = target_user + and idempotency_key = idem_key; + + if current_row.status = 'completed' then + return jsonb_build_object( + 'outcome', 'completed', + 'proposal_json', current_row.proposal_json, + 'usage_input', current_row.usage_input, + 'usage_output', current_row.usage_output + ); + end if; + + raise exception 'router attempt for user % and key % is not claimed', target_user, idem_key + using errcode = '55000'; +end; +$$; + +-- Releases a claim immediately after a definitive provider failure, so a +-- retry does not have to wait out the lease to try again. +create function public.router_attempt_fail( + target_user uuid, + idem_key text +) +returns void +language plpgsql +security invoker +set search_path = public +as $$ +begin + update public.router_attempts + set status = 'failed' + where user_id = target_user + and idempotency_key = idem_key + and status = 'claimed'; +end; +$$; + +revoke execute on function public.router_attempt_claim(uuid, text, integer) from public, anon, authenticated; +grant execute on function public.router_attempt_claim(uuid, text, integer) to service_role; + +revoke execute on function public.router_attempt_complete(uuid, text, text, integer, integer) from public, anon, authenticated; +grant execute on function public.router_attempt_complete(uuid, text, text, integer, integer) to service_role; + +revoke execute on function public.router_attempt_fail(uuid, text) from public, anon, authenticated; +grant execute on function public.router_attempt_fail(uuid, text) to service_role; diff --git a/supabase/tests/database/router_attempt_claim.test.sql b/supabase/tests/database/router_attempt_claim.test.sql new file mode 100644 index 0000000..8ae5ca3 --- /dev/null +++ b/supabase/tests/database/router_attempt_claim.test.sql @@ -0,0 +1,184 @@ +begin; + +select plan(23); + +-- Only service_role can claim/complete/fail router attempts. +select ok( + not has_function_privilege('anon', 'public.router_attempt_claim(uuid, text, integer)', 'execute'), + 'anon cannot claim router attempts' +); +select ok( + not has_function_privilege('authenticated', 'public.router_attempt_claim(uuid, text, integer)', 'execute'), + 'authenticated cannot claim router attempts' +); +select ok( + has_function_privilege('service_role', 'public.router_attempt_claim(uuid, text, integer)', 'execute'), + 'service_role can claim router attempts' +); +select ok( + not has_function_privilege('anon', 'public.router_attempt_complete(uuid, text, text, integer, integer)', 'execute'), + 'anon cannot complete router attempts' +); +select ok( + has_function_privilege('service_role', 'public.router_attempt_complete(uuid, text, text, integer, integer)', 'execute'), + 'service_role can complete router attempts' +); +select ok( + not has_function_privilege('anon', 'public.router_attempt_fail(uuid, text)', 'execute'), + 'anon cannot fail router attempts' +); +select ok( + has_function_privilege('service_role', 'public.router_attempt_fail(uuid, text)', 'execute'), + 'service_role can fail router attempts' +); +select ok( + not has_table_privilege('anon', 'public.router_attempts', 'select'), + 'anon has no direct router_attempts privileges' +); +select ok( + not has_table_privilege('authenticated', 'public.router_attempts', 'select'), + 'authenticated has no direct router_attempts privileges' +); + +insert into auth.users ( + id, aud, role, email, raw_app_meta_data, raw_user_meta_data, is_anonymous, created_at, updated_at +) +values ( + '00000000-0000-0000-0000-000000000401', 'authenticated', 'authenticated', + 'router-attempt-claim@example.invalid', '{}'::jsonb, '{}'::jsonb, false, now(), now() +); + +-- The first claim for a key durably owns provider invocation. +select is( + public.router_attempt_claim('00000000-0000-0000-0000-000000000401', 'router:attempt-1'), + jsonb_build_object('outcome', 'claimed'), + 'the first claim for an absent key is durably claimed' +); + +-- A second claim for the same still-claimed key must not also claim it. +select is( + public.router_attempt_claim('00000000-0000-0000-0000-000000000401', 'router:attempt-1'), + jsonb_build_object('outcome', 'in_progress'), + 'a live claim is reported in_progress, never re-claimed' +); + +-- Recording the provider outcome makes it durably completed. +select is( + public.router_attempt_complete( + '00000000-0000-0000-0000-000000000401', 'router:attempt-1', + '{"action":"openProject"}', 42, 13 + ), + jsonb_build_object( + 'outcome', 'completed', + 'proposal_json', '{"action":"openProject"}', + 'usage_input', 42, + 'usage_output', 13 + ), + 'completing a claimed attempt durably records the provider outcome' +); + +-- Replaying completion on an already-completed attempt is idempotent. +select is( + public.router_attempt_complete( + '00000000-0000-0000-0000-000000000401', 'router:attempt-1', + '{"action":"different"}', 1, 1 + ), + jsonb_build_object( + 'outcome', 'completed', + 'proposal_json', '{"action":"openProject"}', + 'usage_input', 42, + 'usage_output', 13 + ), + 'completing an already-completed attempt is a no-op that returns the stored outcome' +); + +-- A later claim on a completed key returns the stored result instead of +-- reclaiming, so a retry can skip straight to a debit retry. +select is( + public.router_attempt_claim('00000000-0000-0000-0000-000000000401', 'router:attempt-1'), + jsonb_build_object( + 'outcome', 'completed', + 'proposal_json', '{"action":"openProject"}', + 'usage_input', 42, + 'usage_output', 13 + ), + 'claiming an already-completed key returns the stored outcome, not a new claim' +); + +-- A definitive provider failure releases the claim immediately. +select public.router_attempt_claim('00000000-0000-0000-0000-000000000401', 'router:attempt-2'); +select public.router_attempt_fail('00000000-0000-0000-0000-000000000401', 'router:attempt-2'); +select is( + public.router_attempt_claim('00000000-0000-0000-0000-000000000401', 'router:attempt-2'), + jsonb_build_object('outcome', 'claimed'), + 'a failed attempt can be re-claimed immediately, without waiting out a lease' +); + +-- A stale claim (past its lease) is recoverable. +update public.router_attempts +set claimed_at = now() - interval '1 hour' +where user_id = '00000000-0000-0000-0000-000000000401' + and idempotency_key = 'router:attempt-2'; +select is( + public.router_attempt_claim('00000000-0000-0000-0000-000000000401', 'router:attempt-2', 30), + jsonb_build_object('outcome', 'claimed'), + 'a claim past its lease is recoverable by a later caller' +); + +-- A claim still within its lease is not recoverable. +select is( + public.router_attempt_claim('00000000-0000-0000-0000-000000000401', 'router:attempt-2', 30), + jsonb_build_object('outcome', 'in_progress'), + 'a claim within its lease stays in_progress' +); + +-- Completing an attempt that was never claimed fails loudly. +select throws_ok( + $$select public.router_attempt_complete('00000000-0000-0000-0000-000000000401', 'router:never-claimed', 'x', 1, 1)$$, + '55000', + 'router attempt for user 00000000-0000-0000-0000-000000000401 and key router:never-claimed is not claimed', + 'completing an attempt with no matching claim raises' +); + +-- Per-user isolation: two users can independently claim the same literal key. +insert into auth.users ( + id, aud, role, email, raw_app_meta_data, raw_user_meta_data, is_anonymous, created_at, updated_at +) +values ( + '00000000-0000-0000-0000-000000000402', 'authenticated', 'authenticated', + 'router-attempt-claim-b@example.invalid', '{}'::jsonb, '{}'::jsonb, false, now(), now() +); +select is( + public.router_attempt_claim('00000000-0000-0000-0000-000000000402', 'router:attempt-1'), + jsonb_build_object('outcome', 'claimed'), + 'a different user can independently claim the same literal idempotency key' +); + +-- Input validation. +select throws_ok( + $$select public.router_attempt_claim('00000000-0000-0000-0000-000000000401', '')$$, + '22023', + 'idem_key must be a non-empty string', + 'claim rejects an empty idempotency key' +); +select throws_ok( + $$select public.router_attempt_claim('00000000-0000-0000-0000-000000000401', 'router:attempt-3', 0)$$, + '22023', + 'lease_seconds must be a positive integer', + 'claim rejects a non-positive lease' +); +select throws_ok( + $$select public.router_attempt_complete('00000000-0000-0000-0000-000000000401', 'router:attempt-1', '', 1, 1)$$, + '22023', + 'new_proposal_json must be a non-empty string', + 'complete rejects an empty proposal_json' +); +select throws_ok( + $$select public.router_attempt_complete('00000000-0000-0000-0000-000000000401', 'router:attempt-1', 'x', -1, 1)$$, + '22023', + 'new_usage_input and new_usage_output must be non-negative integers', + 'complete rejects negative usage' +); + +select finish(); +rollback;