Merge remote-tracking branch 'origin/master' into codex/truncated-design

# Conflicts:
#	docs/rfc/INDEX.md
This commit is contained in:
Dudu-0223
2026-07-14 11:06:41 +08:00
35 changed files with 1230 additions and 73 deletions

View File

@@ -26,6 +26,7 @@ Packages are grouped by modular role at `packages/<group>/<pkg>/`. The group dir
| [`cordis/`](cordis/README.md) | Self-referential runtime toolset: inspect the live runtime's plugins and services, mount/unmount model-written plugins ([design](../docs/rfc/implemented/feature/2026-07-08-self-referential-cordis-toolset.md)) | Product — stable surface |
| [`hooks/`](hooks/README.md) | Hook bridges + the shared Claude Code / Codex wire-protocol library | Product — stable surface |
| [`session-persistence/`](session-persistence/README.md) | Persistence capability family: the seam + JSONL/SQLite backends | Product — stable surface |
| [`session-query/`](session-query/README.md) | Session retrieval family: logical corpus, surface records, and bounded exact reads | Product — stable surface |
| [`ui/`](ui/README.md) | Editor/client integration surfaces: ACP bridge, JSON-RPC SDK server, app packages, user-approval and user-interaction seams, ask-user tool | Product — stable surface |
| [`support/`](support/README.md) | Dev/test/example infrastructure (invariants, replay adapter, subagent mock) | Support — lower compatibility expectations |
| [`util/`](util/README.md) | Low-level zero-dependency primitives shared across groups (branding, timeout, retention) | Support — small, stable, harness-dep-free |

View File

@@ -161,6 +161,15 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
'abstract list(): Promise<SessionHeader[]>',
],
},
{
key: 'sessionQuery',
summary: 'Live-preferred logical-corpus and exact-event read service.',
methods: [
'listSessions(): Promise<SessionRecord[]>',
'async listEvents(sessionId: SessionId): Promise<SessionEventRecord[]>',
'async readEvent(request: SessionEventReadRequest): Promise<SessionEventWindow>',
],
},
{
key: 'sessions',
summary: 'In-memory session store (`ctx.sessions`).',
@@ -819,10 +828,26 @@ export const TYPE_API: readonly TypeApiEntry[] = [
name: 'SessionEventMap',
declaration: 'export interface SessionEventMap {\n \'turn/start\': {\n turn: number;\n trigger: TurnTrigger;\n };\n \'turn/end\': {\n turn: number;\n reason: TurnEndReason;\n };\n \'step/start\': {\n turn: number;\n step: number;\n };\n \'step/end\': {\n turn: number;\n step: number;\n };\n \'user/message\': {\n content: ContentBlock[];\n source: MessageSource;\n };\n \'prompt/blocked\': {\n content: ContentBlock[];\n source: MessageSource;\n reason: string;\n };\n \'context/message\': {\n content: ContentBlock[];\n source: MessageSource;\n };\n \'assistant/chunk\': {\n turn: number;\n step: number;\n chunk: StreamChunk;\n };\n \'assistant/message\': {\n turn: number;\n step: number;\n content: ContentBlock[];\n usage?: TokenUsage;\n };\n \'tool/call\': {\n turn: number;\n step: number;\n callId: CallId;\n name: string;\n arguments: string;\n };\n \'tool/result\': {\n turn: number;\n step: number;\n callId: CallId;\n content: ContentBlock[];\n isError: boolean;\n error?: {\n name: string;\n code: string;\n };\n meta?: unknown;\n };\n \'steering/message\': {\n turn: number;\n content: ContentBlock[];\n source: MessageSource;\n };\n \'todo/write\': {\n todos: TodoItem[];\n };\n \'request/header\': {\n header: E /* …truncated — full shape in source */',
},
{
name: 'SessionEventReadRequest',
declaration: 'export interface SessionEventReadRequest {\n sessionId: SessionId;\n seq: number;\n before?: number;\n after?: number;\n}',
},
{
name: 'SessionEventRecord',
declaration: 'export interface SessionEventRecord {\n sessionId: SessionId;\n seq: number;\n type: SessionEventType;\n time: number;\n surface: SessionEventSurface;\n}',
},
{
name: 'SessionEventSurface',
declaration: 'export type SessionEventSurface = \'current\' | \'shadowed\' | \'log-only\';',
},
{
name: 'SessionEventType',
declaration: 'export type SessionEventType = keyof SessionEventMap;',
},
{
name: 'SessionEventWindow',
declaration: 'export interface SessionEventWindow {\n session: SessionHeader;\n target: SessionEvent;\n events: SessionEvent[];\n startSeq: number;\n endSeq: number;\n}',
},
{
name: 'SessionForkSource',
declaration: 'export type SessionForkSource = Session | SessionId;',
@@ -835,6 +860,10 @@ export const TYPE_API: readonly TypeApiEntry[] = [
name: 'SessionId',
declaration: 'export type SessionId = Branded<\'SessionId\'>;',
},
{
name: 'SessionRecord',
declaration: 'export interface SessionRecord {\n header: SessionHeader;\n live: boolean;\n persisted: boolean;\n}',
},
{
name: 'SkillCandidate',
declaration: 'export interface SkillCandidate extends SkillSummary {\n readonly rank: number;\n readonly locator: unknown;\n readonly path?: string;\n readonly metadata?: Readonly<Record<string, unknown>>;\n}',

View File

@@ -49,6 +49,7 @@ Durable values need one accepted representation, not a check followed by a secon
- `SurfaceOp` — how a surface node entered the linked list: `'append'` (normal tail append) or `{ op: 'replace', start, end }` (replace nodes from `start` through `end` inclusive — both must be valid surface node seqs; `start === end` replaces a single node). Used by compaction to shadow old nodes without deleting them.
- `SurfaceIntent` — `{ surfaceOp: SurfaceOp; sourceEventSeqs?: number[] }`, the required third parameter to `session.append()` for surface-eligible types.
- `SurfaceNode` — `{ seq: number; prev: number | null; next: number | null }`, one node in the surface linked list.
- `foldSurface(events)` — replay the canonical surface transitions into detached current nodes and actual replacement ranges, rejecting surface-eligible events that lack their mandatory marker. `SurfaceManager` shares the same transitions while retaining its incremental cache.
- `isSurfaceEvent(event)` / `isSurfaceEligibleType(type)` — the first narrows a `SessionEvent` to a fully-formed surface node (type is surface-eligible AND `surfaceOp` present); the second is the type-only check (is this one of the five `SurfaceEventType` values?), used to detect a surface-eligible event MISSING its marker — e.g. when validating a seed/load log.
### Request-header reconstruction (`request-header.ts`)

View File

@@ -22,8 +22,8 @@ export * from './types.ts'
export { isJsonValue, snapshotJsonValue } from './json.ts'
export type { JsonValue } from './json.ts'
export { interruptedTurnClosers } from './repair.ts'
export type { SurfaceNode } from './surface.ts'
export { isSurfaceEvent, isSurfaceEligibleType } from './surface.ts'
export type { SurfaceFoldReplacement, SurfaceFoldResult, SurfaceNode } from './surface.ts'
export { foldSurface, isSurfaceEvent, isSurfaceEligibleType } from './surface.ts'
export { isToolPairingBalanced } from './tool-pairing.ts'
export { applyHeaderDelta, canonicalHeader, diffHeader, foldRequestHeader, headerEquals } from './request-header.ts'

View File

@@ -61,6 +61,131 @@ export interface SurfaceNode {
next: number | null
}
/** One replacement operation observed while folding a session surface. */
export interface SurfaceFoldReplacement {
/** Seq of the event that replaced the prior surface range. */
seq: number
/** Declared inclusive start seq of the replaced surface range. */
start: number
/** Declared inclusive end seq of the replaced surface range. */
end: number
/** Actual surface nodes removed by the operation, in surface order. */
shadowedSeqs: number[]
}
/** Complete result of replaying the surface operations in a session log. */
export interface SurfaceFoldResult {
/** Current surface nodes in linked-list order. */
nodes: SurfaceNode[]
/** Replacement operations in event order. */
replacements: SurfaceFoldReplacement[]
}
/** Mutable state shared by the incremental manager and the full-log fold. */
interface SurfaceFoldState {
nodes: SurfaceNode[]
nodeBySeq: Map<number, SurfaceNode>
replaceGeneration: number
}
/** Create an empty surface fold state. */
function createFoldState(replaceGeneration = 0): SurfaceFoldState {
return {
nodes: [],
nodeBySeq: new Map(),
replaceGeneration,
}
}
/** Apply one event and return replacement metadata only when one occurred. */
function applySurfaceEvent(
state: SurfaceFoldState,
event: SessionEvent,
): SurfaceFoldReplacement | undefined {
if (!isSurfaceEligibleType(event.type)) return
if (!isSurfaceEvent(event)) {
throw new Error(`surface event "${event.type}" (seq ${event.seq}) carries no surfaceOp marker`)
}
if (event.surfaceOp === 'append') {
const tail = state.nodes.length > 0 ? state.nodes[state.nodes.length - 1] : undefined
const node: SurfaceNode = { seq: event.seq, prev: tail?.seq ?? null, next: null }
if (tail) tail.next = event.seq
state.nodes.push(node)
state.nodeBySeq.set(event.seq, node)
return
}
return {
seq: event.seq,
start: event.surfaceOp.start,
end: event.surfaceOp.end,
shadowedSeqs: replaceSurface(state, event.seq, event.surfaceOp),
}
}
/** Apply one positional replacement and return the nodes it removed. */
function replaceSurface(
state: SurfaceFoldState,
newSeq: number,
op: Extract<SurfaceOp, { op: 'replace' }>,
): number[] {
const startNode = state.nodeBySeq.get(op.start)
if (!startNode) {
throw new Error(`surface replace: start seq ${op.start} not found in surface`)
}
const endNode = state.nodeBySeq.get(op.end)
if (!endNode) {
throw new Error(`surface replace: end seq ${op.end} not found in surface`)
}
const startIdx = state.nodes.indexOf(startNode)
const endIdx = state.nodes.indexOf(endNode)
if (startIdx > endIdx) {
throw new Error(`surface replace: start seq ${op.start} (index ${startIdx}) is after end seq ${op.end} (index ${endIdx})`)
}
const removed = state.nodes.splice(startIdx, endIdx - startIdx + 1)
for (const node of removed) state.nodeBySeq.delete(node.seq)
const prevNode = startIdx > 0 ? state.nodes[startIdx - 1] : undefined
const nextNode = startIdx < state.nodes.length ? state.nodes[startIdx] : undefined
const newNode: SurfaceNode = {
seq: newSeq,
prev: prevNode?.seq ?? null,
next: nextNode?.seq ?? null,
}
if (prevNode) prevNode.next = newSeq
if (nextNode) nextNode.prev = newSeq
state.nodes.splice(startIdx, 0, newNode)
state.nodeBySeq.set(newSeq, newNode)
state.replaceGeneration += 1
return removed.map(node => node.seq)
}
/**
* Replay a complete session log through the canonical surface fold.
*
* The returned arrays and nodes are detached snapshots. The incremental
* {@link SurfaceManager} uses the same transition functions, so query read
* models cannot disagree with `deriveMessages()` about replacement ranges.
* @param events - session events in contiguous seq order.
* @returns the current surface and every positional replacement.
* @throws when a surface-eligible event lacks its mandatory `surfaceOp`, or a
* replacement names nodes that are absent or reversed on the current surface.
*/
export function foldSurface(events: readonly SessionEvent[]): SurfaceFoldResult {
const state = createFoldState()
const replacements: SurfaceFoldReplacement[] = []
for (const event of events) {
const replacement = applySurfaceEvent(state, event)
if (replacement !== undefined) replacements.push(replacement)
}
return {
nodes: state.nodes.map(node => ({ ...node })),
replacements,
}
}
/**
* Maintains a cached linked list of surface nodes, rebuilt lazily from
* `surfaceOp` markers in the event log. Because the log is append-only, it
@@ -69,16 +194,11 @@ export interface SurfaceNode {
* whole log.
*/
export class SurfaceManager {
/** Surface nodes in linked-list order (head to tail). Empty until first access. */
private _nodes: SurfaceNode[] = []
/** Map from event seq → node. */
private _nodeBySeq = new Map<number, SurfaceNode>()
/** Incremental state shared with the complete surface fold. */
private _state = createFoldState()
/** The last processed seq. -1 forces a full rebuild on first access. */
private _lastProcessedSeq = -1
/** Rewrite generation — see {@link replaceGeneration}. */
private _replaceGeneration = 0
constructor(private log: readonly SessionEvent[]) {}
/**
@@ -88,11 +208,9 @@ export class SurfaceManager {
*/
invalidate(): void {
this._lastProcessedSeq = -1
this._nodes = []
this._nodeBySeq.clear()
// A wholesale rebuild is a rewrite: bump the generation so incremental
// consumers (the session's derived-message cache) discard their view.
this._replaceGeneration += 1
this._state = createFoldState(this._state.replaceGeneration + 1)
}
/**
@@ -106,13 +224,13 @@ export class SurfaceManager {
*/
get replaceGeneration(): number {
if (this._lastProcessedSeq < this.log.length - 1) this._processDelta()
return this._replaceGeneration
return this._state.replaceGeneration
}
/** The surface nodes in linked-list order (head to tail). */
get nodes(): readonly SurfaceNode[] {
if (this._lastProcessedSeq < this.log.length - 1) this._processDelta()
return this._nodes
return this._state.nodes
}
/**
@@ -124,61 +242,8 @@ export class SurfaceManager {
// Index is bounded by i < this.log.length — never undefined.
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
const event = this.log[i]!
// isSurfaceEvent checks event.type first (is it a surface-eligible type?)
// then checks that surfaceOp is present. Only after both pass do we treat
// it as a SurfaceEvent with mandatory surfaceOp.
if (!isSurfaceEvent(event)) continue
if (event.surfaceOp === 'append') {
const tail = this._nodes.length > 0 ? this._nodes[this._nodes.length - 1] : undefined
const node: SurfaceNode = { seq: event.seq, prev: tail?.seq ?? null, next: null }
if (tail) tail.next = event.seq
this._nodes.push(node)
this._nodeBySeq.set(event.seq, node)
} else {
this._replace(event.seq, event.surfaceOp)
}
applySurfaceEvent(this._state, event)
}
this._lastProcessedSeq = this.log.length - 1
}
/** Apply a replace operation to the in-progress surface. */
private _replace(
newSeq: number,
op: Extract<SurfaceOp, { op: 'replace' }>,
): void {
const startNode = this._nodeBySeq.get(op.start)
if (!startNode) {
throw new Error(`surface replace: start seq ${op.start} not found in surface`)
}
const endNode = this._nodeBySeq.get(op.end)
if (!endNode) {
throw new Error(`surface replace: end seq ${op.end} not found in surface`)
}
const startIdx = this._nodes.indexOf(startNode)
const endIdx = this._nodes.indexOf(endNode)
if (startIdx > endIdx) {
throw new Error(`surface replace: start seq ${op.start} (index ${startIdx}) is after end seq ${op.end} (index ${endIdx})`)
}
// Remove shadowed nodes from `[startIdx, endIdx]` inclusive.
const count = endIdx - startIdx + 1
const removed = this._nodes.splice(startIdx, count)
for (const r of removed) this._nodeBySeq.delete(r.seq)
// Insert the new node where the removed range was.
const prevNode = startIdx > 0 ? this._nodes[startIdx - 1] : undefined
const nextNode = startIdx < this._nodes.length ? this._nodes[startIdx] : undefined
const newNode: SurfaceNode = {
seq: newSeq,
prev: prevNode?.seq ?? null,
next: nextNode?.seq ?? null,
}
if (prevNode) prevNode.next = newSeq
if (nextNode) nextNode.prev = newSeq
this._nodes.splice(startIdx, 0, newNode)
this._nodeBySeq.set(newSeq, newNode)
this._replaceGeneration += 1
}
}

View File

@@ -1,6 +1,6 @@
import { describe, expect, it } from 'vitest'
import type { SessionEvent, SurfaceEvent, SurfaceEventType } from '@deepseek-ai/dsh-session'
import { Session, SessionId, isSurfaceEligibleType, isSurfaceEvent } from '@deepseek-ai/dsh-session'
import { Session, SessionId, foldSurface, isSurfaceEligibleType, isSurfaceEvent } from '@deepseek-ai/dsh-session'
import { CallId } from '@deepseek-ai/dsh-llm'
/** Build a minimal session with turn boundaries and a single user message. */
@@ -14,6 +14,59 @@ function surfaceSession(): Session {
}
describe('SurfaceManager', () => {
it('shares exact nodes and nested replacement ranges with foldSurface', () => {
const s = new Session(SessionId('shared-fold'))
s.append('user/message', { content: [{ type: 'text', text: 'a' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
s.append('user/message', { content: [{ type: 'text', text: 'b' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
s.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'summary' }] }, { surfaceOp: { op: 'replace', start: 0, end: 0 }, sourceEventSeqs: [0] })
s.append('assistant/message', { turn: 1, step: 2, content: [{ type: 'text', text: 'summary 2' }] }, { surfaceOp: { op: 'replace', start: 2, end: 1 }, sourceEventSeqs: [2, 1] })
const folded = foldSurface(s.events)
expect(folded.nodes).toEqual(s.surface.nodes)
expect(folded.replacements).toEqual([
{ seq: 2, start: 0, end: 0, shadowedSeqs: [0] },
{ seq: 3, start: 2, end: 1, shadowedSeqs: [2, 1] },
])
folded.nodes[0]!.next = 99
folded.replacements[0]!.shadowedSeqs.push(99)
expect(s.surface.nodes).toEqual([{ seq: 3, prev: null, next: null }])
expect(foldSurface(s.events).replacements[0]!.shadowedSeqs).toEqual([0])
})
it('does not retain fold-only replacement history in incremental state', () => {
const s = new Session(SessionId('incremental-state'))
s.append('user/message', { content: [{ type: 'text', text: 'a' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
s.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'b' }] }, { surfaceOp: { op: 'replace', start: 0, end: 0 } })
expect(s.surface.nodes).toEqual([{ seq: 1, prev: null, next: null }])
const manager = s.surface as unknown as { _state: object }
expect(Object.hasOwn(manager._state, 'replacements')).toBe(false)
expect(foldSurface(s.events).replacements).toEqual([
{ seq: 1, start: 0, end: 0, shadowedSeqs: [0] },
])
})
it('foldSurface reports the same invalid replacement failures as the incremental manager', () => {
const s = new Session(SessionId('shared-fold-invalid'))
s.append('user/message', { content: [{ type: 'text', text: 'a' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
s.append('assistant/message', { turn: 1, step: 1, content: [] }, { surfaceOp: { op: 'replace', start: 42, end: 0 }, sourceEventSeqs: [0] })
expect(() => foldSurface(s.events)).toThrow(/start seq 42 not found/)
expect(() => s.surface.nodes).toThrow(/start seq 42 not found/)
})
it('foldSurface rejects a surface-eligible event without its mandatory marker', () => {
const malformed: SessionEvent = {
type: 'user/message',
seq: 0,
time: 1,
data: { content: [{ type: 'text', text: 'hidden' }], source: { kind: 'user' } },
}
expect(() => foldSurface([malformed]))
.toThrow(/surface event "user\/message" \(seq 0\) carries no surfaceOp marker/)
})
it('rebuilds a linked list from surfaceOp: append markers', () => {
const s = surfaceSession()
const nodes = s.surface.nodes

View File

@@ -0,0 +1,9 @@
# session-query/ — session retrieval capability family
Trusted exact reads over live and durable session logs. Phase one contains one interface package that owns `ctx.sessionQuery`, logical-corpus precedence, surface classification, and bounded event reads.
| Package | Role | ctx key |
|---|---|---|
| [`session-query/`](session-query/README.md) | Logical-corpus and exact-event read service | `ctx.sessionQuery` |
The family is independent of compaction: it reads the canonical session log but does not participate in compaction policy or execution. Full-text search remains proposed as a phase-two SQLite package rather than a speculative provider seam in this interface package.

View File

@@ -0,0 +1,23 @@
# @deepseek-ai/dsh-session-query
Exact session-history retrieval through `ctx.sessionQuery`. The service presents live `ctx.sessions` and an optional, dynamically mounted `ctx.sessionPersistence` as one logical corpus. Matching ids produce one record: live events win, while `live` and `persisted` report both source availabilities. Conflicting immutable headers fail with `SESSION_QUERY_SOURCE_CONFLICT`.
This is trusted context-wide infrastructure. It performs no caller authorization; a future model tool or UI must constrain which sessions its caller may inspect.
## Reads
- `listSessions()` reads current persistence metadata, merges live records with live precedence, and returns cloned records in deterministic newest-first order.
- `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.
- `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`.
Persistence is optional and may mount or unmount dynamically. A cross-corpus list fails with `SESSION_QUERY_PERSISTENCE_FAILED` while mounted persistence is unreadable. A read targeting a known live session does not consult persistence, so durable backend health cannot make current in-memory history unreadable. Persisted exact reads list before loading, and reject a metadata mismatch rather than combining inconsistent observations.
`SessionQueryError.code` is a closed union: `SESSION_QUERY_EVENT_NOT_FOUND`, `SESSION_QUERY_INVALID_CONFIG`, `SESSION_QUERY_INVALID_SURFACE`, `SESSION_QUERY_INVALID_WINDOW`, `SESSION_QUERY_PERSISTENCE_FAILED`, `SESSION_QUERY_SESSION_NOT_FOUND`, and `SESSION_QUERY_SOURCE_CONFLICT`.
## Configuration
| Key | Default | Contract |
|---|---:|---|
| `readWindowMax` | `50` | Maximum `before` or `after` raw-event count. |
This phase deliberately has no filters, lineage/provenance traversal, extraction registry, search-provider protocol, index synchronization, or model-facing tool. Full-text search belongs beside its first real implementation; the proposed SQLite package and its single transaction/reconciliation owner are described in the [phase-two RFC](../../../docs/rfc/proposed/feature/2026-07-10-sqlite-session-query-provider.md).

View File

@@ -0,0 +1,44 @@
{
"name": "@deepseek-ai/dsh-session-query",
"description": "Live-preferred exact session-history retrieval service (ctx.sessionQuery)",
"version": "0.0.1",
"private": true,
"type": "module",
"main": "lib/index.js",
"types": "lib/types/index.d.ts",
"exports": {
".": {
"types": "./lib/types/index.d.ts",
"default": "./lib/index.js"
},
"./src/*": "./src/*",
"./package.json": "./package.json"
},
"files": [
"lib/index.js",
"lib/types/**/*.d.ts",
"lib/types/**/*.d.ts.map",
"src"
],
"license": "BSD-3-Clause",
"peerDependencies": {
"@deepseek-ai/dsh-llm": "^0.0.1",
"@deepseek-ai/dsh-session": "^0.0.1",
"@deepseek-ai/dsh-session-persistence": "^0.0.1",
"cordis": "^4.0.0-rc.6"
},
"peerDependenciesMeta": {
"@deepseek-ai/dsh-session-persistence": {
"optional": true
}
},
"dependencies": {
"schemastery": "^3.18.0"
},
"devDependencies": {
"@deepseek-ai/dsh-llm": "workspace:^",
"@deepseek-ai/dsh-session": "workspace:^",
"@deepseek-ai/dsh-session-persistence": "workspace:^",
"cordis": "^4.0.0-rc.6"
}
}

View File

@@ -0,0 +1,33 @@
/** Public configuration and typed failures for session-query. */
import { HarnessError } from '@deepseek-ai/dsh-llm'
/** Default maximum `before`/`after` raw-event window. */
export const SESSION_QUERY_READ_WINDOW_MAX = 50
/** Configuration for exact session-query reads. */
export interface Config {
/** Maximum accepted raw read context on either side. Defaults to 50. */
readWindowMax?: number
}
/** Stable machine-routable failure taxonomy for exact session reads. */
export type SessionQueryErrorCode =
| 'SESSION_QUERY_EVENT_NOT_FOUND'
| 'SESSION_QUERY_INVALID_CONFIG'
| 'SESSION_QUERY_INVALID_SURFACE'
| 'SESSION_QUERY_INVALID_WINDOW'
| 'SESSION_QUERY_PERSISTENCE_FAILED'
| '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)
}
}

View File

@@ -0,0 +1,136 @@
/** Live/persisted logical-corpus resolution for session-query. */
import type { Context } from 'cordis'
import type { Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import type SessionPersistence from '@deepseek-ai/dsh-session-persistence'
import type { SessionRecord } from './types.ts'
import { SessionQueryError } from './config.ts'
/** Detached source selected for one exact read. */
export interface LogicalSession {
/** Cloned source header. */
header: SessionHeader
/** Cloned raw event log. */
events: SessionEvent[]
}
/** Resolves a live-preferred corpus against the persistence service mounted now. */
export class SessionCorpus {
private _persistence: SessionPersistence | undefined
constructor(private readonly _ctx: Context) {
_ctx.effect(() => {
const fiber = _ctx.inject(['sessionPersistence'], (childCtx: Context) => {
const service = childCtx.sessionPersistence
this._persistence = service
childCtx.effect(() => () => {
/* v8 ignore next -- a stale optional-service disposer cannot clear a replacement */
if (this._persistence === service) this._persistence = undefined
}, 'sessionQuery.persistenceBinding')
})
return () => void fiber.dispose()
}, 'sessionQuery.optionalPersistence')
}
/**
* List the complete logical corpus with live precedence and cloned headers.
* @returns records in deterministic newest-first order.
*/
async listSessions(): Promise<SessionRecord[]> {
const persistence = this._persistence
const persisted = persistence === undefined ? [] : await listPersisted(persistence)
const records = new Map<SessionId, SessionRecord>()
for (const header of persisted) {
records.set(header.id, { header: structuredClone(header), live: false, persisted: true })
}
for (const session of this._ctx.sessions.list()) {
const durable = records.get(session.id)
if (durable !== undefined) assertCompatibleHeaders(session.header, durable.header)
records.set(session.id, {
header: structuredClone(session.header),
live: true,
persisted: durable !== undefined,
})
}
return [...records.values()].sort(compareSessions)
}
/**
* Load one logical source, preferring a detached live snapshot.
*
* 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.
* @returns detached live-preferred header and events.
*/
async load(sessionId: SessionId): Promise<LogicalSession> {
const live = this._ctx.sessions.get(sessionId)
if (live !== undefined) return snapshotLive(live)
const persistence = this._persistence
if (persistence === undefined) throw notFound(sessionId)
const listed = (await listPersisted(persistence)).find(header => header.id === sessionId)
if (listed === undefined) throw notFound(sessionId)
let loaded: Awaited<ReturnType<SessionPersistence['load']>>
try {
loaded = await persistence.load(sessionId)
} catch (error: unknown) {
throw new SessionQueryError(
`failed to load session "${sessionId}": ${errorMessage(error)}`,
'SESSION_QUERY_PERSISTENCE_FAILED',
{ cause: error },
)
}
assertCompatibleHeaders(loaded.meta, listed)
return {
header: structuredClone(loaded.meta),
events: loaded.events.map(event => structuredClone(event)),
}
}
}
async function listPersisted(persistence: SessionPersistence): Promise<SessionHeader[]> {
try {
return await persistence.list()
} catch (error: unknown) {
throw new SessionQueryError(
`session persistence listing failed: ${errorMessage(error)}`,
'SESSION_QUERY_PERSISTENCE_FAILED',
{ cause: error },
)
}
}
function snapshotLive(session: Session): LogicalSession {
return {
header: structuredClone(session.header),
events: session.events.map(event => structuredClone(event)),
}
}
function assertCompatibleHeaders(a: SessionHeader, b: SessionHeader): void {
if (
a.version !== b.version
|| a.id !== b.id
|| a.createdAt !== b.createdAt
|| a.cwd !== b.cwd
|| a.parentSession !== b.parentSession
|| a.seedLength !== b.seedLength
) {
throw new SessionQueryError(
`live and persisted headers conflict for session "${a.id}"`,
'SESSION_QUERY_SOURCE_CONFLICT',
)
}
}
function compareSessions(a: SessionRecord, b: SessionRecord): number {
return b.header.createdAt - a.header.createdAt || a.header.id.localeCompare(b.header.id)
}
function notFound(sessionId: SessionId): SessionQueryError {
return new SessionQueryError(`session "${sessionId}" not found`, 'SESSION_QUERY_SESSION_NOT_FOUND')
}
function errorMessage(error: unknown): string {
return error instanceof Error ? error.message : 'unknown error'
}

View File

@@ -0,0 +1,136 @@
/**
* Exact session-history reads over live and optionally persisted logs.
*
* @module @deepseek-ai/dsh-session-query
*/
import { Context, Service } from 'cordis'
import z from 'schemastery'
import { foldSurface } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
import type {
SessionEventReadRequest,
SessionEventRecord,
SessionEventWindow,
SessionRecord,
} from './types.ts'
import {
SESSION_QUERY_READ_WINDOW_MAX,
SessionQueryError,
type Config,
} from './config.ts'
import { SessionCorpus } from './corpus.ts'
export type * from './types.ts'
export type { Config, SessionQueryErrorCode } from './config.ts'
export { SESSION_QUERY_READ_WINDOW_MAX, SessionQueryError } from './config.ts'
declare module 'cordis' {
interface Context {
sessionQuery: SessionQueryService
}
}
/** Live-preferred logical-corpus and exact-event read service. */
export class SessionQueryService extends Service {
static inject = ['sessions']
static Config: z<Config> = z.object({
readWindowMax: z.number().step(1).min(0).default(SESSION_QUERY_READ_WINDOW_MAX),
})
private readonly _readWindowMax: number
private readonly _corpus: SessionCorpus
constructor(ctx: Context, config: Config = {}) {
super(ctx, 'sessionQuery')
this._readWindowMax = config.readWindowMax ?? SESSION_QUERY_READ_WINDOW_MAX
if (!Number.isInteger(this._readWindowMax) || this._readWindowMax < 0) {
throw new SessionQueryError(
'session-query: readWindowMax must be a non-negative integer',
'SESSION_QUERY_INVALID_CONFIG',
)
}
this._corpus = new SessionCorpus(ctx)
}
/**
* List the complete logical corpus using live-preferred records.
* @returns deterministic newest-first cloned session records.
*/
listSessions(): Promise<SessionRecord[]> {
return this._corpus.listSessions()
}
/**
* List lightweight raw-log event records for one logical session.
* @param sessionId - live-preferred session id to read.
* @returns event records in ascending seq order.
*/
async listEvents(sessionId: SessionId): Promise<SessionEventRecord[]> {
const loaded = await this._corpus.load(sessionId)
return eventRecords(sessionId, loaded.events)
}
/**
* Read one full event plus a bounded raw-log context window.
* @param request - target session/seq and context sizes.
* @returns cloned target and neighboring events.
*/
async readEvent(request: SessionEventReadRequest): Promise<SessionEventWindow> {
const before = this._readWindow('before', request.before)
const after = this._readWindow('after', request.after)
const loaded = await this._corpus.load(request.sessionId)
const target = loaded.events[request.seq]
if (target === undefined || target.seq !== request.seq) {
throw new SessionQueryError(
`session "${request.sessionId}" has no event at seq ${request.seq}`,
'SESSION_QUERY_EVENT_NOT_FOUND',
)
}
const startSeq = Math.max(0, request.seq - before)
const endSeq = Math.min(loaded.events.length - 1, request.seq + after)
return {
session: loaded.header,
target,
events: loaded.events.slice(startSeq, endSeq + 1),
startSeq,
endSeq,
}
}
private _readWindow(name: 'before' | 'after', value: number | undefined): number {
if (value === undefined) return 0
if (!Number.isInteger(value) || value < 0 || value > this._readWindowMax) {
throw new SessionQueryError(
`${name} must be an integer between 0 and ${this._readWindowMax}`,
'SESSION_QUERY_INVALID_WINDOW',
)
}
return value
}
}
function eventRecords(sessionId: SessionId, events: readonly SessionEvent[]): SessionEventRecord[] {
let folded: ReturnType<typeof foldSurface>
try {
folded = foldSurface(events)
} catch (error: unknown) {
throw new SessionQueryError(
/* v8 ignore next -- foldSurface throws Error instances */
`invalid session surface: ${error instanceof Error ? error.message : 'unknown error'}`,
'SESSION_QUERY_INVALID_SURFACE',
{ cause: error },
)
}
const current = new Set(folded.nodes.map(node => node.seq))
const shadowed = new Set(folded.replacements.flatMap(replacement => replacement.shadowedSeqs))
return events.map(event => ({
sessionId,
seq: event.seq,
type: event.type,
time: event.time,
surface: current.has(event.seq) ? 'current' : shadowed.has(event.seq) ? 'shadowed' : 'log-only',
}))
}
export default SessionQueryService

View File

@@ -0,0 +1,60 @@
/**
* Public records for exact reads over the live-preferred logical session corpus.
*
* @module @deepseek-ai/dsh-session-query/types
*/
import type { SessionEvent, SessionEventType, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
/** Whether an event is current model context, replaced context, or raw-log-only. */
export type SessionEventSurface = 'current' | 'shadowed' | 'log-only'
/** Lightweight identity and source availability for one logical session. */
export interface SessionRecord {
/** Cloned session header selected from the live-preferred corpus. */
header: SessionHeader
/** Whether the id currently exists in `ctx.sessions`. */
live: boolean
/** Whether the active persistence backend currently materializes the id. */
persisted: boolean
}
/** Lightweight metadata for one event within a logical session. */
export interface SessionEventRecord {
/** Session that owns the event. */
sessionId: SessionId
/** Monotonic event seq within the session. */
seq: number
/** Discriminant of the session event. */
type: SessionEventType
/** Event timestamp in Unix epoch milliseconds. */
time: number
/** Event placement in the folded session surface. */
surface: SessionEventSurface
}
/** Request for one event plus raw neighboring log context. */
export interface SessionEventReadRequest {
/** Session that owns the target event. */
sessionId: SessionId
/** Target event seq. */
seq: number
/** Number of preceding raw events to include. */
before?: number
/** Number of following raw events to include. */
after?: number
}
/** Full target event and a bounded raw-log window. */
export interface SessionEventWindow {
/** Cloned header for the live-preferred source read. */
session: SessionHeader
/** Full cloned target event. */
target: SessionEvent
/** Full cloned events from `startSeq` through `endSeq`. */
events: SessionEvent[]
/** First seq included in `events`. */
startSeq: number
/** Last seq included in `events`. */
endSeq: number
}

View File

@@ -0,0 +1,270 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import SessionStore, { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionHeader, SessionId as SessionIdType } from '@deepseek-ai/dsh-session'
import SessionPersistence from '@deepseek-ai/dsh-session-persistence'
import SessionQueryService, {
type SessionQueryErrorCode,
} from '@deepseek-ai/dsh-session-query'
function header(id: string, createdAt = 1, extra: Partial<SessionHeader> = {}): SessionHeader {
return { version: SESSION_FORMAT_VERSION, id: SessionId(id), createdAt, ...extra }
}
function eventLog(text = 'hello'): SessionEvent[] {
return [{
type: 'user/message',
seq: 0,
time: 10,
data: { content: [{ type: 'text', text }], source: { kind: 'user' } },
surfaceOp: 'append',
}]
}
class TestPersistence extends SessionPersistence {
static entries = new Map<SessionIdType, { meta: SessionHeader; events: SessionEvent[] }>()
static listFailure: unknown
static loadFailure: unknown
static afterList: (() => void) | undefined
static reset(entries: readonly { meta: SessionHeader; events: SessionEvent[] }[] = []): void {
this.entries = new Map(entries.map(entry => [entry.meta.id, structuredClone(entry)]))
this.listFailure = undefined
this.loadFailure = undefined
this.afterList = undefined
}
create(meta: SessionHeader): Promise<void> {
TestPersistence.entries.set(meta.id, { meta: structuredClone(meta), events: [] })
return Promise.resolve()
}
append(id: SessionIdType, events: readonly SessionEvent[]): Promise<void> {
const entry = TestPersistence.entries.get(id)
if (entry === undefined) return Promise.reject(new Error('missing test session'))
entry.events.push(...structuredClone(events))
return Promise.resolve()
}
load(id: SessionIdType): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
if (TestPersistence.loadFailure !== undefined) return rejectUnknown(TestPersistence.loadFailure)
const entry = TestPersistence.entries.get(id)
if (entry === undefined) return Promise.reject(new Error('missing test session'))
return Promise.resolve(structuredClone(entry))
}
list(): Promise<SessionHeader[]> {
if (TestPersistence.listFailure !== undefined) return rejectUnknown(TestPersistence.listFailure)
const headers = [...TestPersistence.entries.values()].map(entry => structuredClone(entry.meta))
TestPersistence.afterList?.()
return Promise.resolve(headers)
}
}
async function liveContext(config: ConstructorParameters<typeof SessionQueryService>[1] = {}): Promise<Context> {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(SessionQueryService, config)
return ctx
}
function expectCode(code: SessionQueryErrorCode): Error {
return expect.objectContaining({ code }) as Error
}
function rejectUnknown<T>(reason: unknown): Promise<T> {
return new Promise<T>((_resolve, reject) => {
// Exercise containment for an implementation that violates the Error rejection convention.
// eslint-disable-next-line @typescript-eslint/prefer-promise-reject-errors
reject(reason)
})
}
describe('session-query exact reads', () => {
it('lists live sessions deterministically and returns detached headers', async () => {
const ctx = await liveContext()
const older = ctx.sessions.create(SessionId('older'), { meta: { createdAt: 1 } })
ctx.sessions.create(SessionId('z'), { meta: { createdAt: 2 } })
ctx.sessions.create(SessionId('a'), { meta: { createdAt: 2 } })
const records = await ctx.sessionQuery.listSessions()
expect(records.map(record => record.header.id)).toEqual([SessionId('a'), SessionId('z'), older.id])
expect(records.every(record => record.live && !record.persisted)).toBe(true)
Object.assign(records[2]!.header, { createdAt: 99 })
expect(older.header.createdAt).toBe(1)
})
it('classifies current, shadowed, and raw-log-only events through foldSurface', async () => {
const ctx = await liveContext()
const session = ctx.sessions.create(SessionId('surface'))
const first = session.append(
'user/message',
{ content: [{ type: 'text', text: 'first' }], source: { kind: 'user' } },
{ surfaceOp: 'append' },
)
session.append('assistant/chunk', {
turn: 1,
step: 1,
chunk: { type: 'text-delta', index: 0, text: 'draft' },
})
session.append(
'assistant/message',
{ turn: 1, step: 1, content: [{ type: 'text', text: 'replacement' }] },
{ surfaceOp: { op: 'replace', start: first.seq, end: first.seq } },
)
expect((await ctx.sessionQuery.listEvents(session.id)).map(record => record.surface))
.toEqual(['shadowed', 'log-only', 'current'])
})
it('returns a bounded detached raw-event window and validates the request', async () => {
const ctx = await liveContext({ readWindowMax: 1 })
const session = ctx.sessions.create(SessionId('window'), { meta: { cwd: '/work' } })
for (const text of ['one', 'two', 'three']) {
session.append(
'user/message',
{ content: [{ type: 'text', text }], source: { kind: 'user' } },
{ surfaceOp: 'append' },
)
}
const result = await ctx.sessionQuery.readEvent({ sessionId: session.id, seq: 1, before: 1, after: 1 })
expect([result.startSeq, result.endSeq, result.target.seq]).toEqual([0, 2, 1])
expect(result.session).toEqual(session.header)
Object.assign(result.session, { createdAt: -1 })
if (result.events[0]?.type !== 'user/message') throw new Error('expected user message')
result.events[0].data.content = []
expect(session.header.createdAt).not.toBe(-1)
expect(session.events[0]?.type === 'user/message' && session.events[0].data.content).toHaveLength(1)
await expect(ctx.sessionQuery.readEvent({ sessionId: session.id, seq: 9 }))
.rejects.toThrow(expectCode('SESSION_QUERY_EVENT_NOT_FOUND'))
for (const request of [
{ sessionId: session.id, seq: 0, before: -1 },
{ sessionId: session.id, seq: 0, before: 2 },
{ sessionId: session.id, seq: 0, after: 0.5 },
]) {
await expect(ctx.sessionQuery.readEvent(request)).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_WINDOW'))
}
})
it('merges authoritative persistence with live precedence and detects conflicts', async () => {
const shared = header('shared', 3, { cwd: '/same' })
const durable = header('durable', 2)
TestPersistence.reset([
{ meta: shared, events: eventLog('persisted') },
{ meta: durable, events: eventLog('durable') },
])
const ctx = await liveContext()
const live = ctx.sessions.create(shared.id, { meta: { createdAt: 3, cwd: '/same' } })
live.append(
'user/message',
{ content: [{ type: 'text', text: 'live' }], source: { kind: 'user' } },
{ surfaceOp: 'append' },
)
const persistence = await ctx.plugin(TestPersistence)
expect((await ctx.sessionQuery.listSessions()).map(record => [record.header.id, record.live, record.persisted]))
.toEqual([[shared.id, true, true], [durable.id, false, true]])
const liveRead = await ctx.sessionQuery.readEvent({ sessionId: shared.id, seq: 0 })
expect(liveRead.target.type === 'user/message' && liveRead.target.data.content[0])
.toMatchObject({ text: 'live' })
await expect(ctx.sessionQuery.readEvent({ sessionId: durable.id, seq: 0 }))
.resolves.toMatchObject({ session: durable })
const sharedEntry = TestPersistence.entries.get(shared.id)!
sharedEntry.meta = { ...sharedEntry.meta, cwd: '/conflict' }
await expect(ctx.sessionQuery.listSessions()).rejects.toThrow(expectCode('SESSION_QUERY_SOURCE_CONFLICT'))
await persistence.dispose()
await expect(ctx.sessionQuery.listSessions()).resolves.toEqual([
{ header: shared, live: true, persisted: false },
])
})
it('keeps known live reads independent from persistence health', async () => {
TestPersistence.reset()
const ctx = await liveContext()
const live = ctx.sessions.create(SessionId('live'))
live.append(
'user/message',
{ content: [{ type: 'text', text: 'available' }], source: { kind: 'user' } },
{ surfaceOp: 'append' },
)
await ctx.plugin(TestPersistence)
TestPersistence.listFailure = new Error('list unavailable')
TestPersistence.loadFailure = new Error('load unavailable')
await expect(ctx.sessionQuery.listEvents(live.id)).resolves.toHaveLength(1)
await expect(ctx.sessionQuery.readEvent({ sessionId: live.id, seq: 0 })).resolves.toMatchObject({ target: { seq: 0 } })
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'))
})
it('reports absent sessions, persisted load failures, and persisted header conflicts', async () => {
const durable = header('durable')
TestPersistence.reset([{ meta: durable, events: eventLog() }])
const ctx = await liveContext()
await expect(ctx.sessionQuery.listEvents(SessionId('absent')))
.rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
await ctx.plugin(TestPersistence)
await expect(ctx.sessionQuery.listEvents(SessionId('absent')))
.rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
TestPersistence.loadFailure = 'raw failure'
await expect(ctx.sessionQuery.listEvents(durable.id))
.rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
TestPersistence.loadFailure = undefined
const durableEntry = TestPersistence.entries.get(durable.id)!
durableEntry.meta = { ...durableEntry.meta, cwd: '/changed-after-list' }
TestPersistence.afterList = () => {
const listedEntry = TestPersistence.entries.get(durable.id)!
listedEntry.meta = { ...listedEntry.meta, cwd: '/changed-during-read' }
}
await expect(ctx.sessionQuery.listEvents(durable.id))
.rejects.toThrow(expectCode('SESSION_QUERY_SOURCE_CONFLICT'))
})
it('turns malformed surfaces and direct invalid config into typed errors', async () => {
const ctx = await liveContext()
const session = ctx.sessions.create(SessionId('bad-surface'))
session.append(
'assistant/message',
{ turn: 1, step: 1, content: [] },
{ surfaceOp: { op: 'replace', start: 9, end: 9 } },
)
await expect(ctx.sessionQuery.listEvents(session.id))
.rejects.toThrow(expectCode('SESSION_QUERY_INVALID_SURFACE'))
const persisted = header('bad-persisted-surface')
TestPersistence.reset([{
meta: persisted,
events: [{
type: 'user/message',
seq: 0,
time: 1,
data: { content: [{ type: 'text', text: 'hidden' }], source: { kind: 'user' } },
}],
}])
const persistence = await ctx.plugin(TestPersistence)
await expect(ctx.sessionQuery.listEvents(persisted.id))
.rejects.toThrow(expectCode('SESSION_QUERY_INVALID_SURFACE'))
await persistence.dispose()
const direct = new Context()
await direct.plugin(SessionStore)
expect(new SessionQueryService(direct)).toBeInstanceOf(SessionQueryService)
const invalid = new Context()
await invalid.plugin(SessionStore)
expect(() => new SessionQueryService(invalid, { readWindowMax: -1 }))
.toThrow(expectCode('SESSION_QUERY_INVALID_CONFIG'))
})
it('leaves the optional persistence dependency optional', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const fiber = await ctx.plugin(SessionQueryService)
expect(ctx.sessionQuery).toBeInstanceOf(SessionQueryService)
await fiber.dispose()
expect(ctx.sessionQuery).toBeUndefined()
})
})

View File

@@ -0,0 +1,30 @@
{
"extends": "../../../tsconfig.base.json",
"compilerOptions": {
"rootDir": "src",
"outDir": "lib/types"
},
"include": [
"src"
],
"references": [
{
"path": "../../../vendor/cosmokit"
},
{
"path": "../../../vendor/cordis"
},
{
"path": "../../../vendor/schemastery"
},
{
"path": "../../llm/llm"
},
{
"path": "../../core/session"
},
{
"path": "../../session-persistence/session-persistence"
}
]
}