From 516d1d4eb63578c8bd1938472c2257cac67bf384 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 23:07:44 +0000 Subject: [PATCH] fix(adapters): write learned facts under the profile's lock and limit Profile consolidation wrote facts beside the user profile repository: without its per-user lock, its duplicate check at write time, or its limit of 500 facts. A long-running profile could grow past the limit and then refuse the user's own additions. Consolidation now takes the same lock, drops facts the profile gained since the run read it, and keeps only what still fits. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01MRtGG9ARfqqhma3t85Ro2u --- .../profile-consolidation.postgres.test.ts | 36 ++++++++++++++++++- .../adapters/src/profile-consolidation.ts | 24 +++++++++++-- packages/db/src/user-profile.ts | 7 ++-- 3 files changed, 62 insertions(+), 5 deletions(-) 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"