diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index d4679119..ad85fffd 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/internal/cloud/server-workbench-http.test.ts @@ -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) { diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 97caea78..75e1d381 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -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 })); diff --git a/internal/db/runtime-store.ts b/internal/db/runtime-store.ts index 26d66788..45c2bd3a 100644 --- a/internal/db/runtime-store.ts +++ b/internal/db/runtime-store.ts @@ -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; }