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

fix(workbench): avoid readiness and SSE stalls (#1951)

Browse Source
This commit is contained in:
Lyon
2026-06-23 05:30:57 +08:00
committed by GitHub
parent 4cab31c341
commit 5e1cc8ade5
2 changed files with 62 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/server-workbench-http.ts
+61 -15
View File
@@ -74,23 +74,30 @@ export async function handleWorkbenchReadModelHttp(request, response, url, optio
}
export async function handleWorkbenchRealtimeHttp(request, response, url, options = {}) {
if (request.method !== "GET") return methodNotAllowed(response, "GET");
const perf = options.backendPerformance;
const auth = perf ? await perf.measure("workbench_auth", () => authenticateWorkbenchRead(request, response, options)) : await authenticateWorkbenchRead(request, response, options);
if (!auth) return;
try {
if (request.method !== "GET") return methodNotAllowed(response, "GET");
const requestedSessionId = safeSessionId(url.searchParams.get("sessionId") ?? url.searchParams.get("includeSessionId"));
const requestedTraceId = safeTraceId(url.searchParams.get("traceId"));
const heartbeatMs = parsePositiveInteger(options.env?.HWLAB_WORKBENCH_SSE_HEARTBEAT_MS, DEFAULT_WORKBENCH_SSE_HEARTBEAT_MS);
attachWorkbenchRealtimeOtelContext(request, {
sessionId: requestedSessionId,
traceId: requestedTraceId,
heartbeatMs,
outboxMode: typeof (options.runtimeStore ?? null)?.readWorkbenchProjectionOutbox === "function"
});
const perf = options.backendPerformance;
const auth = perf ? await perf.measure("workbench_auth", () => authenticateWorkbenchRead(request, response, options)) : await authenticateWorkbenchRead(request, response, options);
if (!auth) return;
if (url.searchParams.has("projectId") || url.searchParams.has("workspaceId")) {
sendJson(response, 400, workbenchError("workbench_authority_removed", "Workbench read model is keyed by sessionId/threadId/traceId only."));
return;
}
if (url.searchParams.has("projectId") || url.searchParams.has("workspaceId")) {
sendJson(response, 400, workbenchError("workbench_authority_removed", "Workbench read model is keyed by sessionId/threadId/traceId only."));
return;
}
const requestedSessionId = safeSessionId(url.searchParams.get("sessionId") ?? url.searchParams.get("includeSessionId"));
const requestedTraceId = safeTraceId(url.searchParams.get("traceId"));
const heartbeatMs = parsePositiveInteger(options.env?.HWLAB_WORKBENCH_SSE_HEARTBEAT_MS, DEFAULT_WORKBENCH_SSE_HEARTBEAT_MS);
const traceStore = options.traceStore ?? defaultCodeAgentTraceStore;
const readModel = createWorkbenchReadModel(options, auth.actor);
let closed = false;
const cleanup = [];
const traceStore = options.traceStore ?? defaultCodeAgentTraceStore;
const readModel = createWorkbenchReadModel(options, auth.actor);
let closed = false;
const cleanup = [];
response.writeHead(200, {
"content-type": "text/event-stream; charset=utf-8",
@@ -231,6 +238,9 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option
closed = true;
for (const item of cleanup.splice(0)) item();
});
} catch (error) {
handleWorkbenchRealtimeFailure(request, response, url, options, error);
}
}
function attachWorkbenchRealtimeOtelContext(request, fields = {}) {
@@ -268,6 +278,42 @@ function emitWorkbenchRealtimeAcceptedOtelSpan(request, env = process.env) {
}).catch(() => undefined);
}
function handleWorkbenchRealtimeFailure(request, response, url, options = {}, error) {
const classified = classifyWorkbenchReadModelFailure(error);
const context = request?.hwlabHttpRequestContext;
if (context) {
context.route = "/v1/workbench/events";
context.otelAttributes = {
...(context.otelAttributes ?? {}),
"workbench.route": "events",
"workbench.sse.lifecycle": "failed",
"error.code": classified.code,
"error.layer": "workbench.realtime"
};
}
logWorkbenchReadModelFailure(options.logger ?? console, {
event: "workbench_realtime_route_failed",
ok: false,
route: "/v1/workbench/events",
statusCode: classified.statusCode,
errorCode: classified.code,
errorName: error?.name ?? "Error",
message: error instanceof Error ? error.message : String(error ?? "unknown"),
valuesRedacted: true
});
if (response.headersSent && !response.destroyed) {
try {
response.write(`event: workbench.error\n`);
response.write(`data: ${JSON.stringify({ contractVersion: "workbench-events-v1", serverSentAt: new Date().toISOString(), type: "error", error: { code: classified.code, message: classified.message, valuesRedacted: true } })}\n\n`);
response.end();
} catch {
try { response.destroy(); } catch {}
}
return;
}
if (!response.destroyed) sendJson(response, classified.statusCode, workbenchError(classified.code, classified.message, { route: "/v1/workbench/events", errorName: error?.name ?? "Error" }));
}
function realtimeEventCreatedAt(payload) {
const candidates = [payload?.event?.createdAt, payload?.snapshot?.lastEvent?.createdAt, payload?.turn?.updatedAt];
for (const value of candidates) {
internal/cloud/server.ts
+1 -1
View File
@@ -595,7 +595,7 @@ async function handleRestAdapter(request, response, url, options) {
const [dbProbe, codeAgent, runtime] = await Promise.all([
observeHttpRoutePhase(request, options, "v1.db_readiness", () => buildDbRuntimeReadiness(options.env ?? process.env, dbProbeOptions)),
observeHttpRoutePhase(request, options, "v1.code_agent_availability", () => describeCodeAgentAvailability(options.env ?? process.env, options)),
observeHttpRoutePhase(request, options, "v1.runtime_readiness", () => runtimeReadiness(options.runtimeStore))
observeHttpRoutePhase(request, options, "v1.runtime_readiness", () => runtimeReadinessSnapshot(options.runtimeStore))
]);
const db = applyRuntimeDbReadinessLayers(dbProbe, runtime);
const readiness = buildCloudApiReadiness({ db, codeAgent, runtime });