diff --git a/bun.lock b/bun.lock index a2679f9..9e2ffc9 100644 --- a/bun.lock +++ b/bun.lock @@ -80,7 +80,7 @@ }, "packages/plugins/nexus-service": { "name": "@microvoid/convax-plugin-nexus-service", - "version": "0.4.0", + "version": "0.4.1", "dependencies": { "@microvoid/convax-nexus-mcp": "workspace:*", }, @@ -275,7 +275,7 @@ }, "packages/tools/nexus-mcp": { "name": "@microvoid/convax-nexus-mcp", - "version": "0.3.13", + "version": "0.3.14", "bin": { "convax-nexus-mcp": "dist/convax-nexus-mcp", }, diff --git a/packages/plugins/nexus-service/convax-package.json b/packages/plugins/nexus-service/convax-package.json index 382a2b2..33bd4f2 100644 --- a/packages/plugins/nexus-service/convax-package.json +++ b/packages/plugins/nexus-service/convax-package.json @@ -4,11 +4,11 @@ "id": "nexus-service", "name": "Convax Account", "description": "Connects Convax to your Convax Account for OpenRouter chat and live image-model generation through short-lived Data Tokens.", - "version": "0.4.0", + "version": "0.4.1", "companions": [ { "command": "convax-nexus-mcp", - "version": "0.3.13", + "version": "0.3.14", "source": "packages/tools/nexus-mcp", "targets": [ { diff --git a/packages/plugins/nexus-service/package.json b/packages/plugins/nexus-service/package.json index 3e548bc..69b4c08 100644 --- a/packages/plugins/nexus-service/package.json +++ b/packages/plugins/nexus-service/package.json @@ -1,6 +1,6 @@ { "name": "@microvoid/convax-plugin-nexus-service", - "version": "0.4.0", + "version": "0.4.1", "private": true, "type": "module", "dependencies": { diff --git a/packages/plugins/nexus-service/package/manifest.json b/packages/plugins/nexus-service/package/manifest.json index ed0caba..45568d2 100644 --- a/packages/plugins/nexus-service/package/manifest.json +++ b/packages/plugins/nexus-service/package/manifest.json @@ -3,7 +3,7 @@ "id": "nexus-service", "name": "Convax Account", "description": "Connects Convax to your Convax Account for OpenRouter chat and live image-model generation through short-lived Data Tokens.", - "version": "0.4.0", + "version": "0.4.1", "contributes": { "generation": { "models": [ @@ -18,7 +18,11 @@ "title": "Nexus ยท OpenRouter Image", "description": "Generate an image with an image-output model available through the connected Nexus OpenRouter Provider.", "output": "image", - "acceptedInputs": [] + "acceptedInputs": [], + "recovery": { + "schema": "convax.generation-lro/1", + "mode": "long-running-operation" + } } ] }, diff --git a/packages/tools/nexus-mcp/README.md b/packages/tools/nexus-mcp/README.md index 10896a1..67da6bf 100644 --- a/packages/tools/nexus-mcp/README.md +++ b/packages/tools/nexus-mcp/README.md @@ -15,20 +15,29 @@ until the service adds a dedicated video Usage Inspector and quota settlement mo When the live image catalog is bounded, the companion marks its model field so a compatible host can present each image model as a direct choice. If that catalog cannot be loaded, has no image models, or exceeds the bounded choice limit, the -companion omits image generation from `tools/list` instead of exposing a free-text -model field. +companion keeps the fixed Service and LLM tools available but omits image generation +instead of exposing a free-text model field. The companion keeps the validated image catalog, Provider route, and short-lived Data Token together in one bounded in-memory route. The `tools/call` that follows `tools/list` reuses that exact route and never repeats authorization, Provider, or -model discovery inside the generation call. Route credentials remain private to -the companion and are invalidated before their token refresh window. +model discovery inside the generation call. The selected concrete model is checked +against that route immediately before execution; a stale catalog is invalidated and +reloaded instead of silently using another model. Route credentials remain private +to the companion and are invalidated before their token refresh window. -Each image request sends the host operation id as `x-nexus-request-id` for safe -diagnostic correlation. A rejected request exposes only its HTTP status, an -allow-listed Gateway error code, and that locally generated operation id to the MCP -caller. Response-provided identifiers, raw upstream messages, response bodies, -prompts, and tokens never cross that diagnostic boundary. +Each image request sends the host operation id as both `idempotency-key` and +`x-nexus-request-id`. Before contacting Nexus, the companion durably records the +operation and stable task receipt in its Host-provisioned private recovery +directory. Restart recovery reuses the same operation id and request fingerprint; +a completed result is byte-verified and replayed from the private journal without +contacting Nexus again. A changed request, result digest, or stored artifact fails +closed. + +Canvas receives one bounded failure message. Exact bounded HTTP status, allowlisted +Gateway code, operation-correlated request id, and safe cause-chain messages stay in +Main/sidecar logs. Response-provided identifiers, raw upstream messages, response +bodies, prompts, tokens, and native paths never cross that diagnostic boundary. The companion owns PKCE and the loopback callback, stores the rotating Nexus refresh grant in a private user file, and exposes only a random loopback Gateway credential to @@ -39,7 +48,8 @@ Nexus Plan, quota, subscription state, and allowlisted Checkout Plans from the authorized User API. AI budget balances are displayed in USD; during a rolling Nexus upgrade, legacy micro-USD quota fields are converted locally instead of being shown as raw quota units. The fixed `service.checkout` tool accepts only one advertised -Plan key, reuses a private idempotency key for retries, and returns a Checkout URL +Plan key, single-flights concurrent creation for that account and Plan, reuses a +private idempotency key for transport retries, and returns a Checkout URL only to Convax Main for strict HTTPS validation and system-browser navigation. Redirect Checkouts open the Provider URL directly. QR Checkouts open the trusted Nexus Account Portal, which restores that Checkout and renders its native payment @@ -47,8 +57,8 @@ code without exposing a non-HTTPS Provider URL to the Convax host. Neither the retry record nor any Renderer-facing value contains a Nexus token, Provider Product id, payment credential, or Checkout URL. -During a rolling Nexus API upgrade, a deployment may still return the base Access -shape without embedded Plan or Billing metadata. The companion continues to verify -the Workspace, Provider, and dedicated Quota endpoint in that case, reports Plan -and Billing as unavailable, and enables Checkout only after Nexus advertises its -authoritative allowlisted Plans. +The authoritative base Access shape may omit optional Plan or Billing metadata; in +that case those optional fields are unavailable by definition. Failure of any +required Access, Provider, or Quota request fails the complete `service.status` +operation rather than returning a partial status. Sign-out clears the durable +Checkout attempt even when remote revoke fails. diff --git a/packages/tools/nexus-mcp/package.json b/packages/tools/nexus-mcp/package.json index 89d22dd..50d53f8 100644 --- a/packages/tools/nexus-mcp/package.json +++ b/packages/tools/nexus-mcp/package.json @@ -1,6 +1,6 @@ { "name": "@microvoid/convax-nexus-mcp", - "version": "0.3.13", + "version": "0.3.14", "private": true, "license": "MIT", "type": "module", diff --git a/packages/tools/nexus-mcp/src/image-generator.ts b/packages/tools/nexus-mcp/src/image-generator.ts index 1aa8d4a..3e001a5 100644 --- a/packages/tools/nexus-mcp/src/image-generator.ts +++ b/packages/tools/nexus-mcp/src/image-generator.ts @@ -1,27 +1,35 @@ -import { createHash } from "node:crypto"; -import { constants } from "node:fs"; -import { lstat, open, rm } from "node:fs/promises"; -import path from "node:path"; - import { asRecord, generationCallSchema, type GenerationArtifact, type GenerationCall, } from "./contracts.ts"; -import type { NexusImageRoute } from "./nexus-client.ts"; +import { + NexusImageHttpError, + type NexusImageRoute, +} from "./nexus-client.ts"; +import { + nexusGenerationRequestFingerprint, + NexusOperationStore, + type NexusOperationRecord, + type PreparedNexusOperation, + type RunningNexusOperation, + type StoredNexusGenerationRequest, +} from "./operation-store.ts"; const maximumPromptBytes = 20_000; const maximumImageBytes = 12 * 1024 * 1024; const maximumTotalImageBytes = 32 * 1024 * 1024; const maximumImages = 8; -const maximumTrackedOperations = 256; const modelIdPattern = /^~?[A-Za-z0-9]+(?:[._/:-][A-Za-z0-9]+)*$/u; const operationIdPattern = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/u; +const digestPattern = /^[a-f0-9]{64}$/u; interface TrackedOperation { + controller: AbortController; fingerprint: string; - result: Promise; + requestDigest: string; + result: Promise; } interface DecodedImage { @@ -31,97 +39,354 @@ interface DecodedImage { } export class NexusImageGenerator { - readonly #operations = new Map(); + readonly #tracked = new Map(); - generate( + constructor( + private readonly operations: NexusOperationStore, + private readonly resumeCompletion: ( + request: StoredNexusGenerationRequest, + operationId: string, + signal: AbortSignal, + ) => Promise, + ) {} + + async generate( value: unknown, - resolveRoute: () => NexusImageRoute, + metadata: { operationId: string; requestDigest: string }, + resolveRoute: ( + modelId: string, + signal: AbortSignal, + ) => Promise, signal: AbortSignal, + onSubmitted: (taskId: string) => void, ): Promise { const call = parseGenerationCall(value); - const fingerprint = createHash("sha256") - .update(JSON.stringify(call)) - .digest("hex"); - const existing = this.#operations.get(call.operation_id); + if ( + call.operation_id !== metadata.operationId || + !digestPattern.test(metadata.requestDigest) + ) { + throw new Error("Nexus generation recovery metadata is invalid"); + } + + const existing = await this.operations.find(call.operation_id); + let record: NexusOperationRecord; if (existing) { - if (existing.fingerprint !== fingerprint) { - throw new Error( - "Nexus generation operation id was reused with different input", - ); + this.#assertCallMatches(existing, call, metadata.requestDigest); + record = + existing.status === "prepared" || existing.status === "running" + ? await this.#track( + existing, + (executionSignal) => + this.resumeCompletion( + existing.request, + call.operation_id, + executionSignal, + ), + signal, + onSubmitted, + ) + : existing; + } else { + const route = await resolveRoute(call.model, signal); + const model = route.models.find(({ id }) => id === call.model); + if (!model) { + throw new Error("The selected Nexus image model is unavailable"); } - return existing.result; + const request: StoredNexusGenerationRequest = { + model: model.id, + outputModalities: [...model.outputModalities], + prompt: call.prompt, + }; + const prepared = await this.operations.createPrepared({ + fingerprint: nexusGenerationRequestFingerprint(request), + operationId: call.operation_id, + request, + requestDigest: metadata.requestDigest, + }); + record = await this.#track( + prepared, + (executionSignal) => + route.complete( + model, + call.prompt, + call.operation_id, + executionSignal, + ), + signal, + onSubmitted, + ); } - if (this.#operations.size >= maximumTrackedOperations) { - throw new Error( - "Nexus generation operation history is full; restart the companion before generating again", + + if (record.status === "succeeded") { + return this.operations.materialize( + call.operation_id, + record, + call.output_directory, ); } - const result = this.#generate(call, resolveRoute, signal); - this.#operations.set(call.operation_id, { fingerprint, result }); - return result; + if (record.status === "failed") throw new Error(record.error.message); + if (record.status === "cancelled") throw abortError(); + throw new Error("Nexus generation did not reach a terminal state"); } - async #generate( - call: GenerationCall, - resolveRoute: () => NexusImageRoute, + async get( + operationId: string, + requestDigest: string, + taskId?: string, + ): Promise { + const record = await this.operations.find(operationId); + if (!record) return null; + this.#assertIdentity(record, requestDigest, taskId); + return record; + } + + async wait( + operationId: string, + requestDigest: string, + taskId: string | undefined, signal: AbortSignal, - ): Promise { - if (signal.aborted) throw abortError(); - const outputDirectory = await validateOutputDirectory( - call.output_directory, - ); - if (signal.aborted) throw abortError(); - const route = resolveRoute(); - const model = route.models.find(({ id }) => id === call.model); - if (!model) { - throw new Error("The selected Nexus image model is unavailable"); + ): Promise { + const record = await this.get(operationId, requestDigest, taskId); + if ( + !record || + (record.status !== "prepared" && record.status !== "running") + ) { + return record; } - const response = await route.complete( - model, - call.prompt, - call.operation_id, + return this.#track( + record, + (executionSignal) => + this.resumeCompletion(record.request, operationId, executionSignal), signal, ); - if (signal.aborted) throw abortError(); - const images = parseImages(response); - const artifacts: GenerationArtifact[] = []; - const createdPaths: string[] = []; - try { - for (let index = 0; index < images.length; index += 1) { - if (signal.aborted) throw abortError(); - const image = images[index]!; - const name = `nexus-${safeStem(call.operation_id)}-${index + 1}.${image.extension}`; - const outputPath = path.join(outputDirectory, name); - const handle = await open( - outputPath, - constants.O_WRONLY | - constants.O_CREAT | - constants.O_EXCL | - constants.O_NOFOLLOW, - 0o600, + } + + async cancel( + operationId: string, + requestDigest: string, + taskId?: string, + ): Promise { + const record = await this.get(operationId, requestDigest, taskId); + if ( + !record || + record.status === "succeeded" || + record.status === "failed" || + record.status === "cancelled" + ) { + return record; + } + const tracked = this.#tracked.get(operationId); + if (tracked) { + tracked.controller.abort("Nexus generation was cancelled"); + await tracked.result.catch(() => undefined); + return this.get(operationId, requestDigest, taskId); + } + return this.operations.cancel(operationId, record); + } + + async result( + operationId: string, + requestDigest: string, + taskId: string, + resultDigest: string, + outputDirectory: string, + ) { + const record = await this.get(operationId, requestDigest, taskId); + if ( + record?.status !== "succeeded" || + record.resultDigest !== resultDigest + ) { + throw new Error("Nexus generation result is not available"); + } + return { + artifacts: await this.operations.materialize( + operationId, + record, + outputDirectory, + ), + resultDigest, + }; + } + + async acknowledge( + operationId: string, + requestDigest: string, + taskId?: string, + resultDigest?: string, + ) { + const record = await this.get(operationId, requestDigest, taskId); + if (!record) return; + if ( + record.status === "prepared" || + record.status === "running" || + (record.status === "succeeded" && + record.resultDigest !== resultDigest) || + (record.status !== "succeeded" && resultDigest !== undefined) + ) { + throw new Error("Nexus generation acknowledgement is invalid"); + } + await this.operations.acknowledge(operationId, record); + } + + #track( + record: PreparedNexusOperation | RunningNexusOperation, + complete: (signal: AbortSignal) => Promise, + signal: AbortSignal, + onSubmitted?: (taskId: string) => void, + ) { + const existing = this.#tracked.get(record.operationId); + if (existing) { + if ( + existing.fingerprint !== record.fingerprint || + existing.requestDigest !== record.requestDigest + ) { + throw new Error( + "Nexus generation operation id was reused with different input", ); - createdPaths.push(outputPath); - try { - await handle.writeFile(image.bytes); - await handle.sync(); - } finally { - await handle.close(); - } - // Convax owns output_directory and admits only portable paths relative - // to it. Keep the native path private to this writer and return the - // generated file name through the MCP result contract. - artifacts.push({ mimeType: image.mimeType, name, path: name }); } - return artifacts; + return existing.result; + } + const controller = new AbortController(); + const result = this.#perform( + record, + complete, + AbortSignal.any([signal, controller.signal]), + onSubmitted, + ); + const tracked: TrackedOperation = { + controller, + fingerprint: record.fingerprint, + requestDigest: record.requestDigest, + result, + }; + this.#tracked.set(record.operationId, tracked); + const cleanup = () => { + if (this.#tracked.get(record.operationId) === tracked) { + this.#tracked.delete(record.operationId); + } + }; + void result.then(cleanup, cleanup); + return result; + } + + async #perform( + initial: PreparedNexusOperation | RunningNexusOperation, + complete: (signal: AbortSignal) => Promise, + signal: AbortSignal, + onSubmitted?: (taskId: string) => void, + ): Promise { + const running = + initial.status === "running" + ? initial + : await this.operations.markRunning(initial.operationId, initial); + onSubmitted?.(running.taskId); + try { + if (signal.aborted) throw abortError(); + const response = await complete(signal); + if (signal.aborted) throw abortError(); + return await this.operations.succeed( + running.operationId, + running, + parseImages(response), + ); } catch (error) { - await Promise.all( - createdPaths.map((outputPath) => - rm(outputPath, { force: true }).catch(() => undefined), - ), + // Process shutdown intentionally leaves the durable running record in + // place. The next sidecar reattaches with the same operation id. + if (signal.aborted && signal.reason === "MCP server is closing") { + throw error; + } + if (signal.aborted) { + return this.operations.cancel(running.operationId, running); + } + console.error( + "[nexus] image generation failed", + imageGenerationFailureDiagnostics(error), + ); + return this.operations.fail( + running.operationId, + running, + safeStoredFailure(error), ); - throw error; } } + + #assertCallMatches( + record: NexusOperationRecord, + call: GenerationCall, + requestDigest: string, + ) { + this.#assertIdentity(record, requestDigest); + if ( + record.request.model !== call.model || + record.request.prompt !== call.prompt || + record.fingerprint !== nexusGenerationRequestFingerprint(record.request) + ) { + throw new Error( + "Nexus generation operation id was reused with different input", + ); + } + } + + #assertIdentity( + record: NexusOperationRecord, + requestDigest: string, + taskId?: string, + ) { + if ( + !digestPattern.test(requestDigest) || + record.requestDigest !== requestDigest || + (taskId !== undefined && record.taskId !== taskId) + ) { + throw new Error("Nexus generation recovery identity changed"); + } + } +} + +function safeStoredFailure(error: unknown) { + if (error instanceof NexusImageHttpError) { + return { + code: "gateway_rejected", + message: `Nexus rejected image generation with HTTP ${error.status}.`, + }; + } + return { + code: "generation_failed", + message: "Nexus image generation failed.", + }; +} + +function imageGenerationFailureDiagnostics(error: unknown) { + const nexus = + error instanceof NexusImageHttpError + ? { + code: error.code, + requestId: error.requestId, + status: error.status, + } + : undefined; + const diagnostics: Array<{ message: string; name: string }> = []; + let current: unknown = error; + for (let depth = 0; depth < 4 && current !== undefined; depth += 1) { + if (!(current instanceof Error)) { + diagnostics.push({ + message: String(current) + .replace(/[\u0000-\u001f\u007f]+/gu, " ") + .slice(0, 1_000), + name: typeof current, + }); + break; + } + diagnostics.push({ + message: current.message + .replace(/[\u0000-\u001f\u007f]+/gu, " ") + .slice(0, 1_000), + name: /^[A-Za-z][A-Za-z0-9._-]{0,63}$/u.test(current.name) + ? current.name + : "Error", + }); + current = current.cause; + } + return { diagnostics, ...(nexus === undefined ? {} : { nexus }) }; } function parseGenerationCall(value: unknown): GenerationCall { @@ -182,16 +447,6 @@ function parseGenerationCall(value: unknown): GenerationCall { }; } -async function validateOutputDirectory(value: string) { - if (!path.isAbsolute(value)) - throw new Error("Nexus generation output directory must be absolute"); - const metadata = await lstat(value); - if (!metadata.isDirectory() || metadata.isSymbolicLink()) { - throw new Error("Nexus generation output directory is invalid"); - } - return value; -} - function parseImages(value: unknown): readonly DecodedImage[] { const response = asRecord(value, "Nexus image response"); if ( @@ -322,15 +577,6 @@ function trimmedString( return value; } -function safeStem(operationId: string) { - const stem = operationId - .toLowerCase() - .replaceAll(/[^a-z0-9_-]/gu, "-") - .replaceAll(/-+/gu, "-") - .slice(0, 32); - return stem || "image"; -} - function abortError() { return new DOMException("Nexus image generation was cancelled", "AbortError"); } diff --git a/packages/tools/nexus-mcp/src/llm-gateway.ts b/packages/tools/nexus-mcp/src/llm-gateway.ts index ac907c8..1e16e97 100644 --- a/packages/tools/nexus-mcp/src/llm-gateway.ts +++ b/packages/tools/nexus-mcp/src/llm-gateway.ts @@ -154,8 +154,9 @@ export class NexusLlmGateway { const headers = new Headers({ "Cache-Control": "no-store" }); const contentType = upstream.headers.get("content-type"); if (contentType) headers.set("Content-Type", contentType); - const requestId = upstream.headers.get("x-request-id"); - if (requestId) headers.set("X-Request-Id", requestId.slice(0, 191)); + copyBoundedHeader(upstream.headers, headers, "retry-after"); + copyBoundedHeader(upstream.headers, headers, "x-generation-id"); + copyBoundedHeader(upstream.headers, headers, "x-request-id"); return new Response(upstream.body, { headers, status: upstream.status }); } catch { if (request.signal.aborted) @@ -169,3 +170,18 @@ export class NexusLlmGateway { this.#server = undefined; } } + +function copyBoundedHeader( + source: Headers, + target: Headers, + name: string, +) { + const value = source.get(name); + if ( + value && + value.length <= 191 && + !/[\u0000-\u001f\u007f]/u.test(value) + ) { + target.set(name, value); + } +} diff --git a/packages/tools/nexus-mcp/src/mcp-server.ts b/packages/tools/nexus-mcp/src/mcp-server.ts index 8c0ebd2..1a42678 100644 --- a/packages/tools/nexus-mcp/src/mcp-server.ts +++ b/packages/tools/nexus-mcp/src/mcp-server.ts @@ -1,4 +1,12 @@ -import { asRecord, type JsonRpcRequest, type ToolResult } from "./contracts.ts"; +import os from "node:os"; +import path from "node:path"; + +import { + asRecord, + type GenerationArtifact, + type JsonRpcRequest, + type ToolResult, +} from "./contracts.ts"; import { NexusAuthorization } from "./authorization.ts"; import { NexusCheckoutStore } from "./checkout-store.ts"; import { @@ -9,12 +17,26 @@ import { } from "./nexus-client.ts"; import { NexusImageGenerator } from "./image-generator.ts"; import { NexusLlmGateway } from "./llm-gateway.ts"; +import { + nexusGenerationResult, + NexusOperationStore, + type NexusOperationRecord, +} from "./operation-store.ts"; import { NexusPluginService } from "./plugin-service.ts"; import { NexusSessionStore } from "./session-store.ts"; const protocolVersion = "2025-03-26"; const maximumRequestBytes = 4 * 1024 * 1024; const imageModelCatalogTtlMs = 60_000; +const generationLroMethods = new Set([ + "convax/generation/operations/get", + "convax/generation/operations/wait", + "convax/generation/operations/cancel", + "convax/generation/operations/result", + "convax/generation/operations/acknowledge", +]); +const generationLroRequestSchema = "convax.generation-lro-request/1"; +const generationLroSnapshotSchema = "convax.generation-lro-snapshot/1"; const emptyInputSchema = { additionalProperties: false, properties: {}, @@ -178,7 +200,7 @@ function isJsonRpcRequest(value: unknown): value is JsonRpcRequest { export function publicImageGenerationErrorMessage(error: unknown) { if (!(error instanceof NexusImageHttpError)) { - return "Nexus image generation failed. Check Nexus before retrying because the upstream task result may be unknown."; + return "Nexus image generation failed."; } const details = [ `HTTP ${error.status}`, @@ -202,6 +224,7 @@ export interface NexusMcpServerOptions { checkouts?: NexusCheckoutStore; client?: NexusClientOptions; environment?: Readonly>; + operationStore?: NexusOperationStore; send?: (value: unknown) => void; sessions?: NexusSessionStore; } @@ -215,6 +238,7 @@ export class NexusMcpServer { readonly #sendValue: (value: unknown) => void; readonly #service: NexusPluginService; readonly #imageGenerator: NexusImageGenerator; + readonly #operationStore: NexusOperationStore; #activeImageRoute: NexusImageRoute | undefined; #imageRouteEpoch = 0; #imageRouteExpiresAt = 0; @@ -237,7 +261,24 @@ export class NexusMcpServer { checkouts, ); this.#gateway = new NexusLlmGateway(client); - this.#imageGenerator = new NexusImageGenerator(); + this.#operationStore = + options.operationStore ?? + new NexusOperationStore( + nexusOperationDirectory(options.environment ?? process.env), + ); + this.#imageGenerator = new NexusImageGenerator( + this.#operationStore, + (request, operationId, signal) => + client.imageCompletion( + { + id: request.model, + outputModalities: request.outputModalities, + }, + request.prompt, + operationId, + signal, + ), + ); this.#sendValue = options.send ?? ((value) => { @@ -280,9 +321,21 @@ export class NexusMcpServer { } #dispatch(value: unknown) { - const handler = this.#handle(value); + const handler = this.#handle(value).catch((error: unknown) => { + console.error("[nexus] MCP request handler failed", errorDiagnostics(error)); + const requestId = + isJsonRpcRequest(value) && + (typeof value.id === "string" || + (typeof value.id === "number" && Number.isFinite(value.id))) + ? value.id + : undefined; + if (requestId !== undefined) { + this.#sendError(requestId, -32_603, "Internal error"); + } + }); this.#handlers.add(handler); - void handler.finally(() => this.#handlers.delete(handler)); + const cleanup = () => this.#handlers.delete(handler); + void handler.then(cleanup, cleanup); } async #handle(value: unknown) { @@ -313,12 +366,25 @@ export class NexusMcpServer { return; } this.#sendResult(value.id, { - capabilities: { tools: {} }, + capabilities: { + experimental: { + "convax/generation-lro": { + binding: await this.#operationStore.binding(), + mode: "long-running-operation", + schema: "convax.generation-lro/1", + }, + }, + tools: {}, + }, protocolVersion, - serverInfo: { name: "convax-nexus-mcp", version: "0.3.13" }, + serverInfo: { name: "convax-nexus-mcp", version: "0.3.14" }, }); return; } + if (generationLroMethods.has(value.method)) { + await this.#callGenerationRecovery({ ...value, id: value.id }); + return; + } if (value.method === "tools/list") { this.#sendResult(value.id, { tools: await this.#listedTools() }); return; @@ -387,11 +453,26 @@ export class NexusMcpServer { input.plan_key as string, ); } else if (params.name === "image.generate") { + const operation = generationOperationMetadata(params); structuredContent = { artifacts: await this.#imageGenerator.generate( input, - () => this.#currentImageRoute(), + operation, + (modelId, signal) => this.#imageRouteForModel(modelId, signal), controller.signal, + (taskId) => { + if (!operation.progressToken) return; + this.#sendValue({ + jsonrpc: "2.0", + method: "notifications/progress", + params: { + event: "submitted", + progressToken: operation.progressToken, + schema: "convax.generation-lifecycle/1", + taskId, + }, + }); + }, ), }; } else if (params.name === "llm.models.list") { @@ -399,14 +480,24 @@ export class NexusMcpServer { } else { structuredContent = await this.#gateway.start(); } - this.#sendResult(request.id, { - content: [{ text: "Nexus service operation completed.", type: "text" }], - structuredContent, - } satisfies ToolResult); + this.#sendResult( + request.id, + toolName === "image.generate" + ? nexusGenerationResult( + structuredContent.artifacts as GenerationArtifact[], + ) + : ({ + content: [ + { text: "Nexus service operation completed.", type: "text" }, + ], + structuredContent, + } satisfies ToolResult), + ); } catch (error) { const cancelled = controller.signal.aborted; console.error( cancelled ? "[nexus] request cancelled" : "[nexus] request failed", + errorDiagnostics(error), ); this.#sendResult(request.id, { content: [ @@ -426,6 +517,93 @@ export class NexusMcpServer { } } + async #callGenerationRecovery( + request: JsonRpcRequest & { id: number | string }, + ) { + const controller = new AbortController(); + this.#inflight.set(request.id, controller); + try { + const input = generationRecoveryRequest( + request.params, + request.method === "convax/generation/operations/result", + ); + if (request.method === "convax/generation/operations/get") { + this.#sendResult( + request.id, + recoverySnapshot( + await this.#imageGenerator.get( + input.operationId, + input.requestDigest, + input.taskId, + ), + ), + ); + return; + } + if (request.method === "convax/generation/operations/wait") { + this.#sendResult( + request.id, + recoverySnapshot( + await this.#imageGenerator.wait( + input.operationId, + input.requestDigest, + input.taskId, + controller.signal, + ), + ), + ); + return; + } + if (request.method === "convax/generation/operations/cancel") { + this.#sendResult( + request.id, + recoverySnapshot( + await this.#imageGenerator.cancel( + input.operationId, + input.requestDigest, + input.taskId, + ), + ), + ); + return; + } + if (request.method === "convax/generation/operations/result") { + if ( + !input.taskId || + !input.resultDigest || + !input.outputDirectory + ) { + throw new Error("Nexus generation result request is incomplete"); + } + const replay = await this.#imageGenerator.result( + input.operationId, + input.requestDigest, + input.taskId, + input.resultDigest, + input.outputDirectory, + ); + this.#sendResult(request.id, { + result: nexusGenerationResult(replay.artifacts), + resultDigest: replay.resultDigest, + schema: "convax.generation-lro-result/1", + }); + return; + } + await this.#imageGenerator.acknowledge( + input.operationId, + input.requestDigest, + input.taskId, + input.resultDigest, + ); + this.#sendResult(request.id, { + acknowledged: true, + schema: "convax.generation-lro-acknowledgement/1", + }); + } finally { + this.#inflight.delete(request.id); + } + } + async #listedTools() { try { const route = await this.#loadImageRoute(); @@ -438,6 +616,21 @@ export class NexusMcpServer { } } + async #imageRouteForModel(modelId: string, signal: AbortSignal) { + if (signal.aborted) { + throw new DOMException("Request was cancelled", "AbortError"); + } + let route = await this.#loadImageRoute(); + if (!route.models.some(({ id }) => id === modelId)) { + this.#invalidateImageRoute(); + route = await this.#loadImageRoute(); + } + if (!route.models.some(({ id }) => id === modelId)) { + throw new Error("The selected Nexus image model is unavailable"); + } + return route; + } + #loadImageRoute(): Promise { if ( this.#imageRouteRequest && @@ -480,19 +673,6 @@ export class NexusMcpServer { return request; } - #currentImageRoute(): NexusImageRoute { - const route = this.#activeImageRoute; - if ( - !route || - Date.now() >= this.#imageRouteExpiresAt || - !route.isCurrent() - ) { - this.#invalidateImageRoute(); - throw new Error("Nexus image models must be refreshed before generation"); - } - return route; - } - #invalidateImageRoute() { this.#imageRouteEpoch += 1; this.#activeImageRoute = undefined; @@ -536,3 +716,168 @@ export class NexusMcpServer { } } } + +function nexusOperationDirectory( + environment: Readonly>, +) { + const hostDirectory = environment.CONVAX_GENERATION_LRO_DIRECTORY?.trim(); + if (hostDirectory) { + if (!path.isAbsolute(hostDirectory)) { + throw new Error("Nexus generation recovery directory must be absolute"); + } + return hostDirectory; + } + const configured = environment.XDG_CONFIG_HOME?.trim(); + const root = + configured && path.isAbsolute(configured) + ? configured + : path.join(environment.HOME || os.homedir(), ".config"); + return path.join(root, "convax", "nexus", "image-operations"); +} + +function generationOperationMetadata(params: Record) { + const meta = asRecord(params._meta, "generation metadata"); + const operation = asRecord( + meta.convaxGeneration, + "generation operation metadata", + ); + if ( + operation.schema !== "convax.generation-operation/1" || + operation.recovery !== "required" || + typeof operation.operationId !== "string" || + !/^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/u.test(operation.operationId) || + typeof operation.requestDigest !== "string" || + !/^[a-f0-9]{64}$/u.test(operation.requestDigest) || + (meta.progressToken !== undefined && + (typeof meta.progressToken !== "string" || + meta.progressToken.length < 1 || + meta.progressToken.length > 512 || + /[\u0000-\u001f\u007f]/u.test(meta.progressToken))) + ) { + throw new Error("Nexus generation operation metadata is invalid"); + } + return { + operationId: operation.operationId, + ...(typeof meta.progressToken === "string" + ? { progressToken: meta.progressToken } + : {}), + requestDigest: operation.requestDigest, + }; +} + +interface GenerationRecoveryRequestValue { + operationId: string; + outputDirectory?: string; + requestDigest: string; + resultDigest?: string; + taskId?: string; +} + +function generationRecoveryRequest( + value: unknown, + allowOutputDirectory: boolean, +): GenerationRecoveryRequestValue { + const input = asRecord(value, "generation recovery request"); + const allowed = new Set([ + "operationId", + "outputDirectory", + "requestDigest", + "resultDigest", + "schema", + "taskId", + ]); + if ( + Object.keys(input).some((key) => !allowed.has(key)) || + input.schema !== generationLroRequestSchema || + typeof input.operationId !== "string" || + !/^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/u.test(input.operationId) || + typeof input.requestDigest !== "string" || + !/^[a-f0-9]{64}$/u.test(input.requestDigest) || + (input.taskId !== undefined && + (typeof input.taskId !== "string" || + !/^[A-Za-z0-9][A-Za-z0-9._:-]{0,511}$/u.test(input.taskId))) || + (input.resultDigest !== undefined && + (typeof input.resultDigest !== "string" || + !/^[a-f0-9]{64}$/u.test(input.resultDigest))) || + (input.outputDirectory !== undefined && + (!allowOutputDirectory || + typeof input.outputDirectory !== "string" || + !path.isAbsolute(input.outputDirectory) || + input.outputDirectory.includes("\0") || + input.outputDirectory.length > 4_096)) + ) { + throw new Error("Nexus generation recovery request is invalid"); + } + return { + operationId: input.operationId, + ...(typeof input.outputDirectory === "string" + ? { outputDirectory: input.outputDirectory } + : {}), + requestDigest: input.requestDigest, + ...(typeof input.resultDigest === "string" + ? { resultDigest: input.resultDigest } + : {}), + ...(typeof input.taskId === "string" ? { taskId: input.taskId } : {}), + }; +} + +function recoverySnapshot(record: NexusOperationRecord | null) { + if (!record) { + return { schema: generationLroSnapshotSchema, status: "absent" }; + } + if (record.status === "prepared") { + return { schema: generationLroSnapshotSchema, status: "prepared" }; + } + if (record.status === "running") { + return { + schema: generationLroSnapshotSchema, + status: "running", + taskId: record.taskId, + }; + } + if (record.status === "succeeded") { + return { + resultDigest: record.resultDigest, + schema: generationLroSnapshotSchema, + status: "succeeded", + taskId: record.taskId, + }; + } + if (record.status === "failed") { + return { + error: { ...record.error }, + schema: generationLroSnapshotSchema, + status: "failed", + taskId: record.taskId, + }; + } + return { + schema: generationLroSnapshotSchema, + status: "cancelled", + taskId: record.taskId, + }; +} + +function errorDiagnostics(error: unknown) { + const diagnostics: Array<{ message: string; name: string }> = []; + let current: unknown = error; + for (let depth = 0; depth < 4 && current !== undefined; depth += 1) { + if (!(current instanceof Error)) { + diagnostics.push({ + message: String(current).replace(/[\u0000-\u001f\u007f]+/gu, " ").slice(0, 1_000), + name: typeof current, + }); + break; + } + diagnostics.push({ + message: current.message + .replace(/[\u0000-\u001f\u007f]+/gu, " ") + .slice(0, 1_000), + name: /^[A-Za-z][A-Za-z0-9._-]{0,63}$/u.test(current.name) + ? current.name + : "Error", + }); + current = current.cause; + } + return { diagnostics }; +} diff --git a/packages/tools/nexus-mcp/src/nexus-client.ts b/packages/tools/nexus-mcp/src/nexus-client.ts index 4a64a98..243dd7f 100644 --- a/packages/tools/nexus-mcp/src/nexus-client.ts +++ b/packages/tools/nexus-mcp/src/nexus-client.ts @@ -399,6 +399,7 @@ export class NexusClient { headers: { authorization: `Bearer ${context.dataToken}`, "content-type": "application/json", + "idempotency-key": operationId, "x-nexus-request-id": operationId, }, method: "POST", diff --git a/packages/tools/nexus-mcp/src/operation-store.ts b/packages/tools/nexus-mcp/src/operation-store.ts new file mode 100644 index 0000000..e8ab495 --- /dev/null +++ b/packages/tools/nexus-mcp/src/operation-store.ts @@ -0,0 +1,752 @@ +import { createHash, randomUUID } from "node:crypto"; +import { constants } from "node:fs"; +import { + chmod, + link, + lstat, + mkdir, + open, + readdir, + rename, + rm, +} from "node:fs/promises"; +import path from "node:path"; + +import type { GenerationArtifact, ToolResult } from "./contracts.ts"; + +export const generationRecoveryResultText = + "Nexus image generation completed."; + +export interface StoredNexusGenerationRequest { + model: string; + outputModalities: readonly string[]; + prompt: string; +} + +export interface NexusStoredImage { + bytes: Uint8Array; + extension: "jpg" | "png" | "webp"; + mimeType: "image/jpeg" | "image/png" | "image/webp"; +} + +interface NexusOperationBase { + createdAt: string; + fingerprint: string; + operationId: string; + request: StoredNexusGenerationRequest; + requestDigest: string; + taskId: string; +} + +export interface PreparedNexusOperation extends NexusOperationBase { + status: "prepared"; +} + +export interface RunningNexusOperation extends NexusOperationBase { + status: "running"; +} + +export interface SucceededNexusOperation extends NexusOperationBase { + artifacts: readonly StoredNexusArtifact[]; + resultDigest: string; + status: "succeeded"; +} + +export interface FailedNexusOperation extends NexusOperationBase { + error: { code: string; message: string }; + status: "failed"; +} + +export interface CancelledNexusOperation extends NexusOperationBase { + status: "cancelled"; +} + +export type NexusOperationRecord = + | PreparedNexusOperation + | RunningNexusOperation + | SucceededNexusOperation + | FailedNexusOperation + | CancelledNexusOperation; + +export interface StoredNexusArtifact extends GenerationArtifact { + sha256: string; + size: number; +} + +export interface NexusOperationStoreOptions { + maximumOperations?: number; +} + +const operationSchema = "nexus.image-operation/2"; +const bindingSchema = "nexus.image-operation-binding/1"; +const maximumOperationsDefault = 4_096; +const maximumRecordBytes = 128 * 1024; +const maximumArtifactBytes = 12 * 1024 * 1024; +const maximumTotalArtifactBytes = 32 * 1024 * 1024; +const digestPattern = /^[a-f0-9]{64}$/u; +const operationIdPattern = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/u; +const taskIdPattern = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,511}$/u; +const errorCodePattern = /^[a-z][a-z0-9._-]{0,63}$/u; + +function noFollowFlag() { + const value = constants.O_NOFOLLOW as number | undefined; + if (!Number.isSafeInteger(value) || value === 0) { + throw new Error("Secure no-follow file access is unavailable"); + } + return value as number; +} + +function exclusiveWriteFlags() { + return constants.O_CREAT | constants.O_EXCL | constants.O_WRONLY | noFollowFlag(); +} + +function isCurrentUserOwner(uid: number) { + return typeof process.getuid !== "function" || uid === process.getuid(); +} + +function isRecord(value: unknown): value is Record { + return Boolean(value) && typeof value === "object" && !Array.isArray(value); +} + +function stableJson(value: unknown): string { + if (Array.isArray(value)) return `[${value.map(stableJson).join(",")}]`; + if (isRecord(value)) { + return `{${Object.keys(value) + .filter((key) => value[key] !== undefined) + .sort() + .map((key) => `${JSON.stringify(key)}:${stableJson(value[key])}`) + .join(",")}}`; + } + return JSON.stringify(value) ?? "null"; +} + +function resultForArtifacts(artifacts: readonly GenerationArtifact[]): ToolResult { + return { + content: [{ text: generationRecoveryResultText, type: "text" }], + structuredContent: { + artifacts: artifacts.map((artifact) => ({ ...artifact })), + }, + }; +} + +export function nexusGenerationResult( + artifacts: readonly GenerationArtifact[], +): ToolResult { + return resultForArtifacts(artifacts); +} + +function resultDigest(artifacts: readonly StoredNexusArtifact[]) { + return createHash("sha256") + .update( + stableJson({ + artifacts: artifacts.map((artifact) => ({ + file: { + relativePath: artifact.path, + sha256: artifact.sha256, + size: artifact.size, + }, + mimeType: artifact.mimeType, + name: artifact.name, + path: artifact.path, + })), + content: [ + { + text: generationRecoveryResultText, + type: "text", + }, + ], + }), + ) + .digest("hex"); +} + +export function nexusGenerationRequestFingerprint( + request: StoredNexusGenerationRequest, +) { + return createHash("sha256").update(stableJson(request)).digest("hex"); +} + +export class NexusOperationStore { + readonly #maximumOperations: number; + + constructor( + readonly directory: string, + options: NexusOperationStoreOptions = {}, + ) { + if (!path.isAbsolute(directory)) { + throw new Error("Nexus operation state directory must be absolute"); + } + this.#maximumOperations = + options.maximumOperations ?? maximumOperationsDefault; + if ( + !Number.isSafeInteger(this.#maximumOperations) || + this.#maximumOperations < 1 + ) { + throw new Error("Nexus operation store limit is invalid"); + } + } + + async binding(): Promise { + await this.#ensurePrivateDirectory(this.directory); + const bindingPath = path.join(this.directory, ".binding.json"); + try { + const value = await this.#readBoundedJson(bindingPath, 4_096); + if ( + !isRecord(value) || + value.schema !== bindingSchema || + typeof value.binding !== "string" || + !/^nexus-image-[a-f0-9]{32}$/u.test(value.binding) + ) { + throw new Error("Nexus recovery binding is invalid"); + } + return value.binding; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + } + const binding = `nexus-image-${randomUUID().replaceAll("-", "")}`; + try { + await this.#writeJsonExclusive(bindingPath, { + binding, + schema: bindingSchema, + }); + return binding; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; + const value = await this.#readBoundedJson(bindingPath, 4_096); + if ( + !isRecord(value) || + value.schema !== bindingSchema || + typeof value.binding !== "string" || + !/^nexus-image-[a-f0-9]{32}$/u.test(value.binding) + ) { + throw new Error("Nexus recovery binding is invalid"); + } + return value.binding; + } + } + + async find(operationId: string): Promise { + this.#requireOperationId(operationId); + await this.#ensurePrivateDirectory(this.directory); + try { + return parseRecordEnvelope( + await this.#readBoundedJson( + this.#recordPath(operationId), + maximumRecordBytes, + ), + operationId, + ); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return null; + throw error; + } + } + + async createPrepared(input: { + fingerprint: string; + operationId: string; + request: StoredNexusGenerationRequest; + requestDigest: string; + }): Promise { + this.#requireOperationId(input.operationId); + if (!digestPattern.test(input.fingerprint) || !digestPattern.test(input.requestDigest)) { + throw new Error("Nexus generation digest is invalid"); + } + await this.#ensurePrivateDirectory(this.directory); + if ((await this.#recordCount()) >= this.#maximumOperations) { + throw new Error("Nexus operation history is full"); + } + const record: PreparedNexusOperation = { + createdAt: new Date().toISOString(), + fingerprint: input.fingerprint, + operationId: input.operationId, + request: cloneRequest(input.request), + requestDigest: input.requestDigest, + status: "prepared", + taskId: `nexus_task_${randomUUID().replaceAll("-", "")}`, + }; + await this.#writeRecord(input.operationId, record, false); + return record; + } + + async markRunning( + operationId: string, + expected: PreparedNexusOperation | RunningNexusOperation, + ): Promise { + if (expected.status === "running") return expected; + const record: RunningNexusOperation = { ...expected, status: "running" }; + await this.#replaceMatching(operationId, expected, record); + return record; + } + + async succeed( + operationId: string, + expected: RunningNexusOperation, + images: readonly NexusStoredImage[], + ): Promise { + if (images.length < 1 || images.length > 8) { + throw new Error("Nexus generation artifact count is invalid"); + } + const total = images.reduce((sum, image) => sum + image.bytes.byteLength, 0); + if (total > maximumTotalArtifactBytes) { + throw new Error("Nexus generation artifacts are too large"); + } + const artifactDirectory = this.#artifactDirectory(operationId); + await rm(artifactDirectory, { force: true, recursive: true }); + await this.#ensurePrivateDirectory(artifactDirectory); + const artifacts: StoredNexusArtifact[] = []; + try { + for (let index = 0; index < images.length; index += 1) { + const image = images[index]!; + if (image.bytes.byteLength < 1 || image.bytes.byteLength > maximumArtifactBytes) { + throw new Error("Nexus generation artifact is too large"); + } + const name = `nexus-${this.#safeStem(operationId)}-${index + 1}.${image.extension}`; + const artifactPath = path.join(artifactDirectory, name); + const handle = await open(artifactPath, exclusiveWriteFlags(), 0o600); + try { + await handle.chmod(0o600); + await handle.writeFile(image.bytes); + await handle.sync(); + } finally { + await handle.close(); + } + artifacts.push({ + mimeType: image.mimeType, + name, + path: name, + sha256: createHash("sha256").update(image.bytes).digest("hex"), + size: image.bytes.byteLength, + }); + } + await this.#syncDirectory(artifactDirectory); + const record: SucceededNexusOperation = { + ...expected, + artifacts, + resultDigest: resultDigest(artifacts), + status: "succeeded", + }; + await this.#replaceMatching(operationId, expected, record); + return record; + } catch (error) { + await rm(artifactDirectory, { force: true, recursive: true }).catch( + () => undefined, + ); + throw error; + } + } + + async fail( + operationId: string, + expected: PreparedNexusOperation | RunningNexusOperation, + error: { code: string; message: string }, + ): Promise { + if ( + !errorCodePattern.test(error.code) || + !error.message || + error.message.length > 200 || + error.message !== error.message.trim() || + /[\u0000-\u001f\u007f]/u.test(error.message) + ) { + throw new Error("Nexus generation failure is invalid"); + } + const record: FailedNexusOperation = { + ...expected, + error: { ...error }, + status: "failed", + }; + await this.#replaceMatching(operationId, expected, record); + return record; + } + + async cancel( + operationId: string, + expected: PreparedNexusOperation | RunningNexusOperation, + ): Promise { + const record: CancelledNexusOperation = { + ...expected, + status: "cancelled", + }; + await this.#replaceMatching(operationId, expected, record); + return record; + } + + async materialize( + operationId: string, + record: SucceededNexusOperation, + outputDirectory: string, + ): Promise { + if (!path.isAbsolute(outputDirectory)) { + throw new Error("Nexus generation output directory must be absolute"); + } + const outputInfo = await lstat(outputDirectory); + if (!outputInfo.isDirectory() || outputInfo.isSymbolicLink()) { + throw new Error("Nexus generation output directory is invalid"); + } + const artifacts: GenerationArtifact[] = []; + for (const artifact of record.artifacts) { + const sourcePath = path.join( + this.#artifactDirectory(operationId), + artifact.name, + ); + const source = await open( + sourcePath, + constants.O_RDONLY | noFollowFlag(), + ); + try { + const sourceInfo = await source.stat(); + if ( + !sourceInfo.isFile() || + sourceInfo.size !== artifact.size || + (sourceInfo.mode & 0o077) !== 0 || + !isCurrentUserOwner(sourceInfo.uid) + ) { + throw new Error("Nexus stored generation artifact changed"); + } + const bytes = new Uint8Array(await source.readFile()); + if ( + createHash("sha256").update(bytes).digest("hex") !== artifact.sha256 + ) { + throw new Error("Nexus stored generation artifact changed"); + } + const targetPath = path.join(outputDirectory, artifact.name); + try { + const target = await open(targetPath, exclusiveWriteFlags(), 0o600); + try { + await target.chmod(0o600); + await target.writeFile(bytes); + await target.sync(); + } finally { + await target.close(); + } + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; + const existing = await open( + targetPath, + constants.O_RDONLY | noFollowFlag(), + ); + try { + const existingBytes = new Uint8Array(await existing.readFile()); + if ( + createHash("sha256").update(existingBytes).digest("hex") !== + artifact.sha256 + ) { + throw new Error("Nexus generation output already exists"); + } + } finally { + await existing.close(); + } + } + } finally { + await source.close(); + } + artifacts.push({ + mimeType: artifact.mimeType, + name: artifact.name, + path: artifact.path, + }); + } + return artifacts; + } + + async acknowledge(operationId: string, expected: NexusOperationRecord) { + const current = await this.find(operationId); + if (!current || !sameRecordIdentity(current, expected)) { + throw new Error("Nexus generation operation changed before acknowledgement"); + } + await rm(this.#recordPath(operationId)); + await rm(this.#artifactDirectory(operationId), { + force: true, + recursive: true, + }); + await this.#syncDirectory(this.directory); + } + + async #replaceMatching( + operationId: string, + expected: NexusOperationRecord, + next: NexusOperationRecord, + ) { + const current = await this.find(operationId); + if (!current || !sameRecordIdentity(current, expected)) { + throw new Error("Nexus generation operation changed"); + } + await this.#writeRecord(operationId, next, true); + } + + async #recordCount() { + return (await readdir(this.directory, { withFileTypes: true })).filter( + (entry) => entry.isFile() && entry.name.endsWith(".json") && entry.name !== ".binding.json", + ).length; + } + + #operationDigest(operationId: string) { + return createHash("sha256").update(operationId).digest("hex"); + } + + #recordPath(operationId: string) { + return path.join(this.directory, `${this.#operationDigest(operationId)}.json`); + } + + #artifactDirectory(operationId: string) { + return path.join(this.directory, "artifacts", this.#operationDigest(operationId)); + } + + #safeStem(operationId: string) { + const readable = operationId + .toLowerCase() + .replaceAll(/[^a-z0-9_-]/gu, "-") + .replaceAll(/-+/gu, "-") + .replaceAll(/^-|-$/gu, "") + .slice(0, 24); + return `${readable || "operation"}-${this.#operationDigest(operationId).slice(0, 12)}`; + } + + #requireOperationId(operationId: string) { + if (!operationIdPattern.test(operationId)) { + throw new Error("Nexus generation operation id is invalid"); + } + } + + async #writeRecord( + operationId: string, + record: NexusOperationRecord, + replace: boolean, + ) { + const normalized = parseRecordEnvelope( + { operationId, record, schema: operationSchema }, + operationId, + ); + const targetPath = this.#recordPath(operationId); + const temporaryPath = path.join( + this.directory, + `.${this.#operationDigest(operationId)}-${randomUUID()}.tmp`, + ); + await this.#writeJsonExclusive(temporaryPath, { + operationId, + record: normalized, + schema: operationSchema, + }); + try { + if (replace) { + await rename(temporaryPath, targetPath); + } else { + await link(temporaryPath, targetPath); + await rm(temporaryPath); + } + await chmod(targetPath, 0o600); + await this.#syncDirectory(this.directory); + } finally { + await rm(temporaryPath, { force: true }).catch(() => undefined); + } + } + + async #writeJsonExclusive(filePath: string, value: unknown) { + const serialized = `${JSON.stringify(value)}\n`; + if (Buffer.byteLength(serialized, "utf8") > maximumRecordBytes) { + throw new Error("Nexus operation record is too large"); + } + const handle = await open(filePath, exclusiveWriteFlags(), 0o600); + try { + await handle.chmod(0o600); + await handle.writeFile(serialized, "utf8"); + await handle.sync(); + } finally { + await handle.close(); + } + } + + async #readBoundedJson(filePath: string, maximumBytes: number) { + const handle = await open(filePath, constants.O_RDONLY | noFollowFlag()); + try { + const info = await handle.stat(); + if ( + !info.isFile() || + info.size < 1 || + info.size > maximumBytes || + (info.mode & 0o077) !== 0 || + !isCurrentUserOwner(info.uid) + ) { + throw new Error("Nexus operation record is not private and bounded"); + } + return JSON.parse(await handle.readFile("utf8")) as unknown; + } finally { + await handle.close(); + } + } + + async #ensurePrivateDirectory(directory: string) { + await mkdir(directory, { mode: 0o700, recursive: true }); + const info = await lstat(directory); + if ( + !info.isDirectory() || + info.isSymbolicLink() || + !isCurrentUserOwner(info.uid) + ) { + throw new Error("Nexus operation state directory is not host-owned"); + } + await chmod(directory, 0o700); + } + + async #syncDirectory(directory: string) { + if (process.platform === "win32") return; + const handle = await open(directory, "r"); + try { + await handle.sync(); + } finally { + await handle.close(); + } + } +} + +function cloneRequest( + request: StoredNexusGenerationRequest, +): StoredNexusGenerationRequest { + if ( + typeof request.model !== "string" || + !request.model || + request.model.length > 191 || + typeof request.prompt !== "string" || + request.prompt.length > 20_000 || + !Array.isArray(request.outputModalities) || + request.outputModalities.length < 1 || + request.outputModalities.length > 8 || + request.outputModalities.some( + (value) => + typeof value !== "string" || + !value || + value.length > 32 || + /[\u0000-\u001f\u007f]/u.test(value), + ) + ) { + throw new Error("Nexus stored generation request is invalid"); + } + return { + model: request.model, + outputModalities: [...request.outputModalities], + prompt: request.prompt, + }; +} + +function parseRecordEnvelope( + value: unknown, + operationId: string, +): NexusOperationRecord { + if (!isRecord(value)) throw new Error("Nexus operation record is invalid"); + if ( + value.schema !== operationSchema || + value.operationId !== operationId || + !isRecord(value.record) + ) { + throw new Error("Nexus operation record is invalid"); + } + const input = value.record; + if ( + typeof input.createdAt !== "string" || + !Number.isFinite(Date.parse(input.createdAt)) || + typeof input.fingerprint !== "string" || + !digestPattern.test(input.fingerprint) || + typeof input.requestDigest !== "string" || + !digestPattern.test(input.requestDigest) || + typeof input.taskId !== "string" || + !taskIdPattern.test(input.taskId) + ) { + throw new Error("Nexus operation record is invalid"); + } + const base: NexusOperationBase = { + createdAt: input.createdAt, + fingerprint: input.fingerprint, + operationId, + request: cloneRequest(input.request as StoredNexusGenerationRequest), + requestDigest: input.requestDigest, + taskId: input.taskId, + }; + if (input.status === "prepared" || input.status === "running") { + return { ...base, status: input.status }; + } + if (input.status === "cancelled") return { ...base, status: "cancelled" }; + if (input.status === "failed") { + if ( + !isRecord(input.error) || + typeof input.error.code !== "string" || + !errorCodePattern.test(input.error.code) || + typeof input.error.message !== "string" || + !input.error.message || + input.error.message.length > 200 || + input.error.message !== input.error.message.trim() || + /[\u0000-\u001f\u007f]/u.test(input.error.message) + ) { + throw new Error("Nexus operation failure record is invalid"); + } + return { + ...base, + error: { code: input.error.code, message: input.error.message }, + status: "failed", + }; + } + if ( + input.status !== "succeeded" || + typeof input.resultDigest !== "string" || + !digestPattern.test(input.resultDigest) || + !Array.isArray(input.artifacts) || + input.artifacts.length < 1 || + input.artifacts.length > 8 + ) { + throw new Error("Nexus operation record is invalid"); + } + const artifacts = input.artifacts.map(parseArtifact); + if (resultDigest(artifacts) !== input.resultDigest) { + throw new Error("Nexus operation result digest is invalid"); + } + return { + ...base, + artifacts, + resultDigest: input.resultDigest, + status: "succeeded", + }; +} + +function parseArtifact(value: unknown): StoredNexusArtifact { + if (!isRecord(value)) throw new Error("Nexus operation artifact is invalid"); + if ( + !["image/jpeg", "image/png", "image/webp"].includes( + value.mimeType as string, + ) || + typeof value.name !== "string" || + !value.name || + value.name.length > 255 || + value.path !== value.name || + path.basename(value.name) !== value.name || + typeof value.sha256 !== "string" || + !digestPattern.test(value.sha256) || + !Number.isSafeInteger(value.size) || + (value.size as number) < 1 || + (value.size as number) > maximumArtifactBytes + ) { + throw new Error("Nexus operation artifact is invalid"); + } + return { + mimeType: value.mimeType as string, + name: value.name, + path: value.name, + sha256: value.sha256, + size: value.size as number, + }; +} + +function sameRecordIdentity( + left: NexusOperationRecord, + right: NexusOperationRecord, +) { + return ( + left.createdAt === right.createdAt && + left.fingerprint === right.fingerprint && + left.operationId === right.operationId && + left.requestDigest === right.requestDigest && + left.taskId === right.taskId && + left.status === right.status && + (left.status !== "succeeded" || + (right.status === "succeeded" && + left.resultDigest === right.resultDigest)) + ); +} diff --git a/packages/tools/nexus-mcp/src/plugin-service.ts b/packages/tools/nexus-mcp/src/plugin-service.ts index aa47812..219c975 100644 --- a/packages/tools/nexus-mcp/src/plugin-service.ts +++ b/packages/tools/nexus-mcp/src/plugin-service.ts @@ -30,6 +30,12 @@ function disconnected( } export class NexusPluginService { + readonly #checkoutFlights = new Map< + string, + Promise + >(); + #checkoutBarrier: Promise = Promise.resolve(); + constructor( private readonly authorization: NexusAuthorization, private readonly client: NexusClient, @@ -129,19 +135,13 @@ export class NexusPluginService { unit: "USD", }, }; - } catch { - return configured - ? { - account: unavailable, - billing: unavailable, - credential: { configured: true, verification: "failed" }, - credits: unavailable, - plan: unavailable, - schema: pluginServiceStatusSchema, - state: "attention", - usage: unavailable, - } - : disconnected("failed"); + } catch (error) { + throw new Error( + configured + ? "Nexus service status could not be verified" + : "Nexus credential state could not be read", + { cause: error }, + ); } } @@ -165,8 +165,11 @@ export class NexusPluginService { async signOut() { this.authorization.cancel(); - await this.client.signOut(); - await this.checkouts.clear(); + try { + await this.client.signOut(); + } finally { + await this.checkouts.clear(); + } return disconnected(); } @@ -179,21 +182,48 @@ export class NexusPluginService { ) { throw new Error("The selected Nexus Plan is not available"); } - const attempt = await this.checkouts.begin(access.access.id, planKey); - const checkout = await this.client.createCheckout( - planKey, - attempt.idempotencyKey, - ); - await this.checkouts.write({ - ...attempt, - checkoutId: checkout.checkoutId, - status: checkout.status, + const flightKey = `${access.access.id}\0${planKey}`; + const existing = this.#checkoutFlights.get(flightKey); + if (existing) return existing; + const flight = this.#serializeCheckout(async () => { + const attempt = await this.checkouts.begin(access.access.id, planKey); + const checkout = await this.client.createCheckout( + planKey, + attempt.idempotencyKey, + ); + await this.checkouts.write({ + ...attempt, + checkoutId: checkout.checkoutId, + status: checkout.status, + }); + return { + checkout_id: checkout.checkoutId, + checkout_url: checkout.browserUrl, + schema: pluginServiceCheckoutSchema, + }; }); - return { - checkout_id: checkout.checkoutId, - checkout_url: checkout.browserUrl, - schema: pluginServiceCheckoutSchema, - }; + this.#checkoutFlights.set(flightKey, flight); + try { + return await flight; + } finally { + if (this.#checkoutFlights.get(flightKey) === flight) { + this.#checkoutFlights.delete(flightKey); + } + } + } + + async #serializeCheckout(operation: () => Promise): Promise { + const previous = this.#checkoutBarrier; + let release!: () => void; + this.#checkoutBarrier = new Promise((resolve) => { + release = resolve; + }); + await previous; + try { + return await operation(); + } finally { + release(); + } } async #pending(workspaceAccessId: string, currentPlanKey: string) { @@ -224,14 +254,10 @@ export class NexusPluginService { async #refreshCheckout( attempt: NexusCheckoutAttempt, ): Promise { - try { - const status = await this.client.checkoutStatus(attempt.checkoutId!); - const updated = { ...attempt, status: status.status }; - await this.checkouts.write(updated); - return updated; - } catch { - return attempt; - } + const status = await this.client.checkoutStatus(attempt.checkoutId!); + const updated = { ...attempt, status: status.status }; + await this.checkouts.write(updated); + return updated; } } diff --git a/packages/tools/nexus-mcp/test/image-generator.test.ts b/packages/tools/nexus-mcp/test/image-generator.test.ts index 63876f3..f73f7d3 100644 --- a/packages/tools/nexus-mcp/test/image-generator.test.ts +++ b/packages/tools/nexus-mcp/test/image-generator.test.ts @@ -4,10 +4,33 @@ import os from "node:os"; import path from "node:path"; import { NexusImageGenerator } from "../src/image-generator.ts"; +import { + nexusGenerationRequestFingerprint, + NexusOperationStore, +} from "../src/operation-store.ts"; const roots: string[] = []; const png = Buffer.from([137, 80, 78, 71, 13, 10, 26, 10, 0, 0, 0, 0]); +function imageResponse() { + return { + choices: [ + { + message: { + images: [ + { + image_url: { + url: `data:image/png;base64,${png.toString("base64")}`, + }, + type: "image_url", + }, + ], + }, + }, + ], + }; +} + afterEach(async () => { await Promise.all( roots @@ -22,8 +45,15 @@ describe("NexusImageGenerator", () => { path.join(os.tmpdir(), "convax-nexus-image-"), ); roots.push(root); + const output = path.join(root, "output"); + await fs.mkdir(output); let completions = 0; - const generator = new NexusImageGenerator(); + const generator = new NexusImageGenerator( + new NexusOperationStore(path.join(root, "operations")), + async () => { + throw new Error("A completed operation must not resume upstream"); + }, + ); const route = { isCurrent: () => true, maximumAgeMs: 60_000, @@ -50,29 +80,14 @@ describe("NexusImageGenerator", () => { }); expect(operationId).toBe("operation-123"); completions += 1; - return { - choices: [ - { - message: { - images: [ - { - image_url: { - url: `data:image/png;base64,${png.toString("base64")}`, - }, - type: "image_url", - }, - ], - }, - }, - ], - }; + return imageResponse(); }, }; const input = { model: "openai/gpt-image-1", operation_id: "operation-123", output: "image", - output_directory: root, + output_directory: output, prompt: "Draw a small red circle.", references: [], schema: "convax.generation-call/1", @@ -80,15 +95,19 @@ describe("NexusImageGenerator", () => { const first = await generator.generate( input, - () => route, + { operationId: "operation-123", requestDigest: "a".repeat(64) }, + async () => route, new AbortController().signal, + () => undefined, ); const second = await generator.generate( input, - () => { + { operationId: "operation-123", requestDigest: "a".repeat(64) }, + async () => { throw new Error("A completed operation must not resolve a new route"); }, new AbortController().signal, + () => undefined, ); expect(completions).toBe(1); @@ -96,11 +115,15 @@ describe("NexusImageGenerator", () => { expect(first).toHaveLength(1); expect(first[0]).toMatchObject({ mimeType: "image/png", - name: "nexus-operation-123-1.png", - path: "nexus-operation-123-1.png", }); + expect(first[0]!.name).toMatch( + /^nexus-operation-123-[a-f0-9]{12}-1\.png$/u, + ); + expect(first[0]!.path).toMatch( + /^nexus-operation-123-[a-f0-9]{12}-1\.png$/u, + ); expect(path.isAbsolute(first[0]!.path)).toBe(false); - expect(await fs.readFile(path.join(root, first[0]!.path))).toEqual(png); + expect(await fs.readFile(path.join(output, first[0]!.path))).toEqual(png); }); test("rejects a model that is no longer in the image-output catalog", async () => { @@ -108,8 +131,13 @@ describe("NexusImageGenerator", () => { path.join(os.tmpdir(), "convax-nexus-image-"), ); roots.push(root); + const output = path.join(root, "output"); + await fs.mkdir(output); let completions = 0; - const generator = new NexusImageGenerator(); + const generator = new NexusImageGenerator( + new NexusOperationStore(path.join(root, "operations")), + async () => ({}), + ); const route = { isCurrent: () => true, maximumAgeMs: 60_000, @@ -126,15 +154,167 @@ describe("NexusImageGenerator", () => { model: "openai/gpt-image-1", operation_id: "operation-456", output: "image", - output_directory: root, + output_directory: output, prompt: "Draw a circle.", references: [], schema: "convax.generation-call/1", }, - () => route, + { operationId: "operation-456", requestDigest: "b".repeat(64) }, + async () => route, new AbortController().signal, + () => undefined, ), ).rejects.toThrow("unavailable"); expect(completions).toBe(0); }); + + test("reattaches a durable running operation after restart and never resubmits a completed operation", async () => { + const root = await fs.mkdtemp( + path.join(os.tmpdir(), "convax-nexus-image-restart-"), + ); + roots.push(root); + const directory = path.join(root, "operations"); + const output = path.join(root, "output"); + const replayOutput = path.join(root, "replay-output"); + await Promise.all([fs.mkdir(output), fs.mkdir(replayOutput)]); + const request = { + model: "openai/gpt-image-1", + outputModalities: ["image", "text"], + prompt: "Draw a durable circle.", + } as const; + const firstStore = new NexusOperationStore(directory); + const prepared = await firstStore.createPrepared({ + fingerprint: nexusGenerationRequestFingerprint(request), + operationId: "operation-restart-123", + request, + requestDigest: "c".repeat(64), + }); + const running = await firstStore.markRunning(prepared.operationId, prepared); + const call = { + model: request.model, + operation_id: prepared.operationId, + output: "image", + output_directory: output, + prompt: request.prompt, + references: [], + schema: "convax.generation-call/1", + }; + let resumeCalls = 0; + const submitted: string[] = []; + const restarted = new NexusImageGenerator( + new NexusOperationStore(directory), + async (persisted, operationId) => { + resumeCalls += 1; + expect(persisted).toEqual(request); + expect(operationId).toBe(prepared.operationId); + return imageResponse(); + }, + ); + const artifacts = await restarted.generate( + call, + { + operationId: prepared.operationId, + requestDigest: prepared.requestDigest, + }, + async () => { + throw new Error("A durable operation must not resolve a new route"); + }, + new AbortController().signal, + (taskId) => submitted.push(taskId), + ); + expect(resumeCalls).toBe(1); + expect(submitted).toEqual([running.taskId]); + expect(await fs.readFile(path.join(output, artifacts[0]!.path))).toEqual( + png, + ); + + const completedProcess = new NexusImageGenerator( + new NexusOperationStore(directory), + async () => { + throw new Error("A completed operation must not contact Nexus"); + }, + ); + const replayed = await completedProcess.generate( + { ...call, output_directory: replayOutput }, + { + operationId: prepared.operationId, + requestDigest: prepared.requestDigest, + }, + async () => { + throw new Error("A completed operation must not resolve a route"); + }, + new AbortController().signal, + () => undefined, + ); + expect(replayed).toEqual(artifacts); + expect(resumeCalls).toBe(1); + expect( + await fs.readFile(path.join(replayOutput, replayed[0]!.path)), + ).toEqual(png); + }); + + test("leaves an operation running during process shutdown so the next process can resume it", async () => { + const root = await fs.mkdtemp( + path.join(os.tmpdir(), "convax-nexus-image-shutdown-"), + ); + roots.push(root); + const directory = path.join(root, "operations"); + const output = path.join(root, "output"); + await fs.mkdir(output); + let submitted!: () => void; + const didSubmit = new Promise((resolve) => { + submitted = resolve; + }); + const controller = new AbortController(); + const generator = new NexusImageGenerator( + new NexusOperationStore(directory), + async () => { + throw new Error("A new operation must use its loaded route"); + }, + ); + const operation = generator.generate( + { + model: "openai/gpt-image-1", + operation_id: "operation-shutdown-123", + output: "image", + output_directory: output, + prompt: "Draw a circle through restart.", + references: [], + schema: "convax.generation-call/1", + }, + { + operationId: "operation-shutdown-123", + requestDigest: "d".repeat(64), + }, + async () => ({ + complete: async (_model, _prompt, _operationId, signal) => + new Promise((_, reject) => { + signal.addEventListener( + "abort", + () => reject(new DOMException("closed", "AbortError")), + { once: true }, + ); + }), + isCurrent: () => true, + maximumAgeMs: 60_000, + models: [ + { + id: "openai/gpt-image-1", + name: "GPT Image 1", + outputModalities: ["image", "text"], + }, + ], + }), + controller.signal, + () => submitted(), + ); + await didSubmit; + controller.abort("MCP server is closing"); + await expect(operation).rejects.toThrow("closed"); + + const persisted = await new NexusOperationStore(directory).find( + "operation-shutdown-123", + ); + expect(persisted?.status).toBe("running"); + }); }); diff --git a/packages/tools/nexus-mcp/test/llm-gateway.test.ts b/packages/tools/nexus-mcp/test/llm-gateway.test.ts index 4114aa6..43b02b9 100644 --- a/packages/tools/nexus-mcp/test/llm-gateway.test.ts +++ b/packages/tools/nexus-mcp/test/llm-gateway.test.ts @@ -25,16 +25,26 @@ describe("NexusLlmGateway", () => { authorization: request.headers.get("authorization"), body: await request.json(), }); - return Response.json({ - choices: [ - { - finish_reason: "stop", - index: 0, - message: { content: "Nexus connected", role: "assistant" }, + return Response.json( + { + choices: [ + { + finish_reason: "stop", + index: 0, + message: { content: "Nexus connected", role: "assistant" }, + }, + ], + model: testModelId, + }, + { + headers: { + "retry-after": "2", + "x-generation-id": "generation-123", + "x-request-id": "request-123", + "x-secret-debug": "must-not-cross-loopback", }, - ], - model: testModelId, - }); + }, + ); }, }); servers.push(upstream); @@ -60,6 +70,10 @@ describe("NexusLlmGateway", () => { }), }); expect(response.status).toBe(200); + expect(response.headers.get("retry-after")).toBe("2"); + expect(response.headers.get("x-generation-id")).toBe("generation-123"); + expect(response.headers.get("x-request-id")).toBe("request-123"); + expect(response.headers.get("x-secret-debug")).toBeNull(); expect(await response.json()).toMatchObject({ model: testModelId }); expect(upstreamRequests).toEqual([ { diff --git a/packages/tools/nexus-mcp/test/mcp-server.test.ts b/packages/tools/nexus-mcp/test/mcp-server.test.ts index fe36b1d..4c13de2 100644 --- a/packages/tools/nexus-mcp/test/mcp-server.test.ts +++ b/packages/tools/nexus-mcp/test/mcp-server.test.ts @@ -10,6 +10,10 @@ import { tools, } from "../src/mcp-server.ts"; import { NexusImageHttpError } from "../src/nexus-client.ts"; +import { + nexusGenerationRequestFingerprint, + NexusOperationStore, +} from "../src/operation-store.ts"; import type { NexusSessionStore } from "../src/session-store.ts"; const fixedToolNames = [ @@ -24,9 +28,24 @@ const fixedToolNames = [ "llm.gateway.start", ] as const; -async function listedToolNames( +function generationMeta( + operationId: string, + requestDigest = "a".repeat(64), +) { + return { + convaxGeneration: { + operationId, + recovery: "required", + requestDigest, + schema: "convax.generation-operation/1", + }, + progressToken: `progress-${operationId}`, + }; +} + +async function listedToolResponse( modelCatalogResponse: () => Response, -): Promise { +) { const responses: unknown[] = []; const sessions = { async read() { @@ -93,16 +112,279 @@ async function listedToolNames( await running; expect(responses).toHaveLength(1); - const response = responses[0] as { - result: { tools: Array<{ name: string }> }; - }; - return response.result.tools.map(({ name }) => name); + return responses[0]; } test("Nexus companion keeps only fixed Service and LLM tools without a live image catalog", () => { expect(tools.map(({ name }) => name)).toEqual([...fixedToolNames]); }); +test("initialize advertises one stable durable image-generation recovery binding", async () => { + const root = await fs.mkdtemp( + path.join(os.tmpdir(), "convax-nexus-mcp-initialize-"), + ); + const responses: unknown[] = []; + const operationStore = new NexusOperationStore( + path.join(root, "operations"), + ); + const server = new NexusMcpServer({ + operationStore, + send: (value) => responses.push(value), + }); + let controller!: ReadableStreamDefaultController; + const input = new ReadableStream({ + start(value) { + controller = value; + }, + }); + const running = server.run(input); + try { + controller.enqueue( + new TextEncoder().encode( + `${JSON.stringify({ + id: 1, + jsonrpc: "2.0", + method: "initialize", + params: { protocolVersion: "2025-03-26" }, + })}\n`, + ), + ); + for (let attempt = 0; attempt < 100 && responses.length < 1; attempt += 1) { + await Bun.sleep(10); + } + expect(responses[0]).toMatchObject({ + id: 1, + result: { + capabilities: { + experimental: { + "convax/generation-lro": { + mode: "long-running-operation", + schema: "convax.generation-lro/1", + }, + }, + }, + }, + }); + const advertisedBinding = ( + responses[0] as { + result: { + capabilities: { + experimental: { + "convax/generation-lro": { binding: string }; + }; + }; + }; + } + ).result.capabilities.experimental["convax/generation-lro"].binding; + expect(advertisedBinding).toMatch(/^nexus-image-[a-f0-9]{32}$/u); + expect(advertisedBinding).toBe(await operationStore.binding()); + } finally { + await server.shutdown(1_000); + await running; + await fs.rm(root, { force: true, recursive: true }); + } +}); + +test("generation LRO endpoints recover, materialize, and acknowledge a persisted result", async () => { + const root = await fs.mkdtemp( + path.join(os.tmpdir(), "convax-nexus-mcp-recovery-"), + ); + const output = path.join(root, "output"); + await fs.mkdir(output); + const operationStore = new NexusOperationStore( + path.join(root, "operations"), + ); + const request = { + model: "openai/gpt-image-1", + outputModalities: ["image", "text"], + prompt: "Draw a persisted circle.", + } as const; + const prepared = await operationStore.createPrepared({ + fingerprint: nexusGenerationRequestFingerprint(request), + operationId: "operation-mcp-recovery-123", + request, + requestDigest: "f".repeat(64), + }); + const runningRecord = await operationStore.markRunning( + prepared.operationId, + prepared, + ); + const png = Buffer.from([137, 80, 78, 71, 13, 10, 26, 10, 0, 0, 0, 0]); + const succeeded = await operationStore.succeed( + runningRecord.operationId, + runningRecord, + [{ bytes: png, extension: "png", mimeType: "image/png" }], + ); + const responses: unknown[] = []; + const server = new NexusMcpServer({ + operationStore, + send: (value) => responses.push(value), + }); + let controller!: ReadableStreamDefaultController; + const input = new ReadableStream({ + start(value) { + controller = value; + }, + }); + const running = server.run(input); + const base = { + operationId: succeeded.operationId, + requestDigest: succeeded.requestDigest, + schema: "convax.generation-lro-request/1", + taskId: succeeded.taskId, + }; + try { + controller.enqueue( + new TextEncoder().encode( + `${JSON.stringify({ + id: 1, + jsonrpc: "2.0", + method: "convax/generation/operations/get", + params: base, + })}\n`, + ), + ); + for (let attempt = 0; attempt < 100 && responses.length < 1; attempt += 1) { + await Bun.sleep(10); + } + expect(responses[0]).toMatchObject({ + id: 1, + result: { + resultDigest: succeeded.resultDigest, + schema: "convax.generation-lro-snapshot/1", + status: "succeeded", + taskId: succeeded.taskId, + }, + }); + + controller.enqueue( + new TextEncoder().encode( + `${JSON.stringify({ + id: 2, + jsonrpc: "2.0", + method: "convax/generation/operations/result", + params: { + ...base, + outputDirectory: output, + resultDigest: succeeded.resultDigest, + }, + })}\n`, + ), + ); + for (let attempt = 0; attempt < 100 && responses.length < 2; attempt += 1) { + await Bun.sleep(10); + } + expect(responses[1]).toMatchObject({ + id: 2, + result: { + result: { + structuredContent: { + artifacts: [ + { + mimeType: "image/png", + path: succeeded.artifacts[0]!.path, + }, + ], + }, + }, + resultDigest: succeeded.resultDigest, + schema: "convax.generation-lro-result/1", + }, + }); + expect( + await fs.readFile(path.join(output, succeeded.artifacts[0]!.path)), + ).toEqual(png); + + controller.enqueue( + new TextEncoder().encode( + `${JSON.stringify({ + id: 3, + jsonrpc: "2.0", + method: "convax/generation/operations/acknowledge", + params: { ...base, resultDigest: succeeded.resultDigest }, + })}\n`, + ), + ); + for (let attempt = 0; attempt < 100 && responses.length < 3; attempt += 1) { + await Bun.sleep(10); + } + expect(responses[2]).toMatchObject({ + id: 3, + result: { + acknowledged: true, + schema: "convax.generation-lro-acknowledgement/1", + }, + }); + expect(await operationStore.find(succeeded.operationId)).toBeNull(); + } finally { + await server.shutdown(1_000); + await running; + await fs.rm(root, { force: true, recursive: true }); + } +}); + +test("one rejected recovery handler does not stop later service requests", async () => { + const responses: unknown[] = []; + const server = new NexusMcpServer({ + send: (value) => responses.push(value), + sessions: { + async read() { + return null; + }, + async write() {}, + } as unknown as NexusSessionStore, + }); + let controller!: ReadableStreamDefaultController; + const input = new ReadableStream({ + start(value) { + controller = value; + }, + }); + const running = server.run(input); + controller.enqueue( + new TextEncoder().encode( + `${JSON.stringify({ + id: 1, + jsonrpc: "2.0", + method: "convax/generation/operations/get", + params: {}, + })}\n`, + ), + ); + controller.enqueue( + new TextEncoder().encode( + `${JSON.stringify({ + id: 2, + jsonrpc: "2.0", + method: "tools/call", + params: { arguments: {}, name: "service.status" }, + })}\n`, + ), + ); + for (let attempt = 0; attempt < 100 && responses.length < 2; attempt += 1) { + await Bun.sleep(10); + } + expect( + responses.find((response) => (response as { id?: unknown }).id === 1), + ).toMatchObject({ + error: { code: -32_603, message: "Internal error" }, + id: 1, + }); + expect( + responses.find((response) => (response as { id?: unknown }).id === 2), + ).toMatchObject({ + id: 2, + result: { + structuredContent: { + credential: { configured: false }, + state: "disconnected", + }, + }, + }); + await server.shutdown(1_000); + await running; +}); + test("Nexus image generation projects current image models as a bounded select", () => { const tool = imageGenerationTool([ { id: "microsoft/mai-image-2.5-pro", name: "MAI Image 2.5 Pro" }, @@ -136,15 +418,22 @@ test("Nexus image generation rejects unbounded model catalogs", () => { ).toThrow("Nexus image model catalog is outside the bounded choice limit"); }); -test("tools/list hides image generation when the image catalog request fails", async () => { +test("tools/list keeps fixed service tools when the image catalog request fails", async () => { expect( - await listedToolNames(() => new Response("Unavailable", { status: 503 })), - ).toEqual(fixedToolNames); + await listedToolResponse( + () => new Response("Unavailable", { status: 503 }), + ), + ).toMatchObject({ + id: 1, + result: { + tools: fixedToolNames.map((name) => ({ name })), + }, + }); }); -test("tools/list hides image generation when the live catalog has no image models", async () => { +test("tools/list keeps fixed service tools when the live catalog has no usable image models", async () => { expect( - await listedToolNames(() => + await listedToolResponse(() => Response.json({ data: [ { @@ -155,12 +444,17 @@ test("tools/list hides image generation when the live catalog has no image model ], }), ), - ).toEqual(fixedToolNames); + ).toMatchObject({ + id: 1, + result: { + tools: fixedToolNames.map((name) => ({ name })), + }, + }); }); -test("tools/list hides image generation when the live image catalog exceeds 64 choices", async () => { +test("tools/list keeps fixed service tools when the live image catalog exceeds 64 choices", async () => { expect( - await listedToolNames(() => + await listedToolResponse(() => Response.json({ data: Array.from({ length: 65 }, (_, index) => ({ architecture: { output_modalities: ["image", "text"] }, @@ -169,7 +463,12 @@ test("tools/list hides image generation when the live image catalog exceeds 64 c })), }), ), - ).toEqual(fixedToolNames); + ).toMatchObject({ + id: 1, + result: { + tools: fixedToolNames.map((name) => ({ name })), + }, + }); }); test("Nexus image diagnostics expose only typed bounded HTTP fields", () => { @@ -187,9 +486,7 @@ test("Nexus image diagnostics expose only typed bounded HTTP fields", () => { const generic = publicImageGenerationErrorMessage( new Error("raw secret-token secret-prompt /private/output/path"), ); - expect(generic).toBe( - "Nexus image generation failed. Check Nexus before retrying because the upstream task result may be unknown.", - ); + expect(generic).toBe("Nexus image generation failed."); expect(generic).not.toContain("secret-token"); expect(generic).not.toContain("secret-prompt"); expect(generic).not.toContain("/private/output/path"); @@ -340,6 +637,9 @@ test("MCP excludes automatic routers and image-only routes from live generation test("image.generate never refreshes a missing route inside tools/call", async () => { const responses: unknown[] = []; let networkRequests = 0; + const operationDirectory = await fs.mkdtemp( + path.join(os.tmpdir(), "convax-nexus-mcp-no-route-"), + ); const server = new NexusMcpServer({ client: { fetch: async () => { @@ -347,6 +647,7 @@ test("image.generate never refreshes a missing route inside tools/call", async ( throw new Error("Unexpected network request"); }, }, + operationStore: new NexusOperationStore(operationDirectory), send: (value) => responses.push(value), sessions: { async read() { @@ -369,6 +670,7 @@ test("image.generate never refreshes a missing route inside tools/call", async ( jsonrpc: "2.0", method: "tools/call", params: { + _meta: generationMeta("operation-without-route"), arguments: { model: "openai/gpt-image-1", operation_id: "operation-without-route", @@ -395,17 +697,21 @@ test("image.generate never refreshes a missing route inside tools/call", async ( await server.shutdown(1_000); await running; + await fs.rm(operationDirectory, { force: true, recursive: true }); }); -test("image.generate reuses the route loaded by tools/list", async () => { +test("image.generate uses the card-selected model from the route cached by tools/list", async () => { const outputDirectory = await fs.mkdtemp( path.join(os.tmpdir(), "convax-nexus-mcp-route-"), ); const png = Buffer.from([137, 80, 78, 71, 13, 10, 26, 10, 0, 0, 0, 0]); const requests = { completions: 0, + idempotencyKey: "", models: 0, providers: 0, + requestId: "", + selectedModel: "", tokens: 0, }; const responses: unknown[] = []; @@ -422,8 +728,9 @@ test("image.generate reuses the route loaded by tools/list", async () => { } as unknown as NexusSessionStore; const server = new NexusMcpServer({ client: { - fetch: async (input) => { - const request = input instanceof Request ? input : new Request(input); + fetch: async (input, init) => { + const request = + input instanceof Request ? input : new Request(input, init); const url = new URL(request.url); if (url.pathname.endsWith("/auth/token")) { requests.tokens += 1; @@ -459,11 +766,23 @@ test("image.generate reuses the route loaded by tools/list", async () => { id: "openai/gpt-image-1", name: "GPT Image 1", }, + { + architecture: { output_modalities: ["image", "text"] }, + id: "openai/gpt-image-2", + name: "GPT Image 2", + }, ], }); } if (url.pathname.endsWith("/chat/completions")) { requests.completions += 1; + requests.idempotencyKey = + request.headers.get("idempotency-key") ?? ""; + requests.requestId = + request.headers.get("x-nexus-request-id") ?? ""; + requests.selectedModel = ( + (await request.json()) as { model?: string } + ).model ?? ""; return Response.json({ choices: [ { @@ -485,6 +804,9 @@ test("image.generate reuses the route loaded by tools/list", async () => { }, now: () => new Date("2026-07-27T08:00:00.000Z"), }, + operationStore: new NexusOperationStore( + path.join(outputDirectory, ".operations"), + ), send: (value) => responses.push(value), sessions, }); @@ -520,8 +842,9 @@ test("image.generate reuses the route loaded by tools/list", async () => { jsonrpc: "2.0", method: "tools/call", params: { + _meta: generationMeta("operation-route-123"), arguments: { - model: "openai/gpt-image-1", + model: "openai/gpt-image-2", operation_id: "operation-route-123", output: "image", output_directory: outputDirectory, @@ -534,34 +857,63 @@ test("image.generate reuses the route loaded by tools/list", async () => { })}\n`, ), ); - for (let attempt = 0; attempt < 100 && responses.length < 2; attempt += 1) { + for ( + let attempt = 0; + attempt < 100 && + !responses.some((response) => (response as { id?: unknown }).id === 2); + attempt += 1 + ) { await Bun.sleep(10); } - expect(responses).toHaveLength(2); - expect(responses[1]).toMatchObject({ + const generatedResponse = responses.find( + (response) => (response as { id?: unknown }).id === 2, + ); + expect(generatedResponse).toMatchObject({ id: 2, result: { structuredContent: { artifacts: [ { mimeType: "image/png", - name: "nexus-operation-route-123-1.png", - path: "nexus-operation-route-123-1.png", }, ], }, }, }); + const generatedPath = ( + generatedResponse as { + result: { + structuredContent: { artifacts: Array<{ path: string }> }; + }; + } + ).result.structuredContent.artifacts[0]!.path; + expect(generatedPath).toMatch( + /^nexus-operation-route-123-[a-f0-9]{12}-1\.png$/u, + ); expect( - await fs.readFile( - path.join(outputDirectory, "nexus-operation-route-123-1.png"), + responses.find( + (response) => + (response as { method?: unknown }).method === + "notifications/progress", ), + ).toMatchObject({ + params: { + event: "submitted", + progressToken: "progress-operation-route-123", + schema: "convax.generation-lifecycle/1", + }, + }); + expect( + await fs.readFile(path.join(outputDirectory, generatedPath)), ).toEqual(png); expect(requests).toEqual({ completions: 1, + idempotencyKey: "operation-route-123", models: 1, providers: 1, + requestId: "operation-route-123", + selectedModel: "openai/gpt-image-2", tokens: 1, }); } finally { @@ -571,7 +923,7 @@ test("image.generate reuses the route loaded by tools/list", async () => { } }); -test("image.generate returns bounded correlated HTTP diagnostics", async () => { +test("image.generate returns one bounded failure message without leaking upstream diagnostics", async () => { const outputDirectory = await fs.mkdtemp( path.join(os.tmpdir(), "convax-nexus-mcp-server-"), ); @@ -644,6 +996,9 @@ test("image.generate returns bounded correlated HTTP diagnostics", async () => { }, now: () => new Date("2026-07-27T08:00:00.000Z"), }, + operationStore: new NexusOperationStore( + path.join(outputDirectory, ".operations"), + ), send: (value) => responses.push(value), sessions, }); @@ -669,6 +1024,7 @@ test("image.generate returns bounded correlated HTTP diagnostics", async () => { jsonrpc: "2.0", method: "tools/call", params: { + _meta: generationMeta("operation-123", "b".repeat(64)), arguments: { model: "openai/gpt-image-1", operation_id: "operation-123", @@ -683,17 +1039,24 @@ test("image.generate returns bounded correlated HTTP diagnostics", async () => { })}\n`, ), ); - for (let attempt = 0; attempt < 100 && responses.length < 2; attempt += 1) { + for ( + let attempt = 0; + attempt < 100 && + !responses.some((response) => (response as { id?: unknown }).id === 2); + attempt += 1 + ) { await Bun.sleep(10); } - expect(responses).toHaveLength(2); - const serialized = JSON.stringify(responses[1]); - expect(responses[1]).toMatchObject({ + const failedResponse = responses.find( + (response) => (response as { id?: unknown }).id === 2, + ); + const serialized = JSON.stringify(failedResponse); + expect(failedResponse).toMatchObject({ result: { content: [ { - text: "Nexus rejected image generation (HTTP 409, code metering_unsupported, request id operation-123). This image route is not enabled for Nexus metering; contact Nexus support before trying again.", + text: "Nexus image generation failed.", type: "text", }, ], @@ -712,6 +1075,9 @@ test("image.generate returns bounded correlated HTTP diagnostics", async () => { test("service.sign_out invalidates a prepared image route before revoke completes", async () => { const responses: unknown[] = []; + const operationDirectory = await fs.mkdtemp( + path.join(os.tmpdir(), "convax-nexus-mcp-sign-out-"), + ); let completions = 0; let revokeStarted = false; let resolveRevoke!: (response: Response) => void; @@ -784,6 +1150,7 @@ test("service.sign_out invalidates a prepared image route before revoke complete }, now: () => new Date("2026-07-27T08:00:00.000Z"), }, + operationStore: new NexusOperationStore(operationDirectory), send: (value) => responses.push(value), sessions, }); @@ -835,6 +1202,7 @@ test("service.sign_out invalidates a prepared image route before revoke complete jsonrpc: "2.0", method: "tools/call", params: { + _meta: generationMeta("operation-after-sign-out", "c".repeat(64)), arguments: { model: "openai/gpt-image-1", operation_id: "operation-after-sign-out", @@ -896,6 +1264,7 @@ test("service.sign_out invalidates a prepared image route before revoke complete resolveRevoke(new Response(null, { status: 204 })); await server.shutdown(1_000); await running; + await fs.rm(operationDirectory, { force: true, recursive: true }); } }); @@ -1025,14 +1394,14 @@ test("service.sign_out keeps a stale pending image catalog hidden", async () => await Bun.sleep(10); } - const listed = responses.find( - (response) => (response as { id?: unknown }).id === 1, - ) as { - result: { tools: Array<{ name: string }> }; - }; - expect(listed.result.tools.map(({ name }) => name)).toEqual([ - ...fixedToolNames, - ]); + expect( + responses.find((response) => (response as { id?: unknown }).id === 1), + ).toMatchObject({ + id: 1, + result: { + tools: fixedToolNames.map((name) => ({ name })), + }, + }); const serialized = JSON.stringify(responses); expect(serialized).not.toContain("access-token-for-stale-catalog"); expect(serialized).not.toContain("data-token-for-stale-catalog"); diff --git a/packages/tools/nexus-mcp/test/nexus-client.test.ts b/packages/tools/nexus-mcp/test/nexus-client.test.ts index d97845a..a1e6824 100644 --- a/packages/tools/nexus-mcp/test/nexus-client.test.ts +++ b/packages/tools/nexus-mcp/test/nexus-client.test.ts @@ -813,6 +813,7 @@ describe("NexusClient", () => { const imageRequests: Array<{ authorization: string | null; body: unknown; + idempotencyKey: string | null; requestId: string | null; }> = []; let imageAttempts = 0; @@ -847,6 +848,7 @@ describe("NexusClient", () => { imageRequests.push({ authorization: headers.get("authorization"), body: JSON.parse(String(init?.body)), + idempotencyKey: headers.get("idempotency-key"), requestId: headers.get("x-nexus-request-id"), }); imageAttempts += 1; @@ -925,6 +927,7 @@ describe("NexusClient", () => { model: "openai/gpt-image-1", stream: false, }, + idempotencyKey: "operation-123", requestId: "operation-123", }); diff --git a/packages/tools/nexus-mcp/test/operation-store.test.ts b/packages/tools/nexus-mcp/test/operation-store.test.ts new file mode 100644 index 0000000..f82d36d --- /dev/null +++ b/packages/tools/nexus-mcp/test/operation-store.test.ts @@ -0,0 +1,113 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { createHash } from "node:crypto"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; + +import { + nexusGenerationRequestFingerprint, + NexusOperationStore, +} from "../src/operation-store.ts"; + +const roots: string[] = []; +const png = Buffer.from([137, 80, 78, 71, 13, 10, 26, 10, 0, 0, 0, 0]); + +afterEach(async () => { + await Promise.all( + roots + .splice(0) + .map((root) => fs.rm(root, { force: true, recursive: true })), + ); +}); + +describe("NexusOperationStore", () => { + test("keeps one stable recovery binding and replays a completed result after restart", async () => { + const root = await fs.mkdtemp( + path.join(os.tmpdir(), "convax-nexus-operations-"), + ); + roots.push(root); + const directory = path.join(root, "operations"); + const first = new NexusOperationStore(directory); + const request = { + model: "openai/gpt-image-1", + outputModalities: ["image", "text"], + prompt: "Draw a blue circle.", + } as const; + const prepared = await first.createPrepared({ + fingerprint: nexusGenerationRequestFingerprint(request), + operationId: "operation-persisted-123", + request, + requestDigest: "a".repeat(64), + }); + const running = await first.markRunning(prepared.operationId, prepared); + const succeeded = await first.succeed(running.operationId, running, [ + { + bytes: png, + extension: "png", + mimeType: "image/png", + }, + ]); + const firstBinding = await first.binding(); + + const restarted = new NexusOperationStore(directory); + expect(await restarted.binding()).toBe(firstBinding); + expect(await restarted.find(prepared.operationId)).toEqual(succeeded); + + const output = path.join(root, "output"); + await fs.mkdir(output); + const artifacts = await restarted.materialize( + prepared.operationId, + succeeded, + output, + ); + expect(artifacts).toHaveLength(1); + expect(await fs.readFile(path.join(output, artifacts[0]!.path))).toEqual( + png, + ); + + await restarted.acknowledge(prepared.operationId, succeeded); + expect(await restarted.find(prepared.operationId)).toBeNull(); + }); + + test("fails closed when persisted result bytes change", async () => { + const root = await fs.mkdtemp( + path.join(os.tmpdir(), "convax-nexus-operations-"), + ); + roots.push(root); + const store = new NexusOperationStore(path.join(root, "operations")); + const request = { + model: "openai/gpt-image-1", + outputModalities: ["image", "text"], + prompt: "Draw a red circle.", + } as const; + const prepared = await store.createPrepared({ + fingerprint: nexusGenerationRequestFingerprint(request), + operationId: "operation-tampered-123", + request, + requestDigest: "b".repeat(64), + }); + const running = await store.markRunning(prepared.operationId, prepared); + const succeeded = await store.succeed(running.operationId, running, [ + { + bytes: png, + extension: "png", + mimeType: "image/png", + }, + ]); + const artifactRoot = path.join( + store.directory, + "artifacts", + createHash("sha256").update(prepared.operationId).digest("hex"), + ); + await fs.writeFile( + path.join(artifactRoot, succeeded.artifacts[0]!.name), + Buffer.from("tampered"), + ); + const output = path.join(root, "output"); + await fs.mkdir(output); + + await expect( + store.materialize(prepared.operationId, succeeded, output), + ).rejects.toThrow("stored generation artifact changed"); + }); +}); diff --git a/packages/tools/nexus-mcp/test/plugin-service.test.ts b/packages/tools/nexus-mcp/test/plugin-service.test.ts index ce367b5..d3e0faa 100644 --- a/packages/tools/nexus-mcp/test/plugin-service.test.ts +++ b/packages/tools/nexus-mcp/test/plugin-service.test.ts @@ -172,7 +172,7 @@ describe("NexusPluginService", () => { }); }); - test("keeps a deployed base Access response connected while Plan and Billing remain unavailable", async () => { + test("treats an authoritative base Access response without optional Plan and Billing fields as connected", async () => { const root = await fs.mkdtemp( path.join(os.tmpdir(), "convax-nexus-service-"), ); @@ -253,4 +253,155 @@ describe("NexusPluginService", () => { "The selected Nexus Plan is not available", ); }); + + test("fails the whole status request when a required Nexus query fails", async () => { + const root = await fs.mkdtemp( + path.join(os.tmpdir(), "convax-nexus-service-"), + ); + roots.push(root); + const service = new NexusPluginService( + {} as NexusAuthorization, + { + async access() { + return { + access: { + accessStartsAt: "2026-07-26T08:00:00.000Z", + id: "26010000-0000-4000-8000-000000000005", + planId: "26010000-0000-4000-8000-000000000004", + status: "ACTIVE", + }, + subject: "pairwise-subject", + workspace: { + id: "26010000-0000-4000-8000-000000000003", + name: "Convax", + slug: "convax", + }, + }; + }, + async providers() { + throw new Error("provider catalog unavailable"); + }, + async quota() { + return { + availableUnits: "1000000", + consumedUnits: "0", + periodEnd: "2026-08-26T08:00:00.000Z", + }; + }, + } as unknown as NexusClient, + { + async read() { + return { + nexusOrigin: "https://nexus.test", + refreshToken: "refresh-token-with-sufficient-length", + schema: "convax.nexus-refresh-grant/1", + workspaceSlug: "convax", + }; + }, + } as unknown as NexusSessionStore, + new NexusCheckoutStore({ XDG_CONFIG_HOME: root }), + ); + + await expect(service.status()).rejects.toThrow( + "Nexus service status could not be verified", + ); + }); + + test("single-flights concurrent Checkout creation and clears its durable state on sign-out failure", async () => { + const root = await fs.mkdtemp( + path.join(os.tmpdir(), "convax-nexus-service-"), + ); + roots.push(root); + const checkouts = new NexusCheckoutStore({ XDG_CONFIG_HOME: root }); + let createCalls = 0; + let releaseCheckout!: () => void; + const checkoutGate = new Promise((resolve) => { + releaseCheckout = resolve; + }); + const client = { + async access() { + return { + access: { + accessStartsAt: "2026-07-26T08:00:00.000Z", + id: "26010000-0000-4000-8000-000000000005", + planId: "26010000-0000-4000-8000-000000000004", + status: "ACTIVE", + }, + billing: { + availablePlans: [ + { + billingInterval: "MONTH", + id: "26010000-0000-4000-8000-000000000006", + key: "pro", + name: "Pro", + }, + ], + checkoutAvailable: true, + }, + plan: { + billingInterval: "MONTH", + id: "26010000-0000-4000-8000-000000000004", + key: "free", + name: "Free", + }, + subject: "pairwise-subject", + workspace: { + id: "26010000-0000-4000-8000-000000000003", + name: "Convax", + slug: "convax", + }, + }; + }, + async createCheckout() { + createCalls += 1; + await checkoutGate; + return { + browserUrl: "https://checkout.creem.test/session/shared", + checkoutId: "26010000-0000-4000-8000-000000000009", + status: "CREATED", + }; + }, + async signOut() { + throw new Error("remote revoke failed"); + }, + } as unknown as NexusClient; + const service = new NexusPluginService( + { cancel() {} } as unknown as NexusAuthorization, + client, + { + async read() { + return { + nexusOrigin: "https://nexus.test", + refreshToken: "refresh-token-with-sufficient-length", + schema: "convax.nexus-refresh-grant/1", + workspaceSlug: "convax", + }; + }, + } as unknown as NexusSessionStore, + checkouts, + ); + + const first = service.checkout("pro"); + const second = service.checkout("pro"); + for (let attempt = 0; attempt < 100 && createCalls === 0; attempt += 1) { + await Bun.sleep(1); + } + expect(createCalls).toBe(1); + releaseCheckout(); + expect(await Promise.all([first, second])).toEqual([ + { + checkout_id: "26010000-0000-4000-8000-000000000009", + checkout_url: "https://checkout.creem.test/session/shared", + schema: "convax.plugin-service-checkout/1", + }, + { + checkout_id: "26010000-0000-4000-8000-000000000009", + checkout_url: "https://checkout.creem.test/session/shared", + schema: "convax.plugin-service-checkout/1", + }, + ]); + expect(await checkouts.read()).not.toBeNull(); + await expect(service.signOut()).rejects.toThrow("remote revoke failed"); + expect(await checkouts.read()).toBeNull(); + }); });