diff --git a/internal/dev-entrypoint/cloud-web-proxy.mjs b/internal/dev-entrypoint/cloud-web-proxy.mjs index 8ccfb6e5..8bb66c7a 100644 --- a/internal/dev-entrypoint/cloud-web-proxy.mjs +++ b/internal/dev-entrypoint/cloud-web-proxy.mjs @@ -76,6 +76,14 @@ export function proxyCloudApiRequest({ let timeout = null; let streamTransformEnded = false; + const endStreamTransform = () => { + if (!streamTransformEnded && streamTransform?.end) { + streamTransformEnded = true; + return streamTransform.end(); + } + return null; + }; + const settle = (callback, value) => { if (settled) return; settled = true; @@ -107,10 +115,7 @@ export function proxyCloudApiRequest({ writeStreamOutput(response, streamTransform?.write ? streamTransform.write(chunk) : chunk); }); upstreamResponse.on("end", () => { - if (!streamTransformEnded && streamTransform?.end) { - streamTransformEnded = true; - writeStreamOutput(response, streamTransform.end()); - } + writeStreamOutput(response, endStreamTransform()); response.end(); settle(resolve, { statusCode: upstreamStatusCode, streaming: true, streamTransformStats: streamTransform?.stats }); }); @@ -123,14 +128,15 @@ export function proxyCloudApiRequest({ settle(reject, error); }); upstreamResponse.on("close", () => { - if (!streamTransformEnded && streamTransform?.end) { - streamTransformEnded = true; - writeStreamOutput(response, streamTransform.end()); - } + writeStreamOutput(response, endStreamTransform()); if (!response.writableEnded) response.end(); settle(resolve, { statusCode: upstreamStatusCode, streaming: true, streamTransformStats: streamTransform?.stats }); }); - response.on("close", () => upstream.destroy()); + response.on("close", () => { + endStreamTransform(); + upstream.destroy(); + settle(resolve, { statusCode: upstreamStatusCode, streaming: true, errorCode: "client_closed", streamTransformStats: streamTransform?.stats }); + }); return; } diff --git a/internal/dev-entrypoint/cloud-web-proxy.test.mjs b/internal/dev-entrypoint/cloud-web-proxy.test.mjs index 7ff21ff9..90effc2a 100644 --- a/internal/dev-entrypoint/cloud-web-proxy.test.mjs +++ b/internal/dev-entrypoint/cloud-web-proxy.test.mjs @@ -144,6 +144,62 @@ test("cloud web proxy does not reject after SSE headers are already sent", async } }); +test("cloud web proxy resolves stream transform stats when the client closes SSE", async () => { + const upstream = createServer((request, response) => { + request.resume(); + response.writeHead(200, { "content-type": "text/event-stream; charset=utf-8" }); + response.write('data: {"directory":"/workspace","payload":{"type":"session.status"}}\n\n'); + }); + await listen(upstream); + + let proxyResult = null; + const transform = { + stats: { enabled: true, writes: 0, ended: false }, + write(chunk) { + this.stats.writes += 1; + return chunk; + }, + end() { + this.stats.ended = true; + return ""; + } + }; + const proxy = createServer((request, response) => { + const url = new URL(request.url || "/", "http://cloud-web.local"); + proxyCloudApiRequest({ + target: new URL(url.pathname + url.search, serverUrl(upstream)), + request, + response, + timeoutMs: 1000, + forceStream: true, + streamTransform: transform + }).then((result) => { + proxyResult = result; + }).catch((error) => response.destroy(error)); + }); + await listen(proxy); + + try { + const response = await fetch(`${serverUrl(proxy)}/global/event`, { + headers: { accept: "text/event-stream" } + }); + assert.equal(response.status, 200); + const reader = response.body.getReader(); + const first = await reader.read(); + assert.equal(first.done, false); + await reader.cancel(); + await waitFor(() => proxyResult !== null, "expected proxy result after client close"); + + assert.equal(proxyResult.statusCode, 200); + assert.equal(proxyResult.streaming, true); + assert.equal(proxyResult.errorCode, "client_closed"); + assert.deepEqual(proxyResult.streamTransformStats, { enabled: true, writes: 1, ended: true }); + } finally { + await close(proxy); + await close(upstream); + } +}); + test("cloud web proxy buffers normal JSON and sets content-length", async () => { const upstream = createServer((request, response) => { request.resume(); @@ -189,6 +245,15 @@ function close(server) { }); } +async function waitFor(predicate, message, { timeoutMs = 500, intervalMs = 10 } = {}) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) return; + await new Promise((resolve) => setTimeout(resolve, intervalMs)); + } + assert.fail(message); +} + function serverUrl(server) { const address = server.address(); return `http://127.0.0.1:${address.port}`; diff --git a/internal/dev-entrypoint/cloud-web-runtime.test.mjs b/internal/dev-entrypoint/cloud-web-runtime.test.mjs index ff4a24b2..573a9431 100644 --- a/internal/dev-entrypoint/cloud-web-runtime.test.mjs +++ b/internal/dev-entrypoint/cloud-web-runtime.test.mjs @@ -825,6 +825,88 @@ test("cloud web OpenCode proxy emits bounded OTLP span", async () => { } }); +test("cloud web OpenCode event stream emits rewrite OTLP stats when the client closes", async () => { + const otelBodies = []; + const otel = createServer(async (request, response) => { + let body = ""; + for await (const chunk of request) body += chunk; + otelBodies.push(JSON.parse(body)); + response.writeHead(200, { "content-type": "application/json" }); + response.end("{}\n"); + }); + await listen(otel); + const restoreEnv = withEnv({ OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: `${serverUrl(otel)}/v1/traces` }); + + const cloudApi = createServer((request, response) => { + request.resume(); + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ authenticated: request.headers.cookie === "hwlab_session=session-a" })); + }); + const opencodeResponses = []; + const opencode = createServer((request, response) => { + request.resume(); + opencodeResponses.push(response); + response.writeHead(200, { "content-type": "text/event-stream; charset=utf-8" }); + response.write('data: {"directory":"/workspace","payload":{"type":"session.status","properties":{"sessionID":"ses_1"}}}\n\n'); + }); + await listen(cloudApi); + await listen(opencode); + + const cloudWeb = createCloudWebServer({ + serviceId: "hwlab-cloud-web", + roots: [], + cloudApiBaseUrl: serverUrl(cloudApi), + cloudApiProxyTimeoutMs: 1000, + opencodeUpstreamUrl: serverUrl(opencode), + opencodeProxyHost: "127.0.0.1", + opencodeProxyTimeoutMs: 1000, + opencodeUsername: "oc_user", + opencodePassword: "oc_password", + opencodeEventDirectoryRewrite: { from: "/workspace", to: "/" }, + healthPayload: () => ({ status: "ok" }), + sendJson(response, statusCode, body) { + const payload = JSON.stringify(body); + response.writeHead(statusCode, { "content-type": "application/json", "content-length": Buffer.byteLength(payload) }); + response.end(payload); + } + }); + await listen(cloudWeb); + + try { + const response = await fetch(`${serverUrl(cloudWeb)}/global/event`, { + headers: { accept: "text/event-stream", cookie: "hwlab_session=session-a" } + }); + assert.equal(response.status, 200); + const reader = response.body.getReader(); + const first = await reader.read(); + assert.equal(first.done, false); + const text = new TextDecoder().decode(first.value); + assert.match(text, /"directory":"\/"/u); + await reader.cancel(); + + await waitFor(() => otelBodies.length > 0, "expected OpenCode event stream OTel span"); + const spans = otelBodies.flatMap((body) => body.resourceSpans?.flatMap((resourceSpan) => resourceSpan.scopeSpans?.flatMap((scopeSpan) => scopeSpan.spans ?? []) ?? []) ?? []); + const span = spans.find((item) => item.name === "opencode.proxy.request"); + assert.ok(span, "expected opencode.proxy.request span"); + const attrs = new Map((span.attributes ?? []).map((entry) => [entry.key, entry.value?.stringValue ?? entry.value?.intValue ?? entry.value?.boolValue])); + assert.equal(attrs.get("http.route"), "/global/event"); + assert.equal(attrs.get("opencode.proxy.streaming"), true); + assert.equal(attrs.get("opencode.proxy.sse.directory_rewrite_enabled"), true); + assert.equal(attrs.get("opencode.proxy.sse.directory_rewrite_from"), "/workspace"); + assert.equal(attrs.get("opencode.proxy.sse.directory_rewrite_to"), "/"); + assert.equal(attrs.get("opencode.proxy.sse.directory_rewrite_data_lines"), "1"); + assert.equal(attrs.get("opencode.proxy.sse.directory_rewrite_events"), "1"); + assert.equal(attrs.get("opencode.proxy.sse.directory_rewrite_json_errors"), "0"); + } finally { + for (const response of opencodeResponses) response.destroy(); + restoreEnv(); + await close(cloudWeb); + await close(opencode); + await close(cloudApi); + await close(otel); + } +}); + test("cloud web OpenCode proxy accepts short-lived tickets minted by the shell", async () => { const restoreEnv = withEnv({ HWLAB_CLOUD_WEB_DISPLAY_TIME_ZONE: "Asia/Shanghai",