diff --git a/packages/host/apiproxy/src/api-proxy.ts b/packages/host/apiproxy/src/api-proxy.ts index 698d77a57f..e25a30b94a 100644 --- a/packages/host/apiproxy/src/api-proxy.ts +++ b/packages/host/apiproxy/src/api-proxy.ts @@ -46,7 +46,7 @@ const DEFAULT_MAX_MESSAGES = 50 /** Provider work budget: at most 100 calls and 2,000 inspected hits. */ const SESSION_SEARCH_PROVIDER_CALL_LIMIT = 100 -/** Bound cold-log stat fan-out so an aborted search stops launching new work. */ +/** Bound cold-log stat fan-out and settle each started batch before cancellation returns. */ const COLD_SUMMARY_BATCH_SIZE = 16 /** Surface message event types (the pagination counting unit). */ @@ -615,10 +615,23 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro for (let offset = 0; offset < cold.length; offset += COLD_SUMMARY_BATCH_SIZE) { signal?.throwIfAborted() const batch = cold.slice(offset, offset + COLD_SUMMARY_BATCH_SIZE) - items.push(...await Promise.all( + const settled = await Promise.allSettled( batch.map(meta => summarizeCold(persistence, meta, signal)), - )) + ) + const summaries: SessionSummary[] = [] + let rejected = false + let failure: unknown + for (const result of settled) { + if (result.status === 'fulfilled') { + summaries.push(result.value) + } else if (!rejected) { + rejected = true + failure = result.reason + } + } + if (rejected) throw failure signal?.throwIfAborted() + items.push(...summaries) } } items.sort((a, b) => b.updatedAt - a.updatedAt) diff --git a/packages/host/apiproxy/tests/api-proxy-search.spec.ts b/packages/host/apiproxy/tests/api-proxy-search.spec.ts index e8229ddfca..c3644af5af 100644 --- a/packages/host/apiproxy/tests/api-proxy-search.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-search.spec.ts @@ -6,6 +6,7 @@ import { describe, expect, it, vi } from 'vitest' import { Context } from 'cordis' +import { stat } from 'node:fs/promises' import AgentRegistry from '@deepseek-ai/dsh-agent' import SessionStore from '@deepseek-ai/dsh-session' import type { SessionHeader, SessionId } from '@deepseek-ai/dsh-session' @@ -19,6 +20,11 @@ import type { RpcRequest } from '@deepseek-ai/dsh-host-apiproxy/api' import { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api' import { createApiProxy } from '@deepseek-ai/dsh-host-apiproxy' +vi.mock('node:fs/promises', async (importOriginal) => { + const actual = await importOriginal() + return { ...actual, stat: vi.fn(actual.stat) } +}) + const sid = (value: string): SessionId => value as SessionId const defaults = { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' } @@ -777,6 +783,48 @@ describe('session.search', () => { expect(searchSessions).not.toHaveBeenCalled() }) + it('awaits every started cold-summary stat before returning cancellation', async () => { + const ctx = await baseContext() + const controller = new AbortController() + const cold = Array.from({ length: 16 }, (_, index) => header(`cold-${index}`, `/cold-${index}`)) + const statGates = cold.map(() => Promise.withResolvers<{ mtimeMs: number }>()) + const statMock = vi.mocked(stat) + statMock.mockClear() + for (const gate of statGates) { + statMock.mockImplementationOnce((() => gate.promise) as never) + } + ctx.provide('sessionPersistence', { + list: () => Promise.resolve(cold), + locate: (meta: SessionHeader) => ({ kind: 'jsonl', path: `/logs/${meta.id}.jsonl` }), + } as never) + const searchSessions = vi.fn() + ctx.provide('sessionQuery', { searchSessions } as never) + + let settled = false + const responsePromise = createApiProxy(ctx, defaults).sessions.search( + request('cancel-during-cold-stats'), + controller.signal, + ).finally(() => { + settled = true + }) + await vi.waitFor(() => { + expect(statMock).toHaveBeenCalledTimes(16) + }) + + controller.abort() + statGates[0]!.resolve({ mtimeMs: 101 }) + await new Promise(resolve => setImmediate(resolve)) + expect(settled).toBe(false) + + for (const gate of statGates.slice(1)) gate.resolve({ mtimeMs: 102 }) + const response = await responsePromise + expect(response.result).toMatchObject({ + ok: false, + error: { code: 'cancelled' }, + }) + expect(searchSessions).not.toHaveBeenCalled() + }) + it('maps missing composition, query cancellation, and provider failure', async () => { const missingCtx = await baseContext() missingCtx.sessions.create(sid('visible'), { meta: header('visible') })