diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 5f226f55b41..69424f93503 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -1037,8 +1037,7 @@ "key-failover.test.ts": "adapters", "key-login-live-update.test.ts": "oauth", "key-login-preserves-model-costs.test.ts": "oauth", - "jev-decision.test.ts": "routing", - "jev-decision-destination.test.ts": "routing", + "jev-decision.test.ts": "routing", "jev-decision-destination.test.ts": "routing", "jev-service-exchange.test.ts": "routing", "jev-decision-provider-combo.test.ts": "routing", "jev-typesafe-golden.test.ts": "routing", "jev-model-backend.test.ts": "routing", diff --git a/src/combos/jev-service-exchange.ts b/src/combos/jev-service-exchange.ts new file mode 100644 index 00000000000..835141f1397 --- /dev/null +++ b/src/combos/jev-service-exchange.ts @@ -0,0 +1,281 @@ +import { readBoundedResponseBytes } from "../lib/bounded-body"; +import { + providerOutboundPost, + providerRedirectError, +} from "../lib/provider-outbound"; +import { + isKeychainReference, + keychainReferenceBelongsToProvider, + resolveProviderApiKey, +} from "../providers/api-key-resolve"; +import { providerMatchesRegistryTransport } from "../providers/registry"; +import type { OcxConfig, OcxProviderConfig } from "../types"; +import type { JevDecision, ResolveJevDecisionOptions } from "./jev"; +import { + isSystemOneEndpoint, + jevDecisionEndpointUrl, + JEV_DECISION_TIMEOUT_DEFAULT_MS, + JEV_DECISION_TIMEOUT_MAX_MS, + JEV_DECISION_TIMEOUT_MIN_MS, +} from "./jev-decision-contract"; + +export const JEV_PROVIDER_ID = "jev"; +export const JEV_API_URL = "https://api.typesafe.ai/v1/systemone"; +export const JEV_MODEL = "jev-latest"; + +/** Environment names that hold TypeSafe credentials; a self-hosted row may never reference them. */ +const TYPESAFE_ENV_KEYS = new Set(["TYPESAFE_API_KEY", "JEV_API_KEY"]); +export const JEV_MAX_REQUEST_BYTES = 65_536; +export const JEV_MAX_RESPONSE_BYTES = 65_536; +const JEV_OUTBOUND_DEPENDENCIES = { + isCanonicalUrl: (name: string, url: string) => name === JEV_PROVIDER_ID && url === JEV_API_URL, + // Self-hosted decision models (Ollama tev1) listen on plain HTTP loopback. The outbound wrapper + // still requires the row's own allowPrivateNetwork and a literal local address for that. + allowLocalCleartextPost: true, +}; + +/** Every non-applied outcome of one exchange: a local refusal, a transport failure or an invalid answer. */ +export type JevDecisionFailureGate = Exclude; + +/** + * Transport options shared with the route resolver. `signal` is the caller's cancellation and always + * rejects with its own reason; `timeoutMs` is the separate decision deadline that yields a `timeout` gate. + */ +export type JevServiceExchangeOptions = Pick; + +/** Request shape only: question builders never receive the destination or credential. */ +export interface JevDecisionEndpointShape { + model: string; + descriptiveCriteria: boolean; +} + +/** The decision deadline: the configured value when in bounds, otherwise the four-second default. */ +export function jevDecisionTimeoutMs(value: number | undefined): number { + return value !== undefined + && Number.isInteger(value) + && value >= JEV_DECISION_TIMEOUT_MIN_MS + && value <= JEV_DECISION_TIMEOUT_MAX_MS + ? value + : JEV_DECISION_TIMEOUT_DEFAULT_MS; +} + +/** Retain a registry-matching JEV row's transport policy; ignore a retargeted row for hosted requests. */ +function canonicalJevProvider(config: OcxConfig): OcxProviderConfig { + const configured = config.providers[JEV_PROVIDER_ID]; + if (configured && providerMatchesRegistryTransport(JEV_PROVIDER_ID, configured)) return configured; + return { + adapter: "jev-decision", + baseUrl: JEV_API_URL, + authMode: "key", + liveModels: false, + }; +} + +interface JevDecisionEndpoint { + name: string; + provider: OcxProviderConfig; + url: string; + model: string; + apiKey: string | undefined; + /** Self-hosted System One services accept only string option descriptions. */ + descriptiveCriteria: boolean; +} + +/** Recognize the $NAME and ${NAME} credential references before applying self-hosted ownership rules. */ +function envReferenceName(value: string): string | undefined { + const braced = /^\$\{(\w+)\}$/.exec(value); + if (braced) return braced[1]; + return value.startsWith("$") ? value.slice(1) : undefined; +} + +/** + * A self-hosted row may carry only its own secret: never a reference to the TypeSafe environment + * keys, and never a keychain entry that belongs to another provider. `null` means refused. + */ +function selfHostedApiKey(name: string, apiKey: string | undefined): string | undefined | null { + if (!apiKey) return undefined; + const envName = envReferenceName(apiKey); + if (envName !== undefined && TYPESAFE_ENV_KEYS.has(envName)) return null; + if (isKeychainReference(apiKey) && !keychainReferenceBelongsToProvider(apiKey, name)) return null; + return resolveProviderApiKey(apiKey)?.trim() || undefined; +} + +/** + * Resolve where one decision request goes and which credential it may carry. + * + * The `jev` id stays pinned to the canonical TypeSafe URL and `jev-latest`: its row key is used + * only while the row still matches the registry transport, and the environment fallbacks exist + * only for that URL. A retargeted `jev` row therefore keeps today's behavior instead of becoming a + * custom destination. Any other id must be an enabled `jev-decision` row whose baseUrl is a + * full HTTPS decision endpoint (or a local HTTP `/systemone` endpoint) and which names its own + * model; only its own key may accompany it, so no TypeSafe credential can reach a self-hosted + * service. The URL is `jevDecisionEndpointUrl`, shared with the management surfaces: an HTTPS + * path is sent exactly as configured and only `/systemone` normalizes trailing slashes. + * `undefined` means no usable decision service (reported through the existing `missing_key` + * gate); `null` means the request's destination scope refused it before any credential access. + */ +function jevDecisionEndpoint( + config: OcxConfig, + decisionProvider: string, + isDestinationAllowed?: ResolveJevDecisionOptions["isDestinationAllowed"], +): JevDecisionEndpoint | undefined | null { + const configured = Object.hasOwn(config.providers, decisionProvider) + ? config.providers[decisionProvider] + : undefined; + if (configured?.disabled === true) return undefined; + if (decisionProvider === JEV_PROVIDER_ID) { + if (isDestinationAllowed?.(JEV_PROVIDER_ID, JEV_MODEL) === false) return null; + const configuredOwnsJev = configured + && providerMatchesRegistryTransport(JEV_PROVIDER_ID, configured); + const apiKey = ( + configuredOwnsJev ? resolveProviderApiKey(configured.apiKey)?.trim() : undefined + ) || process.env.TYPESAFE_API_KEY?.trim() + || process.env.JEV_API_KEY?.trim(); + if (!apiKey) return undefined; + return { + name: JEV_PROVIDER_ID, + provider: canonicalJevProvider(config), + url: JEV_API_URL, + model: JEV_MODEL, + apiKey, + descriptiveCriteria: false, + }; + } + if (configured?.adapter !== "jev-decision" || typeof configured.baseUrl !== "string") return undefined; + const url = jevDecisionEndpointUrl(configured.baseUrl); + if (!url || !isSystemOneEndpoint(url)) return undefined; + // `jev-latest` is TypeSafe's model name; a self-hosted host must name its own. + const model = configured.defaultModel?.trim() || configured.models?.[0]?.trim(); + if (!model) return undefined; + if (isDestinationAllowed?.(decisionProvider, model) === false) return null; + const apiKey = selfHostedApiKey(decisionProvider, configured.apiKey); + if (apiKey === null) return undefined; + return { + name: decisionProvider, + provider: configured, + url, + model, + apiKey, + descriptiveCriteria: true, + }; +} + +/** Run the question builder: a refusal keeps its gate; a throw or an over-cap body is `invalid`. */ +function prepareRequestBody( + prepare: (endpoint: JevDecisionEndpointShape) => { body: string } | JevDecisionFailureGate, + shape: JevDecisionEndpointShape, +): { body: string } | { gate: JevDecisionFailureGate } { + try { + const prepared = prepare(shape); + if (typeof prepared === "string") return { gate: prepared }; + const body = prepared.body; + if (new TextEncoder().encode(body).byteLength > JEV_MAX_REQUEST_BYTES) return { gate: "invalid" }; + return { body }; + } catch { + return { gate: "invalid" }; + } +} + +/** + * One bounded System One round-trip, independent of the decision question and its choice policy. + * Destination authorization precedes credential access and `prepare`; request bytes, deadline, + * redirects and UTF-8 JSON are bounded here. `parse` validates the question's answer inside the + * cancellation boundary. Local preparation/parser failures are `invalid`. + * + * This function owns caller cancellation: an aborted `options.signal` rejects with its reason by + * identity, never as a gate. It is checked before endpoint, credential or `prepare` work, after + * endpoint resolution and preparation (including failures), after POST, redirect inspection, HTTP + * error-body cleanup and response reads, and when `parse` returns or throws. Only the caller's signal counts; expiry of + * the separate decision deadline remains a `timeout` gate. + */ +export async function exchangeJevDecision( + options: JevServiceExchangeOptions, + prepare: (endpoint: JevDecisionEndpointShape) => { body: string } | JevDecisionFailureGate, + parse: (payload: unknown) => T, +): Promise<{ value: T } | { gate: JevDecisionFailureGate }> { + if (options.signal?.aborted) throw options.signal.reason; + + let endpoint: JevDecisionEndpoint | undefined | null; + try { + endpoint = jevDecisionEndpoint(options.config, options.decisionProvider ?? JEV_PROVIDER_ID, options.isDestinationAllowed); + } catch (error) { + // Authorization and credential callbacks may cancel the caller and then throw; cancellation wins. + if (options.signal?.aborted) throw options.signal.reason; + throw error; + } + if (options.signal?.aborted) throw options.signal.reason; + if (endpoint === null) return { gate: "invalid" }; + if (!endpoint) return { gate: "missing_key" }; + + const request = prepareRequestBody(prepare, { model: endpoint.model, descriptiveCriteria: endpoint.descriptiveCriteria }); + if (options.signal?.aborted) throw options.signal.reason; + if ("gate" in request) return request; + const requestBody = request.body; + + const timeoutMs = jevDecisionTimeoutMs(options.timeoutMs); + const timeoutSignal = AbortSignal.timeout(timeoutMs); + const signal = options.signal + ? AbortSignal.any([options.signal, timeoutSignal]) + : timeoutSignal; + const post = options.post ?? providerOutboundPost; + + try { + const response = await post( + endpoint.name, + endpoint.provider, + endpoint.url, + { + headers: { + ...(endpoint.apiKey ? { Authorization: `Bearer ${endpoint.apiKey}` } : {}), + "Content-Type": "application/json", + }, + body: requestBody, + signal, + }, + JEV_OUTBOUND_DEPENDENCIES, + ); + if (options.signal?.aborted) throw options.signal.reason; + + const redirectError = await providerRedirectError(response, endpoint.url); + if (options.signal?.aborted) throw options.signal.reason; + if (redirectError) return { gate: "redirect" }; + if (!response.ok) { + try { void response.body?.cancel().catch(() => undefined); } catch { /* best effort */ } + if (options.signal?.aborted) throw options.signal.reason; + return { gate: "http" }; + } + + const bounded = await readBoundedResponseBytes(response, { + maxBytes: JEV_MAX_RESPONSE_BYTES, + signal, + }); + if (options.signal?.aborted) throw options.signal.reason; + if (bounded.oversized) return { gate: "malformed" }; + + let payload: unknown; + try { + const text = new TextDecoder("utf-8", { fatal: true }).decode(bounded.bytes); + payload = JSON.parse(text); + } catch { + return { gate: "malformed" }; + } + + let parsed: T; + try { + parsed = parse(payload); + } catch { + if (options.signal?.aborted) throw options.signal.reason; + return { gate: "invalid" }; + } + if (options.signal?.aborted) throw options.signal.reason; + return { value: parsed }; + } catch (error) { + if (options.signal?.aborted) throw options.signal.reason; + if (timeoutSignal.aborted + || (error instanceof DOMException && error.name === "TimeoutError")) { + return { gate: "timeout" }; + } + return { gate: "network" }; + } +} diff --git a/src/combos/jev.ts b/src/combos/jev.ts index 9b24861302b..ddc37b517d6 100644 --- a/src/combos/jev.ts +++ b/src/combos/jev.ts @@ -1,42 +1,26 @@ -import { readBoundedResponseBytes } from "../lib/bounded-body"; +import type { providerOutboundPost } from "../lib/provider-outbound"; +import type { OcxComboDefaultEffort, OcxConfig } from "../types"; import { - providerOutboundPost, - providerRedirectError, -} from "../lib/provider-outbound"; -import { - isKeychainReference, - keychainReferenceBelongsToProvider, - resolveProviderApiKey, -} from "../providers/api-key-resolve"; -import { providerMatchesRegistryTransport } from "../providers/registry"; -import type { OcxComboDefaultEffort, OcxConfig, OcxProviderConfig } from "../types"; -import { jevDecisionEndpointUrl } from "./jev-decision-contract"; -import { - isSystemOneEndpoint, - JEV_DECISION_TIMEOUT_DEFAULT_MS, - JEV_DECISION_TIMEOUT_MAX_MS, - JEV_DECISION_TIMEOUT_MIN_MS, JEV_MAX_CANDIDATE_FIELD_CHARS, } from "./types"; -export const JEV_PROVIDER_ID = "jev"; -export const JEV_API_URL = "https://api.typesafe.ai/v1/systemone"; -export const JEV_MODEL = "jev-latest"; +import { + exchangeJevDecision, + JEV_PROVIDER_ID, +} from "./jev-service-exchange"; +export { + JEV_API_URL, + JEV_MODEL, + JEV_PROVIDER_ID, + JEV_MAX_REQUEST_BYTES, + JEV_MAX_RESPONSE_BYTES, + jevDecisionTimeoutMs, +} from "./jev-service-exchange"; export const JEV_MAX_CANDIDATES = 64; /** Ollama's System One choice questions accept 2..26 options; other services share the floor. */ const SELF_HOSTED_MIN_OPTIONS = 2; const SELF_HOSTED_MAX_OPTIONS = 26; -/** Environment names that hold TypeSafe credentials; a self-hosted row may never reference them. */ -const TYPESAFE_ENV_KEYS = new Set(["TYPESAFE_API_KEY", "JEV_API_KEY"]); -export const JEV_MAX_REQUEST_BYTES = 65_536; -export const JEV_MAX_RESPONSE_BYTES = 65_536; -const JEV_OUTBOUND_DEPENDENCIES = { - isCanonicalUrl: (name: string, url: string) => name === JEV_PROVIDER_ID && url === JEV_API_URL, - // Self-hosted decision models (Ollama tev1) listen on plain HTTP loopback. The outbound wrapper - // still requires the row's own allowPrivateNetwork and a literal local address for that. - allowLocalCleartextPost: true, -}; const TASK_CHARS = 500; const TASK_HEAD_CHARS = 320; @@ -384,6 +368,7 @@ function hasImageContent(item: Record): boolean { return item.content.some(part => isRecord(part) && (part.type === "input_image" || part.type === "image_url")); } +/** Extract bounded task/assistant/tool evidence, omit protected envelopes, and attach configured target notes. */ export function buildJevState(body: unknown, candidates: readonly JevCandidate[] = []): Record { const input = isRecord(body) ? body.input : undefined; let task = ""; @@ -445,6 +430,7 @@ export function buildJevState(body: unknown, candidates: readonly JevCandidate[] }; } +/** Require task text, an image signal, or a tool-output tail; assistant context and operator notes alone do not suffice. */ export function hasJevDecisionState(state: Record): boolean { if (typeof state.task === "string" && state.task.trim()) return true; if (isRecord(state.signals) && state.signals.has_image === true) return true; @@ -476,6 +462,7 @@ function candidateOptions(candidates: readonly JevCandidate[]): Map JEV_MAX_CANDIDATES) return false; return candidates.every(candidate => [candidate.key, candidate.provider, candidate.model] @@ -550,6 +537,7 @@ export function buildJevRouteQuestion( }; } +/** Project only allowlisted, nonnegative safe-integer usage counters; omit missing or wholly invalid usage. */ export function jevUsage(payload: Record): Record | undefined { if (!isRecord(payload.usage)) return undefined; const usage: Record = {}; @@ -560,6 +548,7 @@ export function jevUsage(payload: Record): Record 0 ? usage : undefined; } +/** Validate an allowlisted route choice and any complete probability distribution; throw on invalid answers. */ export function parseJevDecision( payload: unknown, candidates: readonly JevCandidate[], @@ -612,6 +601,7 @@ export function parseJevDecision( }; } +/** Preserve the caller's eligible target/effort fallback while recording the failed gate, backend, and latency. */ export function fallbackDecision( fallback: ResolveJevDecisionOptions["fallback"], gate: Exclude, @@ -621,113 +611,6 @@ export function fallbackDecision( return { backend, ...fallback, gate, latencyMs }; } -/** The decision deadline: the configured value when in bounds, otherwise the four-second default. */ -export function jevDecisionTimeoutMs(value: number | undefined): number { - return value !== undefined - && Number.isInteger(value) - && value >= JEV_DECISION_TIMEOUT_MIN_MS - && value <= JEV_DECISION_TIMEOUT_MAX_MS - ? value - : JEV_DECISION_TIMEOUT_DEFAULT_MS; -} - -function canonicalJevProvider(config: OcxConfig): OcxProviderConfig { - const configured = config.providers[JEV_PROVIDER_ID]; - if (configured && providerMatchesRegistryTransport(JEV_PROVIDER_ID, configured)) return configured; - return { - adapter: "jev-decision", - baseUrl: JEV_API_URL, - authMode: "key", - liveModels: false, - }; -} - -interface JevDecisionEndpoint { - name: string; - provider: OcxProviderConfig; - url: string; - model: string; - apiKey: string | undefined; - /** Self-hosted System One services accept only string option descriptions. */ - descriptiveCriteria: boolean; -} - -function envReferenceName(value: string): string | undefined { - const braced = /^\$\{(\w+)\}$/.exec(value); - if (braced) return braced[1]; - return value.startsWith("$") ? value.slice(1) : undefined; -} - -/** - * A self-hosted row may carry only its own secret: never a reference to the TypeSafe environment - * keys, and never a keychain entry that belongs to another provider. `null` means refused. - */ -function selfHostedApiKey(name: string, apiKey: string | undefined): string | undefined | null { - if (!apiKey) return undefined; - const envName = envReferenceName(apiKey); - if (envName !== undefined && TYPESAFE_ENV_KEYS.has(envName)) return null; - if (isKeychainReference(apiKey) && !keychainReferenceBelongsToProvider(apiKey, name)) return null; - return resolveProviderApiKey(apiKey)?.trim() || undefined; -} - -/** - * Resolve where one decision request goes and which credential it may carry. - * - * The `jev` id stays pinned to the canonical TypeSafe URL and `jev-latest`: its row key is used - * only while the row still matches the registry transport, and the environment fallbacks exist - * only for that URL. A retargeted `jev` row therefore keeps today's behavior instead of becoming a - * custom destination. Any other id must be an enabled `jev-decision` row whose baseUrl is a - * full HTTPS decision endpoint (or a local HTTP `/systemone` endpoint) and names its own model; only its own key may accompany it, so no - * TypeSafe credential can reach a self-hosted service. `undefined` means no usable decision - * service (reported through the existing `missing_key` gate); `null` means the request's - * destination scope refused it before any credential access. - */ -function jevDecisionEndpoint( - config: OcxConfig, - decisionProvider: string, - isDestinationAllowed?: ResolveJevDecisionOptions["isDestinationAllowed"], -): JevDecisionEndpoint | undefined | null { - const configured = Object.hasOwn(config.providers, decisionProvider) - ? config.providers[decisionProvider] - : undefined; - if (configured?.disabled === true) return undefined; - if (decisionProvider === JEV_PROVIDER_ID) { - if (isDestinationAllowed?.(JEV_PROVIDER_ID, JEV_MODEL) === false) return null; - const configuredOwnsJev = configured - && providerMatchesRegistryTransport(JEV_PROVIDER_ID, configured); - const apiKey = ( - configuredOwnsJev ? resolveProviderApiKey(configured.apiKey)?.trim() : undefined - ) || process.env.TYPESAFE_API_KEY?.trim() - || process.env.JEV_API_KEY?.trim(); - if (!apiKey) return undefined; - return { - name: JEV_PROVIDER_ID, - provider: canonicalJevProvider(config), - url: JEV_API_URL, - model: JEV_MODEL, - apiKey, - descriptiveCriteria: false, - }; - } - if (configured?.adapter !== "jev-decision" || typeof configured.baseUrl !== "string") return undefined; - const url = jevDecisionEndpointUrl(configured.baseUrl); - if (!url || !isSystemOneEndpoint(url)) return undefined; - // `jev-latest` is TypeSafe's model name; a self-hosted host must name its own. - const model = configured.defaultModel?.trim() || configured.models?.[0]?.trim(); - if (!model) return undefined; - if (isDestinationAllowed?.(decisionProvider, model) === false) return null; - const apiKey = selfHostedApiKey(decisionProvider, configured.apiKey); - if (apiKey === null) return undefined; - return { - name: decisionProvider, - provider: configured, - url, - model, - apiKey, - descriptiveCriteria: true, - }; -} - /** * Ask the configured JEV decision service for one allowlisted target/effort decision. * @@ -745,98 +628,29 @@ export async function resolveJevDecision(options: ResolveJevDecisionOptions): Pr if (options.candidates.length === 0) return failed("no_choices"); if (!candidatesFitRequestBounds(options.candidates)) return failed("invalid"); - const endpoint = jevDecisionEndpoint(options.config, options.decisionProvider ?? JEV_PROVIDER_ID, options.isDestinationAllowed); - if (endpoint === null) return failed("invalid"); - if (!endpoint) return failed("missing_key"); - - let requestBody: string; - try { + const exchanged = await exchangeJevDecision(options, (endpoint) => { if (endpoint.descriptiveCriteria) { // Self-hosted choice questions accept 2..26 options; decide locally instead of spending a // round-trip on a request the service will refuse. const optionCount = candidateOptions(options.candidates).size; - if (optionCount < SELF_HOSTED_MIN_OPTIONS) return failed("no_choices"); - if (optionCount > SELF_HOSTED_MAX_OPTIONS) return failed("invalid"); + if (optionCount < SELF_HOSTED_MIN_OPTIONS) return "no_choices"; + if (optionCount > SELF_HOSTED_MAX_OPTIONS) return "invalid"; } const state = buildJevState(options.body, options.candidates); - if (!hasJevDecisionState(state)) return failed("no_state"); - requestBody = JSON.stringify({ + if (!hasJevDecisionState(state)) return "no_state"; + return { body: JSON.stringify({ model: endpoint.model, state, questions: buildJevRouteQuestion(options.candidates, { descriptiveCriteria: endpoint.descriptiveCriteria }), - }); - if (new TextEncoder().encode(requestBody).byteLength > JEV_MAX_REQUEST_BYTES) return failed("invalid"); - } catch { - return failed("invalid"); - } - - const timeoutMs = jevDecisionTimeoutMs(options.timeoutMs); - const timeoutSignal = AbortSignal.timeout(timeoutMs); - const signal = options.signal - ? AbortSignal.any([options.signal, timeoutSignal]) - : timeoutSignal; - const post = options.post ?? providerOutboundPost; - - try { - const response = await post( - endpoint.name, - endpoint.provider, - endpoint.url, - { - headers: { - ...(endpoint.apiKey ? { Authorization: `Bearer ${endpoint.apiKey}` } : {}), - "Content-Type": "application/json", - }, - body: requestBody, - signal, - }, - JEV_OUTBOUND_DEPENDENCIES, - ); - if (options.signal?.aborted) throw options.signal.reason; - - const redirectError = await providerRedirectError(response, endpoint.url); - if (redirectError) return failed("redirect"); - if (!response.ok) { - try { void response.body?.cancel().catch(() => undefined); } catch { /* best effort */ } - return failed("http"); - } - - const bounded = await readBoundedResponseBytes(response, { - maxBytes: JEV_MAX_RESPONSE_BYTES, - signal, - }); - if (options.signal?.aborted) throw options.signal.reason; - if (bounded.oversized) return failed("malformed"); - - let payload: unknown; - try { - const text = new TextDecoder("utf-8", { fatal: true }).decode(bounded.bytes); - payload = JSON.parse(text); - } catch { - return failed("malformed"); - } - - let parsed: ReturnType; - try { - parsed = parseJevDecision(payload, options.candidates); - } catch { - return failed("invalid"); - } - if (options.signal?.aborted) throw options.signal.reason; - return { - backend, - ...parsed, - gate: "apply", - latencyMs: Math.max(0, now() - startedAt), - }; - } catch (error) { - if (options.signal?.aborted) throw options.signal.reason; - if (timeoutSignal.aborted - || (error instanceof DOMException && error.name === "TimeoutError")) { - return failed("timeout"); - } - return failed("network"); - } + }) }; + }, (payload) => parseJevDecision(payload, options.candidates)); + if ("gate" in exchanged) return failed(exchanged.gate); + return { + backend, + ...exchanged.value, + gate: "apply", + latencyMs: Math.max(0, now() - startedAt), + }; } export type JevDecisionProbeResult = diff --git a/structure/providers/jev-decision.md b/structure/providers/jev-decision.md index 041ade80770..802f0d226e4 100644 --- a/structure/providers/jev-decision.md +++ b/structure/providers/jev-decision.md @@ -30,8 +30,9 @@ rows reuse `missing_key`, and `decisionTimeoutMs` (1000..120000) replaces the 4 Decision backends. `src/combos/jev-dispatch.ts` derives the backend from the Combo and never stores it: `decisionModel` set means `model`, a `decisionProvider` other than `jev` means `systemone`, and -neither means `typesafe`; setting both is a config error. The System One path is `src/combos/jev.ts` -unchanged, so the TypeSafe request bytes stay pinned by `tests/fixtures/jev-typesafe-request-golden.json`. +neither means `typesafe`; setting both is a config error. `src/combos/jev.ts` builds and validates +the route question through `src/combos/jev-service-exchange.ts`, the shared question-agnostic HTTP +exchange. TypeSafe request bytes stay pinned by `tests/fixtures/jev-typesafe-request-golden.json`. `src/combos/jev-model-backend.ts` asks an ordinary opencodex route for `{"choice":""}` over the same bounded state and option map, under the same deadline, bounds, and fail-open gates; `src/combos/jev-decision-contract.ts` holds the constants the GUI shares. The server glue @@ -58,10 +59,23 @@ the JEV credential are excluded. All three text samples omit recognized Codex pr and Claude Code `` blocks before clipping, including nested and unclosed blocks. Only an envelope-only `codex_internal_context` goal outside a reminder may supply a fallback task; reminder-only text supplies no task. Reminder-free inputs keep their existing sampling behavior. -It owns the joint target/effort choice map, strict response -validation, canonical `jev-latest` destination, default four-second deadline, no-redirect policy, bounded response, -and caller-cancellation propagation. Missing credentials or safe state, transport failures, and invalid -answers fail open to the first eligible target; no response can escape the configured choice map. +It owns the joint target/effort choice map and strict answer +validation, including the complete probability distribution. The service exchange owns the canonical +`jev-latest` destination, the self-hosted URL (shared `jevDecisionEndpointUrl`: an HTTPS path as +configured, `/systemone` trailing slashes normalized), credentials, 64 KiB serialized request and response caps, default +four-second deadline, no-redirect policy, and caller-cancellation ownership. It rejects an already +aborted caller by reason identity before endpoint, credential or preparation work. It checks again +after endpoint resolution and preparation (including failures), POST, redirect inspection, HTTP +error-body cleanup and response reads, and when parsing returns or throws; caller cancellation takes +precedence over the local outcome at those checkpoints. HTTP error-body cancellation is best effort +and never awaited. The separate decision deadline remains a `timeout` gate. Its +request builder receives only `model` and `descriptiveCriteria` after authorization/credential +resolution; its answer parser runs inside the cancellation boundary. Without caller cancellation, +question-specific local refusals and invalid answers retain their existing gates. `tests/routing/jev-service-exchange.test.ts` exercises this +seam independently of route questions; `tests/routing/jev-typesafe-golden.test.ts` keeps the +unchanged route bytes authoritative. Missing credentials or safe state, transport failures, and +invalid answers fail open to the first eligible target; no response can escape the configured choice +map. The direct TypeSafe and System One decision destinations are checked against the parent API key's resolved provider/model scope before reading decision credentials or extracting state. A denied optional decision uses the existing fail-open inference target without sending a decision request; diff --git a/structure/transports/inventory.md b/structure/transports/inventory.md index 762992198ee..733fade3166 100644 --- a/structure/transports/inventory.md +++ b/structure/transports/inventory.md @@ -191,12 +191,12 @@ resolved-address policy error still rejects. Proxy mode logs once that the proxy cannot be pinned. Private destinations additionally require allowPrivateNetwork plus NO_PROXY. `providerOutboundPost` refuses every non-HTTPS URL before any executor or DNS work. The one exception is the caller opt-in `allowLocalCleartextPost`, used only by the self-hosted JEV decision client in -`src/combos/jev.ts`: it admits `http:` solely when the row sets `allowPrivateNetwork: true` itself -(a registry default does not count) and the host is exactly `localhost` or an address literal in the -narrow `localCleartextAddressAllowed` set (127/8, ::1, ::ffff:127.0.0.0/104, 10/8, 172.16/12, -192.168/16, fc00::/7). Every resolved answer must stay in that set, any applicable proxy (including -the DNS-failure proxy degradation) is refused, and an injected executor must receive an address -literal. Every other destination keeps the HTTPS-only gate. +`src/combos/jev-service-exchange.ts`: it admits `http:` solely when the row sets +`allowPrivateNetwork: true` itself (a registry default does not count) and the host is exactly +`localhost` or an address literal in the narrow `localCleartextAddressAllowed` set (127/8, ::1, +::ffff:127.0.0.0/104, 10/8, 172.16/12, 192.168/16, fc00::/7). Every resolved answer must stay in +that set, any applicable proxy (including the DNS-failure proxy degradation) is refused, and an +injected executor must receive an address literal. Every other destination keeps the HTTPS-only gate. Every request through this wrapper is proxy-originated, so it fills a default `User-Agent: opencodex` when the request headers name no User-Agent of their own; registry diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index e7953e2947a..4d8b46431df 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -636,6 +636,7 @@ "issue-914-transport-attribution.test.ts": "codex-integration", "key-attribution.test.ts": "usage", "key-failover.test.ts": "adapters", "key-login-live-update.test.ts": "oauth", "key-login-preserves-model-costs.test.ts": "oauth", "jev-decision.test.ts": "routing", + "jev-service-exchange.test.ts": "routing", "jev-decision-destination.test.ts": "routing", "jev-decision-provider-combo.test.ts": "routing", "jev-typesafe-golden.test.ts": "routing", "jev-model-backend.test.ts": "routing", "jev-provider.test.ts": "providers", "keyring-smoke.test.ts": "ci-workflows", diff --git a/tests/routing/jev-service-exchange.test.ts b/tests/routing/jev-service-exchange.test.ts new file mode 100644 index 00000000000..664eb54893f --- /dev/null +++ b/tests/routing/jev-service-exchange.test.ts @@ -0,0 +1,527 @@ +import { describe, expect, test } from "bun:test"; +import { parseJevDecision, resolveJevDecision, type JevCandidate } from "../../src/combos/jev"; +import { + exchangeJevDecision, + JEV_API_URL, + type JevDecisionEndpointShape, + type JevServiceExchangeOptions, +} from "../../src/combos/jev-service-exchange"; +import type { OcxConfig, OcxProviderConfig } from "../../src/types"; + +type Post = NonNullable; +const row: OcxProviderConfig = { + adapter: "jev-decision", baseUrl: "https://decider.example/v1/systemone/", + defaultModel: "tev1:4b", apiKey: "fixture-own-key", +}; +const config = (name = "decider", provider = row): OcxConfig => ({ + port: 0, defaultProvider: "a", providers: { [name]: provider }, +}); +const options = (extra: Partial = {}): JevServiceExchangeOptions => ({ + config: config(), decisionProvider: "decider", ...extra, +}); +const prepare = (endpoint: JevDecisionEndpointShape) => ({ body: JSON.stringify({ + model: endpoint.model, state: { task: "A synthetic question." }, + questions: { test: { type: "choice", criteria: { yes: "Yes", no: "No" } } }, +}) }); +const parse = (payload: unknown) => payload; +const okPost: Post = async () => Response.json({ answers: { test: { choice: "yes" } } }); + +// Exercise the seam directly so future question builders cannot bypass the transport contract. +describe("bounded JEV service exchange", () => { + for (const name of ["jev", "decider"]) { + test(`${name}: destination denial precedes credential access and request-state extraction`, async () => { + const seen: string[] = []; + const provider: OcxProviderConfig = { + ...row, baseUrl: name === "jev" ? JEV_API_URL : row.baseUrl, + get apiKey(): string { throw new Error("credential must not be read"); }, + }; + let prepared = false; + let sends = 0; + const opts = options({ config: config(name, provider), decisionProvider: name, + isDestinationAllowed(providerName, model) { seen.push(`${providerName}/${model}`); return false; }, + post: async () => { sends++; return okPost("", row, "", { body: "" }); }, + }); + expect(await exchangeJevDecision(opts, () => { prepared = true; throw new Error("state must not be read"); }, parse)) + .toEqual({ gate: "invalid" }); + expect(prepared).toBe(false); + expect(seen).toEqual([name === "jev" ? "jev/jev-latest" : "decider/tev1:4b"]); + expect(sends).toBe(0); + + const body = { get input(): string { throw new Error("state must not be read"); } }; + expect(await resolveJevDecision({ ...opts, body, candidates, fallback })) + .toMatchObject({ ...fallback, gate: "invalid" }); + }); + } + + test("builders see only model/criteria shape; another question uses the same outbound guard", async () => { + const shapes: JevDecisionEndpointShape[] = []; + const result = await exchangeJevDecision(options({ post: async (name, provider, url, init, deps) => { + expect(name).toBe("decider"); + expect(provider).toBe(row); + expect(url).toBe("https://decider.example/v1/systemone"); + expect(new Headers(init.headers).get("authorization")).toBe("Bearer fixture-own-key"); + expect(new Headers(init.headers).get("content-type")).toBe("application/json"); + expect(init.body).toBe(prepare({ model: "tev1:4b", descriptiveCriteria: true }).body); + expect(deps?.isCanonicalUrl?.(name, url)).toBe(false); + expect(deps?.isCanonicalUrl?.("jev", JEV_API_URL)).toBe(true); + expect(deps?.allowLocalCleartextPost).toBe(true); + return Response.json({ answers: { test: { choice: "yes" } } }); + } }), endpoint => { shapes.push(endpoint); return prepare(endpoint); }, parse); + expect(shapes).toEqual([{ model: "tev1:4b", descriptiveCriteria: true }]); + expect(result).toEqual({ value: { answers: { test: { choice: "yes" } } } }); + }); + + test("keyless self-hosted requests never borrow canonical environment keys", async () => { + const previous = process.env.TYPESAFE_API_KEY; + process.env.TYPESAFE_API_KEY = "fixture-typesafe-key"; + try { + expect(await exchangeJevDecision(options({ config: config("decider", { ...row, apiKey: undefined }), + post: async (_name, _provider, _url, init) => { + expect(new Headers(init.headers).has("authorization")).toBe(false); + expect(String(init.body)).not.toContain("fixture-typesafe-key"); + return Response.json({}); + }, + }), prepare, parse)).toEqual({ value: {} }); + } finally { + if (previous === undefined) delete process.env.TYPESAFE_API_KEY; + else process.env.TYPESAFE_API_KEY = previous; + } + }); + + for (const apiKey of ["$TYPESAFE_API_KEY", "${TYPESAFE_API_KEY}", "$JEV_API_KEY", "${JEV_API_KEY}", "keychain:jev"]) { + test(`refuses foreign credential reference ${apiKey} before preparation or POST`, async () => { + expect(await exchangeJevDecision(options({ config: config("decider", { ...row, apiKey }), + post: async () => { throw new Error("unexpected send"); }, + }), () => { throw new Error("unexpected preparation"); }, parse)).toEqual({ gate: "missing_key" }); + }); + } + + // The 64 KiB caps are pinned as literals, not through the exported constants, so changing a cap fails here. + test("enforces serialized UTF-8 request byte cap, including its exact boundary", async () => { + let sends = 0; + const opts = options({ post: async () => { sends++; return Response.json({}); } }); + const boundary = "é".repeat(32_768); // 2 bytes each: exactly 65_536 bytes + let bodyReads = 0; + expect(await exchangeJevDecision(opts, () => ({ get body() { bodyReads++; return boundary; } }), parse)).toEqual({ value: {} }); + expect(bodyReads).toBe(1); // Preparation samples the body once, as before the cancellation fix. + expect(await exchangeJevDecision(opts, () => ({ body: `${boundary}x` }), parse)).toEqual({ gate: "invalid" }); + expect(sends).toBe(1); + }); + + test("redirect/HTTP responses are canceled without parsing or retaining upstream detail", async () => { + for (const [status, gate] of [[302, "redirect"], [307, "redirect"], [402, "http"]] as const) { + let canceled = false; + let parsed = false; + const stream = new ReadableStream({ cancel() { canceled = true; } }); + expect(await exchangeJevDecision(options({ post: async () => new Response(stream, { + status, headers: { location: "https://other.example/" }, + }) }), prepare, () => { parsed = true; })).toEqual({ gate }); + expect(canceled).toBe(true); + expect(parsed).toBe(false); + } + }); + + test("bounds response bytes at 64 KiB and rejects malformed JSON and UTF-8", async () => { + const boundary = `"${"x".repeat(65_534)}"`; // exactly 65_536 bytes + expect(new TextEncoder().encode(boundary).byteLength).toBe(65_536); + expect(await exchangeJevDecision(options({ post: async () => new Response(boundary) }), prepare, parse)) + .toEqual({ value: "x".repeat(65_534) }); + let canceled = false; + const stream = new ReadableStream({ + start(controller) { controller.enqueue(new Uint8Array(65_537)); }, + cancel() { canceled = true; }, + }); + expect(await exchangeJevDecision(options({ post: async () => new Response(stream) }), prepare, parse)) + .toEqual({ gate: "malformed" }); + expect(canceled).toBe(true); + for (const body of ["not-json", new Uint8Array([0x22, 0xc3, 0x28, 0x22])]) { + expect(await exchangeJevDecision(options({ post: async () => new Response(body) }), prepare, parse)) + .toEqual({ gate: "malformed" }); + } + }); + + test("local preparation and answer validation retain their existing gates", async () => { + for (const gate of ["no_choices", "no_state", "invalid"] as const) { + expect(await exchangeJevDecision(options(), () => gate, parse)).toEqual({ gate }); + } + expect(await exchangeJevDecision(options(), () => { throw new Error("private preparation detail"); }, parse)) + .toEqual({ gate: "invalid" }); + expect(await exchangeJevDecision(options({ post: okPost }), prepare, () => { throw new Error("private parser detail"); })) + .toEqual({ gate: "invalid" }); + expect(await exchangeJevDecision(options({ post: async () => { throw new TypeError("private network detail"); } }), prepare, parse)) + .toEqual({ gate: "network" }); + }); + + test("route parser still refuses malformed probabilities through the shared exchange", async () => { + for (const probabilities of [null, { "a/m1:low": 1 }, { "a/m1:low": 0.4, "a/m2:low": 0.2 }, + { "a/m1:low": 0.2, "a/m2:low": 0.8 }, { "a/m1:low": -0.1, "a/m2:low": 1.1 }]) { + const payload = { answers: { route: { choice: "a/m1:low", probabilities } } }; + expect(await exchangeJevDecision(options({ post: async () => Response.json(payload) }), prepare, + answer => parseJevDecision(answer, candidates))).toEqual({ gate: "invalid" }); + } + }); + + test("caller cancellation wins by identity when already aborted, after POST, during read and parsing", async () => { + const reason = { caller: "stopped" }; + for (const phase of ["pre", "post", "read", "parse"] as const) { + const caller = new AbortController(); + if (phase === "pre") caller.abort(reason); + let canceled = false; + const post: Post = async () => { + if (phase === "post") caller.abort(reason); + if (phase === "read") return new Response(new ReadableStream({ + pull() { caller.abort(reason); }, cancel() { canceled = true; }, + }, { highWaterMark: 0 })); + return Response.json({}); + }; + await expect(exchangeJevDecision(options({ signal: caller.signal, post }), prepare, payload => { + if (phase === "parse") caller.abort(reason); + return payload; + })).rejects.toBe(reason); + if (phase === "read") expect(canceled).toBe(true); + } + }); + + // Cancellation ownership: whenever the caller's signal is aborted at a checkpoint, the caller's reason + // wins by identity over every local gate. Each case aborts from inside the callback under test and then + // refuses or throws, so only a check made after that callback returns can still preserve the reason. + test("an already-aborted caller rejects by identity before authorization, credential access, preparation or POST", async () => { + const reason = { caller: "stopped before start" }; + for (const variant of ["self-hosted", "canonical", "missing-key", "unusable", "foreign-credential", "prepare-invalid", "prepare-no-state"] as const) { + for (const allowed of [true, false]) { + const counts = { authorizations: 0, credentialReads: 0, preparations: 0, sends: 0 }; + const credential = variant === "missing-key" ? undefined : variant === "foreign-credential" ? "$TYPESAFE_API_KEY" : "fixture-own-key"; + const canonical = variant === "canonical" || variant === "missing-key"; + const provider: OcxProviderConfig = canonical + ? { adapter: "jev-decision", baseUrl: JEV_API_URL, get apiKey() { counts.credentialReads++; return credential; } } + : { ...row, get apiKey() { counts.credentialReads++; return credential; } }; + const name = canonical ? "jev" : "decider"; + const caller = new AbortController(); + caller.abort(reason); + const opts = options({ + config: variant === "unusable" ? { port: 0, defaultProvider: "a", providers: {} } : config(name, provider), + decisionProvider: name, + signal: caller.signal, + isDestinationAllowed() { counts.authorizations++; return allowed; }, + post: async () => { counts.sends++; return Response.json({}); }, + }); + await expect(exchangeJevDecision(opts, () => { + counts.preparations++; + if (variant === "prepare-invalid") return "invalid"; + if (variant === "prepare-no-state") return "no_state"; + return prepare({ model: "m", descriptiveCriteria: true }); + }, parse)).rejects.toBe(reason); + expect(counts).toEqual({ authorizations: 0, credentialReads: 0, preparations: 0, sends: 0 }); + } + } + }); + + test("a caller abort inside authorization or credential resolution beats the local refusal or throw", async () => { + const stages = ["deny", "deny-then-throw", "allow", "foreign-credential", "credential-throw", "credential-unusable"] as const; + const previous = [process.env.TYPESAFE_API_KEY, process.env.JEV_API_KEY]; + delete process.env.TYPESAFE_API_KEY; + delete process.env.JEV_API_KEY; + try { + for (const stage of stages) { + const reason = { caller: `stopped during ${stage}` }; + const caller = new AbortController(); + const counts = { preparations: 0, sends: 0 }; + const credential = (): string | undefined => { + caller.abort(reason); + if (stage === "credential-throw") throw new Error("synthetic credential failure"); + return stage === "foreign-credential" ? "$TYPESAFE_API_KEY" : undefined; + }; + const unusable = stage === "credential-unusable"; + const keyed = stage.startsWith("credential") || stage === "foreign-credential"; + const name = unusable ? "jev" : "decider"; + const provider: OcxProviderConfig = unusable + ? { adapter: "jev-decision", baseUrl: JEV_API_URL, get apiKey() { return credential(); } } + : { ...row, get apiKey() { return keyed ? credential() : "fixture-own-key"; } }; + const opts = options({ + config: config(name, provider), decisionProvider: name, signal: caller.signal, + ...(keyed ? {} : { isDestinationAllowed() { + caller.abort(reason); + if (stage === "deny-then-throw") throw new Error("synthetic authorization failure"); + return stage === "allow"; + } }), + post: async () => { counts.sends++; return Response.json({}); }, + }); + await expect(exchangeJevDecision(opts, () => { counts.preparations++; return prepare({ model: "m", descriptiveCriteria: true }); }, parse)) + .rejects.toBe(reason); + expect(counts).toEqual({ preparations: 0, sends: 0 }); + } + } finally { + for (const [i, key] of ["TYPESAFE_API_KEY", "JEV_API_KEY"].entries()) { + if (previous[i] === undefined) delete process.env[key]; + else process.env[key] = previous[i]; + } + } + }); + + test("an authorization failure without cancellation keeps its own error and a destination denial stays a gate", async () => { + const failure = new Error("synthetic authorization failure"); + const caller = new AbortController(); + await expect(exchangeJevDecision(options({ signal: caller.signal, isDestinationAllowed() { throw failure; } }), prepare, parse)) + .rejects.toBe(failure); + expect(await exchangeJevDecision(options({ signal: caller.signal, isDestinationAllowed: () => false }), prepare, parse)) + .toEqual({ gate: "invalid" }); + }); + + test("a caller abort inside preparation beats every local refusal and prevents the POST", async () => { + for (const phase of ["no_choices", "no_state", "invalid", "oversized", "throw", "valid"] as const) { + const reason = { caller: `stopped during ${phase} preparation` }; + const caller = new AbortController(); + let sends = 0; + await expect(exchangeJevDecision(options({ signal: caller.signal, post: async () => { sends++; return Response.json({}); } }), () => { + caller.abort(reason); + if (phase === "throw") throw new Error("synthetic invalid request"); + if (phase === "oversized") return { body: "x".repeat(65_537) }; + if (phase === "valid") return { body: "{}" }; + return phase; + }, parse)).rejects.toBe(reason); + expect(sends).toBe(0); + } + }); + + test("a caller abort inside the answer parser beats the invalid gate; other parser failures stay invalid", async () => { + for (const outcome of ["throw", "return"] as const) { + const reason = { caller: `stopped in parser then ${outcome}` }; + const caller = new AbortController(); + await expect(exchangeJevDecision(options({ signal: caller.signal, post: okPost }), prepare, payload => { + caller.abort(reason); + if (outcome === "throw") throw new Error("synthetic rejected answer"); + return payload; + })).rejects.toBe(reason); + } + const idle = new AbortController(); + expect(await exchangeJevDecision(options({ signal: idle.signal, post: okPost }), prepare, () => { throw new Error("synthetic rejected answer"); })) + .toEqual({ gate: "invalid" }); + expect(await exchangeJevDecision(options({ signal: idle.signal, post: okPost }), prepare, parse)) + .toEqual({ value: { answers: { test: { choice: "yes" } } } }); + }); + + test("a caller abort while a redirect body is canceled still wins over the redirect gate", async () => { + const reason = { caller: "stopped while canceling a redirect body" }; + const caller = new AbortController(); + const body = new ReadableStream({ cancel() { caller.abort(reason); } }); + await expect(exchangeJevDecision(options({ signal: caller.signal, post: async () => new Response(body, { + status: 302, headers: { location: "https://other.example/" }, + }) }), prepare, parse)).rejects.toBe(reason); + }); + + test("caller abort during HTTP body cleanup rejects by identity in the exchange and actual route", async () => { + for (const resolveRoute of [false, true]) for (const cleanup of ["return", "throw", "pending"] as const) { + const caller = new AbortController(); + const reason = { caller: `stopped during HTTP cleanup: ${cleanup}` }; + let cancels = 0; + let parses = 0; + const post: Post = async () => new Response(new ReadableStream({ + cancel() { + cancels++; + caller.abort(reason); + if (cleanup === "throw") throw new Error("synthetic cleanup failure"); + if (cleanup === "pending") return new Promise(() => {}); + }, + }), { status: 402 }); + const opts = options({ signal: caller.signal, post }); + const pending = resolveRoute + ? resolveJevDecision({ ...opts, body: { input: "A synthetic task." }, candidates, fallback }) + : exchangeJevDecision(opts, prepare, payload => { parses++; return payload; }); + await expect(pending).rejects.toBe(reason); + expect(cancels).toBe(1); + expect(parses).toBe(0); + } + const caller = new AbortController(); + const reason = { caller: "stopped in a synchronously throwing cancel method" }; + const response = new Response("synthetic HTTP error", { status: 402 }); + response.body!.cancel = () => { caller.abort(reason); throw new Error("synthetic synchronous cleanup failure"); }; + await expect(exchangeJevDecision(options({ signal: caller.signal, post: async () => response }), prepare, parse)).rejects.toBe(reason); + }); + + test("HTTP cleanup remains best effort and non-waiting without caller cancellation", async () => { + for (const resolveRoute of [false, true]) for (const cleanup of ["return", "throw", "pending"] as const) { + const caller = new AbortController(); + let cancels = 0; + let parses = 0; + const opts = options({ signal: caller.signal, post: async () => new Response(new ReadableStream({ + cancel() { + cancels++; + if (cleanup === "throw") throw new Error("synthetic cleanup failure"); + if (cleanup === "pending") return new Promise(() => {}); + }, + }), { status: 402 }) }); + const result = resolveRoute + ? await resolveJevDecision({ ...opts, body: { input: "A synthetic task." }, candidates, fallback }) + : await exchangeJevDecision(opts, prepare, payload => { parses++; return payload; }); + expect(result).toMatchObject({ gate: "http" }); + if (resolveRoute) expect(result).toMatchObject(fallback); + expect(caller.signal.aborted).toBe(false); + expect(cancels).toBe(1); + expect(parses).toBe(0); + } + // Even a non-conforming cancel method that throws synchronously is still best effort. + const response = new Response("synthetic HTTP error", { status: 402 }); + response.body!.cancel = () => { throw new Error("synthetic synchronous cleanup failure"); }; + expect(await exchangeJevDecision(options({ post: async () => response }), prepare, parse)).toEqual({ gate: "http" }); + }); + + test("caller abort during oversized-response cleanup beats the malformed gate", async () => { + const caller = new AbortController(); + const reason = { caller: "stopped during oversized-response cleanup" }; + let cancels = 0; + const body = new ReadableStream({ + start(controller) { controller.enqueue(new Uint8Array(65_537)); }, + cancel() { cancels++; caller.abort(reason); }, + }); + await expect(exchangeJevDecision(options({ signal: caller.signal, post: async () => new Response(body) }), prepare, parse)) + .rejects.toBe(reason); + expect(cancels).toBe(1); + }); + + test("caller cancellation stays separate from the decision deadline", async () => { + const descriptor = Object.getOwnPropertyDescriptor(AbortSignal, "timeout")!; + let deadline = new AbortController(); + Object.defineProperty(AbortSignal, "timeout", { configurable: true, value() { return deadline.signal; } }); + try { + const expired = new DOMException("deadline", "TimeoutError"); + const idle = new AbortController(); + expect(await exchangeJevDecision(options({ signal: idle.signal, post: async () => { deadline.abort(expired); throw expired; } }), prepare, parse)) + .toEqual({ gate: "timeout" }); + deadline = new AbortController(); + expect(await exchangeJevDecision(options({ signal: idle.signal, post: okPost }), prepare, () => { + deadline.abort(expired); + throw new Error("synthetic rejected answer"); + })).toEqual({ gate: "invalid" }); + expect(idle.signal.aborted).toBe(false); + deadline = new AbortController(); + const reason = { caller: "stopped as the deadline expired" }; + const caller = new AbortController(); + await expect(exchangeJevDecision(options({ signal: caller.signal, post: async () => { + deadline.abort(expired); + caller.abort(reason); + throw expired; + } }), prepare, parse)).rejects.toBe(reason); + } finally { Object.defineProperty(AbortSignal, "timeout", descriptor); } + }); + + test("caller abort reaches the in-flight POST signal and settles long before the deadline", async () => { + const caller = new AbortController(); + const reason = { caller: "stopped in flight" }; + const release = new AbortController(); // lets a regressed build unwind instead of leaking a pending POST + let posted: AbortSignal | undefined; + let started!: () => void; + const postStarted = new Promise(resolve => { started = resolve; }); + const post: Post = (_name, _provider, _url, init) => { + posted = init.signal ?? undefined; + return new Promise((_resolve, reject) => { + init.signal?.addEventListener("abort", () => reject(init.signal?.reason), { once: true }); + release.signal.addEventListener("abort", () => reject(new Error("test released")), { once: true }); + started(); + }); + }; + const pending = exchangeJevDecision(options({ signal: caller.signal, timeoutMs: 120_000, post }), prepare, parse); + await postStarted; + expect(posted?.aborted).toBe(false); + caller.abort(reason); + let timer: ReturnType | undefined; + const outcome = await Promise.race([ + pending.then(() => "resolved", error => (error === reason ? "reason" : "other")), + new Promise(resolve => { timer = setTimeout(() => resolve("still pending"), 1_000); }), + ]); + clearTimeout(timer); + release.abort(); + await pending.catch(() => undefined); + expect(outcome).toBe("reason"); + expect(posted?.aborted).toBe(true); + expect(posted?.reason).toBe(reason); + }); + + test("deadline normalization and expiry apply to both POST and body read", async () => { + const descriptor = Object.getOwnPropertyDescriptor(AbortSignal, "timeout")!; + const deadlines: number[] = []; + let deadline = new AbortController(); + Object.defineProperty(AbortSignal, "timeout", { configurable: true, value(ms: number) { + deadlines.push(ms); return deadline.signal; + } }); + try { + for (const timeoutMs of [undefined, 999, 1_000, 120_000, 120_001, 1_500.5]) { + expect(await exchangeJevDecision(options({ timeoutMs, post: okPost }), prepare, parse)).toHaveProperty("value"); + } + expect(deadlines).toEqual([4_000, 4_000, 1_000, 120_000, 4_000, 4_000]); + const reason = new DOMException("deadline", "TimeoutError"); + expect(await exchangeJevDecision(options({ post: async () => { deadline.abort(reason); throw reason; } }), prepare, parse)) + .toEqual({ gate: "timeout" }); + deadline = new AbortController(); + let canceled = false; + expect(await exchangeJevDecision(options({ post: async () => new Response(new ReadableStream({ + pull() { deadline.abort(reason); }, cancel() { canceled = true; }, + }, { highWaterMark: 0 })) }), prepare, parse)).toEqual({ gate: "timeout" }); + expect(canceled).toBe(true); + } finally { Object.defineProperty(AbortSignal, "timeout", descriptor); } + }); + + // The destination is the shared decision-endpoint contract (jevDecisionEndpointUrl, #6731), not a copy. + describe("decision endpoint URL contract", () => { + const answer = { answers: { test: { choice: "yes" }, route: { choice: "a/m2:low" } } }; + async function sendTo(baseUrl: string) { + const sent: Array<{ url: string; body: string }> = []; + const post: Post = async (_name, _provider, url, init) => { + expect(new Headers(init.headers).get("authorization")).toBe("Bearer fixture-own-key"); + sent.push({ url, body: String(init.body) }); + return Response.json(answer); + }; + const opts = options({ config: config("decider", { ...row, baseUrl }), post }); + const direct = await exchangeJevDecision(opts, prepare, parse); + const route = await resolveJevDecision({ ...opts, body: { input: "Choose a target." }, candidates, fallback }); + return { sent, direct, route }; + } + + test.each([ + ["https://decider.example/v1/decisions", "https://decider.example/v1/decisions"], + ["https://decider.example/v1/decisions/", "https://decider.example/v1/decisions/"], + ["https://decider.example/v1/decisions//", "https://decider.example/v1/decisions//"], + [" https://decider.example/v1/decisions/ ", "https://decider.example/v1/decisions/"], + ["https://Decider.Example/v1/Decisions/", "https://Decider.Example/v1/Decisions/"], + ["https://decider.example/v1/systemone/extra", "https://decider.example/v1/systemone/extra"], + ["https://decider.example/v1/systemone", "https://decider.example/v1/systemone"], + ["https://decider.example/v1/systemone/", "https://decider.example/v1/systemone"], + ["https://decider.example/v1/systemone//", "https://decider.example/v1/systemone"], + ["http://127.0.0.1:11434/v1/systemone/", "http://127.0.0.1:11434/v1/systemone"], + ])("baseUrl %j is sent to %j through the exchange and the route", async (baseUrl, expected) => { + const { sent, direct, route } = await sendTo(baseUrl); + expect(sent.map(request => request.url)).toEqual([expected, expected]); + expect(direct).toEqual({ value: answer }); + expect(route.gate).toBe("apply"); + }); + + test("the request body does not depend on the endpoint path", async () => { + const bodies = new Set(); + for (const baseUrl of ["https://decider.example/v1/decisions/", "https://decider.example/v1/systemone/"]) { + const { sent } = await sendTo(baseUrl); + expect(sent).toHaveLength(2); + sent.forEach(request => bodies.add(request.body)); + } + expect(bodies.size).toBe(2); // one exchange body and one route body, identical across both endpoints + }); + + test.each([ + "https://decider.example/v1/systemone?mode=x", "https://decider.example/v1/decisions?", + "https://decider.example/v1/systemone#part", "https://decider.example/v1/decisions#", + "https://user:pass@example.test/v1/systemone", "https://@decider.example/v1/decisions", + "https:\t//@decider.example/v1/decisions", "https://decider.example/v1/system\tone", + "http://decider.example/v1/systemone", "http://127.0.0.1:11434/v1/decisions", + "ftp://decider.example/v1/systemone", "not a url", + ])("baseUrl %j is refused before any send", async (baseUrl) => { + const { sent, direct, route } = await sendTo(baseUrl); + expect(sent).toEqual([]); + expect(direct).toEqual({ gate: "missing_key" }); + expect(route.gate).toBe("missing_key"); + }); + }); +}); + +const candidates: JevCandidate[] = [ + { key: "a/m1", provider: "a", model: "m1", reasoningEfforts: ["low"] }, + { key: "a/m2", provider: "a", model: "m2", reasoningEfforts: ["low"] }, +]; +const fallback = { targetKey: "a/m1", effort: "low" as const };