fix: cancel session authorization reads

docs/cordis-catalog/services.md

@@ -994,9 +994,10 @@ abstract searchEvents( request: SessionEventSearchRequest, exec?: SessionSearchE
 
 /**
  * List the complete logical corpus using live-preferred records.
+ * @param signal - optional cancellation for persistence listing.
  * @returns deterministic newest-first cloned session records.
  */
-listSessions(): Promise<SessionRecord[]>
+listSessions(signal?: AbortSignal): Promise<SessionRecord[]>
 
 /**
  * Read and replay-validate one complete logical session log without making it live.
@@ -1009,9 +1010,10 @@ async readSession(sessionId: SessionId): Promise<SessionLogSnapshot>
 /**
  * Filter the complete logical corpus with provider-independent predicates.
  * @param filters - ANDed session metadata and availability clauses.
+ * @param signal - optional cancellation for persistence listing.
  * @returns matching cloned records in deterministic newest-first order.
  */
-async filterSessions(filters: readonly SessionResultFilter[]): Promise<SessionRecord[]>
+async filterSessions( filters: readonly SessionResultFilter[], signal?: AbortSignal, ): Promise<SessionRecord[]>
 
 /**
  * Fold the latest log-backed title from one live-preferred logical session.

packages/cordis/tool-cordis/src/api-catalog.ts

@@ -495,16 +495,16 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
         jsDoc: '/**\n * Search events within one live-preferred logical session.\n * @param request - target session, query text, filters, page size, and cursor.\n * @param exec - optional cancellation control.\n * @returns matching event hits and their target header from one indexed generation.\n */',
       },
       {
-        signature: 'listSessions(): Promise<SessionRecord[]>',
+        signature: 'listSessions(signal?: AbortSignal): Promise<SessionRecord[]>',
-        jsDoc: '/**\n * List the complete logical corpus using live-preferred records.\n * @returns deterministic newest-first cloned session records.\n */',
+        jsDoc: '/**\n * List the complete logical corpus using live-preferred records.\n * @param signal - optional cancellation for persistence listing.\n * @returns deterministic newest-first cloned session records.\n */',
       },
       {
         signature: 'async readSession(sessionId: SessionId): Promise<SessionLogSnapshot>',
         jsDoc: '/**\n * Read and replay-validate one complete logical session log without making it live.\n * @param sessionId - live or persisted session id to read.\n * @returns cloned header and complete raw event log from one observation.\n * @throws when persistence, header compatibility, or replay validation fails.\n */',
       },
       {
-        signature: 'async filterSessions(filters: readonly SessionResultFilter[]): Promise<SessionRecord[]>',
+        signature: 'async filterSessions( filters: readonly SessionResultFilter[], signal?: AbortSignal, ): Promise<SessionRecord[]>',
-        jsDoc: '/**\n * Filter the complete logical corpus with provider-independent predicates.\n * @param filters - ANDed session metadata and availability clauses.\n * @returns matching cloned records in deterministic newest-first order.\n */',
+        jsDoc: '/**\n * Filter the complete logical corpus with provider-independent predicates.\n * @param filters - ANDed session metadata and availability clauses.\n * @param signal - optional cancellation for persistence listing.\n * @returns matching cloned records in deterministic newest-first order.\n */',
       },
       {
         signature: 'async readTitle( sessionId: SessionId, signal?: AbortSignal, ): Promise<SessionTitleSnapshot | undefined>',

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

@@ -4,9 +4,9 @@
 
 ## Reads
 
-- `listSessions()` reads current persistence metadata, merges live records with live precedence, and returns cloned records in deterministic newest-first order.
+- `listSessions(signal?)` reads current persistence metadata, merges live records with live precedence, and returns cloned records in deterministic newest-first order.
 - `readSession(sessionId)` returns one complete detached raw log after the same core replay validation used by resume; it never enters the session into the live store.
-- `filterSessions(filters)` applies provider-independent session metadata and availability predicates to that same cloned logical corpus.
+- `filterSessions(filters, signal?)` applies provider-independent session metadata and availability predicates to that same cloned logical corpus.
 - `filterEvents(sessionId, filters)` extracts first-party semantic documents and applies provider-independent metadata and literal-text predicates in ascending seq order.
 - `readTitleSnapshots(sessionIds, signal?)` resolves unique ids from one live-preferred corpus observation, passes cancellation through persisted listing and inspection, and returns ordered per-session settlements so one missing or malformed title source does not discard its peers. Each live source is folded directly, and each persisted worker folds to a detached header/title result and releases the full log before dequeuing another id. Cancellation rejects the whole batch. `readTitleSnapshot(sessionId, signal?)` is the one-observation view; `readTitle(sessionId, signal?)` returns only its optional folded `session/title`.
 - `listEvents(sessionId)` loads the live-preferred raw log and classifies each event as `current`, `shadowed`, or `log-only` with the shared `dsh-session` surface fold.

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

@@ -52,11 +52,14 @@ export class SessionCorpus {
 
   /**
    * List the complete logical corpus with live precedence and cloned headers.
+   * @param signal - optional cancellation for persistence listing.
    * @returns records in deterministic newest-first order.
    */
-  async listSessions(): Promise<SessionRecord[]> {
+  async listSessions(signal?: AbortSignal): Promise<SessionRecord[]> {
+    signal?.throwIfAborted()
     const persistence = this._persistence
-    const persisted = persistence === undefined ? [] : await listPersisted(persistence)
+    const persisted = persistence === undefined ? [] : await listPersisted(persistence, signal)
+    signal?.throwIfAborted()
     const records = new Map<SessionId, SessionRecord>()
     for (const header of persisted) {
       records.set(header.id, { header: structuredClone(header), live: false, persisted: true })
@@ -239,6 +242,7 @@ async function listPersisted(
   try {
     return await persistence.list(signal)
   } catch (error: unknown) {
+    if (signal?.aborted) signal.throwIfAborted()
     throw new SessionQueryError(
       `session persistence listing failed: ${errorMessage(error)}`,
       'SESSION_QUERY_PERSISTENCE_FAILED',

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

@@ -115,10 +115,11 @@ export abstract class SessionQueryService extends Service {
 
   /**
    * List the complete logical corpus using live-preferred records.
+   * @param signal - optional cancellation for persistence listing.
    * @returns deterministic newest-first cloned session records.
    */
-  listSessions(): Promise<SessionRecord[]> {
+  listSessions(signal?: AbortSignal): Promise<SessionRecord[]> {
-    return this._corpus.listSessions()
+    return this._corpus.listSessions(signal)
   }
 
   /**
@@ -139,11 +140,15 @@ export abstract class SessionQueryService extends Service {
   /**
    * Filter the complete logical corpus with provider-independent predicates.
    * @param filters - ANDed session metadata and availability clauses.
+   * @param signal - optional cancellation for persistence listing.
    * @returns matching cloned records in deterministic newest-first order.
    */
-  async filterSessions(filters: readonly SessionResultFilter[]): Promise<SessionRecord[]> {
+  async filterSessions(
+    filters: readonly SessionResultFilter[],
+    signal?: AbortSignal,
+  ): Promise<SessionRecord[]> {
     const ownedFilters = materializeSessionResultFilters(filters)
-    return this._filterSessions(ownedFilters)
+    return this._filterSessions(ownedFilters, signal)
   }
 
   /**
@@ -220,8 +225,11 @@ export abstract class SessionQueryService extends Service {
     return this._filterEvents(sessionId, ownedFilters)
   }
 
-  private async _filterSessions(filters: readonly SessionResultFilter[]): Promise<SessionRecord[]> {
+  private async _filterSessions(
-    return filterSessionResults(await this._corpus.listSessions(), filters)
+    filters: readonly SessionResultFilter[],
+    signal?: AbortSignal,
+  ): Promise<SessionRecord[]> {
+    return filterSessionResults(await this._corpus.listSessions(signal), filters)
   }
 
   private async _filterEvents(

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

@@ -130,6 +130,98 @@ function rejectUnknown<T>(reason: unknown): Promise<T> {
   })
 }
 
+const cancellableSessionListings = [
+  {
+    name: 'listSessions',
+    run: (ctx: Context, signal: AbortSignal) => ctx.sessionQuery.listSessions(signal),
+  },
+  {
+    name: 'filterSessions',
+    run: (ctx: Context, signal: AbortSignal) => ctx.sessionQuery.filterSessions([], signal),
+  },
+] as const
+
+describe.each(cancellableSessionListings)('$name cancellation', ({ run }) => {
+  it('preserves an exact pre-abort reason without entering persistence', async () => {
+    TestPersistence.reset()
+    const ctx = await liveContext()
+    await ctx.plugin(TestPersistence)
+    const controller = new AbortController()
+    const reason = new Error('session listing cancelled before start')
+    controller.abort(reason)
+
+    await expect(run(ctx, controller.signal)).rejects.toBe(reason)
+    expect(TestPersistence.listCalls).toBe(0)
+    expect(TestPersistence.listSignals).toEqual([])
+  })
+
+  it('forwards in-flight cancellation and waits for persistence cleanup before rejecting', async () => {
+    TestPersistence.reset()
+    const ctx = await liveContext()
+    await ctx.plugin(TestPersistence)
+    const controller = new AbortController()
+    const reason = new Error('session listing cancelled in flight')
+    const started = Promise.withResolvers<undefined>()
+    const abortObserved = Promise.withResolvers<undefined>()
+    const cleanup = Promise.withResolvers<undefined>()
+    let active = false
+    TestPersistence.listOverride = async (signal) => {
+      if (signal === undefined) throw new Error('expected persistence listing signal')
+      active = true
+      const aborted = new Promise<void>((resolve) => {
+        signal.addEventListener('abort', () => { resolve() }, { once: true })
+      })
+      started.resolve(undefined)
+      await aborted
+      abortObserved.resolve(undefined)
+      await cleanup.promise
+      active = false
+      signal.throwIfAborted()
+      return []
+    }
+
+    const pending = run(ctx, controller.signal)
+    let settled = false
+    void pending.then(
+      () => { settled = true },
+      () => { settled = true },
+    )
+    await started.promise
+    controller.abort(reason)
+    await abortObserved.promise
+
+    expect(settled).toBe(false)
+    expect(active).toBe(true)
+    expect(TestPersistence.listSignals).toEqual([controller.signal])
+
+    cleanup.resolve(undefined)
+    await expect(pending).rejects.toBe(reason)
+    expect(active).toBe(false)
+  })
+
+  it('preserves cancellation after a persistence implementation ignores the signal', async () => {
+    TestPersistence.reset()
+    const ctx = await liveContext()
+    await ctx.plugin(TestPersistence)
+    const controller = new AbortController()
+    const reason = new Error('session listing cancelled before persistence returned')
+    const started = Promise.withResolvers<undefined>()
+    const listing = Promise.withResolvers<SessionHeader[]>()
+    TestPersistence.listOverride = (_signal) => {
+      started.resolve(undefined)
+      return listing.promise
+    }
+
+    const pending = run(ctx, controller.signal)
+    await started.promise
+    controller.abort(reason)
+    listing.resolve([])
+
+    await expect(pending).rejects.toBe(reason)
+    expect(TestPersistence.listSignals).toEqual([controller.signal])
+  })
+})
+
 describe('session-query exact reads', () => {
   it('returns a detached replay-valid full log and rejects a corrupt persisted seed', async () => {
     const valid = header('valid-log', 2)

packages/session-query/tool-session-query/package.json

@@ -51,6 +51,7 @@
     "@deepseek-ai/dsh-session-title": "workspace:^",
     "@deepseek-ai/dsh-system-prompt": "workspace:^",
     "@deepseek-ai/dsh-timeout": "workspace:^",
+    "@deepseek-ai/dsh-timeout-policy": "workspace:^",
     "@deepseek-ai/dsh-tools": "workspace:^",
     "cordis": "^4.0.0-rc.7"
   }

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

@@ -312,7 +312,7 @@ async function authorizeTarget(
   const records = await ctx.sessionQuery.filterSessions([
     { kind: 'id', values: [target] },
     { kind: 'cwd', values: [cwd] },
-  ])
+  ], signal)
   signal.throwIfAborted()
   if (records.length !== 1) throw unauthorizedTarget()
 }
@@ -805,7 +805,7 @@ async function authorizeSessionIds(
   const records = await ctx.sessionQuery.filterSessions([
     { kind: 'id', values: other },
     { kind: 'cwd', values: [cwd] },
-  ])
+  ], signal)
   signal.throwIfAborted()
   for (const record of records) authorized.add(record.header.id)
   return authorized

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

@@ -2,7 +2,8 @@ import { afterEach, describe, expect, it, vi } from 'vitest'
 import { Context, type Fiber } from 'cordis'
 import type { Agent } from '@deepseek-ai/dsh-agent'
 import { CallId, HarnessError } from '@deepseek-ai/dsh-llm'
-import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
+import { MAX_TIMER_DELAY_MS, TimeoutReason } from '@deepseek-ai/dsh-timeout'
+import * as TimeoutPolicy from '@deepseek-ai/dsh-timeout-policy'
 import SessionStore, {
   SESSION_FORMAT_VERSION,
   SessionId,
@@ -30,6 +31,7 @@ import * as ToolSessionQuery from '@deepseek-ai/dsh-tool-session-query'
 const activeContexts: Context[] = []
 
 afterEach(async () => {
+  vi.useRealTimers()
   vi.restoreAllMocks()
   for (const ctx of activeContexts.splice(0)) await ctx.fiber.dispose()
   FakeQuery.reset()
@@ -194,12 +196,14 @@ interface Mounted {
 async function mount(
   config: ToolSessionQuery.Config = {},
   callerCwd: string | null = '/work',
+  enforceTimeout = false,
 ): Promise<Mounted> {
   const ctx = new Context()
   activeContexts.push(ctx)
   await ctx.plugin(SessionStore)
   await ctx.plugin(SystemPrompt)
   await ctx.plugin(ToolRegistry)
+  if (enforceTimeout) await ctx.plugin(TimeoutPolicy)
   await ctx.plugin(FakeQuery)
   const fiber = await ctx.plugin(ToolSessionQuery, config)
   const caller = createSession(ctx, 'caller', callerCwd ?? undefined, 10)
@@ -1145,6 +1149,123 @@ describe('search paging, prior-history bounds, titles, and cancellation', () =>
     expect(text(result)).not.toContain('title unavailable')
   })
 
+  it('forwards caller cancellation into direct-target authorization and waits for cleanup', async () => {
+    const mounted = await mount()
+    const target = createSession(mounted.ctx, 'stalled-direct-authorization', '/work')
+    const controller = new AbortController()
+    const cancellation = new SessionQueryError(
+      'direct-target authorization cancelled',
+      'SESSION_QUERY_ABORTED',
+    )
+    const started = Promise.withResolvers<undefined>()
+    const abortObserved = Promise.withResolvers<undefined>()
+    const cleanup = Promise.withResolvers<undefined>()
+    let active = false
+    const filterSessions = vi.spyOn(mounted.ctx.sessionQuery, 'filterSessions')
+      .mockImplementation(async (_filters, signal) => {
+        if (signal === undefined) throw new Error('expected authorization signal')
+        active = true
+        const aborted = new Promise<void>((resolve) => {
+          signal.addEventListener('abort', () => { resolve() }, { once: true })
+        })
+        started.resolve(undefined)
+        await aborted
+        abortObserved.resolve(undefined)
+        await cleanup.promise
+        active = false
+        signal.throwIfAborted()
+        return []
+      })
+
+    const pending = mounted.call(
+      'session_event_search',
+      { session_id: target.id, query: 'needle' },
+      { signal: controller.signal },
+    )
+    let settled = false
+    void pending.then(
+      () => { settled = true },
+      () => { settled = true },
+    )
+    await started.promise
+    controller.abort(cancellation)
+    await abortObserved.promise
+
+    expect(settled).toBe(false)
+    expect(active).toBe(true)
+    expect(filterSessions.mock.calls[0]?.[1]).toBe(controller.signal)
+    expect(controller.signal.reason).toBe(cancellation)
+    expect(FakeQuery.eventRequests).toEqual([])
+
+    cleanup.resolve(undefined)
+    const result = await pending
+    expect(active).toBe(false)
+    expect(errorCode(result)).toBe('SESSION_QUERY_ABORTED')
+    expect(text(result)).toBe('Error: direct-target authorization cancelled')
+    expect(FakeQuery.eventRequests).toEqual([])
+  })
+
+  it('forwards the search deadline into parent authorization and times out only after cleanup', async () => {
+    vi.useFakeTimers()
+    const timeoutMs = 1_234
+    const mounted = await mount({ searchTimeoutMs: timeoutMs }, '/work', true)
+    const parent = createSession(mounted.ctx, 'stalled-parent-authorization', '/work')
+    FakeQuery.sessionSearch = () => Promise.resolve({
+      items: [sessionHit('authorized-child', '/work', 'needle', parent.id)],
+    })
+    const upstream = new AbortController()
+    const started = Promise.withResolvers<undefined>()
+    const abortObserved = Promise.withResolvers<undefined>()
+    const cleanup = Promise.withResolvers<undefined>()
+    let active = false
+    let deadlineSignal: AbortSignal | undefined
+    const filterSessions = vi.spyOn(mounted.ctx.sessionQuery, 'filterSessions')
+      .mockImplementation(async (_filters, signal) => {
+        if (signal === undefined) throw new Error('expected authorization signal')
+        deadlineSignal = signal
+        active = true
+        const aborted = new Promise<void>((resolve) => {
+          signal.addEventListener('abort', () => { resolve() }, { once: true })
+        })
+        started.resolve(undefined)
+        await aborted
+        abortObserved.resolve(undefined)
+        await cleanup.promise
+        active = false
+        signal.throwIfAborted()
+        return []
+      })
+
+    const pending = mounted.call(
+      'session_search',
+      { query: 'needle' },
+      { signal: upstream.signal },
+    )
+    let settled = false
+    void pending.then(
+      () => { settled = true },
+      () => { settled = true },
+    )
+    await started.promise
+    await vi.advanceTimersByTimeAsync(timeoutMs)
+    await abortObserved.promise
+
+    expect(settled).toBe(false)
+    expect(active).toBe(true)
+    expect(deadlineSignal).toBeDefined()
+    expect(deadlineSignal).not.toBe(upstream.signal)
+    expect(filterSessions.mock.calls[0]?.[1]).toBe(deadlineSignal)
+    expect(FakeQuery.searchSignals).toEqual([deadlineSignal])
+    expect(deadlineSignal?.reason).toBeInstanceOf(TimeoutReason)
+    expect(deadlineSignal?.reason).toMatchObject({ code: 'TOOL_TIMEOUT', timeoutMs })
+
+    cleanup.resolve(undefined)
+    const result = await pending
+    expect(active).toBe(false)
+    expect(errorCode(result)).toBe('TOOL_TIMEOUT')
+    expect(text(result)).toBe(`Error: tool call timed out after ${timeoutMs}ms`)
+  })
+
   it('passes the exact execution signal to every FTS page and stops on cancellation', async () => {
     const mounted = await mount()
     const controller = new AbortController()

pnpm-lock.yaml

@@ -2954,6 +2954,9 @@ importers:
       '@deepseek-ai/dsh-timeout':
         specifier: workspace:^
         version: link:../../util/timeout
+      '@deepseek-ai/dsh-timeout-policy':
+        specifier: workspace:^
+        version: link:../../timeout/timeout-policy
       '@deepseek-ai/dsh-tools':
         specifier: workspace:^
         version: link:../../core/tools