diff --git a/packages/adapters/src/profile-consolidation.postgres.test.ts b/packages/adapters/src/profile-consolidation.postgres.test.ts index 187008e..1690474 100644 --- a/packages/adapters/src/profile-consolidation.postgres.test.ts +++ b/packages/adapters/src/profile-consolidation.postgres.test.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import type { AgentRunRequest, AgentRuntime, BackgroundJob } from "@rakazo/adapter-kit"; import type { PrismaClient } from "@rakazo/db"; -import { bootstrapUserSpace, createDb, ensureMainBot } from "@rakazo/db"; +import { bootstrapUserSpace, createDb, ensureMainBot, USER_FACT_LIMIT } from "@rakazo/db"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { EmulatorEmbeddingProvider, emulatedEmbedding } from "./embedding-emulator.js"; import { @@ -288,6 +288,40 @@ describePostgres("profile consolidation (PostgreSQL)", () => { ]); }); + it("never pushes the profile past its fact limit", async () => { + const user = await newUser(); + await addMessages(user.main.threadId, [ + { role: "user", text: "I run, swim and cycle every week.", at: ago(HOUR) }, + ]); + await prisma.userFact.createMany({ + data: Array.from({ length: USER_FACT_LIMIT - 1 }, (_, index) => ({ + userId: user.userId, + category: "habits", + value: `Known habit ${index}`, + source: "user", + })), + }); + const { runtime } = runtimeReplying( + JSON.stringify({ + facts: [ + { category: "habits", value: "Runs every week." }, + { category: "habits", value: "Swims every week." }, + { category: "habits", value: "Cycles every week." }, + ], + summary: "", + }), + ); + + await consolidateProfile(consolidateDeps(runtime), user.userId); + + expect(await prisma.userFact.count({ where: { userId: user.userId } })).toBe(USER_FACT_LIMIT); + expect( + await prisma.userFact.count({ where: { userId: user.userId, source: "consolidation" } }), + ).toBe(1); + const profile = await prisma.userProfile.findUniqueOrThrow({ where: { userId: user.userId } }); + expect(profile.consolidatedAt).not.toBeNull(); + }); + it("schedules users with new user activity whose last consolidation is a day old", async () => { const now = new Date(); const fresh = await newUser(); diff --git a/packages/adapters/src/profile-consolidation.ts b/packages/adapters/src/profile-consolidation.ts index 2831229..1a11f38 100644 --- a/packages/adapters/src/profile-consolidation.ts +++ b/packages/adapters/src/profile-consolidation.ts @@ -13,6 +13,7 @@ import { } from "@rakazo/contracts"; import { blocksToAgentHistoryText } from "@rakazo/core"; import type { PrismaClient } from "@rakazo/db"; +import { lockUserProfile, normalizeUserFactValue, USER_FACT_LIMIT } from "@rakazo/db"; import { getLogger } from "@rakazo/logging"; import { z } from "zod"; import { formatCurrentTimeInstruction } from "./current-time.js"; @@ -420,6 +421,8 @@ export async function consolidateProfile( } const written = await deps.prisma.$transaction(async (tx) => { + // The same lock as every other profile writer, so the checks below see their facts. + await lockUserProfile(tx, userId); // Compare-and-set: a concurrent run or a summary edit by the user wins over this run. const advanced = await tx.userProfile.updateMany({ where: { userId, consolidatedAt: profile.consolidatedAt, summary: profile.summary }, @@ -429,9 +432,26 @@ export async function consolidateProfile( }, }); if (advanced.count === 0) return false; - if (facts.length > 0) { + // Facts saved since this run read the profile win, and learned facts never push the + // profile past its limit, which would lock the user out of adding their own. + const current = await tx.userFact.findMany({ where: { userId }, select: { value: true } }); + const known = new Set(current.map((fact) => normalizeUserFactValue(fact.value))); + const room = Math.max(0, USER_FACT_LIMIT - current.length); + const unknown = facts.filter((fact) => { + const key = normalizeUserFactValue(fact.value); + if (known.has(key)) return false; + known.add(key); + return true; + }); + if (unknown.length > room) { + getLogger().info( + `profile.consolidate kept ${room} of ${unknown.length} new facts for user ${userId}: the profile is full`, + ); + } + const fitting = unknown.slice(0, room); + if (fitting.length > 0) { await tx.userFact.createMany({ - data: facts.map((fact) => { + data: fitting.map((fact) => { const sensitive = SENSITIVE_CATEGORIES.has(fact.category); return { userId, diff --git a/packages/db/src/user-profile.ts b/packages/db/src/user-profile.ts index 7946257..c8f80f3 100644 --- a/packages/db/src/user-profile.ts +++ b/packages/db/src/user-profile.ts @@ -161,8 +161,11 @@ export async function getUserProfileState( return { profile: profileDto(profile), facts: facts.map(factDto) }; } -/** Serializes one user's profile writes so duplicate checks and the fact limit hold under concurrency. */ -async function lockUserProfile(tx: Prisma.TransactionClient, userId: string): Promise { +/** + * Serializes one user's profile writes so duplicate checks and the fact limit hold under + * concurrency. Every writer of facts takes it inside its transaction. + */ +export async function lockUserProfile(tx: Prisma.TransactionClient, userId: string): Promise { const key = `user-profile:${userId}`; await tx.$queryRaw(Prisma.sql` SELECT pg_advisory_xact_lock(hashtextextended(${key}, 0))::text AS "lock"