fix: thin workbench session list projection

This commit is contained in:
lyon
2026-06-20 13:24:49 +08:00
parent 96704ea1d7
commit 6468263672
6 changed files with 161 additions and 3 deletions
@@ -1046,6 +1046,8 @@ test("workbench session list overlaps include lookup with the page query", async
assert.equal(sessions.status, 200);
assert.equal(sessions.body.sessions[0].sessionId, included.id);
assert.equal(sessions.body.hasMore, true);
assert.equal(queries.find((query) => query.sessionOrder === "updated_desc")?.sessionProjection, "summary");
assert.equal(queries.find((query) => query.sessionId === included.id)?.sessionProjection, undefined);
} finally {
pageReleased = true;
releasePage();
+1 -1
View File
@@ -322,7 +322,7 @@ async function handleWorkbenchSessionList(response, url, options, actor) {
const includeSessionId = safeSessionId(url.searchParams.get("includeSessionId"));
const includeRouteId = includeSessionId ?? safeConversationId(url.searchParams.get("includeSessionId"));
const readModel = createWorkbenchReadModel(options, actor);
const pageQuery = { ownerUserId: actor.role === "admin" ? undefined : actor.id, limit: limit + offset + 1, families: WORKBENCH_SESSION_LIST_PAGE_FAMILIES, sessionOrder: "updated_desc" };
const pageQuery = { ownerUserId: actor.role === "admin" ? undefined : actor.id, limit: limit + offset + 1, families: WORKBENCH_SESSION_LIST_PAGE_FAMILIES, sessionOrder: "updated_desc", sessionProjection: "summary" };
const includeQueryPromise = includeRouteId ? queryWorkbenchSessionInclude(readModel, includeRouteId, perf) : null;
const pageResult = await (perf ? perf.measure("workbench_session_page_query", () => readModel.queryFacts(pageQuery)) : readModel.queryFacts(pageQuery));
if (pageResult.error) return sendJson(response, 503, workbenchProjectionStoreError(pageResult.error));
+90
View File
@@ -868,6 +868,57 @@ test("configured postgres runtime reuses proven readiness for Workbench durable
assert.equal(queryClient.calls.filter((call) => call.sql.startsWith("SELECT session_json FROM workbench_sessions")).length, 2);
});
test("configured postgres runtime can query thin Workbench session summaries", async () => {
const queryClient = createFakePostgresClient({ migrationReady: true });
const store = createConfiguredCloudRuntimeStore({
env: {
HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres",
HWLAB_CLOUD_RUNTIME_DURABLE: "true"
},
dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab",
queryClient,
now: () => "2026-06-20T10:00:00.000Z"
});
await store.writeWorkbenchFacts({
facts: {
sessions: [{
sessionId: "ses_summary_projection",
ownerUserId: "usr_summary_projection",
projectId: "prj_summary_projection",
conversationId: "cnv_summary_projection",
threadId: "thr_summary_projection",
status: "running",
lastTraceId: "trc_summary_projection",
projectedSeq: 7,
sourceSeq: 7,
sourceEventId: "evt_summary_projection",
terminal: false,
sealed: false,
sessionJson: {
providerProfile: "codex-fast",
messages: [{ role: "user", text: "large payload".repeat(500) }],
prompt: "should stay out of list projection"
},
createdAt: "2026-06-20T09:59:00.000Z",
updatedAt: "2026-06-20T10:00:00.000Z"
}]
}
});
queryClient.calls.length = 0;
const loaded = await store.queryWorkbenchFacts({ families: ["sessions"], sessionProjection: "summary", limit: 10 });
const session = loaded.facts.sessions[0];
const readCall = queryClient.calls.find((call) => call.sql.includes("FROM workbench_sessions"));
assert.equal(session.sessionId, "ses_summary_projection");
assert.equal(session.providerProfile, "codex-fast");
assert.deepEqual(session.sessionJson, { providerProfile: "codex-fast", valuesRedacted: true, secretMaterialStored: false });
assert.equal(session.sessionJson.messages, undefined);
assert.ok(readCall?.sql.startsWith("SELECT session_id, owner_user_id"));
assert.equal(readCall.sql.includes("SELECT session_json FROM workbench_sessions"), false);
assert.ok(readCall.sql.includes("providerProfile"));
});
test("configured postgres runtime queries requested workbench fact families concurrently", async () => {
const expectedReads = 4;
let started = 0;
@@ -1140,6 +1191,10 @@ function createFakePostgresClient({
rows.sort((left, right) => String(left.updated_at).localeCompare(String(right.updated_at)) || String(left.trace_id).localeCompare(String(right.trace_id)));
return { rows: rows.map((record) => ({ projection_json: record.projection_json })) };
}
if (sql.startsWith("SELECT session_id, owner_user_id") && sql.includes("FROM workbench_sessions")) {
if (beforeWorkbenchFactRead) await beforeWorkbenchFactRead(sql, params);
return workbenchSessionSummaryRows(state.workbench_sessions, sql, params, readErrorCode);
}
if (sql.startsWith("SELECT session_json FROM workbench_sessions")) {
if (beforeWorkbenchFactRead) await beforeWorkbenchFactRead(sql, params);
return workbenchFactRows(state.workbench_sessions, "session_json", sql, params, readErrorCode);
@@ -1310,6 +1365,41 @@ function workbenchFactRows(map, jsonColumn, sql, params, readErrorCode) {
return { rows: Number.isInteger(limit) && limit > 0 ? rows.slice(0, limit) : rows };
}
function workbenchSessionSummaryRows(map, sql, params, readErrorCode) {
if (readErrorCode) {
const error = new Error("workbench fact read query failed");
error.code = readErrorCode;
throw error;
}
const direction = sql.includes("ORDER BY updated_at DESC") ? -1 : 1;
const rows = [...map.values()]
.filter((record) => sqlRecordMatches(record, sql, params))
.sort((left, right) => direction * (String(left.updated_at).localeCompare(String(right.updated_at)) || String(left.session_id).localeCompare(String(right.session_id))))
.map((record) => {
const session = JSON.parse(record.session_json || "{}");
const providerProfile = session.providerProfile ?? session.sessionJson?.providerProfile ?? null;
return {
session_id: record.session_id,
owner_user_id: record.owner_user_id,
project_id: record.project_id,
conversation_id: record.conversation_id,
thread_id: record.thread_id,
status: record.status,
last_trace_id: record.last_trace_id,
projected_seq: record.projected_seq,
source_seq: record.source_seq,
source_event_id: record.source_event_id,
terminal: record.terminal,
sealed: record.sealed,
created_at: record.created_at,
updated_at: record.updated_at,
provider_profile: providerProfile
};
});
const limit = sql.includes(" LIMIT $") ? Number(params.at(-1)) : null;
return { rows: Number.isInteger(limit) && limit > 0 ? rows.slice(0, limit) : rows };
}
function sqlRecordMatches(record, sql, params) {
for (const column of Object.keys(record)) {
const match = sql.match(new RegExp(`(?:^|[^a-z_])${column} = \\$(\\d+)`, "u"));
+56 -1
View File
@@ -1326,12 +1326,14 @@ export class PostgresCloudRuntimeStore {
return result.rows.map((row) => parseJsonColumn(row[jsonColumn], null)).filter(Boolean);
}
const orderDirection = workbenchFactOrder(params, family) === "updated_desc" ? "DESC" : "ASC";
let sql = `SELECT ${jsonColumn} FROM ${table}${clauses.length ? ` WHERE ${clauses.join(" AND ")}` : ""} ORDER BY updated_at ${orderDirection}`;
const useSessionSummaryProjection = table === "workbench_sessions" && params.sessionProjection === "summary";
let sql = `SELECT ${useSessionSummaryProjection ? workbenchSessionSummarySelectClause() : jsonColumn} FROM ${table}${clauses.length ? ` WHERE ${clauses.join(" AND ")}` : ""} ORDER BY updated_at ${orderDirection}`;
if (limit) {
queryParams.push(limit);
sql += ` LIMIT $${queryParams.length}`;
}
const result = await this.queryDurableReadRows(method, sql, queryParams);
if (useSessionSummaryProjection) return result.rows.map(workbenchSessionSummaryFactFromRow).filter(Boolean);
return result.rows.map((row) => parseJsonColumn(row[jsonColumn], null)).filter(Boolean);
}
@@ -2227,6 +2229,59 @@ function workbenchFactFamilyForTable(table) {
return "unknown";
}
function workbenchSessionSummarySelectClause() {
return [
"session_id",
"owner_user_id",
"project_id",
"conversation_id",
"thread_id",
"status",
"last_trace_id",
"projected_seq",
"source_seq",
"source_event_id",
"terminal",
"sealed",
"created_at",
"updated_at",
"COALESCE(NULLIF(session_json::jsonb ->> 'providerProfile', ''), NULLIF(session_json::jsonb #>> '{sessionJson,providerProfile}', '')) AS provider_profile"
].join(", ");
}
function workbenchSessionSummaryFactFromRow(row = {}) {
const sessionId = textOr(row.session_id, "");
if (!sessionId) return null;
const providerProfile = textOr(row.provider_profile, "");
return {
id: sessionId,
sessionId,
ownerUserId: textOr(row.owner_user_id, null),
projectId: textOr(row.project_id, null),
conversationId: textOr(row.conversation_id, null),
threadId: textOr(row.thread_id, null),
status: textOr(row.status, "unknown"),
lastTraceId: textOr(row.last_trace_id, null),
projectedSeq: nonNegativeInteger(row.projected_seq),
sourceSeq: nonNegativeInteger(row.source_seq),
sourceEventId: textOr(row.source_event_id, null),
terminal: row.terminal === true || row.terminal === "t",
sealed: row.sealed === true || row.sealed === "t",
...(providerProfile ? {
providerProfile,
sessionJson: {
providerProfile,
valuesRedacted: true,
secretMaterialStored: false
}
} : {}),
createdAt: textOr(row.created_at, null),
updatedAt: textOr(row.updated_at, row.created_at ?? null),
valuesPrinted: false,
valuesRedacted: true
};
}
function workbenchFactOrder(params = {}, family = "") {
const value = params[`${family}Order`] ?? params.order;
return value === "updated_desc" ? "updated_desc" : "updated_asc";
@@ -64,6 +64,17 @@ test("Workbench submit failures do not pollute first visible output SLI", () =>
assert.ok(events.some((event) => event.kind === "workbench_journey" && event.journey === "submit_to_failure" && event.outcome === "network"));
});
test("Workbench session switch treats visible empty sessions as successful first paint", () => {
resetWorkbenchPerformanceForTest();
startWorkbenchSessionSwitch({ sessionId: "ses_empty_visible", source: "rail", targetState: "empty", cache: "cold" });
acknowledgeWorkbenchVisible({ messages: [], activeSessionId: "ses_empty_visible", detailLoading: false });
const events = drainWorkbenchPerformanceEventsForTest();
const switchEvent = events.find((event) => event.kind === "workbench_journey" && event.journey === "session_switch_first_visible");
assert.equal(switchEvent?.outcome, "ok");
assert.equal(switchEvent?.targetState, "empty");
});
test("Workbench submit first visible waits for terminal final or tool output", () => {
resetWorkbenchPerformanceForTest();
const wallBase = Date.now();
@@ -375,7 +375,7 @@ export function acknowledgeWorkbenchVisible(input: { messages: ChatMessage[]; ac
const session = activeSessionId ? sessionSwitches.get(activeSessionId) : null;
if (session && !session.firstVisibleReported && !input.detailLoading) {
session.firstVisibleReported = true;
enqueue({ kind: "workbench_journey", journey: "session_switch_first_visible", route: session.route, source: session.source, targetState: session.targetState, cache: session.cache, visibility: visibilityState(), outcome: input.messages.length > 0 ? "ok" : "empty", valueMs: Math.max(0, now - session.startAt) });
enqueue({ kind: "workbench_journey", journey: "session_switch_first_visible", route: session.route, source: session.source, targetState: session.targetState, cache: session.cache, visibility: visibilityState(), outcome: "ok", valueMs: Math.max(0, now - session.startAt) });
}
for (const message of input.messages) {
const traceId = safeText(message.traceId ?? message.runnerTrace?.traceId);