fix(memory): read note access from an unlocked snapshot and defer unavailable cleanup (#9102)

* fix(memory): read note access from an unlocked snapshot and defer unavailable cleanup

- Resolve memory note access in a read-only repeatable-read transaction with no row or advisory locks; the shared Project helpers gain a snapshot mode and keep their hold behavior for every existing caller.
- Resolve workspace access once per check and derive shared mutation from the effective permission.
- Read the projects flag before the transaction.
- Defer memory cleanup hourly without spending attempts when the worker has no cleanup endpoint or the processor lacks the key or destination, instead of dead-lettering.
- Drop the credential and source-env equality checks; replay the enqueuing source env and accept rows queued with the retired fingerprint field.
- Enqueue cleanup events in one bulk insert.

* fix(memory): keep the Projects-off refusal after the chat checks and defer cleanup on pre-route 401

* refactor(memory): reuse the shared role check and one cleanup wait path

* test(memory): use plain latches the app TypeScript lib supports

* fix(memory): accept additive fields in note access and cleanup answers

Author: waleedlatif1

apps/sim/lib/core/application/workspace-authorization.ts

@@ -191,7 +191,10 @@ export function requireAllowedWorkspacePrincipal<O extends WorkspaceOperation>(
   }
 }
 
-function requirePermission(permission: PermissionType | null, required: PermissionType): void {
+function requirePermission(
+  permission: PermissionType | null,
+  required: PermissionType
+): asserts permission is PermissionType {
   if (permission === null) {
     throw new NoWorkspaceAccessError()
   }
@@ -324,7 +327,7 @@ export async function requireCurrentHumanRole<C extends WorkspaceAuthorizationCo
   context: C,
   required: PermissionType,
   options?: WorkspaceAuthorizationOptions<C>
-): Promise<void> {
+): Promise<PermissionType> {
   const permission = await resolveEffectiveWorkspacePermission(
     userId,
     context.workspaceId,
@@ -333,6 +336,7 @@ export async function requireCurrentHumanRole<C extends WorkspaceAuthorizationCo
     { forUpdate: options?.forUpdate }
   )
   requirePermission(permission, required)
+  return permission
 }
 
 /**

apps/sim/lib/mothership/generated/memory-notes-access.ts

@@ -3,12 +3,13 @@
 
 import { z } from "zod";
 
-const permissionsSchema = z.strictObject({
+// Answers tolerate fields a newer peer adds; requests stay strict.
+const permissionsSchema = z.object({
   read: z.literal(true),
   privateMutation: z.literal(true),
   sharedMutation: z.boolean(),
 });
-const projectSchema = z.strictObject({
+const projectSchema = z.object({
   scope: z.literal("project"),
   id: z.uuid(),
   name: z.string(),
@@ -25,17 +26,17 @@ export const MemoryNotesAccessInput = z.strictObject({
 });
 export type MemoryNotesAccessInput = z.infer<typeof MemoryNotesAccessInput>;
 
-export const MemoryNotesAccessResult = z.strictObject({
+export const MemoryNotesAccessResult = z.object({
   success: z.literal(true),
   version: z.literal(1),
-  actor: z.strictObject({ userId: z.string().min(1), chatId: z.uuid() }),
+  actor: z.object({ userId: z.string().min(1), chatId: z.uuid() }),
   target: z.discriminatedUnion("scope", [
-    z.strictObject({
+    z.object({
       scope: z.literal("personal"),
       permissions: permissionsSchema.extend({ sharedMutation: z.literal(false) }),
     }),
     projectSchema,
-    z.strictObject({
+    z.object({
       scope: z.literal("workspace"),
       id: z.uuid(),
       name: z.string(),
@@ -55,4 +56,4 @@ export const MemoryCleanupInput = z.strictObject({
   ]),
 });
 export type MemoryCleanupInput = z.infer<typeof MemoryCleanupInput>;
-export const MemoryCleanupResult = z.strictObject({ success: z.literal(true) });
+export const MemoryCleanupResult = z.object({ success: z.literal(true) });

apps/sim/lib/mothership/memories/application/resolve-access.integration.ts

@@ -12,6 +12,7 @@ import {
 } from '@sim/db/schema'
 import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures'
 import { featureFlagsMock, featureFlagsMockFns } from '@sim/testing/mocks/feature-flags.mock'
+import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
 import { and, eq, inArray } from 'drizzle-orm'
 import { NextRequest } from 'next/server'
@@ -26,7 +27,7 @@ import {
   resolveMemoryNotesAccess,
 } from '@/lib/mothership/memories/application/resolve-access'
 import { executeSimControl } from '@/lib/mothership/transport/control'
-import { splitForkProject } from '@/lib/projects/membership'
+import { lockProject, splitForkProject } from '@/lib/projects/membership'
 import { POST } from '@/app/api/mothership/memories/access/route'
 
 vi.mock('@/lib/core/config/feature-flags', () => featureFlagsMock)
@@ -255,6 +256,41 @@ it('resolves overlapping workspace checks that share one Project without deadloc
     expect(result.target).toMatchObject({ scope: 'workspace', currentProject: { id: projectId } })
 })
 
+it('answers while a writer holds the account row and the Project exclusively', async () => {
+  let release = () => {}
+  const released = new Promise<void>((resolve) => {
+    release = resolve
+  })
+  let held = () => {}
+  const holding = new Promise<void>((resolve) => {
+    held = resolve
+  })
+  const writer = db.transaction(async (tx) => {
+    await tx.select({ id: user.id }).from(user).where(eq(user.id, actorId)).for('update')
+    await lockProject(tx, projectId)
+    held()
+    await released
+  })
+  await holding
+  try {
+    for (const organizationChat of [false, true]) {
+      const result = await Promise.race([
+        resolveMemoryNotesAccess.execute({
+          principal: principal(organizationChat),
+          input: { target: { scope: 'workspace', workspaceId }, action: 'read' },
+        }),
+        sleep(2_000).then(() => {
+          throw new Error('Memory access waited on a writer')
+        }),
+      ])
+      expect(result.target).toMatchObject({ currentProject: { id: projectId } })
+    }
+  } finally {
+    release()
+    await writer
+  }
+})
+
 it('does not turn write in one environment into shared Project write', async () => {
   await db
     .update(permissions)

apps/sim/lib/mothership/memories/application/resolve-access.ts

@@ -1,15 +1,12 @@
 import type { OrganizationDelegatedPrincipal, SubjectDelegatedPrincipal } from '@sim/auth/principal'
 import { db } from '@sim/db'
-import { copilotChats, workspace } from '@sim/db/schema'
+import { copilotChats, project, workspace } from '@sim/db/schema'
+import { permissionSatisfies } from '@sim/platform-authz/workspace'
 import { and, eq, isNull } from 'drizzle-orm'
 import { defineWorkspaceOperation } from '@/lib/core/application'
 import { authorizeOrganizationOperation } from '@/lib/core/application/organization-authorization'
 import { defineOrganizationOperation } from '@/lib/core/application/organization-operation'
-import {
-  InsufficientWorkspacePermissionsError,
-  requireCurrentHumanRole,
-} from '@/lib/core/application/workspace-authorization'
-import { isFeatureEnabled } from '@/lib/core/config/feature-flags'
+import { requireCurrentHumanRole } from '@/lib/core/application/workspace-authorization'
 import { OrchestrationError } from '@/lib/core/orchestration/types'
 import type { DbTransaction } from '@/lib/db/types'
 import { defineAuthorizedChatUseCase } from '@/lib/mothership/chat/application/authorized-chat-use-case'
@@ -21,15 +18,13 @@ import type {
 } from '@/lib/mothership/generated/memory-notes-access'
 import { assertWorkspaceCapability } from '@/lib/permission-groups/capability-assertions'
 import { PermissionGroupCapabilityError } from '@/lib/permission-groups/capability-error'
-import { acquirePermissionGroupOrgReadLock } from '@/lib/permission-groups/locks'
 import { loadProjectAccess } from '@/lib/projects/application/authorization'
 import {
   requireCurrentCopilotProjectInvocation,
   requireProjectFileOwnerCapabilities,
   requireProjectFileOwnerRole,
 } from '@/lib/projects/files/application/authorization'
-import { lockWorkspaceProject } from '@/lib/projects/membership'
-import { requireProjectApiEnabled } from '@/lib/projects/rollout.server'
+import { isProjectsEnabled, requireProjectApiEnabled } from '@/lib/projects/rollout.server'
 import { requireCurrentFileSubject } from '@/lib/workspace-files/application/subject'
 
 export const MEMORY_NOTES_ACCESS_AUDIENCE = 'sim:memory-access'
@@ -52,6 +47,18 @@ const organizationOperation = defineOrganizationOperation({
   delegationAudience: MEMORY_NOTES_ACCESS_AUDIENCE,
 })
 type MemoryPrincipal = SubjectDelegatedPrincipal | OrganizationDelegatedPrincipal
+/** The worker acts on the answer after commit, so a consistent unlocked snapshot is enough. */
+const READ_SNAPSHOT = { isolationLevel: 'repeatable read', accessMode: 'read only' } as const
+
+async function workspaceProject(tx: DbTransaction, workspaceId: string) {
+  const [row] = await tx
+    .select({ id: project.id, organizationId: project.organizationId })
+    .from(workspace)
+    .innerJoin(project, eq(project.id, workspace.projectId))
+    .where(eq(workspace.id, workspaceId))
+    .limit(1)
+  return row ?? null
+}
 
 async function workspaceAccess(tx: DbTransaction, userId: string, workspaceId: string) {
   const [target] = await tx
@@ -63,25 +70,17 @@ async function workspaceAccess(tx: DbTransaction, userId: string, workspaceId: s
     })
     .from(workspace)
     .where(and(eq(workspace.id, workspaceId), isNull(workspace.archivedAt)))
-    .for('share')
     .limit(1)
   if (!target) throw new OrchestrationError('not_found', 'Workspace not found')
-  await requireCurrentHumanRole(userId, target, 'read', { executor: tx })
+  const permission = await requireCurrentHumanRole(userId, target, 'read', { executor: tx })
   await assertWorkspaceCapability(
     userId,
     workspaceId,
     'copilot.use',
     target.workspaceOrganizationId,
     tx
   )
-  let sharedMutation = true
-  try {
-    await requireCurrentHumanRole(userId, target, 'write', { executor: tx })
-  } catch (error) {
-    if (!(error instanceof InsufficientWorkspacePermissionsError)) throw error
-    sharedMutation = false
-  }
-  return { target, sharedMutation }
+  return { target, sharedMutation: permissionSatisfies(permission, 'write') }
 }
 
 async function projectAccess(
@@ -99,22 +98,21 @@ async function projectAccess(
       ...(scope.organizationId ? { organizationId: scope.organizationId } : {}),
       ...(scope.workspaceId ? { workspaceId: scope.workspaceId } : {}),
     },
-    'shared'
+    'snapshot'
   )
   if (scope.organizationId !== access.record.organizationId)
     throw new OrchestrationError('not_found', 'Project not found in this conversation')
-  if (access.record.organizationId)
-    await acquirePermissionGroupOrgReadLock(tx, access.record.organizationId)
   await requireCurrentCopilotProjectInvocation(
     tx,
     {
       subjectUserId: principal.subjectUserId,
       invocation: { kind: 'chat', chatId },
     },
     access,
-    scope
+    scope,
+    'snapshot'
   )
-  const owner = await requireProjectFileOwnerRole(tx, principal, access, 'read')
+  const owner = await requireProjectFileOwnerRole(tx, principal, access, 'read', 'snapshot')
   let sharedMutation = owner.canWrite
   if (sharedMutation) {
     try {
@@ -167,8 +165,9 @@ const resolveAccess = defineAuthorizedChatUseCase({
     },
   },
   async execute({ principal, context, input }): Promise<MemoryNotesAccessResult> {
+    const projectsEnabled = input.target.scope !== 'personal' && (await isProjectsEnabled())
     return db.transaction(async (tx) => {
-      await requireCurrentFileSubject(tx, principal)
+      await requireCurrentFileSubject(tx, principal, 'snapshot')
       const [chat] = await tx
         .select({
           userId: copilotChats.userId,
@@ -179,7 +178,6 @@ const resolveAccess = defineAuthorizedChatUseCase({
         })
         .from(copilotChats)
         .where(and(eq(copilotChats.id, context.chatId), isNull(copilotChats.deletedAt)))
-        .for('share')
         .limit(1)
       if (
         !chat ||
@@ -204,9 +202,11 @@ const resolveAccess = defineAuthorizedChatUseCase({
       }
       const origin =
         chat.workspaceId && input.target.scope !== 'personal'
-          ? await lockWorkspaceProject(tx, chat.workspaceId)
+          ? await workspaceProject(tx, chat.workspaceId)
           : null
-      if (chat.workspaceId) await workspaceAccess(tx, principal.subjectUserId, chat.workspaceId)
+      const originAccess = chat.workspaceId
+        ? await workspaceAccess(tx, principal.subjectUserId, chat.workspaceId)
+        : null
       const scope = {
         organizationId: chat.organizationId ?? origin?.organizationId ?? null,
         ...(chat.workspaceId ? { workspaceId: chat.workspaceId } : {}),
@@ -218,7 +218,7 @@ const resolveAccess = defineAuthorizedChatUseCase({
           permissions: { read: true, privateMutation: true, sharedMutation: false },
         }
       } else if (input.target.scope === 'project') {
-        await requireProjectApiEnabled()
+        if (!projectsEnabled) await requireProjectApiEnabled()
         if (chat.workspaceId && origin?.id !== input.target.projectId)
           throw new OrchestrationError('not_found', 'Project not found in this conversation')
         target = await projectAccess(tx, principal, context.chatId, input.target.projectId, scope)
@@ -227,13 +227,15 @@ const resolveAccess = defineAuthorizedChatUseCase({
           throw new OrchestrationError('not_found', 'Workspace not found in this conversation')
         const membership = chat.workspaceId
           ? origin
-          : await lockWorkspaceProject(tx, input.target.workspaceId)
+          : await workspaceProject(tx, input.target.workspaceId)
         if (!membership || membership.organizationId !== scope.organizationId)
           throw new OrchestrationError('not_found', 'Workspace not found in this conversation')
-        const access = await workspaceAccess(tx, principal.subjectUserId, input.target.workspaceId)
+        const access =
+          originAccess ??
+          (await workspaceAccess(tx, principal.subjectUserId, input.target.workspaceId))
         if (access.target.workspaceOrganizationId !== scope.organizationId)
           throw new OrchestrationError('not_found', 'Workspace not found in this conversation')
-        const currentProject = (await isFeatureEnabled('projects'))
+        const currentProject = projectsEnabled
           ? await projectAccess(tx, principal, context.chatId, membership.id, scope)
           : null
         target = {
@@ -252,7 +254,7 @@ const resolveAccess = defineAuthorizedChatUseCase({
         actor: { userId: principal.subjectUserId, chatId: context.chatId },
         target,
       }
-    })
+    }, READ_SNAPSHOT)
   },
 })
 

apps/sim/lib/mothership/memories/cleanup-outbox.test.ts

@@ -15,7 +15,7 @@ vi.mock('@/lib/mothership/server/agent-url', () => mothershipAgentUrlMock)
 
 import { enqueueMemoryCleanup } from '@/lib/mothership/memories/cleanup-outbox'
 
-const { mockEnqueueOutboxEvent } = outboxServiceMockFns
+const { mockEnqueueOutboxEvents } = outboxServiceMockFns
 const executor = { insert: vi.fn() }
 const targets = [
   { kind: 'private_owner' as const, userId: 'user-1' },
@@ -35,17 +35,6 @@ describe('enqueueMemoryCleanup', () => {
 
   it('queues nothing when this Sim has no worker credential', async () => {
     await enqueueMemoryCleanup(executor, targets)
-    expect(mockEnqueueOutboxEvent).not.toHaveBeenCalled()
-  })
-
-  it('queues one credential-bound event per destination and target', async () => {
-    setEnv({ COPILOT_API_KEY: 'configured-key' })
-    await enqueueMemoryCleanup(executor, targets)
-    expect(mockEnqueueOutboxEvent).toHaveBeenCalledTimes(targets.length)
-    for (const [, , payload] of mockEnqueueOutboxEvent.mock.calls)
-      expect(payload).toMatchObject({
-        baseUrl: 'https://agent.example.test',
-        credentialFingerprint: expect.any(String),
-      })
+    expect(mockEnqueueOutboxEvents).not.toHaveBeenCalled()
   })
 })