fix: cancel exact session observations

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

@@ -11,11 +11,11 @@
 - `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.
 - `readSurface(sessionId)` returns one cloned header, raw-log capture boundary, and the complete folded current surface in model-history order. A live session wins over persistence; compaction is observed before or after its replacement append, never as a synthetic mixture.
-- `readEvent(request)` returns a cloned header, the full target event, and a bounded raw-seq window. `before` and `after` default to zero and may not exceed `readWindowMax`.
-- `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.
+- `readEvent(request, signal?)` returns a cloned header, the full target event, and a bounded raw-seq window. `before` and `after` default to zero and may not exceed `readWindowMax`.
+- `traceSession(sessionId, signal?)` 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, signal?)` 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 `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.
+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 read, event trace, or event read 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. Lineage-trace cancellation is passed to persisted listing; event-trace and event-read cancellation is passed to persisted listing and inspection. Each waits for the started backend call to settle, then rejects with the signal's exact reason even when the backend ignored that signal. A pre-aborted known-live title read, event trace, or event read rejects before folding or snapshotting without consulting persistence. 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
 

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

@@ -82,23 +82,37 @@ export class SessionCorpus {
    * A known live target never consults persistence, so an optional backend's
    * failure cannot make current in-memory history unreadable.
    * @param sessionId - session to resolve.
+   * @param signal - optional cancellation for persisted source resolution.
    * @returns detached live-preferred header and events.
    */
-  async load(sessionId: SessionId): Promise<LogicalSession> {
+  async load(sessionId: SessionId, signal?: AbortSignal): Promise<LogicalSession> {
+    signal?.throwIfAborted()
     const live = this._ctx.sessions.get(sessionId)
-    if (live !== undefined) return snapshotLive(live)
+    if (live !== undefined) {
+      const snapshot = snapshotLive(live)
+      signal?.throwIfAborted()
+      return snapshot
+    }
     const persistence = this._persistence
     if (persistence === undefined) throw notFound(sessionId)
-    const listed = (await listPersisted(persistence)).find(header => header.id === sessionId)
+    const listed = (await listPersisted(persistence, signal)).find(header => header.id === sessionId)
+    signal?.throwIfAborted()
     if (listed === undefined) throw notFound(sessionId)
-    const loaded = await inspectPersisted(persistence, sessionId)
+    const loaded = await inspectPersisted(persistence, sessionId, signal)
+    signal?.throwIfAborted()
     const attached = this._ctx.sessions.get(sessionId)
-    if (attached !== undefined) return snapshotLive(attached)
+    if (attached !== undefined) {
+      const snapshot = snapshotLive(attached)
+      signal?.throwIfAborted()
+      return snapshot
+    }
     assertSessionHeadersCompatible(loaded.meta, listed)
-    return {
+    const snapshot = {
       header: structuredClone(loaded.meta),
       events: loaded.events.map(event => structuredClone(event)),
     }
+    signal?.throwIfAborted()
+    return snapshot
   }
 
   /**

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

@@ -272,22 +272,26 @@ export abstract class SessionQueryService extends Service {
   /**
    * Trace known ancestry and descendants from one corpus observation.
    * @param sessionId - logical session id to trace.
+   * @param signal - optional cancellation for persistence listing.
    * @returns a complete lineage or an explicit unresolved parent boundary.
    * @throws when corpus resolution fails, the target is absent, or its known ancestry cycles.
    */
-  async traceSession(sessionId: SessionId): Promise<SessionLineageTrace> {
-    const records = await this._corpus.listSessions()
+  async traceSession(sessionId: SessionId, signal?: AbortSignal): Promise<SessionLineageTrace> {
+    const records = await this._corpus.listSessions(signal)
+    signal?.throwIfAborted()
     return tracing.traceSession(records, sessionId)
   }
 
   /**
    * Trace one event's direct positional and provenance relationships.
    * @param request - target session id and event seq.
+   * @param signal - optional cancellation for persisted source resolution.
    * @returns source header, direct links, and the target's positional replacement chain.
    * @throws when source resolution fails, the target is absent, or surface/provenance validation fails.
    */
-  async traceEvent(request: SessionEventTraceRequest): Promise<SessionEventTraceObservation> {
-    const loaded = await this._corpus.load(request.sessionId)
+  async traceEvent(request: SessionEventTraceRequest, signal?: AbortSignal): Promise<SessionEventTraceObservation> {
+    const loaded = await this._corpus.load(request.sessionId, signal)
+    signal?.throwIfAborted()
     return {
       session: loaded.header,
       ...tracing.traceEvent(request.sessionId, loaded.events, request.seq),
@@ -297,14 +301,15 @@ export abstract class SessionQueryService extends Service {
   /**
    * Read one full event plus a bounded raw-log context window.
    * @param request - target session/seq and context sizes.
+   * @param signal - optional cancellation for persisted source resolution.
    * @returns cloned target and neighboring events.
    */
-  async readEvent(request: SessionEventReadRequest): Promise<SessionEventWindow> {
+  async readEvent(request: SessionEventReadRequest, signal?: AbortSignal): Promise<SessionEventWindow> {
     const before = this._readWindow('before', request.before)
     const after = this._readWindow('after', request.after)
     const sessionId = request.sessionId
     const seq = request.seq
-    return this._readEvent(sessionId, seq, before, after)
+    return this._readEvent(sessionId, seq, before, after, signal)
   }
 
   private async _readEvent(
@@ -312,8 +317,10 @@ export abstract class SessionQueryService extends Service {
     seq: number,
     before: number,
     after: number,
+    signal?: AbortSignal,
   ): Promise<SessionEventWindow> {
-    const loaded = await this._corpus.load(sessionId)
+    const loaded = await this._corpus.load(sessionId, signal)
+    signal?.throwIfAborted()
     const target = loaded.events[seq]
     if (target === undefined || target.seq !== seq) {
       throw new SessionQueryError(

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

@@ -142,6 +142,34 @@ const cancellableSessionListings = [
   },
 ] as const
 
+interface CancellableExactRead {
+  readonly name: 'traceSession' | 'traceEvent' | 'readEvent'
+  readonly inspects: boolean
+  readonly run: (
+    ctx: Context,
+    sessionId: SessionIdType,
+    signal: AbortSignal,
+  ) => Promise<unknown>
+}
+
+const cancellableExactReads: readonly CancellableExactRead[] = [
+  {
+    name: 'traceSession',
+    inspects: false,
+    run: (ctx, sessionId, signal) => ctx.sessionQuery.traceSession(sessionId, signal),
+  },
+  {
+    name: 'traceEvent',
+    inspects: true,
+    run: (ctx, sessionId, signal) => ctx.sessionQuery.traceEvent({ sessionId, seq: 0 }, signal),
+  },
+  {
+    name: 'readEvent',
+    inspects: true,
+    run: (ctx, sessionId, signal) => ctx.sessionQuery.readEvent({ sessionId, seq: 0 }, signal),
+  },
+] as const
+
 describe.each(cancellableSessionListings)('$name cancellation', ({ run }) => {
   it('preserves an exact pre-abort reason without entering persistence', async () => {
     TestPersistence.reset()
@@ -223,6 +251,167 @@ describe.each(cancellableSessionListings)('$name cancellation', ({ run }) => {
   })
 })
 
+describe.each(cancellableExactReads)('$name cancellation', ({ inspects, run }) => {
+  it('preserves an exact pre-abort reason without entering persistence', async () => {
+    const persisted = header('pre-aborted-exact-read')
+    TestPersistence.reset([{ meta: persisted, events: eventLog() }])
+    const ctx = await liveContext()
+    await ctx.plugin(TestPersistence)
+    const controller = new AbortController()
+    const reason = new Error('exact read cancelled before start')
+    controller.abort(reason)
+
+    await expect(run(ctx, persisted.id, controller.signal)).rejects.toBe(reason)
+    expect(TestPersistence.listCalls).toBe(0)
+    expect(TestPersistence.inspectCalls).toEqual([])
+  })
+
+  it('forwards in-flight list cancellation and waits for cleanup before rejecting', async () => {
+    const persisted = header('cancelled-exact-list')
+    TestPersistence.reset([{ meta: persisted, events: eventLog() }])
+    const ctx = await liveContext()
+    await ctx.plugin(TestPersistence)
+    const controller = new AbortController()
+    const reason = new Error('exact read list 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 exact-read 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, persisted.id, 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])
+    expect(TestPersistence.inspectCalls).toEqual([])
+
+    cleanup.resolve(undefined)
+    await expect(pending).rejects.toBe(reason)
+    expect(active).toBe(false)
+  })
+
+  it('waits for an ignoring backend to return before preserving the abort reason', async () => {
+    const persisted = header('ignored-exact-signal')
+    const entry = { meta: persisted, events: eventLog() }
+    TestPersistence.reset([entry])
+    const ctx = await liveContext()
+    await ctx.plugin(TestPersistence)
+    const controller = new AbortController()
+    const reason = new Error('exact read cancelled while backend ignored signal')
+    const started = Promise.withResolvers<undefined>()
+    const release = Promise.withResolvers<undefined>()
+    let active = false
+    if (inspects) {
+      TestPersistence.inspectOverride = async () => {
+        active = true
+        started.resolve(undefined)
+        await release.promise
+        active = false
+        return structuredClone(entry)
+      }
+    } else {
+      TestPersistence.listOverride = async () => {
+        active = true
+        started.resolve(undefined)
+        await release.promise
+        active = false
+        return [structuredClone(persisted)]
+      }
+    }
+
+    const pending = run(ctx, persisted.id, controller.signal)
+    let settled = false
+    void pending.then(
+      () => { settled = true },
+      () => { settled = true },
+    )
+    await started.promise
+    controller.abort(reason)
+
+    expect(settled).toBe(false)
+    expect(active).toBe(true)
+    expect(TestPersistence.listSignals).toEqual([controller.signal])
+    expect(TestPersistence.inspectSignals).toEqual(inspects ? [controller.signal] : [])
+
+    release.resolve(undefined)
+    await expect(pending).rejects.toBe(reason)
+    expect(active).toBe(false)
+  })
+})
+
+describe.each(cancellableExactReads.filter(read => read.inspects))(
+  '$name persisted inspection cancellation',
+  ({ run }) => {
+    it('forwards cancellation and waits for inspection cleanup before rejecting', async () => {
+      const persisted = header('cancelled-exact-inspect')
+      TestPersistence.reset([{ meta: persisted, events: eventLog() }])
+      const ctx = await liveContext()
+      await ctx.plugin(TestPersistence)
+      const controller = new AbortController()
+      const reason = new Error('exact read inspection cancelled in flight')
+      const started = Promise.withResolvers<undefined>()
+      const abortObserved = Promise.withResolvers<undefined>()
+      const cleanup = Promise.withResolvers<undefined>()
+      let active = false
+      TestPersistence.inspectOverride = async (_sessionId, signal) => {
+        if (signal === undefined) throw new Error('expected exact-read inspection 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()
+        throw new Error('unreachable after exact-read cancellation')
+      }
+
+      const pending = run(ctx, persisted.id, 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])
+      expect(TestPersistence.inspectSignals).toEqual([controller.signal])
+
+      cleanup.resolve(undefined)
+      await expect(pending).rejects.toBe(reason)
+      expect(active).toBe(false)
+    })
+  },
+)
+
 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)
@@ -862,9 +1051,15 @@ describe('session-query exact reads', () => {
     await ctx.plugin(TestPersistence)
     TestPersistence.listFailure = new Error('list unavailable')
     TestPersistence.inspectFailure = new Error('inspect unavailable')
+    const signal = new AbortController().signal
 
     await expect(ctx.sessionQuery.listEvents(live.id)).resolves.toHaveLength(2)
-    await expect(ctx.sessionQuery.readEvent({ sessionId: live.id, seq: 1 })).resolves.toMatchObject({ target: { seq: 1 } })
+    await expect(ctx.sessionQuery.traceEvent({ sessionId: live.id, seq: 1 }, signal))
+      .resolves.toMatchObject({ session: { id: live.id }, target: { seq: 1 } })
+    await expect(ctx.sessionQuery.readEvent({ sessionId: live.id, seq: 1 }, signal))
+      .resolves.toMatchObject({ target: { seq: 1 } })
+    expect(TestPersistence.listSignals).toEqual([])
+    expect(TestPersistence.inspectSignals).toEqual([])
     await expect(ctx.sessionQuery.listSessions()).rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
     await expect(ctx.sessionQuery.listEvents(SessionId('durable'))).rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
   })

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. 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.
+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. Every exact executor passes its unchanged execution signal through authorization and the service trace/read, so cancellation waits for cooperative persistence cleanup and retains the signal's exact reason. 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

@@ -428,7 +428,7 @@ async function executeSessionTrace(
   await authorizeTarget(ctx, caller, sessionId, exec.signal)
   let trace: SessionLineageTrace
   try {
-    trace = await ctx.sessionQuery.traceSession(sessionId)
+    trace = await ctx.sessionQuery.traceSession(sessionId, exec.signal)
   } catch (error: unknown) {
     exec.signal.throwIfAborted()
     if (error instanceof SessionQueryError && error.code === 'SESSION_QUERY_INVALID_LINEAGE') {
@@ -471,7 +471,7 @@ async function executeEventTrace(
   const caller = callerOf(exec)
   const sessionId = targetId(args, caller)
   await authorizeTarget(ctx, caller, sessionId, exec.signal)
-  const trace = await ctx.sessionQuery.traceEvent({ sessionId, seq: args.seq })
+  const trace = await ctx.sessionQuery.traceEvent({ sessionId, seq: args.seq }, exec.signal)
   exec.signal.throwIfAborted()
   assertObservedTargetAuthorized(caller, sessionId, trace.session)
   const title = await readTitle(ctx, caller, sessionId, exec.signal)
@@ -494,7 +494,7 @@ async function executeEventRead(
     seq: args.seq,
     ...args.before === undefined ? {} : { before: args.before },
     ...args.after === undefined ? {} : { after: args.after },
-  })
+  }, exec.signal)
   exec.signal.throwIfAborted()
   assertObservedTargetAuthorized(caller, sessionId, window.session)
   const title = await readTitle(ctx, caller, sessionId, exec.signal)

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

@@ -661,6 +661,78 @@ describe('workspace authority and lineage redaction', () => {
     expect(text(result)).toBe(`Error: ${message}`)
   })
 
+  it.each([
+    'session_trace',
+    'session_event_trace',
+    'session_event_read',
+  ] as const)('forwards the exact signal to %s and waits for service cleanup', async (toolName) => {
+    const mounted = await mount()
+    const target = createSession(mounted.ctx, `cancelled-${toolName}`, '/work')
+    target.append(
+      'user/message',
+      { content: [{ type: 'text', text: 'pending exact read' }], source: { kind: 'user' } },
+      { surfaceOp: 'append' },
+    )
+    const controller = new AbortController()
+    const cancellation = new SessionQueryError(
+      `${toolName} cancelled`,
+      'SESSION_QUERY_ABORTED',
+    )
+    const started = Promise.withResolvers<undefined>()
+    const abortObserved = Promise.withResolvers<undefined>()
+    const cleanup = Promise.withResolvers<undefined>()
+    let observedSignal: AbortSignal | undefined
+    let active = false
+    const holdExactRead = async (signal?: AbortSignal): Promise<never> => {
+      if (signal === undefined) throw new Error('expected exact tool execution signal')
+      observedSignal = 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()
+      throw new Error('unreachable after exact tool cancellation')
+    }
+    if (toolName === 'session_trace') {
+      vi.spyOn(mounted.ctx.sessionQuery, 'traceSession')
+        .mockImplementation((_sessionId, signal) => holdExactRead(signal))
+    } else if (toolName === 'session_event_trace') {
+      vi.spyOn(mounted.ctx.sessionQuery, 'traceEvent')
+        .mockImplementation((_request, signal) => holdExactRead(signal))
+    } else {
+      vi.spyOn(mounted.ctx.sessionQuery, 'readEvent')
+        .mockImplementation((_request, signal) => holdExactRead(signal))
+    }
+    const args = toolName === 'session_trace'
+      ? { session_id: target.id }
+      : { session_id: target.id, seq: 0 }
+
+    const pending = mounted.call(toolName, args, { 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(observedSignal).toBe(controller.signal)
+
+    cleanup.resolve(undefined)
+    const result = await pending
+    expect(active).toBe(false)
+    expect(errorCode(result)).toBe('SESSION_QUERY_ABORTED')
+    expect(text(result)).toBe(`Error: ${toolName} cancelled`)
+  })
+
   it('preserves caller cancellation while a lineage trace is pending', async () => {
     const mounted = await mount()
     const target = createSession(mounted.ctx, 'cancelled-trace-target', '/work')