mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
fa7252449a10a4b0d12908bf7a54e856efde2358
pikasTech-HWLAB/internal/cloud/server-workbench-http.test.ts
T

1082 lines
51 KiB
TypeScript
Raw Blame History

// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p0-projector-resume; PJ2026-010403 API契约 draft-2026-06-18-r1.
// Responsibility: Workbench read model, durable projection diagnostics, and pure-GET regression tests.
import assert from "node:assert/strict";
import { test } from "bun:test";
import { createCloudApiServer } from "./server.ts";
import { createBackendPerformanceStore } from "./backend-performance.ts";
import { createCodeAgentTraceStore } from "./code-agent-trace-store.ts";
import { createCodeAgentChatResultStore } from "./server-code-agent-http.ts";
import { createWorkbenchTurnProjection, projectionDiagnostics } from "./workbench-turn-projection.ts";
const ACTOR = { id: "usr_workbench_reader", username: "reader", displayName: "Reader", role: "user", status: "active" };
test("workbench projection diagnostics exposes AgentRun refresh errors without terminal inference", () => {
const traceId = "trc_projection_refresh_degraded";
const result = {
traceId,
status: "running",
agentRun: { runId: "run_projection_refresh_degraded", commandId: "cmd_projection_refresh_degraded", status: "running" }
};
const trace = { traceId, status: "running", events: [], eventCount: 0, updatedAt: "2026-06-18T01:00:00.000Z" };
const projection = createWorkbenchTurnProjection({ traceId, result, trace });
const diagnostics = projectionDiagnostics({
traceId,
result,
trace,
projection,
refreshError: { code: "agentrun_result_timeout", message: "状态更新超时，AgentRun 结果暂不可见。", retryable: true, timeoutMs: 2500 }
});
assert.equal(projection.status, "running");
assert.equal(diagnostics.projectionStatus, "blocked");
assert.equal(diagnostics.projectionHealth, "degraded");
assert.equal(diagnostics.blocker.code, "agentrun_result_timeout");
assert.equal(diagnostics.blocker.message, "状态更新超时，AgentRun 结果暂不可见。");
assert.equal(diagnostics.sourceRunId, "run_projection_refresh_degraded");
assert.equal(diagnostics.sourceCommandId, "cmd_projection_refresh_degraded");
});
test("workbench projection diagnostics keeps projecting health distinct from caught-up", () => {
const traceId = "trc_projection_health_projecting";
const result = {
traceId,
status: "running",
agentRun: { runId: "run_projection_health_projecting", commandId: "cmd_projection_health_projecting", status: "running" }
};
const trace = {
traceId,
status: "running",
events: [{ traceId, seq: 7, sourceSeq: 7, type: "backend_status", status: "running", createdAt: "2026-06-19T04:40:00.000Z" }],
eventCount: 1,
updatedAt: "2026-06-19T04:40:00.000Z"
};
const projection = createWorkbenchTurnProjection({ traceId, result, trace });
const diagnostics = projectionDiagnostics({ traceId, result, trace, projection });
assert.equal(projection.status, "running");
assert.equal(diagnostics.projectionStatus, "projecting");
assert.equal(diagnostics.projectionHealth, "projecting");
});
test("workbench trace event page exposes monotonic cursor range for restored mixed events", async () => {
const traceId = "trc_workbench_trace_mixed_cursor";
const session = {
id: "ses_workbench_trace_mixed_cursor",
projectId: "prj_workbench_authority_session",
agentId: "hwlab-code-agent",
status: "completed",
ownerUserId: ACTOR.id,
conversationId: "cnv_workbench_trace_mixed_cursor",
threadId: "thread-workbench-trace-mixed-cursor",
lastTraceId: traceId,
updatedAt: "2026-06-19T04:10:17.060Z",
session: { sessionStatus: "completed", lastTraceId: traceId, messages: [{ role: "user", text: "trace", traceId }] }
};
const mixedEvents = [
{ seq: 1, source: "agentrun", sourceSeq: 1, type: "backend", status: "running", label: "agentrun:backend:run-created", createdAt: "2026-06-19T01:27:27.092Z" },
{ seq: 5, source: "agentrun", sourceSeq: 1, type: "backend", status: "running", label: "agentrun:backend:run-created", createdAt: "2026-06-19T01:27:27.092Z" },
{ seq: 6, source: "agentrun", sourceSeq: 4, type: "backend", status: "running", label: "agentrun:backend:runner-image-work-ready-smoke", createdAt: "2026-06-19T01:27:34.128Z" },
{ seq: 2, source: "agentrun", sourceSeq: 4, type: "backend", status: "running", label: "agentrun:backend:runner-image-work-ready-smoke", createdAt: "2026-06-19T01:27:34.128Z" },
{ seq: 3, source: "agentrun", sourceSeq: 5, type: "backend", status: "running", label: "agentrun:backend:run-claimed", createdAt: "2026-06-19T01:27:34.324Z" },
{ seq: 7, source: "agentrun", sourceSeq: 5, type: "backend", status: "running", label: "agentrun:backend:run-claimed", createdAt: "2026-06-19T01:27:34.324Z" },
{ seq: 4, type: "result", status: "completed", label: "agentrun:result:completed", terminal: true, createdAt: "2026-06-19T01:27:34.324Z" },
{ seq: 1, type: "request", status: "accepted", label: "agentrun:request:accepted", createdAt: "2026-06-19T04:10:15.362Z" },
{ seq: 2, type: "backend", status: "running", label: "agentrun:run:reused", createdAt: "2026-06-19T04:10:15.664Z" },
{ seq: 3, type: "backend", status: "running", label: "agentrun:command:created", createdAt: "2026-06-19T04:10:16.013Z" },
{ seq: 4, type: "backend", status: "running", label: "agentrun:runner-job:ensured", createdAt: "2026-06-19T04:10:17.060Z" }
];
const accessController = {
store: {
async getAgentSession(sessionId) { return sessionId === session.id ? session : null; },
async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; }
},
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
const runtimeStore = {
async queryAgentTraceEvents(input = {}) {
assert.equal(input.traceId, traceId);
return { events: mixedEvents, count: mixedEvents.length };
}
};
const server = createCloudApiServer({ accessController, runtimeStore, codeAgentChatResults: createCodeAgentChatResultStore() });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?limit=25`);
assert.equal(trace.status, 200);
assert.equal(trace.body.events.length, mixedEvents.length);
assert.deepEqual(trace.body.events.map((event) => event.seq), [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]);
assert.equal(trace.body.events[0].label, "agentrun:backend:run-created");
assert.equal(trace.body.events.at(-1).label, "agentrun:runner-job:ensured");
assert.deepEqual(trace.body.range, { afterSeq: 0, fromSeq: 1, toSeq: 11, limit: 25, returned: 11, total: 11 });
assert.equal(trace.body.nextSeq, 11);
assert.equal(trace.body.hasMore, false);
assert.equal(trace.body.nextCursor, null);
const middle = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?cursor=seq:4&limit=3`);
assert.equal(middle.status, 200);
assert.deepEqual(middle.body.events.map((event) => event.seq), [5, 6, 7]);
assert.deepEqual(middle.body.range, { afterSeq: 4, fromSeq: 5, toSeq: 7, limit: 3, returned: 3, total: 11 });
assert.equal(middle.body.hasMore, true);
assert.equal(middle.body.nextSeq, 7);
assert.equal(middle.body.nextCursor, "seq:7");
const empty = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?cursor=seq:999999&limit=5`);
assert.equal(empty.status, 200);
assert.deepEqual(empty.body.events, []);
assert.deepEqual(empty.body.range, { afterSeq: 999999, fromSeq: null, toSeq: null, limit: 5, returned: 0, total: 11 });
assert.equal(empty.body.nextSeq, 999999);
} finally {
await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}
});
test("workbench read model exposes session, messages, turn, and trace without write repair", async () => {
const traceStore = createCodeAgentTraceStore();
const results = createCodeAgentChatResultStore();
const writes = { recordAgentSessionOwner: 0, updateWorkspace: 0 };
const traceId = "trc_workbench_read_model";
const session = {
id: "ses_workbench_read_model",
projectId: "prj_workbench_authority_session",
agentId: "hwlab-code-agent",
status: "completed",
startedAt: "2026-06-17T00:00:00.000Z",
endedAt: "2026-06-17T00:00:03.000Z",
ownerUserId: ACTOR.id,
conversationId: "cnv_workbench_read_model",
threadId: "thread-workbench-read-model",
lastTraceId: traceId,
updatedAt: "2026-06-17T00:00:03.000Z",
session: {
sessionStatus: "completed",
providerProfile: "codex-api",
lastTraceId: traceId,
messages: [
{ role: "user", text: "ping", traceId, createdAt: "2026-06-17T00:00:00.000Z" },
{ role: "assistant", text: "pong", traceId, createdAt: "2026-06-17T00:00:03.000Z" }
],
valuesRedacted: true,
secretMaterialStored: false
}
};
const listInputs = [];
traceStore.append(traceId, { type: "request", status: "accepted", label: "request:accepted" });
traceStore.append(traceId, { type: "result", status: "completed", label: "result:completed", terminal: true });
results.set(traceId, {
status: "completed",
traceId,
ownerUserId: ACTOR.id,
conversationId: session.conversationId,
sessionId: session.id,
threadId: session.threadId,
finalResponse: "pong",
agentRun: { runId: "run_workbench_read_model", commandId: "cmd_workbench_read_model", status: "completed" }
});
const accessController = {
store: {
async listAgentSessionsForUser(input = {}) {
listInputs.push(input);
return [session];
},
async getAgentSession(sessionId) { return sessionId === session.id ? session : null; },
async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; },
async recordAgentSessionOwner() { writes.recordAgentSessionOwner += 1; throw new Error("read model must not write session owner"); },
async updateWorkspace() { writes.updateWorkspace += 1; throw new Error("read model must not update workspace"); }
},
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: results });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const sessions = await getJson(port, "/v1/workbench/sessions");
assert.equal(sessions.status, 200);
assert.equal(sessions.body.contractVersion, "workbench-sessions-v1");
assert.equal(sessions.body.sessions[0].sessionId, session.id);
assert.equal(sessions.body.sessions[0].turnSummary.traceId, traceId);
assert.equal(listInputs[0].projectId, undefined);
assert.equal(listInputs[0].limit, 21);
const staleProject = await getJson(port, `/v1/workbench/sessions?projectId=prj_stale_filter&includeSessionId=${encodeURIComponent(session.id)}`);
assert.equal(staleProject.status, 400);
assert.equal(staleProject.body.error.code, "workbench_authority_removed");
assert.equal(listInputs.length, 1);
const detail = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}`);
assert.equal(detail.status, 200);
assert.equal(detail.body.session.sessionId, session.id);
assert.equal(detail.body.session.messagePageUrl, `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages`);
assert.equal(detail.body.session.messageCount, 2);
const messages = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages?limit=1`);
assert.equal(messages.status, 200);
assert.equal(messages.body.contractVersion, "workbench-message-page-v1");
assert.equal(messages.body.messages.length, 1);
assert.equal(messages.body.messages[0].role, "user");
assert.equal(messages.body.messages[0].text, "ping");
assert.equal(messages.body.messages[0].parts[0].text, "ping");
assert.equal(messages.body.hasMore, true);
const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`);
assert.equal(turn.status, 200);
assert.equal(turn.body.contractVersion, "workbench-turn-snapshot-v1");
assert.equal(turn.body.turn.traceId, traceId);
assert.equal(turn.body.turn.userMessageId, messages.body.messages[0].messageId);
assert.equal(turn.body.turn.terminal, true);
const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?limit=1`);
assert.equal(trace.status, 200);
assert.equal(trace.body.contractVersion, "workbench-trace-events-v1");
assert.equal(trace.body.events.length, 1);
assert.equal(trace.body.hasMore, true);
const emptyTracePage = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?cursor=seq:99`);
assert.equal(emptyTracePage.status, 200);
assert.equal(emptyTracePage.body.events.length, 0);
assert.equal(emptyTracePage.body.range.fromSeq, null);
assert.equal(emptyTracePage.body.range.toSeq, null);
assert.equal(emptyTracePage.body.nextSeq, 99);
assert.deepEqual(writes, { recordAgentSessionOwner: 0, updateWorkspace: 0 });
} finally {
await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}
});
test("workbench read model recovers trace events from durable projection without duplicating final response", async () => {
const traceStore = createCodeAgentTraceStore();
const traceId = "trc_workbench_durable_reload";
const session = {
id: "ses_workbench_durable_reload",
projectId: "prj_hwpod_workbench",
agentId: "hwlab-code-agent",
status: "completed",
startedAt: "2026-06-17T01:00:00.000Z",
endedAt: "2026-06-17T01:00:04.000Z",
ownerUserId: ACTOR.id,
conversationId: "cnv_workbench_durable_reload",
threadId: "thread-workbench-durable-reload",
lastTraceId: traceId,
updatedAt: "2026-06-17T01:00:04.000Z",
session: {
sessionStatus: "completed",
lastTraceId: traceId,
finalResponse: "durable pong",
messages: [
{ role: "user", text: "durable ping", traceId, createdAt: "2026-06-17T01:00:00.000Z" },
{ role: "agent", text: "durable pong", traceId, createdAt: "2026-06-17T01:00:04.000Z" }
],
valuesRedacted: true,
secretMaterialStored: false
}
};
const durableEvents = [
{ traceId, seq: 1, type: "request", status: "accepted", label: "request:accepted", createdAt: "2026-06-17T01:00:00.000Z", valuesPrinted: false },
{ traceId, seq: 2, type: "result", status: "completed", label: "result:completed", terminal: true, createdAt: "2026-06-17T01:00:04.000Z", valuesPrinted: false }
];
const accessController = {
store: {
async getAgentSession(sessionId) { return sessionId === session.id ? session : null; },
async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; }
},
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
let traceQueryCount = 0;
const runtimeStore = {
async queryAgentTraceEvents(params = {}) {
traceQueryCount += 1;
assert.equal(params.traceId, traceId);
return { events: durableEvents, count: durableEvents.length };
}
};
const server = createCloudApiServer({ accessController, traceStore, runtimeStore, codeAgentChatResults: createCodeAgentChatResultStore() });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const messages = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages?limit=10`);
assert.equal(messages.status, 200);
assert.equal(messages.body.total, 2);
assert.deepEqual(messages.body.messages.map((message) => message.role), ["user", "agent"]);
const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`);
assert.equal(turn.status, 200);
assert.equal(turn.body.turn.assistantMessageId, messages.body.messages[1].messageId);
assert.equal(turn.body.turn.trace.eventCount, 2);
assert.equal(turn.body.turn.trace.status, "completed");
const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?limit=10`);
assert.equal(trace.status, 200);
assert.equal(trace.body.events.length, 2);
assert.equal(trace.body.traceStatus, "completed");
assert.equal(trace.body.hasMore, false);
assert.equal(traceQueryCount, 1);
} finally {
await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}
});
test("workbench read model projects terminal result atomically across session, messages, and turn", async () => {
const traceStore = createCodeAgentTraceStore();
const results = createCodeAgentChatResultStore();
const traceId = "trc_workbench_terminal_atomic";
const finalText = "terminal projection final response";
const session = {
id: "ses_workbench_terminal_atomic",
projectId: "prj_hwpod_workbench",
agentId: "hwlab-code-agent",
status: "running",
ownerUserId: ACTOR.id,
conversationId: "cnv_workbench_terminal_atomic",
threadId: "thread-workbench-terminal-atomic",
lastTraceId: traceId,
updatedAt: "2026-06-17T02:10:00.000Z",
session: {
sessionStatus: "running",
lastTraceId: traceId,
messages: [
{ role: "user", text: "terminal atomic ping", traceId, status: "sent", createdAt: "2026-06-17T02:09:58.000Z" },
{ role: "agent", text: "", traceId, status: "running", createdAt: "2026-06-17T02:09:59.000Z" }
],
valuesRedacted: true,
secretMaterialStored: false
}
};
traceStore.append(traceId, { seq: 1, type: "request", status: "accepted", label: "request:accepted" });
traceStore.append(traceId, { seq: 2, type: "result", status: "completed", label: "result:completed", terminal: true });
results.set(traceId, {
status: "completed",
traceId,
ownerUserId: ACTOR.id,
conversationId: session.conversationId,
sessionId: session.id,
threadId: session.threadId,
finalResponse: { text: finalText, status: "completed", traceId },
agentRun: { runId: "run_workbench_terminal_atomic", commandId: "cmd_workbench_terminal_atomic", status: "completed", terminalStatus: "completed" }
});
const accessController = {
store: {
async listAgentSessionsForUser() { return [session]; },
async getAgentSession(sessionId) { return sessionId === session.id ? session : null; },
async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; }
},
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: results });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const sessions = await getJson(port, `/v1/workbench/sessions?includeSessionId=${encodeURIComponent(session.id)}`);
assert.equal(sessions.status, 200);
assert.equal(sessions.body.sessions[0].status, "completed");
assert.equal(sessions.body.sessions[0].running, false);
assert.equal(sessions.body.sessions[0].terminal, true);
const messages = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages?limit=10`);
assert.equal(messages.status, 200);
assert.equal(messages.body.total, 2);
const assistant = messages.body.messages[1];
assert.equal(assistant.role, "agent");
assert.equal(assistant.status, "completed");
assert.equal(assistant.text, finalText);
assert.equal(assistant.parts[0].status, "completed");
assert.equal(assistant.parts[0].text, finalText);
const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`);
assert.equal(turn.status, 200);
assert.equal(turn.body.turn.status, "completed");
assert.equal(turn.body.turn.terminal, true);
assert.equal(turn.body.turn.assistantText, finalText);
assert.equal(turn.body.turn.finalResponse.text, finalText);
} finally {
await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}
});
test("workbench read model does not expose trace-only memory without visible session or result owner", async () => {
const traceStore = createCodeAgentTraceStore();
const traceId = "trc_workbench_trace_only";
traceStore.append(traceId, { type: "request", status: "accepted", label: "request:accepted" });
const accessController = {
store: {
async getAgentSessionByTraceId() { return null; }
},
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: createCodeAgentChatResultStore() });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`);
assert.equal(turn.status, 404);
assert.equal(turn.body.error.code, "workbench_turn_not_found");
const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events`);
assert.equal(trace.status, 404);
assert.equal(trace.body.error.code, "workbench_trace_not_found");
} finally {
await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}
});
test("workbench read model projects current turn running state from trace projection", async () => {
const traceStore = createCodeAgentTraceStore();
const traceId = "trc_workbench_active_state";
const session = {
id: "ses_workbench_active_state",
projectId: "prj_hwpod_workbench",
agentId: "hwlab-code-agent",
status: "active",
ownerUserId: ACTOR.id,
conversationId: "cnv_workbench_active_state",
threadId: "thread-workbench-active-state",
lastTraceId: traceId,
updatedAt: "2026-06-17T00:01:00.000Z",
session: { sessionStatus: "active", lastTraceId: traceId, messages: [{ role: "user", text: "still running", traceId }] }
};
traceStore.append(traceId, { type: "backend", status: "running", label: "runner:created", terminal: false });
const accessController = {
store: {
async listAgentSessionsForUser() { return [session]; }
},
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: createCodeAgentChatResultStore() });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const sessions = await getJson(port, "/v1/workbench/sessions");
assert.equal(sessions.status, 200);
assert.equal(sessions.body.sessions[0].status, "running");
assert.equal(sessions.body.sessions[0].running, true);
assert.equal(sessions.body.sessions[0].terminal, false);
} finally {
await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}
});
test("workbench read model projects current turn running state for idle session summary", async () => {
const traceStore = createCodeAgentTraceStore();
const traceId = "trc_workbench_idle_with_running_trace";
const session = {
id: "ses_workbench_idle_with_running_trace",
projectId: "prj_hwpod_workbench",
agentId: "hwlab-code-agent",
status: "idle",
ownerUserId: ACTOR.id,
conversationId: "cnv_workbench_idle_with_running_trace",
threadId: "thread-workbench-idle-with-running-trace",
lastTraceId: traceId,
updatedAt: "2026-06-17T00:02:00.000Z",
session: { sessionStatus: "idle", lastTraceId: traceId, messages: [{ role: "user", text: "still running despite idle row", traceId, status: "sent" }] }
};
traceStore.append(traceId, { type: "backend", status: "running", label: "runner:created", terminal: false });
const accessController = {
store: {
async listAgentSessionsForUser() { return [session]; }
},
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: createCodeAgentChatResultStore() });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const sessions = await getJson(port, `/v1/workbench/sessions?includeSessionId=${encodeURIComponent(session.id)}`);
assert.equal(sessions.status, 200);
assert.equal(sessions.body.sessions[0].status, "running");
assert.equal(sessions.body.sessions[0].running, true);
assert.equal(sessions.body.sessions[0].terminal, false);
assert.equal(sessions.body.sessions[0].turnSummary.status, "running");
} finally {
await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}
});
test("workbench read model projects completed current turn for idle session summary", async () => {
const traceStore = createCodeAgentTraceStore();
const results = createCodeAgentChatResultStore();
const traceId = "trc_workbench_idle_with_completed_trace";
const finalText = "OK";
const session = {
id: "ses_workbench_idle_with_completed_trace",
projectId: "prj_hwpod_workbench",
agentId: "hwlab-code-agent",
status: "idle",
ownerUserId: ACTOR.id,
conversationId: "cnv_workbench_idle_with_completed_trace",
threadId: "thread-workbench-idle-with-completed-trace",
lastTraceId: traceId,
updatedAt: "2026-06-17T00:03:00.000Z",
session: {
sessionStatus: "idle",
lastTraceId: traceId,
messages: [
{ role: "user", text: "reply OK", traceId, status: "sent" },
{ role: "agent", text: "", traceId, status: "running" }
]
}
};
traceStore.append(traceId, { type: "request", status: "accepted", label: "request:accepted" });
traceStore.append(traceId, { type: "result", status: "completed", label: "result:completed", terminal: true });
results.set(traceId, {
status: "completed",
traceId,
ownerUserId: ACTOR.id,
conversationId: session.conversationId,
sessionId: session.id,
threadId: session.threadId,
finalResponse: finalText,
agentRun: { runId: "run_workbench_idle_completed", commandId: "cmd_workbench_idle_completed", status: "completed" }
});
const accessController = {
store: {
async listAgentSessionsForUser() { return [session]; },
async getAgentSession(sessionId) { return sessionId === session.id ? session : null; },
async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; }
},
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
const backendPerformanceStore = createBackendPerformanceStore({ env: { POD_NAMESPACE: "hwlab-v03", HWLAB_GITOPS_TARGET: "v03", HWLAB_RUNTIME_LANE: "v03", HWLAB_CODE_AGENT_AGENTRUN_PROVIDER_ID: "D601" } });
const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: results, backendPerformanceStore });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const sessions = await getJson(port, `/v1/workbench/sessions?includeSessionId=${encodeURIComponent(session.id)}`);
assert.equal(sessions.status, 200);
assert.equal(sessions.body.sessions[0].status, "completed");
assert.equal(sessions.body.sessions[0].running, false);
assert.equal(sessions.body.sessions[0].terminal, true);
assert.equal(sessions.body.sessions[0].turnSummary.status, "completed");
const messages = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages?limit=10`);
assert.equal(messages.status, 200);
assert.equal(messages.body.messages[1].status, "completed");
assert.equal(messages.body.messages[1].text, finalText);
const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`);
assert.equal(turn.status, 200);
assert.equal(turn.body.turn.status, "completed");
assert.equal(turn.body.turn.running, false);
assert.equal(turn.body.turn.terminal, true);
assert.equal(turn.body.turn.assistantText, finalText);
assert.equal(turn.body.projectionStatus, "caught-up");
assert.equal(turn.body.sourceRunId, "run_workbench_idle_completed");
assert.equal(turn.body.sourceCommandId, "cmd_workbench_idle_completed");
const metricsText = backendPerformanceStore.metricsText();
const lagCountLine = metricsText.split("\n").find((line) => line.startsWith("hwlab_workbench_projection_lag_seconds_count") && line.includes('projection_status="caught_up"'));
const turnGetCountLine = metricsText.split("\n").find((line) => line.startsWith("hwlab_workbench_turn_get_duration_seconds_count") && line.includes('route="/v1/workbench/turns/:traceId"'));
assert.ok(lagCountLine?.endsWith(" 1"));
assert.ok(turnGetCountLine?.endsWith(" 1"));
const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?limit=10`);
assert.equal(trace.status, 200);
assert.equal(trace.body.projectionStatus, "caught-up");
assert.equal(trace.body.sourceRunId, "run_workbench_idle_completed");
assert.equal(trace.body.sourceCommandId, "cmd_workbench_idle_completed");
assert.equal(trace.body.lastProjectedSeq, 2);
} finally {
await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}
});
test("workbench read model lets terminal result override stale running session state", async () => {
const traceStore = createCodeAgentTraceStore();
const results = createCodeAgentChatResultStore();
const traceId = "trc_workbench_stale_running_result_completed";
const finalText = "OK";
const session = {
id: "ses_workbench_stale_running_result_completed",
projectId: "prj_hwpod_workbench",
agentId: "hwlab-code-agent",
status: "running",
ownerUserId: ACTOR.id,
conversationId: "cnv_workbench_stale_running_result_completed",
threadId: "thread-workbench-stale-running-result-completed",
lastTraceId: traceId,
updatedAt: "2026-06-18T16:31:04.000Z",
session: {
sessionStatus: "running",
lastTraceId: traceId,
messages: [
{ role: "user", text: "reply OK", traceId, status: "sent" },
{ role: "agent", text: "", traceId, status: "running" }
]
}
};
traceStore.append(traceId, { type: "backend", status: "running", label: "runner:created", terminal: false });
results.set(traceId, {
status: "completed",
traceId,
ownerUserId: ACTOR.id,
conversationId: session.conversationId,
sessionId: session.id,
threadId: session.threadId,
finalResponse: finalText,
agentRun: { runId: "run_workbench_stale_running_completed", commandId: "cmd_workbench_stale_running_completed", status: "completed" }
});
const accessController = {
store: {
async listAgentSessionsForUser() { return [session]; },
async getAgentSession(sessionId) { return sessionId === session.id ? session : null; },
async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; }
},
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: results });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const sessions = await getJson(port, `/v1/workbench/sessions?includeSessionId=${encodeURIComponent(session.id)}`);
assert.equal(sessions.status, 200);
assert.equal(sessions.body.sessions[0].status, "completed");
assert.equal(sessions.body.sessions[0].running, false);
assert.equal(sessions.body.sessions[0].terminal, true);
assert.equal(sessions.body.sessions[0].turnSummary.status, "completed");
const detail = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}`);
assert.equal(detail.status, 200);
assert.equal(detail.body.session.status, "completed");
assert.equal(detail.body.session.running, false);
assert.equal(detail.body.session.terminal, true);
const messages = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages?limit=10`);
assert.equal(messages.status, 200);
assert.equal(messages.body.messages[1].status, "completed");
assert.equal(messages.body.messages[1].text, finalText);
const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`);
assert.equal(turn.status, 200);
assert.equal(turn.body.status, "completed");
assert.equal(turn.body.turn.status, "completed");
assert.equal(turn.body.turn.running, false);
assert.equal(turn.body.turn.terminal, true);
assert.equal(turn.body.turn.assistantText, finalText);
} finally {
await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}
});
test("workbench session list uses compact trace result without durable trace hydration", async () => {
const traceStore = createCodeAgentTraceStore();
const traceId = "trc_workbench_compact_list_summary";
const finalText = "compact list summary OK";
const session = {
id: "ses_workbench_compact_list_summary",
projectId: "prj_hwpod_workbench",
agentId: "hwlab-code-agent",
status: "running",
ownerUserId: ACTOR.id,
conversationId: "cnv_workbench_compact_list_summary",
threadId: "thread-workbench-compact-list-summary",
lastTraceId: traceId,
updatedAt: "2026-06-19T15:40:00.000Z",
session: {
sessionStatus: "running",
lastTraceId: traceId,
messages: [
{ role: "user", text: "compact summary", traceId, status: "sent" },
{ role: "agent", text: "", traceId, status: "running" }
],
traceResults: {
[traceId]: {
traceId,
status: "completed",
finalResponse: { text: finalText, status: "completed", traceId, valuesPrinted: false },
traceSummary: {
traceId,
source: "agent-session-compact-summary",
sourceEventCount: 42,
terminalStatus: "completed",
agentRun: { runId: "run_workbench_compact_list_summary", commandId: "cmd_workbench_compact_list_summary", lastSeq: 42, valuesPrinted: false },
valuesPrinted: false
},
agentRun: { runId: "run_workbench_compact_list_summary", commandId: "cmd_workbench_compact_list_summary", status: "completed", terminalStatus: "completed", lastSeq: 42, valuesPrinted: false },
updatedAt: "2026-06-19T15:40:00.000Z",
valuesRedacted: true,
secretMaterialStored: false
}
},
valuesRedacted: true,
secretMaterialStored: false
}
};
const accessController = {
store: {
async listAgentSessionsForUser() { return [session]; }
},
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
let durableQueryCount = 0;
const runtimeStore = {
async queryAgentTraceEvents() {
durableQueryCount += 1;
throw new Error("session list must not hydrate durable trace events");
}
};
const server = createCloudApiServer({ accessController, traceStore, runtimeStore, codeAgentChatResults: createCodeAgentChatResultStore() });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const sessions = await getJson(port, `/v1/workbench/sessions?includeSessionId=${encodeURIComponent(session.id)}`);
assert.equal(sessions.status, 200);
assert.equal(sessions.body.sessions[0].status, "completed");
assert.equal(sessions.body.sessions[0].running, false);
assert.equal(sessions.body.sessions[0].terminal, true);
assert.equal(sessions.body.sessions[0].turnSummary.status, "completed");
assert.equal(sessions.body.sessions[0].turnSummary.eventCount, 42);
assert.equal(sessions.body.sessions[0].projectionStatus, "caught-up");
assert.equal(durableQueryCount, 0);
} finally {
await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}
});
test("workbench read model recovers terminal session status from durable trace", async () => {
const traceStore = createCodeAgentTraceStore();
const traceId = "trc_workbench_durable_terminal_after_memory_running";
const session = {
id: "ses_workbench_durable_terminal_after_memory_running",
projectId: "prj_hwpod_workbench",
agentId: "hwlab-code-agent",
status: "running",
ownerUserId: ACTOR.id,
conversationId: "cnv_workbench_durable_terminal_after_memory_running",
threadId: "thread-workbench-durable-terminal-after-memory-running",
lastTraceId: traceId,
updatedAt: "2026-06-18T17:22:00.000Z",
session: {
sessionStatus: "running",
lastTraceId: traceId,
messages: [
{ role: "user", text: "reply OK", traceId, status: "sent" },
{ role: "agent", text: "", traceId, status: "running" }
]
}
};
traceStore.append(traceId, { type: "backend", status: "running", label: "runner:created", terminal: false, seq: 1 });
const durableEvents = [
{ traceId, seq: 1, type: "request", status: "accepted", label: "request:accepted", createdAt: "2026-06-18T17:21:40.000Z", valuesPrinted: false },
{ traceId, seq: 2, type: "result", status: "completed", label: "result:completed", terminal: true, createdAt: "2026-06-18T17:22:00.000Z", valuesPrinted: false }
];
const accessController = {
store: {
async listAgentSessionsForUser() { return [session]; },
async getAgentSession(sessionId) { return sessionId === session.id ? session : null; },
async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; }
},
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
let durableQueryCount = 0;
const runtimeStore = {
async queryAgentTraceEvents(params = {}) {
durableQueryCount += 1;
assert.equal(params.traceId, traceId);
return { events: durableEvents, count: durableEvents.length };
}
};
const server = createCloudApiServer({ accessController, traceStore, runtimeStore, codeAgentChatResults: createCodeAgentChatResultStore() });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const sessions = await getJson(port, `/v1/workbench/sessions?includeSessionId=${encodeURIComponent(session.id)}`);
assert.equal(sessions.status, 200);
assert.equal(sessions.body.sessions[0].status, "running");
assert.equal(sessions.body.sessions[0].running, true);
assert.equal(sessions.body.sessions[0].terminal, false);
assert.equal(durableQueryCount, 0);
const detail = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}`);
assert.equal(detail.status, 200);
assert.equal(detail.body.session.status, "completed");
assert.equal(detail.body.session.running, false);
assert.equal(detail.body.session.terminal, true);
assert.equal(durableQueryCount, 1);
const messages = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages?limit=10`);
assert.equal(messages.status, 200);
assert.equal(messages.body.messages[1].status, "completed");
const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`);
assert.equal(turn.status, 200);
assert.equal(turn.body.status, "completed");
assert.equal(turn.body.turn.status, "completed");
assert.equal(turn.body.turn.running, false);
assert.equal(turn.body.turn.terminal, true);
assert.equal(turn.body.turn.trace.eventCount, 2);
assert.equal(turn.body.projectionStatus, "caught-up");
} finally {
await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}
});
test("workbench read model keeps non-terminal durable tool completed events out of turn status", async () => {
const traceStore = createCodeAgentTraceStore();
const results = createCodeAgentChatResultStore();
const traceId = "trc_workbench_durable_tool_completed_nonterminal";
const session = {
id: "ses_workbench_durable_tool_completed_nonterminal",
projectId: "prj_hwpod_workbench",
agentId: "hwlab-code-agent",
status: "running",
ownerUserId: ACTOR.id,
conversationId: "cnv_workbench_durable_tool_completed_nonterminal",
threadId: "thread-workbench-durable-tool-completed-nonterminal",
lastTraceId: traceId,
updatedAt: "2026-06-18T19:35:51.787Z",
session: {
sessionStatus: "running",
lastTraceId: traceId,
messages: [
{ role: "user", text: "write benchmark", traceId, status: "sent" },
{ role: "agent", text: "", traceId, status: "running" }
],
valuesRedacted: true,
secretMaterialStored: false
}
};
traceStore.append(traceId, { seq: 1, type: "backend_status", status: "running", label: "runner:created", terminal: false });
const durableEvents = [
{ traceId, seq: 1, type: "assistant_message", status: "running", label: "agentrun:assistant:message", terminal: false, message: "正在安装 Python。", createdAt: "2026-06-18T19:35:40.000Z", valuesPrinted: false },
{ traceId, seq: 2, type: "commandExecution", status: "completed", label: "item/commandExecution:completed", terminal: false, command: "ls -ld .", stdout: "OK\n", createdAt: "2026-06-18T19:35:51.787Z", valuesPrinted: false }
];
results.set(traceId, {
status: "running",
traceId,
ownerUserId: ACTOR.id,
conversationId: session.conversationId,
sessionId: session.id,
threadId: session.threadId,
finalResponse: null,
assistantText: null,
agentRun: { runId: "run_workbench_nonterminal_tool_completed", commandId: "cmd_workbench_nonterminal_tool_completed", status: "runner-job-created", runStatus: "pending", commandState: "pending" }
});
const accessController = {
store: {
async listAgentSessionsForUser() { return [session]; },
async getAgentSession(sessionId) { return sessionId === session.id ? session : null; },
async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; }
},
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
const runtimeStore = {
async queryAgentTraceEvents(params = {}) {
assert.equal(params.traceId, traceId);
return { events: durableEvents, count: durableEvents.length };
}
};
const server = createCloudApiServer({ accessController, traceStore, runtimeStore, codeAgentChatResults: results });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const sessions = await getJson(port, `/v1/workbench/sessions?includeSessionId=${encodeURIComponent(session.id)}`);
assert.equal(sessions.status, 200);
assert.equal(sessions.body.sessions[0].status, "running");
assert.equal(sessions.body.sessions[0].running, true);
assert.equal(sessions.body.sessions[0].terminal, false);
const messages = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages?limit=10`);
assert.equal(messages.status, 200);
assert.equal(messages.body.messages[1].status, "running");
assert.equal(messages.body.messages[1].text, "");
const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`);
assert.equal(turn.status, 200);
assert.equal(turn.body.status, "running");
assert.equal(turn.body.turn.status, "running");
assert.equal(turn.body.turn.running, true);
assert.equal(turn.body.turn.terminal, false);
assert.equal(turn.body.turn.finalResponse, null);
assert.equal(turn.body.projectionStatus, "projecting");
const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?limit=10`);
assert.equal(trace.status, 200);
assert.equal(trace.body.events.at(-1).status, "completed");
assert.equal(trace.body.events.at(-1).terminal, false);
assert.equal(trace.body.projectionStatus, "projecting");
} finally {
await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}
});
test("workbench realtime stream accepts session authority without project or workspace", async () => {
const traceStore = createCodeAgentTraceStore();
const results = createCodeAgentChatResultStore();
const traceId = "trc_workbench_realtime";
const session = {
id: "ses_workbench_realtime",
projectId: "prj_hwpod_workbench",
agentId: "hwlab-code-agent",
status: "running",
ownerUserId: ACTOR.id,
conversationId: "cnv_workbench_realtime",
threadId: "thread-workbench-realtime",
lastTraceId: traceId,
updatedAt: "2026-06-17T02:00:00.000Z",
session: { sessionStatus: "running", lastTraceId: traceId, messages: [{ role: "user", text: "stream", traceId }] }
};
traceStore.append(traceId, { type: "request", status: "accepted", label: "request:accepted" });
results.set(traceId, { status: "running", traceId, ownerUserId: ACTOR.id, sessionId: session.id, threadId: session.threadId });
const accessController = {
store: {
async getAgentSession(sessionId) { return sessionId === session.id ? session : null; },
async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; }
},
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: results, env: { HWLAB_WORKBENCH_SSE_HEARTBEAT_MS: "60000" } });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const eventsPromise = getSseEvents(port, `/v1/workbench/events?sessionId=${encodeURIComponent(session.id)}`, 4);
setTimeout(() => traceStore.append(traceId, { type: "backend", status: "running", label: "backend:live" }), 50);
const events = await eventsPromise;
assert.deepEqual(events.slice(0, 3).map((event) => event.event), [
"workbench.connected",
"workbench.trace.snapshot",
"workbench.turn.snapshot"
]);
assert.equal(events[0].data.filters.sessionId, session.id);
assert.equal(events[1].data.traceId, traceId);
assert.equal(events[2].data.turn.traceId, traceId);
assert.equal(events[3].event, "workbench.trace.event");
assert.equal(events[3].data.event.label, "backend:live");
} finally {
await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}
});
test("workbench read model exposes runtime trace projection query failures as projection blockers", async () => {
const traceStore = createCodeAgentTraceStore();
const results = createCodeAgentChatResultStore();
const traceId = "trc_workbench_projection_store_unavailable";
const session = {
id: "ses_workbench_projection_store_unavailable",
projectId: "prj_hwpod_workbench",
agentId: "hwlab-code-agent",
status: "running",
ownerUserId: ACTOR.id,
conversationId: "cnv_workbench_projection_store_unavailable",
threadId: "thread-workbench-projection-store-unavailable",
lastTraceId: traceId,
updatedAt: "2026-06-18T01:10:00.000Z",
session: {
sessionStatus: "running",
lastTraceId: traceId,
messages: [
{ role: "user", text: "projection store unavailable", traceId, status: "sent", createdAt: "2026-06-18T01:09:58.000Z" },
{ role: "agent", text: "", traceId, status: "running", createdAt: "2026-06-18T01:09:59.000Z" }
],
valuesRedacted: true,
secretMaterialStored: false
}
};
traceStore.append(traceId, { seq: 1, type: "backend", status: "running", label: "runner:created", terminal: false, createdAt: "2026-06-18T01:10:00.000Z" });
results.set(traceId, { status: "running", traceId, ownerUserId: ACTOR.id, sessionId: session.id, threadId: session.threadId });
const accessController = {
store: {
async getAgentSession(sessionId) { return sessionId === session.id ? session : null; },
async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; }
},
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
const runtimeStore = {
async queryAgentTraceEvents() {
const error = new Error("runtime query failed");
error.code = "TEST_RUNTIME_QUERY_FAILED";
throw error;
}
};
const server = createCloudApiServer({ accessController, traceStore, runtimeStore, codeAgentChatResults: results });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`);
assert.equal(turn.status, 200);
assert.equal(turn.body.turn.status, "running");
assert.equal(turn.body.projectionHealth, "unavailable");
assert.equal(turn.body.projection.blocker.code, "projection_store_unavailable");
assert.equal(turn.body.blocker.code, "projection_store_unavailable");
const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?limit=10`);
assert.equal(trace.status, 200);
assert.equal(trace.body.events.length, 1);
assert.equal(trace.body.traceStatus, "running");
assert.equal(trace.body.projectionHealth, "unavailable");
assert.equal(trace.body.projection.blocker.code, "projection_store_unavailable");
} finally {
await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}
});
async function getJson(port, path) {
const response = await fetch(`http://127.0.0.1:${port}${path}`);
return {
status: response.status,
body: await response.json()
};
}
async function getSseEvents(port, path, count) {
const controller = new AbortController();
const response = await fetch(`http://127.0.0.1:${port}${path}`, { signal: controller.signal });
assert.equal(response.status, 200);
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
const events = [];
try {
while (events.length < count) {
const { value, done } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
for (;;) {
const index = buffer.indexOf("\n\n");
if (index < 0) break;
const block = buffer.slice(0, index);
buffer = buffer.slice(index + 2);
events.push(parseSseBlock(block));
if (events.length >= count) break;
}
}
} finally {
controller.abort();
}
return events;
}
function parseSseBlock(block) {
const lines = block.split("\n");
const event = lines.find((line) => line.startsWith("event:"))?.slice(6).trim() ?? "message";
const data = lines.filter((line) => line.startsWith("data:")).map((line) => line.slice(5).trim()).join("\n");
return { event, data: JSON.parse(data) };
}
Reference in New Issue View Git Blame Copy Permalink