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

fix(session-query): checkpoint review round 3

Browse Source
This commit is contained in:
Hypatia May
2026-07-11 10:19:34 +08:00
parent 18028cad4f
commit 352ea6cf4f
15 changed files with 153 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
packages/cordis/tool-cordis/src/api-catalog.ts
View File

@@ -737,6 +737,10 @@ export const TYPE_API: readonly TypeApiEntry[] = [
name: 'SessionEventSearchRequest',
declaration: 'export interface SessionEventSearchRequest extends SessionSearchPageRequest {\n sessionId: SessionId;\n query: string;\n filters?: readonly SessionEventResultFilter[];\n}',
},
{
name: 'SessionEventSearchSpec',
declaration: 'export interface SessionEventSearchSpec extends SessionEventSearchRequest {\n limit: number;\n}',
},
{
name: 'SessionEventSurface',
declaration: 'export type SessionEventSurface = \'current\' | \'shadowed\' | \'log-only\';',
@@ -819,7 +823,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [
},
{
name: 'SessionSearchProvider',
declaration: 'export interface SessionSearchProvider {\n readonly id: string;\n status(): SessionSearchProviderStatus;\n persistedInventory(): Promise<readonly SessionPersistedIndexEntry[]>;\n setPersistedActive(active: boolean): Promise<void>;\n replacePersisted(snapshot: SessionIndexSnapshot): Promise<void>;\n removePersisted(sessionId: SessionId): Promise<void>;\n replaceLive(snapshot: SessionIndexSnapshot): Promise<void>;\n removeLive(sessionId: SessionId): Promise<void>;\n searchSessions(request: SessionSearchRequest, exec?: SessionQueryExecContext): Promise<SessionSearchPage<SessionSearchHit>>;\n searchEvents(request: SessionEventSearchRequest, exec?: SessionQueryExecContext): Promise<SessionSearchPage<SessionEventSearchHit>>;\n}',
declaration: 'export interface SessionSearchProvider {\n readonly id: string;\n status(): SessionSearchProviderStatus;\n persistedInventory(): Promise<readonly SessionPersistedIndexEntry[]>;\n setPersistedActive(active: boolean): Promise<void>;\n replacePersisted(snapshot: SessionIndexSnapshot): Promise<void>;\n removePersisted(sessionId: SessionId): Promise<void>;\n replaceLive(snapshot: SessionIndexSnapshot): Promise<void>;\n removeLive(sessionId: SessionId): Promise<void>;\n searchSessions(request: SessionSearchSpec, exec?: SessionQueryExecContext): Promise<SessionSearchPage<SessionSearchHit>>;\n searchEvents(request: SessionEventSearchSpec, exec?: SessionQueryExecContext): Promise<SessionSearchPage<SessionEventSearchHit>>;\n}',
},
{
name: 'SessionSearchProviderStatus',
@@ -829,6 +833,10 @@ export const TYPE_API: readonly TypeApiEntry[] = [
name: 'SessionSearchRequest',
declaration: 'export interface SessionSearchRequest extends SessionSearchPageRequest {\n query: string;\n sessionFilters?: readonly SessionResultFilter[];\n eventFilters?: readonly SessionEventResultFilter[];\n}',
},
{
name: 'SessionSearchSpec',
declaration: 'export interface SessionSearchSpec extends SessionSearchRequest {\n limit: number;\n}',
},
{
name: 'StreamChunk',
declaration: 'export type StreamChunk = {\n type: \'block-start\';\n index: number;\n blockType: ContentBlockType;\n} | {\n type: \'text-delta\';\n index: number;\n text: string;\n} | {\n type: \'reasoning-delta\';\n index: number;\n text: string;\n} | {\n type: \'tool-call-delta\';\n index: number;\n id: CallId;\n name?: string;\n argumentsDelta: string;\n} | {\n type: \'block-end\';\n index: number;\n block: ContentBlock;\n} | {\n type: \'usage\';\n usage: TokenUsage;\n} | {\n type: \'finish\';\n reason: FinishReason;\n};',

6
packages/core/session/tests/session.spec.ts
View File

@@ -374,12 +374,9 @@ describe('SessionStore', () => {
expect(observations).toHaveLength(1)
})
it('contains failing session/removed listeners without starving later observers', async () => {
it('contains rejected session/removed listeners during teardown', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const observed: SessionId[] = []
ctx.on('session/removed', () => { throw new Error('synchronous observer failed') })
ctx.on('session/removed', header => void observed.push(header.id))
ctx.on('session/removed', () => Promise.reject(new Error('observer failed')))
const session = ctx.sessions.prepare(SessionId('contained'))
const detach = ctx.sessions.enter(session)
@@ -388,7 +385,6 @@ describe('SessionStore', () => {
await Promise.resolve()
await Promise.resolve()
expect(ctx.sessions.get(session.id)).toBeUndefined()
expect(observed).toEqual([session.id])
})
it('rolls back the session (and onAppend) when a session/created listener throws (P1-1)', async () => {

3
packages/session-persistence/session-persistence/tests/coordinator-contract.ts
View File

@@ -128,12 +128,11 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
const fix = await makeFixture()
const { ctx, fiber } = await freshCtx(fix)
const observed: Array<{ headerId: SessionId; change: SessionPersistedChange }> = []
ctx.on('session/persisted', () => { throw new Error('synchronous derived read model failed') })
ctx.on('session/persisted', (header, change) => {
observed.push({ headerId: header.id, change: structuredClone(change) })
header.createdAt = -1
return Promise.reject(new Error('derived read model failed'))
})
ctx.on('session/persisted', () => Promise.reject(new Error('asynchronous derived read model failed')))
try {
const m = meta('notifications', WORK)
await ctx.sessionPersistence.create(m)

6
packages/session-query/session-query/README.md
View File

@@ -28,6 +28,12 @@ The service feeds providers two independent layers: a durable persisted base (`p
Persisted snapshots carry a SHA-256 fingerprint over canonical header/events plus the versions of relevant extractors. Reconciliation still loads and hashes canonical logs, but a provider replacement occurs only for a new or changed fingerprint; stale durable inventory entries are removed only while persistence is active and authoritative.
Providers receive resolved `SessionSearchSpec` and `SessionEventSearchSpec` values whose `limit` is required after service defaulting and validation. Public service callers use `SessionSearchRequest` and `SessionEventSearchRequest`, where `limit` remains optional.
## Errors
`SessionQueryError.code` is the closed `SessionQueryErrorCode` union: `SESSION_QUERY_ABORTED`, `SESSION_QUERY_DUPLICATE_EXTRACTOR`, `SESSION_QUERY_DUPLICATE_PROVIDER`, `SESSION_QUERY_EVENT_NOT_FOUND`, `SESSION_QUERY_INDEX_FAILED`, `SESSION_QUERY_INVALID_CONFIG`, `SESSION_QUERY_INVALID_EXTRACTOR`, `SESSION_QUERY_INVALID_FILTER`, `SESSION_QUERY_INVALID_LIMIT`, `SESSION_QUERY_INVALID_LINEAGE`, `SESSION_QUERY_INVALID_QUERY`, `SESSION_QUERY_INVALID_SURFACE`, `SESSION_QUERY_INVALID_WINDOW`, `SESSION_QUERY_PERSISTENCE_FAILED`, `SESSION_QUERY_PROVIDER_AMBIGUOUS`, `SESSION_QUERY_PROVIDER_CONFIGURED_MISSING`, `SESSION_QUERY_PROVIDER_CONFIGURED_UNAVAILABLE`, `SESSION_QUERY_PROVIDER_ERROR`, `SESSION_QUERY_PROVIDER_UNAVAILABLE`, `SESSION_QUERY_SESSION_NOT_FOUND`, and `SESSION_QUERY_SOURCE_CONFLICT`.
## Text extractors
Core extraction indexes semantic message text and reasoning, tool names/arguments/results, blocked prompts, context and steering, todos, and error/status detail. Stream chunks, request headers, and structural-only events contribute no document. Unknown event and content-block types contribute no text until their owner registers a versioned extractor with `registerEventTextExtractor()` or `registerContentTextExtractor()`.

36
packages/session-query/session-query/src/config.ts
View File

@@ -25,5 +25,37 @@ export interface Config {
readWindowMax?: number
}
/** Typed session-query failure with a stable machine-routable code. */
export class SessionQueryError extends HarnessError {}
/** Complete stable machine-routable failure taxonomy for session-query. */
export type SessionQueryErrorCode =
| 'SESSION_QUERY_ABORTED'
| 'SESSION_QUERY_DUPLICATE_EXTRACTOR'
| 'SESSION_QUERY_DUPLICATE_PROVIDER'
| 'SESSION_QUERY_EVENT_NOT_FOUND'
| 'SESSION_QUERY_INDEX_FAILED'
| 'SESSION_QUERY_INVALID_CONFIG'
| 'SESSION_QUERY_INVALID_EXTRACTOR'
| 'SESSION_QUERY_INVALID_FILTER'
| 'SESSION_QUERY_INVALID_LIMIT'
| 'SESSION_QUERY_INVALID_LINEAGE'
| 'SESSION_QUERY_INVALID_QUERY'
| 'SESSION_QUERY_INVALID_SURFACE'
| 'SESSION_QUERY_INVALID_WINDOW'
| 'SESSION_QUERY_PERSISTENCE_FAILED'
| 'SESSION_QUERY_PROVIDER_AMBIGUOUS'
| 'SESSION_QUERY_PROVIDER_CONFIGURED_MISSING'
| 'SESSION_QUERY_PROVIDER_CONFIGURED_UNAVAILABLE'
| 'SESSION_QUERY_PROVIDER_ERROR'
| 'SESSION_QUERY_PROVIDER_UNAVAILABLE'
| 'SESSION_QUERY_SESSION_NOT_FOUND'
| 'SESSION_QUERY_SOURCE_CONFLICT'
/** Typed session-query failure whose `code` is one closed taxonomy member. */
export class SessionQueryError extends HarnessError {
declare readonly code: SessionQueryErrorCode
// The base stores the value; this signature narrows its open string code.
// eslint-disable-next-line @typescript-eslint/no-useless-constructor
constructor(message: string, code: SessionQueryErrorCode, options?: ErrorOptions) {
super(message, code, options)
}
}

2
packages/session-query/session-query/src/index.ts
View File

@@ -40,7 +40,7 @@ import { SessionProviderCoordinator } from './provider.ts'
import { eventRecords, traceEventLog, traceLineage } from './tracing.ts'
export type * from './types.ts'
export type { Config } from './config.ts'
export type { Config, SessionQueryErrorCode } from './config.ts'
export {
SESSION_QUERY_DEFAULT_LIMIT,
SESSION_QUERY_MAX_LIMIT,

9
packages/session-query/session-query/src/provider.ts
View File

@@ -8,12 +8,14 @@ import type {
SessionEventRecord,
SessionEventSearchHit,
SessionEventSearchRequest,
SessionEventSearchSpec,
SessionQueryExecContext,
SessionRecord,
SessionSearchHit,
SessionSearchPage,
SessionSearchProvider,
SessionSearchRequest,
SessionSearchSpec,
} from './types.ts'
import type { Config } from './config.ts'
import { SessionQueryError } from './config.ts'
@@ -28,9 +30,6 @@ interface ProviderState {
liveSync: Map<SessionId, Promise<void>>
}
type NormalizedSessionSearchRequest = SessionSearchRequest & { limit: number }
type NormalizedEventSearchRequest = SessionEventSearchRequest & { limit: number }
/** Coordinates one selected provider against live and persisted corpus layers. */
export class SessionProviderCoordinator {
private readonly _configuredProviderId: string | undefined
@@ -257,7 +256,7 @@ export class SessionProviderCoordinator {
return single
}
private _normalizeSessionSearch(request: SessionSearchRequest): NormalizedSessionSearchRequest {
private _normalizeSessionSearch(request: SessionSearchRequest): SessionSearchSpec {
const query = this._queryText(request.query)
const limit = this._limitValue(request.limit)
filterSessionResults<SessionRecord>([], request.sessionFilters ?? [])
@@ -265,7 +264,7 @@ export class SessionProviderCoordinator {
return { ...request, query, limit }
}
private _normalizeEventSearch(request: SessionEventSearchRequest): NormalizedEventSearchRequest {
private _normalizeEventSearch(request: SessionEventSearchRequest): SessionEventSearchSpec {
const query = this._queryText(request.query)
const limit = this._limitValue(request.limit)
filterEventResults<SessionEventRecord>([], request.filters ?? [])

16
packages/session-query/session-query/src/types.ts
View File

@@ -102,6 +102,18 @@ export interface SessionEventSearchRequest extends SessionSearchPageRequest {
filters?: readonly SessionEventResultFilter[]
}
/** Provider-facing cross-session search spec after service normalization. */
export interface SessionSearchSpec extends SessionSearchRequest {
/** Required page size validated and defaulted by the query service. */
limit: number
}
/** Provider-facing event search spec after service normalization. */
export interface SessionEventSearchSpec extends SessionEventSearchRequest {
/** Required page size validated and defaulted by the query service. */
limit: number
}
/** One lightweight event search hit with provider-produced evidence text. */
export interface SessionEventSearchHit extends SessionEventRecord {
/** Plain-text excerpt explaining the match. */
@@ -281,12 +293,12 @@ export interface SessionSearchProvider {
* @param exec - optional cancellation context.
* @returns one ranked session page.
*/
searchSessions(request: SessionSearchRequest, exec?: SessionQueryExecContext): Promise<SessionSearchPage<SessionSearchHit>>
searchSessions(request: SessionSearchSpec, exec?: SessionQueryExecContext): Promise<SessionSearchPage<SessionSearchHit>>
/**
* Search events within one logical session.
* @param request - target session, query, filters, and pagination.
* @param exec - optional cancellation context.
* @returns one ranked event page.
*/
searchEvents(request: SessionEventSearchRequest, exec?: SessionQueryExecContext): Promise<SessionSearchPage<SessionEventSearchHit>>
searchEvents(request: SessionEventSearchSpec, exec?: SessionQueryExecContext): Promise<SessionSearchPage<SessionEventSearchHit>>
}

35
packages/session-query/session-query/tests/session-query.spec.ts
View File

@@ -12,14 +12,15 @@ import SessionQueryService, {
} from '@deepseek-ai/dsh-session-query'
import type {
SessionEventSearchHit,
SessionEventSearchRequest,
SessionEventSearchSpec,
SessionIndexSnapshot,
SessionQueryErrorCode,
SessionRecord,
SessionSearchHit,
SessionSearchPage,
SessionSearchProvider,
SessionSearchProviderStatus,
SessionSearchRequest,
SessionSearchSpec,
} from '@deepseek-ai/dsh-session-query'
declare module '@deepseek-ai/dsh-llm' {
@@ -98,8 +99,8 @@ class FakeProvider implements SessionSearchProvider {
activeHistory: boolean[] = []
removedPersisted: SessionIdType[] = []
removedLive: SessionIdType[] = []
sessionRequests: SessionSearchRequest[] = []
eventRequests: SessionEventSearchRequest[] = []
sessionRequests: SessionSearchSpec[] = []
eventRequests: SessionEventSearchSpec[] = []
failNextLive = false
failNextPersisted = false
failNextActive = false
@@ -162,12 +163,12 @@ class FakeProvider implements SessionSearchProvider {
return Promise.resolve()
}
searchSessions(request: SessionSearchRequest): Promise<SessionSearchPage<SessionSearchHit>> {
searchSessions(request: SessionSearchSpec): Promise<SessionSearchPage<SessionSearchHit>> {
this.sessionRequests.push(structuredClone(request))
return Promise.resolve(structuredClone(this.sessionPage))
}
searchEvents(request: SessionEventSearchRequest): Promise<SessionSearchPage<SessionEventSearchHit>> {
searchEvents(request: SessionEventSearchSpec): Promise<SessionSearchPage<SessionEventSearchHit>> {
this.eventRequests.push(structuredClone(request))
return Promise.resolve(structuredClone(this.eventPage))
}
@@ -180,7 +181,7 @@ async function liveContext(config: ConstructorParameters<typeof SessionQueryServ
return ctx
}
function expectCode(code: string): Error {
function expectCode(code: SessionQueryErrorCode): Error {
return expect.objectContaining({ code }) as Error
}
@@ -361,8 +362,8 @@ describe('logical corpus reads and traces', () => {
TestPersistence.listFailure = undefined
TestPersistence.loadFailure = new Error('load unavailable')
await expect(ctx.sessionQuery.listEvents(persistedOnly.id)).rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
TestPersistence.loadFailure = new SessionQueryError('typed load failure', 'SESSION_QUERY_TEST_FAILURE')
await expect(ctx.sessionQuery.listEvents(persistedOnly.id)).rejects.toThrow(expectCode('SESSION_QUERY_TEST_FAILURE'))
TestPersistence.loadFailure = new SessionQueryError('typed load failure', 'SESSION_QUERY_EVENT_NOT_FOUND')
await expect(ctx.sessionQuery.listEvents(persistedOnly.id)).rejects.toThrow(expectCode('SESSION_QUERY_EVENT_NOT_FOUND'))
await persistenceFiber.dispose()
TestPersistence.loadFailure = undefined
@@ -682,9 +683,11 @@ describe('semantic text extractors', () => {
session.append('user/message', { content: [{ type: 'test/text', value: 'block note' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
const provider = new FakeProvider()
ctx.sessionQuery.registerSearchProvider(provider)
let disposeEvent!: () => void
let disposeContent!: () => void
const extractorFiber = await ctx.plugin(Object.assign((inner: Context) => {
inner.sessionQuery.registerEventTextExtractor('test/note', { version: 'event-v1', extract: event => [event.data.note] })
inner.sessionQuery.registerContentTextExtractor('test/text', { version: 'block-v1', extract: block => [block.value] })
disposeEvent = inner.sessionQuery.registerEventTextExtractor('test/note', { version: 'event-v1', extract: event => [event.data.note] })
disposeContent = inner.sessionQuery.registerContentTextExtractor('test/text', { version: 'block-v1', extract: block => [block.value] })
}, { inject: ['sessionQuery'] }))
await ctx.sessionQuery.searchEvents({ sessionId: session.id, query: 'x' })
@@ -697,11 +700,13 @@ describe('semantic text extractors', () => {
expect(() => ctx.sessionQuery.registerContentTextExtractor('test/text', { version: ' ', extract: () => [] }))
.toThrow(expectCode('SESSION_QUERY_INVALID_EXTRACTOR'))
await extractorFiber.dispose()
disposeEvent()
disposeContent()
await ctx.sessionQuery.searchEvents({ sessionId: session.id, query: 'x' })
const second = provider.live.get(session.id)
expect(second?.documents).toEqual([])
expect(second?.fingerprint).not.toBe(first?.fingerprint)
await extractorFiber.dispose()
const replacementFiber = await ctx.plugin(Object.assign((inner: Context) => {
inner.sessionQuery.registerEventTextExtractor('test/note', { version: 'event-v2', extract: event => [`replacement ${event.data.note}`] })
@@ -712,6 +717,8 @@ describe('semantic text extractors', () => {
expect(third?.documents.map(document => document.text)).toEqual(['replacement event note', 'replacement block note'])
expect(third?.fingerprint).not.toBe(second?.fingerprint)
await replacementFiber.dispose()
await ctx.sessionQuery.searchEvents({ sessionId: session.id, query: 'x' })
expect(provider.live.get(session.id)?.documents).toEqual([])
})
})
@@ -721,8 +728,8 @@ describe('configuration', () => {
await ctx.plugin(SessionStore)
await expect(ctx.plugin(SessionQueryService, { defaultLimit: 3, maxLimit: 2 }))
.rejects.toThrow(expectCode('SESSION_QUERY_INVALID_CONFIG'))
const error = new SessionQueryError('test', 'SESSION_QUERY_TEST')
expect(error).toMatchObject({ name: 'SessionQueryError', code: 'SESSION_QUERY_TEST' })
const error = new SessionQueryError('test', 'SESSION_QUERY_INVALID_CONFIG')
expect(error).toMatchObject({ name: 'SessionQueryError', code: 'SESSION_QUERY_INVALID_CONFIG' })
})
it('uses constructor defaults and removes the service on plugin disposal', async () => {