diff --git a/docs/architecture.md b/docs/architecture.md
index 81951378e..7f4a0175a 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -78,6 +78,10 @@ Client-side environment selection is a developer convenience, not authorization.
Every structured answer is shown while the model writes it. A page passes a `live` callback to its feature wrapper, and the request adds `stream: true`. Every check before the feature handler (authentication, input, debate context, quota, model routing, source ranking) still answers with plain JSON and its real status. After that, [`supabase/functions/ai/stream.ts`](../supabase/functions/ai/stream.ts) answers with server-sent events: `phase` (`research`, `writing`, `checking`), `delta` with the model's raw text, `reset` when a JSON repair retry starts the answer again, and one final `result` holding the status and body a plain request returns. Only the call whose output the user sees streams; research sub-calls and Jev decisions do not. The joined text is parsed exactly as a non-streamed reply, and streamed usage is billed the same way, marked incomplete if the stream breaks. On the client, [`src/lib/ai/live.ts`](../src/lib/ai/live.ts) reads the stream, [`packages/core/src/ai/partial-json.ts`](../packages/core/src/ai/partial-json.ts) turns the JSON written so far into a draft, and the page renders it with the result's own labels. A JSON body is accepted too, so the client works against a server from before live generation.
+A streamed run keeps its worker alive with `EdgeRuntime.waitUntil` ([`_shared/background.ts`](../supabase/functions/_shared/background.ts)) from the first event until the HTTP layer has read the last one. Supabase otherwise counts the response as sent once its headers are out and may retire the worker mid-run (`EarlyDrop`). Streamed Ask (`ai-stream`) is held the same way until the browser has read its final event. A run still unanswered about 142 s after its request arrived is aborted and gets a retryable timeout `result` instead of a connection cut at the 150 s wall clock; when the aborted run answers within 2 s, that answer is sent instead, so it says `charged: false` when the units went back. Storing the result and waiting for the browser to read it are cut short to finish before the wall clock. The request sends its id as `resultKey`. The function then writes a row to `ai_run_results` when the run starts, and stores the `result` there before sending it ([`ai/run-results.ts`](../supabase/functions/ai/run-results.ts); one row per user and feature, readable only by its owner). When the stream drops, stalls (no keepalive for 45 s), or delivers a result it cannot read, [`src/lib/ai/run-recovery.ts`](../src/lib/ai/run-recovery.ts) polls that row until the run's result appears, then resolves or fails exactly as if the event had arrived. A row still without a status long after it started is a run its worker never finished. A browser Stop, reload, or closed tab still cancels the run. The server cannot tell those from a lost connection, so it stops spending as soon as the connection goes.
+
+A reply that hits its token cap is written again once with twice the cap, after a `reset`. The Generate cap (12,000 tokens) leaves room for a full comprehensive case on Claude Haiku, so a long case finishes on its first attempt. Claude rejects an assistant prefill, so a cut-off reply cannot be continued where it stopped. Sparring calls send a strict `json_schema` (`sparringOutputSchema` in `ai/sparring-persistence.ts`). Anthropic's structured outputs take no numeric or array-size limits, so the 0-25 range, the four categories, and the 12-item lists are enforced by the parser. When a reply parses but has the wrong shape, the log records only the failing field path, never the model's text. The call is then retried once with a repair prompt that names the field. A second failure answers `invalid_output` (502, logged as `invalid_output`), which the browser shows as a malformed answer. The judge also retries once when Claude is briefly unavailable or its stream breaks. The round has already paid, so retries cost the student nothing.
+
For a validated Crossfire next turn, the server returns the reply first with a short-lived signed sidecar token. The client then requests a Jev classification of only the newest completed question-answer exchange. The token is atomically claimed through its own one-use counter, separate from the product's daily AI quota. The optional directness, concession, and follow-up signal is attached by turn ID to the answer; a missing, slow, replayed, forged, or failed sidecar never delays or invalidates the main reply. Draft version 1 remains compatible because turn signals are optional.
### Voice rounds
@@ -154,11 +158,11 @@ Every model call asks for JSON. Calls whose shape the server checks in code (the
`Auto Fast` and `Auto Pro` are virtual selections, not provider model IDs. Auto routes only to Claude, which runs on DebateLab's Anthropic key (BYOK in OpenRouter); other families bill OpenRouter credits, so they run only when a student picks one by hand. Auto Fast is available to Basic accounts and resolves to Claude Haiku 5.5. Auto Pro remains Pro-only and resolves to Claude Sonnet 5.5; it never routes or falls back to Opus. With one candidate per pool the Edge Functions skip the TypeSafe Jev routing call; a pool with more than one candidate (all `anthropic/`, which `src/test/ai/anthropic-only-guard.test.ts` enforces) is resolved through Jev on OpenRouter before calling the concrete writing model. Provider calls, usage rows, and response metadata record the concrete resolved model, while response metadata also retains the requested automatic selection. A Jev router call uses the same OpenRouter key and does not consume a second daily quota unit.
-The daily AI quota counts units, not requests. `checkRateLimit` in [`supabase/functions/_shared/entitlements.ts`](../supabase/functions/_shared/entitlements.ts) reserves the requested model's weight through `consume_ai_quota` (Claude Opus 5.5 counts as 5, Sonnet as 2, cheap models as 1; Auto Pro counts as its most expensive candidate because the quota is reserved before routing). The same call refuses the request with `weekly_budget_reached` once the user's weekly AI spend reaches the weekly budget ($2 by default, calendar week from Monday 00:00 UTC). Weekly spend is the sum of `ai_requests.cap_cost_usd`: each request's full logged charge, including Claude calls billed to our own BYOK Anthropic key, plus Exa and ElevenLabs charges. A BYOK response without the upstream charge counts OpenRouter's fee, the part that is known. When part of a request's cost is unknown, the known part counts; rows with no known cap spend, and rows logged before the column existed, do not count. Cost is logged after the provider answers, so the cap can be passed by about one request. The weights, the budget, and the tier defaults live in [`supabase/functions/_shared/usage-limits.ts`](../supabase/functions/_shared/usage-limits.ts). Voice, import, share, and Crossfire signal quotas keep calling `consume_daily_quota`, one unit per call, and are not gated by the weekly budget (voice spend still counts toward it).
+The daily AI quota counts units, not requests. `checkRateLimit` in [`supabase/functions/_shared/entitlements.ts`](../supabase/functions/_shared/entitlements.ts) reserves the requested model's weight through `consume_ai_quota` (Claude Opus 5.5 counts as 5, Sonnet as 2, cheap models as 1; an Auto selection counts as its most expensive candidate because the quota is reserved before routing. The Auto weights are derived from the candidate pools, so Auto Pro counts as Sonnet's 2 while Sonnet is its only candidate). The same call refuses the request with `weekly_budget_reached` once the user's weekly AI spend reaches the weekly budget ($2 by default, calendar week from Monday 00:00 UTC). Weekly spend is the sum of `ai_requests.cap_cost_usd`: each request's full logged charge, including Claude calls billed to our own BYOK Anthropic key, plus Exa and ElevenLabs charges. A BYOK response without the upstream charge counts OpenRouter's fee, the part that is known. When part of a request's cost is unknown, the known part counts; rows with no known cap spend, and rows logged before the column existed, do not count. Cost is logged after the provider answers, so the cap can be passed by about one request. The weights, the budget, and the tier defaults live in [`supabase/functions/_shared/usage-limits.ts`](../supabase/functions/_shared/usage-limits.ts). Voice, import, share, and Crossfire signal quotas keep calling `consume_daily_quota`, one unit per call, and are not gated by the weekly budget (voice spend still counts toward it).
Crossfire and Sparring are charged per round, not per call. The round's first call reserves the model's units through `claim_ai_session_quota` ([`20261022220000_ai_quota_sessions.sql`](../supabase/migrations/20261022220000_ai_quota_sessions.sql)), keyed by the Sparring `sessionId` or the Crossfire round id the client sends as `quotaSession` beside `input`; later turns, the debrief, and the judge in that round reserve nothing. A round past its call cap or lifetime, or switched to a heavier model, pays again (`SESSION_QUOTA` in `usage-limits.ts`), so a reused id cannot buy unlimited AI. A Crossfire request without `quotaSession` (an older client) is charged per call. Before a round starts, both pages send a `quotaCheck` request, which reserves nothing and answers with the round's cost and today's count, so a round the server would refuse never starts.
-A run that produces nothing costs nothing. When a request fails, times out, or the browser goes away (reload, closed tab, Stop) before `REFUND_MAX_DELIVERED_CHARS` of streamed output reached it and before its answer was saved, `ai` and `ai-stream` give its units back (`refund_daily_quota`, or `refund_ai_session_quota` for the call that paid for a round) and answer with `charged: false`, which the client shows as "You were not charged". The weekly cap still counts what the run cost. The usage line in Settings and on every tool page reads the same counter through `ai_usage_today()` (the caller's own units today), so per-round charging and refunds show there, and it is read again after each AI call (`refreshAIUsage` in `src/lib/query-client.ts`). When the counter cannot be read (an older database), the line is hidden rather than estimated from `ai_requests` rows. `ai_requests.error_code` holds a short code (`provider_error`, `timeout`, `aborted`, `claude_unavailable`, the refusal codes); `logUsage` maps anything else to one of those and sends the raw text, truncated, to the function log only.
+A run that produces nothing costs nothing. When an `ai` request fails on the server (a provider failure, a reply cut off at its cap, an answer that failed validation, the time budget running out) and its answer was not saved, the function gives its units back, however much of it had streamed. When the browser goes away (reload, closed tab, Stop, a dropped connection; the server cannot tell these apart), the run is refunded only before `REFUND_MAX_DELIVERED_CHARS` of streamed output reached the student and before the model finished its answer. Otherwise a student could read most of an answer and press Stop to get the unit back. For the plain-text Ask stream the streamed text is the answer, so `ai-stream` refunds only before `REFUND_MAX_DELIVERED_CHARS` reached the reader, whatever ended the run. A refund goes through `refund_daily_quota`, or `refund_ai_session_quota` for the call that paid for a round, at most once per run, and the answer carries `charged: false`, which the client shows as "You were not charged". The weekly cap still counts what the run cost. The usage line in Settings and on every tool page reads the same counter through `ai_usage_today()` (the caller's own units today), so per-round charging and refunds show there, and it is read again after each AI call (`refreshAIUsage` in `src/lib/query-client.ts`). When the counter cannot be read (an older database), the line is hidden rather than estimated from `ai_requests` rows. `ai_requests.error_code` holds a short code (`provider_error`, `timeout`, `aborted`, `claude_unavailable`, the refusal codes); `logUsage` maps anything else to one of those and sends the raw text, truncated, to the function log only.
## Observability
diff --git a/src/components/features/sparring/SparringRound.tsx b/src/components/features/sparring/SparringRound.tsx
index 47f1eb044..797a543d8 100644
--- a/src/components/features/sparring/SparringRound.tsx
+++ b/src/components/features/sparring/SparringRound.tsx
@@ -167,7 +167,7 @@ export function SparringRound({
{t(`sparring.format.${session.format}`, session.format.toUpperCase())}
- {t('sparring.round.inProgress', 'Round in progress')}
+ {judging && judgingError ? t('sparring.round.scoringStopped', 'Scoring stopped') : t('sparring.round.inProgress', 'Round in progress')}
·
{session.userSide === 'pro' ? t('sparring.round.youArePro', 'You are Pro') : t('sparring.round.youAreCon', 'You are Con')}
@@ -188,7 +188,7 @@ export function SparringRound({
{judging ? t('sparring.round.complete', 'Round complete') : currentTurn ? t('sparring.round.turnOf', 'Turn {{current}} of {{total}}', { current: session.currentTurnIndex + 1, total: session.turns.length }) : t('sparring.round.preparingResult', 'Preparing your result')}
-
{judging ? t('sparring.round.judgeReading', 'The judge is reading the transcript.') : currentTurn ? t(`sparring.turns.${currentTurn.id}`, currentTurn.label) : t('sparring.round.scoring', 'Scoring the round')}
+
{judging ? (judgingError ? t('sparring.round.judgeFailed', 'The judge could not finish scoring.') : t('sparring.round.judgeReading', 'The judge is reading the transcript.')) : currentTurn ? t(`sparring.turns.${currentTurn.id}`, currentTurn.label) : t('sparring.round.scoring', 'Scoring the round')}
diff --git a/src/lib/ai/client.ts b/src/lib/ai/client.ts
index db7b827c1..15512baf8 100644
--- a/src/lib/ai/client.ts
+++ b/src/lib/ai/client.ts
@@ -15,6 +15,7 @@ export {
isAINetworkError,
isAIRateLimitError,
isAIProviderError,
+ isInvalidOutputError,
isTimeoutError,
isAISubscriptionError,
isAIModelRestrictionError,
diff --git a/src/lib/ai/errors.test.ts b/src/lib/ai/errors.test.ts
index 12a47f9dd..899822fd4 100644
--- a/src/lib/ai/errors.test.ts
+++ b/src/lib/ai/errors.test.ts
@@ -1,4 +1,5 @@
import { describe, expect, it } from 'vitest';
+import i18n from '@/i18n';
import {
AIAuthError,
AIModelRestrictionError,
@@ -37,6 +38,25 @@ describe('classifyAIError', () => {
expect(classifyAIError(new TimeoutError())).toMatchObject({ errorName: 'TimeoutError', shouldCaptureException: true });
});
+ it('reads invalid_output as a malformed answer, not a provider failure', async () => {
+ const { isInvalidOutputError } = await import('./errors');
+ const error = mapAIError({ message: 'failed', context: { status: 502 } }, 'req_1', 'openrouter', 'auto-pro', { error: 'The AI answer came back malformed. Please try again.', code: 'invalid_output', charged: false });
+ expect(error.name).toBe('AIResponseError');
+ expect(isInvalidOutputError(error)).toBe(true);
+ expect(error.notCharged).toBe(true);
+ expect(error.message).toBe(i18n.t('aiErrors.invalidOutput'));
+ });
+
+ it('keeps transient provider failures out of exception reports but still reports the rest', () => {
+ const mapped = (payload: Record
, status: number) =>
+ classifyAIError(mapAIError({ message: 'failed', context: { status } }, 'req_1', 'openrouter', 'auto-pro', payload));
+ // A Claude outage, and a failure the server marked worth retrying.
+ expect(mapped({ error: 'Anthropic Claude is temporarily unavailable.', code: 'provider_unavailable' }, 503)).toMatchObject({ errorName: 'AIProviderError', shouldCaptureException: false });
+ expect(mapped({ error: 'AI request failed. Please try again.', code: 'provider_error', retryable: true }, 500)).toMatchObject({ errorName: 'AIProviderError', shouldCaptureException: false });
+ // An answer that failed validation may be a bug.
+ expect(mapped({ error: 'AI request failed. Please try again.', code: 'provider_error' }, 500)).toMatchObject({ errorName: 'AIProviderError', shouldCaptureException: true });
+ });
+
it('recognizes the other browser network messages and an offline navigator', () => {
expect(classifyAIError(new TypeError('NetworkError when attempting to fetch resource')).errorName).toBe('AINetworkError');
expect(classifyAIError(new TypeError('Load failed')).errorName).toBe('AINetworkError');
diff --git a/src/lib/ai/errors.ts b/src/lib/ai/errors.ts
index 755c429ec..8f78731f9 100644
--- a/src/lib/ai/errors.ts
+++ b/src/lib/ai/errors.ts
@@ -229,6 +229,18 @@ export function isAIProviderError(error: unknown): error is AIProviderError {
return error instanceof AIProviderError || (error instanceof Error && error.name === 'AIProviderError');
}
+/** The server could not use the model's answer (`invalid_output`). */
+export function isInvalidOutputError(error: unknown): boolean {
+ return error instanceof AIResponseError && error.details?.code === 'invalid_output';
+}
+
+/** A provider failure the server marked worth retrying as is (an outage, a dropped stream, the time budget). */
+export function isTransientProviderError(error: unknown): boolean {
+ if (!isAIProviderError(error)) return false;
+ const details = (error as AIError).details;
+ return details?.retryable === true || details?.code === 'provider_unavailable';
+}
+
export function isTimeoutError(error: unknown): error is TimeoutError {
return error instanceof TimeoutError || (error instanceof Error && error.name === 'TimeoutError');
}
@@ -263,7 +275,10 @@ export function classifyAIError(error: unknown, userInitiatedAbort = false): AIE
if (error instanceof AIResponseError || (error instanceof Error && error.name === 'AIResponseError')) {
return { errorName: 'AIResponseError', statusCode, shouldCaptureException: true };
}
- if (isAIProviderError(error)) return { errorName: 'AIProviderError', statusCode, shouldCaptureException: true };
+ // An outage or a dropped provider stream is expected now and then and the
+ // student is told to try again; ai_request_failed still records it. Other
+ // provider failures may be bugs and are reported.
+ if (isAIProviderError(error)) return { errorName: 'AIProviderError', statusCode, shouldCaptureException: !isTransientProviderError(error) };
if (error instanceof AIServerError || (error instanceof Error && error.name === 'AIServerError')) {
return { errorName: 'AIServerError', statusCode, shouldCaptureException: true };
}
@@ -378,12 +393,18 @@ function mapAIErrorPayload(
});
}
+ // The model answered, but in a shape the feature could not use, even after
+ // the server's repair retry. Trying again usually works.
+ if (errorCode === 'invalid_output') {
+ return new AIResponseError(i18n.t('aiErrors.invalidOutput'), { requestId, details: { code: errorCode } });
+ }
+
if (isProviderErrorMessage(message) || isProviderErrorCode(errorCode)) {
return new AIProviderError(message, {
statusCode: statusCode || 500,
requestId,
provider,
- details: { model, code: errorCode },
+ details: { model, code: errorCode, ...(payload?.retryable === true ? { retryable: true } : {}) },
});
}
diff --git a/src/lib/ai/live.ts b/src/lib/ai/live.ts
index 4e5fd5be4..e15818f09 100644
--- a/src/lib/ai/live.ts
+++ b/src/lib/ai/live.ts
@@ -24,6 +24,57 @@ const DRAFT_INTERVAL_MS = 50;
*/
export const LIVE_STREAM_DRAIN_MS = 2_000;
+/**
+ * The function sends a keepalive every 15 s. Three missed in a row means the
+ * stream is gone even if the connection still looks open, as when a proxy
+ * holds the tail of a response.
+ */
+export const LIVE_STREAM_STALL_MS = 45_000;
+
+/**
+ * A stalled stream is left open, not cancelled: closing it would make the
+ * function cancel the run whose stored result the caller is about to read.
+ * It is cancelled after this long, past any run's wall clock, or when the
+ * caller aborts.
+ */
+export const LIVE_STREAM_STALLED_OPEN_MS = 160_000;
+
+/**
+ * Why a live stream ended without its answer: the body ended, the
+ * connection dropped, nothing arrived for LIVE_STREAM_STALL_MS, or the
+ * `result` event arrived but could not be read. The caller can still recover
+ * the stored result (run-recovery.ts) before it reports the failure.
+ */
+export type LiveStreamFailure = 'ended' | 'dropped' | 'stalled' | 'unreadable';
+
+export class LiveStreamError extends Error {
+ readonly reason: LiveStreamFailure;
+ /** The read error behind a dropped stream. */
+ readonly cause?: unknown;
+
+ constructor(reason: LiveStreamFailure, cause?: unknown) {
+ super(reason === 'unreadable' ? 'The answer arrived but could not be read.' : 'The response ended before the answer was complete.');
+ this.name = 'LiveStreamError';
+ this.reason = reason;
+ this.cause = cause;
+ }
+}
+
+/** Reads one chunk, or 'stalled' when nothing arrives in time. */
+async function readWithin(reader: ReadableStreamDefaultReader, ms: number): Promise | 'stalled'> {
+ let timer: ReturnType | undefined;
+ try {
+ return await Promise.race([
+ reader.read(),
+ new Promise<'stalled'>((resolve) => {
+ timer = setTimeout(() => resolve('stalled'), ms);
+ }),
+ ]);
+ } finally {
+ if (timer) clearTimeout(timer);
+ }
+}
+
function isPhase(value: unknown): value is LivePhase {
return value === 'research' || value === 'writing' || value === 'checking';
}
@@ -83,6 +134,7 @@ export async function readLiveResponse(
let shown = '';
let timer: ReturnType | null = null;
let result: { status: number; body: unknown } | null = null;
+ let unreadableResult = false;
const showDraft = () => {
if (timer) clearTimeout(timer);
@@ -112,8 +164,11 @@ export async function readLiveResponse(
// cannot arrive after it and look like writing again.
showDraft();
live.onPhase?.(payload.phase);
- } else if (event === 'result' && typeof payload.status === 'number') {
- result = { status: payload.status, body: payload.body };
+ } else if (event === 'result') {
+ // A result that is not the {status, body} the function sends was
+ // damaged on the way; the stored copy may still be read.
+ if (typeof payload.status === 'number') result = { status: payload.status, body: payload.body };
+ else unreadableResult = true;
}
}
};
@@ -129,27 +184,48 @@ export async function readLiveResponse(
// before the server closes, aborts the fetch and drops the body.
let streamEnded = false;
let draining = false;
+ let stalled = false;
const release = () => {
- signal?.removeEventListener('abort', stopReading);
if (timer) clearTimeout(timer);
timer = null;
+ if (stalled) {
+ // Keep the connection, so the run keeps going and stores its result.
+ setTimeout(stopReading, LIVE_STREAM_STALLED_OPEN_MS);
+ return;
+ }
+ signal?.removeEventListener('abort', stopReading);
if (!streamEnded) reader.cancel().catch((): undefined => undefined);
};
+ let failure: LiveStreamError | null = null;
try {
- while (!result) {
+ while (!result && !failure) {
if (signal?.aborted) throw new DOMException('The request was canceled.', 'AbortError');
- const { done, value } = await reader.read();
+ let chunk: Awaited>;
+ try {
+ chunk = await readWithin(reader, LIVE_STREAM_STALL_MS);
+ } catch (error) {
+ if (signal?.aborted) throw new DOMException('The request was canceled.', 'AbortError');
+ // The connection dropped (ERR_CONNECTION_CLOSED, network change).
+ failure = new LiveStreamError('dropped', error);
+ break;
+ }
if (signal?.aborted) throw new DOMException('The request was canceled.', 'AbortError');
- if (done) {
+ if (chunk === 'stalled') {
+ stalled = true;
+ failure = new LiveStreamError('stalled');
+ break;
+ }
+ if (chunk.done) {
streamEnded = true;
break;
}
- const parsed = consumeSSEChunk(buffer, decoder.decode(value, { stream: true }));
+ const parsed = consumeSSEChunk(buffer, decoder.decode(chunk.value, { stream: true }));
buffer = parsed.buffer;
handle(parsed.events);
}
- if (!result) handle(consumeSSEChunk('', `${buffer}${decoder.decode()}\n\n`).events);
+ // A last event without its closing blank line still counts.
+ if (!result && !failure) handle(consumeSSEChunk('', `${buffer}${decoder.decode()}\n\n`).events);
// Hand the answer back before the tail, so a slow close cannot delay the UI.
if (result && !streamEnded) {
@@ -168,7 +244,7 @@ export async function readLiveResponse(
}
const final = result as { status: number; body: unknown } | null;
- if (!final) return new FunctionsFetchError(new Error('The response ended before the answer was complete.'));
+ if (!final) return new FunctionsFetchError(failure ?? new LiveStreamError(unreadableResult ? 'unreadable' : 'ended'));
if (final.status >= 200 && final.status < 300) return final.body as T;
// The same error a plain request would have produced, so error mapping
// (limits, model access, provider failures) does not change.
diff --git a/src/lib/ai/request.ts b/src/lib/ai/request.ts
index be92b61ac..6c6f79b21 100644
--- a/src/lib/ai/request.ts
+++ b/src/lib/ai/request.ts
@@ -36,7 +36,9 @@ import {
} from './errors';
import { parseSourceAssessments } from './sources';
import { utf8ByteLength } from './text';
-import { readLiveResponse, type LiveGeneration } from './live';
+import { LiveStreamError, readLiveResponse, type LiveGeneration } from './live';
+import { recoverRunResult } from './run-recovery';
+import { FunctionsFetchError, FunctionsHttpError } from '@supabase/supabase-js';
import { refreshAIUsage } from '@/lib/query-client';
// Request timeout in milliseconds (200 seconds)
@@ -161,7 +163,8 @@ export async function callSupabaseAI(
config: ProviderConfig,
options: AIRequestOptions = {},
): Promise {
- const requestId = `req_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`;
+ const startedAt = Date.now();
+ const requestId = `req_${startedAt}_${Math.random().toString(36).slice(2, 8)}`;
const startTime = performance.now();
const session = useSessionStore.getState();
@@ -245,10 +248,12 @@ export async function callSupabaseAI(
}
const live = options.live;
- const { data, error } = live
+ let { data, error } = live
? await invokeAuthenticatedFunctionWith>('ai', {
headers: { Authorization: `Bearer ${accessToken}` },
- body: { ...requestBody, stream: true },
+ // The request id names the run's stored result, so a dropped stream
+ // can be recovered below. Servers from before it ignore the field.
+ body: { ...requestBody, stream: true, resultKey: requestId },
signal: controller.signal,
}, (response) => readLiveResponse>(response, live, controller.signal))
: await invokeAuthenticatedFunction>('ai', {
@@ -257,6 +262,28 @@ export async function callSupabaseAI(
signal: controller.signal,
});
+ // The stream dropped before its answer arrived. The server keeps running
+ // while the connection lasts and stores the final result before sending
+ // it, so the answer may be waiting even though the browser never got it.
+ if (live && error instanceof FunctionsFetchError && error.context instanceof LiveStreamError && !browserIsOffline()) {
+ const recovered = await recoverRunResult(feature, requestId, startedAt, controller.signal, error.context.reason === 'stalled' ? Infinity : undefined);
+ if (recovered && recovered.status >= 200 && recovered.status < 300) {
+ captureProductEvent('ai_stream_recovered', { feature, durationMs: Math.round(performance.now() - startTime) });
+ data = recovered.body as Record;
+ error = null;
+ } else if (recovered) {
+ error = new FunctionsHttpError(new Response(JSON.stringify(recovered.body ?? {}), {
+ status: recovered.status,
+ headers: { 'Content-Type': 'application/json' },
+ }));
+ } else {
+ // Not a request the browser could not send: the stream broke on the way.
+ throw error.context.reason === 'unreadable'
+ ? new AIResponseError('The streamed answer could not be read', { requestId })
+ : new AINetworkError(offlineAIMessage(), { requestId });
+ }
+ }
+
const elapsed = Math.round(performance.now() - startTime);
// Log response in development
@@ -313,7 +340,7 @@ export async function callSupabaseAI(
throw new AIProviderError(errorMessage, {
requestId,
provider,
- details: { model: config.model, code: errorCode },
+ details: { model: config.model, code: errorCode, ...(data.retryable === true ? { retryable: true } : {}) },
});
}
diff --git a/src/lib/ai/run-recovery.test.ts b/src/lib/ai/run-recovery.test.ts
new file mode 100644
index 000000000..d13a4782b
--- /dev/null
+++ b/src/lib/ai/run-recovery.test.ts
@@ -0,0 +1,134 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { FunctionsFetchError } from '@supabase/supabase-js';
+import { LiveStreamError } from '@/lib/ai/live';
+import { RUN_RESULT_MISSING_MS, RUN_RESULT_POLL_MS, RUN_RESULT_RUNNING_MS } from '@/lib/ai/run-recovery';
+
+// A live request whose stream drops: the answer the server stored is read
+// back by the id the request sent, instead of failing with "check your
+// connection".
+
+const { invokeWith, rows, queries } = vi.hoisted(() => ({
+ invokeWith: vi.fn(),
+ /** What each successive read of ai_run_results returns. */
+ rows: [] as Array | null>,
+ queries: [] as Array<{ table: string; filters: Record }>,
+}));
+
+vi.mock('@/lib/observability', () => ({ captureException: vi.fn(), captureProductEvent: vi.fn() }));
+vi.mock('@/lib/query-client', () => ({ refreshAIUsage: vi.fn() }));
+vi.mock('@/lib/supabase/client', () => ({
+ getActiveAccessToken: () => 'token',
+ SUPABASE_URL: 'http://test-supabase.local',
+ invokeAuthenticatedFunction: vi.fn(),
+ invokeAuthenticatedFunctionWith: invokeWith,
+ supabase: {
+ auth: { getSession: vi.fn() },
+ from: (table: string) => {
+ const query = { table, filters: {} as Record };
+ queries.push(query);
+ const chain = {
+ select: () => chain,
+ eq: (column: string, value: unknown) => {
+ query.filters[column] = value;
+ return chain;
+ },
+ maybeSingle: async (): Promise<{ data: Record | null; error: null }> => ({ data: (rows.length > 1 ? rows.shift() : rows[0]) ?? null, error: null }),
+ };
+ return chain;
+ },
+ },
+}));
+
+const { callSupabaseAI } = await import('@/lib/ai/request');
+const { AINetworkError, AIProviderError, AIResponseError } = await import('@/lib/ai/errors');
+
+const config = { provider: 'openrouter' as const, model: 'auto-fast' };
+const live = { onDraft: () => {} };
+
+function dropStream(reason: 'dropped' | 'unreadable' = 'dropped') {
+ invokeWith.mockImplementation(async () => ({ data: null, error: new FunctionsFetchError(new LiveStreamError(reason)) }));
+}
+
+const sentResultKey = () => (invokeWith.mock.calls[0][1] as { body: { resultKey?: string } }).body.resultKey;
+
+beforeEach(() => {
+ invokeWith.mockReset();
+ rows.length = 0;
+ queries.length = 0;
+});
+
+afterEach(() => {
+ vi.useRealTimers();
+});
+
+describe('a live request whose stream dropped', () => {
+ it('returns the answer the server stored for it', async () => {
+ invokeWith.mockImplementationOnce(async (_name: string, options: { body: { resultKey: string } }) => {
+ // The server finished and stored this run's result; the stream never delivered it.
+ rows.push({ result_key: options.body.resultKey, status: 200, body: { output: { thesis: 'Kept' } } });
+ return { data: null, error: new FunctionsFetchError(new LiveStreamError('dropped')) };
+ });
+ const result = await callSupabaseAI('generate', { motion: 'Homework' }, config, { live, debateId: null });
+ expect(result).toEqual({ output: { thesis: 'Kept' } });
+ expect(sentResultKey()).toMatch(/^req_\d+_[a-z0-9]+$/);
+ expect(queries[0]).toMatchObject({ table: 'ai_run_results', filters: { feature: 'generate' } });
+ });
+
+ it('waits for a run that is still writing, then returns its answer', async () => {
+ vi.useFakeTimers();
+ let key = '';
+ invokeWith.mockImplementation(async (_name: string, options: { body: { resultKey: string } }) => {
+ key = options.body.resultKey;
+ rows.push({ result_key: key, status: null, body: null }, { result_key: key, status: 200, body: { output: { done: true } } });
+ return { data: null, error: new FunctionsFetchError(new LiveStreamError('stalled')) };
+ });
+ const pending = callSupabaseAI('generate', { motion: 'Homework' }, config, { live, debateId: null });
+ await vi.advanceTimersByTimeAsync(RUN_RESULT_POLL_MS);
+ await expect(pending).resolves.toEqual({ output: { done: true } });
+ });
+
+ it('gives up on a dropped run whose row never finishes, well before the wall clock', async () => {
+ vi.useFakeTimers();
+ invokeWith.mockImplementation(async (_name: string, options: { body: { resultKey: string } }) => {
+ rows.push({ result_key: options.body.resultKey, status: null, body: null });
+ return { data: null, error: new FunctionsFetchError(new LiveStreamError('dropped')) };
+ });
+ let settled: unknown = 'pending';
+ const pending = callSupabaseAI('generate', { motion: 'Homework' }, config, { live, debateId: null }).catch((caught: unknown) => caught);
+ void pending.then((value) => { settled = value; });
+ await vi.advanceTimersByTimeAsync(RUN_RESULT_RUNNING_MS - RUN_RESULT_POLL_MS);
+ expect(settled).toBe('pending');
+ await vi.advanceTimersByTimeAsync(2 * RUN_RESULT_POLL_MS);
+ expect(await pending).toBeInstanceOf(AINetworkError);
+ });
+
+ it('reports the stored failure, including that it was not charged', async () => {
+ invokeWith.mockImplementation(async (_name: string, options: { body: { resultKey: string } }) => {
+ rows.push({ result_key: options.body.resultKey, status: 500, body: { error: 'AI request failed. Please try again.', code: 'provider_error', charged: false } });
+ return { data: null, error: new FunctionsFetchError(new LiveStreamError('dropped')) };
+ });
+ const error = await callSupabaseAI('generate', { motion: 'Homework' }, config, { live, debateId: null }).catch((caught: unknown) => caught);
+ expect(error).toBeInstanceOf(AIProviderError);
+ expect((error as InstanceType).notCharged).toBe(true);
+ });
+
+ it('is a connection failure when nothing was stored, and a damaged answer when the result was unreadable', async () => {
+ vi.useFakeTimers();
+ dropStream('dropped');
+ const pending = callSupabaseAI('generate', { motion: 'Homework' }, config, { live, debateId: null }).catch((caught: unknown) => caught);
+ await vi.advanceTimersByTimeAsync(RUN_RESULT_MISSING_MS + RUN_RESULT_POLL_MS);
+ expect(await pending).toBeInstanceOf(AINetworkError);
+
+ dropStream('unreadable');
+ const damaged = callSupabaseAI('generate', { motion: 'Homework' }, config, { live, debateId: null }).catch((caught: unknown) => caught);
+ await vi.advanceTimersByTimeAsync(RUN_RESULT_MISSING_MS + RUN_RESULT_POLL_MS);
+ expect(await damaged).toBeInstanceOf(AIResponseError);
+ });
+
+ it('does not take another run\'s stored answer', async () => {
+ rows.push({ result_key: 'req_1_someoneelse', status: 200, body: { output: { thesis: 'Other tab' } } });
+ dropStream();
+ const error = await callSupabaseAI('generate', { motion: 'Homework' }, config, { live, debateId: null }).catch((caught: unknown) => caught);
+ expect(error).toBeInstanceOf(AINetworkError);
+ });
+});
diff --git a/src/lib/ai/run-recovery.ts b/src/lib/ai/run-recovery.ts
new file mode 100644
index 000000000..4e700c75a
--- /dev/null
+++ b/src/lib/ai/run-recovery.ts
@@ -0,0 +1,83 @@
+// Recovers a live run's answer after its stream dropped. The ai function
+// keeps each streamed run's final result in ai_run_results under the id the
+// browser sent (supabase/functions/ai/run-results.ts), writing it before it
+// sends the `result` event, so an answer the browser did not receive can
+// still be read.
+
+import { supabase } from '@/lib/supabase/client';
+
+export interface RecoveredRunResult {
+ status: number;
+ body: unknown;
+}
+
+/** No run answers past Supabase's 150 s wall clock, counted from the request. */
+export const RUN_RESULT_WAIT_MS = 155_000;
+export const RUN_RESULT_POLL_MS = 2_500;
+/**
+ * A run the server started has a row within moments. With none after this
+ * long, the request never reached a server that keeps results.
+ */
+export const RUN_RESULT_MISSING_MS = 8_000;
+/**
+ * A run that loses its browser is cancelled and stores its outcome within
+ * seconds, and one still running stores it when it ends. A row still without
+ * a result this long after the drop is a run whose worker is gone. A stalled
+ * stream is left open (live.ts), so its run is still writing and stores its
+ * result when it ends; it waits the whole window instead.
+ */
+export const RUN_RESULT_RUNNING_MS = 30_000;
+
+function wait(ms: number, signal?: AbortSignal): Promise {
+ return new Promise((resolve, reject) => {
+ if (signal?.aborted) {
+ reject(new DOMException('The request was canceled.', 'AbortError'));
+ return;
+ }
+ const timer = setTimeout(() => {
+ signal?.removeEventListener('abort', onAbort);
+ resolve();
+ }, ms);
+ const onAbort = () => {
+ clearTimeout(timer);
+ reject(new DOMException('The request was canceled.', 'AbortError'));
+ };
+ signal?.addEventListener('abort', onAbort, { once: true });
+ });
+}
+
+/**
+ * Waits for the stored result of the run named `resultKey`, which started at
+ * `startedAt` (epoch ms). Resolves with it, or null when there is none to
+ * wait for: no row (an older server, or a request that never arrived), a
+ * newer run of the same feature took the row, the table cannot be read, or
+ * the run outlived any server. Rejects only when `signal` aborts.
+ */
+export async function recoverRunResult(
+ feature: string,
+ resultKey: string,
+ startedAt: number,
+ signal?: AbortSignal,
+ runningGraceMs = RUN_RESULT_RUNNING_MS,
+ now: () => number = Date.now,
+): Promise {
+ const recoveryStartedAt = now();
+ while (true) {
+ if (signal?.aborted) throw new DOMException('The request was canceled.', 'AbortError');
+ const { data, error } = await supabase
+ .from('ai_run_results')
+ .select('result_key,status,body')
+ .eq('feature', feature)
+ .maybeSingle();
+ if (signal?.aborted) throw new DOMException('The request was canceled.', 'AbortError');
+ if (error) return null;
+ const row = data as { result_key?: unknown; status?: unknown; body?: unknown } | null;
+ if (row && row.result_key !== resultKey) return null;
+ if (row && typeof row.status === 'number') return { status: row.status, body: row.body };
+ // No row yet, or a run still writing (status null).
+ if (!row && now() - recoveryStartedAt >= RUN_RESULT_MISSING_MS) return null;
+ if (row && now() - recoveryStartedAt >= runningGraceMs) return null;
+ if (now() - startedAt >= RUN_RESULT_WAIT_MS) return null;
+ await wait(RUN_RESULT_POLL_MS, signal);
+ }
+}
diff --git a/src/lib/observability.ts b/src/lib/observability.ts
index 4745bb775..3b88a77d2 100644
--- a/src/lib/observability.ts
+++ b/src/lib/observability.ts
@@ -74,6 +74,7 @@ const PRODUCT_EVENT_NAMES = new Set([
"ai_request_failed",
"ai_request_started",
"ai_request_succeeded",
+ "ai_stream_recovered",
"content_exported",
"free_tool_used",
"sparring_round_completed",
diff --git a/src/locales/en.json b/src/locales/en.json
index 54a397070..13c624810 100644
--- a/src/locales/en.json
+++ b/src/locales/en.json
@@ -2116,6 +2116,8 @@
"turnOf": "Turn {{current}} of {{total}}",
"preparingResult": "Preparing your result",
"judgeReading": "The judge is reading the transcript.",
+ "judgeFailed": "The judge could not finish scoring.",
+ "scoringStopped": "Scoring stopped",
"scoring": "Scoring the round",
"noTimer": "No timer while judging",
"timeRemaining": "{{time}} remaining",
@@ -2124,6 +2126,7 @@
"resetTurn": "Reset turn",
"clockNote": "The clock uses the saved deadline, not tab focus.",
"scoreFailed": "Could not score this round",
+ "scorecardMalformed": "The scorecard came back malformed. Try scoring again.",
"retryScoring": "Try scoring again",
"judgeWeighing": "Judge is weighing the round",
"judgeDescription": "Reading the claims, responses, and strategic choices. This usually takes one request.",
@@ -2575,6 +2578,7 @@
"unavailable": "{{provider}} is temporarily unavailable. Please try again later.",
"safety": "{{provider}} blocked this request for safety reasons. Try rephrasing your input.",
"unexpectedProvider": "The AI provider returned an unexpected error. Please try again later.",
+ "invalidOutput": "The AI's answer came back malformed. Please try again.",
"unexpected": "An unexpected error occurred",
"offline": "Check your connection and try again.",
"notCharged": "You were not charged for this request."
diff --git a/src/locales/es.json b/src/locales/es.json
index 0e7d96e57..7a8addb11 100644
--- a/src/locales/es.json
+++ b/src/locales/es.json
@@ -2116,6 +2116,8 @@
"turnOf": "Turno {{current}} de {{total}}",
"preparingResult": "Preparando tu resultado",
"judgeReading": "El juez está leyendo la transcripción.",
+ "judgeFailed": "El juez no pudo terminar de puntuar.",
+ "scoringStopped": "Puntuación detenida",
"scoring": "Puntuando la ronda",
"noTimer": "No hay cronómetro durante la evaluación",
"timeRemaining": "{{time}} restantes",
@@ -2124,6 +2126,7 @@
"resetTurn": "Reiniciar turno",
"clockNote": "El cronómetro usa la fecha límite guardada, no el foco de la pestaña.",
"scoreFailed": "No se pudo puntuar esta ronda",
+ "scorecardMalformed": "La tarjeta de puntuación llegó con un formato incorrecto. Vuelve a intentar puntuar.",
"retryScoring": "Intentar puntuar de nuevo",
"judgeWeighing": "El juez está valorando la ronda",
"judgeDescription": "Está leyendo las afirmaciones, respuestas y decisiones estratégicas. Esto suele requerir una solicitud.",
@@ -2575,6 +2578,7 @@
"unavailable": "{{provider}} no está disponible por ahora. Inténtalo más tarde.",
"safety": "{{provider}} bloqueó esta solicitud por seguridad. Prueba a reformularla.",
"unexpectedProvider": "El proveedor de IA devolvió un error inesperado. Inténtalo más tarde.",
+ "invalidOutput": "La respuesta de la IA llegó con un formato incorrecto. Inténtalo de nuevo.",
"unexpected": "Se produjo un error inesperado",
"offline": "Comprueba tu conexión e inténtalo de nuevo.",
"notCharged": "No se te ha cobrado esta solicitud."
diff --git a/src/locales/hu.json b/src/locales/hu.json
index dbd7cf5e6..e99eb07ac 100644
--- a/src/locales/hu.json
+++ b/src/locales/hu.json
@@ -2116,6 +2116,8 @@
"turnOf": "{{current}}. kör a(z) {{total}} közül",
"preparingResult": "Eredmény előkészítése",
"judgeReading": "A bíró olvassa az átiratot.",
+ "judgeFailed": "A bíró nem tudta befejezni az értékelést.",
+ "scoringStopped": "Az értékelés leállt",
"scoring": "Forduló értékelése",
"noTimer": "Értékelés közben nincs időmérő",
"timeRemaining": "{{time}} van hátra",
@@ -2124,6 +2126,7 @@
"resetTurn": "Kör visszaállítása",
"clockNote": "Az óra a mentett határidőt használja, nem a böngészőfül fókuszát.",
"scoreFailed": "Nem sikerült értékelni ezt a fordulót",
+ "scorecardMalformed": "Az értékelőlap hibás formátumban érkezett. Próbáld újra az értékelést.",
"retryScoring": "Értékelés újrapróbálása",
"judgeWeighing": "A bíró mérlegeli a fordulót",
"judgeDescription": "Az állításokat, válaszokat és stratégiai döntéseket olvassa. Ehhez általában egy kérés kell.",
@@ -2575,6 +2578,7 @@
"unavailable": "{{provider}}: átmenetileg nem érhető el. Próbáld újra később.",
"safety": "{{provider}}: biztonsági okokból elutasította a kérést. Próbáld másképp megfogalmazni.",
"unexpectedProvider": "Az MI-szolgáltató váratlan hibát adott. Próbáld újra később.",
+ "invalidOutput": "Az MI válasza hibás formátumban érkezett. Próbáld újra.",
"unexpected": "Váratlan hiba történt",
"offline": "Ellenőrizd a kapcsolatot, és próbáld újra.",
"notCharged": "Ezt a kérést nem számoltuk el."
diff --git a/src/pages/SparringPage.tsx b/src/pages/SparringPage.tsx
index 7fb228bf5..fe1bcae52 100644
--- a/src/pages/SparringPage.tsx
+++ b/src/pages/SparringPage.tsx
@@ -5,7 +5,7 @@ import { ArrowLeft, CircleHelp } from 'lucide-react';
import { useLocation, useNavigate } from 'react-router-dom';
import { useSparringStore } from '@/lib/stores/sparring-store';
import { useSettingsStore } from '@/lib/stores/settings-store';
-import { generateSparringTurn, isAISubscriptionError, judgeSparringSession } from '@/lib/ai/client';
+import { AIError, generateSparringTurn, isAISubscriptionError, isInvalidOutputError, judgeSparringSession } from '@/lib/ai/client';
import { getUserFacingAIError } from '@/lib/ai/error-display';
import type { SparringRespondInput, SparringScoreInput, SparringSpeech } from '@/lib/ai/schemas/sparring';
import { SparringSetup, type SparringSetupValues } from '@/components/features/sparring/SparringSetup';
@@ -287,7 +287,9 @@ export default function SparringPage() {
.catch((error: unknown) => {
if (judgeRequestRef.current !== requestKey) return;
handleSubscriptionError(error);
- setJudgingError(readableError(error, t('sparring.round.judgeRequestFailed', 'The judge request failed. Check your connection and try again.')));
+ setJudgingError(isInvalidOutputError(error)
+ ? `${t('sparring.round.scorecardMalformed', 'The scorecard came back malformed. Try scoring again.')}${error instanceof AIError && error.notCharged ? ` ${t('aiErrors.notCharged')}` : ''}`
+ : readableError(error, t('sparring.round.judgeRequestFailed', 'The judge request failed. Check your connection and try again.')));
})
.finally(() => {
if (judgeRequestRef.current === requestKey) {
diff --git a/src/test/ai/live.test.ts b/src/test/ai/live.test.ts
index d9331bcb6..e9ef21640 100644
--- a/src/test/ai/live.test.ts
+++ b/src/test/ai/live.test.ts
@@ -1,6 +1,6 @@
import { afterEach, describe, expect, it, vi } from 'vitest';
import { FunctionsFetchError, FunctionsHttpError } from '@supabase/supabase-js';
-import { LIVE_STREAM_DRAIN_MS, readLiveResponse } from '@/lib/ai/live';
+import { LIVE_STREAM_DRAIN_MS, LIVE_STREAM_STALL_MS, LIVE_STREAM_STALLED_OPEN_MS, LiveStreamError, readLiveResponse } from '@/lib/ai/live';
function eventStream(chunks: string[]): Response {
const encoder = new TextEncoder();
@@ -171,4 +171,57 @@ describe('draining the body after result', () => {
await vi.advanceTimersByTimeAsync(1);
expect(cancelled).toBe(true);
});
+
+ it('reads a large result split across many reads, and one sent without its closing blank line', async () => {
+ const body = { output: { thesis: 'x'.repeat(5_000) }, discoveredSources: [{ title: 'é ünicode ✓', content: 'y'.repeat(20_000) }] };
+ const whole = event('delta', { text: '{"thesis"' }) + event('result', { status: 200, body });
+ // Byte-level splits land inside multi-byte characters too.
+ const bytes = new TextEncoder().encode(whole);
+ const split = new Response(new ReadableStream({
+ start(controller) {
+ for (let index = 0; index < bytes.length; index += 997) controller.enqueue(bytes.slice(index, index + 997));
+ controller.close();
+ },
+ }), { headers: { 'Content-Type': 'text/event-stream' } });
+ await expect(readLiveResponse(split, { onDraft: () => {} })).resolves.toEqual(body);
+
+ const unterminated = eventStream([`event: result\ndata: ${JSON.stringify({ status: 200, body })}`]);
+ await expect(readLiveResponse(unterminated, { onDraft: () => {} })).resolves.toEqual(body);
+ });
+
+ it('tells a damaged result, a dropped connection, and a silent stream apart instead of throwing', async () => {
+ const damaged = await readLiveResponse(eventStream(['event: result\ndata: {"status":200,"body":{"out\n\n']), { onDraft: () => {} });
+ expect(damaged).toBeInstanceOf(FunctionsFetchError);
+ expect((damaged as FunctionsFetchError).context).toMatchObject({ name: 'LiveStreamError', reason: 'unreadable' });
+
+ const encoder = new TextEncoder();
+ const dropped = new Response(new ReadableStream({
+ start(controller) {
+ controller.enqueue(encoder.encode(event('delta', { text: '{"a": "b' })));
+ setTimeout(() => controller.error(new TypeError('network error')), 5);
+ },
+ }), { headers: { 'Content-Type': 'text/event-stream' } });
+ const droppedError = await readLiveResponse(dropped, { onDraft: () => {} });
+ expect((droppedError as FunctionsFetchError).context).toBeInstanceOf(LiveStreamError);
+ expect((droppedError as FunctionsFetchError).context).toMatchObject({ reason: 'dropped' });
+
+ vi.useFakeTimers();
+ let cancelled = false;
+ const silent = new Response(new ReadableStream({
+ start(controller) {
+ controller.enqueue(encoder.encode(event('delta', { text: '{"a"' })));
+ },
+ cancel() {
+ cancelled = true;
+ },
+ }), { headers: { 'Content-Type': 'text/event-stream' } });
+ const pending = readLiveResponse(silent, { onDraft: () => {} });
+ await vi.advanceTimersByTimeAsync(LIVE_STREAM_STALL_MS);
+ const stalled = await pending;
+ expect((stalled as FunctionsFetchError).context).toMatchObject({ reason: 'stalled' });
+ // Left open, so the function does not cancel the run being recovered.
+ expect(cancelled).toBe(false);
+ await vi.advanceTimersByTimeAsync(LIVE_STREAM_STALLED_OPEN_MS);
+ expect(cancelled).toBe(true);
+ });
});
diff --git a/src/test/sparring/sparring-round.test.tsx b/src/test/sparring/sparring-round.test.tsx
index 028a80022..b75e5b695 100644
--- a/src/test/sparring/sparring-round.test.tsx
+++ b/src/test/sparring/sparring-round.test.tsx
@@ -53,6 +53,21 @@ describe('SparringRound', () => {
expect(useSparringStore.getState().currentSession?.timerRunning).toBe(true);
});
+ it('says scoring stopped when the judge fails, and goes back to reading on a retry', () => {
+ const session = useSparringStore.getState().startSession({ motion: 'This house would make public transport free.', userSide: 'pro', format: 'standard', difficulty: 'medium' });
+ const round = (judgingError: string | null) => undefined} onRetryJudging={() => undefined} />;
+ const view = render(round('The AI provider returned an unexpected error.'));
+ expect(screen.getByText('Scoring stopped')).toBeInTheDocument();
+ expect(screen.getByText('The judge could not finish scoring.')).toBeInTheDocument();
+ expect(screen.queryByText('The judge is reading the transcript.')).not.toBeInTheDocument();
+ expect(screen.queryByText('Round in progress')).not.toBeInTheDocument();
+
+ // Try scoring again clears the error while the new request runs.
+ view.rerender(round(null));
+ expect(screen.getByText('The judge is reading the transcript.')).toBeInTheDocument();
+ expect(screen.queryByText('Scoring stopped')).not.toBeInTheDocument();
+ });
+
it('does not nest a second main landmark inside the app layout', () => {
renderRound();
expect(document.querySelector('main')).toBeNull();
diff --git a/supabase/README.md b/supabase/README.md
index a0f962dcd..0931e6bbf 100644
--- a/supabase/README.md
+++ b/supabase/README.md
@@ -30,8 +30,8 @@ supabase/tests/ pgTAP database and policy tests
### AI and practice
-- `ai`: authenticated endpoint for generate, attack, brainstorm, factoids, Ask, Crossfire, and sparring. It enforces supported providers and controls, access limits, research bounds, output validation, usage accounting, and sparring integrity. With `stream: true` it streams the model's output as server-sent events and ends with the same body as a JSON request (see `ai/stream.ts`).
-- `ai-stream`: authenticated server-sent events endpoint for Ask. It emits deltas, a final event, or an error and records usage.
+- `ai`: authenticated endpoint for generate, attack, brainstorm, factoids, Ask, Crossfire, and sparring. It enforces supported providers and controls, access limits, research bounds, output validation, usage accounting, and sparring integrity. With `stream: true` it streams the model's output as server-sent events and ends with the same body as a JSON request (see `ai/stream.ts`). A streamed run holds its worker with `EdgeRuntime.waitUntil` until the last event is read. When the request sends a `resultKey`, the run's result is also written to `ai_run_results` before it is sent, so the browser can recover it after a dropped stream (`ai/run-results.ts`).
+- `ai-stream`: authenticated server-sent events endpoint for Ask. It emits deltas, a final event, or an error and records usage, holding its worker with `EdgeRuntime.waitUntil` until the stream closes.
- `voice`: authenticated text-to-speech for the opponent in voice rounds. It accepts one segment of at most 1,000 characters with the token `ai` signed for it, so it speaks only server-written text. It answers `voice_unavailable` without `ELEVENLABS_API_KEY`, enforces the per-user daily voice cap through `consume_daily_quota` under the `voice` key, calls ElevenLabs (`eleven_v4` for sparring, `eleven_v4_turbo` for Crossfire), and records each request in `ai_requests` as `voice_sparring` or `voice_crossfire`. Every failure is a JSON error with a code, and the browser then reads the reply with its own voice.
`ai/index.ts` is the entrypoint and dispatcher; `ai/features/` holds one handler per feature, `ai/provider.ts` the OpenRouter call, JSON repair, and the retry for a reply cut off at its cap (`_shared/output-caps.ts` holds the caps), `ai/request.ts` input hygiene, and `ai/runtime.ts` the environment and service clients. Each feature's prompt lives in `_shared/prompts/.ts`, with shared rules in `_shared/prompts/common.ts` and composition in `_shared/prompts/system.ts`.
diff --git a/supabase/functions/_shared/auto-model-router.test.ts b/supabase/functions/_shared/auto-model-router.test.ts
index 3c7870682..a076951df 100644
--- a/supabase/functions/_shared/auto-model-router.test.ts
+++ b/supabase/functions/_shared/auto-model-router.test.ts
@@ -150,3 +150,14 @@ Deno.test("router bounds state and propagates caller cancellation", async () =>
}
throw new Error("routing cancellation was converted to a fallback");
});
+
+Deno.test("Auto charges its most expensive candidate, so Auto Pro costs what Sonnet costs", async () => {
+ const { quotaUnitsFor } = await import("./usage-limits.ts");
+ // Written out by hand, not read from the table under test.
+ const ownUnits: Record = { "anthropic/claude-haiku-5.5": 1, "anthropic/claude-sonnet-5.5": 2, "anthropic/claude-opus-5.5": 5 };
+ for (const [selection, pool] of [[AUTO_FAST, FAST_CANDIDATES], [AUTO_PRO, PRO_CANDIDATES]] as const) {
+ const expected = Math.max(...Object.keys(pool).map((id) => ownUnits[id] ?? 1));
+ if (quotaUnitsFor(selection) !== expected) throw new Error(`${selection} reserves ${quotaUnitsFor(selection)}, its pool's most expensive candidate costs ${expected}`);
+ }
+ if (quotaUnitsFor(AUTO_PRO) !== 2) throw new Error(`Auto Pro reserves ${quotaUnitsFor(AUTO_PRO)}; its only candidate, Sonnet, costs 2`);
+});
diff --git a/supabase/functions/_shared/background.ts b/supabase/functions/_shared/background.ts
new file mode 100644
index 000000000..9036f5ca2
--- /dev/null
+++ b/supabase/functions/_shared/background.ts
@@ -0,0 +1,91 @@
+// Keeps an Edge Function worker alive while work it started is still running.
+//
+// Supabase counts a streamed Response as answered once its headers go out,
+// and can then retire the worker ("EarlyDrop") with the stream still being
+// written, unless the work is registered with EdgeRuntime.waitUntil. Every
+// streamed run registers its whole lifecycle here: the run, the final event,
+// the result row, and the close.
+
+interface EdgeRuntimeLike {
+ waitUntil?: (promise: Promise) => void;
+}
+
+// Supabase stops a worker 150 s after its request arrived.
+export const WORKER_WALL_CLOCK_MS = 150_000;
+// Kept free at the end of it: the settle below, and slack for the runtime.
+const WALL_CLOCK_MARGIN_MS = 2_000;
+
+/** Milliseconds a worker whose request arrived at `receivedAt` may still hold on, never negative. */
+export function workerTimeLeft(receivedAt: number, now = Date.now()): number {
+ return Math.max(0, WORKER_WALL_CLOCK_MS - WALL_CLOCK_MARGIN_MS - (now - receivedAt));
+}
+
+/** Resolves when `work` settles or after `ms`, whichever is first; the timer never outlives it. */
+export async function waitAtMost(work: Promise, ms: number): Promise {
+ let timer: ReturnType | undefined;
+ try {
+ await Promise.race([
+ work.then(() => undefined, () => undefined),
+ new Promise((resolve) => {
+ timer = setTimeout(resolve, Math.max(0, ms));
+ }),
+ ]);
+ } finally {
+ if (timer) clearTimeout(timer);
+ }
+}
+
+/**
+ * Hands `source` to the HTTP layer one chunk per read. `drained` resolves once
+ * the reader has taken the last chunk and asked for more, or went away. A
+ * stream enqueued from `start` closes with its tail still queued; this tells
+ * the worker when that tail has actually left.
+ */
+export function trackDrain(source: ReadableStream): { body: ReadableStream; drained: Promise } {
+ const reader = source.getReader();
+ let markDrained!: () => void;
+ const drained = new Promise((resolve) => {
+ markDrained = resolve;
+ });
+ const body = new ReadableStream({
+ async pull(controller) {
+ try {
+ const { value, done } = await reader.read();
+ if (done) {
+ controller.close();
+ markDrained();
+ } else {
+ controller.enqueue(value);
+ }
+ } catch (error) {
+ controller.error(error);
+ markDrained();
+ }
+ },
+ async cancel(reason) {
+ markDrained();
+ await reader.cancel(reason);
+ },
+ }, { highWaterMark: 0 });
+ return { body, drained };
+}
+
+/**
+ * Registers `work` with EdgeRuntime.waitUntil when the runtime provides it
+ * (deployed and `supabase functions serve`); a no-op in tests and plain Deno.
+ * Returns `work` unchanged. A rejection is still the caller's to handle; the
+ * runtime only sees a settled promise. `settleMs` holds the worker that much
+ * longer, for bytes the HTTP layer took but has not written yet.
+ */
+export function keepWorkerAlive(work: Promise, settleMs = 0): Promise {
+ const runtime = (globalThis as { EdgeRuntime?: EdgeRuntimeLike }).EdgeRuntime;
+ if (typeof runtime?.waitUntil === "function") {
+ try {
+ const settled = work.then(() => undefined, () => undefined);
+ runtime.waitUntil(settleMs > 0 ? settled.then(() => new Promise((resolve) => setTimeout(resolve, settleMs))) : settled);
+ } catch (error) {
+ console.warn("[RUNTIME] waitUntil failed:", error instanceof Error ? error.message : "unknown");
+ }
+ }
+ return work;
+}
diff --git a/supabase/functions/_shared/entitlements.ts b/supabase/functions/_shared/entitlements.ts
index 9a811ae81..426e1ee8d 100644
--- a/supabase/functions/_shared/entitlements.ts
+++ b/supabase/functions/_shared/entitlements.ts
@@ -301,8 +301,8 @@ export async function peekRateLimit(userId: string, model?: ModelSelection): Pro
/**
* Gives back exactly what checkRateLimit reserved for a request that produced
- * nothing: the same `units`, which follow the requested selection (Auto Pro
- * reserves 5 even when it resolves to Haiku). A session call that did not pay
+ * nothing: the same `units`, which follow the requested selection (an Auto
+ * selection reserves its most expensive candidate). A session call that did not pay
* has nothing to give back, and a session charge is returned only while no
* other call has run on it. Returns whether units went back. A failed refund
* is logged, never thrown, so the student still gets the error.
diff --git a/supabase/functions/_shared/output-caps.ts b/supabase/functions/_shared/output-caps.ts
index c2f4be837..4f87b3787 100644
--- a/supabase/functions/_shared/output-caps.ts
+++ b/supabase/functions/_shared/output-caps.ts
@@ -19,8 +19,13 @@ export const REPLY_TOKEN_CAPS = {
sparringSpeech: 2_500,
/** The Sparring scorecard: four categories, moments, areas, verdict. */
sparringScore: 3_000,
- /** A full case, any length setting. */
- generate: 6_000,
+ /**
+ * A full case, any length setting. Claude Haiku can write past 6,000 on a
+ * comprehensive case; a cut-off reply is then written again from the start
+ * at twice the cap, which took Auto Fast cases past the function's time
+ * limit. The cap only bounds the reply, so room costs nothing unless used.
+ */
+ generate: 12_000,
attack: 4_000,
brainstorm: 4_000,
factoids: 4_000,
diff --git a/supabase/functions/_shared/usage-limits.ts b/supabase/functions/_shared/usage-limits.ts
index 703dd195d..d946088e4 100644
--- a/supabase/functions/_shared/usage-limits.ts
+++ b/supabase/functions/_shared/usage-limits.ts
@@ -4,22 +4,34 @@
// and rate_limits rows; these values decide how fast a request uses them up.
import { AUTO_FAST, AUTO_PRO, type ModelSelection } from "./model-catalog.ts";
+import { FAST_CANDIDATES, PRO_CANDIDATES } from "./auto-model-router.ts";
/** Quota units a model without an entry below costs. */
export const DEFAULT_QUOTA_UNITS = 1;
/**
* How many daily requests one call to a model counts as. Unlisted models
- * cost DEFAULT_QUOTA_UNITS. The quota is reserved before Auto picks a model,
- * so Auto selections are charged as their most expensive candidate.
+ * cost DEFAULT_QUOTA_UNITS.
*/
-export const MODEL_QUOTA_UNITS: Readonly> = {
+const MODEL_UNITS: Readonly> = {
"anthropic/claude-opus-5.5": 5,
"openai/gpt-6.1-sol": 3,
"anthropic/claude-sonnet-5.5": 2,
"deepseek/deepseek-v4-pro-0813": 2,
- [AUTO_PRO]: 5,
- [AUTO_FAST]: 1,
+};
+
+const modelUnits = (model: string) => MODEL_UNITS[model] ?? DEFAULT_QUOTA_UNITS;
+
+/**
+ * The quota is reserved before Auto picks a model, so an Auto selection costs
+ * its most expensive candidate. Derived from the pools, so a pool change
+ * (Auto Pro dropping Opus) cannot leave Auto charging for a model it no
+ * longer uses.
+ */
+export const MODEL_QUOTA_UNITS: Readonly> = {
+ ...MODEL_UNITS,
+ [AUTO_PRO]: Math.max(...Object.keys(PRO_CANDIDATES).map(modelUnits)),
+ [AUTO_FAST]: Math.max(...Object.keys(FAST_CANDIDATES).map(modelUnits)),
};
/**
@@ -55,8 +67,11 @@ export const SESSION_QUOTA: Readonly[] = [];
+ const runtime = globalThis as { EdgeRuntime?: { waitUntil: (promise: Promise) => void } };
+ runtime.EdgeRuntime = { waitUntil: (promise) => registered.push(promise) };
try {
const response = await handleRequest(
@@ -149,7 +152,15 @@ Deno.test("streaming Ask forwards the shared prompt and returns provider SSE", a
throw new Error("streaming handler did not return SSE");
}
+ // The answer is finished and closed within milliseconds, but the worker
+ // stays registered until the browser has read the final event, well past
+ // the old fixed 1 s settle.
+ if (registered.length !== 1) throw new Error(`expected one waitUntil registration, got ${registered.length}`);
+ const held = registered[0];
+ const early = await Promise.race([held.then(() => "released"), new Promise((resolve) => setTimeout(() => resolve("held"), 1_500))]);
+ if (early !== "held") throw new Error("the worker was released before the browser read the final event");
const body = await response.text();
+ await held;
assertIncludes(body, 'event: delta\ndata: {"text":"Szia"}');
assertIncludes(body, 'event: delta\ndata: {"text":" Adam"}');
assertIncludes(body, "event: final");
@@ -192,6 +203,7 @@ Deno.test("streaming Ask forwards the shared prompt and returns provider SSE", a
throw new Error("provider did not receive the current Ask message");
}
} finally {
+ delete runtime.EdgeRuntime;
globalThis.fetch = originalFetch;
}
});
diff --git a/supabase/functions/ai-stream/index.ts b/supabase/functions/ai-stream/index.ts
index 2388ec880..2b463fa05 100644
--- a/supabase/functions/ai-stream/index.ts
+++ b/supabase/functions/ai-stream/index.ts
@@ -24,6 +24,7 @@ import { HttpError, jsonError, requireUser } from "../_shared/auth.ts";
import { getCorsHeaders } from "../_shared/cors.ts";
import { checkRateLimit, logUsage, refundQuotaReservation } from "../_shared/entitlements.ts";
import { REFUND_MAX_DELIVERED_CHARS } from "../_shared/usage-limits.ts";
+import { keepWorkerAlive, trackDrain, waitAtMost, workerTimeLeft } from "../_shared/background.ts";
import {
CLAUDE_UNAVAILABLE_CODE,
CLAUDE_UNAVAILABLE_MESSAGE,
@@ -53,6 +54,13 @@ import { resolveModelSelection } from "../_shared/auto-model-router.ts";
import { outputTokenLimit, REPLY_TOKEN_CAPS } from "../_shared/output-caps.ts";
import { JEV_MODEL } from "../_shared/jev.ts";
+// After the close, hold the worker at most this long for the browser to take
+// the last bytes, and never past the wall clock. A client that stopped
+// reading must not hold it to 150 s.
+const DRAIN_CAP_MS = 10_000;
+// Bytes the HTTP layer took on the last read still have to reach the socket.
+const STREAM_SETTLE_MS = 500;
+
type ChatRole = "user" | "assistant";
type ChatMessage = {
@@ -106,6 +114,7 @@ export function buildMessages(message: string, history: ChatMessage[], mode: str
export async function handleRequest(req: Request): Promise {
+ const receivedAt = Date.now();
const corsHeaders = getCorsHeaders(req.headers.get("Origin"));
if (req.method === "OPTIONS") {
@@ -251,8 +260,9 @@ export async function handleRequest(req: Request): Promise {
const history = normalizeHistory(input.history);
const startedAt = Date.now();
- const stream = new ReadableStream({
- async start(controller) {
+ let answering!: Promise;
+ const source = new ReadableStream({
+ start: (controller) => answering = (async () => {
const encoder = new TextEncoder();
let accumulated = "";
let streamUsageRaw: Record | undefined;
@@ -439,10 +449,15 @@ export async function handleRequest(req: Request): Promise {
reader?.releaseLock();
try { controller.close(); } catch { /* already closed */ }
}
- },
+ })(),
});
+ // Supabase counts the answer as sent once headers are out and may retire
+ // the worker mid-answer (EarlyDrop) unless the work is registered. It stays
+ // registered until the browser has read the final event.
+ const { body: events, drained } = trackDrain(source);
+ keepWorkerAlive(answering.then(() => waitAtMost(drained, Math.min(DRAIN_CAP_MS, workerTimeLeft(receivedAt)))), STREAM_SETTLE_MS);
- return new Response(stream, {
+ return new Response(events, {
headers: {
...corsHeaders,
"Content-Type": "text/event-stream; charset=utf-8",
diff --git a/supabase/functions/ai/features/sparring.ts b/supabase/functions/ai/features/sparring.ts
index 3449e095c..59628e7f6 100644
--- a/supabase/functions/ai/features/sparring.ts
+++ b/supabase/functions/ai/features/sparring.ts
@@ -1,13 +1,15 @@
import { logUsage } from "../../_shared/entitlements.ts";
-import { mergeNormalizedUsage } from "../../_shared/cost.ts";
-import { callAI } from "../provider.ts";
+import { mergeNormalizedUsage, type NormalizedUsage, selectObservedUsage } from "../../_shared/cost.ts";
+import { ClaudeProviderUnavailableError } from "../../_shared/provider-routing.ts";
+import { AIResponseValidationError, callAI, isIncompleteProviderStream, missingOpenRouterUsage, withJsonRepairPrompt } from "../provider.ts";
import { cleanGeneratedText } from "../text.ts";
import { buildUserPrompt } from "../prompt-builder.ts";
import {
completeClaimedSparringScoreSession,
createSparringSpeechIntegrityToken,
+ checkSparringProviderOutput,
normalizeSparringOutput,
- parseSparringProviderOutput,
+ sparringOutputSchema,
type SparringDatabase,
type SparringRequest,
} from "../sparring-persistence.ts";
@@ -24,6 +26,21 @@ export interface SparringHandlerOptions {
onScoreSaved: () => void;
}
+// One more attempt runs inside the same request when the first gave no usable
+// answer for a reason another try can fix. A reply in the wrong shape (it
+// parsed, but a field is missing or wrong) gets a repair prompt naming the
+// field; for the judge, a provider outage or a dropped provider stream gets a
+// plain retry. A Sparring round has already paid, so the retry costs the
+// student nothing and spares them "Try scoring again". A failed save is not
+// retried, and neither is a run that has used up this much of its time.
+const RETRY_BEFORE_MS = 60_000;
+
+function transientProviderFailure(error: unknown): boolean {
+ if (isIncompleteProviderStream(error)) return true;
+ // 401, 402 and 403 are the key or its credit, which a retry does not fix.
+ return error instanceof ClaudeProviderUnavailableError && (error.status === null || error.status === 429 || error.status >= 500);
+}
+
// SPARRING: transcript-only opponent replies and scorecards. No research
// branch runs here, and score persistence follows output validation.
export async function handleSparring(ctx: FeatureContext, options: SparringHandlerOptions): Promise {
@@ -31,31 +48,71 @@ export async function handleSparring(ctx: FeatureContext, options: SparringHandl
const { userId, provider, model, reasoningEffort, workSignal, startTime, researchUsageForLog, recordMainResult, buildSystemPromptFor } = ctx;
const { request: sparringRequest, database: sparringDatabase, claimToken: sparringScoreClaimToken } = options;
if (!sparringRequest) throw new Error("Invalid sparring request");
+ const action = sparringRequest.action;
const systemPrompt = await buildSystemPromptFor(feature);
const userPrompt = buildUserPrompt(feature, sparringRequest as unknown as Record, "");
const messages = [{ role: "user", content: userPrompt }];
- const aiResult = recordMainResult(await callAI(
+ const outputSchema = sparringOutputSchema(action);
+ const callModel = (repairPath?: string) => callAI(
provider,
model,
- systemPrompt,
- messages,
+ repairPath === undefined
+ ? systemPrompt
+ : `${systemPrompt}\n\nJSON REPAIR INSTRUCTION: The previous answer did not match the required shape at "${repairPath}". Return only one JSON object with exactly the fields above, every list item a plain string.`,
+ repairPath === undefined ? messages : withJsonRepairPrompt(messages),
reasoningEffort,
- sparringRequest.action === "score" ? "sparringScore" : "sparringSpeech",
+ action === "score" ? "sparringScore" : "sparringSpeech",
workSignal,
ctx.stream,
- ));
- const output = cleanGeneratedText(aiResult.output);
- const normalizedOutput = normalizeSparringOutput(output, sparringRequest.action);
- const parsedOutput = parseSparringProviderOutput(normalizedOutput, sparringRequest.action);
- if (!parsedOutput) throw new Error("AI response did not match the expected sparring schema");
+ outputSchema,
+ );
+ const attempt = async (repairPath?: string, earlierUsage?: NormalizedUsage) => {
+ const result = await callModel(repairPath);
+ const usage = mergeNormalizedUsage([earlierUsage, result.usage]);
+ const checked = checkSparringProviderOutput(normalizeSparringOutput(cleanGeneratedText(result.output), action), action);
+ if (!checked.ok) {
+ // The field path only: the reply is the student's round and is never logged.
+ console.warn(`[SPARRING] ${action} reply failed its shape at ${checked.path}${repairPath === undefined ? "" : " after the repair retry"}`);
+ throw new AIResponseValidationError(checked.path, usage);
+ }
+ return { result: recordMainResult({ ...result, usage }), parsed: checked.value };
+ };
+
+ let answered: Awaited>;
+ try {
+ answered = await attempt();
+ } catch (error) {
+ const wrongShape = error instanceof AIResponseValidationError;
+ const retry = wrongShape || (action === "score" && transientProviderFailure(error));
+ if (!retry || workSignal.aborted || Date.now() - startTime > RETRY_BEFORE_MS) throw error;
+ const firstUsage = selectObservedUsage(error as { usage?: NormalizedUsage; completedUsage?: NormalizedUsage }, false)
+ ?? missingOpenRouterUsage("sparring_attempt");
+ console.warn(`[SPARRING] ${action} attempt failed (${error instanceof Error ? error.name : "unknown"}); trying once more`);
+ // The first attempt's draft is void; the answer is written again.
+ ctx.stream?.reset();
+ try {
+ answered = await attempt(wrongShape ? error.path : undefined, firstUsage);
+ } catch (retryError) {
+ // Bill the first attempt with whatever the second one cost. A shape
+ // failure already carries both.
+ if (retryError && typeof retryError === "object" && !(retryError instanceof AIResponseValidationError)) {
+ const withUsage = retryError as { usage?: NormalizedUsage; completedUsage?: NormalizedUsage };
+ withUsage.usage = mergeNormalizedUsage([firstUsage, withUsage.usage ?? missingOpenRouterUsage("sparring_retry")]);
+ withUsage.completedUsage = mergeNormalizedUsage([firstUsage, withUsage.completedUsage]);
+ }
+ throw retryError;
+ }
+ }
+ const aiResult = answered.result;
+ const parsedOutput = answered.parsed;
// In a voice round the speech may carry audio tags. They go to the voice
// only: the content that is shown, signed, and saved has none.
let voice: VoicePayload | undefined;
if (parsedOutput.action === "respond" && ctx.voice) {
const reply = await prepareVoiceReply(userId, feature, parsedOutput.speech.content);
- if (!reply.clean) throw new Error("AI response did not match the expected sparring schema");
+ if (!reply.clean) throw new AIResponseValidationError("speech.content", aiResult.usage);
parsedOutput.speech.content = reply.clean;
voice = reply.voice;
}
diff --git a/supabase/functions/ai/index.ts b/supabase/functions/ai/index.ts
index f79a0d58d..65b4bb8d9 100644
--- a/supabase/functions/ai/index.ts
+++ b/supabase/functions/ai/index.ts
@@ -95,13 +95,19 @@ import {
} from "./runtime.ts";
import {
type AIResult,
+ AIResponseValidationError,
callAI,
createRequestBudgetSignal,
type GenerationStream,
+ InvalidJsonResponseError,
+ isIncompleteProviderStream,
modelForRequest,
OPENROUTER_PROVIDER,
+ REQUEST_BUDGET_MS,
} from "./provider.ts";
import { streamFeatureResponse } from "./stream.ts";
+import { createRunResultRecorder, parseResultKey, type RunResultsDatabase } from "./run-results.ts";
+import { workerTimeLeft } from "../_shared/background.ts";
import {
isRecord,
MAX_REQUEST_BODY_BYTES,
@@ -138,8 +144,14 @@ export {
MAX_CROSSFIRE_TURN_CHARS,
} from "./features/crossfire-input.ts";
+// A streamed run that has not answered this long after its request arrived
+// is answered with a timeout: past the provider budget, before Supabase's
+// 150 s wall clock cuts the connection.
+const STREAM_DEADLINE_MS = REQUEST_BUDGET_MS + 12_000;
+
export async function handleRequest(req: Request): Promise {
const requestId = generateRequestId();
+ const receivedAt = Date.now();
const corsHeaders = getCorsHeaders(req.headers.get("origin"));
if (req.method === "OPTIONS") {
@@ -212,6 +224,10 @@ export async function handleRequest(req: Request): Promise {
// Live generation: answer with Server-Sent Events (see stream.ts). Older
// servers ignore the flag and answer with JSON, which clients also accept.
const streamRequested = body.stream === true;
+ // The browser's id for a streamed run. With it the final result is kept in
+ // ai_run_results, so a browser that loses the stream can still read it.
+ // Older clients send none and stream as before.
+ const resultKey = streamRequested ? parseResultKey(body.resultKey) : null;
// The per-user custom system prompt was replaced by each Current Debate's
// settings and instructions. A customSystemPrompt field from an older
// client is ignored.
@@ -587,6 +603,12 @@ export async function handleRequest(req: Request): Promise {
});
}
+ // Every path that gives the units back goes through this, so a run is
+ // refunded at most once, however its failure is reached.
+ const reservation = rateLimit;
+ let refund: Promise | undefined;
+ const refundOnce = () => refund ??= refundQuotaReservation(userId, reservation);
+
const routingStartedAt = Date.now();
try {
const resolution = await resolveModelSelection({
@@ -614,7 +636,8 @@ export async function handleRequest(req: Request): Promise {
const message = error instanceof Error ? error.message : "Model routing failed";
log(requestId, "ERROR", "Model routing failed", { error: message });
// Nothing ran, so nothing is charged.
- await refundQuotaReservation(userId, rateLimit);
+ await refundOnce();
+ await logUsage(userId, feature, provider, requestedModel, undefined, undefined, "error", "routing_failed");
return badRequest(message.includes("Pro") ? message : "Unable to route this request", corsHeaders, message.includes("Pro") ? 403 : 500);
}
@@ -656,7 +679,8 @@ export async function handleRequest(req: Request): Promise {
if (rankingUsage) await logUsage(userId, "jev_source_ranking", provider, JEV_MODEL, rankingUsage, undefined, ranked.jevUsage ? "error" : "success", ranked.jevUsage ? "jev_incomplete" : undefined);
} catch {
await interruptWorkspaceThread();
- await refundQuotaReservation(userId, rateLimit);
+ await refundOnce();
+ await logUsage(userId, feature, provider, model, undefined, undefined, "error", "source_ranking_failed");
return badRequest("Unable to rank sources", corsHeaders, 500);
}
}
@@ -751,6 +775,9 @@ export async function handleRequest(req: Request): Promise {
// The streamed answer on screen is complete: the model sent its
// finish_reason. A repair retry's reset voids it.
let modelFinished = false;
+ // The handler wrote this run's ai_requests row. Every handler logs it right
+ // before ctx.respond, so a failure after that must not log a second one.
+ let usageLogged = false;
const countDelivered = (stream?: GenerationStream): GenerationStream | undefined => stream && {
...stream,
delta: (text) => {
@@ -790,9 +817,12 @@ export async function handleRequest(req: Request): Promise {
markTruncated: () => {
responseTruncated = true;
},
- respond: async (payload) => new Response(JSON.stringify(await successBody(payload)), {
- headers: { ...corsHeaders, "Content-Type": "application/json" },
- }),
+ respond: async (payload) => {
+ usageLogged = true;
+ return new Response(JSON.stringify(await successBody(payload)), {
+ headers: { ...corsHeaders, "Content-Type": "application/json" },
+ });
+ },
};
switch (feature) {
@@ -846,6 +876,7 @@ export async function handleRequest(req: Request): Promise {
const latencyMs = Date.now() - startTime;
await logUsage(userId, feature, provider, model, mergeNormalizedUsage([fallbackResult.usage, ...researchUsageForLog()]), latencyMs);
+ usageLogged = true;
return new Response(JSON.stringify(await successBody({ output })), { headers: { ...corsHeaders, "Content-Type": "application/json" } });
} catch (err) {
const message = err instanceof Error ? err.message : "Unknown error";
@@ -879,51 +910,73 @@ export async function handleRequest(req: Request): Promise {
// Claude runs only on the Anthropic key, with no fallback provider. When
// it fails the student gets an outage message.
const claudeUnavailable = err instanceof ClaudeProviderUnavailableError;
+ // The model answered, but not in a usable shape even after its repair
+ // retry. Not a provider outage: logged and answered as invalid_output.
+ const invalidOutput = err instanceof AIResponseValidationError || err instanceof InvalidJsonResponseError;
const aborted = workSignal?.aborted === true;
const timedOut = aborted && (workSignal.reason as { name?: unknown } | undefined)?.name === "TimeoutError";
const errorUsage = mergeNormalizedUsage([
mainErrorUsage,
...researchUsageForLog(),
]);
- // A run that produced nothing costs nothing: a provider failure, a
- // response that failed validation, the request budget running out, or
- // the browser leaving (reload, closed tab, Stop) before the student saw
- // a usable part of the answer. A run whose answer was saved, or that
- // streamed REFUND_MAX_DELIVERED_CHARS or more, stays charged, and so
- // does one whose streamed answer the model had finished, however the
- // request ended afterwards (a disconnect, a failed read of the usage
- // trailer): a short answer may already be on screen. A non-streamed
- // call the browser left after the provider answered stays charged too.
- const answerFinished = modelFinished || (aborted && Boolean(completedMainUsage));
- const refundable = !outputSaved && !answerFinished && deliveredChars < REFUND_MAX_DELIVERED_CHARS;
+ // A run that failed on the server costs nothing, however much of it had
+ // streamed: a provider failure, a reply cut off at its cap, a response
+ // that failed validation, or the request budget running out. The
+ // student sees an error instead of the answer. A run the browser left
+ // (reload, closed tab, Stop, a dropped connection; the server cannot
+ // tell these apart) is refunded only before the student saw a usable
+ // part of the answer: one that streamed REFUND_MAX_DELIVERED_CHARS or
+ // more, or whose answer the model had finished, stays charged, as does
+ // a non-streamed call the browser left after the provider answered.
+ // Otherwise a student could read most of a case and press Stop to get
+ // the unit back. A run whose answer was saved always stays charged.
+ const clientLeft = aborted && !timedOut;
+ const answerKept = clientLeft && (modelFinished || Boolean(completedMainUsage) || deliveredChars >= REFUND_MAX_DELIVERED_CHARS);
+ const refundable = !outputSaved && !answerKept;
// Refund before answering, so a retry sent straight away sees it.
- const refunded = refundable && await refundQuotaReservation(userId, rateLimit);
+ const refunded = refundable && await refundOnce();
// The weekly cap counts real spend, so a refunded run's cost (BYOK
// provider charges included) still counts. Only a Claude refusal that cost nothing is logged at 0.
const freeRefusal = claudeUnavailable && !aborted && !err.afterOutput && !completedMainUsage && !usageError?.completedUsage &&
!researchDidWork && !researchAccountingIncomplete && !chargedAgainstCap(errorUsage);
- await logUsage(
- userId,
- feature,
- provider,
- model,
- freeRefusal ? excludeFromCap(errorUsage) : errorUsage,
- latencyMs,
- aborted && !timedOut ? "cancelled" : "error",
- claudeUnavailable ? "claude_unavailable" : timedOut ? "timeout" : aborted ? "aborted" : "provider_error",
- );
+ // One row per run: a handler that already logged its success row (the
+ // failure came while answering) is not logged again.
+ if (!usageLogged) {
+ await logUsage(
+ userId,
+ feature,
+ provider,
+ model,
+ freeRefusal ? excludeFromCap(errorUsage) : errorUsage,
+ latencyMs,
+ aborted && !timedOut ? "cancelled" : "error",
+ claudeUnavailable ? "claude_unavailable" : timedOut ? "timeout" : aborted ? "aborted" : invalidOutput ? "invalid_output" : "provider_error",
+ );
+ }
// Tells the client to say "You were not charged": the units went back,
// or this call ran inside a session that had already paid.
const notCharged = refunded || rateLimit.charged === false ? { charged: false } : {};
+ // A provider outage, a cut-off provider stream, or the time budget is
+ // worth trying again as it is; the browser handles those quietly. Other
+ // failures (an answer that failed validation, a save that failed) may
+ // be bugs and are reported.
+ const retryable = claudeUnavailable || timedOut || isIncompleteProviderStream(err) ? { retryable: true } : {};
if (claudeUnavailable) {
- return new Response(JSON.stringify({ error: CLAUDE_UNAVAILABLE_MESSAGE, code: CLAUDE_UNAVAILABLE_CODE, ...notCharged }), { status: CLAUDE_UNAVAILABLE_STATUS, headers: { ...corsHeaders, "Content-Type": "application/json" } });
+ return new Response(JSON.stringify({ error: CLAUDE_UNAVAILABLE_MESSAGE, code: CLAUDE_UNAVAILABLE_CODE, ...notCharged, ...retryable }), { status: CLAUDE_UNAVAILABLE_STATUS, headers: { ...corsHeaders, "Content-Type": "application/json" } });
}
- return new Response(JSON.stringify({ error: "AI request failed. Please try again.", code: "provider_error", ...notCharged }), { status: 500, headers: { ...corsHeaders, "Content-Type": "application/json" } });
+ if (invalidOutput && !aborted) {
+ return new Response(JSON.stringify({ error: "The AI answer came back malformed. Please try again.", code: "invalid_output", ...notCharged }), { status: 502, headers: { ...corsHeaders, "Content-Type": "application/json" } });
+ }
+ return new Response(JSON.stringify({ error: "AI request failed. Please try again.", code: "provider_error", ...notCharged, ...retryable }), { status: 500, headers: { ...corsHeaders, "Content-Type": "application/json" } });
}
};
if (!streamRequested) return await runFeature();
- return streamFeatureResponse(corsHeaders, streamCancel, runFeature);
+ return streamFeatureResponse(corsHeaders, streamCancel, runFeature, {
+ recorder: resultKey ? createRunResultRecorder(supabase as unknown as RunResultsDatabase, userId, feature, resultKey) : undefined,
+ deadlineMs: STREAM_DEADLINE_MS - (Date.now() - receivedAt),
+ endMs: workerTimeLeft(receivedAt),
+ });
}
if (import.meta.main) {
diff --git a/supabase/functions/ai/provider.ts b/supabase/functions/ai/provider.ts
index 3d655d471..a49172eaf 100644
--- a/supabase/functions/ai/provider.ts
+++ b/supabase/functions/ai/provider.ts
@@ -65,6 +65,40 @@ export class OutputTruncatedError extends Error {
}
}
+/**
+ * A reply that parsed as JSON but not into the shape the feature needs, even
+ * after its repair retry. `path` names the first wrong field for the log;
+ * the reply itself is never logged. The call completed, so it is billed.
+ */
+export class AIResponseValidationError extends Error {
+ readonly path: string;
+ usage?: NormalizedUsage;
+ completedUsage?: NormalizedUsage;
+
+ constructor(path: string, usage?: NormalizedUsage) {
+ super("AI response did not match the expected shape");
+ this.name = "AIResponseValidationError";
+ this.path = path;
+ this.usage = usage;
+ this.completedUsage = usage;
+ }
+}
+
+/** The provider's stream stopped before the model ended its reply: a dropped upstream connection. */
+export class IncompleteProviderStreamError extends Error {
+ usage?: NormalizedUsage;
+
+ constructor(usage?: NormalizedUsage) {
+ super("OpenRouter stream ended before completion");
+ this.name = "IncompleteProviderStreamError";
+ this.usage = usage;
+ }
+}
+
+export function isIncompleteProviderStream(error: unknown): boolean {
+ return error instanceof IncompleteProviderStreamError;
+}
+
/**
* Parses provider content, attaching the attempt's token usage to a parse
* failure so callAI's repair retry can fold it into the final accounting.
@@ -517,10 +551,23 @@ export async function readOpenRouterStream(
}
};
+ // The model ended its reply and only the usage trailer was lost: the answer
+ // is whole, so it is kept and its usage marked incomplete.
+ let trailerLost = false;
try {
while (!done) {
if (signal.aborted) throw signal.reason ?? new DOMException("The operation was aborted", "AbortError");
- const { done: ended, value } = await reader.read();
+ let read: ReadableStreamReadResult;
+ try {
+ read = await reader.read();
+ } catch (readError) {
+ if (finishReason && finishReason !== "length" && text && !signal.aborted) {
+ trailerLost = true;
+ break;
+ }
+ throw readError;
+ }
+ const { done: ended, value } = read;
if (ended) break;
totalBytes += value.byteLength;
if (totalBytes > MAX_PROVIDER_RESPONSE_BYTES * MAX_STREAM_FRAME_OVERHEAD) {
@@ -534,7 +581,7 @@ export async function readOpenRouterStream(
if (done) break;
}
}
- if (!done) {
+ if (!done && !trailerLost) {
buffer += decoder.decode();
for (const line of buffer.split("\n")) {
processLine(line);
@@ -553,9 +600,11 @@ export async function readOpenRouterStream(
void reader.cancel().catch(() => undefined);
}
if (!done && !finishReason) {
- const error = new Error("OpenRouter stream ended before completion") as Error & { usage?: NormalizedUsage };
- error.usage = usageRaw ? { ...normalizeOpenRouterUsage(usageRaw), incomplete: true } : missingOpenRouterUsage("stream");
- throw error;
+ throw new IncompleteProviderStreamError(usageRaw ? { ...normalizeOpenRouterUsage(usageRaw), incomplete: true } : missingOpenRouterUsage("stream"));
+ }
+ if (trailerLost) {
+ console.warn("[OpenRouter] Stream dropped after the reply finished; usage is incomplete");
+ return { text, usage: usageRaw ? { ...normalizeOpenRouterUsage(usageRaw), incomplete: true } : missingOpenRouterUsage("stream"), finishReason };
}
return { text, usage: usageRaw ? normalizeOpenRouterUsage(usageRaw) : missingOpenRouterUsage("stream"), finishReason };
}
diff --git a/supabase/functions/ai/quota-sessions_test.ts b/supabase/functions/ai/quota-sessions_test.ts
index 5f85da3ed..a16feeb43 100644
--- a/supabase/functions/ai/quota-sessions_test.ts
+++ b/supabase/functions/ai/quota-sessions_test.ts
@@ -22,6 +22,8 @@ const SPARRING_SESSION = "5e551000-0000-4000-8000-000000000009";
interface Harness {
rpcs: Array<{ name: string; body: Record }>;
+ /** Chat completion request bodies, in order. */
+ providerCalls: Array>;
usageRows: Array>;
/** What claim_ai_session_quota answers. */
claim: Record;
@@ -29,7 +31,7 @@ interface Harness {
reply: { content: string } | { status: number } | { hangAfter: string; finished?: boolean; thenFail?: boolean };
}
-const harness: Harness = { rpcs: [], usageRows: [], claim: {}, reply: { content: "" } };
+const harness: Harness = { rpcs: [], providerCalls: [], usageRows: [], claim: {}, reply: { content: "" } };
function json(body: unknown, status = 200): Response {
return new Response(JSON.stringify(body), { status, headers: { "Content-Type": "application/json" } });
@@ -84,6 +86,7 @@ function install() {
}
if (url.href === "https://openrouter.ai/api/v1/chat/completions") {
const body = JSON.parse(String(init?.body));
+ harness.providerCalls.push(body);
const reply = harness.reply;
if ("status" in reply) return json({ error: { message: "Provider overloaded" } }, reply.status);
if ("hangAfter" in reply) return new Response(hangingStream(reply.hangAfter, reply.finished === true, init?.signal, reply.thenFail === true), { headers: { "Content-Type": "text/event-stream" } });
@@ -102,6 +105,7 @@ function install() {
function reset(reply: Harness["reply"], claim: Record = { allowed: true, charged: true, units: 1, current_count: 1, denied_by: null, charge_id: 1 }) {
harness.rpcs.length = 0;
+ harness.providerCalls.length = 0;
harness.usageRows.length = 0;
harness.reply = reply;
harness.claim = claim;
@@ -232,6 +236,36 @@ Deno.test("a failed call refunds the session only when it was the call that paid
}
});
+Deno.test("a Sparring reply in the wrong shape gets one repair retry, then fails as invalid_output and is refunded once", async () => {
+ const originalWarn = console.warn;
+ const originalError = console.error;
+ console.warn = () => undefined;
+ console.error = () => undefined;
+ install();
+ try {
+ // Valid JSON, but the speech has no strategy, on the first try and the repair.
+ reset({ content: JSON.stringify({ action: "respond", speech: { content: "Who pays for the buses?", targetedPoints: [] } }) });
+ const response = await send({ feature: "sparring", input: sparringInput });
+ assertEquals(response.status, 502);
+ const body = await response.json();
+ assertEquals([body.code, body.charged], ["invalid_output", false]);
+ assertEquals(harness.providerCalls.length, 2, "one first attempt and one repair retry");
+ const repairMessages = harness.providerCalls[1].messages as Array<{ role: string; content: unknown }>;
+ assertEquals(String(repairMessages.at(-1)?.content).includes("Return only valid JSON matching this schema"), true);
+ for (const call of harness.providerCalls) {
+ assertEquals((call.response_format as { type?: string; json_schema?: { name?: string; strict?: boolean } }).json_schema?.name, "sparring_speech");
+ }
+ assertEquals(rpcNames(), ["claim_ai_session_quota", "refund_ai_session_quota"]);
+ assertEquals(harness.usageRows.filter((row) => row.feature === "sparring").map((row) => row.error_code), ["invalid_output"]);
+ // Both attempts are billed in that one row.
+ assertEquals(harness.usageRows.find((row) => row.feature === "sparring")?.input_tokens, 20);
+ } finally {
+ globalThis.fetch = originalFetch;
+ console.warn = originalWarn;
+ console.error = originalError;
+ }
+});
+
Deno.test("a streamed run the browser cancels is refunded only before a usable answer", async () => {
const originalError = console.error;
console.error = () => undefined;
@@ -333,6 +367,25 @@ Deno.test("a finished answer stays charged when the provider connection drops be
}
});
+Deno.test("a run that fails on the server is refunded however much of it streamed", async () => {
+ const originalError = console.error;
+ console.error = () => undefined;
+ install();
+ try {
+ // The provider connection breaks mid-answer, after a long stretch reached
+ // the browser. The student gets an error, not the answer.
+ const partial = `{"response":"${"x".repeat(REFUND_MAX_DELIVERED_CHARS + 300)}`;
+ reset({ hangAfter: partial, finished: false, thenFail: true });
+ const text = await (await send({ feature: "ask", input: { message: "A question", mode: "general", webSearch: false }, stream: true })).text();
+ assertEquals(text.includes('"charged":false'), true);
+ assertEquals(rpcNames(), ["consume_ai_quota", "refund_daily_quota"]);
+ assertEquals(harness.usageRows.filter((row) => row.feature === "ask").length, 1);
+ } finally {
+ globalThis.fetch = originalFetch;
+ console.error = originalError;
+ }
+});
+
Deno.test("an empty streamed reply is refunded even though the model said it finished", async () => {
const originalError = console.error;
console.error = () => undefined;
diff --git a/supabase/functions/ai/run-results.ts b/supabase/functions/ai/run-results.ts
new file mode 100644
index 000000000..be0fbbef3
--- /dev/null
+++ b/supabase/functions/ai/run-results.ts
@@ -0,0 +1,102 @@
+// Keeps a streamed run's final answer in public.ai_run_results
+// (20261023140000_ai_run_results.sql) so a browser whose stream dropped can
+// read it (src/lib/ai/run-recovery.ts). Writes never fail the run: a missing
+// table (function deployed before the migration) or a slow database only
+// costs the recovery, never the answer.
+
+const RESULT_KEY = /^[A-Za-z0-9._:-]{1,128}$/;
+// Either write is a single-row upsert or update; past this the answer is sent
+// without waiting, as it was before the table existed.
+const WRITE_TIMEOUT_MS = 4_000;
+// The finish waits this long for a slow start write to land, so the start
+// cannot arrive after the finish and leave a finished run looking unfinished.
+const BEGIN_SETTLE_MS = 10_000;
+
+/** The browser's id for a run, as the table accepts it, or null. */
+export function parseResultKey(value: unknown): string | null {
+ return typeof value === "string" && RESULT_KEY.test(value) ? value : null;
+}
+
+export interface RunResultRecorder {
+ /** Marks the run as started; a row still unfinished later is a run the worker lost. */
+ begin: () => Promise;
+ /**
+ * Stores what the `result` event carries, before it is sent. Waits at most
+ * `budgetMs` in all, so a slow database cannot hold the answer past the
+ * worker's wall clock.
+ */
+ finish: (result: Record, budgetMs?: number) => Promise;
+}
+
+type WriteResult = PromiseLike<{ error: { message?: string; code?: string } | null }>;
+
+interface RunResultsTable {
+ upsert: (row: Record, options: { onConflict: string }) => WriteResult;
+ update: (row: Record) => {
+ eq: (column: string, value: string) => {
+ eq: (column: string, value: string) => { eq: (column: string, value: string) => WriteResult };
+ };
+ };
+}
+
+export interface RunResultsDatabase {
+ from: (table: "ai_run_results") => RunResultsTable;
+}
+
+async function bounded(label: string, write: () => WriteResult, timeoutMs = WRITE_TIMEOUT_MS): Promise {
+ let timer: ReturnType | undefined;
+ try {
+ const outcome = await Promise.race([
+ Promise.resolve(write()),
+ new Promise<"timeout">((resolve) => {
+ timer = setTimeout(() => resolve("timeout"), timeoutMs);
+ }),
+ ]);
+ if (outcome === "timeout") console.warn(`[AI_RUN_RESULT] ${label} timed out`);
+ else if (outcome.error) console.warn(`[AI_RUN_RESULT] ${label} failed:`, outcome.error.code ?? outcome.error.message?.slice(0, 200));
+ } catch (error) {
+ console.warn(`[AI_RUN_RESULT] ${label} failed:`, error instanceof Error ? error.message.slice(0, 200) : "unknown");
+ } finally {
+ if (timer) clearTimeout(timer);
+ }
+}
+
+export function createRunResultRecorder(
+ database: RunResultsDatabase,
+ userId: string,
+ feature: string,
+ resultKey: string,
+): RunResultRecorder {
+ // The start write itself, which may outlive the begin() wait.
+ let started: Promise = Promise.resolve();
+ return {
+ begin: () => bounded("begin", () => {
+ const write = Promise.resolve(database.from("ai_run_results").upsert({
+ user_id: userId,
+ feature,
+ result_key: resultKey,
+ status: null,
+ body: null,
+ started_at: new Date().toISOString(),
+ finished_at: null,
+ }, { onConflict: "user_id,feature" }));
+ started = write.catch(() => undefined);
+ return write;
+ }),
+ finish: async (result, budgetMs = BEGIN_SETTLE_MS + WRITE_TIMEOUT_MS) => {
+ const until = Date.now() + budgetMs;
+ const left = (cap: number) => Math.min(cap, Math.max(0, until - Date.now()));
+ await bounded("begin settle", () => started.then(() => ({ error: null })), left(BEGIN_SETTLE_MS));
+ await bounded("finish", () => database.from("ai_run_results")
+ .update({
+ status: typeof result.status === "number" ? result.status : 500,
+ body: result.body ?? null,
+ finished_at: new Date().toISOString(),
+ })
+ .eq("user_id", userId)
+ .eq("feature", feature)
+ // A newer run of the same feature took the row: leave its row alone.
+ .eq("result_key", resultKey), left(WRITE_TIMEOUT_MS));
+ },
+ };
+}
diff --git a/supabase/functions/ai/sparring-judge_test.ts b/supabase/functions/ai/sparring-judge_test.ts
new file mode 100644
index 000000000..b47c06366
--- /dev/null
+++ b/supabase/functions/ai/sparring-judge_test.ts
@@ -0,0 +1,206 @@
+// The Sparring judge retries once inside the request when its first attempt
+// gave no scorecard for a reason another try can fix, bills both attempts in
+// one usage row, voids the first attempt's streamed draft, and does not
+// retry a failure a retry cannot fix (the key's credit).
+
+import { assertEquals, assertRejects } from "jsr:@std/assert@1.0.19";
+
+for (const [key, value] of Object.entries({
+ SUPABASE_URL: "https://supabase.test",
+ SUPABASE_ANON_KEY: "test-anon-key",
+ SUPABASE_SERVICE_ROLE_KEY: "test-service-role-key",
+ SPARRING_INTEGRITY_SECRET: "test-sparring-secret",
+ OPENROUTER_API_KEY: "test-openrouter-key",
+})) Deno.env.set(key, value);
+
+const { handleSparring } = await import("./features/sparring.ts");
+const { AIResponseValidationError } = await import("./provider.ts");
+const { narrowSparringDatabase } = await import("./runtime.ts");
+type FeatureContext = import("./features/types.ts").FeatureContext;
+type SparringRequest = import("./sparring-persistence.ts").SparringRequest;
+
+const MODEL = "anthropic/claude-sonnet-5.5";
+const SCORECARD = JSON.stringify({
+ action: "score",
+ scorecard: {
+ userScore: 1,
+ aiScore: 1,
+ categories: [
+ { name: "Logic", userScore: 20, aiScore: 10, feedback: "Logic feedback" },
+ { name: "Evidence", userScore: 20, aiScore: 15, feedback: "Evidence feedback" },
+ { name: "Delivery", userScore: 20, aiScore: 14, feedback: "Delivery feedback" },
+ { name: "Strategy", userScore: 20, aiScore: 13, feedback: "Strategy feedback" },
+ ],
+ overallFeedback: "The user controlled the comparison.",
+ },
+ keyMoments: ["The first clash"],
+ improvementAreas: ["Make the warrant explicit"],
+ winner: "user",
+ verdict: "User edge",
+});
+
+const request = {
+ action: "score",
+ sessionId: "00000000-0000-4000-8000-000000000031",
+ motion: "This house would regulate social media algorithms",
+ userSide: "pro",
+ format: "ld",
+ difficulty: "medium",
+ language: "en",
+ speeches: [{
+ id: "speech-1",
+ turnId: "ld-1",
+ speaker: "user",
+ side: "pro",
+ role: "Affirmative constructive",
+ content: "Algorithms shape what people believe.",
+ createdAt: 1,
+ timeLimitSeconds: 360,
+ timeUsedSeconds: 20,
+ }],
+} as unknown as SparringRequest;
+
+type Reply = { content: string } | { status: number };
+
+function install(replies: Reply[]) {
+ const calls: Array> = [];
+ const usageRows: Array> = [];
+ const originalFetch = globalThis.fetch;
+ globalThis.fetch = (async (input: Request | URL | string, init?: RequestInit) => {
+ const url = new URL(input instanceof Request ? input.url : String(input));
+ if (url.pathname.endsWith("/rest/v1/ai_requests")) {
+ usageRows.push(JSON.parse(String(init?.body)));
+ return new Response(null, { status: 201 });
+ }
+ if (url.href === "https://openrouter.ai/api/v1/chat/completions") {
+ calls.push(JSON.parse(String(init?.body)));
+ const reply = replies[Math.min(calls.length - 1, replies.length - 1)];
+ if ("status" in reply) return Response.json({ error: { message: "upstream" } }, { status: reply.status });
+ if (calls.at(-1)?.stream === true) {
+ const frames = [
+ { choices: [{ delta: { content: reply.content }, finish_reason: "stop" }] },
+ { choices: [], usage: { prompt_tokens: 10, completion_tokens: 5, cost: 0 } },
+ ];
+ return new Response(`${frames.map((frame) => `data: ${JSON.stringify(frame)}\n\n`).join("")}data: [DONE]\n\n`, { headers: { "Content-Type": "text/event-stream" } });
+ }
+ return Response.json({ choices: [{ message: { content: reply.content }, finish_reason: "stop" }], usage: { prompt_tokens: 10, completion_tokens: 5, cost: 0 } });
+ }
+ throw new Error(`Unexpected request: ${url}`);
+ }) as typeof fetch;
+ return { calls, usageRows, restore: () => { globalThis.fetch = originalFetch; } };
+}
+
+function context(events: string[]): FeatureContext {
+ return {
+ req: new Request("http://127.0.0.1/functions/v1/ai"),
+ userId: "user-1",
+ provider: "openrouter",
+ model: MODEL,
+ reasoningEffort: "low",
+ workSignal: new AbortController().signal,
+ voice: false,
+ stream: { delta: () => {}, reset: () => events.push("reset"), phase: () => {} },
+ input: {},
+ sources: [],
+ sourceAssessments: [],
+ sourceDecisionCandidates: [],
+ researchDeps: {} as FeatureContext["researchDeps"],
+ startTime: Date.now(),
+ researchUsageForLog: () => [],
+ recordMainResult: (result) => result,
+ buildSystemPromptFor: () => Promise.resolve("system"),
+ markTruncated: () => {},
+ respond: (payload) => Promise.resolve(Response.json(payload)),
+ };
+}
+
+const savedScorecards: unknown[] = [];
+const database = narrowSparringDatabase({
+ from: () => {
+ const query = {
+ update: (row: Record) => {
+ savedScorecards.push(row.final_evaluation);
+ return query;
+ },
+ eq: () => query,
+ select: () => query,
+ maybeSingle: () => Promise.resolve({ data: { id: request.sessionId }, error: null }),
+ };
+ return query;
+ },
+});
+
+const options = { request, database, claimToken: "claim-token", onScoreSaved: () => {} };
+
+Deno.test("a scorecard in the wrong shape gets one repair retry naming the field, and both attempts are billed", async () => {
+ const harness = install([{ content: JSON.stringify({ action: "score", scorecard: {} }) }, { content: SCORECARD }]);
+ const events: string[] = [];
+ savedScorecards.length = 0;
+ const originalWarn = console.warn;
+ console.warn = () => undefined;
+ try {
+ const response = await handleSparring(context(events), options);
+ assertEquals(response.status, 200);
+ assertEquals(harness.calls.length, 2);
+ for (const call of harness.calls) {
+ const format = call.response_format as { type?: string; json_schema?: { name?: string; strict?: boolean } };
+ assertEquals([format.type, format.json_schema?.name, format.json_schema?.strict], ["json_schema", "sparring_scorecard", true]);
+ }
+ const [system, ...rest] = harness.calls[1].messages as Array<{ role: string; content: unknown }>;
+ assertEquals(JSON.stringify(system.content).includes('did not match the required shape at \\"scorecard.categories\\"'), true, "the repair prompt does not name the field");
+ assertEquals(String(rest.at(-1)?.content).includes("Return only valid JSON matching this schema"), true);
+ assertEquals(events, ["reset"], "the first attempt's draft was not voided");
+ assertEquals(savedScorecards.length, 1);
+ assertEquals(harness.usageRows.length, 1, "exactly one usage row for the run");
+ assertEquals([harness.usageRows[0].input_tokens, harness.usageRows[0].output_tokens], [20, 10]);
+ } finally {
+ console.warn = originalWarn;
+ harness.restore();
+ }
+});
+
+Deno.test("a provider outage gets the retry; an exhausted key does not", async () => {
+ const originalError = console.error;
+ const originalWarn = console.warn;
+ console.error = () => undefined;
+ console.warn = () => undefined;
+ try {
+ const outage = install([{ status: 503 }, { content: SCORECARD }]);
+ try {
+ assertEquals((await handleSparring(context([]), options)).status, 200);
+ assertEquals(outage.calls.length, 2);
+ } finally {
+ outage.restore();
+ }
+
+ const noCredit = install([{ status: 402 }, { content: SCORECARD }]);
+ try {
+ await assertRejects(() => handleSparring(context([]), options));
+ assertEquals(noCredit.calls.length, 1, "a key out of credit was retried");
+ } finally {
+ noCredit.restore();
+ }
+ } finally {
+ console.error = originalError;
+ console.warn = originalWarn;
+ }
+});
+
+Deno.test("a scorecard still in the wrong shape after its repair fails with the typed error and the field", async () => {
+ // The fixture places verdict beside the scorecard; an empty one cannot be repaired.
+ const malformed = { content: JSON.stringify({ ...JSON.parse(SCORECARD), verdict: "" }) };
+ const harness = install([malformed, malformed, { content: SCORECARD }]);
+ const originalWarn = console.warn;
+ console.warn = () => undefined;
+ try {
+ const error = await assertRejects(() => handleSparring(context([]), options));
+ assertEquals(error instanceof AIResponseValidationError, true);
+ assertEquals((error as InstanceType).path, "scorecard.verdict");
+ assertEquals(harness.calls.length, 2, "only one repair retry");
+ assertEquals(harness.usageRows.length, 0, "the entrypoint, not the handler, logs a failed run");
+ assertEquals((error as InstanceType).usage?.inputTokens, 20);
+ } finally {
+ console.warn = originalWarn;
+ harness.restore();
+ }
+});
diff --git a/supabase/functions/ai/sparring-persistence.test.ts b/supabase/functions/ai/sparring-persistence.test.ts
index d38e9aa0c..e5f6ac246 100644
--- a/supabase/functions/ai/sparring-persistence.test.ts
+++ b/supabase/functions/ai/sparring-persistence.test.ts
@@ -1,4 +1,5 @@
import {
+ checkSparringProviderOutput,
claimSparringScoreSession,
completeClaimedSparringScoreSession,
createSparringSpeechIntegrityToken,
@@ -82,6 +83,84 @@ Deno.test("normalizes misplaced score fields and derives totals and winner", ()
}
});
+Deno.test("repairs a stray overallFeedback and object list items, the shapes Sonnet returns", () => {
+ const raw = {
+ action: "score",
+ scorecard: {
+ userScore: 0,
+ aiScore: 0,
+ categories: [
+ { name: "Logic", userScore: 20, aiScore: 10, feedback: "Logic feedback" },
+ { name: "Evidence", userScore: 20, aiScore: 15, feedback: "Evidence feedback" },
+ { name: "Delivery", userScore: 20, aiScore: 14, feedback: "Delivery feedback" },
+ { name: "Strategy", userScore: 20, aiScore: 13, feedback: "Strategy feedback" },
+ ],
+ keyMoments: [{ moment: "Con R1 dropped the cost warrant" }, { text: "Pro R2 weighed probability" }, "A plain moment"],
+ improvementAreas: [{ description: "Signpost the second contention" }],
+ winner: "user",
+ verdict: "User edge",
+ },
+ overallFeedback: "The user controlled the comparison.",
+ };
+ const parsed = parseSparringProviderOutput(normalizeSparringOutput(raw, "score"), "score");
+ if (!parsed || parsed.action !== "score") throw new Error("the repaired scorecard was rejected");
+ if (parsed.scorecard.overallFeedback !== "The user controlled the comparison.") throw new Error("overallFeedback was not moved into the scorecard");
+ if (JSON.stringify(parsed.scorecard.keyMoments) !== JSON.stringify(["Con R1 dropped the cost warrant", "Pro R2 weighed probability", "A plain moment"])) {
+ throw new Error(`keyMoments were not flattened: ${JSON.stringify(parsed.scorecard.keyMoments)}`);
+ }
+ if (parsed.scorecard.improvementAreas[0] !== "Signpost the second contention") throw new Error("improvementAreas were not flattened");
+
+ // An object without a known text field is not guessed at.
+ const unknownItem = { ...raw, scorecard: { ...raw.scorecard, keyMoments: [{ note: "?" }] } };
+ const checked = checkSparringProviderOutput(normalizeSparringOutput(unknownItem, "score"), "score");
+ if (checked.ok || checked.path !== "scorecard.keyMoments[0]") throw new Error(`unexpected result: ${JSON.stringify(checked)}`);
+});
+
+Deno.test("a rejected reply names the failing field and never echoes the model's text", () => {
+ const card = (scorecard: Record) => ({
+ action: "score",
+ scorecard: {
+ userScore: 80,
+ aiScore: 52,
+ categories: [
+ { name: "Logic", userScore: 20, aiScore: 10, feedback: "SECRET feedback" },
+ { name: "Evidence", userScore: 20, aiScore: 15, feedback: "f" },
+ { name: "Delivery", userScore: 20, aiScore: 14, feedback: "f" },
+ { name: "Strategy", userScore: 20, aiScore: 13, feedback: "f" },
+ ],
+ overallFeedback: "SECRET overall",
+ keyMoments: ["k"],
+ improvementAreas: ["i"],
+ winner: "user",
+ verdict: "SECRET verdict",
+ ...scorecard,
+ },
+ });
+ const pathFor = (value: unknown, action: "score" | "respond" = "score") => {
+ const checked = checkSparringProviderOutput(value, action);
+ if (checked.ok) throw new Error("an invalid reply was accepted");
+ if (/SECRET/.test(checked.path)) throw new Error("the path carried model text");
+ return checked.path;
+ };
+ const categories = card({}).scorecard.categories;
+ const cases: Array<[unknown, string]> = [
+ [card({ verdict: undefined }), "scorecard.verdict"],
+ [card({ categories: categories.slice(0, 3) }), "scorecard.categories.length"],
+ [card({ categories: [categories[0], { ...categories[1], userScore: 30 }, categories[2], categories[3]] }), "scorecard.categories[1].userScore"],
+ [card({ categories: [categories[0], categories[0], categories[2], categories[3]] }), "scorecard.categories.name"],
+ [card({ keyMoments: Array(13).fill("k") }), "scorecard.keyMoments.length"],
+ [card({ winner: "judge" }), "scorecard.winner"],
+ [{ action: "respond", speech: { content: "SECRET speech", targetedPoints: [] } }, "action"],
+ ];
+ for (const [value, path] of cases) {
+ const found = pathFor(value);
+ if (found !== path) throw new Error(`expected ${path}, got ${found}`);
+ }
+ if (pathFor({ action: "respond", speech: { content: "SECRET speech", targetedPoints: [] } }, "respond") !== "speech.strategy") {
+ throw new Error("a speech without a strategy did not name speech.strategy");
+ }
+});
+
Deno.test("rejects invalid sparring request invariants", () => {
const responseTurn = parseSparringRequest({
...request,
diff --git a/supabase/functions/ai/sparring-persistence.ts b/supabase/functions/ai/sparring-persistence.ts
index ab6cdc380..012064bbf 100644
--- a/supabase/functions/ai/sparring-persistence.ts
+++ b/supabase/functions/ai/sparring-persistence.ts
@@ -1,3 +1,5 @@
+import type { JsonOutputSchema } from "../_shared/research.ts";
+
const UUID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
const MAX_MOTION_LENGTH = 1_200;
const MAX_LANGUAGE_LENGTH = 32;
@@ -404,43 +406,81 @@ export function parseSparringRequest(value: unknown): { request: SparringRequest
};
}
-function parseScoreCategory(value: unknown): SparringScorecardRecord["categories"][number] | null {
- if (!isRecord(value)) return null;
- const names: SparringScorecardRecord["categories"][number]["name"][] = ["Logic", "Evidence", "Delivery", "Strategy"];
- const name = typeof value.name === "string" ? names.find((candidate) => candidate === value.name) : undefined;
+const SCORE_CATEGORY_NAMES = ["Logic", "Evidence", "Delivery", "Strategy"] as const;
+const MAX_SCORE_LIST_ITEMS = 12;
+const MAX_TARGETED_POINTS = 8;
+
+/** A provider answer that fails the shape: `path` names the first failing field, never its content. */
+type Checked = { ok: true; value: T } | { ok: false; path: string };
+
+const fail = (path: string): { ok: false; path: string } => ({ ok: false, path });
+
+function checkScoreCategory(value: unknown, path: string): Checked {
+ if (!isRecord(value)) return fail(path);
+ const name = typeof value.name === "string" ? SCORE_CATEGORY_NAMES.find((candidate) => candidate === value.name) : undefined;
+ if (!name) return fail(`${path}.name`);
+ if (!integerInRange(value.userScore, 0, 25)) return fail(`${path}.userScore`);
+ if (!integerInRange(value.aiScore, 0, 25)) return fail(`${path}.aiScore`);
const feedback = cleanString(value.feedback, 2_000);
- if (!name || !feedback || !integerInRange(value.userScore, 0, 25) || !integerInRange(value.aiScore, 0, 25)) return null;
- return { name, userScore: value.userScore, aiScore: value.aiScore, feedback };
+ if (!feedback) return fail(`${path}.feedback`);
+ return { ok: true, value: { name, userScore: value.userScore, aiScore: value.aiScore, feedback } };
}
-export function parseSparringScorecard(value: unknown): SparringScorecardRecord | null {
- if (!isRecord(value)) return null;
- const categories = Array.isArray(value.categories) ? value.categories.map(parseScoreCategory) : null;
- if (!categories || categories.length !== 4 || categories.some((category) => category === null)) return null;
- const parsedCategories = categories.filter((category): category is SparringScorecardRecord["categories"][number] => category !== null);
- if (parsedCategories.length !== 4) return null;
- const requiredNames = new Set(["Logic", "Evidence", "Delivery", "Strategy"]);
- const categoryNames = parsedCategories.map((category) => category.name);
- if (new Set(categoryNames).size !== 4 || categoryNames.some((name) => !requiredNames.has(name))) return null;
+function checkStringList(value: unknown, path: string, maxItems: number, maxLength: number): Checked {
+ if (!Array.isArray(value)) return fail(path);
+ if (value.length > maxItems) return fail(`${path}.length`);
+ const items: string[] = [];
+ for (const [index, item] of value.entries()) {
+ const clean = cleanString(item, maxLength);
+ if (!clean) return fail(`${path}[${index}]`);
+ items.push(clean);
+ }
+ return { ok: true, value: items };
+}
+
+/** Validates a scorecard, or names the first field (as `scorecard.`) that is wrong. */
+export function checkSparringScorecard(value: unknown): Checked {
+ if (!isRecord(value)) return fail("scorecard");
+ if (!Array.isArray(value.categories)) return fail("scorecard.categories");
+ if (value.categories.length !== 4) return fail("scorecard.categories.length");
+ const categories: SparringScorecardRecord["categories"] = [];
+ for (const [index, item] of value.categories.entries()) {
+ const category = checkScoreCategory(item, `scorecard.categories[${index}]`);
+ if (!category.ok) return category;
+ categories.push(category.value);
+ }
+ if (new Set(categories.map((category) => category.name)).size !== 4) return fail("scorecard.categories.name");
const overallFeedback = cleanString(value.overallFeedback, 4_000);
+ if (!overallFeedback) return fail("scorecard.overallFeedback");
+ const keyMoments = checkStringList(value.keyMoments, "scorecard.keyMoments", MAX_SCORE_LIST_ITEMS, 1_000);
+ if (!keyMoments.ok) return keyMoments;
+ const improvementAreas = checkStringList(value.improvementAreas, "scorecard.improvementAreas", MAX_SCORE_LIST_ITEMS, 1_000);
+ if (!improvementAreas.ok) return improvementAreas;
+ if (!integerInRange(value.userScore, 0, 100)) return fail("scorecard.userScore");
+ if (!integerInRange(value.aiScore, 0, 100)) return fail("scorecard.aiScore");
+ if (value.winner !== "user" && value.winner !== "ai" && value.winner !== "tie") return fail("scorecard.winner");
const verdict = cleanString(value.verdict, 2_000);
- const keyMoments = Array.isArray(value.keyMoments) ? value.keyMoments.map((item) => cleanString(item, 1_000)) : null;
- const improvementAreas = Array.isArray(value.improvementAreas) ? value.improvementAreas.map((item) => cleanString(item, 1_000)) : null;
- if (!overallFeedback || !verdict || !keyMoments || !improvementAreas || keyMoments.length > 12 || improvementAreas.length > 12 || keyMoments.some((item) => !item) || improvementAreas.some((item) => !item)) return null;
- if (!integerInRange(value.userScore, 0, 100) || !integerInRange(value.aiScore, 0, 100)) return null;
- if (value.winner !== "user" && value.winner !== "ai" && value.winner !== "tie") return null;
+ if (!verdict) return fail("scorecard.verdict");
return {
- userScore: value.userScore,
- aiScore: value.aiScore,
- categories: parsedCategories,
- overallFeedback,
- keyMoments: keyMoments.filter((item): item is string => Boolean(item)),
- improvementAreas: improvementAreas.filter((item): item is string => Boolean(item)),
- winner: value.winner,
- verdict,
+ ok: true,
+ value: {
+ userScore: value.userScore,
+ aiScore: value.aiScore,
+ categories,
+ overallFeedback,
+ keyMoments: keyMoments.value,
+ improvementAreas: improvementAreas.value,
+ winner: value.winner,
+ verdict,
+ },
};
}
+export function parseSparringScorecard(value: unknown): SparringScorecardRecord | null {
+ const checked = checkSparringScorecard(value);
+ return checked.ok ? checked.value : null;
+}
+
function scorecardMatchesDerivedValues(scorecard: SparringScorecardRecord): boolean {
const userScore = scorecard.categories.reduce((total, category) => total + category.userScore, 0);
const aiScore = scorecard.categories.reduce((total, category) => total + category.aiScore, 0);
@@ -448,42 +488,63 @@ function scorecardMatchesDerivedValues(scorecard: SparringScorecardRecord): bool
return scorecard.userScore === userScore && scorecard.aiScore === aiScore && scorecard.winner === winner;
}
-export function parseSparringProviderOutput(value: unknown, expectedAction: SparringRequest["action"]): SparringProviderOutput | null {
- if (!isRecord(value) || value.action !== expectedAction) return null;
+/** Validates a provider answer for the action, or names the first field that is wrong. */
+export function checkSparringProviderOutput(value: unknown, expectedAction: SparringRequest["action"]): Checked {
+ if (!isRecord(value)) return fail("$");
+ if (value.action !== expectedAction) return fail("action");
if (expectedAction === "respond") {
- if (!isRecord(value.speech)) return null;
+ if (!isRecord(value.speech)) return fail("speech");
const content = cleanString(value.speech.content, MAX_SPEECH_CONTENT_LENGTH);
+ if (!content) return fail("speech.content");
const strategy = cleanString(value.speech.strategy, 2_000);
- const targetedPoints = Array.isArray(value.speech.targetedPoints)
- ? value.speech.targetedPoints.map((point) => cleanString(point, 500))
- : null;
- if (!content || !strategy || !targetedPoints || targetedPoints.length > 8 || targetedPoints.some((point) => !point)) return null;
- return {
- action: "respond",
- speech: { content, strategy, targetedPoints: targetedPoints.filter((point): point is string => Boolean(point)) },
- };
+ if (!strategy) return fail("speech.strategy");
+ const targetedPoints = checkStringList(value.speech.targetedPoints, "speech.targetedPoints", MAX_TARGETED_POINTS, 500);
+ if (!targetedPoints.ok) return targetedPoints;
+ return { ok: true, value: { action: "respond", speech: { content, strategy, targetedPoints: targetedPoints.value } } };
}
- const scorecard = parseSparringScorecard(value.scorecard);
- return scorecard ? { action: "score", scorecard } : null;
+ const scorecard = checkSparringScorecard(value.scorecard);
+ return scorecard.ok ? { ok: true, value: { action: "score", scorecard: scorecard.value } } : scorecard;
+}
+
+export function parseSparringProviderOutput(value: unknown, expectedAction: SparringRequest["action"]): SparringProviderOutput | null {
+ const checked = checkSparringProviderOutput(value, expectedAction);
+ return checked.ok ? checked.value : null;
+}
+
+// Narrative list items some models write as objects instead of strings.
+const LIST_ITEM_TEXT_FIELDS = ["text", "moment", "description"] as const;
+
+function flattenListItems(value: unknown): unknown {
+ if (!Array.isArray(value)) return value;
+ return value.map((item) => {
+ if (!isRecord(item)) return item;
+ const field = LIST_ITEM_TEXT_FIELDS.find((name) => typeof item[name] === "string");
+ return field ? item[field] : item;
+ });
}
/**
- * Repair the one common provider shape error without filling in omitted data.
- * Some JSON models place scorecard narrative fields beside `scorecard` rather
- * than inside it. Only known fields are moved, then totals and winner are
- * derived from the four validated category scores.
+ * Repair the common provider shape errors without filling in omitted data.
+ * Some models place scorecard narrative fields beside `scorecard` rather than
+ * inside it, or write a key moment or improvement area as an object holding
+ * its text. Only known fields are moved, and only a text, moment, or
+ * description string is taken from an object; then totals and winner are
+ * derived from the four validated category scores. An answer that still does
+ * not parse is returned with those repairs, so its failing field is the real one.
*/
export function normalizeSparringOutput(value: unknown, expectedAction: SparringRequest["action"]): unknown {
if (expectedAction !== "score" || !isRecord(value) || value.action !== "score" || !isRecord(value.scorecard)) return value;
const scorecard = { ...value.scorecard };
- const movableFields = ["keyMoments", "improvementAreas", "winner", "verdict"] as const;
+ const movableFields = ["overallFeedback", "keyMoments", "improvementAreas", "winner", "verdict"] as const;
for (const field of movableFields) {
if (scorecard[field] === undefined && value[field] !== undefined) scorecard[field] = value[field];
}
+ scorecard.keyMoments = flattenListItems(scorecard.keyMoments);
+ scorecard.improvementAreas = flattenListItems(scorecard.improvementAreas);
const parsed = parseSparringScorecard(scorecard);
- if (!parsed) return value;
+ if (!parsed) return { action: "score", scorecard };
const userScore = parsed.categories.reduce((total, category) => total + category.userScore, 0);
const aiScore = parsed.categories.reduce((total, category) => total + category.aiScore, 0);
const winner = userScore === aiScore ? "tie" : userScore > aiScore ? "user" : "ai";
@@ -493,6 +554,82 @@ export function normalizeSparringOutput(value: unknown, expectedAction: Sparring
};
}
+const PROSE = { type: "string" };
+const STRING_LIST = (description: string) => ({ type: "array", items: PROSE, description });
+
+/**
+ * The strict json_schema each sparring call is constrained to. Anthropic's
+ * structured outputs take no numeric or array-size constraints, so the 0-25
+ * category range, the four categories, and the list limits are stated in
+ * descriptions and enforced by checkSparringProviderOutput.
+ */
+export function sparringOutputSchema(action: SparringRequest["action"]): JsonOutputSchema {
+ if (action === "respond") {
+ return {
+ name: "sparring_speech",
+ schema: {
+ type: "object",
+ properties: {
+ action: { type: "string", enum: ["respond"] },
+ speech: {
+ type: "object",
+ properties: {
+ content: PROSE,
+ strategy: PROSE,
+ targetedPoints: STRING_LIST(`At most ${MAX_TARGETED_POINTS} transcript points`),
+ },
+ required: ["content", "strategy", "targetedPoints"],
+ additionalProperties: false,
+ },
+ },
+ required: ["action", "speech"],
+ additionalProperties: false,
+ },
+ };
+ }
+ const score = (description: string) => ({ type: "integer", description });
+ return {
+ name: "sparring_scorecard",
+ schema: {
+ type: "object",
+ properties: {
+ action: { type: "string", enum: ["score"] },
+ scorecard: {
+ type: "object",
+ properties: {
+ userScore: score("Sum of the user's four category scores, 0-100"),
+ aiScore: score("Sum of the AI's four category scores, 0-100"),
+ categories: {
+ type: "array",
+ description: "Exactly four categories, one each of Logic, Evidence, Delivery, Strategy",
+ items: {
+ type: "object",
+ properties: {
+ name: { type: "string", enum: [...SCORE_CATEGORY_NAMES] },
+ userScore: score("Integer 0-25"),
+ aiScore: score("Integer 0-25"),
+ feedback: PROSE,
+ },
+ required: ["name", "userScore", "aiScore", "feedback"],
+ additionalProperties: false,
+ },
+ },
+ overallFeedback: PROSE,
+ keyMoments: STRING_LIST(`At most ${MAX_SCORE_LIST_ITEMS} key moments, each a plain string`),
+ improvementAreas: STRING_LIST(`At most ${MAX_SCORE_LIST_ITEMS} improvement areas, each a plain string`),
+ winner: { type: "string", enum: ["user", "ai", "tie"] },
+ verdict: PROSE,
+ },
+ required: ["userScore", "aiScore", "categories", "overallFeedback", "keyMoments", "improvementAreas", "winner", "verdict"],
+ additionalProperties: false,
+ },
+ },
+ required: ["action", "scorecard"],
+ additionalProperties: false,
+ },
+ };
+}
+
export function sparringPersona(difficulty: SparringRequest["difficulty"]): "coach" | "analytical" | "aggressive" {
return difficulty === "easy" ? "coach" : difficulty === "medium" ? "analytical" : "aggressive";
}
diff --git a/supabase/functions/ai/stream.test.ts b/supabase/functions/ai/stream.test.ts
index 5e17453d1..d79ca9def 100644
--- a/supabase/functions/ai/stream.test.ts
+++ b/supabase/functions/ai/stream.test.ts
@@ -35,6 +35,8 @@ type ProviderReply = { content: string; finishReason?: string } | { status: numb
interface Harness {
providerBodies: Array>;
usageRows: Array>;
+ /** Writes to ai_run_results, in order: the method, the query, the row. */
+ runResultWrites: Array<{ method: string; query: string; row: Record }>;
restore: () => void;
}
@@ -58,7 +60,7 @@ function sseBody(content: string, finishReason = "stop"): ReadableStream { globalThis.fetch = originalFetch; } };
+ const harness: Harness = { providerBodies: [], usageRows: [], runResultWrites: [], restore: () => { globalThis.fetch = originalFetch; } };
globalThis.fetch = (async (input: Request | URL | string, init?: RequestInit) => {
const url = new URL(input instanceof Request ? input.url : String(input));
if (url.origin === "https://supabase.test") {
@@ -74,6 +76,10 @@ function install(replies: ProviderReply[]): Harness {
if (url.pathname.endsWith("/rest/v1/rpc/claim_ai_session_quota")) return jsonResponse([{ allowed: true, charged: true, units: 1, current_count: 1, charge_id: 1 }]);
if (/\/rest\/v1\/rpc\/refund_(daily_quota|ai_session_quota)$/.test(url.pathname)) return jsonResponse(1);
if (url.pathname.includes("/rest/v1/global_prompts")) return jsonResponse({ prompt_text: null });
+ if (url.pathname.includes("/rest/v1/ai_run_results")) {
+ harness.runResultWrites.push({ method: init?.method ?? "GET", query: decodeURIComponent(url.search), row: JSON.parse(String(init?.body)) });
+ return new Response(null, { status: 204 });
+ }
if (url.pathname.includes("/rest/v1/ai_requests")) {
const row = JSON.parse(String(init?.body));
harness.usageRows.push(Array.isArray(row) ? row[0] : row);
@@ -121,11 +127,11 @@ const REPLY = JSON.stringify({
speech: { content: "Free fares sound generous. Who pays for the buses? \"Nobody\" is not an answer.", strategy: "Press the funding gap.", targetedPoints: ["price barrier"] },
});
-function aiRequest(stream: boolean): Request {
+function aiRequest(stream: boolean, extra: Record = {}): Request {
return new Request("http://127.0.0.1/functions/v1/ai", {
method: "POST",
headers: { Authorization: "Bearer test-user-token", "Content-Type": "application/json" },
- body: JSON.stringify({ model: "openai/gpt-6-luna", feature: "sparring", input: respondInput, ...(stream ? { stream: true } : {}) }),
+ body: JSON.stringify({ model: "openai/gpt-6-luna", feature: "sparring", input: respondInput, ...(stream ? { stream: true } : {}), ...extra }),
});
}
@@ -287,3 +293,123 @@ Deno.test("frame overhead does not count against the answer's size cap", async (
const { text } = await readOpenRouterStream(body, ignore, new AbortController().signal);
assert(text.length === tokens * 5, `the answer was cut: ${text.length}`);
});
+
+Deno.test("a streamed run with a result key stores its result before sending it", async () => {
+ const harness = install([{ content: REPLY }]);
+ try {
+ const events = await readEvents(await handleRequest(aiRequest(true, { resultKey: "req_1_abc" })));
+ const result = events.at(-1);
+ assert(result?.event === "result" && result.data.status === 200, "the run did not succeed");
+ const [begin, finish] = harness.runResultWrites;
+ assert(harness.runResultWrites.length === 2, `expected a start and a finish write, got ${JSON.stringify(harness.runResultWrites)}`);
+ assert(begin.method === "POST" && begin.query.includes("on_conflict=user_id,feature"), `the start was not an upsert: ${JSON.stringify(begin)}`);
+ assert(begin.row.result_key === "req_1_abc" && begin.row.status === null && begin.row.user_id === USER, `unexpected start row: ${JSON.stringify(begin.row)}`);
+ assert(finish.method === "PATCH" && finish.query.includes("result_key=eq.req_1_abc"), `the finish could overwrite a newer run: ${JSON.stringify(finish)}`);
+ assert(finish.row.status === 200 && JSON.stringify(finish.row.body) === JSON.stringify(result.data.body), "the stored result is not the one sent");
+ } finally {
+ harness.restore();
+ }
+
+ // Older clients send no key and nothing is stored.
+ const plain = install([{ content: REPLY }]);
+ try {
+ await readEvents(await handleRequest(aiRequest(true)));
+ assert(plain.runResultWrites.length === 0, "a run without a result key was stored");
+ } finally {
+ plain.restore();
+ }
+});
+
+Deno.test("the worker is held until the browser has read the last event", async () => {
+ const registered: Promise[] = [];
+ const runtime = globalThis as { EdgeRuntime?: { waitUntil: (promise: Promise) => void } };
+ runtime.EdgeRuntime = { waitUntil: (promise) => registered.push(promise) };
+ const harness = install([{ content: REPLY }]);
+ try {
+ const response = await handleRequest(aiRequest(true));
+ assert(registered.length >= 2, `expected the run and the stream registered, got ${registered.length}`);
+ const held = Promise.all(registered);
+ const early = await Promise.race([held.then(() => "released"), new Promise((resolve) => setTimeout(() => resolve("held"), 50))]);
+ assert(early === "held", "the worker was released before the browser read the stream");
+ const events = await readEvents(response);
+ assert(events.at(-1)?.event === "result", "the stream did not end with its result");
+ await held;
+ } finally {
+ delete runtime.EdgeRuntime;
+ harness.restore();
+ }
+});
+
+Deno.test("a run that outlives the stream's deadline ends with its own refunded answer, stored first", async () => {
+ const { streamFeatureResponse } = await import("./stream.ts");
+ const cancel = new AbortController();
+ const stored: Array> = [];
+ const timedOut = { error: "AI request failed. Please try again.", code: "provider_error", charged: false, retryable: true };
+ const response = streamFeatureResponse({}, cancel, (stream) => new Promise((resolve) => {
+ stream.delta("{\"partial\":");
+ // Like a step that ignores the budget until it is aborted; its error path
+ // then refunds and says so.
+ cancel.signal.addEventListener("abort", () => resolve(new Response(JSON.stringify(timedOut), { status: 500 })));
+ }), {
+ deadlineMs: 30,
+ recorder: {
+ begin: async () => {},
+ finish: async (payload) => {
+ // Slow storage: the result must still wait for it.
+ await new Promise((resolve) => setTimeout(resolve, 20));
+ stored.push(payload);
+ },
+ },
+ });
+ const reader = response.body!.getReader();
+ const decoder = new TextDecoder();
+ let text = "";
+ let storedWhenResultArrived: number | undefined;
+ while (true) {
+ const { value, done } = await reader.read();
+ if (done) break;
+ text += decoder.decode(value, { stream: true });
+ if (storedWhenResultArrived === undefined && text.includes("event: result")) storedWhenResultArrived = stored.length;
+ }
+ assert(storedWhenResultArrived === 1, "the result was sent before it was stored");
+ const result = JSON.parse(/event: result\ndata: (.+)\n\n$/.exec(text)?.[1] ?? "null");
+ assert(result?.status === 500 && result.body.retryable === true, `expected a retryable failure, got ${JSON.stringify(result)}`);
+ assert(result.body.charged === false, `the refunded run's answer lost charged:false: ${JSON.stringify(result)}`);
+ assert((cancel.signal.reason as { name?: string })?.name === "TimeoutError", "the run was not aborted as a timeout");
+ assert(JSON.stringify(stored[0]) === JSON.stringify(result), "the stored result differs from the sent one");
+});
+
+Deno.test("a run that ignores the deadline abort still ends with a retryable timeout", async () => {
+ const { streamFeatureResponse } = await import("./stream.ts");
+ const cancel = new AbortController();
+ // Never settles, as a step stuck past its abort would.
+ const response = streamFeatureResponse({}, cancel, () => new Promise(() => {}), { deadlineMs: 10 });
+ const events = await readEvents(response);
+ const result = events.at(-1);
+ assert(result?.event === "result" && result.data.status === 500, `expected a failure result, got ${JSON.stringify(result)}`);
+ const body = result.data.body as Record;
+ assert(body.retryable === true && !("charged" in body), `a run that never answered cannot claim a refund: ${JSON.stringify(body)}`);
+});
+
+Deno.test("a hung database cannot hold the result past the worker's end", async () => {
+ const { streamFeatureResponse } = await import("./stream.ts");
+ const { createRunResultRecorder } = await import("./run-results.ts");
+ const hung = new Promise<{ error: null }>(() => {});
+ const database = {
+ from: () => ({
+ upsert: () => hung,
+ update: () => ({ eq: () => ({ eq: () => ({ eq: () => hung }) }) }),
+ }),
+ };
+ const began = Date.now();
+ const response = streamFeatureResponse({}, new AbortController(), async () => new Response("{\"ok\":true}"), {
+ recorder: createRunResultRecorder(database, USER, "sparring", "req_hung"),
+ // Without the end the recorder would wait 4 s for the start write and
+ // up to 14 s more to finish.
+ endMs: 1_200,
+ });
+ const events = await readEvents(response);
+ const elapsed = Date.now() - began;
+ assert(events.at(-1)?.event === "result" && events.at(-1)?.data.status === 200, "the answer was not sent");
+ assert(elapsed < 1_500, `the result waited ${elapsed} ms on a hung database`);
+});
diff --git a/supabase/functions/ai/stream.ts b/supabase/functions/ai/stream.ts
index be9e1792d..311c3d38e 100644
--- a/supabase/functions/ai/stream.ts
+++ b/supabase/functions/ai/stream.ts
@@ -9,10 +9,33 @@
// `result` comes last, exactly once, and carries the status and JSON body the
// non-streamed request would have answered with. Every check that runs before
// the feature handler still answers with plain JSON and its real status.
+//
+// The worker is kept alive (EdgeRuntime.waitUntil) from the first event until
+// the HTTP layer has taken the last byte. Supabase otherwise counts a stream
+// as answered once its headers are out and may retire the worker mid-run.
+// With a recorder, `result` is stored before it is sent, so a browser that
+// loses the stream can still read it. Storing it and waiting for the browser
+// to read it are cut short to end before the worker's wall clock.
+import { keepWorkerAlive } from "../_shared/background.ts";
import type { GenerationPhase, GenerationStream } from "./provider.ts";
+import type { RunResultRecorder } from "./run-results.ts";
const HEARTBEAT_MS = 15_000;
+// After `result`, hold the worker at most this long for the browser to take
+// the last bytes. A client that stopped reading must not hold it to 150 s.
+const DRAIN_CAP_MS = 10_000;
+// Bytes the HTTP layer took on the last read still have to reach the socket.
+const SETTLE_MS = 500;
+// After a deadline abort, how long the run's own error path gets to answer.
+// Its answer says whether the units went back (`charged: false`).
+const DEADLINE_GRACE_MS = 2_000;
+// Kept free after the result is stored, for the browser to read it.
+const SEND_RESERVE_MS = 1_000;
+
+const FAILED_RESULT = { status: 500, body: { error: "AI request failed. Please try again.", code: "provider_error" } };
+// Running out of time is worth another try; the browser says so.
+const DEADLINE_RESULT = { status: 500, body: { ...FAILED_RESULT.body, retryable: true } };
function sse(event: string, payload: Record): string {
return `event: ${event}\ndata: ${JSON.stringify(payload)}\n\n`;
@@ -29,6 +52,35 @@ async function resultPayload(response: Response): Promise(work: Promise, ms: number | undefined, fallback: F): Promise {
+ if (ms === undefined) return await work;
+ let timer: ReturnType | undefined;
+ try {
+ return await Promise.race([work, new Promise((resolve) => {
+ timer = setTimeout(() => resolve(fallback), Math.max(0, ms));
+ })]);
+ } finally {
+ if (timer) clearTimeout(timer);
+ }
+}
+
+export interface StreamFeatureOptions {
+ /** Stores the final result before it is sent; the browser sent a result key. */
+ recorder?: RunResultRecorder;
+ /**
+ * Milliseconds from now after which the stream stops waiting for the run,
+ * aborts it, and sends a timeout result, so a run that outlives its budget
+ * ends with an answer instead of a dropped connection at the wall clock.
+ */
+ deadlineMs?: number;
+ /**
+ * Milliseconds from now by which the worker must be done. Storing the
+ * result and waiting for the browser to read it are bounded by what is left.
+ */
+ endMs?: number;
+}
+
/**
* Runs a feature with a stream sink and returns its events as the response.
* `cancel` aborts when the browser goes away so provider work stops with it.
@@ -37,49 +89,112 @@ export function streamFeatureResponse(
corsHeaders: Record,
cancel: AbortController,
run: (stream: GenerationStream) => Promise,
+ options: StreamFeatureOptions = {},
): Response {
+ const endAt = options.endMs === undefined ? undefined : Date.now() + options.endMs;
+ const timeLeft = (fallback: number) => endAt === undefined ? fallback : Math.min(fallback, Math.max(0, endAt - Date.now()));
const encoder = new TextEncoder();
+ // Events wait here until the HTTP layer reads them. The body below hands
+ // out one chunk per read, so when it closes the last event has been taken.
+ const queue: Uint8Array[] = [];
+ let ended = false;
+ let gone = false;
+ let wake: (() => void) | undefined;
+ let markDrained!: () => void;
+ const drained = new Promise((resolve) => {
+ markDrained = resolve;
+ });
+ const push = (chunk: string) => {
+ if (ended || gone) return;
+ queue.push(encoder.encode(chunk));
+ wake?.();
+ };
let heartbeat: ReturnType | undefined;
+ const stop = () => {
+ clearInterval(heartbeat);
+ heartbeat = undefined;
+ };
+
const body = new ReadableStream({
- async start(controller) {
- let open = true;
- const write = (chunk: string) => {
- if (!open) return;
- try {
- controller.enqueue(encoder.encode(chunk));
- } catch {
- open = false;
- }
- };
- // Research can run for a minute without output; keep proxies from
- // treating the connection as idle.
- heartbeat = setInterval(() => write(": keepalive\n\n"), HEARTBEAT_MS);
- let lastPhase: GenerationPhase | undefined;
- const stream: GenerationStream = {
- delta: (text) => write(sse("delta", { text })),
- reset: () => write(sse("reset", {})),
- phase: (phase) => {
- if (phase === lastPhase) return;
- lastPhase = phase;
- write(sse("phase", { phase }));
- },
- };
- try {
- write(sse("result", await resultPayload(await run(stream))));
- } catch (error) {
- console.error("[AI_STREAM] Stream failed:", error instanceof Error ? error.message.slice(0, 2000) : "Unknown error");
- write(sse("result", { status: 500, body: { error: "AI request failed. Please try again.", code: "provider_error" } }));
- } finally {
- clearInterval(heartbeat);
- open = false;
- try { controller.close(); } catch { /* already closed */ }
- }
+ pull(controller) {
+ return new Promise((resolve) => {
+ const deliver = () => {
+ wake = undefined;
+ const chunk = queue.shift();
+ if (chunk) controller.enqueue(chunk);
+ else if (ended) {
+ controller.close();
+ markDrained();
+ } else {
+ wake = deliver;
+ return;
+ }
+ resolve();
+ };
+ deliver();
+ });
},
cancel() {
- clearInterval(heartbeat);
+ gone = true;
+ queue.length = 0;
+ wake = undefined;
+ stop();
+ markDrained();
cancel.abort();
},
- });
+ }, { highWaterMark: 0 });
+
+ let lastPhase: GenerationPhase | undefined;
+ const stream: GenerationStream = {
+ delta: (text) => push(sse("delta", { text })),
+ reset: () => push(sse("reset", {})),
+ phase: (phase) => {
+ if (phase === lastPhase) return;
+ lastPhase = phase;
+ push(sse("phase", { phase }));
+ },
+ };
+
+ const lifecycle = (async () => {
+ // Research can run for a minute without output; keep proxies from
+ // treating the connection as idle.
+ heartbeat = setInterval(() => push(": keepalive\n\n"), HEARTBEAT_MS);
+ const began = options.recorder?.begin() ?? Promise.resolve();
+ const running = keepWorkerAlive(run(stream).then(resultPayload));
+ let payload: Record;
+ try {
+ const outcome = await within(running, options.deadlineMs, "deadline" as const);
+ if (outcome === "deadline") {
+ console.error("[AI_STREAM] Run passed its deadline; answering with a timeout");
+ // The run's own error path refunds and logs it once it sees the abort,
+ // and its answer carries `charged: false` when the units went back. A
+ // step that ignores the abort gets the plain timeout instead.
+ cancel.abort(new DOMException("The request ran past its time budget", "TimeoutError"));
+ const answered = await within(running.catch(() => undefined), DEADLINE_GRACE_MS, undefined);
+ payload = answered ?? DEADLINE_RESULT;
+ } else {
+ payload = outcome;
+ }
+ } catch (error) {
+ console.error("[AI_STREAM] Stream failed:", error instanceof Error ? error.message.slice(0, 2000) : "Unknown error");
+ payload = FAILED_RESULT;
+ }
+ if (options.recorder) {
+ // Stored before it is sent: a browser that loses the stream from here
+ // on can still read it. A slow database gets only the time left before
+ // the end, less what the browser needs to read the result.
+ const recordBy = endAt === undefined ? undefined : endAt - SEND_RESERVE_MS;
+ const recordLeft = () => recordBy === undefined ? undefined : Math.max(0, recordBy - Date.now());
+ await within(began, recordLeft(), undefined);
+ await options.recorder.finish(payload, recordLeft());
+ }
+ push(sse("result", payload));
+ stop();
+ ended = true;
+ wake?.();
+ await within(drained, timeLeft(DRAIN_CAP_MS), undefined);
+ })();
+ keepWorkerAlive(lifecycle, SETTLE_MS);
return new Response(body, {
headers: {
diff --git a/supabase/migrations/20261023140000_ai_run_results.sql b/supabase/migrations/20261023140000_ai_run_results.sql
new file mode 100644
index 000000000..78d92a2ac
--- /dev/null
+++ b/supabase/migrations/20261023140000_ai_run_results.sql
@@ -0,0 +1,77 @@
+-- The final answer of a streamed AI run, kept so a browser whose stream
+-- dropped can still read it (src/lib/ai/run-recovery.ts).
+--
+-- Additive only. The ai Edge Function (supabase/functions/ai/stream.ts) writes
+-- a row when a streamed run starts (status NULL) and fills in the status and
+-- body the stream's `result` event carries before it sends that event. A row
+-- left with status NULL long after started_at is a run the worker never
+-- finished. A function deployed before this migration finds no table, logs a
+-- warning, and streams as before.
+--
+-- One row per user and feature: the next run of a feature replaces the last,
+-- so the table stays bounded without a cleanup job. Only the service role
+-- writes it; a signed-in user reads their own rows.
+CREATE TABLE IF NOT EXISTS public.ai_run_results (
+ user_id uuid NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE,
+ feature text NOT NULL CHECK (feature IN ('generate', 'attack', 'brainstorm', 'factoids', 'ask', 'sparring', 'crossfire')),
+ -- The id the browser sent with the run; recovery reads only a matching row.
+ result_key text NOT NULL CHECK (result_key ~ '^[A-Za-z0-9._:-]{1,128}$'),
+ status integer CHECK (status BETWEEN 100 AND 599),
+ body jsonb,
+ started_at timestamptz NOT NULL DEFAULT now(),
+ finished_at timestamptz,
+ PRIMARY KEY (user_id, feature)
+);
+
+COMMENT ON TABLE public.ai_run_results IS
+ 'Latest streamed AI run per user and feature, written by the ai Edge Function so a dropped stream can recover its result.';
+
+ALTER TABLE public.ai_run_results ENABLE ROW LEVEL SECURITY;
+REVOKE ALL ON TABLE public.ai_run_results FROM PUBLIC, anon, authenticated;
+GRANT SELECT ON TABLE public.ai_run_results TO authenticated;
+GRANT SELECT, INSERT, UPDATE, DELETE ON TABLE public.ai_run_results TO service_role;
+
+CREATE POLICY "ai_run_results_select_own" ON public.ai_run_results
+ FOR SELECT TO authenticated
+ USING (user_id = (SELECT auth.uid()));
+
+-- Account erase also removes kept AI run results. The body is unchanged from
+-- 20260926090000_current_debates.sql apart from the ai_run_results line.
+CREATE OR REPLACE FUNCTION public.clear_current_user_data()
+RETURNS void
+LANGUAGE plpgsql
+SECURITY DEFINER
+SET search_path = public
+AS $$
+DECLARE
+ v_user_id uuid := auth.uid();
+BEGIN
+ IF v_user_id IS NULL THEN
+ RAISE EXCEPTION 'authentication required';
+ END IF;
+
+ PERFORM pg_advisory_xact_lock(hashtextextended(v_user_id::text, 0));
+ INSERT INTO public.user_data_clear_markers (user_id, cleared_at)
+ VALUES (v_user_id, clock_timestamp())
+ ON CONFLICT (user_id) DO UPDATE SET cleared_at = EXCLUDED.cleared_at;
+
+ DELETE FROM public.cases WHERE user_id = v_user_id;
+ DELETE FROM public.attacks WHERE user_id = v_user_id;
+ DELETE FROM public.brainstorms WHERE user_id = v_user_id;
+ DELETE FROM public.factoids WHERE user_id = v_user_id;
+ DELETE FROM public.chats WHERE user_id = v_user_id;
+ DELETE FROM public.flowcharts WHERE user_id = v_user_id;
+ DELETE FROM public.sparring_sessions WHERE user_id = v_user_id;
+ DELETE FROM public.crossfires WHERE user_id = v_user_id;
+ DELETE FROM public.evidence_cards WHERE user_id = v_user_id;
+ DELETE FROM public.sources WHERE user_id = v_user_id;
+ DELETE FROM public.source_folders WHERE user_id = v_user_id;
+ DELETE FROM public.api_keys WHERE user_id = v_user_id;
+ DELETE FROM public.current_debates WHERE user_id = v_user_id;
+ DELETE FROM public.debates WHERE user_id = v_user_id;
+ DELETE FROM public.ai_run_results WHERE user_id = v_user_id;
+END;
+$$;
+
+REVOKE ALL ON FUNCTION public.clear_current_user_data() FROM PUBLIC, anon;
+GRANT EXECUTE ON FUNCTION public.clear_current_user_data() TO authenticated;
diff --git a/supabase/tests/account_deletion_test.sql b/supabase/tests/account_deletion_test.sql
index 682736b25..04086b51e 100644
--- a/supabase/tests/account_deletion_test.sql
+++ b/supabase/tests/account_deletion_test.sql
@@ -111,6 +111,7 @@ INSERT INTO public.ai_requests (user_id, feature, provider, model)
VALUES ('ac000000-0000-4000-8000-00000000000a', 'brainstorm', 'openrouter', 'model');
INSERT INTO public.daily_usage_counters (user_id, usage_date) VALUES ('ac000000-0000-4000-8000-00000000000a', current_date);
INSERT INTO public.ai_quota_sessions (user_id, feature, session_key) VALUES ('ac000000-0000-4000-8000-00000000000a', 'crossfire', 'cf-leaver');
+INSERT INTO public.ai_run_results (user_id, feature, result_key, status, body) VALUES ('ac000000-0000-4000-8000-00000000000a', 'generate', 'req_1_leaver', 200, '{"output":{}}');
INSERT INTO public.rate_limits (user_id, limit_type, limit_value) VALUES ('ac000000-0000-4000-8000-00000000000a', 'daily', 5);
INSERT INTO public.api_keys (user_id, provider, key_ciphertext) VALUES ('ac000000-0000-4000-8000-00000000000a', 'openai', 'LEAVER-CIPHERTEXT');
INSERT INTO public.user_roles (user_id, role) VALUES ('ac000000-0000-4000-8000-00000000000a', 'admin');
diff --git a/supabase/tests/ai_run_results_test.sql b/supabase/tests/ai_run_results_test.sql
new file mode 100644
index 000000000..5758fc5d7
--- /dev/null
+++ b/supabase/tests/ai_run_results_test.sql
@@ -0,0 +1,69 @@
+BEGIN;
+SELECT plan(6);
+
+-- Returns how many rows a statement changed; a privilege or RLS rejection changes none.
+CREATE FUNCTION pg_temp.rows_changed(statement text)
+RETURNS bigint LANGUAGE plpgsql AS $$
+DECLARE changed bigint;
+BEGIN
+ EXECUTE statement;
+ GET DIAGNOSTICS changed = ROW_COUNT;
+ RETURN changed;
+EXCEPTION WHEN insufficient_privilege THEN
+ RETURN 0;
+END;
+$$;
+
+INSERT INTO auth.users (id)
+VALUES
+ ('a7000000-0000-4000-8000-000000000001'),
+ ('a7000000-0000-4000-8000-000000000002');
+
+SELECT ok(
+ NOT has_table_privilege('anon', 'public.ai_run_results', 'SELECT,INSERT,UPDATE,DELETE,TRUNCATE')
+ AND NOT has_table_privilege('authenticated', 'public.ai_run_results', 'INSERT,UPDATE,DELETE,TRUNCATE')
+ AND has_table_privilege('service_role', 'public.ai_run_results', 'SELECT')
+ AND has_table_privilege('service_role', 'public.ai_run_results', 'INSERT')
+ AND has_table_privilege('service_role', 'public.ai_run_results', 'UPDATE')
+ AND has_table_privilege('service_role', 'public.ai_run_results', 'DELETE'),
+ 'only the service role writes AI run results; anon cannot read them'
+);
+
+INSERT INTO public.ai_run_results (user_id, feature, result_key, status, body, finished_at)
+VALUES
+ ('a7000000-0000-4000-8000-000000000001', 'generate', 'req_1_own', 200, '{"output":{"case":"own"}}', now()),
+ ('a7000000-0000-4000-8000-000000000002', 'generate', 'req_1_other', 200, '{"output":{"case":"other"}}', now());
+
+SET LOCAL ROLE authenticated;
+SELECT set_config('request.jwt.claim.sub', 'a7000000-0000-4000-8000-000000000001', true);
+SELECT set_config('request.jwt.claim.role', 'authenticated', true);
+SELECT is(
+ (SELECT array_agg(result_key) FROM public.ai_run_results),
+ ARRAY['req_1_own'],
+ 'a signed-in user reads only their own run result'
+);
+SELECT is(
+ pg_temp.rows_changed($$UPDATE public.ai_run_results SET status = 500
+ WHERE user_id = 'a7000000-0000-4000-8000-000000000001'$$),
+ 0::bigint,
+ 'a signed-in user cannot rewrite their run result'
+);
+SELECT is(
+ pg_temp.rows_changed($$INSERT INTO public.ai_run_results (user_id, feature, result_key, status, body)
+ VALUES ('a7000000-0000-4000-8000-000000000001', 'attack', 'req_forged', 200, '{}')$$),
+ 0::bigint,
+ 'a signed-in user cannot plant a run result'
+);
+SELECT lives_ok(
+ $$SELECT public.clear_current_user_data()$$,
+ 'clearing data runs with the new table'
+);
+RESET ROLE;
+SELECT is(
+ (SELECT array_agg(result_key) FROM public.ai_run_results),
+ ARRAY['req_1_other'],
+ 'clearing data removes the user''s run results and no one else''s'
+);
+
+SELECT * FROM finish();
+ROLLBACK;