diff --git a/src/capability-proxy.ts b/src/capability-proxy.ts index 73e96c6..05f2be8 100644 --- a/src/capability-proxy.ts +++ b/src/capability-proxy.ts @@ -1,5 +1,6 @@ import type { Agent as HttpAgent, IncomingMessage, request as httpRequest } from 'node:http' import type { Socket } from 'node:net' +import { Readable } from 'node:stream' import { apiTokenMintInstructions, cliBootstrapInstructions, @@ -43,6 +44,15 @@ import { readPackageVersion } from './package-info.js' */ export const capabilityProxySessionPath = 'v1/capability-proxy/session' export const capabilityProxyCallPath = 'v1/capability-proxy/call' +/** Streaming ambient/secret fetch hop (raw body; not JSON `/call`). */ +export const capabilityProxyGatewayFetchPath = 'v1/capability-proxy/gateway-fetch' + +/** Allowlisted metadata headers for {@link capabilityProxyGatewayFetchPath}. */ +export const gatewayFetchUrlHeaderName = 'X-Kody-Gateway-Url' +export const gatewayFetchMethodHeaderName = 'X-Kody-Gateway-Method' +export const gatewayFetchPackageIdHeaderName = 'X-Kody-Gateway-Package-Id' +export const gatewayFetchHeadersHeaderName = 'X-Kody-Gateway-Headers' +export const secretResolutionHeaderName = 'x-kody-secret-resolution' /** * Origin runs a placeholder `fetch` under a 5-minute deadline that starts only @@ -371,6 +381,183 @@ export async function callCapabilityProxy( return isRecord(body) ? body.result : undefined } +export type CapabilityProxyGatewayFetchStreamHeaders = { + url: string + method?: string + packageId?: string + /** Outbound header name → string (may include secret placeholders). */ + headers?: Record + /** Outbound `Content-Type` (set on the Open API request). */ + contentType?: string | null + /** `on` | `headers-only` | `off`. */ + secretResolution?: string | null +} + +/** + * Streaming CapabilityProxy hop: `POST /v1/capability-proxy/gateway-fetch`. + * The Open API body is the outbound fetch body (no JSON/base64 wrapper). + * Returns the upstream `Response` (or an Open API JSON error envelope). + * + * Uses Node `http(s)` so duplex streaming works with the same 7-minute idle + * timeout as JSON CapabilityProxy calls (undici's global fetch caps at 300s). + */ +export async function callCapabilityProxyGatewayFetchStream( + input: CapabilityProxyClientInput & { + gateway: CapabilityProxyGatewayFetchStreamHeaders + /** Raw outbound body stream (may be empty for GET/HEAD). */ + body: IncomingMessage | Readable | null + contentLength?: string | null + }, +): Promise { + assertTokenSafeApiUrl(input.apiUrl) + const url = withOrgQuery( + capabilityProxyUrl(input.apiUrl, capabilityProxyGatewayFetchPath), + input.org, + ) + const timeoutMs = capabilityProxyHttpTimeoutMs + const connectTimeoutMs = capabilityProxyConnectTimeoutMs + const useEnvProxy = nodeUsesEnvProxy() + if (useEnvProxy && !nodeAgentHonorsEnvProxy()) { + throw new Error( + `Node ${process.version} cannot route CapabilityProxy requests through HTTP_PROXY / HTTPS_PROXY: env-proxy support needs Node 22.21 or 24.5 and later. Upgrade Node or unset NODE_USE_ENV_PROXY / --use-env-proxy.`, + ) + } + const transports = await loadTransports(useEnvProxy) + const transport = transports[url.protocol as keyof Transports] + if (!transport) { + throw new TypeError(`Unsupported protocol for the Kody API: ${url.protocol}`) + } + const headers: Record> = { + accept: '*/*', + authorization: `Bearer ${input.token}`, + 'user-agent': `${cliName}/${readPackageVersion()}`, + [gatewayFetchUrlHeaderName]: input.gateway.url, + } + if (input.gateway.method) { + headers[gatewayFetchMethodHeaderName] = input.gateway.method + } + if (input.gateway.packageId) { + headers[gatewayFetchPackageIdHeaderName] = input.gateway.packageId + } + if (input.gateway.headers && Object.keys(input.gateway.headers).length > 0) { + headers[gatewayFetchHeadersHeaderName] = JSON.stringify(input.gateway.headers) + } + if (input.gateway.secretResolution) { + headers[secretResolutionHeaderName] = input.gateway.secretResolution + } + if (input.gateway.contentType) { + headers['content-type'] = input.gateway.contentType + } + if (input.contentLength) { + headers['content-length'] = input.contentLength + } + const signal = input.signal + return await new Promise((resolve, reject) => { + const request = transport.request( + url, + { + agent: transport.agent, + method: 'POST', + headers, + }, + (response) => { + clearTimeout(connectTimer) + resolve(nodeIncomingToWebResponse(response)) + }, + ) + const connectTimer = setTimeout(() => { + request.destroy( + new Error( + `Kody API did not connect within ${connectTimeoutMs}ms (${url.pathname}).`, + ), + ) + }, connectTimeoutMs) + request.on('socket', (socket) => { + const clearConnectTimer = () => clearTimeout(connectTimer) + if (url.protocol === 'https:') { + if (tlsHandshakePending(socket) === false) { + clearConnectTimer() + return + } + socket.once('secureConnect', clearConnectTimer) + return + } + if (!socket.connecting) { + clearConnectTimer() + return + } + socket.once('connect', clearConnectTimer) + }) + request.setTimeout(timeoutMs, () => + request.destroy( + new Error(`Kody API sent nothing for ${timeoutMs}ms (${url.pathname}).`), + ), + ) + request.on('error', (error) => { + clearTimeout(connectTimer) + reject( + error instanceof CapabilityProxyError + ? error + : new CapabilityProxyError( + `Could not reach the Kody API at ${url.origin} (${describeNetworkError(error)}). Check --api-url / KODY_API_URL and your network.`, + ), + ) + }) + if (signal) { + const abort = () => { + clearTimeout(connectTimer) + request.destroy(abortError(signal.reason)) + } + if (signal.aborted) { + abort() + } else { + signal.addEventListener('abort', abort, { once: true }) + request.once('close', () => signal.removeEventListener('abort', abort)) + } + } + if (input.body) { + input.body.pipe(request) + input.body.on('error', (error) => { + clearTimeout(connectTimer) + request.destroy(error) + }) + } else { + request.end() + } + }) +} + +/** Convert a Node response into a streaming Web `Response`. */ +export function nodeIncomingToWebResponse(response: IncomingMessage): Response { + const headers = new Headers() + for (const [name, value] of Object.entries(response.headers)) { + if (Array.isArray(value)) { + for (const entry of value) headers.append(name, entry) + } else if (value !== undefined) { + headers.set(name, value) + } + } + const status = response.statusCode ?? 0 + const bodyless = + status === 101 || status === 103 || status === 204 || status === 205 || status === 304 + if (bodyless) { + response.resume() + return new Response(null, { + status, + statusText: response.statusMessage ?? '', + headers, + }) + } + return new Response( + Readable.toWeb(response) as import('node:stream/web').ReadableStream, + { + status, + statusText: response.statusMessage ?? '', + headers, + }, + ) +} + /** * Preflight ADR 0068 call-map picks via `kody.credentialClientCheck` before * local code runs. Throws on a structured check failure; returns warnings for diff --git a/src/local-execute.ts b/src/local-execute.ts index 3468edb..7c3839c 100644 --- a/src/local-execute.ts +++ b/src/local-execute.ts @@ -2,13 +2,7 @@ import { spawn, type ChildProcess } from 'node:child_process' import { randomBytes } from 'node:crypto' import { rmSync } from 'node:fs' import { mkdtemp, rm, writeFile } from 'node:fs/promises' -import { - createServer, - request as httpRequest, - type IncomingMessage, - type Server, -} from 'node:http' -import type { AddressInfo } from 'node:net' +import { request as httpRequest, type IncomingMessage } from 'node:http' import { tmpdir } from 'node:os' import { join } from 'node:path' import type { Readable } from 'node:stream' @@ -20,6 +14,11 @@ import { openCapabilityProxySession, type CapabilityProxySession, } from './capability-proxy.js' +import { forwardLocalGatewayFetchStream } from './local-gateway-fetch-forward.js' +import { + startLocalRuntimeBridge, + type LocalRuntimeBridge, +} from './local-runtime-bridge.js' import { missingCredentialCallMapFeatureMessage, sessionSupportsCredentialCallMap, @@ -50,8 +49,8 @@ import { resolveMockFetchGateway, tryHandleMockGatewayFetchCall, } from './mock/mock-fetch-gateway.js' -import type { ToolCallResult } from './tool-result.js' import { redact } from './redact.js' +import type { ToolCallResult } from './tool-result.js' import { ensureWorkerdBinary } from './workerd-binary.js' export { @@ -108,8 +107,6 @@ export type LocalExecuteInput = { type SandboxResponse = { result?: unknown; error?: string; logs?: Array } -type Bridge = { port: number; close: () => Promise } - const workerdStartTimeoutMs = 30_000 const workerdExitGraceMs = 1_000 @@ -174,7 +171,7 @@ export async function runLocalExecute(input: LocalExecuteInput): Promise child?.kill('SIGKILL') const releaseSignals = cleanupOnSignal(() => { @@ -197,9 +194,9 @@ export async function runLocalExecute(input: LocalExecuteInput): Promise { + forwardCall: async (call, signal) => { const mocked = tryHandleMockGatewayFetchCall({ gateway: mockFetchGateway, path: call.path, @@ -217,6 +214,15 @@ export async function runLocalExecute(input: LocalExecuteInput): Promise { + await forwardLocalGatewayFetchStream({ + request, + response, + signal, + client, + ...(mockFetchGateway ? { mockFetchGateway } : {}), + }) + }, }) const files = { entry: 'entry.js', user: 'main.js', runtime: 'runtime.js' } let packageModules = await writePackageModuleFiles(workDir, packageGraph) @@ -444,65 +450,6 @@ function postToWorkerd(input: { }) } -async function startBridge(input: { - runSecret: string - forward: ( - call: { path: Array; args: Array }, - signal: AbortSignal, - ) => Promise -}): Promise { - const inFlight = new AbortController() - const server: Server = createServer(async (request, response) => { - const reply = (status: number, body: unknown) => { - response.writeHead(status, { 'content-type': 'application/json' }) - response.end(JSON.stringify(body)) - } - if ( - request.method !== 'POST' || - request.url !== '/call' || - request.headers[runSecretHeader] !== input.runSecret - ) { - reply(403, { error: 'Forbidden' }) - return - } - try { - const call = parseBridgeCall(await readBody(request)) - const result = await input.forward(call, inFlight.signal) - reply(200, { result }) - } catch (error) { - reply(200, { error: redact(error instanceof Error ? error.message : String(error)) }) - } - }) - await new Promise((resolve, reject) => { - server.once('error', reject) - server.listen(0, '127.0.0.1', () => resolve()) - }) - return { - port: (server.address() as AddressInfo).port, - close: () => - new Promise((resolve) => { - inFlight.abort() - server.closeAllConnections() - server.close(() => resolve()) - }), - } -} - -function parseBridgeCall(body: string): { path: Array; args: Array } { - const parsed = JSON.parse(body) as { path?: unknown; args?: unknown } - if ( - !Array.isArray(parsed.path) || - parsed.path.length === 0 || - !parsed.path.every((segment) => typeof segment === 'string') - ) { - throw new Error('Invalid kody:runtime call path.') - } - return { - path: parsed.path as Array, - args: Array.isArray(parsed.args) ? parsed.args : [], - } -} - function assertCredentialCallMapSupported( session: CapabilityProxySession, input: LocalExecuteInput, diff --git a/src/local-gateway-fetch-forward.ts b/src/local-gateway-fetch-forward.ts new file mode 100644 index 0000000..5a37fc2 --- /dev/null +++ b/src/local-gateway-fetch-forward.ts @@ -0,0 +1,213 @@ +/** + * Node-side handlers for the workerd → bridge `POST /gateway-fetch` hop: + * mock MSW path (buffered) or prod CapabilityProxy streaming hop. + */ + +import type { IncomingMessage, ServerResponse } from 'node:http' +import { + callCapabilityProxyGatewayFetchStream, + gatewayFetchHeadersHeaderName, + gatewayFetchMethodHeaderName, + gatewayFetchPackageIdHeaderName, + gatewayFetchUrlHeaderName, + secretResolutionHeaderName, + type CapabilityProxyClientInput, +} from './capability-proxy.js' +import { + readBodyBuffer, + writeWebResponse, +} from './local-runtime-bridge.js' +import type { MockFetchGateway } from './mock/mock-fetch-gateway.js' +import { runSecretHeader } from './local-runtime-source.js' + +const gatewayMetaHeaderNames = [ + gatewayFetchUrlHeaderName, + gatewayFetchMethodHeaderName, + gatewayFetchPackageIdHeaderName, + gatewayFetchHeadersHeaderName, + secretResolutionHeaderName, + 'content-type', + 'content-length', +] as const + +function headerValue( + headers: IncomingMessage['headers'], + name: string, +): string | undefined { + const value = headers[name.toLowerCase()] + if (Array.isArray(value)) return value[0] + return value +} + +export async function forwardLocalGatewayFetchStream(input: { + request: IncomingMessage + response: ServerResponse + signal: AbortSignal + client: CapabilityProxyClientInput + mockFetchGateway?: MockFetchGateway +}): Promise { + const url = headerValue(input.request.headers, gatewayFetchUrlHeaderName)?.trim() + if (!url) { + await writeWebResponse( + input.response, + Response.json( + { + error: { + code: 'invalid_request', + message: + 'X-Kody-Gateway-Url is required and must be a non-empty absolute URL.', + }, + }, + { status: 400, headers: { 'Cache-Control': 'no-store' } }, + ), + ) + return + } + + if (input.mockFetchGateway) { + const body = await readBodyBuffer(input.request) + const method = + headerValue(input.request.headers, gatewayFetchMethodHeaderName)?.trim() || + 'GET' + const packageId = headerValue( + input.request.headers, + gatewayFetchPackageIdHeaderName, + )?.trim() + const rawHeaders = headerValue( + input.request.headers, + gatewayFetchHeadersHeaderName, + ) + let headers: Record = {} + if (rawHeaders) { + try { + const parsed = JSON.parse(rawHeaders) as unknown + if (parsed != null && typeof parsed === 'object' && !Array.isArray(parsed)) { + headers = Object.fromEntries( + Object.entries(parsed as Record).flatMap( + ([key, value]) => + typeof value === 'string' ? [[key, value] as const] : [], + ), + ) + } + } catch { + await writeWebResponse( + input.response, + Response.json( + { + error: { + code: 'invalid_request', + message: + 'X-Kody-Gateway-Headers must be valid JSON: an object of header names to string values.', + }, + }, + { status: 400, headers: { 'Cache-Control': 'no-store' } }, + ), + ) + return + } + } + const contentType = headerValue(input.request.headers, 'content-type') + if (contentType) headers['content-type'] = contentType + const resolution = headerValue( + input.request.headers, + secretResolutionHeaderName, + ) + if (resolution) headers[secretResolutionHeaderName] = resolution + const result = await input.mockFetchGateway.handle( + { + ...(packageId ? { packageId } : {}), + request: { + url, + method, + headers, + ...(body.byteLength > 0 + ? { bodyBase64: body.toString('base64') } + : {}), + }, + }, + input.signal, + ) + const nullBody = + result.status === 101 || + result.status === 103 || + result.status === 204 || + result.status === 205 || + result.status === 304 + const bytes = Buffer.from(result.bodyBase64 ?? '', 'base64') + await writeWebResponse( + input.response, + new Response(nullBody ? null : bytes, { + status: result.status, + statusText: result.statusText, + headers: result.headers, + }), + ) + return + } + + const method = headerValue(input.request.headers, gatewayFetchMethodHeaderName) + const packageId = headerValue( + input.request.headers, + gatewayFetchPackageIdHeaderName, + ) + const rawHeaders = headerValue( + input.request.headers, + gatewayFetchHeadersHeaderName, + ) + let outboundHeaders: Record | undefined + if (rawHeaders) { + try { + const parsed = JSON.parse(rawHeaders) as unknown + if (parsed != null && typeof parsed === 'object' && !Array.isArray(parsed)) { + outboundHeaders = Object.fromEntries( + Object.entries(parsed as Record).flatMap(([key, value]) => + typeof value === 'string' ? [[key, value] as const] : [], + ), + ) + } + } catch { + await writeWebResponse( + input.response, + Response.json( + { + error: { + code: 'invalid_request', + message: + 'X-Kody-Gateway-Headers must be valid JSON: an object of header names to string values.', + }, + }, + { status: 400, headers: { 'Cache-Control': 'no-store' } }, + ), + ) + return + } + } + const contentType = headerValue(input.request.headers, 'content-type') + const secretResolution = headerValue( + input.request.headers, + secretResolutionHeaderName, + ) + const contentLength = headerValue(input.request.headers, 'content-length') + const upstream = await callCapabilityProxyGatewayFetchStream({ + ...input.client, + signal: input.signal, + gateway: { + url, + ...(method ? { method } : {}), + ...(packageId ? { packageId } : {}), + ...(outboundHeaders ? { headers: outboundHeaders } : {}), + ...(contentType ? { contentType } : {}), + ...(secretResolution ? { secretResolution } : {}), + }, + body: input.request, + ...(contentLength ? { contentLength } : {}), + }) + await writeWebResponse(input.response, upstream) +} + +/** Headers the workerd bridge may forward (excludes the run secret). */ +export function isGatewayFetchForwardHeader(name: string): boolean { + const lower = name.toLowerCase() + if (lower === runSecretHeader) return false + return gatewayMetaHeaderNames.some((allowed) => allowed.toLowerCase() === lower) +} diff --git a/src/local-host-runtime-surface.ts b/src/local-host-runtime-surface.ts new file mode 100644 index 0000000..396e95e --- /dev/null +++ b/src/local-host-runtime-surface.ts @@ -0,0 +1,82 @@ +/** + * Hosted CapabilityProxy / local-execute host runtime surface that `--local` + * must cover. Keep in sync with kentcdodds/kody: + * + * - `packages/worker/src/open-api/capability-proxy.ts` (JSON `/call` specials) + * - `POST /v1/capability-proxy/gateway-fetch` (`gatewayFetchStream`) + * - `docs/contributing/contracts/cli-streaming-gateway-fetch.md` + * - `docs/contributing/contracts/cli-local-credential-runtime.md` + * + * Generic MCP tools (`kody.emailSend`, …) hop through the JSON Proxy and are + * not listed here. Prefix families (`packageStorage*`, `packageSecret*`, + * `credentialClient*`, `kody.mcp.*`) are listed once as patterns. + */ + +/** Special-cased CapabilityProxy JSON `/call` paths (exact). */ +export const hostedCapabilityProxySpecialFunctions = [ + 'kody.authenticatedFetch', + 'kody.gatewayFetch', + 'kody.oauthClientCredentials', + 'workflows.create', +] as const + +/** Prefix families served by CapabilityProxy JSON `/call`. */ +export const hostedCapabilityProxySpecialPrefixes = [ + 'kody.packageStorage', + 'kody.packageSecret', + 'kody.credentialClient', + 'kody.mcp.', +] as const + +/** + * Host methods that cannot use the JSON `/call` Proxy hop. The CLI workerd + * `kody:runtime` must implement these (streaming body / real `Response`). + */ +export const hostedLocalHostSpecialFunctions = ['kody.gatewayFetchStream'] as const + +/** + * Full audit list: every hosted special the CLI `--local` host must support + * (Proxy hop, dedicated host method, or documented prefix family). + */ +export const hostedLocalExecuteRuntimeSurface = [ + ...hostedCapabilityProxySpecialFunctions, + ...hostedCapabilityProxySpecialPrefixes, + ...hostedLocalHostSpecialFunctions, +] as const + +export type HostedLocalExecuteRuntimeSurface = + (typeof hostedLocalExecuteRuntimeSurface)[number] + +/** + * CLI `--local` implementations that are more than the generic JSON Proxy. + * Drift tests assert these markers appear in the workerd runtime source. + */ +export const localHostImplementedRuntimeMarkers = [ + { + id: 'kody.gatewayFetchStream', + /** Dedicated streaming host method (not Proxy → `/call`). */ + sourceMarker: 'gatewayFetchStream', + bridgeMarker: '/gateway-fetch', + }, + { + id: 'kody.authenticatedFetch', + sourceMarker: 'createAuthenticatedFetch', + bridgeMarker: null, + }, + { + id: 'kody.oauthClientCredentials', + sourceMarker: 'oauthClientCredentials', + bridgeMarker: null, + }, + { + id: 'kody.credentialClient*', + sourceMarker: 'credentialClientOpen', + bridgeMarker: null, + }, + { + id: 'kody.gatewayFetch', + /** JSON hop via Proxy; shim calls `kody.gatewayFetch`. */ + sourceMarker: "__kodyNamespace(['kody'])", + bridgeMarker: '/call', + }, +] as const diff --git a/src/local-package-run.ts b/src/local-package-run.ts index bd4f856..b7ce9c2 100644 --- a/src/local-package-run.ts +++ b/src/local-package-run.ts @@ -7,13 +7,7 @@ import { spawn, type ChildProcess } from 'node:child_process' import { randomBytes } from 'node:crypto' import { rmSync } from 'node:fs' import { mkdtemp, rm, writeFile } from 'node:fs/promises' -import { - createServer, - request as httpRequest, - type IncomingMessage, - type Server, -} from 'node:http' -import type { AddressInfo } from 'node:net' +import { request as httpRequest, type IncomingMessage } from 'node:http' import { tmpdir } from 'node:os' import { join } from 'node:path' import type { Readable } from 'node:stream' @@ -24,11 +18,16 @@ import { } from './capability-proxy.js' import { apiTokenEnvVar } from './defaults.js' import { bundleLocalUserModule } from './local-bundle.js' +import { forwardLocalGatewayFetchStream } from './local-gateway-fetch-forward.js' import { localExecuteGatewayFetchShimModuleName, resolveLocalPackageGraph, writePackageModuleFiles, } from './local-package-graph.js' +import { + startLocalRuntimeBridge, + type LocalRuntimeBridge, +} from './local-runtime-bridge.js' import { createRuntimeFlatAliasSource, createWorkerdConfig, @@ -84,7 +83,6 @@ export type LocalPackageRunInput = { } type SandboxResponse = { result?: unknown; error?: string; logs?: Array } -type Bridge = { port: number; close: () => Promise } const workerdStartTimeoutMs = 30_000 const workerdExitGraceMs = 1_000 @@ -111,7 +109,7 @@ export async function runLocalPackageModule( const runSecret = randomBytes(32).toString('hex') const workDir = await mkdtemp(join(tmpdir(), 'kody-pkg-')) let child: ChildProcess | undefined - let bridge: Bridge | undefined + let bridge: LocalRuntimeBridge | undefined const stopWorkerd = () => child?.kill('SIGKILL') const releaseSignals = cleanupOnSignal(() => { stopWorkerd() @@ -171,9 +169,9 @@ export async function runLocalPackageModule( ...(input.packageRoot ? { packageRoot: input.packageRoot } : {}), onStatus: input.onStatus, }) - bridge = await startBridge({ + bridge = await startLocalRuntimeBridge({ runSecret, - forward: async (call, signal) => { + forwardCall: async (call, signal) => { const mocked = tryHandleMockGatewayFetchCall({ gateway: mockFetchGateway, path: call.path, @@ -187,6 +185,15 @@ export async function runLocalPackageModule( signal, }) }, + forwardGatewayFetchStream: async ({ request, response, signal }) => { + await forwardLocalGatewayFetchStream({ + request, + response, + signal, + client, + ...(mockFetchGateway ? { mockFetchGateway } : {}), + }) + }, }) const files = { entry: 'entry.js', user: 'main.js', runtime: 'runtime.js' } let packageModules = await writePackageModuleFiles( @@ -394,65 +401,6 @@ function postToWorkerd(input: { }) } -async function startBridge(input: { - runSecret: string - forward: ( - call: { path: Array; args: Array }, - signal: AbortSignal, - ) => Promise -}): Promise { - const inFlight = new AbortController() - const server: Server = createServer(async (request, response) => { - const reply = (status: number, body: unknown) => { - response.writeHead(status, { 'content-type': 'application/json' }) - response.end(JSON.stringify(body)) - } - if ( - request.method !== 'POST' || - request.url !== '/call' || - request.headers[runSecretHeader] !== input.runSecret - ) { - reply(403, { error: 'Forbidden' }) - return - } - try { - const call = parseBridgeCall(await readBody(request)) - const result = await input.forward(call, inFlight.signal) - reply(200, { result }) - } catch (error) { - reply(200, { error: redact(error instanceof Error ? error.message : String(error)) }) - } - }) - await new Promise((resolve, reject) => { - server.once('error', reject) - server.listen(0, '127.0.0.1', () => resolve()) - }) - return { - port: (server.address() as AddressInfo).port, - close: () => - new Promise((resolve) => { - inFlight.abort() - server.closeAllConnections() - server.close(() => resolve()) - }), - } -} - -function parseBridgeCall(body: string): { path: Array; args: Array } { - const parsed = JSON.parse(body) as { path?: unknown; args?: unknown } - if ( - !Array.isArray(parsed.path) || - parsed.path.length === 0 || - !parsed.path.every((segment) => typeof segment === 'string') - ) { - throw new Error('Invalid kody:runtime call path.') - } - return { - path: parsed.path as Array, - args: Array.isArray(parsed.args) ? parsed.args : [], - } -} - async function readBody(stream: IncomingMessage): Promise { const chunks: Array = [] for await (const chunk of stream) { diff --git a/src/local-runtime-bridge.ts b/src/local-runtime-bridge.ts new file mode 100644 index 0000000..b1a49f6 --- /dev/null +++ b/src/local-runtime-bridge.ts @@ -0,0 +1,189 @@ +/** + * Loopback HTTP bridge between local workerd and the CLI host process. + * + * - `POST /call` — JSON CapabilityProxy hop (`{ path, args }` → `{ result|error }`) + * - `POST /gateway-fetch` — streaming gateway-fetch hop (raw body in/out) + */ + +import { + createServer, + type IncomingMessage, + type Server, + type ServerResponse, +} from 'node:http' +import type { AddressInfo } from 'node:net' +import { Readable } from 'node:stream' +import { redact } from './redact.js' +import { runSecretHeader } from './local-runtime-source.js' + +export const localBridgeCallPath = '/call' +export const localBridgeGatewayFetchPath = '/gateway-fetch' + +export type LocalRuntimeBridge = { + port: number + close: () => Promise +} + +export type LocalRuntimeBridgeCall = { + path: Array + args: Array +} + +export type LocalRuntimeBridgeInput = { + runSecret: string + forwardCall: ( + call: LocalRuntimeBridgeCall, + signal: AbortSignal, + ) => Promise + /** + * Streaming gateway-fetch. Must write a complete HTTP response to + * `response` (status, headers, body) — including Open API error envelopes. + */ + forwardGatewayFetchStream: (input: { + request: IncomingMessage + response: ServerResponse + signal: AbortSignal + }) => Promise +} + +export async function startLocalRuntimeBridge( + input: LocalRuntimeBridgeInput, +): Promise { + const inFlight = new AbortController() + const server: Server = createServer(async (request, response) => { + const replyJson = ( + status: number, + body: unknown, + headers: Record = {}, + ) => { + response.writeHead(status, { + 'content-type': 'application/json', + ...headers, + }) + response.end(JSON.stringify(body)) + } + /** Open API-shaped error so workerd `gatewayFetchStream` throws (needs no-store). */ + const replyGatewayFetchError = (status: number, message: string, code: string) => { + replyJson( + status, + { error: { code, message } }, + { 'Cache-Control': 'no-store' }, + ) + } + if (request.headers[runSecretHeader] !== input.runSecret) { + replyJson(403, { error: 'Forbidden' }) + return + } + if (request.method === 'POST' && request.url === localBridgeCallPath) { + try { + const call = parseBridgeCall(await readBodyText(request)) + const result = await input.forwardCall(call, inFlight.signal) + replyJson(200, { result }) + } catch (error) { + replyJson(200, { + error: redact(error instanceof Error ? error.message : String(error)), + }) + } + return + } + if (request.method === 'POST' && request.url === localBridgeGatewayFetchPath) { + try { + await input.forwardGatewayFetchStream({ + request, + response, + signal: inFlight.signal, + }) + } catch (error) { + if (response.headersSent) { + response.destroy( + error instanceof Error ? error : new Error(String(error)), + ) + return + } + replyGatewayFetchError( + 502, + redact(error instanceof Error ? error.message : String(error)), + 'gateway_fetch_failed', + ) + } + return + } + replyJson(403, { error: 'Forbidden' }) + }) + await new Promise((resolve, reject) => { + server.once('error', reject) + server.listen(0, '127.0.0.1', () => resolve()) + }) + return { + port: (server.address() as AddressInfo).port, + close: () => + new Promise((resolve) => { + inFlight.abort() + server.closeAllConnections() + server.close(() => resolve()) + }), + } +} + +export function parseBridgeCall(body: string): LocalRuntimeBridgeCall { + const parsed = JSON.parse(body) as { path?: unknown; args?: unknown } + if ( + !Array.isArray(parsed.path) || + parsed.path.length === 0 || + !parsed.path.every((segment) => typeof segment === 'string') + ) { + throw new Error('Invalid kody:runtime call path.') + } + return { + path: parsed.path as Array, + args: Array.isArray(parsed.args) ? parsed.args : [], + } +} + +export async function readBodyText(stream: IncomingMessage): Promise { + const chunks: Array = [] + for await (const chunk of stream) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)) + } + return Buffer.concat(chunks).toString('utf8') +} + +export async function readBodyBuffer(stream: IncomingMessage): Promise { + const chunks: Array = [] + for await (const chunk of stream) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)) + } + return Buffer.concat(chunks) +} + +/** Write a Web `Response` onto a Node `ServerResponse` (streams the body). */ +export async function writeWebResponse( + response: ServerResponse, + web: Response, +): Promise { + const headers: Record> = {} + web.headers.forEach((value, name) => { + const existing = headers[name] + if (existing === undefined) { + headers[name] = value + return + } + headers[name] = Array.isArray(existing) + ? [...existing, value] + : [existing, value] + }) + response.writeHead(web.status, web.statusText, headers) + if (!web.body) { + response.end() + return + } + const nodeStream = Readable.fromWeb( + web.body as import('node:stream/web').ReadableStream, + ) + await new Promise((resolve, reject) => { + nodeStream.on('error', reject) + response.on('error', reject) + response.on('finish', () => resolve()) + nodeStream.pipe(response) + }) +} diff --git a/src/local-runtime-source.ts b/src/local-runtime-source.ts index 81724a0..6abe371 100644 --- a/src/local-runtime-source.ts +++ b/src/local-runtime-source.ts @@ -49,21 +49,139 @@ export function createRuntimeFlatAliasSource(): string { return `export * from ${runtime};\nexport { default } from ${runtime};\n` } -export function createLocalRuntimeModuleSource(): string { +/** Workerd-side `kody.gatewayFetchStream` + bridge install helpers (shared shape). */ +export function localGatewayFetchStreamRuntimeSource(options: { + /** Error when the JSON/call bridge is missing. */ + callUnavailableMessage: string +}): string { return ` -import { AsyncLocalStorage } from 'node:async_hooks'; - let __kodyBridge = null; export function __kodyInstallLocalBridge(bridge) { - __kodyBridge = bridge; + // Object form: { call, gatewayFetch }. Legacy function form still works for /call. + __kodyBridge = + typeof bridge === 'function' ? { call: bridge, gatewayFetch: null } : bridge; } function __kodyCall(path, args) { - if (__kodyBridge === null) { - throw new Error('kody:runtime is only available while an execute --local run is in progress.'); + if (__kodyBridge === null || typeof __kodyBridge.call !== 'function') { + throw new Error(${JSON.stringify(options.callUnavailableMessage)}); } - return __kodyBridge(path, args); + return __kodyBridge.call(path, args); +} + +const __kodyGatewayNullBodyStatuses = new Set([101, 103, 204, 205, 304]); +const __kodySecretResolutionHeader = 'x-kody-secret-resolution'; + +/** + * Streaming CapabilityProxy hop (POST /v1/capability-proxy/gateway-fetch via the + * loopback bridge). Returns a real Response — not the JSON gatewayFetch envelope. + * See kentcdodds/kody docs/contributing/contracts/cli-streaming-gateway-fetch.md. + */ +async function gatewayFetchStream(input) { + if (__kodyBridge === null || typeof __kodyBridge.gatewayFetch !== 'function') { + throw new Error( + 'Local execute needs kody.gatewayFetchStream on the host runtime to stream a secret-aware fetch body (multipart, large, or ReadableStream). Upgrade @kodycodes/cli, or buffer the body under 4 MiB for the JSON gatewayFetch hop.', + ); + } + const record = input != null && typeof input === 'object' && !Array.isArray(input) ? input : null; + if (!record) { + throw new Error('gatewayFetchStream requires a single object argument: { request, packageId? }.'); + } + const requestValue = record.request; + if (requestValue == null || typeof requestValue !== 'object' || Array.isArray(requestValue)) { + throw new Error('gatewayFetchStream requires request: { url, method?, headers?, body? }.'); + } + const url = typeof requestValue.url === 'string' ? requestValue.url.trim() : ''; + if (!url) { + throw new Error('gatewayFetchStream request.url must be a non-empty absolute URL.'); + } + const method = + typeof requestValue.method === 'string' && requestValue.method.trim() + ? requestValue.method.trim() + : 'GET'; + const packageId = + typeof record.packageId === 'string' && record.packageId.trim() + ? record.packageId.trim() + : undefined; + const rawHeaders = + requestValue.headers != null && + typeof requestValue.headers === 'object' && + !Array.isArray(requestValue.headers) + ? { ...requestValue.headers } + : {}; + const outboundHeaders = {}; + let contentType = null; + let secretResolution = null; + for (const [key, value] of Object.entries(rawHeaders)) { + if (typeof value !== 'string') continue; + const lower = key.toLowerCase(); + if (lower === 'content-type') { + contentType = value; + continue; + } + if (lower === __kodySecretResolutionHeader) { + secretResolution = value; + continue; + } + outboundHeaders[key] = value; + } + const headers = { + 'X-Kody-Gateway-Url': url, + 'X-Kody-Gateway-Method': method, + }; + if (packageId) headers['X-Kody-Gateway-Package-Id'] = packageId; + if (Object.keys(outboundHeaders).length > 0) { + headers['X-Kody-Gateway-Headers'] = JSON.stringify(outboundHeaders); + } + if (contentType) headers['Content-Type'] = contentType; + if (secretResolution) headers[__kodySecretResolutionHeader] = secretResolution; + const body = Object.hasOwn(requestValue, 'body') ? requestValue.body : null; + const init = { + method: 'POST', + headers, + }; + const upper = method.toUpperCase(); + if (body != null && upper !== 'GET' && upper !== 'HEAD') { + init.body = body; + init.duplex = 'half'; + } + if (requestValue.signal != null) init.signal = requestValue.signal; + if (requestValue.redirect != null) init.redirect = requestValue.redirect; + const response = await __kodyBridge.gatewayFetch(init); + const cacheControl = response.headers.get('cache-control') ?? ''; + const contentTypeHeader = response.headers.get('content-type') ?? ''; + if (response.status === 413 && !contentTypeHeader.includes('application/json')) { + throw new Error( + 'Request body exceeds the Worker upload limit of 104857600 bytes (100 MiB).', + ); + } + if (cacheControl.includes('no-store') && contentTypeHeader.includes('application/json')) { + const text = await response.clone().text(); + try { + const parsed = JSON.parse(text); + if (parsed != null && typeof parsed === 'object' && parsed.error != null) { + const err = parsed.error; + const message = + typeof err === 'string' + ? err + : typeof err?.message === 'string' + ? err.message + : 'gatewayFetchStream failed.'; + throw new Error(message); + } + } catch (error) { + if (!(error instanceof SyntaxError)) throw error; + } + } + if (__kodyGatewayNullBodyStatuses.has(response.status)) { + return new Response(null, { + status: response.status, + statusText: response.statusText, + headers: response.headers, + }); + } + return response; } function __kodyNamespace(path) { @@ -74,6 +192,13 @@ function __kodyNamespace(path) { return undefined; } if (property === 'toString' || property === 'inspect') return () => label; + if ( + path.length === 1 && + path[0] === 'kody' && + property === 'gatewayFetchStream' + ) { + return gatewayFetchStream; + } return __kodyNamespace([...path, property]); }, apply(_target, _thisArg, args) { @@ -81,6 +206,17 @@ function __kodyNamespace(path) { }, }); } +`.trimStart() +} + +export function createLocalRuntimeModuleSource(): string { + return ` +import { AsyncLocalStorage } from 'node:async_hooks'; + +${localGatewayFetchStreamRuntimeSource({ + callUnavailableMessage: + 'kody:runtime is only available while an execute --local run is in progress.', +})} export const kody = __kodyNamespace(['kody']); @@ -313,15 +449,26 @@ export default { } __kodyClaimed = true; const { params } = await request.json(); - __kodyInstallLocalBridge(async (path, args) => { - const response = await env.KODY_BRIDGE.fetch('http://kody-bridge/call', { - method: 'POST', - headers: { 'content-type': 'application/json', [__kodyRunSecretHeader]: secret }, - body: JSON.stringify({ path, args }), - }); - const data = await response.json(); - if (data.error) throw new Error(data.error); - return data.result; + __kodyInstallLocalBridge({ + call: async (path, args) => { + const response = await env.KODY_BRIDGE.fetch('http://kody-bridge/call', { + method: 'POST', + headers: { 'content-type': 'application/json', [__kodyRunSecretHeader]: secret }, + body: JSON.stringify({ path, args }), + }); + const data = await response.json(); + if (data.error) throw new Error(data.error); + return data.result; + }, + gatewayFetch: async (init) => { + const headers = new Headers(init?.headers); + headers.set(__kodyRunSecretHeader, secret); + return await env.KODY_BRIDGE.fetch('http://kody-bridge/gateway-fetch', { + ...init, + method: 'POST', + headers, + }); + }, }); const logs = []; const nativeConsole = globalThis.console; diff --git a/src/mock/mock-supported.ts b/src/mock/mock-supported.ts index 7eb7971..fc4e46e 100644 --- a/src/mock/mock-supported.ts +++ b/src/mock/mock-supported.ts @@ -65,5 +65,6 @@ export function isMockCapabilityPath(pathKey: string): pathKey is MockCapability export const mockCapabilityProxyRoutes = [ 'GET /v1/capability-proxy/session', 'POST /v1/capability-proxy/call', + 'POST /v1/capability-proxy/gateway-fetch', 'POST /v1/local-execute/package-graph', ] as const diff --git a/src/package-app-dev.ts b/src/package-app-dev.ts index d66ca1a..c22b851 100644 --- a/src/package-app-dev.ts +++ b/src/package-app-dev.ts @@ -27,6 +27,11 @@ import { callCapabilityProxy, } from './capability-proxy.js' import { apiTokenEnvVar } from './defaults.js' +import { forwardLocalGatewayFetchStream } from './local-gateway-fetch-forward.js' +import { + startLocalRuntimeBridge, + type LocalRuntimeBridge, +} from './local-runtime-bridge.js' import { createDevBundler, type DevBundle } from './dev/dev-bundler.js' import { parseWorkerdStdoutLine, type DevEvent, type DevRequestTag } from './dev/dev-events.js' import { devErrorHeader, devRequestIdHeader } from './dev/dev-protocol.js' @@ -143,8 +148,6 @@ export type PackageAppDevServer = { close: () => Promise } -type Bridge = { port: number; close: () => Promise } - type Generation = { id: number dir: string @@ -258,9 +261,9 @@ export async function startPackageAppDev( onStatus: options.onStatus, }) - const bridge: Bridge = await startBridge({ + const bridge: LocalRuntimeBridge = await startLocalRuntimeBridge({ runSecret, - forward: async (call, signal) => { + forwardCall: async (call, signal) => { const mocked = tryHandleMockGatewayFetchCall({ gateway: mockFetchGateway, path: call.path, @@ -274,6 +277,15 @@ export async function startPackageAppDev( signal, }) }, + forwardGatewayFetchStream: async ({ request, response, signal }) => { + await forwardLocalGatewayFetchStream({ + request, + response, + signal, + client, + ...(mockFetchGateway ? { mockFetchGateway } : {}), + }) + }, }) const entryPath = resolvePackagePath(options.manifest, options.manifest.app.entry) @@ -823,71 +835,3 @@ function joinUrlPath(base: string, path: string): string { if (!base) return rest return `${base}${rest === '/' ? '' : rest}` || base } - -async function startBridge(input: { - runSecret: string - forward: ( - call: { path: Array; args: Array }, - signal: AbortSignal, - ) => Promise -}): Promise { - const inFlight = new AbortController() - const server: Server = createServer(async (request, response) => { - const reply = (status: number, body: unknown) => { - response.writeHead(status, { 'content-type': 'application/json' }) - response.end(JSON.stringify(body)) - } - if ( - request.method !== 'POST' || - request.url !== '/call' || - request.headers[runSecretHeader] !== input.runSecret - ) { - reply(403, { error: 'Forbidden' }) - return - } - try { - const parsed = JSON.parse(await readBody(request)) as { - path?: unknown - args?: unknown - } - if ( - !Array.isArray(parsed.path) || - parsed.path.length === 0 || - !parsed.path.every((segment) => typeof segment === 'string') - ) { - throw new Error('Invalid kody:runtime call path.') - } - const result = await input.forward( - { - path: parsed.path as Array, - args: Array.isArray(parsed.args) ? parsed.args : [], - }, - inFlight.signal, - ) - reply(200, { result }) - } catch (error) { - reply(200, { error: redact(error instanceof Error ? error.message : String(error)) }) - } - }) - await new Promise((resolve, reject) => { - server.once('error', reject) - server.listen(0, '127.0.0.1', () => resolve()) - }) - return { - port: (server.address() as AddressInfo).port, - close: () => - new Promise((resolve) => { - inFlight.abort() - server.closeAllConnections() - server.close(() => resolve()) - }), - } -} - -async function readBody(stream: IncomingMessage): Promise { - const chunks: Array = [] - for await (const chunk of stream) { - chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)) - } - return Buffer.concat(chunks).toString('utf8') -} diff --git a/src/package-local-runtime-source.ts b/src/package-local-runtime-source.ts index 0b60dea..2b06f6d 100644 --- a/src/package-local-runtime-source.ts +++ b/src/package-local-runtime-source.ts @@ -9,6 +9,7 @@ import { kodyRuntimeStorageSymbolDescription, localExecuteCompatibilityDate, localExecuteCompatibilityFlags, + localGatewayFetchStreamRuntimeSource, localWorkerModuleNames, runSecretEnvVar, runSecretHeader, @@ -87,34 +88,10 @@ export function packageStorage() { return ` import { AsyncLocalStorage } from 'node:async_hooks'; -let __kodyBridge = null; - -export function __kodyInstallLocalBridge(bridge) { - __kodyBridge = bridge; -} - -function __kodyCall(path, args) { - if (__kodyBridge === null) { - throw new Error('kody:runtime is only available while a local package run is in progress.'); - } - return __kodyBridge(path, args); -} - -function __kodyNamespace(path) { - const label = '[KodyRuntime:' + path.join('.') + ']'; - return new Proxy(function () {}, { - get(_target, property) { - if (typeof property === 'symbol' || property === 'then' || property === 'toJSON') { - return undefined; - } - if (property === 'toString' || property === 'inspect') return () => label; - return __kodyNamespace([...path, property]); - }, - apply(_target, _thisArg, args) { - return __kodyCall(path, args); - }, - }); -} +${localGatewayFetchStreamRuntimeSource({ + callUnavailableMessage: + 'kody:runtime is only available while a local package run is in progress.', +})} export const kody = __kodyNamespace(['kody']); @@ -344,15 +321,26 @@ export default { } __kodyClaimed = true; const { params } = await request.json(); - __kodyInstallLocalBridge(async (path, args) => { - const response = await env.KODY_BRIDGE.fetch('http://kody-bridge/call', { - method: 'POST', - headers: { 'content-type': 'application/json', [__kodyRunSecretHeader]: secret }, - body: JSON.stringify({ path, args }), - }); - const data = await response.json(); - if (data.error) throw new Error(data.error); - return data.result; + __kodyInstallLocalBridge({ + call: async (path, args) => { + const response = await env.KODY_BRIDGE.fetch('http://kody-bridge/call', { + method: 'POST', + headers: { 'content-type': 'application/json', [__kodyRunSecretHeader]: secret }, + body: JSON.stringify({ path, args }), + }); + const data = await response.json(); + if (data.error) throw new Error(data.error); + return data.result; + }, + gatewayFetch: async (init) => { + const headers = new Headers(init?.headers); + headers.set(__kodyRunSecretHeader, secret); + return await env.KODY_BRIDGE.fetch('http://kody-bridge/gateway-fetch', { + ...init, + method: 'POST', + headers, + }); + }, }); const logs = []; const nativeConsole = globalThis.console; @@ -471,15 +459,26 @@ export default { if (!secret || request.headers.get(__kodyRunSecretHeader) !== secret) { return new Response('Forbidden', { status: 403 }); } - __kodyInstallLocalBridge(async (path, args) => { - const response = await env.KODY_BRIDGE.fetch('http://kody-bridge/call', { - method: 'POST', - headers: { 'content-type': 'application/json', [__kodyRunSecretHeader]: secret }, - body: JSON.stringify({ path, args }), - }); - const data = await response.json(); - if (data.error) throw new Error(data.error); - return data.result; + __kodyInstallLocalBridge({ + call: async (path, args) => { + const response = await env.KODY_BRIDGE.fetch('http://kody-bridge/call', { + method: 'POST', + headers: { 'content-type': 'application/json', [__kodyRunSecretHeader]: secret }, + body: JSON.stringify({ path, args }), + }); + const data = await response.json(); + if (data.error) throw new Error(data.error); + return data.result; + }, + gatewayFetch: async (init) => { + const headers = new Headers(init?.headers); + headers.set(__kodyRunSecretHeader, secret); + return await env.KODY_BRIDGE.fetch('http://kody-bridge/gateway-fetch', { + ...init, + method: 'POST', + headers, + }); + }, }); const url = new URL(request.url); const tag = { diff --git a/test/local-execute.test.ts b/test/local-execute.test.ts index 374f2c7..3af098e 100644 --- a/test/local-execute.test.ts +++ b/test/local-execute.test.ts @@ -72,23 +72,45 @@ let server: Server before(async () => { server = createServer(async (request, response) => { - let raw = '' - for await (const chunk of request) raw += chunk - const body = raw ? JSON.parse(raw) : null + const send = (status: number, payload: unknown) => { + response.writeHead(status, { 'content-type': 'application/json' }) + response.end(JSON.stringify(payload)) + } + const isGatewayFetch = request.url === '/v1/capability-proxy/gateway-fetch' + const chunks: Array = [] + for await (const chunk of request) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)) + } + const raw = Buffer.concat(chunks).toString('utf8') + const body = isGatewayFetch + ? { + gatewayUrl: request.headers['x-kody-gateway-url'] ?? null, + gatewayMethod: request.headers['x-kody-gateway-method'] ?? null, + gatewayHeaders: request.headers['x-kody-gateway-headers'] ?? null, + contentType: request.headers['content-type'] ?? null, + rawBody: raw, + } + : raw + ? JSON.parse(raw) + : null proxyRequests.push({ method: request.method ?? '', url: request.url ?? '', authorization: request.headers.authorization ?? null, body, }) - const send = (status: number, payload: unknown) => { - response.writeHead(status, { 'content-type': 'application/json' }) - response.end(JSON.stringify(payload)) - } if (request.headers.authorization !== `Bearer ${goodToken}`) { send(401, { error: { code: 'invalid_token', message: 'unknown token' } }) return } + if (isGatewayFetch) { + response.writeHead(201, { + 'content-type': 'text/plain', + 'x-upstream': 'streamed', + }) + response.end(`echo:${raw}`) + return + } if (request.url === '/v1/capability-proxy/session') { send(sessionStatus.status, sessionStatus.body) return @@ -476,6 +498,85 @@ test('createLocalRuntimeModuleSource exports credential and credentials for ad h ) }) +test('createLocalRuntimeModuleSource implements kody.gatewayFetchStream as a host method', () => { + const source = createLocalRuntimeModuleSource() + assert.match(source, /async function gatewayFetchStream/) + assert.match(source, /property === 'gatewayFetchStream'/) + assert.match(source, /X-Kody-Gateway-Url/) + assert.match(createLocalEntrySource(), /gateway-fetch/) +}) + +test( + 'runLocalExecute kody.gatewayFetchStream hops POST /v1/capability-proxy/gateway-fetch and returns a Response', + { timeout: 180_000 }, + async () => { + reset() + const code = ` +import { kody } from 'kody:runtime' +export default async function main() { + const response = await kody.gatewayFetchStream({ + packageId: 'pkg-stream', + request: { + url: 'https://api.example.com/upload', + method: 'POST', + headers: { + authorization: 'Bearer {{secret:demoToken|scope=user}}', + 'content-type': 'text/plain', + }, + body: 'stream-body-bytes', + }, + }) + return { + status: response.status, + text: await response.text(), + upstream: response.headers.get('x-upstream'), + typeofGatewayFetchStream: typeof kody.gatewayFetchStream, + } +} +` + const result = await runLocalExecute({ + code, + token: goodToken, + apiUrl, + }) + assert.equal(result.isError, false, JSON.stringify(result)) + const structured = result.structuredContent as { + result: { + status: number + text: string + upstream: string | null + typeofGatewayFetchStream: string + } + } + assert.equal(structured.result.typeofGatewayFetchStream, 'function') + assert.equal(structured.result.status, 201) + assert.equal(structured.result.text, 'echo:stream-body-bytes') + assert.equal(structured.result.upstream, 'streamed') + const hop = proxyRequests.find((entry) => entry.url === '/v1/capability-proxy/gateway-fetch') + assert.ok(hop) + assert.equal(hop?.authorization, `Bearer ${goodToken}`) + assert.deepEqual(hop?.body, { + gatewayUrl: 'https://api.example.com/upload', + gatewayMethod: 'POST', + gatewayHeaders: JSON.stringify({ + authorization: 'Bearer {{secret:demoToken|scope=user}}', + }), + contentType: 'text/plain', + rawBody: 'stream-body-bytes', + }) + assert.equal( + proxyRequests.some( + (entry) => + entry.url === '/v1/capability-proxy/call' && + (entry.body as { path?: Array } | null)?.path?.join('.') === + 'kody.gatewayFetchStream', + ), + false, + 'gatewayFetchStream must not use the JSON /call hop', + ) + }, +) + test( 'runLocalExecute exposes credential() on kody:runtime and hops credentialClientOpen without secret values', { timeout: 180_000 }, diff --git a/test/local-gateway-fetch-stream.test.ts b/test/local-gateway-fetch-stream.test.ts new file mode 100644 index 0000000..5490167 --- /dev/null +++ b/test/local-gateway-fetch-stream.test.ts @@ -0,0 +1,211 @@ +import assert from 'node:assert/strict' +import { createServer } from 'node:http' +import type { AddressInfo } from 'node:net' +import { Readable } from 'node:stream' +import { test } from 'node:test' +import { + callCapabilityProxyGatewayFetchStream, + gatewayFetchHeadersHeaderName, + gatewayFetchMethodHeaderName, + gatewayFetchUrlHeaderName, +} from '../src/capability-proxy.js' +import { forwardLocalGatewayFetchStream } from '../src/local-gateway-fetch-forward.js' +import { + localBridgeGatewayFetchPath, + startLocalRuntimeBridge, + writeWebResponse, +} from '../src/local-runtime-bridge.js' +import { runSecretHeader } from '../src/local-runtime-source.js' +import { createMockFetchGateway } from '../src/mock/mock-fetch-gateway.js' + +test('callCapabilityProxyGatewayFetchStream posts metadata headers and streams the body', async () => { + const chunks: Array = [] + let requestHeaders: Record | undefined> = {} + const server = createServer((request, response) => { + requestHeaders = { ...request.headers } + request.on('data', (chunk: Buffer) => chunks.push(chunk)) + request.on('end', () => { + response.writeHead(201, { + 'content-type': 'text/plain', + 'x-upstream': '1', + }) + response.end('upstream-ok') + }) + }) + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)) + const { port } = server.address() as AddressInfo + try { + const body = Readable.from([Buffer.from('hello-stream')]) + const response = await callCapabilityProxyGatewayFetchStream({ + apiUrl: `http://127.0.0.1:${port}`, + token: 'kody_at_test', + gateway: { + url: 'https://example.com/upload', + method: 'POST', + packageId: 'pkg-1', + headers: { authorization: 'Bearer {{secret:demo}}' }, + contentType: 'text/plain', + secretResolution: 'on', + }, + body, + contentLength: String(Buffer.byteLength('hello-stream')), + }) + assert.equal(response.status, 201) + assert.equal(await response.text(), 'upstream-ok') + assert.equal(requestHeaders[gatewayFetchUrlHeaderName.toLowerCase()], 'https://example.com/upload') + assert.equal(requestHeaders[gatewayFetchMethodHeaderName.toLowerCase()], 'POST') + assert.equal(requestHeaders['content-type'], 'text/plain') + assert.equal( + requestHeaders[gatewayFetchHeadersHeaderName.toLowerCase()], + JSON.stringify({ authorization: 'Bearer {{secret:demo}}' }), + ) + assert.equal(Buffer.concat(chunks).toString('utf8'), 'hello-stream') + assert.equal(requestHeaders.authorization, 'Bearer kody_at_test') + } finally { + server.close() + } +}) + +test('local runtime bridge streams gateway-fetch to the mock MSW gateway', async () => { + const { http, HttpResponse } = await import('msw') + const gateway = createMockFetchGateway({ + handlers: [ + http.post('https://api.example.com/upload', async ({ request }) => { + const text = await request.text() + return HttpResponse.json({ ok: true, echo: text }, { status: 202 }) + }), + ], + }) + const bridge = await startLocalRuntimeBridge({ + runSecret: 'secret', + forwardCall: async () => { + throw new Error('JSON /call should not be used for gatewayFetchStream') + }, + forwardGatewayFetchStream: async ({ request, response, signal }) => { + await forwardLocalGatewayFetchStream({ + request, + response, + signal, + client: { + apiUrl: 'http://127.0.0.1:9', + token: 'kody_at_unused', + }, + mockFetchGateway: gateway, + }) + }, + }) + try { + const response = await fetch( + `http://127.0.0.1:${bridge.port}${localBridgeGatewayFetchPath}`, + { + method: 'POST', + headers: { + [runSecretHeader]: 'secret', + [gatewayFetchUrlHeaderName]: 'https://api.example.com/upload', + [gatewayFetchMethodHeaderName]: 'POST', + 'content-type': 'text/plain', + }, + body: 'payload-bytes', + duplex: 'half', + } as RequestInit, + ) + assert.equal(response.status, 202) + assert.deepEqual(await response.json(), { ok: true, echo: 'payload-bytes' }) + } finally { + await bridge.close() + } +}) + +test('bridge gateway-fetch failures return Open API error envelopes (no-store)', async () => { + const bridge = await startLocalRuntimeBridge({ + runSecret: 'secret', + forwardCall: async () => null, + forwardGatewayFetchStream: async () => { + throw new Error('upstream blew up') + }, + }) + try { + const response = await fetch( + `http://127.0.0.1:${bridge.port}${localBridgeGatewayFetchPath}`, + { + method: 'POST', + headers: { + [runSecretHeader]: 'secret', + [gatewayFetchUrlHeaderName]: 'https://api.example.com/upload', + }, + body: 'x', + duplex: 'half', + } as RequestInit, + ) + assert.equal(response.status, 502) + assert.match(response.headers.get('cache-control') ?? '', /no-store/) + assert.deepEqual(await response.json(), { + error: { code: 'gateway_fetch_failed', message: 'upstream blew up' }, + }) + } finally { + await bridge.close() + } +}) + +test('mock gateway-fetch null-body statuses do not throw', async () => { + const gateway = createMockFetchGateway({ + handlers: [], + passthroughAll: true, + fetchFn: (async () => + new Response(null, { status: 204, statusText: 'No Content' })) as typeof fetch, + }) + const bridge = await startLocalRuntimeBridge({ + runSecret: 'secret', + forwardCall: async () => null, + forwardGatewayFetchStream: async ({ request, response, signal }) => { + await forwardLocalGatewayFetchStream({ + request, + response, + signal, + client: { apiUrl: 'http://127.0.0.1:9', token: 'kody_at_unused' }, + mockFetchGateway: gateway, + }) + }, + }) + try { + const response = await fetch( + `http://127.0.0.1:${bridge.port}${localBridgeGatewayFetchPath}`, + { + method: 'POST', + headers: { + [runSecretHeader]: 'secret', + [gatewayFetchUrlHeaderName]: 'https://api.example.com/empty', + [gatewayFetchMethodHeaderName]: 'POST', + }, + body: '', + duplex: 'half', + } as RequestInit, + ) + assert.equal(response.status, 204) + assert.equal(await response.text(), '') + } finally { + await bridge.close() + } +}) + +test('writeWebResponse pipes status, headers, and body', async () => { + const server = createServer(async (_request, response) => { + await writeWebResponse( + response, + new Response('piped', { + status: 203, + headers: { 'x-test': 'yes', 'content-type': 'text/plain' }, + }), + ) + }) + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)) + const { port } = server.address() as AddressInfo + try { + const response = await fetch(`http://127.0.0.1:${port}/`) + assert.equal(response.status, 203) + assert.equal(response.headers.get('x-test'), 'yes') + assert.equal(await response.text(), 'piped') + } finally { + server.close() + } +}) diff --git a/test/local-host-runtime-surface.test.ts b/test/local-host-runtime-surface.test.ts new file mode 100644 index 0000000..0300097 --- /dev/null +++ b/test/local-host-runtime-surface.test.ts @@ -0,0 +1,110 @@ +import assert from 'node:assert/strict' +import { readFileSync } from 'node:fs' +import { dirname, join } from 'node:path' +import { test } from 'node:test' +import { fileURLToPath } from 'node:url' +import { + hostedCapabilityProxySpecialFunctions, + hostedLocalExecuteRuntimeSurface, + hostedLocalHostSpecialFunctions, + localHostImplementedRuntimeMarkers, +} from '../src/local-host-runtime-surface.js' +import { + createLocalEntrySource, + createLocalRuntimeModuleSource, + localGatewayFetchStreamRuntimeSource, +} from '../src/local-runtime-source.js' +import { createPackageLocalRuntimeModuleSource } from '../src/package-local-runtime-source.js' +import { localBridgeGatewayFetchPath } from '../src/local-runtime-bridge.js' +import { + capabilityProxyGatewayFetchPath, + gatewayFetchHeadersHeaderName, + gatewayFetchMethodHeaderName, + gatewayFetchPackageIdHeaderName, + gatewayFetchUrlHeaderName, +} from '../src/capability-proxy.js' + +const srcDir = join(dirname(fileURLToPath(import.meta.url)), '../src') + +test('hosted local-execute runtime surface lists gatewayFetchStream as a special host method', () => { + assert.deepEqual([...hostedLocalHostSpecialFunctions], ['kody.gatewayFetchStream']) + assert.ok(hostedLocalExecuteRuntimeSurface.includes('kody.gatewayFetchStream')) + assert.ok(hostedCapabilityProxySpecialFunctions.includes('kody.gatewayFetch')) + assert.ok(hostedCapabilityProxySpecialFunctions.includes('kody.authenticatedFetch')) + assert.ok(hostedCapabilityProxySpecialFunctions.includes('kody.oauthClientCredentials')) +}) + +test('CLI --local runtime source implements every hosted special host marker (drift guard)', () => { + const localRuntime = createLocalRuntimeModuleSource() + const packageRuntime = createPackageLocalRuntimeModuleSource({ + packageContext: { + packageId: 'pkg', + kodyId: 'demo', + appBasePath: '', + hostedUrl: 'http://127.0.0.1/', + assetBasePath: '/_assets', + clientModuleUrl: null, + }, + mockStorage: true, + }) + const entry = createLocalEntrySource() + const bridgeSource = readFileSync(join(srcDir, 'local-runtime-bridge.ts'), 'utf8') + const capabilitySource = readFileSync(join(srcDir, 'capability-proxy.ts'), 'utf8') + const forwardSource = readFileSync(join(srcDir, 'local-gateway-fetch-forward.ts'), 'utf8') + + for (const marker of localHostImplementedRuntimeMarkers) { + assert.match( + localRuntime, + new RegExp(escapeRegExp(marker.sourceMarker)), + `local runtime missing ${marker.id} marker ${marker.sourceMarker}`, + ) + assert.match( + packageRuntime, + new RegExp(escapeRegExp(marker.sourceMarker)), + `package-local runtime missing ${marker.id} marker ${marker.sourceMarker}`, + ) + if (marker.bridgeMarker === '/gateway-fetch') { + assert.match(entry, /gateway-fetch/) + assert.match(bridgeSource, /gateway-fetch/) + assert.equal(localBridgeGatewayFetchPath, '/gateway-fetch') + assert.match(capabilitySource, /capability-proxy\/gateway-fetch/) + assert.equal(capabilityProxyGatewayFetchPath, 'v1/capability-proxy/gateway-fetch') + assert.match(forwardSource, /callCapabilityProxyGatewayFetchStream/) + } + if (marker.bridgeMarker === '/call') { + assert.match(entry, /kody-bridge\/call/) + assert.match(bridgeSource, /\/call/) + } + } +}) + +test('gatewayFetchStream is a real host method on the kody Proxy (not a JSON /call hop)', () => { + const source = localGatewayFetchStreamRuntimeSource({ + callUnavailableMessage: 'unavailable', + }) + assert.match(source, /async function gatewayFetchStream/) + assert.match(source, /property === 'gatewayFetchStream'/) + assert.match(source, /X-Kody-Gateway-Url/) + assert.match(source, /init\.duplex = 'half'/) + assert.doesNotMatch( + source, + /gatewayFetchStream[\s\S]{0,80}__kodyCall\(\['kody', 'gatewayFetchStream'\]/, + ) + assert.equal(gatewayFetchUrlHeaderName, 'X-Kody-Gateway-Url') + assert.equal(gatewayFetchMethodHeaderName, 'X-Kody-Gateway-Method') + assert.equal(gatewayFetchPackageIdHeaderName, 'X-Kody-Gateway-Package-Id') + assert.equal(gatewayFetchHeadersHeaderName, 'X-Kody-Gateway-Headers') +}) + +test('hosted surface list stays alphabetically grouped and unique (edit-guard)', () => { + const ids = [...hostedLocalExecuteRuntimeSurface] + assert.equal(ids.length, new Set(ids).size, 'duplicate runtime surface ids') + // Special host methods that are not Proxy JSON hops must stay listed. + for (const id of hostedLocalHostSpecialFunctions) { + assert.ok(ids.includes(id), `missing special host function ${id}`) + } +}) + +function escapeRegExp(value: string): string { + return value.replace(/[.*+?^${}()|[\]\\]/g, '\\$&') +}