fix: serialize paginated session searches

packages/session-query/session-query-sqlite/README.md

@@ -28,6 +28,7 @@ The database is disposable but reset is guarded: every recognized schema version
 | `maxLimit` | `100` | Largest accepted request page size; at most `Number.MAX_SAFE_INTEGER - 1`. |
 | `snippetChars` | `240` | Maximum snippet length in Unicode code points. |
 | `readWindowMax` | `50` | Maximum `before` or `after` raw-event count for inherited `readEvent()`. |
+| `persistedInspectConcurrency` | `4` | Maximum concurrent persisted-log inspections for inherited batch reads; must be a positive safe integer. |
 
 ## Tokenizer and limits
 

packages/session-query/session-query-sqlite/src/index.ts

@@ -15,6 +15,7 @@ import type {
   SessionPersistenceSnapshot,
 } from '@deepseek-ai/dsh-session-persistence'
 import SessionQueryService, {
+  SESSION_QUERY_DEFAULT_PERSISTED_INSPECT_CONCURRENCY,
   SESSION_QUERY_READ_WINDOW_MAX,
   SessionQueryError,
   SessionSearchCursor,
@@ -87,6 +88,8 @@ export interface Config extends SessionQueryConfig {
   maxLimit?: number
   /** Maximum snippet length in Unicode code points. Defaults to 240. */
   snippetChars?: number
+  /** Maximum concurrent persisted-log inspections in one inherited batch read. Defaults to 4. */
+  persistedInspectConcurrency?: number
 }
 
 interface ResolvedConfig {
@@ -96,6 +99,7 @@ interface ResolvedConfig {
   maxLimit: number
   snippetChars: number
   readWindowMax: number
+  persistedInspectConcurrency: number
 }
 
 interface ObservedSession {
@@ -176,6 +180,11 @@ export class SessionQuerySqlite extends SessionQueryService {
     maxLimit: z.number().step(1).min(1).max(SQLITE_MAX_PAGE_LIMIT).default(SESSION_QUERY_SQLITE_MAX_LIMIT),
     snippetChars: z.number().step(1).min(1).default(SESSION_QUERY_SQLITE_SNIPPET_CHARS),
     readWindowMax: z.number().step(1).min(0).default(SESSION_QUERY_READ_WINDOW_MAX),
+    persistedInspectConcurrency: z.number()
+      .step(1)
+      .min(1)
+      .max(Number.MAX_SAFE_INTEGER)
+      .default(SESSION_QUERY_DEFAULT_PERSISTED_INSPECT_CONCURRENCY),
   })
 
   /** Validated and defaulted backend configuration. */
@@ -937,6 +946,8 @@ function resolveConfig(config: Config): ResolvedConfig {
     maxLimit: config.maxLimit ?? SESSION_QUERY_SQLITE_MAX_LIMIT,
     snippetChars: config.snippetChars ?? SESSION_QUERY_SQLITE_SNIPPET_CHARS,
     readWindowMax: config.readWindowMax ?? SESSION_QUERY_READ_WINDOW_MAX,
+    persistedInspectConcurrency: config.persistedInspectConcurrency
+      ?? SESSION_QUERY_DEFAULT_PERSISTED_INSPECT_CONCURRENCY,
   }
   if (typeof resolved.path !== 'string' || resolved.path.trim().length === 0) {
     throw invalidConfig('path must not be blank')
@@ -947,6 +958,12 @@ function resolveConfig(config: Config): ResolvedConfig {
   if (!Number.isInteger(resolved.readWindowMax) || resolved.readWindowMax < 0) {
     throw invalidConfig('readWindowMax must be a non-negative integer')
   }
+  if (
+    !Number.isSafeInteger(resolved.persistedInspectConcurrency)
+    || resolved.persistedInspectConcurrency < 1
+  ) {
+    throw invalidConfig('persistedInspectConcurrency must be a positive safe integer')
+  }
   if (resolved.defaultLimit > resolved.maxLimit) {
     throw invalidConfig('defaultLimit must be less than or equal to maxLimit')
   }

packages/session-query/session-query-sqlite/tests/sqlite.spec.ts

@@ -13,6 +13,7 @@ import SessionQuerySqlite, {
   SESSION_QUERY_SQLITE_SCHEMA_VERSION,
 } from '@deepseek-ai/dsh-session-query-sqlite'
 import {
+  SESSION_QUERY_DEFAULT_PERSISTED_INSPECT_CONCURRENCY,
   SessionQueryError,
   SessionSearchCursor,
   type SessionAvailability,
@@ -167,6 +168,29 @@ async function liveContext(config: ConstructorParameters<typeof SessionQuerySqli
 }
 
 describe('SQLite session search', () => {
+  it('defaults and validates persisted inspection concurrency through its Cordis config', async () => {
+    const defaultCtx = await liveContext()
+    expect((defaultCtx.sessionQuery as SessionQuerySqlite).config.persistedInspectConcurrency)
+      .toBe(SESSION_QUERY_DEFAULT_PERSISTED_INSPECT_CONCURRENCY)
+
+    const configuredValue = 2
+    const configured = new SessionQuerySqlite.Config({
+      path: ':memory:',
+      persistedInspectConcurrency: configuredValue,
+    })
+    expect(configured.persistedInspectConcurrency).toBe(configuredValue)
+    const configuredCtx = await liveContext(configured)
+    expect((configuredCtx.sessionQuery as SessionQuerySqlite).config.persistedInspectConcurrency)
+      .toBe(configuredValue)
+
+    for (const persistedInspectConcurrency of [0, Number.MAX_SAFE_INTEGER + 1]) {
+      expect(() => new SessionQuerySqlite.Config({
+        path: ':memory:',
+        persistedInspectConcurrency,
+      })).toThrow()
+    }
+  })
+
   it('searches two-character Unicode61 tokens in live-only sessions', async () => {
     const ctx = await liveContext({ path: ':memory:', snippetChars: 20 })
     const session = ctx.sessions.create(SessionId('live'), {
@@ -486,6 +510,8 @@ describe('SQLite session search', () => {
       { path: ':memory:', maxLimit: 1e100 },
       { path: ':memory:', snippetChars: 0 },
       { path: ':memory:', readWindowMax: -1 },
+      { path: ':memory:', persistedInspectConcurrency: 0 },
+      { path: ':memory:', persistedInspectConcurrency: Number.MAX_SAFE_INTEGER + 1 },
       { path: ':memory:', defaultLimit: 3, maxLimit: 2 },
       { path: ':memory:', journalMode: 'memory' },
     ]) {

packages/session-query/session-query/README.md

@@ -15,7 +15,7 @@
 - `traceSession(sessionId)` reads the corpus once and returns immediate-to-outward ancestors plus deterministic recursive descendant trees. `complete: false` identifies the first missing parent; a target-connected cycle fails with `SESSION_QUERY_INVALID_LINEAGE`.
 - `traceEvent(request)` loads the logical log once and returns its cloned source header with direct positional replacements and direct logged provenance. `replacementChain` follows positional replacers to the final replacement; provenance links remain non-transitive.
 
-Persistence is optional and may mount or unmount dynamically. Cross-corpus listing and lineage tracing fail with `SESSION_QUERY_PERSISTENCE_FAILED` while mounted persistence is unreadable. A title, event read, or trace targeting a known live session does not consult persistence, so durable backend health cannot make current in-memory state unreadable. Persisted title and event operations list before loading and reject a metadata mismatch rather than combining inconsistent observations. A batch title observation performs one metadata listing, inspects its unique persisted ids with at most four workers, and preserves each title's own observed header for downstream authorization. Cancellation starts no queued inspections and rejects only after already-started workers settle. `listSessions()` remains lightweight and does not load logs or index titles.
+Persistence is optional and may mount or unmount dynamically. Cross-corpus listing and lineage tracing fail with `SESSION_QUERY_PERSISTENCE_FAILED` while mounted persistence is unreadable. A title, event read, or trace targeting a known live session does not consult persistence, so durable backend health cannot make current in-memory state unreadable. Persisted title and event operations list before loading and reject a metadata mismatch rather than combining inconsistent observations. A batch title observation performs one metadata listing, inspects its unique persisted ids with at most `persistedInspectConcurrency` workers, and preserves each title's own observed header for downstream authorization. Cancellation starts no queued inspections and rejects only after already-started workers settle. `listSessions()` remains lightweight and does not load logs or index titles.
 
 ## Filtering and extraction
 
@@ -38,6 +38,7 @@ The package has no provider coordinator, fallback implementation, or standalone
 | Key | Default | Contract |
 |---|---:|---|
 | `readWindowMax` | `50` | Maximum `before` or `after` raw-event count. |
+| `persistedInspectConcurrency` | `4` | Maximum concurrent persisted-log inspections in one batch read; must be a positive safe integer. |
 
 ## Model Experience
 

packages/session-query/session-query/src/config.ts

@@ -5,10 +5,15 @@ import { HarnessError } from '@deepseek-ai/dsh-llm'
 /** Default maximum `before`/`after` raw-event window. */
 export const SESSION_QUERY_READ_WINDOW_MAX = 50
 
+/** Default maximum number of concurrent persisted-log inspections in one batch read. */
+export const SESSION_QUERY_DEFAULT_PERSISTED_INSPECT_CONCURRENCY = 4
+
 /** Backend-independent configuration inherited by every session-query implementation. */
 export interface Config {
   /** Maximum accepted raw read context on either side. Defaults to 50. */
   readWindowMax?: number
+  /** Maximum concurrent persisted-log inspections in one batch read. Defaults to 4. */
+  persistedInspectConcurrency?: number
 }
 
 /** Stable machine-routable failure taxonomy for session reads, traces, and search. */

packages/session-query/session-query/src/corpus.ts

@@ -28,15 +28,15 @@ export type LogicalProjectionResult<Value> =
   | { sessionId: SessionId; status: 'fulfilled'; value: Value }
   | { sessionId: SessionId; status: 'rejected'; reason: unknown }
 
-/** Bound persisted observation fan-out for public batch title reads. */
-const PERSISTED_INSPECT_CONCURRENCY = 4
-
 /** Resolves a live-preferred corpus against the persistence service mounted now. */
 export class SessionCorpus {
   private _persistence: SessionPersistence | undefined
   private readonly _optionalPersistenceFiber: Fiber
 
-  constructor(private readonly _ctx: Context) {
+  constructor(
+    private readonly _ctx: Context,
+    private readonly _persistedInspectConcurrency: number,
+  ) {
     this._optionalPersistenceFiber = _ctx.inject(['sessionPersistence'], (childCtx: Context) => {
       const service = childCtx.sessionPersistence
       this._persistence = service
@@ -188,7 +188,7 @@ export class SessionCorpus {
         await resolvePersisted(unresolved[index] as SessionId)
       }
     }
-    const workerCount = Math.min(PERSISTED_INSPECT_CONCURRENCY, unresolved.length)
+    const workerCount = Math.min(this._persistedInspectConcurrency, unresolved.length)
     const settlements = await Promise.allSettled(
       Array.from({ length: workerCount }, () => worker()),
     )

packages/session-query/session-query/src/index.ts

@@ -31,6 +31,7 @@ import type {
   SessionTitleObservationResult,
 } from './types.ts'
 import {
+  SESSION_QUERY_DEFAULT_PERSISTED_INSPECT_CONCURRENCY,
   SESSION_QUERY_READ_WINDOW_MAX,
   SessionQueryError,
   type Config,
@@ -48,7 +49,11 @@ import * as tracing from './tracing.ts'
 export type * from './types.ts'
 export { SessionSearchCursor } from './cursor.ts'
 export type { Config, SessionQueryErrorCode } from './config.ts'
-export { SESSION_QUERY_READ_WINDOW_MAX, SessionQueryError } from './config.ts'
+export {
+  SESSION_QUERY_DEFAULT_PERSISTED_INSPECT_CONCURRENCY,
+  SESSION_QUERY_READ_WINDOW_MAX,
+  SessionQueryError,
+} from './config.ts'
 export { extractSessionEventText } from './extraction.ts'
 export { buildSessionEventRecords, buildSessionEventSearchDocuments } from './documents.ts'
 export {
@@ -88,7 +93,15 @@ export abstract class SessionQueryService extends Service {
         'SESSION_QUERY_INVALID_CONFIG',
       )
     }
-    this._corpus = new SessionCorpus(ctx)
+    const persistedInspectConcurrency = config.persistedInspectConcurrency
+      ?? SESSION_QUERY_DEFAULT_PERSISTED_INSPECT_CONCURRENCY
+    if (!Number.isSafeInteger(persistedInspectConcurrency) || persistedInspectConcurrency < 1) {
+      throw new SessionQueryError(
+        'session-query: persistedInspectConcurrency must be a positive safe integer',
+        'SESSION_QUERY_INVALID_CONFIG',
+      )
+    }
+    this._corpus = new SessionCorpus(ctx, persistedInspectConcurrency)
   }
 
   /**

packages/session-query/session-query/tests/session-query.spec.ts

@@ -4,6 +4,7 @@ import SessionStore, { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/ds
 import type { SessionEvent, SessionHeader, SessionId as SessionIdType } from '@deepseek-ai/dsh-session'
 import SessionPersistence, { SessionPersistenceRevision } from '@deepseek-ai/dsh-session-persistence'
 import SessionQueryService, {
+  SESSION_QUERY_DEFAULT_PERSISTED_INSPECT_CONCURRENCY,
   type SessionEventSurface,
   type SessionQueryErrorCode,
 } from '@deepseek-ai/dsh-session-query'
@@ -372,7 +373,7 @@ describe('session-query exact reads', () => {
 
     const results = await ctx.sessionQuery.readTitleSnapshots(entries.map(entry => entry.meta.id))
 
-    expect(maximum).toBe(4)
+    expect(maximum).toBe(SESSION_QUERY_DEFAULT_PERSISTED_INSPECT_CONCURRENCY)
     expect(TestPersistence.listCalls).toBe(1)
     expect(TestPersistence.inspectCalls).toEqual(entries.map(entry => entry.meta.id))
     expect(results.map(result => result.sessionId)).toEqual(entries.map(entry => entry.meta.id))
@@ -466,7 +467,8 @@ describe('session-query exact reads', () => {
       events: eventLog(`queued-${index}`),
     }))
     TestPersistence.reset(entries)
-    const ctx = await liveContext()
+    const persistedInspectConcurrency = 2
+    const ctx = await liveContext({ persistedInspectConcurrency })
     await ctx.plugin(TestPersistence)
     const controller = new AbortController()
     const reason = new Error('cancel queued title batch')
@@ -490,17 +492,21 @@ describe('session-query exact reads', () => {
       () => { batchSettled = true },
       () => { batchSettled = true },
     )
-    await vi.waitFor(() => { expect(TestPersistence.inspectCalls).toHaveLength(4) })
+    await vi.waitFor(() => {
+      expect(TestPersistence.inspectCalls).toHaveLength(persistedInspectConcurrency)
+    })
     controller.abort(reason)
-    await vi.waitFor(() => { expect(abortsObserved).toBe(4) })
+    await vi.waitFor(() => { expect(abortsObserved).toBe(persistedInspectConcurrency) })
 
     expect(batchSettled).toBe(false)
-    expect(TestPersistence.inspectCalls).toEqual(entries.slice(0, 4).map(entry => entry.meta.id))
+    expect(TestPersistence.inspectCalls)
+      .toEqual(entries.slice(0, persistedInspectConcurrency).map(entry => entry.meta.id))
     for (const release of releases) release()
 
     await expect(pending).rejects.toBe(reason)
-    expect(inspectionsSettled).toBe(4)
-    expect(TestPersistence.inspectCalls).toEqual(entries.slice(0, 4).map(entry => entry.meta.id))
+    expect(inspectionsSettled).toBe(persistedInspectConcurrency)
+    expect(TestPersistence.inspectCalls)
+      .toEqual(entries.slice(0, persistedInspectConcurrency).map(entry => entry.meta.id))
   })
 
   it('passes cancellation into a stalled persisted title listing and rejects with its reason', async () => {
@@ -907,10 +913,16 @@ describe('session-query exact reads', () => {
     const direct = new Context()
     await direct.plugin(SessionStore)
     expect(new TestSessionQueryService(direct)).toBeInstanceOf(SessionQueryService)
-    const invalid = new Context()
-    await invalid.plugin(SessionStore)
-    expect(() => new TestSessionQueryService(invalid, { readWindowMax: -1 }))
-      .toThrow(expectCode('SESSION_QUERY_INVALID_CONFIG'))
+    for (const config of [
+      { readWindowMax: -1 },
+      { persistedInspectConcurrency: 0 },
+      { persistedInspectConcurrency: Number.MAX_SAFE_INTEGER + 1 },
+    ]) {
+      const invalid = new Context()
+      await invalid.plugin(SessionStore)
+      expect(() => new TestSessionQueryService(invalid, config))
+        .toThrow(expectCode('SESSION_QUERY_INVALID_CONFIG'))
+    }
   })
 
   it('leaves the optional persistence dependency optional', async () => {

packages/session-query/tool-session-query/README.md

@@ -9,7 +9,7 @@ Workspace-authorized model tools over `ctx.sessionQuery`. The package depends on
 | `maxSearchResults` | `100` | Maximum authorized non-self hits collected across internal provider pages |
 | `searchTimeoutMs` | `30000` | Cooperative deadline attached to both full-text search tools |
 
-The caller comes exclusively from `ToolExecution.exec.agent`. Cross-session access requires exact equality between the target and caller session `cwd` values; a caller without `cwd` can inspect only itself. Search never exposes provider cursors, offsets, page sizes, or a model-controlled limit. Timestamps at the tool boundary require an explicit `Z` or numeric offset and become inclusive epoch-millisecond filters.
+The caller comes exclusively from `ToolExecution.exec.agent`. Cross-session access requires exact equality between the target and caller session `cwd` values; a caller without `cwd` can inspect only itself. Search never exposes provider cursors, offsets, page sizes, or a model-controlled limit. Because one search consumes generation-bound provider cursors internally, both search tools execute exclusively with sibling tool calls; the three exact trace/read tools opt into parallel execution. Timestamps at the tool boundary require an explicit `Z` or numeric offset and become inclusive epoch-millisecond filters.
 
 `session_search` always omits the caller session. A current-session `session_event_search` stops immediately before the step that invoked it, so the active assistant output and logged tool call cannot match themselves. Direct targets are authorized before trace, event, or title reads. Lineage output replaces unauthorized ancestor and descendant boundaries with markers that contain no hidden session id.
 

packages/session-query/tool-session-query/src/index.ts

@@ -211,7 +211,6 @@ export function apply(ctx: Context, config: Config): void {
     parameters: SESSION_SEARCH_PARAMETERS,
     output: TEXT_OUTPUT,
     timeoutMs: resolved.searchTimeoutMs,
-    isConcurrencySafe: () => true,
     execute: (args, exec) => executeSessionSearch(ctx, args, exec, resolved.maxSearchResults),
     presentCall: presentSessionSearchCall,
   }))
@@ -222,7 +221,6 @@ export function apply(ctx: Context, config: Config): void {
     parameters: EVENT_SEARCH_PARAMETERS,
     output: TEXT_OUTPUT,
     timeoutMs: resolved.searchTimeoutMs,
-    isConcurrencySafe: () => true,
     execute: (args, exec) => executeEventSearch(ctx, args, exec, resolved.maxSearchResults),
     presentCall: presentEventSearchCall,
   }))

packages/session-query/tool-session-query/tests/tool-session-query.spec.ts

@@ -248,15 +248,13 @@ describe('registration and schemas', () => {
     expect(sessionSchema?.parameters).not.toHaveProperty('properties.cwd')
     expect(mounted.ctx.tools.get('session_search')?.timeoutMs).toBe(1234)
     expect(mounted.ctx.tools.get('session_trace')?.timeoutMs).toBeUndefined()
-    const safeArgs: Record<string, unknown> = {
-      session_search: { query: 'q' },
-      session_event_search: { query: 'q' },
+    const parallelArgs: Record<string, unknown> = {
       session_trace: {},
       session_event_trace: { seq: 0 },
       session_event_read: { seq: 0 },
     }
-    for (const name of names) {
-      expect(mounted.ctx.tools.get(name)?.isConcurrencySafe?.(safeArgs[name])).toBe(true)
+    for (const [name, args] of Object.entries(parallelArgs)) {
+      expect(mounted.ctx.tools.get(name)?.isConcurrencySafe?.(args)).toBe(true)
     }
     expect(mounted.ctx.tools.get('session_search')?.output.render({}, 'rendered'))
       .toEqual([{ type: 'text', text: 'rendered' }])
@@ -287,6 +285,27 @@ describe('registration and schemas', () => {
       .not.toContain('tool:session-query')
   })
 
+  it('keeps generation-bound searches exclusive while exact observations remain parallel', async () => {
+    const mounted = await mount()
+    const classifications = [
+      ['session_search', { query: 'q' }, 'exclusive'],
+      ['session_event_search', { query: 'q' }, 'exclusive'],
+      ['session_trace', {}, 'parallel'],
+      ['session_event_trace', { seq: 0 }, 'parallel'],
+      ['session_event_read', { seq: 0 }, 'parallel'],
+    ] as const
+
+    for (const [name, args, kind] of classifications) {
+      expect(mounted.ctx.tools.executionMode({
+        name,
+        arguments: args,
+        callId: CallId(`mode-${name}`),
+        signal: new AbortController().signal,
+        agent: fakeAgent(mounted.caller),
+      })).toEqual({ kind })
+    }
+  })
+
   it('fails invalid direct config before registering anything', async () => {
     const mounted = await mount()
     for (const maxSearchResults of [0, 1.5, Number.NaN]) {