Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(web): converge session search boundaries (round 9)

Browse Source
This commit is contained in:
Hypatia May
2026-07-27 14:46:08 +08:00
parent ba2925c704
commit 0aa7f8c5cf
29 changed files with 629 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/host/apiproxy/README.i18n.yaml
View File

@@ -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: e61de41a14294b8c1601e5be8cab19fdf780916d
README.zh.md: 7e49ad49aaa9356412f90b37237b06ce65776e1c
README.md: ada4c7c39d7e4dd1b785a8883096551d9f703826
README.zh.md: 82aa35a2b5d366c1999e209480c56a679bac988c

4
packages/host/apiproxy/README.md
View File

@@ -14,9 +14,9 @@ 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 asks the optional `ctx.sessionQuery` service for globally ranked current-surface user, assistant, and steering matches in pages capped at 20 hits, consumes 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. Returned snippets contain at most 240 Unicode code points; a malformed non-string provider snippet fails closed instead of crossing the RPC boundary. Keeping the authorization set in Host memory avoids SQLite's variable ceiling for large valid corpora without weakening visibility or ranking.
`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, consumes 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. Provider pages start at 20 hits; when a first-page request rejects that limit, the gateway probes 10, 5, 2, then 1 and retains the learned size for continuation and stale-generation restarts. Returned snippets contain at most 240 Unicode code points; a malformed non-string provider snippet fails closed at the Host, and the response schema independently rejects an oversized snippet at each client boundary. Keeping the authorization set in Host memory avoids SQLite's variable ceiling for large valid corpora without weakening visibility or ranking.
A stale continuation discards every partial result, deduplication entry, and cursor from that provider attempt, then restarts at the first page against the original list-derived visibility snapshot. Stale retries share the same limit of at most 100 provider calls (and therefore at most 2,000 inspected hits); an oversized page, a repeated continuation cursor, or a still-unexhausted stream at that call budget fails closed as an `internal` business error. The carrier request signal cancels persistence listing, cold-summary collection, and every search call, including a stale rejection observed concurrently with cancellation. A deployment without the service, or any unrecovered index/query failure, also returns an `internal` business error so clients can retain metadata-only matches.
A stale continuation discards every partial result, deduplication entry, and cursor from that provider attempt, then restarts at the first page against the original list-derived visibility snapshot without discarding the learned provider page size. Limit probes and stale retries share the same limit of at most 100 provider calls (and therefore at most 2,000 inspected hits); a page larger than its requested limit, a repeated continuation cursor, or a still-unexhausted stream at that call budget fails closed as an `internal` business error. The carrier request signal cancels persistence listing, cold-summary collection, and every search call, including a limit or stale rejection observed concurrently with cancellation. A deployment without the service, or any unrecovered index/query failure, also 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.

4
packages/host/apiproxy/README.zh.md
View File

@@ -14,9 +14,9 @@ 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` 所列会话为范围的有界内容搜索投影。网关向可选的 `ctx.sessionQuery` 服务请求全局排序后的当前 surface user、assistant 和 steering（中途引导）匹配项，每页至多 20 个命中，并持续消费该结果流，直到获得至多 20 个可见会话／snippet 对及一个前瞻项；返回前仍会依据从列表推导的授权集合重新校验每个命中。返回的 snippet 最多包含 240 个 Unicode 码点；如果提供方返回格式错误的非字符串 snippet，系统会直接失败，而不会让它越过 RPC 边界。将授权集合保留在宿主内存中，可在不削弱可见性或排序的前提下避开有效大型语料库的 SQLite 变量上限。
`session.search` 是以 `session.list` 所列会话为范围的有界内容搜索投影。网关向可选的 `ctx.sessionQuery` 服务请求全局排序后的当前 surface user、assistant 和 steering（中途引导）匹配项，并持续消费该结果流，直到获得至多 20 个可见会话／snippet 对及一个前瞻项；返回前仍会依据从列表推导的授权集合重新校验每个命中。提供方分页初始请求 20 个命中；如果第一页请求因这一上限被拒绝，网关会依次探测 10、5、2、1，并在续传和陈旧世代重启中沿用探测所得的页面大小。返回的 snippet 最多包含 240 个 Unicode 码点；如果提供方返回格式错误的非字符串 snippet，系统会在宿主侧直接失败，响应 schema 则会在每个客户端边界独立拒绝超长的 snippet。将授权集合保留在宿主内存中，可在不削弱可见性或排序的前提下避开有效大型语料库的 SQLite 变量上限。
陈旧的续传会丢弃该提供方尝试中的所有部分结果、去重条目和游标，然后依据最初从列表推导的可见性快照从第一页重新开始。陈旧重试共用最多 100 次提供方调用的限制（因此最多检查 2,000 个命中）；如果单页命中数超限、续传游标重复，或用尽该调用预算后结果流仍未耗尽，都会直接返回 `internal` 业务错误，不返回部分结果。载体请求信号可取消持久化列表枚举、冷会话摘要收集和每一次搜索调用；即使同时收到陈旧拒绝，也以取消为准。部署若未挂载该服务，或索引／查询故障无法恢复，也会返回 `internal` 业务错误，以便客户端保留仅基于元数据的匹配项。
陈旧的续传会丢弃该提供方尝试中的所有部分结果、去重条目和游标，然后依据最初从列表推导的可见性快照从第一页重新开始，但不会丢弃探测所得的提供方页面大小。上限探测与陈旧重试共用最多 100 次提供方调用的限制（因此最多检查 2,000 个命中）；如果某页命中数超过其请求的上限、续传游标重复，或用尽该调用预算后结果流仍未耗尽，都会直接返回 `internal` 业务错误，不返回部分结果。载体请求信号可取消持久化列表枚举、冷会话摘要收集和每一次搜索调用；即使同时收到上限拒绝或陈旧拒绝，也以取消为准。部署若未挂载该服务，或索引／查询故障无法恢复，也会返回 `internal` 业务错误，以便客户端保留仅基于元数据的匹配项。
`command.*` 与 `skill.*` 领域向客户端暴露宿主命令注册表和技能目录。每个方法都通过 `sessionId` 寻址一个会话的 Agent（被服务的会话必有 Agent；`command.*` 经由与 `session.*` 相同的路径恢复冷会话，而 `skill.list` 从会话头解析项目根目录，不触碰 Agent 注册表）。`command.execute` 在宿主侧运行一条斜杠命令行并返回脱耦结果；载体的请求信号可取消正在运行的处理器。`host/commands-changed` 是目录失效帧：客户端重新拉取 `command.list` 而不是做差分。

17
packages/host/apiproxy/src/api-proxy.ts
View File

@@ -674,6 +674,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
const seenCursors = new Set<SessionSearchCursor>()
let cursor: SessionSearchCursor | undefined
let providerCallCount = 0
let providerPageLimit = SESSION_SEARCH_LIMIT
while (authorized.length <= SESSION_SEARCH_LIMIT) {
if (isAborted(signal)) return cancelled()
if (providerCallCount >= SESSION_SEARCH_PROVIDER_CALL_LIMIT) {
@@ -683,6 +684,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
}
providerCallCount++
const requestedCursor = cursor
const requestedPageLimit = providerPageLimit
let page
try {
page = await sessionQuery.searchSessions({
@@ -691,11 +693,20 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
{ kind: 'type', values: ['user/message', 'assistant/message', 'steering/message'] },
{ kind: 'surface', values: ['current'] },
],
limit: SESSION_SEARCH_LIMIT,
limit: requestedPageLimit,
...requestedCursor === undefined ? {} : { cursor: requestedCursor },
}, { signal })
} catch (error: unknown) {
if (isAborted(signal)) return cancelled()
if (
requestedCursor === undefined
&& error instanceof SessionQueryError
&& error.code === 'SESSION_QUERY_INVALID_LIMIT'
&& requestedPageLimit > 1
) {
providerPageLimit = Math.max(1, Math.floor(requestedPageLimit / 2))
continue
}
if (
requestedCursor !== undefined
&& error instanceof SessionQueryError
@@ -711,9 +722,9 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
}
if (isAborted(signal)) return cancelled()
const providerItemCount = page.items.length
if (providerItemCount > SESSION_SEARCH_LIMIT) {
if (providerItemCount > requestedPageLimit) {
throw new Error(
`session search provider returned ${providerItemCount} items; maximum is ${SESSION_SEARCH_LIMIT}`,
`session search provider returned ${providerItemCount} items; maximum is ${requestedPageLimit}`,
)
}
// Host visibility is the authorization boundary. Consume the

25
packages/host/apiproxy/src/api/sessions.schema.ts
View File

@@ -58,6 +58,26 @@ export const sessionListValueSchema = z.object({
const SESSION_SEARCH_QUERY_MAX_CHARS = 500
/** Product response bound validated independently by every client carrier. */
const SESSION_SEARCH_RESULT_LIMIT = 20
/** Maximum response snippet length in Unicode code points. */
const SESSION_SEARCH_SNIPPET_MAX_CODE_POINTS = 240
/** Early-exit Unicode code-point bound without materializing an iterator result. */
function hasAtMostCodePoints(value: string, maximum: number): boolean {
let count = 0
let offset = 0
while (offset < value.length) {
if (count === maximum) return false
const first = value.charCodeAt(offset)
const paired = first >= 0xD800
&& first <= 0xDBFF
&& offset + 1 < value.length
&& value.charCodeAt(offset + 1) >= 0xDC00
&& value.charCodeAt(offset + 1) <= 0xDFFF
offset += paired ? 2 : 1
count++
}
return true
}
/** session.search request payload. */
export const sessionSearchRequestSchema = z.object({
@@ -68,7 +88,10 @@ export const sessionSearchRequestSchema = z.object({
/** One session.search result. */
export const sessionSearchItemSchema = z.object({
sessionId: sessionIdSchema,
snippet: z.string(),
snippet: z.string().refine(
snippet => hasAtMostCodePoints(snippet, SESSION_SEARCH_SNIPPET_MAX_CODE_POINTS),
{ message: `search snippet must contain at most ${SESSION_SEARCH_SNIPPET_MAX_CODE_POINTS} Unicode code points` },
),
}) satisfies z.ZodType<Wire<SessionSearchItem>>
/** session.search response value. */

192
packages/host/apiproxy/tests/api-proxy-search.spec.ts
View File

@@ -225,16 +225,70 @@ describe('session.search', () => {
expect(searchSessions.mock.calls[1]?.[0]).toMatchObject({ cursor: 'page-2' })
})
it('fails closed after 100 provider calls with distinct continuation cursors', async () => {
it('learns a provider maxLimit of 10 and collects the 20-item result plus lookahead', 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 invalidLimit = new SessionQueryError(
'provider accepts at most 10 items',
'SESSION_QUERY_INVALID_LIMIT',
)
const searchSessions = vi.fn((providerRequest: SessionSearchRequest) => {
const limit = providerRequest.limit
if (limit === undefined) throw new Error('Host search must request an explicit provider limit')
if (limit > 10) return Promise.reject(invalidLimit)
const offset = providerRequest.cursor === undefined
? 0
: Number.parseInt(providerRequest.cursor.slice('offset-'.length), 10)
const end = Math.min(items.length, offset + limit)
return Promise.resolve({
items: items.slice(offset, end),
...end < items.length ? { nextCursor: `offset-${end}` } : {},
})
})
ctx.provide('sessionQuery', { searchSessions } as never)
const response = await createApiProxy(ctx, defaults).sessions.search(
request('adaptive-page-limit'),
new AbortController().signal,
)
expect(response.result).toMatchObject({
ok: true,
value: { hasMore: true },
})
if (!response.result.ok) throw new Error('unreachable')
expect(response.result.value.items.map(item => item.sessionId))
.toEqual(items.slice(0, 20).map(item => item.header.id))
expect(searchSessions.mock.calls.map(([providerRequest]) => ({
limit: providerRequest.limit,
cursor: providerRequest.cursor,
}))).toEqual([
{ limit: 20, cursor: undefined },
{ limit: 10, cursor: undefined },
{ limit: 10, cursor: 'offset-10' },
{ limit: 10, cursor: 'offset-20' },
])
})
it('counts a page-limit probe inside the 100-call budget', async () => {
const ctx = await baseContext()
ctx.sessions.create(sid('visible'), { meta: header('visible') })
let pageNumber = 0
const invalidLimit = new SessionQueryError(
'provider accepts at most 10 items',
'SESSION_QUERY_INVALID_LIMIT',
)
const searchSessions = vi.fn((providerRequest: SessionSearchRequest) => {
pageNumber++
expect(providerRequest.limit).toBe(20)
if (searchSessions.mock.calls.length === 1) {
expect(providerRequest).toMatchObject({ limit: 20 })
return Promise.reject(invalidLimit)
}
expect(providerRequest.limit).toBe(10)
return Promise.resolve({
items: [],
nextCursor: `page-${pageNumber}`,
nextCursor: `page-${searchSessions.mock.calls.length}`,
})
})
ctx.provide('sessionQuery', { searchSessions } as never)
@@ -251,7 +305,7 @@ describe('session.search', () => {
expect(searchSessions).toHaveBeenCalledTimes(100)
})
it('restarts a stale continuation from one fresh generation and keeps the visibility snapshot', async () => {
it('restarts a stale continuation with its learned limit and original visibility snapshot', async () => {
const ctx = await baseContext()
const oldOnly = hit('old-only', 0)
const shared = hit('shared', 1)
@@ -265,25 +319,37 @@ describe('session.search', () => {
'provider generation changed',
'SESSION_QUERY_STALE_CURSOR',
)
const invalidLimit = new SessionQueryError(
'provider accepts at most 10 items',
'SESSION_QUERY_INVALID_LIMIT',
)
const searchSessions = vi.fn((providerRequest: SessionSearchRequest) => {
switch (searchSessions.mock.calls.length) {
case 1:
expect(providerRequest).toMatchObject({ limit: 20 })
expect(providerRequest).not.toHaveProperty('cursor')
return Promise.reject(invalidLimit)
case 2:
expect(providerRequest).toMatchObject({ limit: 10 })
expect(providerRequest).not.toHaveProperty('cursor')
return Promise.resolve({
items: [oldOnly, shared],
nextCursor: 'old-cursor',
})
case 2:
case 3:
expect(providerRequest).toMatchObject({ limit: 10 })
expect(providerRequest.cursor).toBe('old-cursor')
ctx.sessions.create(late.header.id, { meta: late.header })
return Promise.reject(stale)
case 3:
case 4:
expect(providerRequest).toMatchObject({ limit: 10 })
expect(providerRequest).not.toHaveProperty('cursor')
return Promise.resolve({
items: [freshFirst, shared],
nextCursor: 'old-cursor',
})
case 4:
case 5:
expect(providerRequest).toMatchObject({ limit: 10 })
expect(providerRequest.cursor).toBe('old-cursor')
return Promise.resolve({ items: [freshLast, late] })
default:
@@ -308,7 +374,7 @@ describe('session.search', () => {
hasMore: false,
},
})
expect(searchSessions).toHaveBeenCalledTimes(4)
expect(searchSessions).toHaveBeenCalledTimes(5)
})
it('counts continuous stale restarts against the 100-call budget', async () => {
@@ -394,6 +460,82 @@ describe('session.search', () => {
expect(searchSessions).toHaveBeenCalledOnce()
})
it('does not adapt an invalid-limit continuation failure', async () => {
const ctx = await baseContext()
ctx.sessions.create(sid('visible'), { meta: header('visible') })
const searchSessions = vi.fn()
.mockResolvedValueOnce({ items: [], nextCursor: 'page-2' })
.mockRejectedValueOnce(new SessionQueryError(
'continuation limit is invalid',
'SESSION_QUERY_INVALID_LIMIT',
))
ctx.provide('sessionQuery', { searchSessions } as never)
const response = await createApiProxy(ctx, defaults).sessions.search(
request('continuation-invalid-limit'),
new AbortController().signal,
)
expect(response.result).toMatchObject({
ok: false,
error: { code: 'internal' },
})
expect(searchSessions).toHaveBeenCalledTimes(2)
expect(searchSessions.mock.calls.map(([providerRequest]) => (
providerRequest as SessionSearchRequest
).limit))
.toEqual([20, 20])
})
it('stops page-limit adaptation at one item', async () => {
const ctx = await baseContext()
ctx.sessions.create(sid('visible'), { meta: header('visible') })
const searchSessions = vi.fn((providerRequest: SessionSearchRequest) => Promise.reject(
new SessionQueryError(
`provider rejects ${providerRequest.limit}`,
'SESSION_QUERY_INVALID_LIMIT',
),
))
ctx.provide('sessionQuery', { searchSessions } as never)
const response = await createApiProxy(ctx, defaults).sessions.search(
request('minimum-page-limit'),
new AbortController().signal,
)
expect(response.result).toMatchObject({
ok: false,
error: { code: 'internal' },
})
expect(searchSessions.mock.calls.map(([providerRequest]) => providerRequest.limit))
.toEqual([20, 10, 5, 2, 1])
})
it('gives abort priority over a coincident invalid first-page limit', async () => {
const ctx = await baseContext()
ctx.sessions.create(sid('visible'), { meta: header('visible') })
const controller = new AbortController()
const searchSessions = vi.fn(() => {
controller.abort()
return Promise.reject(new SessionQueryError(
'provider rejects 20',
'SESSION_QUERY_INVALID_LIMIT',
))
})
ctx.provide('sessionQuery', { searchSessions } as never)
const response = await createApiProxy(ctx, defaults).sessions.search(
request('abort-invalid-limit'),
controller.signal,
)
expect(response.result).toMatchObject({
ok: false,
error: { code: 'cancelled' },
})
expect(searchSessions).toHaveBeenCalledOnce()
})
it('rejects an oversized provider page before iterating its items', async () => {
const ctx = await baseContext()
ctx.sessions.create(sid('visible'), { meta: header('visible') })
@@ -415,6 +557,36 @@ describe('session.search', () => {
expect(iterate).not.toHaveBeenCalled()
})
it('uses the learned provider limit for the overproduction guard', async () => {
const ctx = await baseContext()
ctx.sessions.create(sid('visible'), { meta: header('visible') })
const oversized = new Array<SessionSearchHit>(11)
const iterate = vi.fn(() => oversized.values())
Object.defineProperty(oversized, Symbol.iterator, { value: iterate })
const searchSessions = vi.fn((providerRequest: SessionSearchRequest) => {
if (providerRequest.limit === 20) {
return Promise.reject(new SessionQueryError(
'provider accepts at most 10 items',
'SESSION_QUERY_INVALID_LIMIT',
))
}
return Promise.resolve({ items: oversized })
})
ctx.provide('sessionQuery', { searchSessions } as never)
const response = await createApiProxy(ctx, defaults).sessions.search(
request('adapted-oversized-page'),
new AbortController().signal,
)
expect(response.result.ok).toBe(false)
if (response.result.ok) throw new Error('unreachable')
expect(response.result.error).toMatchObject({ code: 'internal' })
expect(response.result.error.message).toContain('returned 11 items; maximum is 10')
expect(searchSessions).toHaveBeenCalledTimes(2)
expect(iterate).not.toHaveBeenCalled()
})
it('bounds provider snippets to 240 Unicode code points without splitting astral text', async () => {
const ctx = await baseContext()
const visible = hit('visible')

14
packages/host/apiproxy/tests/client-handler.spec.ts
View File

@@ -101,6 +101,20 @@ describe('unary round trip', () => {
})
})
it('rejects an overlong session-search snippet at the client value boundary', async () => {
const api = scriptedApi({
sessions: {
search: request => ok(request, {
items: [{ sessionId: sid('s1'), snippet: '😀'.repeat(241) }],
hasMore: false,
}),
},
})
await expect(client(api).sessions.search({ query: 'message' }))
.rejects.toThrow(/240 Unicode code points/)
})
it('routes workspace rename and insertSessionBefore through the wire', async () => {
const api = scriptedApi()
const c = client(api)

8
packages/host/apiproxy/tests/rpc-schemas.spec.ts
View File

@@ -132,6 +132,14 @@ describe('sessions domain schemas', () => {
items: [{ sessionId: 's1', snippet: 'matching text' }],
hasMore: true,
})
expect(sessionSearchValueSchema.parse({
items: [{ sessionId: 's1', snippet: '😀'.repeat(240) }],
hasMore: false,
}).items[0]?.snippet).toBe('😀'.repeat(240))
expect(() => sessionSearchValueSchema.parse({
items: [{ sessionId: 's1', snippet: '😀'.repeat(241) }],
hasMore: false,
})).toThrow(/240 Unicode code points/)
expect(() => sessionSearchValueSchema.parse({
items: [{ sessionId: '', snippet: 'matching text' }],
hasMore: false,