diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index f47b4d6829e8..d49e4cc637ed 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -28,6 +28,7 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; import * as SqlClient from "effect/sql/SqlClient"; +import * as Statement from "effect/sql/Statement"; import { projectThreadAwarenessV2 } from "@t3tools/shared/agentAwareness"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; @@ -383,6 +384,53 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { assert.deepEqual(yield* store.getTurnStartHistory(threadId, [RunId.make("run:other")]), []); }), ); + it.effect("answers empty record filters without reading the filtered tables", () => + Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const threadId = yield* addRolledBackRecoveryCandidate("empty-record-filters"); + const runId = (yield* store.getThreadProjection(threadId)).runs[0]!.id; + const queries: Array = []; + const recordQueries: Statement.Transformer = (statement) => { + queries.push(statement.compile()[0]); + return Effect.succeed(statement); + }; + const read = (filter: ProjectionStore.ProjectionRecordFilter) => + store + .getThreadRecords(threadId, ["runs", "messages", "turnItems"], filter) + .pipe(Effect.provideService(Statement.CurrentTransformer, recordQueries)); + + const emptyFilters = [ + [{ turnItemTypes: [] }, "turnItems", "turn_items"], + [{ turnItemStatuses: [] }, "turnItems", "turn_items"], + [{ turnItemRunIds: [] }, "turnItems", "turn_items"], + [{ runIds: [] }, "runs", "runs"], + [{ messageIds: [] }, "messages", "messages"], + [{ messageRoles: [] }, "messages", "messages"], + [{ messageRunIds: [] }, "messages", "messages"], + ] as const; + for (const [filter, field, table] of emptyFilters) { + queries.length = 0; + assert.isEmpty((yield* read(filter))[field]); + assert.isFalse( + queries.some((query) => + new RegExp(`FROM orchestration_v2_projection_${table}\\b`).test(query), + ), + `${JSON.stringify(filter)} read ${table}`, + ); + } + + const selected = yield* read({ turnItemTypes: ["command_execution"], runIds: [runId] }); + assert.deepEqual( + selected.turnItems.map((item) => item.type), + ["command_execution"], + ); + assert.deepEqual( + selected.runs.map((run) => run.id), + [runId], + ); + assert.isEmpty((yield* read({ turnItemTypes: ["user_message"] })).turnItems); + }), + ); it.effect("preserves stored provider usage when a terminal update omits it", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 07d847e2b91d..404f3b02c4f2 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -955,6 +955,11 @@ type ShellRunItemCountRow = { const encodeIdList = Schema.encodeSync(Schema.fromJsonString(Schema.Array(Schema.String))); +// An empty list filter selects no rows, but SQLite still walks every row of +// the thread to apply it. Callers pass `turnItemTypes: []` to skip items. +const matchesNothing = (...filters: ReadonlyArray | undefined>) => + filters.some((values) => values?.length === 0); + const encodeThreadPayload = Schema.encodeEffect( Schema.fromJsonString(OrchestrationV2AppThreadJsonSchema), ); @@ -2661,7 +2666,9 @@ export const layer: Layer.Layer = } const boundedTurnItemRows = - fields !== undefined && !fields.includes("turnItems") + (fields !== undefined && !fields.includes("turnItems")) || + (window === undefined && + matchesNothing(filter?.turnItemTypes, filter?.turnItemStatuses, filter?.turnItemRunIds)) ? [] : window === undefined ? yield* sql` @@ -2874,7 +2881,8 @@ export const layer: Layer.Layer = contextTransferRows, ] = yield* Effect.all([ decodeThreadPayload(threadRow.payload_json), - fields !== undefined && !fields.includes("runs") + (fields !== undefined && !fields.includes("runs")) || + (window === undefined && matchesNothing(filter?.runIds)) ? Effect.succeed([]) : window === undefined ? sql` @@ -3024,7 +3032,9 @@ export const layer: Layer.Layer = OR provider_turn_id IN (SELECT value FROM json_each(${cohortProviderTurnIds}))) ORDER BY created_at ASC, runtime_request_id ASC `, - fields !== undefined && !fields.includes("messages") + (fields !== undefined && !fields.includes("messages")) || + (window === undefined && + matchesNothing(filter?.messageIds, filter?.messageRoles, filter?.messageRunIds)) ? Effect.succeed([]) : window === undefined ? sql`