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

fix(workbench): stream session outbox events over sse (#1962)

Browse Source
This commit is contained in:
Lyon
2026-06-23 10:20:50 +08:00
committed by GitHub
parent 78a57dc98c
commit ee8181fb52
1 changed files with 22 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/server-workbench-http.ts
+22 -11
View File
@@ -144,34 +144,45 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option
emitWorkbenchRealtimeAcceptedOtelSpan(request, options.env);
if (activeTraceId) {
await (perf ? perf.measure("workbench_initial_trace", () => writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "initial" })) : writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "initial" }));
const runtimeStore = options.runtimeStore ?? null;
const outboxPollMs = parsePositiveInteger(options.env?.HWLAB_WORKBENCH_SSE_OUTBOX_POLL_MS, 2000);
let outboxCursor = 0;
if (typeof runtimeStore?.readWorkbenchProjectionOutbox === "function") {
}
const runtimeStore = options.runtimeStore ?? null;
const outboxPollMs = parsePositiveInteger(options.env?.HWLAB_WORKBENCH_SSE_OUTBOX_POLL_MS, 2000);
let outboxCursor = 0;
const outboxTraceSnapshots = new Set(activeTraceId ? [activeTraceId] : []);
if (typeof runtimeStore?.readWorkbenchProjectionOutbox === "function" && (streamSessionId || activeTraceId)) {
const pollOutbox = async () => {
if (closed || response.destroyed) return;
try {
const rows = await runtimeStore.readWorkbenchProjectionOutbox({ afterSeq: outboxCursor, limit: 50, traceId: activeTraceId });
const query = { afterSeq: outboxCursor, limit: 50 };
if (streamSessionId) query.sessionId = streamSessionId;
else query.traceId = activeTraceId;
const rows = await runtimeStore.readWorkbenchProjectionOutbox(query);
for (const row of rows) {
if (closed || response.destroyed) break;
outboxCursor = row.outboxSeq;
const rowTraceId = safeTraceId(row.traceId) ?? activeTraceId;
const rowSessionId = safeSessionId(row.sessionId ?? streamSessionId) ?? streamSessionId;
if (rowTraceId && !outboxTraceSnapshots.has(rowTraceId)) {
outboxTraceSnapshots.add(rowTraceId);
await writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: rowTraceId, reason: "outbox-session" });
}
if (row.commitType === "terminal") {
writeEvent("workbench.trace.snapshot", {
type: "trace.snapshot",
sessionId: row.sessionId ?? streamSessionId,
sessionId: rowSessionId,
threadId: streamThreadId,
traceId: row.traceId ?? activeTraceId,
traceId: rowTraceId,
reason: "outbox-terminal",
cursor: { traceSeq: row.projectedSeq, outboxSeq: row.outboxSeq },
projectionStatus: { terminal: row.terminal, sealed: row.sealed }
});
void writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "terminal" });
if (rowTraceId) void writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: rowTraceId, reason: "terminal" });
} else {
writeEvent("workbench.trace.event", {
type: "trace.event",
sessionId: row.sessionId ?? streamSessionId,
sessionId: rowSessionId,
threadId: streamThreadId,
traceId: row.traceId ?? activeTraceId,
traceId: rowTraceId,
cursor: { traceSeq: row.projectedSeq, outboxSeq: row.outboxSeq }
});
}
@@ -186,7 +197,7 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option
await pollOutbox();
const outboxTimer = setInterval(() => { void pollOutbox(); }, Math.max(500, outboxPollMs));
cleanup.push(() => clearInterval(outboxTimer));
} else {
} else if (activeTraceId) {
const unsubscribe = traceStore.subscribe(activeTraceId, (event, snapshot) => {
if (event) {
writeEvent("workbench.trace.event", {