fix: reduce workbench read-model fact scans (#1676)
This commit is contained in:
Lyon
@@ -313,11 +313,17 @@ test("workbench read model exposes session, messages, turn, and trace without wr
|
||||
assert.equal(sessions.body.sessions[0].turnSummary.traceId, traceId);
|
||||
assert.equal(factQueries[0].projectId, undefined);
|
||||
assert.equal(factQueries[0].limit, 21);
|
||||
assert.equal(factQueries[0].sessionOrder, "updated_desc");
|
||||
assert.deepEqual(factQueries[0].families, ["sessions"]);
|
||||
assert.deepEqual(factQueries[1].families, ["messages", "parts", "turns"]);
|
||||
assert.deepEqual(factQueries[1].sessionIds, [session.id]);
|
||||
assert.deepEqual(factQueries[2].families, ["checkpoints"]);
|
||||
assert.deepEqual(factQueries[2].traceIds, [traceId]);
|
||||
|
||||
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(factQueries.length, 1);
|
||||
assert.equal(factQueries.length, 3);
|
||||
|
||||
const detail = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}`);
|
||||
assert.equal(detail.status, 200);
|
||||
@@ -1369,21 +1375,44 @@ function normalizeTestEvents(events, session, { traceId }) {
|
||||
}
|
||||
|
||||
function filterFacts(facts, params = {}) {
|
||||
const families = testFactFamilySet(params);
|
||||
const filtered = {
|
||||
sessions: facts.sessions.filter((record) => matchesFact(record, params, ["sessionId", "ownerUserId", "projectId", "conversationId", "threadId", "status"]) && (!params.traceId || record.lastTraceId === params.traceId)),
|
||||
messages: facts.messages.filter((record) => matchesFact(record, params, ["messageId", "sessionId", "turnId", "traceId", "role", "status"])),
|
||||
parts: facts.parts.filter((record) => matchesFact(record, params, ["partId", "messageId", "sessionId", "turnId", "traceId", "partType", "status"])),
|
||||
turns: facts.turns.filter((record) => matchesFact(record, params, ["turnId", "sessionId", "traceId", "messageId", "status"])),
|
||||
traceEvents: facts.traceEvents.filter((record) => matchesFact(record, params, ["id", "traceId", "sessionId", "turnId", "messageId", "eventType"])),
|
||||
checkpoints: facts.checkpoints.filter((record) => matchesFact(record, params, ["traceId", "sessionId", "turnId", "runId", "commandId", "projectionStatus", "projectionHealth"]))
|
||||
sessions: families.has("sessions") ? facts.sessions.filter((record) => matchesFact(record, params, ["sessionId", "ownerUserId", "projectId", "conversationId", "threadId", "status"]) && matchesTraceFact(record.lastTraceId, params)) : [],
|
||||
messages: families.has("messages") ? facts.messages.filter((record) => matchesFact(record, params, ["messageId", "sessionId", "turnId", "traceId", "role", "status"])) : [],
|
||||
parts: families.has("parts") ? facts.parts.filter((record) => matchesFact(record, params, ["partId", "messageId", "sessionId", "turnId", "traceId", "partType", "status"])) : [],
|
||||
turns: families.has("turns") ? facts.turns.filter((record) => matchesFact(record, params, ["turnId", "sessionId", "traceId", "messageId", "status"])) : [],
|
||||
traceEvents: families.has("traceEvents") ? facts.traceEvents.filter((record) => matchesFact(record, params, ["id", "traceId", "sessionId", "turnId", "messageId", "eventType"])) : [],
|
||||
checkpoints: families.has("checkpoints") ? facts.checkpoints.filter((record) => matchesFact(record, params, ["traceId", "sessionId", "turnId", "runId", "commandId", "projectionStatus", "projectionHealth"])) : []
|
||||
};
|
||||
if (params.sessionOrder === "updated_desc") {
|
||||
filtered.sessions.sort((left, right) => String(right.updatedAt ?? "").localeCompare(String(left.updatedAt ?? "")) || String(right.sessionId ?? "").localeCompare(String(left.sessionId ?? "")));
|
||||
}
|
||||
const limit = Number.parseInt(params.limit ?? "", 10);
|
||||
if (!Number.isInteger(limit) || limit <= 0) return filtered;
|
||||
return Object.fromEntries(Object.entries(filtered).map(([key, value]) => [key, value.slice(0, limit)]));
|
||||
}
|
||||
|
||||
function matchesFact(record, params, fields) {
|
||||
return fields.every((field) => params[field] === undefined || params[field] === null || record[field] === params[field]);
|
||||
return fields.every((field) => {
|
||||
if (params[field] !== undefined && params[field] !== null && record[field] !== params[field]) return false;
|
||||
const values = Array.isArray(params[`${field}s`]) ? params[`${field}s`].map((item) => String(item ?? "")).filter(Boolean) : [];
|
||||
return values.length === 0 || values.includes(String(record[field] ?? ""));
|
||||
});
|
||||
}
|
||||
|
||||
function matchesTraceFact(value, params) {
|
||||
if (params.traceId && value !== params.traceId) return false;
|
||||
const traceIds = Array.isArray(params.traceIds) ? params.traceIds.map((item) => String(item ?? "")).filter(Boolean) : [];
|
||||
return traceIds.length === 0 || traceIds.includes(String(value ?? ""));
|
||||
}
|
||||
|
||||
function testFactFamilySet(params = {}) {
|
||||
const all = ["sessions", "messages", "parts", "turns", "traceEvents", "checkpoints"];
|
||||
const raw = params.families ?? params.factFamilies;
|
||||
if (!raw) return new Set(all);
|
||||
const values = Array.isArray(raw) ? raw : String(raw).split(",");
|
||||
const selected = values.map((item) => String(item ?? "").trim()).filter((item) => all.includes(item));
|
||||
return new Set(selected.length ? selected : all);
|
||||
}
|
||||
|
||||
function mergeFacts(target, source) {
|
||||
|
||||
@@ -20,6 +20,8 @@ const DEFAULT_PAGE_LIMIT = 50;
|
||||
const DEFAULT_SESSION_LIST_LIMIT = 20;
|
||||
const MAX_PAGE_LIMIT = 100;
|
||||
const DEFAULT_WORKBENCH_SSE_HEARTBEAT_MS = 15000;
|
||||
const WORKBENCH_SESSION_LIST_PAGE_FAMILIES = Object.freeze(["sessions"]);
|
||||
const WORKBENCH_SESSION_DETAIL_FAMILIES = Object.freeze(["sessions", "messages", "parts", "turns", "checkpoints"]);
|
||||
export async function handleWorkbenchReadModelHttp(request, response, url, options = {}) {
|
||||
try {
|
||||
const perf = options.backendPerformance;
|
||||
@@ -318,12 +320,13 @@ 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 pageResult = await (perf ? perf.measure("workbench_session_page_query", () => readModel.queryFacts({ ownerUserId: actor.role === "admin" ? undefined : actor.id, limit: limit + offset + 1 })) : readModel.queryFacts({ ownerUserId: actor.role === "admin" ? undefined : actor.id, limit: limit + offset + 1 }));
|
||||
const pageQuery = { ownerUserId: actor.role === "admin" ? undefined : actor.id, limit: limit + offset + 1, families: WORKBENCH_SESSION_LIST_PAGE_FAMILIES, sessionOrder: "updated_desc" };
|
||||
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));
|
||||
let pageFacts = pageResult.facts;
|
||||
let naturalPage = visibleFactSessions(pageFacts, actor).slice(offset, offset + limit + 1);
|
||||
if (includeRouteId && !naturalPage.some((session) => factSessionMatchesRouteId(session, includeRouteId))) {
|
||||
const includeResult = await (perf ? perf.measure("workbench_session_include_query", () => queryFactsByRouteId(readModel, includeRouteId)) : queryFactsByRouteId(readModel, includeRouteId));
|
||||
const includeResult = await (perf ? perf.measure("workbench_session_include_query", () => queryFactsByRouteId(readModel, includeRouteId, { families: WORKBENCH_SESSION_LIST_PAGE_FAMILIES })) : queryFactsByRouteId(readModel, includeRouteId, { families: WORKBENCH_SESSION_LIST_PAGE_FAMILIES }));
|
||||
if (includeResult.error) return sendJson(response, 503, workbenchProjectionStoreError(includeResult.error));
|
||||
const included = visibleFactSessions(includeResult.facts, actor)[0] ?? null;
|
||||
if (included) {
|
||||
@@ -332,6 +335,9 @@ async function handleWorkbenchSessionList(response, url, options, actor) {
|
||||
}
|
||||
}
|
||||
const pageSessions = naturalPage.slice(0, limit);
|
||||
const summaryResult = await (perf ? perf.measure("workbench_session_summary_query", () => queryFactsForSessionSummaries(readModel, pageSessions)) : queryFactsForSessionSummaries(readModel, pageSessions));
|
||||
if (summaryResult.error) return sendJson(response, 503, workbenchProjectionStoreError(summaryResult.error));
|
||||
pageFacts = mergeFactSets(pageFacts, summaryResult.facts);
|
||||
const summaries = pageSessions.map((session) => factSessionSummary(session, pageFacts)).filter(Boolean);
|
||||
const hasMore = naturalPage.length > limit;
|
||||
sendJson(response, 200, {
|
||||
@@ -355,14 +361,45 @@ function sessionMatchesRouteId(session, routeId) {
|
||||
return Boolean(conversationId && session?.conversationId === conversationId);
|
||||
}
|
||||
|
||||
async function queryFactsByRouteId(readModel, routeId) {
|
||||
async function queryFactsByRouteId(readModel, routeId, query = {}) {
|
||||
const baseQuery = { ...query, limit: query.limit ?? MAX_PAGE_LIMIT };
|
||||
const sessionId = safeSessionId(routeId);
|
||||
if (sessionId) return readModel.queryFacts({ sessionId, limit: MAX_PAGE_LIMIT });
|
||||
if (sessionId) return readModel.queryFacts({ ...baseQuery, sessionId });
|
||||
const conversationId = safeConversationId(routeId);
|
||||
if (conversationId) return readModel.queryFacts({ conversationId, limit: MAX_PAGE_LIMIT });
|
||||
if (conversationId) {
|
||||
const sessionResult = await readModel.queryFacts({ ...baseQuery, families: WORKBENCH_SESSION_LIST_PAGE_FAMILIES, conversationId });
|
||||
if (sessionResult.error || queryRequestsOnlySessionFacts(baseQuery)) return sessionResult;
|
||||
const resolvedSessionId = factSessionId(factArray(sessionResult.facts?.sessions)[0] ?? null);
|
||||
if (!resolvedSessionId) return sessionResult;
|
||||
const relatedResult = await readModel.queryFacts({ ...baseQuery, sessionId: resolvedSessionId });
|
||||
if (relatedResult.error) return relatedResult;
|
||||
const facts = mergeFactSets(sessionResult.facts, relatedResult.facts);
|
||||
return { facts, count: Object.values(facts).reduce((sum, rows) => sum + rows.length, 0), durable: relatedResult.durable ?? sessionResult.durable, persistence: relatedResult.persistence ?? sessionResult.persistence ?? null };
|
||||
}
|
||||
return { facts: emptyFactSet(), count: 0, durable: false, persistence: null };
|
||||
}
|
||||
|
||||
function queryRequestsOnlySessionFacts(query = {}) {
|
||||
const families = Array.isArray(query.families) ? query.families : [];
|
||||
return families.length === 1 && families[0] === "sessions";
|
||||
}
|
||||
|
||||
async function queryFactsForSessionSummaries(readModel, sessions = []) {
|
||||
const sessionIds = uniqueText(sessions.map(factSessionId));
|
||||
const traceIds = uniqueText(sessions.map(factLastTraceId));
|
||||
if (sessionIds.length === 0 && traceIds.length === 0) return { facts: emptyFactSet(), count: 0, durable: false, persistence: null };
|
||||
const bySession = sessionIds.length > 0
|
||||
? await readModel.queryFacts({ sessionIds, families: ["messages", "parts", "turns"] })
|
||||
: { facts: emptyFactSet(), count: 0, durable: false, persistence: null };
|
||||
if (bySession.error) return bySession;
|
||||
const byTrace = traceIds.length > 0
|
||||
? await readModel.queryFacts({ traceIds, families: ["checkpoints"] })
|
||||
: { facts: emptyFactSet(), count: 0, durable: false, persistence: bySession.persistence ?? null };
|
||||
if (byTrace.error) return byTrace;
|
||||
const facts = mergeFactSets(bySession.facts, byTrace.facts);
|
||||
return { facts, count: Object.values(facts).reduce((sum, rows) => sum + rows.length, 0), durable: bySession.durable ?? byTrace.durable, persistence: bySession.persistence ?? byTrace.persistence ?? null };
|
||||
}
|
||||
|
||||
function visibleFactSessions(facts, actor) {
|
||||
return factArray(facts?.sessions)
|
||||
.filter((session) => canActorReadFactSession(session, actor))
|
||||
@@ -766,9 +803,13 @@ function factArray(value) {
|
||||
return Array.isArray(value) ? value : [];
|
||||
}
|
||||
|
||||
function uniqueText(values = []) {
|
||||
return [...new Set(values.map((value) => textValue(value)).filter(Boolean))];
|
||||
}
|
||||
|
||||
async function handleWorkbenchSessionDetail(response, options, actor, sessionId) {
|
||||
const readModel = createWorkbenchReadModel(options, actor);
|
||||
const result = await queryFactsByRouteId(readModel, sessionId);
|
||||
const result = await queryFactsByRouteId(readModel, sessionId, { families: WORKBENCH_SESSION_DETAIL_FAMILIES });
|
||||
if (result.error) return sendJson(response, 503, workbenchProjectionStoreError(result.error));
|
||||
const session = visibleFactSessions(result.facts, actor)[0] ?? null;
|
||||
if (!session) return sendJson(response, 404, workbenchError("workbench_session_not_found", "Workbench session is not visible to the current actor.", { sessionId }));
|
||||
@@ -784,7 +825,7 @@ async function handleWorkbenchSessionDetail(response, options, actor, sessionId)
|
||||
|
||||
async function handleWorkbenchMessagePage(response, url, options, actor, sessionId) {
|
||||
const readModel = createWorkbenchReadModel(options, actor);
|
||||
const result = await queryFactsByRouteId(readModel, sessionId);
|
||||
const result = await queryFactsByRouteId(readModel, sessionId, { families: WORKBENCH_SESSION_DETAIL_FAMILIES });
|
||||
if (result.error) return sendJson(response, 503, workbenchProjectionStoreError(result.error));
|
||||
const session = visibleFactSessions(result.facts, actor)[0] ?? null;
|
||||
if (!session) return sendJson(response, 404, workbenchError("workbench_session_not_found", "Workbench session is not visible to the current actor.", { sessionId }));
|
||||
|
||||
@@ -69,6 +69,7 @@ const postgresRuntimeReadIndexes = Object.freeze([
|
||||
"CREATE INDEX IF NOT EXISTS idx_workbench_projection_state_status_retry_updated ON workbench_projection_state(projection_status, next_retry_at, updated_at)",
|
||||
"CREATE INDEX IF NOT EXISTS idx_workbench_projection_state_run_command ON workbench_projection_state(run_id, command_id)",
|
||||
"CREATE INDEX IF NOT EXISTS idx_workbench_projection_state_session_updated ON workbench_projection_state(session_id, updated_at DESC)",
|
||||
"CREATE INDEX IF NOT EXISTS idx_workbench_sessions_updated ON workbench_sessions(updated_at DESC, session_id)",
|
||||
"CREATE INDEX IF NOT EXISTS idx_workbench_sessions_owner_updated ON workbench_sessions(owner_user_id, updated_at DESC, session_id)",
|
||||
"CREATE INDEX IF NOT EXISTS idx_workbench_sessions_status_updated ON workbench_sessions(status, updated_at DESC, session_id)",
|
||||
"CREATE INDEX IF NOT EXISTS idx_workbench_messages_session_updated ON workbench_messages(session_id, updated_at ASC, message_id)",
|
||||
@@ -604,6 +605,7 @@ export class CloudRuntimeStore {
|
||||
}
|
||||
|
||||
queryWorkbenchFacts(params = {}) {
|
||||
const families = workbenchFactFamilySet(params);
|
||||
const sessions = [...this.workbenchSessions.values()].filter((record) => matchesWorkbenchSessionFactQuery(record, params));
|
||||
const messages = [...this.workbenchMessages.values()].filter((record) => matchesQuery(record, params, ["messageId", "sessionId", "turnId", "traceId", "role", "status"]));
|
||||
const parts = [...this.workbenchParts.values()].filter((record) => matchesQuery(record, params, ["partId", "messageId", "sessionId", "turnId", "traceId", "partType", "status"]));
|
||||
@@ -611,12 +613,12 @@ export class CloudRuntimeStore {
|
||||
const traceEvents = [...this.workbenchTraceEvents.values()].filter((record) => matchesQuery(record, params, ["id", "traceId", "sessionId", "turnId", "messageId", "eventType"]));
|
||||
const checkpoints = [...this.workbenchProjectionCheckpoints.values()].filter((record) => matchesQuery(record, params, ["traceId", "sessionId", "turnId", "runId", "commandId", "projectionStatus", "projectionHealth"]));
|
||||
const facts = {
|
||||
sessions: limitResults(sessions.sort(compareWorkbenchFactRecords), params.limit),
|
||||
messages: limitResults(messages.sort(compareWorkbenchFactRecords), params.limit),
|
||||
parts: limitResults(parts.sort(compareWorkbenchPartFactRecords), params.limit),
|
||||
turns: limitResults(turns.sort(compareWorkbenchFactRecords), params.limit),
|
||||
traceEvents: limitResults(traceEvents.sort(compareWorkbenchTraceEventFacts), params.limit),
|
||||
checkpoints: limitResults(checkpoints.sort(compareWorkbenchFactRecords), params.limit)
|
||||
sessions: families.has("sessions") ? limitResults(sortWorkbenchFactRows(sessions, params, "sessions", compareWorkbenchFactRecords), params.limit) : [],
|
||||
messages: families.has("messages") ? limitResults(sortWorkbenchFactRows(messages, params, "messages", compareWorkbenchFactRecords), params.limit) : [],
|
||||
parts: families.has("parts") ? limitResults(sortWorkbenchFactRows(parts, params, "parts", compareWorkbenchPartFactRecords), params.limit) : [],
|
||||
turns: families.has("turns") ? limitResults(sortWorkbenchFactRows(turns, params, "turns", compareWorkbenchFactRecords), params.limit) : [],
|
||||
traceEvents: families.has("traceEvents") ? limitResults(sortWorkbenchFactRows(traceEvents, params, "traceEvents", compareWorkbenchTraceEventFacts), params.limit) : [],
|
||||
checkpoints: families.has("checkpoints") ? limitResults(sortWorkbenchFactRows(checkpoints, params, "checkpoints", compareWorkbenchFactRecords), params.limit) : []
|
||||
};
|
||||
return {
|
||||
facts,
|
||||
@@ -1212,13 +1214,14 @@ export class PostgresCloudRuntimeStore {
|
||||
|
||||
async queryWorkbenchFacts(params = {}) {
|
||||
await this.assertReadyForDurableReads("workbench.facts.query");
|
||||
const families = workbenchFactFamilySet(params);
|
||||
const facts = {
|
||||
sessions: await this.queryWorkbenchFactRows("workbench.sessions.query", "workbench_sessions", "session_json", params, { sessionId: "session_id", ownerUserId: "owner_user_id", projectId: "project_id", conversationId: "conversation_id", threadId: "thread_id", traceId: "last_trace_id", lastTraceId: "last_trace_id", status: "status" }),
|
||||
messages: await this.queryWorkbenchFactRows("workbench.messages.query", "workbench_messages", "message_json", params, { messageId: "message_id", sessionId: "session_id", turnId: "turn_id", traceId: "trace_id", role: "role", status: "status" }),
|
||||
parts: await this.queryWorkbenchFactRows("workbench.parts.query", "workbench_parts", "part_json", params, { partId: "part_id", messageId: "message_id", sessionId: "session_id", turnId: "turn_id", traceId: "trace_id", partType: "part_type", status: "status" }),
|
||||
turns: await this.queryWorkbenchFactRows("workbench.turns.query", "workbench_turns", "turn_json", params, { turnId: "turn_id", sessionId: "session_id", traceId: "trace_id", messageId: "message_id", status: "status" }),
|
||||
traceEvents: await this.queryWorkbenchFactRows("workbench.trace-events.query", "workbench_trace_events", "event_json", params, { id: "id", traceId: "trace_id", sessionId: "session_id", turnId: "turn_id", messageId: "message_id", eventType: "event_type" }),
|
||||
checkpoints: await this.queryWorkbenchFactRows("workbench.projection-checkpoints.query", "workbench_projection_checkpoints", "checkpoint_json", params, { traceId: "trace_id", sessionId: "session_id", turnId: "turn_id", runId: "run_id", commandId: "command_id", projectionStatus: "projection_status", projectionHealth: "projection_health" })
|
||||
sessions: families.has("sessions") ? await this.queryWorkbenchFactRows("workbench.sessions.query", "workbench_sessions", "session_json", params, { sessionId: "session_id", ownerUserId: "owner_user_id", projectId: "project_id", conversationId: "conversation_id", threadId: "thread_id", traceId: "last_trace_id", lastTraceId: "last_trace_id", status: "status" }) : [],
|
||||
messages: families.has("messages") ? await this.queryWorkbenchFactRows("workbench.messages.query", "workbench_messages", "message_json", params, { messageId: "message_id", sessionId: "session_id", turnId: "turn_id", traceId: "trace_id", role: "role", status: "status" }) : [],
|
||||
parts: families.has("parts") ? await this.queryWorkbenchFactRows("workbench.parts.query", "workbench_parts", "part_json", params, { partId: "part_id", messageId: "message_id", sessionId: "session_id", turnId: "turn_id", traceId: "trace_id", partType: "part_type", status: "status" }) : [],
|
||||
turns: families.has("turns") ? await this.queryWorkbenchFactRows("workbench.turns.query", "workbench_turns", "turn_json", params, { turnId: "turn_id", sessionId: "session_id", traceId: "trace_id", messageId: "message_id", status: "status" }) : [],
|
||||
traceEvents: families.has("traceEvents") ? await this.queryWorkbenchFactRows("workbench.trace-events.query", "workbench_trace_events", "event_json", params, { id: "id", traceId: "trace_id", sessionId: "session_id", turnId: "turn_id", messageId: "message_id", eventType: "event_type" }) : [],
|
||||
checkpoints: families.has("checkpoints") ? await this.queryWorkbenchFactRows("workbench.projection-checkpoints.query", "workbench_projection_checkpoints", "checkpoint_json", params, { traceId: "trace_id", sessionId: "session_id", turnId: "turn_id", runId: "run_id", commandId: "command_id", projectionStatus: "projection_status", projectionHealth: "projection_health" }) : []
|
||||
};
|
||||
return withPersistence({ facts, count: Object.values(facts).reduce((sum, rows) => sum + rows.length, 0) }, this.summary());
|
||||
}
|
||||
@@ -1228,12 +1231,20 @@ export class PostgresCloudRuntimeStore {
|
||||
const clauses = [];
|
||||
for (const [field, column] of Object.entries(columnMap)) {
|
||||
const value = textOr(params[field], "");
|
||||
if (!value) continue;
|
||||
queryParams.push(value);
|
||||
clauses.push(`${column} = $${queryParams.length}`);
|
||||
if (value) {
|
||||
queryParams.push(value);
|
||||
clauses.push(`${column} = $${queryParams.length}`);
|
||||
}
|
||||
const values = textList(params[`${field}s`]);
|
||||
if (values.length > 0) {
|
||||
queryParams.push(values);
|
||||
clauses.push(`${column} = ANY($${queryParams.length})`);
|
||||
}
|
||||
}
|
||||
const limit = positiveIntegerOrNull(params.limit);
|
||||
let sql = `SELECT ${jsonColumn} FROM ${table}${clauses.length ? ` WHERE ${clauses.join(" AND ")}` : ""} ORDER BY updated_at ASC`;
|
||||
const family = workbenchFactFamilyForTable(table);
|
||||
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}`;
|
||||
if (limit) {
|
||||
queryParams.push(limit);
|
||||
sql += ` LIMIT $${queryParams.length}`;
|
||||
@@ -2048,6 +2059,39 @@ function compareWorkbenchTraceEventFacts(left, right) {
|
||||
String(left?.id ?? "").localeCompare(String(right?.id ?? ""));
|
||||
}
|
||||
|
||||
function sortWorkbenchFactRows(records, params = {}, family, fallbackCompare) {
|
||||
const sorted = [...records];
|
||||
if (workbenchFactOrder(params, family) === "updated_desc") {
|
||||
return sorted.sort((left, right) => String(right?.updatedAt ?? right?.createdAt ?? "").localeCompare(String(left?.updatedAt ?? left?.createdAt ?? "")) || fallbackCompare(left, right));
|
||||
}
|
||||
return sorted.sort(fallbackCompare);
|
||||
}
|
||||
|
||||
const WORKBENCH_FACT_FAMILIES = Object.freeze(["sessions", "messages", "parts", "turns", "traceEvents", "checkpoints"]);
|
||||
|
||||
function workbenchFactFamilySet(params = {}) {
|
||||
const raw = params.families ?? params.factFamilies;
|
||||
if (raw === undefined || raw === null || raw === "") return new Set(WORKBENCH_FACT_FAMILIES);
|
||||
const values = Array.isArray(raw) ? raw : String(raw).split(",");
|
||||
const selected = values.map((value) => String(value ?? "").trim()).filter((value) => WORKBENCH_FACT_FAMILIES.includes(value));
|
||||
return new Set(selected.length > 0 ? selected : WORKBENCH_FACT_FAMILIES);
|
||||
}
|
||||
|
||||
function workbenchFactFamilyForTable(table) {
|
||||
if (table === "workbench_sessions") return "sessions";
|
||||
if (table === "workbench_messages") return "messages";
|
||||
if (table === "workbench_parts") return "parts";
|
||||
if (table === "workbench_turns") return "turns";
|
||||
if (table === "workbench_trace_events") return "traceEvents";
|
||||
if (table === "workbench_projection_checkpoints") return "checkpoints";
|
||||
return "unknown";
|
||||
}
|
||||
|
||||
function workbenchFactOrder(params = {}, family = "") {
|
||||
const value = params[`${family}Order`] ?? params.order;
|
||||
return value === "updated_desc" ? "updated_desc" : "updated_asc";
|
||||
}
|
||||
|
||||
function traceEventLevel(value) {
|
||||
const text = textOr(value, "info").toLowerCase();
|
||||
if (["failed", "failure", "error", "timeout"].includes(text)) return "error";
|
||||
@@ -2100,6 +2144,11 @@ function textOr(value, fallback) {
|
||||
return text || fallback;
|
||||
}
|
||||
|
||||
function textList(value) {
|
||||
const raw = Array.isArray(value) ? value : [];
|
||||
return [...new Set(raw.map((item) => textOr(item, "")).filter(Boolean))];
|
||||
}
|
||||
|
||||
function asObject(value, label) {
|
||||
if (!value || typeof value !== "object" || Array.isArray(value)) {
|
||||
throw new HwlabProtocolError(`${label} must be a JSON object`, {
|
||||
@@ -2162,6 +2211,10 @@ function matchesQuery(record, params, fields) {
|
||||
if (params[field] !== undefined && params[field] !== null && record[field] !== params[field]) {
|
||||
return false;
|
||||
}
|
||||
const values = textList(params[`${field}s`]);
|
||||
if (values.length > 0 && !values.includes(textOr(record[field], ""))) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user