mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: rewrite opencode event directory for live ui

Browse Source
This commit is contained in:
UniDesk Codex
2026-06-30 14:17:03 +08:00
parent ffb5645f3c
commit 573eb1de90
4 changed files with 165 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/deploy.yaml
+2
View File
@@ -585,6 +585,8 @@ lanes:
HWLAB_CLOUD_WEB_OPENCODE_UPSTREAM_URL: http://opencode-server.hwlab-v03.svc.cluster.local:4096
HWLAB_CLOUD_WEB_OPENCODE_USERNAME: secretRef:hwlab-opencode-server-auth/username
HWLAB_CLOUD_WEB_OPENCODE_PASSWORD: secretRef:hwlab-opencode-server-auth/password
HWLAB_CLOUD_WEB_OPENCODE_EVENT_DIRECTORY_FROM: /workspace
HWLAB_CLOUD_WEB_OPENCODE_EVENT_DIRECTORY_TO: /
OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: http://otel-collector.platform-infra.svc.cluster.local:4318/v1/traces
OTEL_SERVICE_NAME: hwlab-cloud-web
observable: true
internal/dev-entrypoint/cloud-web-proxy.mjs
+23 -4
View File
@@ -67,12 +67,14 @@ export function proxyCloudApiRequest({
body = "",
timeoutMs,
forceStream = false,
extraResponseHeaders = {}
extraResponseHeaders = {},
streamTransform = null
}) {
return new Promise((resolve, reject) => {
let timedOut = false;
let settled = false;
let timeout = null;
let streamTransformEnded = false;
const settle = (callback, value) => {
if (settled) return;
@@ -102,11 +104,15 @@ export function proxyCloudApiRequest({
response.flushHeaders?.();
upstreamResponse.on("data", (chunk) => {
armTimeout();
response.write(chunk);
writeStreamOutput(response, streamTransform?.write ? streamTransform.write(chunk) : chunk);
});
upstreamResponse.on("end", () => {
if (!streamTransformEnded && streamTransform?.end) {
streamTransformEnded = true;
writeStreamOutput(response, streamTransform.end());
}
response.end();
settle(resolve, { statusCode: upstreamStatusCode, streaming: true });
settle(resolve, { statusCode: upstreamStatusCode, streaming: true, streamTransformStats: streamTransform?.stats });
});
upstreamResponse.on("error", (error) => {
if (response.headersSent) {
@@ -117,8 +123,12 @@ export function proxyCloudApiRequest({
settle(reject, error);
});
upstreamResponse.on("close", () => {
if (!streamTransformEnded && streamTransform?.end) {
streamTransformEnded = true;
writeStreamOutput(response, streamTransform.end());
}
if (!response.writableEnded) response.end();
settle(resolve, { statusCode: upstreamStatusCode, streaming: true });
settle(resolve, { statusCode: upstreamStatusCode, streaming: true, streamTransformStats: streamTransform?.stats });
});
response.on("close", () => upstream.destroy());
return;
@@ -157,6 +167,15 @@ export function proxyCloudApiRequest({
});
}
function writeStreamOutput(response, output) {
if (output === undefined || output === null || output === "") return;
if (Array.isArray(output)) {
for (const item of output) writeStreamOutput(response, item);
return;
}
response.write(output);
}
function mergeResponseHeaders(baseHeaders = {}, extraHeaders = {}) {
const result = { ...baseHeaders };
for (const [key, value] of Object.entries(extraHeaders)) {
internal/dev-entrypoint/cloud-web-runtime.mjs
+88 -6
View File
@@ -63,7 +63,8 @@ export function createCloudWebServer({
max: 2400000
}),
opencodeUsername = optionalRuntimeConfigEnv("HWLAB_CLOUD_WEB_OPENCODE_USERNAME") || "",
opencodePassword = optionalRuntimeConfigEnv("HWLAB_CLOUD_WEB_OPENCODE_PASSWORD") || ""
opencodePassword = optionalRuntimeConfigEnv("HWLAB_CLOUD_WEB_OPENCODE_PASSWORD") || "",
opencodeEventDirectoryRewrite = opencodeEventDirectoryRewriteFromEnv()
}) {
return createServer(async (request, response) => {
attachRequestErrorHandlers({ request, response, serviceId });
@@ -78,7 +79,7 @@ export function createCloudWebServer({
return;
}
if (isOpencodeProxyRequest(request, url, opencodeProxyHost)) {
await proxyOpencodeRequest({ request, response, url, cloudApiBaseUrl, cloudApiProxyTimeoutMs, opencodeUpstreamUrl, opencodeProxyTimeoutMs, opencodeUsername, opencodePassword, serviceId, sendJson });
await proxyOpencodeRequest({ request, response, url, cloudApiBaseUrl, cloudApiProxyTimeoutMs, opencodeUpstreamUrl, opencodeProxyTimeoutMs, opencodeUsername, opencodePassword, opencodeEventDirectoryRewrite, serviceId, sendJson });
return;
}
if (await handleCloudWebAuth({ request, response, url, cloudApiBaseUrl, cloudApiProxyTimeoutMs, serviceId, sendJson })) {
@@ -304,6 +305,71 @@ function opencodeProxyHostFromEnv() {
}
}
function opencodeEventDirectoryRewriteFromEnv() {
const from = optionalRuntimeConfigEnv("HWLAB_CLOUD_WEB_OPENCODE_EVENT_DIRECTORY_FROM");
const to = optionalRuntimeConfigEnv("HWLAB_CLOUD_WEB_OPENCODE_EVENT_DIRECTORY_TO");
if (!from || !to || from === to) return null;
return { from, to };
}
function opencodeEventDirectoryTransformForTarget(target, rewrite) {
if (!rewrite || target?.pathname !== "/global/event") return null;
return opencodeEventDirectoryRewriteTransform(rewrite);
}
function opencodeEventDirectoryRewriteTransform(rewrite) {
const stats = {
enabled: true,
from: rewrite.from,
to: rewrite.to,
dataLines: 0,
rewrittenEvents: 0,
jsonErrors: 0
};
let pending = "";
return {
stats,
write(chunk) {
pending += Buffer.isBuffer(chunk) ? chunk.toString("utf8") : String(chunk ?? "");
const output = [];
let newlineIndex = pending.indexOf("\n");
while (newlineIndex >= 0) {
const line = pending.slice(0, newlineIndex + 1);
pending = pending.slice(newlineIndex + 1);
output.push(rewriteOpencodeEventDirectoryLine(line, rewrite, stats));
newlineIndex = pending.indexOf("\n");
}
return output;
},
end() {
if (!pending) return "";
const line = pending;
pending = "";
return rewriteOpencodeEventDirectoryLine(line, rewrite, stats);
}
};
}
function rewriteOpencodeEventDirectoryLine(line, rewrite, stats) {
const match = String(line).match(/^(data:\s*)(.*?)(\r?\n)?$/u);
if (!match) return line;
const [, prefix, payload, lineEnding = ""] = match;
const text = payload.trim();
if (!text) return line;
stats.dataLines += 1;
try {
const parsed = JSON.parse(text);
if (parsed && typeof parsed === "object" && parsed.directory === rewrite.from) {
parsed.directory = rewrite.to;
stats.rewrittenEvents += 1;
return `${prefix}${JSON.stringify(parsed)}${lineEnding}`;
}
} catch {
stats.jsonErrors += 1;
}
return line;
}
function validateDisplayTimeZone(timeZone) {
try {
new Intl.DateTimeFormat("en-US", { timeZone }).format(new Date(0));
@@ -663,7 +729,7 @@ export async function proxyCloudApi({ request, response, url, cloudApiBaseUrl, c
}
}
async function proxyOpencodeRequest({ request, response, url, cloudApiBaseUrl, cloudApiProxyTimeoutMs, opencodeUpstreamUrl, opencodeProxyTimeoutMs, opencodeUsername, opencodePassword, serviceId, sendJson }) {
async function proxyOpencodeRequest({ request, response, url, cloudApiBaseUrl, cloudApiProxyTimeoutMs, opencodeUpstreamUrl, opencodeProxyTimeoutMs, opencodeUsername, opencodePassword, opencodeEventDirectoryRewrite = null, serviceId, sendJson }) {
const traceContext = cloudWebTraceContext(request);
const startedAtMs = Date.now();
if (!cloudApiBaseUrl) {
@@ -733,6 +799,7 @@ async function proxyOpencodeRequest({ request, response, url, cloudApiBaseUrl, c
}
const target = opencodeTargetUrl(url, opencodeUpstreamUrl);
const streamTransform = opencodeEventDirectoryTransformForTarget(target, opencodeEventDirectoryRewrite);
const extraResponseHeaders = {
...cloudWebTraceHeaders(traceContext, serviceId),
...(ticketAuth.ticket ? { "set-cookie": opencodeTicketSetCookie(ticketAuth.ticket) } : {})
@@ -748,7 +815,8 @@ async function proxyOpencodeRequest({ request, response, url, cloudApiBaseUrl, c
response,
body: request.method === "GET" || request.method === "HEAD" ? "" : body,
timeoutMs: opencodeProxyTimeoutMs,
extraResponseHeaders
extraResponseHeaders,
streamTransform
});
emitOpencodeProxySpanAsync({
traceContext,
@@ -761,7 +829,8 @@ async function proxyOpencodeRequest({ request, response, url, cloudApiBaseUrl, c
errorCode: proxyResult?.errorCode || "",
ticketAuth,
sessionStatusCode: session.statusCode || 0,
streaming: proxyResult?.streaming === true
streaming: proxyResult?.streaming === true,
streamTransformStats: proxyResult?.streamTransformStats
});
} catch (error) {
if (response.headersSent || response.writableEnded) {
@@ -916,7 +985,7 @@ function opencodeTraceDiagnostic(traceContext, extra = {}) {
};
}
function emitOpencodeProxySpanAsync({ traceContext, serviceId, request, url, target = null, startedAtMs, statusCode = 0, errorCode = "", ticketAuth = null, sessionStatusCode = 0, streaming = false } = {}) {
function emitOpencodeProxySpanAsync({ traceContext, serviceId, request, url, target = null, startedAtMs, statusCode = 0, errorCode = "", ticketAuth = null, sessionStatusCode = 0, streaming = false, streamTransformStats = null } = {}) {
const endpoint = otelTracesEndpoint(process.env);
if (!endpoint || !traceContext?.traceId || !traceContext?.spanId) return;
const endedAtMs = Date.now();
@@ -944,6 +1013,7 @@ function emitOpencodeProxySpanAsync({ traceContext, serviceId, request, url, tar
"opencode.proxy.ticket_accepted": Boolean(ticketAuth?.ticket),
"opencode.proxy.session_status_code": sessionStatusCode || undefined,
"opencode.proxy.streaming": streaming === true,
...opencodeStreamTransformStatsAttributes(streamTransformStats),
valuesPrinted: false
}
};
@@ -961,6 +1031,18 @@ function emitOpencodeProxySpanAsync({ traceContext, serviceId, request, url, tar
}, 0);
}
function opencodeStreamTransformStatsAttributes(stats) {
if (!stats?.enabled) return {};
return {
"opencode.proxy.sse.directory_rewrite_enabled": true,
"opencode.proxy.sse.directory_rewrite_from": stats.from,
"opencode.proxy.sse.directory_rewrite_to": stats.to,
"opencode.proxy.sse.directory_rewrite_data_lines": stats.dataLines,
"opencode.proxy.sse.directory_rewrite_events": stats.rewrittenEvents,
"opencode.proxy.sse.directory_rewrite_json_errors": stats.jsonErrors
};
}
function opencodeProxyRoute(pathname) {
const path = String(pathname || "/");
if (path === "/global/event") return "/global/event";
internal/dev-entrypoint/cloud-web-runtime.test.mjs
+52
View File
@@ -705,6 +705,58 @@ test("cloud web OpenCode proxy injects upstream Basic Auth without forwarding HW
}
});
test("cloud web OpenCode proxy rewrites configured event stream directory", async () => {
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 opencode = 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","properties":{"sessionID":"ses_1","status":{"type":"idle"}}}}\n\n');
response.write('data: {"directory":"/other","payload":{"type":"server.heartbeat","properties":{}}}\n\n');
response.end();
});
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 text = await response.text();
assert.match(text, /"directory":"\/"/u);
assert.doesNotMatch(text, /"directory":"\/workspace"/u);
assert.match(text, /"directory":"\/other"/u);
} finally {
await close(cloudWeb);
await close(opencode);
await close(cloudApi);
}
});
test("cloud web OpenCode proxy emits bounded OTLP span", async () => {
const otelBodies = [];
const otel = createServer(async (request, response) => {