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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MRtGG9ARfqqhma3t85Ro2u

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();

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,

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<void> {
+/**
+ * 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<void> {
   const key = `user-profile:${userId}`;
   await tx.$queryRaw(Prisma.sql`
     SELECT pg_advisory_xact_lock(hashtextextended(${key}, 0))::text AS "lock"