fix(engine): scope observer DM listing queries

CHANGELOG.md

@@ -16,7 +16,11 @@ This project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 Packages without a separate changelog are covered by the cross-package notes below.
 
-## [Unreleased]
+## [Unreleased - Patch]
+
+### Fixed
+
+- Conversation-scoped observer links load only their allowed direct messages instead of enriching every DM in the workspace first.
 
 ## [10.0.1] - 2026-10-11
 

packages/engine/CHANGELOG.md

@@ -7,7 +7,11 @@ See the [root changelog](../../CHANGELOG.md) for cross-package release highlight
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [Unreleased]
+## [Unreleased - Patch]
+
+### Fixed
+
+- `GET /v1/dm/conversations/all` applies an observer's conversation allowlist before participant and message enrichment, keeping narrowly scoped links fast in large workspaces.
 
 ## [10.0.1] - 2026-10-11
 

packages/engine/src/__tests__/conformance/connectObserver.test.ts

@@ -8,7 +8,7 @@ async function mintConnectObserver(
   stack: TestStack,
   workspaceKey: string,
   name: string,
-  filters: { include_dms: true; agent_ids?: string[] } = { include_dms: true },
+  filters: { include_dms: true; agent_ids?: string[]; dm_conversation_ids?: string[] } = { include_dms: true },
 ) {
   const response = await stack.app.request('/v1/observer-tokens', {
     method: 'POST',
@@ -57,11 +57,24 @@ describe('Relay Connect observer capability', () => {
     const room = await createWorkspace(stack.app, 'connect-many-dms');
     const alice = await registerAgent(stack.app, room.workspaceKey, 'alice');
     const peers = Array.from({ length: 105 }, (_, index) => `peer-${String(index).padStart(3, '0')}`);
+    const conversationIds: string[] = [];
     for (const peer of peers) {
       await registerAgent(stack.app, room.workspaceKey, peer);
-      await sendDm(stack, alice.token, peer, `hello ${peer}`);
+      const dm = await sendDm(stack, alice.token, peer, `hello ${peer}`);
+      conversationIds.push(dm.data.conversation_id);
     }
-    const observer = await mintConnectObserver(stack, room.workspaceKey, 'connect-many-dms-observer');
+    const observer = await mintConnectObserver(
+      stack,
+      room.workspaceKey,
+      'connect-many-dms-observer',
+      { include_dms: true, dm_conversation_ids: conversationIds },
+    );
+    const scopedObserver = await mintConnectObserver(
+      stack,
+      room.workspaceKey,
+      'connect-one-dm-observer',
+      { include_dms: true, dm_conversation_ids: [conversationIds[0]!] },
+    );
 
     // Emulate D1's 100-bound-parameter ceiling against the SQL Drizzle emits.
     const sqlite = stack.runtime.handle.sqlite;
@@ -90,6 +103,40 @@ describe('Relay Connect observer capability', () => {
         expect(conversation.last_message?.text).toBe(`hello ${peer}`);
       }
     }
+
+    const preparedSql: string[] = [];
+    sqlite.prepare = ((source: string) => {
+      preparedSql.push(source);
+      return prepare(source);
+    }) as typeof sqlite.prepare;
+    let scopedResponse: Response;
+    try {
+      scopedResponse = await stack.app.request('/v1/dm/conversations/all', {
+        headers: { authorization: `Bearer ${scopedObserver.token}` },
+      });
+    } finally {
+      sqlite.prepare = prepare as typeof sqlite.prepare;
+    }
+    expect(scopedResponse.status).toBe(200);
+    const scopedBody = (await scopedResponse.json()) as {
+      data: Array<{ id: string; participants: string[]; message_count: number; last_message: { text: string } | null }>;
+    };
+    expect(scopedBody.data).toEqual([
+      expect.objectContaining({
+        id: conversationIds[0],
+        participants: ['alice', peers[0]],
+        message_count: 1,
+        last_message: expect.objectContaining({ text: `hello ${peers[0]}` }),
+      }),
+    ]);
+
+    const dmListingQueries = preparedSql.filter((source) => (
+      source.includes('from "dm_conversations"')
+      || source.includes('from "dm_participants"')
+      || source.includes('from "messages"')
+    ));
+    expect(dmListingQueries).toHaveLength(5);
+    expect(dmListingQueries[0]).toContain('"dm_conversations"."id" in (?)');
   });
 
   it('opens one room, reads historical and new DMs, and fails closed for writes, other rooms, expiry, revocation, and room deletion', async () => {

packages/engine/src/engine/dmAll.ts

@@ -23,8 +23,19 @@ export function dmFilteredCountConversationChunkSize(agentFilterCount: number):
   return Math.max(1, D1_SAFE_IN_QUERY_CHUNK_SIZE - Math.min(agentFilterCount, DM_COUNT_AGENT_CHUNK_SIZE));
 }
 
-export async function listAllDmConversations(db: Db, workspaceId: string): Promise<WorkspaceDmConversation[]> {
-  const conversations = await db
+export async function listAllDmConversations(
+  db: Db,
+  workspaceId: string,
+  opts: { conversationIds?: string[] } = {},
+): Promise<WorkspaceDmConversation[]> {
+  const scopedConversationIds = opts.conversationIds === undefined
+    ? undefined
+    : [...new Set(opts.conversationIds)];
+  if (scopedConversationIds?.length === 0) {
+    return [];
+  }
+
+  const selectConversations = (conversationIds?: string[]) => db
     .select({
       id: dmConversations.id,
       dmType: dmConversations.dmType,
@@ -33,7 +44,18 @@ export async function listAllDmConversations(db: Db, workspaceId: string): Promi
       createdAt: dmConversations.createdAt,
     })
     .from(dmConversations)
-    .where(eq(dmConversations.workspaceId, workspaceId));
+    .where(and(
+      eq(dmConversations.workspaceId, workspaceId),
+      ...(conversationIds ? [inArray(dmConversations.id, conversationIds)] : []),
+    ));
+
+  // Observer capabilities commonly expose only one conversation in a large
+  // workspace. Apply that scope before any participant/message enrichment so
+  // unauthorized conversations are never loaded. Chunk high-cardinality
+  // capabilities to retain D1's bound-parameter safety.
+  const conversations = scopedConversationIds
+    ? await queryInChunks(scopedConversationIds, selectConversations)
+    : await selectConversations();
 
   if (conversations.length === 0) {
     return [];

packages/engine/src/routes/workspace.ts

@@ -505,9 +505,16 @@ workspaceRoutes.get('/dm/conversations/all', requireWorkspaceRead('dms:read', {
     const db = c.get('db');
     const workspace = c.get('workspace');
     const observer = getObserverTokenFromContext(c);
-    const allowedConversations = (await dmAllEngine.listAllDmConversations(db, workspace.id))
-      .filter((conversation) => observerAllowsConversation(observer, conversation.id));
     const filters = observer ? normalizeObserverFilters(observer.filters) : null;
+    const scopedConversationIds = observer
+      ? (filters?.include_dms ? filters.dm_conversation_ids : [])
+      : undefined;
+    const allowedConversations = (await dmAllEngine.listAllDmConversations(
+      db,
+      workspace.id,
+      { conversationIds: scopedConversationIds },
+    ))
+      .filter((conversation) => observerAllowsConversation(observer, conversation.id));
     const filtersMessages = Boolean(filters?.agent_ids?.length || filters?.created_after);
     const filteredCounts = observer && filtersMessages
       ? await dmAllEngine.countDmMessagesForWorkspaceConversations(