diff --git a/packages/convex/convex/models/facts/facts.test.ts b/packages/convex/convex/models/facts/facts.test.ts index 6030020..20cd152 100644 --- a/packages/convex/convex/models/facts/facts.test.ts +++ b/packages/convex/convex/models/facts/facts.test.ts @@ -4,6 +4,7 @@ import { afterEach, beforeEach, describe, expect, test } from "vitest"; import { api, internal } from "../../_generated/api"; import schema from "../../schema"; import { modules } from "../../test.setup"; +import { listFacts, searchFacts } from "./model"; const issuer = "https://brain.example.test"; @@ -321,4 +322,80 @@ describe("structured durable facts", () => { "Jordan — home city: Seattle.", ]); }); + + test("fills fact result limits after lifecycle filtering", async () => { + const t = convexTest(schema, modules); + const userId = await t.run((ctx) => ctx.db.insert("users", {})); + + await t.run(async (ctx) => { + const subjectEntityId = await ctx.db.insert("entities", { + userId, + key: "person:pagination-test", + kind: "person", + canonicalName: "Pagination Test", + normalizedName: "pagination test", + aliases: [], + normalizedAliases: [], + }); + const insertFact = ( + index: number, + status: "current" | "retracted", + validFrom?: number, + ) => + ctx.db.insert("facts", { + userId, + subjectEntityId, + predicate: "school", + value: { type: "text", value: `School ${index}` }, + statement: `Pagination Test — school: School ${index}.`, + searchText: `pagination sentinel school School ${index}`, + sourceType: "user_stated", + confidence: 1, + isCore: true, + validFrom, + status, + }); + + // Retrievable rows are deliberately older. The scheduled rows exhaust + // the old current-only `take(limit * 5)` window, while the still-newer + // retractions exhaust its historical window. + for (let index = 0; index < 10; index += 1) { + await insertFact(index, "current"); + } + for (let index = 10; index < 70; index += 1) { + await insertFact(index, "current", Date.now() + 86_400_000); + } + for (let index = 70; index < 130; index += 1) { + await insertFact(index, "retracted"); + } + }); + + const recent = await t.run((ctx) => listFacts(ctx, userId, { limit: 10 })); + const core = await t.run((ctx) => + listFacts(ctx, userId, { limit: 10, coreOnly: true }), + ); + const historical = await t.run((ctx) => + listFacts(ctx, userId, { limit: 10, includeHistorical: true }), + ); + const search = await t.run((ctx) => + searchFacts(ctx, userId, "pagination sentinel school", { limit: 10 }), + ); + const historicalSearch = await t.run((ctx) => + searchFacts(ctx, userId, "pagination sentinel school", { + limit: 10, + includeHistorical: true, + }), + ); + + for (const results of [ + recent, + core, + historical, + search, + historicalSearch, + ]) { + expect(results).toHaveLength(10); + expect(results.every((fact) => fact.status !== "retracted")).toBe(true); + } + }); }); diff --git a/packages/convex/convex/models/facts/model.ts b/packages/convex/convex/models/facts/model.ts index 973a867..dcc9d1a 100644 --- a/packages/convex/convex/models/facts/model.ts +++ b/packages/convex/convex/models/facts/model.ts @@ -1,6 +1,7 @@ import type { Infer } from "convex/values"; +import type { Expression, FilterBuilder, NamedTableInfo } from "convex/server"; -import type { Doc, Id } from "../../_generated/dataModel"; +import type { DataModel, Doc, Id } from "../../_generated/dataModel"; import type { MutationCtx, QueryCtx } from "../../_generated/server"; import { assertValidMemoryValidity, @@ -304,6 +305,30 @@ export function isFactRetrievable( ); } +type FactFilterBuilder = FilterBuilder>; + +/** + * Express current business-time validity inside the Convex query so `take` + * counts retrievable rows rather than candidates later discarded in memory. + */ +function currentFactValidityFilter( + q: FactFilterBuilder, + activeAt: number, +): Expression { + const validFrom = q.field("validFrom"); + const validTo = q.field("validTo"); + return q.and( + q.or( + q.eq(validFrom, undefined), + q.lte(validFrom as Expression, activeAt), + ), + q.or( + q.eq(validTo, undefined), + q.gt(validTo as Expression, activeAt), + ), + ); +} + export type RememberFactArgs = { subject: EntitySelector; predicate: string; @@ -532,22 +557,43 @@ export async function listFacts( throw new Error("Fact limit must be a positive integer"); } const limit = Math.min(requested, MAX_FACT_SEARCH_LIMIT); - const facts = options.coreOnly - ? await ctx.db - .query("facts") - .withIndex("by_userId_and_isCore", (q) => - q.eq("userId", userId).eq("isCore", true), - ) - .order("desc") - .take(limit * 5) - : await ctx.db - .query("facts") - .withIndex("by_userId", (q) => q.eq("userId", userId)) - .order("desc") - .take(limit * 5); - const selected = facts - .filter((fact) => isFactRetrievable(fact, options.includeHistorical)) - .slice(0, limit); + const activeAt = Date.now(); + let selected: Doc<"facts">[]; + if (options.includeHistorical) { + selected = options.coreOnly + ? await ctx.db + .query("facts") + .withIndex("by_userId_and_isCore", (q) => + q.eq("userId", userId).eq("isCore", true), + ) + .order("desc") + .filter((q) => q.neq(q.field("status"), "retracted")) + .take(limit) + : await ctx.db + .query("facts") + .withIndex("by_userId", (q) => q.eq("userId", userId)) + .order("desc") + .filter((q) => q.neq(q.field("status"), "retracted")) + .take(limit); + } else { + selected = options.coreOnly + ? await ctx.db + .query("facts") + .withIndex("by_userId_isCore_status", (q) => + q.eq("userId", userId).eq("isCore", true).eq("status", "current"), + ) + .order("desc") + .filter((q) => currentFactValidityFilter(q, activeAt)) + .take(limit) + : await ctx.db + .query("facts") + .withIndex("by_userId_and_status", (q) => + q.eq("userId", userId).eq("status", "current"), + ) + .order("desc") + .filter((q) => currentFactValidityFilter(q, activeAt)) + .take(limit); + } return await Promise.all(selected.map((fact) => hydrateFact(ctx, fact))); } @@ -563,14 +609,20 @@ export async function searchFacts( throw new Error("Fact search limit must be a positive integer"); } const limit = Math.min(requested, MAX_FACT_SEARCH_LIMIT); - const hits = await ctx.db + const activeAt = Date.now(); + const selected = await ctx.db .query("facts") - .withSearchIndex("by_searchText", (q) => - q.search("searchText", cleanedQuery).eq("userId", userId), + .withSearchIndex("by_searchText", (q) => { + const search = q.search("searchText", cleanedQuery).eq("userId", userId); + return options.includeHistorical + ? search + : search.eq("status", "current"); + }) + .filter((q) => + options.includeHistorical + ? q.neq(q.field("status"), "retracted") + : currentFactValidityFilter(q, activeAt), ) - .take(limit * 5); - const selected = hits - .filter((fact) => isFactRetrievable(fact, options.includeHistorical)) - .slice(0, limit); + .take(limit); return await Promise.all(selected.map((fact) => hydrateFact(ctx, fact))); } diff --git a/packages/convex/convex/models/thoughts/coreMemory.test.ts b/packages/convex/convex/models/thoughts/coreMemory.test.ts index 8c4622c..1c6ec45 100644 --- a/packages/convex/convex/models/thoughts/coreMemory.test.ts +++ b/packages/convex/convex/models/thoughts/coreMemory.test.ts @@ -89,6 +89,18 @@ describe("core memories", () => { isCore: true, memoryStatus: "current", }); + // These are newer than the retrievable core set. The previous fixed + // 250-candidate window returned nothing once enough history accumulated. + for (let index = 0; index < 260; index += 1) { + await ctx.db.insert("thoughts", { + userId: ownerId, + content: `Owner retracted core memory ${index}`, + embedding, + metadata, + isCore: true, + memoryStatus: "retracted", + }); + } }); const owner = t.withIdentity({ issuer, subject: ownerId }); diff --git a/packages/convex/convex/models/thoughts/memoryTransition.test.ts b/packages/convex/convex/models/thoughts/memoryTransition.test.ts index 9d8c04a..bb2ad88 100644 --- a/packages/convex/convex/models/thoughts/memoryTransition.test.ts +++ b/packages/convex/convex/models/thoughts/memoryTransition.test.ts @@ -65,6 +65,62 @@ describe("temporal memory transitions", () => { ]); }); + test("fills memory result limits after lifecycle filtering", async () => { + const t = convexTest(schema, modules); + const userId = await t.run((ctx) => ctx.db.insert("users", {})); + + await t.run(async (ctx) => { + const insertMemory = ( + index: number, + memoryStatus: "current" | "retracted" | undefined, + validFrom?: number, + ) => + ctx.db.insert("thoughts", { + userId, + content: `Pagination sentinel memory ${index}`, + embedding, + metadata, + memoryStatus, + validFrom, + }); + + for (let index = 0; index < 10; index += 1) { + await insertMemory(index, undefined); + } + for (let index = 10; index < 70; index += 1) { + await insertMemory(index, "current", Date.now() + 86_400_000); + } + for (let index = 70; index < 130; index += 1) { + await insertMemory(index, "retracted"); + } + }); + + const current = await t.run((ctx) => _listByUser(ctx, userId, 10)); + const historical = await t.run((ctx) => _listByUser(ctx, userId, 10, true)); + const search = await t.query( + internal.models.thoughts.private.searchByText, + { + userId, + query: "pagination sentinel memory", + limit: 10, + activeAt: Date.now(), + }, + ); + + expect(current).toHaveLength(10); + expect(current.every((memory) => memory.memoryStatus === undefined)).toBe( + true, + ); + expect(historical).toHaveLength(10); + expect( + historical.every((memory) => memory.memoryStatus !== "retracted"), + ).toBe(true); + expect(search).toHaveLength(10); + expect(search.every((memory) => memory.memoryStatus === undefined)).toBe( + true, + ); + }); + test("atomically preserves and links a superseded memory", async () => { const t = convexTest(schema, modules); const userId = await t.run((ctx) => ctx.db.insert("users", {})); diff --git a/packages/convex/convex/models/thoughts/model.ts b/packages/convex/convex/models/thoughts/model.ts index dca60a2..323dd17 100644 --- a/packages/convex/convex/models/thoughts/model.ts +++ b/packages/convex/convex/models/thoughts/model.ts @@ -1,12 +1,11 @@ import type { Infer } from "convex/values"; +import type { Expression, FilterBuilder, NamedTableInfo } from "convex/server"; -import type { Id } from "../../_generated/dataModel"; +import type { DataModel, Id } from "../../_generated/dataModel"; import type { MutationCtx, QueryCtx } from "../../_generated/server"; import { assertValidMemoryValidity, isCurrentMemory, - isMemoryActive, - isMemoryRetrievable, safeSupersededValidTo, type MemoryStatus, type MemoryValidity, @@ -26,50 +25,44 @@ type ThoughtProvenance = { export const DEFAULT_CORE_MEMORY_LIMIT = 10; export const MAX_CORE_MEMORY_LIMIT = 25; -const MAX_CORE_MEMORY_CANDIDATES = 250; export async function _findById(ctx: QueryCtx, id: Id<"thoughts">) { return await ctx.db.get(id); } +type ThoughtFilterBuilder = FilterBuilder< + NamedTableInfo +>; + /** - * Pages through an ordered query, keeping only rows that pass `predicate`, - * until `limit` rows are collected or the source is exhausted. - * - * Lifecycle status cannot be filtered at the index: memories written before - * the temporal-memory change have no `memoryStatus` at all, and `undefined` - * means "current". Filtering on the stored value would silently drop every - * legacy memory, so the filter has to run after the read. A fixed over-fetch - * multiplier would instead under-fill the page once an account accumulates - * enough superseded memories, with no signal that anything was dropped — - * paging until the quota is met keeps the result honest either way. + * Express memory lifecycle and business-time validity inside a Convex query. + * Legacy rows omit `memoryStatus`, so undefined remains equivalent to current. + * Applying this before `take` fills the requested result window without trying + * to issue a second `.paginate()` call in the same function execution. */ -export async function collectFiltered( - fetchPage: ( - cursor: string | null, - numItems: number, - ) => Promise<{ page: T[]; isDone: boolean; continueCursor: string }>, - predicate: (row: T) => boolean, - limit: number, - maxPages = 20, -): Promise { - const collected: T[] = []; - const pageSize = Math.min(Math.max(limit * 2, 50), 500); - let cursor: string | null = null; - - for (let page = 0; page < maxPages && collected.length < limit; page += 1) { - const result = await fetchPage(cursor, pageSize); - for (const row of result.page) { - if (predicate(row)) { - collected.push(row); - if (collected.length === limit) break; - } - } - if (result.isDone) break; - cursor = result.continueCursor; +export function memoryRetrievabilityFilter( + q: ThoughtFilterBuilder, + includeHistorical: boolean | undefined, + activeAt: number, +): Expression { + const memoryStatus = q.field("memoryStatus"); + if (includeHistorical) { + return q.neq(memoryStatus, "retracted"); } - return collected; + const validFrom = q.field("validFrom"); + const validTo = q.field("validTo"); + return q.and( + q.or(q.eq(memoryStatus, undefined), q.eq(memoryStatus, "current")), + q.or( + q.eq(validFrom, undefined), + q.lte(validFrom as Expression, activeAt), + ), + q.or( + q.eq(validTo, undefined), + q.gt(validTo as Expression, activeAt), + ), + ); } export async function _listByUser( @@ -78,18 +71,13 @@ export async function _listByUser( limit: number = 20, includeHistorical = false, ) { - const query = () => - ctx.db - .query("thoughts") - .withIndex("by_userId", (q) => q.eq("userId", userId)) - .order("desc"); - const activeAt = Date.now(); - return await collectFiltered( - (cursor, numItems) => query().paginate({ cursor, numItems }), - (memory) => isMemoryRetrievable(memory, includeHistorical, activeAt), - limit, - ); + return await ctx.db + .query("thoughts") + .withIndex("by_userId", (q) => q.eq("userId", userId)) + .order("desc") + .filter((q) => memoryRetrievabilityFilter(q, includeHistorical, activeAt)) + .take(limit); } export async function _listCoreByUser( @@ -105,18 +93,15 @@ export async function _listCoreByUser( throw new Error("Core memory limit must be a positive integer"); } const limit = Math.min(requestedLimit, MAX_CORE_MEMORY_LIMIT); - const candidates = await ctx.db + const activeAt = Date.now(); + return await ctx.db .query("thoughts") .withIndex("by_userId_and_isCore", (q) => q.eq("userId", userId).eq("isCore", true), ) .order("desc") - .take(MAX_CORE_MEMORY_CANDIDATES); - - const activeAt = Date.now(); - return candidates - .filter((memory) => isMemoryActive(memory, activeAt)) - .slice(0, limit); + .filter((q) => memoryRetrievabilityFilter(q, false, activeAt)) + .take(limit); } export async function _insertOne( diff --git a/packages/convex/convex/models/thoughts/private.ts b/packages/convex/convex/models/thoughts/private.ts index f9c17c6..754fff6 100644 --- a/packages/convex/convex/models/thoughts/private.ts +++ b/packages/convex/convex/models/thoughts/private.ts @@ -7,9 +7,8 @@ import { _listCoreByUser, _setCoreStatus, _transitionMemory, - collectFiltered, + memoryRetrievabilityFilter, } from "./model"; -import { isMemoryRetrievable } from "./memoryLifecycle"; import { memorySourceType, thoughtLifecycleFields, @@ -188,18 +187,16 @@ export const searchByText = internalQuery({ ), handler: async (ctx, args) => { const limit = args.limit ?? 50; - const query = () => - ctx.db.query("thoughts").withSearchIndex("by_content", (q) => { + const results = await ctx.db + .query("thoughts") + .withSearchIndex("by_content", (q) => { const base = q.search("content", args.query).eq("userId", args.userId); return args.type ? base.eq("metadata.type", args.type) : base; - }); - - const results = await collectFiltered( - (cursor, numItems) => query().paginate({ cursor, numItems }), - (memory) => - isMemoryRetrievable(memory, args.includeHistorical, args.activeAt), - limit, - ); + }) + .filter((q) => + memoryRetrievabilityFilter(q, args.includeHistorical, args.activeAt), + ) + .take(limit); return results.map(({ embedding: _embedding, ...rest }) => rest); }, diff --git a/packages/convex/convex/models/thoughts/public.ts b/packages/convex/convex/models/thoughts/public.ts index 0a410e5..77f69e4 100644 --- a/packages/convex/convex/models/thoughts/public.ts +++ b/packages/convex/convex/models/thoughts/public.ts @@ -1,13 +1,17 @@ import { query } from "../../_generated/server"; import { v } from "convex/values"; import { requireWebUserId } from "../../lib/webAuth"; -import { isMemoryActive, isMemoryRetrievable } from "./memoryLifecycle"; +import { isMemoryActive } from "./memoryLifecycle"; import { thoughtLifecycleFields, thoughtMetadata, thoughtType, } from "./validators"; -import { _listByUser, _listCoreByUser, collectFiltered } from "./model"; +import { + _listByUser, + _listCoreByUser, + memoryRetrievabilityFilter, +} from "./model"; import { isFactActive } from "../facts/model"; export const listRecent = query({ @@ -33,21 +37,17 @@ export const listRecent = query({ let results; if (args.type) { const limit = args.limit ?? 20; - const query = () => - ctx.db - .query("thoughts") - .withIndex("by_userId_and_type", (q) => - q.eq("userId", userId).eq("metadata.type", args.type!), - ) - .order("desc"); - const activeAt = Date.now(); - results = await collectFiltered( - (cursor, numItems) => query().paginate({ cursor, numItems }), - (memory) => - isMemoryRetrievable(memory, args.includeHistorical, activeAt), - limit, - ); + results = await ctx.db + .query("thoughts") + .withIndex("by_userId_and_type", (q) => + q.eq("userId", userId).eq("metadata.type", args.type!), + ) + .order("desc") + .filter((q) => + memoryRetrievabilityFilter(q, args.includeHistorical, activeAt), + ) + .take(limit); } else { results = await _listByUser( ctx, diff --git a/packages/convex/convex/schema.ts b/packages/convex/convex/schema.ts index 59f05b6..ee07819 100644 --- a/packages/convex/convex/schema.ts +++ b/packages/convex/convex/schema.ts @@ -32,6 +32,7 @@ export default defineSchema({ ]), facts: defineTable(factFields) .index("by_userId", ["userId"]) + .index("by_userId_and_status", ["userId", "status"]) .index("by_userId_subject_predicate_status", [ "userId", "subjectEntityId", @@ -39,9 +40,10 @@ export default defineSchema({ "status", ]) .index("by_userId_and_isCore", ["userId", "isCore"]) + .index("by_userId_isCore_status", ["userId", "isCore", "status"]) .searchIndex("by_searchText", { searchField: "searchText", - filterFields: ["userId"], + filterFields: ["userId", "status"], }), apiKeys: defineTable(apiKeyFields) .index("by_keyHash", ["keyHash"])