From 3aa898d3a0994d3b8964d35142cc2255e0884626 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 15 Oct 2026 19:32:53 -0700 Subject: [PATCH 1/5] 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. --- .../application/resolve-access.integration.ts | 38 ++++++- .../memories/application/resolve-access.ts | 76 +++++++------ .../memories/cleanup-outbox.test.ts | 15 +-- .../lib/mothership/memories/cleanup-outbox.ts | 105 +++++++++++------- .../memories/cleanup.integration.ts | 49 +++++--- .../lib/projects/application/authorization.ts | 9 +- .../files/application/authorization.ts | 45 ++++---- .../workspace-files/application/subject.ts | 12 +- 8 files changed, 221 insertions(+), 128 deletions(-) diff --git a/apps/sim/lib/mothership/memories/application/resolve-access.integration.ts b/apps/sim/lib/mothership/memories/application/resolve-access.integration.ts index 9473a1b62ee..72a6033d0a5 100644 --- a/apps/sim/lib/mothership/memories/application/resolve-access.integration.ts +++ b/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((resolve) => { + release = resolve + }) + let held = () => {} + const holding = new Promise((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) diff --git a/apps/sim/lib/mothership/memories/application/resolve-access.ts b/apps/sim/lib/mothership/memories/application/resolve-access.ts index db6b6592676..2e488482f08 100644 --- a/apps/sim/lib/mothership/memories/application/resolve-access.ts +++ b/apps/sim/lib/mothership/memories/application/resolve-access.ts @@ -1,15 +1,18 @@ 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, + resolveEffectiveWorkspacePermission, +} 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, + NoWorkspaceAccessError, } from '@/lib/core/application/workspace-authorization' -import { isFeatureEnabled } from '@/lib/core/config/feature-flags' 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 +24,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 +53,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 @@ -59,14 +72,19 @@ async function workspaceAccess(tx: DbTransaction, userId: string, workspaceId: s workspaceId: workspace.id, name: workspace.name, workspaceOrganizationId: workspace.organizationId, - allowPersonalApiKeys: workspace.allowPersonalApiKeys, }) .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 resolveEffectiveWorkspacePermission( + userId, + workspaceId, + target.workspaceOrganizationId, + tx + ) + if (!permission) throw new NoWorkspaceAccessError() + if (!permissionSatisfies(permission, 'read')) throw new InsufficientWorkspacePermissionsError() await assertWorkspaceCapability( userId, workspaceId, @@ -74,14 +92,7 @@ async function workspaceAccess(tx: DbTransaction, userId: string, workspaceId: s 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,12 +110,10 @@ 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, { @@ -112,9 +121,10 @@ async function projectAccess( 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 +177,10 @@ const resolveAccess = defineAuthorizedChatUseCase({ }, }, async execute({ principal, context, input }): Promise { + if (input.target.scope === 'project') await requireProjectApiEnabled() + const projectsEnabled = input.target.scope === 'workspace' && (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 +191,6 @@ const resolveAccess = defineAuthorizedChatUseCase({ }) .from(copilotChats) .where(and(eq(copilotChats.id, context.chatId), isNull(copilotChats.deletedAt))) - .for('share') .limit(1) if ( !chat || @@ -204,9 +215,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 +231,6 @@ const resolveAccess = defineAuthorizedChatUseCase({ permissions: { read: true, privateMutation: true, sharedMutation: false }, } } else if (input.target.scope === 'project') { - 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 +239,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 +266,7 @@ const resolveAccess = defineAuthorizedChatUseCase({ actor: { userId: principal.subjectUserId, chatId: context.chatId }, target, } - }) + }, READ_SNAPSHOT) }, }) diff --git a/apps/sim/lib/mothership/memories/cleanup-outbox.test.ts b/apps/sim/lib/mothership/memories/cleanup-outbox.test.ts index 57603f42054..25c1da8673b 100644 --- a/apps/sim/lib/mothership/memories/cleanup-outbox.test.ts +++ b/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() }) }) diff --git a/apps/sim/lib/mothership/memories/cleanup-outbox.ts b/apps/sim/lib/mothership/memories/cleanup-outbox.ts index 3dbaa16ef55..00b3814784e 100644 --- a/apps/sim/lib/mothership/memories/cleanup-outbox.ts +++ b/apps/sim/lib/mothership/memories/cleanup-outbox.ts @@ -1,8 +1,15 @@ -import { createHash } from 'node:crypto' import type { db } from '@sim/db' +import { createLogger } from '@sim/logger' +import { chunkArray } from '@sim/utils/helpers' +import { omit } from '@sim/utils/object' import { z } from 'zod' import { env } from '@/lib/core/config/env' -import { enqueueOutboxEvent, type OutboxHandlerRegistry } from '@/lib/core/outbox/service' +import { + deferOutboxHandler, + enqueueOutboxEvents, + MAX_BULK_ENQUEUE_EVENTS, + type OutboxHandlerRegistry, +} from '@/lib/core/outbox/service' import { SIM_AGENT_API_URL } from '@/lib/mothership/constants' import { MemoryCleanupInput, @@ -11,13 +18,25 @@ import { import { MEMORY_CLEANUP_EVENT } from '@/lib/mothership/memories/cleanup-event' import { fetchGo } from '@/lib/mothership/request/go/fetch' import { mothershipRequestHeaders } from '@/lib/mothership/request/headers' -import { getMothershipSourceEnvHeaders } from '@/lib/mothership/server/agent-url' +import { + getMothershipSourceEnvHeaders, + MOTHERSHIP_SOURCE_ENV_HEADER, +} from '@/lib/mothership/server/agent-url' + +const logger = createLogger('MemoryCleanup') -const cleanupPayloadSchema = z.strictObject({ +/** Waits for a worker or configuration that is not there yet without spending the retry budget. */ +const UNAVAILABLE_RETRY_MS = 60 * 60 * 1000 +/** Statuses meaning the worker does not serve cleanup yet, not that this request failed. */ +const ENDPOINT_MISSING_STATUSES = new Set([404, 405, 501]) + +/** Non-strict so rows enqueued with the retired `credentialFingerprint` field still parse. */ +const cleanupPayloadSchema = z.object({ request: MemoryCleanupInput, baseUrl: z.url(), - credentialFingerprint: z.string().nullable(), - sourceHeaders: z.record(z.string(), z.string()), + sourceHeaders: z.object({ + [MOTHERSHIP_SOURCE_ENV_HEADER]: z.enum(['dev', 'staging', 'prod']).optional(), + }), }) function destinations(): string[] { @@ -31,10 +50,6 @@ function destinations(): string[] { ] } -function credentialFingerprint(): string | null { - return env.COPILOT_API_KEY ? createHash('sha256').update(env.COPILOT_API_KEY).digest('hex') : null -} - /** * Captures configured routes before deletion; notes can outlive a user's environment selection. * Without `COPILOT_API_KEY` this Sim never reached a worker, so it holds no notes to erase and @@ -44,50 +59,62 @@ export async function enqueueMemoryCleanup( executor: Pick, targets: MemoryCleanupInput['target'][] ): Promise { - const fingerprint = credentialFingerprint() - if (!fingerprint) return + if (!env.COPILOT_API_KEY) return const sourceHeaders = getMothershipSourceEnvHeaders() - for (const baseUrl of destinations()) { - for (const target of targets) { - await enqueueOutboxEvent( - executor, - MEMORY_CLEANUP_EVENT, - cleanupPayloadSchema.parse({ - request: { version: 1, target }, - baseUrl, - credentialFingerprint: fingerprint, - sourceHeaders, - }) - ) - } + const payloads = destinations().flatMap((baseUrl) => + targets.map((target) => + cleanupPayloadSchema.parse({ request: { version: 1, target }, baseUrl, sourceHeaders }) + ) + ) + for (const batch of chunkArray(payloads, MAX_BULK_ENQUEUE_EVENTS)) { + await enqueueOutboxEvents(executor, MEMORY_CLEANUP_EVENT, batch) } } +/** + * Sends the processor's current key only to a destination the processor itself configures, and + * replays the enqueuing deployment's source env so the worker erases that deployment's notes. + * A missing key, destination, or worker endpoint waits without consuming attempts: erasure must + * not dead-letter while configuration or the worker rollout catches up. + */ export const memoryCleanupOutboxHandlers = { [MEMORY_CLEANUP_EVENT]: async (rawPayload, context) => { const payload = cleanupPayloadSchema.parse(rawPayload) - if ( - !payload.credentialFingerprint || - payload.credentialFingerprint !== credentialFingerprint() - ) { - throw new Error('Memory cleanup credential binding changed; operator reconciliation required') + if (!env.COPILOT_API_KEY) { + logger.warn('Memory cleanup waiting: COPILOT_API_KEY is not configured', { + eventId: context.eventId, + }) + return deferOutboxHandler('COPILOT_API_KEY not configured', UNAVAILABLE_RETRY_MS, false) } - if ( - !destinations().includes(payload.baseUrl) || - JSON.stringify(payload.sourceHeaders) !== JSON.stringify(getMothershipSourceEnvHeaders()) - ) { - throw new Error( - 'Memory cleanup destination binding changed; operator reconciliation required' - ) + if (!destinations().includes(payload.baseUrl)) { + logger.warn('Memory cleanup waiting: destination is not configured', { + eventId: context.eventId, + baseUrl: payload.baseUrl, + }) + return deferOutboxHandler('Destination not configured', UNAVAILABLE_RETRY_MS, false) } + const sourceEnv = payload.sourceHeaders[MOTHERSHIP_SOURCE_ENV_HEADER] const response = await fetchGo(`${payload.baseUrl}/api/memories/cleanup`, { method: 'POST', - headers: mothershipRequestHeaders(), + headers: { + ...omit(mothershipRequestHeaders(), [MOTHERSHIP_SOURCE_ENV_HEADER]), + ...(sourceEnv ? { [MOTHERSHIP_SOURCE_ENV_HEADER]: sourceEnv } : {}), + }, body: JSON.stringify(payload.request), signal: AbortSignal.any([context.signal, AbortSignal.timeout(10_000)]), redirect: 'error', }) - if (!response.ok) throw new Error(`Memory cleanup failed (${response.status})`) + if (!response.ok) { + await response.body?.cancel().catch(() => {}) + if (ENDPOINT_MISSING_STATUSES.has(response.status)) { + return deferOutboxHandler( + `Memory cleanup endpoint unavailable (${response.status})`, + UNAVAILABLE_RETRY_MS, + false + ) + } + throw new Error(`Memory cleanup failed (${response.status})`) + } MemoryCleanupResult.parse(await response.json()) }, } satisfies OutboxHandlerRegistry diff --git a/apps/sim/lib/mothership/memories/cleanup.integration.ts b/apps/sim/lib/mothership/memories/cleanup.integration.ts index 211f13bd464..dd361ac1b2a 100644 --- a/apps/sim/lib/mothership/memories/cleanup.integration.ts +++ b/apps/sim/lib/mothership/memories/cleanup.integration.ts @@ -11,13 +11,13 @@ const userId = generateId() const workspaceId = generateId() const projectId = generateId() const received: { body: unknown; key: string | string[] | undefined }[] = [] -let fail = true +let status = 503 const server = createServer(async (req, res) => { let body = '' for await (const chunk of req) body += chunk received.push({ body: JSON.parse(body), key: req.headers['x-api-key'] }) - res.writeHead(fail ? 503 : 200, { 'content-type': 'application/json' }) - res.end(JSON.stringify({ success: !fail })) + res.writeHead(status, { 'content-type': 'application/json' }) + res.end(JSON.stringify({ success: status === 200 })) }) let runtime: { deletion: typeof import('@/lib/users/account-deletion') @@ -117,7 +117,7 @@ it('rolls cleanup back with deletion and retries committed account/target erasur expect( (await pending()).every((event) => event.status === 'pending' && event.attempts === 1) ).toBe(true) - fail = false + status = 200 await db .update(outboxEvent) .set({ availableAt: new Date(0) }) @@ -139,18 +139,33 @@ it('rolls cleanup back with deletion and retries committed account/target erasur expect(received.every((request) => request.key === 'memory-cleanup-fixture-key')).toBe(true) const first = events[0] - await db - .update(outboxEvent) - .set({ - status: 'pending', - availableAt: new Date(0), - payload: { ...toRecord(first.payload), credentialFingerprint: 'changed-credential' }, - }) - .where(and(eq(outboxEvent.id, first.id), eq(outboxEvent.status, 'completed'))) + const replay = async (payload: unknown = first.payload) => { + await db + .update(outboxEvent) + .set({ status: 'pending', availableAt: new Date(0), attempts: 9, payload }) + .where(eq(outboxEvent.id, first.id)) + await runtime.outbox.processOutboxEventById( + first.id, + runtime.cleanup.memoryCleanupOutboxHandlers + ) + const [row] = await db.select().from(outboxEvent).where(eq(outboxEvent.id, first.id)) + return row + } + const waitsAnHour = (row: typeof outboxEvent.$inferSelect) => { + expect(row).toMatchObject({ status: 'pending', attempts: 9 }) + expect(row.availableAt.getTime()).toBeGreaterThan(Date.now() + 50 * 60_000) + } + + status = 404 const before = received.length - await runtime.outbox.processOutboxEventById(first.id, runtime.cleanup.memoryCleanupOutboxHandlers) - expect(received).toHaveLength(before) - const [blocked] = await db.select().from(outboxEvent).where(eq(outboxEvent.id, first.id)) - expect(blocked.status).toBe('pending') - expect(blocked.lastError).toContain('credential binding changed') + waitsAnHour(await replay()) + expect(received).toHaveLength(before + 1) + + status = 200 + waitsAnHour(await replay({ ...toRecord(first.payload), baseUrl: 'https://unconfigured.test' })) + expect(received).toHaveLength(before + 1) + + const legacy = await replay({ ...toRecord(first.payload), credentialFingerprint: 'retired' }) + expect(legacy.status).toBe('completed') + expect(received).toHaveLength(before + 2) }, 30_000) diff --git a/apps/sim/lib/projects/application/authorization.ts b/apps/sim/lib/projects/application/authorization.ts index 75fe2291a15..f91e22c75a5 100644 --- a/apps/sim/lib/projects/application/authorization.ts +++ b/apps/sim/lib/projects/application/authorization.ts @@ -192,16 +192,17 @@ function projectVisibility( * Loads and holds canonical Project membership for a resource-specific access policy. `lock` is * the mode the Project is held in: `shared` for a caller that only needs it fenced against * lifecycle and payer changes, `exclusive` for one that will take it exclusively later in the - * transaction, which must never upgrade a shared hold. + * transaction, which must never upgrade a shared hold. `snapshot` holds nothing and relies on + * the caller's read-only snapshot, for a caller that acts on nothing in the transaction. */ export async function loadProjectAccess( tx: DbTransaction, userId: string, input: ProjectAuthorizationInput & { projectId: string }, - lock: 'shared' | 'exclusive' = 'exclusive' + lock: 'shared' | 'exclusive' | 'snapshot' = 'exclusive' ) { if (lock === 'shared') await lockProjectShared(tx, input.projectId) - else await lockProject(tx, input.projectId) + else if (lock === 'exclusive') await lockProject(tx, input.projectId) const [record] = await tx.select().from(project).where(eq(project.id, input.projectId)).limit(1) if ( !record || @@ -210,7 +211,7 @@ export async function loadProjectAccess( throw new OrchestrationError('not_found', 'Project not found') return projectVisibility( record, - await loadProjectMemberships(tx, userId, [record], 'hold'), + await loadProjectMemberships(tx, userId, [record], lock === 'snapshot' ? 'snapshot' : 'hold'), input ) } diff --git a/apps/sim/lib/projects/files/application/authorization.ts b/apps/sim/lib/projects/files/application/authorization.ts index 9d799fa7849..3c2986c6554 100644 --- a/apps/sim/lib/projects/files/application/authorization.ts +++ b/apps/sim/lib/projects/files/application/authorization.ts @@ -98,7 +98,10 @@ function requirePrincipal( } } -/** Rechecks canonical conversation ownership and authoring capability while authorization locks are held. */ +/** + * Rechecks canonical conversation ownership and authoring capability while authorization locks + * are held; `snapshot` reads the same rows unlocked inside the caller's read-only snapshot. + */ export async function requireCurrentCopilotProjectInvocation( tx: DbTransaction, principal: Pick< @@ -106,14 +109,15 @@ export async function requireCurrentCopilotProjectInvocation( 'subjectUserId' | 'invocation' >, access: Awaited>, - scope: Awaited> + scope: Awaited>, + mode: 'hold' | 'snapshot' = 'hold' ) { if (scope.organizationId !== access.record.organizationId) { throw new OrchestrationError('not_found', 'Project not found in this conversation') } let workspaceId: string | null if (principal.invocation.kind === 'chat') { - const [chat] = await tx + const chatQuery = tx .select({ userId: copilotChats.userId, workspaceId: copilotChats.workspaceId, @@ -123,8 +127,8 @@ export async function requireCurrentCopilotProjectInvocation( }) .from(copilotChats) .where(and(eq(copilotChats.id, principal.invocation.chatId), isNull(copilotChats.deletedAt))) - .for('share') .limit(1) + const [chat] = await (mode === 'hold' ? chatQuery.for('share') : chatQuery) if ( !chat || chat.userId !== principal.subjectUserId || @@ -158,7 +162,7 @@ export async function requireCurrentCopilotProjectInvocation( if (!workspaceId || workspaceId !== scope.workspaceId) { throw new OrchestrationError('not_found', 'Project not found in this conversation') } - const [origin] = await tx + const originQuery = tx .select({ workspaceId: workspace.id, workspaceOrganizationId: workspace.organizationId, @@ -166,22 +170,24 @@ export async function requireCurrentCopilotProjectInvocation( }) .from(workspace) .where(and(eq(workspace.id, workspaceId), isNull(workspace.archivedAt))) - .for('share') .limit(1) + const [origin] = await (mode === 'hold' ? originQuery.for('share') : originQuery) if (!origin || origin.workspaceOrganizationId !== scope.organizationId) throw new OrchestrationError('not_found', 'Workspace not found in this conversation') - // Hold origin grants without upgrading the target's existing SHARE locks. - await tx - .select({ id: permissions.id }) - .from(permissions) - .where( - and( - eq(permissions.userId, principal.subjectUserId), - eq(permissions.entityType, 'workspace'), - eq(permissions.entityId, workspaceId) + if (mode === 'hold') { + // Hold origin grants without upgrading the target's existing SHARE locks. + await tx + .select({ id: permissions.id }) + .from(permissions) + .where( + and( + eq(permissions.userId, principal.subjectUserId), + eq(permissions.entityType, 'workspace'), + eq(permissions.entityId, workspaceId) + ) ) - ) - .for('share') + .for('share') + } await requireCurrentHumanRole(principal.subjectUserId, origin, 'read', { executor: tx }) await assertWorkspaceCapability( principal.subjectUserId, @@ -197,10 +203,11 @@ export async function requireProjectFileOwnerRole( tx: DbTransaction, principal: ProjectFilePrincipal | SubjectDelegatedPrincipal | OrganizationDelegatedPrincipal, access: Awaited>, - accessMode: 'read' | 'write' + accessMode: 'read' | 'write', + mode: 'hold' | 'snapshot' = 'hold' ): Promise { if (access.record.archivedAt) throw new OrchestrationError('conflict', 'Project is archived') - await requireCurrentFileSubject(tx, principal) + await requireCurrentFileSubject(tx, principal, mode) const canWrite = access.orgAdmin || access.active.some((row) => row.permission === 'admin') || diff --git a/apps/sim/lib/workspace-files/application/subject.ts b/apps/sim/lib/workspace-files/application/subject.ts index a11bc5906e7..81145e11930 100644 --- a/apps/sim/lib/workspace-files/application/subject.ts +++ b/apps/sim/lib/workspace-files/application/subject.ts @@ -17,19 +17,23 @@ export async function requireFileSubject(principal: FileSubjectPrincipal): Promi return userId } -/** Holds the acting account stable through a file transaction without substituting an owner. */ +/** + * Holds the acting account stable through a file transaction without substituting an owner. + * `snapshot` reads it without a lock, for a caller in a read-only snapshot that acts on nothing. + */ export async function requireCurrentFileSubject( tx: DbTransaction, - principal: FileSubjectPrincipal + principal: FileSubjectPrincipal, + mode: 'hold' | 'snapshot' = 'hold' ): Promise { // actorless-unsupported: Project files and cross-owner copies require a human; their operation registries reject workspace keys and executors. const userId = requirePrincipalSubjectUserId(principal) - const [actor] = await tx + const query = tx .select({ banned: user.banned, banExpires: user.banExpires, suspendedAt: user.suspendedAt }) .from(user) .where(eq(user.id, userId)) - .for('share') .limit(1) + const [actor] = await (mode === 'hold' ? query.for('share') : query) if (!actor || isAccountBlocked(actor)) { throw new OrchestrationError('forbidden', 'User account is suspended') } From afcce9a2e263a4a09e3313ec8be759501e8bae69 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 15 Oct 2026 19:37:19 -0700 Subject: [PATCH 2/5] fix(memory): keep the Projects-off refusal after the chat checks and defer cleanup on pre-route 401 --- .../lib/mothership/memories/application/resolve-access.ts | 6 +++--- apps/sim/lib/mothership/memories/cleanup-outbox.ts | 7 +++++-- apps/sim/lib/projects/rollout.server.ts | 2 +- 3 files changed, 9 insertions(+), 6 deletions(-) diff --git a/apps/sim/lib/mothership/memories/application/resolve-access.ts b/apps/sim/lib/mothership/memories/application/resolve-access.ts index 2e488482f08..5f0e360e56c 100644 --- a/apps/sim/lib/mothership/memories/application/resolve-access.ts +++ b/apps/sim/lib/mothership/memories/application/resolve-access.ts @@ -30,7 +30,7 @@ import { requireProjectFileOwnerCapabilities, requireProjectFileOwnerRole, } from '@/lib/projects/files/application/authorization' -import { isProjectsEnabled, requireProjectApiEnabled } from '@/lib/projects/rollout.server' +import { isProjectsEnabled, ProjectUnavailableError } from '@/lib/projects/rollout.server' import { requireCurrentFileSubject } from '@/lib/workspace-files/application/subject' export const MEMORY_NOTES_ACCESS_AUDIENCE = 'sim:memory-access' @@ -177,8 +177,7 @@ const resolveAccess = defineAuthorizedChatUseCase({ }, }, async execute({ principal, context, input }): Promise { - if (input.target.scope === 'project') await requireProjectApiEnabled() - const projectsEnabled = input.target.scope === 'workspace' && (await isProjectsEnabled()) + const projectsEnabled = input.target.scope !== 'personal' && (await isProjectsEnabled()) return db.transaction(async (tx) => { await requireCurrentFileSubject(tx, principal, 'snapshot') const [chat] = await tx @@ -231,6 +230,7 @@ const resolveAccess = defineAuthorizedChatUseCase({ permissions: { read: true, privateMutation: true, sharedMutation: false }, } } else if (input.target.scope === 'project') { + if (!projectsEnabled) throw new ProjectUnavailableError() 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) diff --git a/apps/sim/lib/mothership/memories/cleanup-outbox.ts b/apps/sim/lib/mothership/memories/cleanup-outbox.ts index 00b3814784e..96ab4c2d54d 100644 --- a/apps/sim/lib/mothership/memories/cleanup-outbox.ts +++ b/apps/sim/lib/mothership/memories/cleanup-outbox.ts @@ -27,8 +27,11 @@ const logger = createLogger('MemoryCleanup') /** Waits for a worker or configuration that is not there yet without spending the retry budget. */ const UNAVAILABLE_RETRY_MS = 60 * 60 * 1000 -/** Statuses meaning the worker does not serve cleanup yet, not that this request failed. */ -const ENDPOINT_MISSING_STATUSES = new Set([404, 405, 501]) +/** + * Statuses meaning the worker does not serve cleanup yet, not that this request failed: a worker + * that predates the route answers 404 to the hosted key and 401 to a self-hosted customer key. + */ +const ENDPOINT_MISSING_STATUSES = new Set([401, 404, 405, 501]) /** Non-strict so rows enqueued with the retired `credentialFingerprint` field still parse. */ const cleanupPayloadSchema = z.object({ diff --git a/apps/sim/lib/projects/rollout.server.ts b/apps/sim/lib/projects/rollout.server.ts index b2c7ffe5d9d..33fb9ac2056 100644 --- a/apps/sim/lib/projects/rollout.server.ts +++ b/apps/sim/lib/projects/rollout.server.ts @@ -1,7 +1,7 @@ import { isFeatureEnabled } from '@/lib/core/config/feature-flags' import { HttpError } from '@/lib/core/utils/http-error' -class ProjectUnavailableError extends HttpError { +export class ProjectUnavailableError extends HttpError { readonly statusCode = 503 constructor() { super('Projects are not enabled on this deployment') From d9738891bf8e7be7f6b8a0c1ec9576e2bbd05d53 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 15 Oct 2026 19:39:15 -0700 Subject: [PATCH 3/5] refactor(memory): reuse the shared role check and one cleanup wait path --- .../application/workspace-authorization.ts | 8 +++-- .../application/resolve-access.integration.ts | 10 ++----- .../memories/application/resolve-access.ts | 24 ++++----------- .../lib/mothership/memories/cleanup-outbox.ts | 29 +++++++------------ apps/sim/lib/projects/rollout.server.ts | 2 +- 5 files changed, 26 insertions(+), 47 deletions(-) diff --git a/apps/sim/lib/core/application/workspace-authorization.ts b/apps/sim/lib/core/application/workspace-authorization.ts index 721c59d7007..67de9320502 100644 --- a/apps/sim/lib/core/application/workspace-authorization.ts +++ b/apps/sim/lib/core/application/workspace-authorization.ts @@ -191,7 +191,10 @@ export function requireAllowedWorkspacePrincipal( } } -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 -): Promise { +): Promise { const permission = await resolveEffectiveWorkspacePermission( userId, context.workspaceId, @@ -333,6 +336,7 @@ export async function requireCurrentHumanRole { - let release = () => {} - const released = new Promise((resolve) => { - release = resolve - }) - let held = () => {} - const holding = new Promise((resolve) => { - held = resolve - }) + const { promise: released, resolve: release } = Promise.withResolvers() + const { promise: holding, resolve: held } = Promise.withResolvers() 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) diff --git a/apps/sim/lib/mothership/memories/application/resolve-access.ts b/apps/sim/lib/mothership/memories/application/resolve-access.ts index 5f0e360e56c..a52ba63bc29 100644 --- a/apps/sim/lib/mothership/memories/application/resolve-access.ts +++ b/apps/sim/lib/mothership/memories/application/resolve-access.ts @@ -1,18 +1,12 @@ import type { OrganizationDelegatedPrincipal, SubjectDelegatedPrincipal } from '@sim/auth/principal' import { db } from '@sim/db' import { copilotChats, project, workspace } from '@sim/db/schema' -import { - permissionSatisfies, - resolveEffectiveWorkspacePermission, -} from '@sim/platform-authz/workspace' +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, - NoWorkspaceAccessError, -} from '@/lib/core/application/workspace-authorization' +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' @@ -30,7 +24,7 @@ import { requireProjectFileOwnerCapabilities, requireProjectFileOwnerRole, } from '@/lib/projects/files/application/authorization' -import { isProjectsEnabled, ProjectUnavailableError } 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' @@ -72,19 +66,13 @@ async function workspaceAccess(tx: DbTransaction, userId: string, workspaceId: s workspaceId: workspace.id, name: workspace.name, workspaceOrganizationId: workspace.organizationId, + allowPersonalApiKeys: workspace.allowPersonalApiKeys, }) .from(workspace) .where(and(eq(workspace.id, workspaceId), isNull(workspace.archivedAt))) .limit(1) if (!target) throw new OrchestrationError('not_found', 'Workspace not found') - const permission = await resolveEffectiveWorkspacePermission( - userId, - workspaceId, - target.workspaceOrganizationId, - tx - ) - if (!permission) throw new NoWorkspaceAccessError() - if (!permissionSatisfies(permission, 'read')) throw new InsufficientWorkspacePermissionsError() + const permission = await requireCurrentHumanRole(userId, target, 'read', { executor: tx }) await assertWorkspaceCapability( userId, workspaceId, @@ -230,7 +218,7 @@ const resolveAccess = defineAuthorizedChatUseCase({ permissions: { read: true, privateMutation: true, sharedMutation: false }, } } else if (input.target.scope === 'project') { - if (!projectsEnabled) throw new ProjectUnavailableError() + 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) diff --git a/apps/sim/lib/mothership/memories/cleanup-outbox.ts b/apps/sim/lib/mothership/memories/cleanup-outbox.ts index 96ab4c2d54d..eb67bcf1267 100644 --- a/apps/sim/lib/mothership/memories/cleanup-outbox.ts +++ b/apps/sim/lib/mothership/memories/cleanup-outbox.ts @@ -74,6 +74,11 @@ export async function enqueueMemoryCleanup( } } +function waitForWorker(reason: string, eventId: string, detail?: Record) { + logger.warn(`Memory cleanup waiting: ${reason}`, { eventId, ...detail }) + return deferOutboxHandler(reason, UNAVAILABLE_RETRY_MS, false) +} + /** * Sends the processor's current key only to a destination the processor itself configures, and * replays the enqueuing deployment's source env so the worker erases that deployment's notes. @@ -83,19 +88,12 @@ export async function enqueueMemoryCleanup( export const memoryCleanupOutboxHandlers = { [MEMORY_CLEANUP_EVENT]: async (rawPayload, context) => { const payload = cleanupPayloadSchema.parse(rawPayload) - if (!env.COPILOT_API_KEY) { - logger.warn('Memory cleanup waiting: COPILOT_API_KEY is not configured', { - eventId: context.eventId, - }) - return deferOutboxHandler('COPILOT_API_KEY not configured', UNAVAILABLE_RETRY_MS, false) - } - if (!destinations().includes(payload.baseUrl)) { - logger.warn('Memory cleanup waiting: destination is not configured', { - eventId: context.eventId, + if (!env.COPILOT_API_KEY) + return waitForWorker('COPILOT_API_KEY is not configured', context.eventId) + if (!destinations().includes(payload.baseUrl)) + return waitForWorker('destination is not configured', context.eventId, { baseUrl: payload.baseUrl, }) - return deferOutboxHandler('Destination not configured', UNAVAILABLE_RETRY_MS, false) - } const sourceEnv = payload.sourceHeaders[MOTHERSHIP_SOURCE_ENV_HEADER] const response = await fetchGo(`${payload.baseUrl}/api/memories/cleanup`, { method: 'POST', @@ -109,13 +107,8 @@ export const memoryCleanupOutboxHandlers = { }) if (!response.ok) { await response.body?.cancel().catch(() => {}) - if (ENDPOINT_MISSING_STATUSES.has(response.status)) { - return deferOutboxHandler( - `Memory cleanup endpoint unavailable (${response.status})`, - UNAVAILABLE_RETRY_MS, - false - ) - } + if (ENDPOINT_MISSING_STATUSES.has(response.status)) + return waitForWorker(`endpoint unavailable (${response.status})`, context.eventId) throw new Error(`Memory cleanup failed (${response.status})`) } MemoryCleanupResult.parse(await response.json()) diff --git a/apps/sim/lib/projects/rollout.server.ts b/apps/sim/lib/projects/rollout.server.ts index 33fb9ac2056..b2c7ffe5d9d 100644 --- a/apps/sim/lib/projects/rollout.server.ts +++ b/apps/sim/lib/projects/rollout.server.ts @@ -1,7 +1,7 @@ import { isFeatureEnabled } from '@/lib/core/config/feature-flags' import { HttpError } from '@/lib/core/utils/http-error' -export class ProjectUnavailableError extends HttpError { +class ProjectUnavailableError extends HttpError { readonly statusCode = 503 constructor() { super('Projects are not enabled on this deployment') From b0c85a26b9b7605755bd521f0cfc4849389d18d5 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 15 Oct 2026 19:46:51 -0700 Subject: [PATCH 4/5] test(memory): use plain latches the app TypeScript lib supports --- .../memories/application/resolve-access.integration.ts | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/apps/sim/lib/mothership/memories/application/resolve-access.integration.ts b/apps/sim/lib/mothership/memories/application/resolve-access.integration.ts index a67758eb328..72a6033d0a5 100644 --- a/apps/sim/lib/mothership/memories/application/resolve-access.integration.ts +++ b/apps/sim/lib/mothership/memories/application/resolve-access.integration.ts @@ -257,8 +257,14 @@ it('resolves overlapping workspace checks that share one Project without deadloc }) it('answers while a writer holds the account row and the Project exclusively', async () => { - const { promise: released, resolve: release } = Promise.withResolvers() - const { promise: holding, resolve: held } = Promise.withResolvers() + let release = () => {} + const released = new Promise((resolve) => { + release = resolve + }) + let held = () => {} + const holding = new Promise((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) From 586fe1b3ad5b3a58d00818f907108ffcb40fddc9 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 15 Oct 2026 20:00:31 -0700 Subject: [PATCH 5/5] fix(memory): accept additive fields in note access and cleanup answers --- .../mothership/generated/memory-notes-access.ts | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/apps/sim/lib/mothership/generated/memory-notes-access.ts b/apps/sim/lib/mothership/generated/memory-notes-access.ts index b4989b14822..b1c414e4986 100644 --- a/apps/sim/lib/mothership/generated/memory-notes-access.ts +++ b/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; -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; -export const MemoryCleanupResult = z.strictObject({ success: z.literal(true) }); +export const MemoryCleanupResult = z.object({ success: z.literal(true) });