fix(web): support large search corpora (round 4)

2026-07-27 13:05:59 +08:00
parent bd42204e53
commit 0dc66d5b77
8 changed files with 96 additions and 39 deletions
--- a/packages/host/apiproxy/README.i18n.yaml
+++ b/packages/host/apiproxy/README.i18n.yaml
@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/host/apiproxy/README.md
-README.md: deb1073c2ff9e7a533e595ee3f5537e649660e5b
-README.zh.md: e3c521d5f6a09414d087e3fb142852e6d3eb0cb7
+README.md: b9f0fcd8506afda733774868d78fe6e851b4fe2a
+README.zh.md: c57990b029b8d1e143396bb13718dbed165e2b47
--- a/packages/host/apiproxy/README.md
+++ b/packages/host/apiproxy/README.md
@@ -14,7 +14,7 @@ The mux stream projects the latest log-backed title as a validated `session/titl

 Workspace and Session lists are separate reconnect baselines. `workspace.create` creates a unique name or adopts an existing directory, `session.create` accepts an optional preallocated Session id, and `host/workspace-changed` plus `host/session-added` carry committed increments in either arrival order. `SessionSummary.blank` and the `host/session-added` frame carry the derived zero-events bit: clients hide blank sessions and reuse them per workspace, flip blank on the first `host/session-status(running:true)`, and treat `session.list` as the reconnect authority; cold summaries are never blank because lazy persistence keeps never-appended sessions out of `list()`.

-`session.search` is a bounded content-search projection over the sessions visible through `session.list`. The gateway passes only those session ids and current-surface user, assistant, and steering messages to the optional `ctx.sessionQuery` service, returns at most 20 session/snippet pairs plus a refine-query bit, and forwards the carrier request signal for cancellation. A deployment without the service, or a failed index/query operation, returns an `internal` business error so clients can retain metadata-only matches.
+`session.search` is a bounded content-search projection over the sessions visible through `session.list`. The gateway asks the optional `ctx.sessionQuery` service for globally ranked current-surface user, assistant, and steering matches, pages that stream until it has at most 20 visible session/snippet pairs plus one lookahead, and revalidates every hit against the list-derived authorization set before returning it. Keeping the authorization set in Host memory avoids SQLite's variable ceiling for large valid corpora without weakening visibility or ranking. The carrier request signal cancels persistence listing, cold-summary collection, and every search page. A deployment without the service, or a failed index/query operation, returns an `internal` business error so clients can retain metadata-only matches.

 The `command.*` and `skill.*` domains expose the host command registry and skill catalog to clients. Every method addresses one session's agent by `sessionId` (a served session always has an Agent; `command.*` resumes cold sessions through the same path as `session.*`, while `skill.list` resolves the project root from the session header without touching the Agent registry). `command.execute` runs a slash-command line host-side and returns a detached result; the carrier's request signal cancels the running handler. `host/commands-changed` is the catalog invalidation frame: clients refetch `command.list` instead of diffing.

--- a/packages/host/apiproxy/README.zh.md
+++ b/packages/host/apiproxy/README.zh.md
@@ -14,7 +14,7 @@ mux 流会在每个已附加会话的订阅基线之后，以及对应的实时

 Workspace 列表与 Session 列表是相互独立的重连基线。`workspace.create` 会创建唯一名称或接纳现有目录，`session.create` 接受可选的预分配 Session id，`host/workspace-changed` 与 `host/session-added` 则以任意到达顺序携带已提交的增量。`SessionSummary.blank` 与 `host/session-added` 帧携带派生的零事件位：客户端隐藏空白会话并按 workspace 复用它们，在首个 `host/session-status(running:true)` 时翻转 blank，并以 `session.list` 作为重连权威；冷会话摘要永远不是空白——惰性持久化让从未追加过事件的会话根本不出现在 `list()` 中。

-`session.search` 是以 `session.list` 所列会话为范围的有界内容搜索投影。网关仅将这些会话 id 以及当前表层中的 user、assistant 和 steering（中途引导）消息传给可选的 `ctx.sessionQuery` 服务，返回至多 20 个会话／snippet 对和一个提示细化查询的标志位，并转发载体请求信号以支持取消。部署若未挂载该服务，或索引／查询操作失败，都会返回 `internal` 业务错误，以便客户端保留仅基于元数据的匹配项。
+`session.search` 是以 `session.list` 所列会话为范围的有界内容搜索投影。网关向可选的 `ctx.sessionQuery` 服务请求全局排序后的当前 surface user、assistant 和 steering（中途引导）匹配项，对该结果流分页，直到获得至多 20 个可见会话／snippet 对及一个前瞻项，并在返回前依据从列表推导的授权集合重新校验每个命中。将授权集合保留在宿主内存中，可在不削弱可见性或排序的前提下避开有效大型语料库的 SQLite 变量上限。载体请求信号可取消持久化列表枚举、冷会话摘要收集和每一页搜索。部署若未挂载该服务，或索引／查询操作失败，都会返回 `internal` 业务错误，以便客户端保留仅基于元数据的匹配项。

 `command.*` 与 `skill.*` 领域向客户端暴露宿主命令注册表和技能目录。每个方法都通过 `sessionId` 寻址一个会话的 Agent（被服务的会话必有 Agent；`command.*` 经由与 `session.*` 相同的路径恢复冷会话，而 `skill.list` 从会话头解析项目根目录，不触碰 Agent 注册表）。`command.execute` 在宿主侧运行一条斜杠命令行并返回脱耦结果；载体的请求信号可取消正在运行的处理器。`host/commands-changed` 是目录失效帧：客户端重新拉取 `command.list` 而不是做差分。

--- a/packages/host/apiproxy/src/api-proxy.ts
+++ b/packages/host/apiproxy/src/api-proxy.ts
@@ -11,7 +11,7 @@ import type { Agent, AgentMessage, AgentMessageId, AgentStatus } from '@deepseek
 import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
 import type { JsonValue, Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
 import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
-import { SessionQueryError } from '@deepseek-ai/dsh-session-query'
+import { SessionQueryError, type SessionSearchCursor } from '@deepseek-ai/dsh-session-query'
 import { foldSessionTitle } from '@deepseek-ai/dsh-session-title'
 import type { Workspace, WorkspaceRecord } from '@deepseek-ai/dsh-workspace'
 import {
@@ -641,30 +641,51 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
          if (isAborted(signal)) return cancelled()
          if (visible.length === 0) return ok(request, { items: [], hasMore: false })
          const visibleIds = new Set(visible.map(item => item.sessionId))
-          const page = await sessionQuery.searchSessions({
-            query: request.payload.query,
-            sessionFilters: [{ kind: 'id', values: [...visibleIds] }],
-            eventFilters: [
-              { kind: 'type', values: ['user/message', 'assistant/message', 'steering/message'] },
-              { kind: 'surface', values: ['current'] },
-            ],
-            limit: SESSION_SEARCH_LIMIT,
-          }, { signal })
-          if (isAborted(signal)) return cancelled()
-          // The filters are the authorization boundary. Re-check the complete
-          // provider provenance before emitting its snippet so a backend
-          // regression cannot pair an allowed header with excluded content.
-          const authorized = page.items.filter(hit =>
-            visibleIds.has(hit.header.id)
-            && hit.bestMatch.sessionId === hit.header.id
-            && hit.bestMatch.surface === 'current'
-            && MESSAGE_TYPES.has(hit.bestMatch.type))
+          const authorized: SessionSearchItem[] = []
+          const acceptedIds = new Set<SessionId>()
+          const seenCursors = new Set<SessionSearchCursor>()
+          let cursor: SessionSearchCursor | undefined
+          while (authorized.length <= SESSION_SEARCH_LIMIT) {
+            if (isAborted(signal)) return cancelled()
+            const page = await sessionQuery.searchSessions({
+              query: request.payload.query,
+              eventFilters: [
+                { kind: 'type', values: ['user/message', 'assistant/message', 'steering/message'] },
+                { kind: 'surface', values: ['current'] },
+              ],
+              limit: SESSION_SEARCH_LIMIT,
+              ...cursor === undefined ? {} : { cursor },
+            }, { signal })
+            if (isAborted(signal)) return cancelled()
+            // Host visibility is the authorization boundary. Consume the
+            // provider's globally ranked stream rather than binding every
+            // visible id into one SQLite statement, then re-check complete
+            // provenance before emitting any snippet.
+            for (const hit of page.items) {
+              if (
+                !visibleIds.has(hit.header.id)
+                || hit.bestMatch.sessionId !== hit.header.id
+                || hit.bestMatch.surface !== 'current'
+                || !MESSAGE_TYPES.has(hit.bestMatch.type)
+                || acceptedIds.has(hit.header.id)
+              ) continue
+              acceptedIds.add(hit.header.id)
+              authorized.push({
+                sessionId: hit.header.id,
+                snippet: hit.bestMatch.snippet,
+              })
+              if (authorized.length > SESSION_SEARCH_LIMIT) break
+            }
+            if (authorized.length > SESSION_SEARCH_LIMIT || page.nextCursor === undefined) break
+            if (seenCursors.has(page.nextCursor)) {
+              throw new Error('session search provider repeated a continuation cursor')
+            }
+            seenCursors.add(page.nextCursor)
+            cursor = page.nextCursor
+          }
          return ok(request, {
-            items: authorized.slice(0, SESSION_SEARCH_LIMIT).map(hit => ({
-              sessionId: hit.header.id,
-              snippet: hit.bestMatch.snippet,
-            })),
-            hasMore: page.nextCursor !== undefined || authorized.length > SESSION_SEARCH_LIMIT,
+            items: authorized.slice(0, SESSION_SEARCH_LIMIT),
+            hasMore: authorized.length > SESSION_SEARCH_LIMIT,
          })
        } catch (error: unknown) {
          if (
--- a/packages/host/apiproxy/tests/api-proxy-search.spec.ts
+++ b/packages/host/apiproxy/tests/api-proxy-search.spec.ts
@@ -107,7 +107,6 @@ describe('session.search', () => {
          },
        },
      ],
-      nextCursor: 'more' as never,
    }))
    ctx.provide('sessionQuery', { searchSessions } as never)
    const api = createApiProxy(ctx, defaults)
@@ -119,7 +118,7 @@ describe('session.search', () => {
      ok: true,
      value: {
        items: [{ sessionId: 'cold', snippet: 'the matching answer' }],
-        hasMore: true,
+        hasMore: false,
      },
    })
    expect(searchSessions).toHaveBeenCalledOnce()
@@ -129,7 +128,6 @@ describe('session.search', () => {
    ]
    expect(query).toEqual({
      query: 'matching answer',
-      sessionFilters: [{ kind: 'id', values: ['live', 'cold'] }],
      eventFilters: [
        {
          kind: 'type',
@@ -179,7 +177,6 @@ describe('session.search', () => {
          withBestMatch(2, { type: 'tool/result' }),
          withBestMatch(3, { type: 'steering/message', snippet: 'allowed snippet' }),
        ],
-        nextCursor: 'more',
      }),
    } as never)

@@ -192,19 +189,25 @@ describe('session.search', () => {
      ok: true,
      value: {
        items: [{ sessionId: 'visible', snippet: 'allowed snippet' }],
-        hasMore: true,
+        hasMore: false,
      },
    })
  })

-  it('enforces the 20-item Host boundary even if a provider overproduces', async () => {
+  it('pages the globally ranked stream until the 20-item Host boundary is known', async () => {
    const ctx = await baseContext()
    const items = Array.from({ length: 21 }, (_, index) => hit(`visible-${index}`, index))
    for (const item of items) {
      ctx.sessions.create(item.header.id, { meta: item.header })
    }
+    const searchSessions = vi.fn()
+      .mockResolvedValueOnce({
+        items: [hit('hidden-ranked-first'), ...items.slice(0, 19)],
+        nextCursor: 'page-2',
+      })
+      .mockResolvedValueOnce({ items: items.slice(19) })
    ctx.provide('sessionQuery', {
-      searchSessions: () => Promise.resolve({ items }),
+      searchSessions,
    } as never)
    const response = await createApiProxy(ctx, defaults).sessions.search(
      request('match'),
@@ -218,6 +221,39 @@ describe('session.search', () => {
    if (!response.result.ok) throw new Error('unreachable')
    expect(response.result.value.items).toHaveLength(20)
    expect(response.result.value.items.at(-1)?.sessionId).toBe('visible-19')
+    expect(searchSessions).toHaveBeenCalledTimes(2)
+    expect(searchSessions.mock.calls[1]?.[0]).toMatchObject({ cursor: 'page-2' })
+  })
+
+  it('keeps visibility sets above SQLite variable limits out of provider bindings', async () => {
+    const ctx = await baseContext()
+    const cold = Array.from(
+      { length: 32_751 },
+      (_, index) => header(`cold-${index}`, `/cold-${index}`),
+    )
+    ctx.provide('sessionPersistence', {
+      list: () => Promise.resolve(cold),
+      locate: () => undefined,
+    } as never)
+    const searchSessions = vi.fn(() => Promise.resolve({
+      items: [hit('cold-32750')],
+    }))
+    ctx.provide('sessionQuery', { searchSessions } as never)
+
+    const response = await createApiProxy(ctx, defaults).sessions.search(
+      request('large corpus'),
+      new AbortController().signal,
+    )
+
+    expect(response.result).toEqual({
+      ok: true,
+      value: {
+        items: [{ sessionId: 'cold-32750', snippet: 'match 0' }],
+        hasMore: false,
+      },
+    })
+    expect(searchSessions).toHaveBeenCalledOnce()
+    expect(searchSessions.mock.calls[0]?.[0]).not.toHaveProperty('sessionFilters')
  })

  it('propagates cancellation through visible-session collection and stops cold-summary work', async () => {