mirror of
https://github.com/deepseek-ai/deepseek-harness.git
synced 2026-09-09 04:02:35 +00:00
409 lines
15 KiB
TypeScript
409 lines
15 KiB
TypeScript
/** Cold-safe Session list and search projection. */
|
|
|
|
import { stat } from 'node:fs/promises'
|
|
import type { Context } from '@deepseek-ai/cordis'
|
|
import { resolveSessionPreset } from '@deepseek-ai/dsh-agent-presets'
|
|
import type { ImageAttachmentLimits } from '@deepseek-ai/dsh-attachment'
|
|
import type { Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
|
|
import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
|
|
import type {} from '@deepseek-ai/dsh-session-projection'
|
|
import type {} from '@deepseek-ai/dsh-session-projection-cache'
|
|
import { SessionQueryError, type SessionSearchCursor } from '@deepseek-ai/dsh-session-query'
|
|
import { TypertRemoteFailure } from '@deepseek-ai/dsh-typert-protocol'
|
|
import { z } from 'zod'
|
|
import {
|
|
SESSION_SEARCH_RESULT_LIMIT,
|
|
SESSION_SEARCH_SNIPPET_MAX_CODE_POINTS,
|
|
} from './types.ts'
|
|
import type {
|
|
SessionListMetadata, SessionProjectionsBlock, SessionProjectionValues, SessionSearchItem,
|
|
SessionSearchValue, SessionSummary,
|
|
} from './types.ts'
|
|
|
|
/** Default maximum artifact size eligible for one cold blankness read. */
|
|
export const DEFAULT_COLD_BLANK_PROBE_MAX_BYTES = 1024
|
|
|
|
const COLD_SUMMARY_BATCH_SIZE = 16
|
|
const SEARCH_PROVIDER_CALL_LIMIT = 100
|
|
const SESSION_SEARCH_QUERY_MAX_CHARS = 500
|
|
const MESSAGE_TYPES = new Set(['user/message', 'assistant/message'])
|
|
|
|
const sessionListMetadataSchema: z.ZodType<SessionListMetadata> = z.object({
|
|
blank: z.boolean(),
|
|
lastPromptAt: z.number().nullable(),
|
|
})
|
|
|
|
const imageLimitsSchema = z.object({
|
|
maxImageBytes: z.number().int().positive(),
|
|
maxImagesPerMessage: z.number().int().positive(),
|
|
maxMessageImageBytes: z.number().int().positive(),
|
|
maxImagePixels: z.number().int().positive(),
|
|
maxImageDimension: z.number().int().positive(),
|
|
mediaTypes: z.array(z.string()),
|
|
}) as unknown as z.ZodType<ImageAttachmentLimits>
|
|
|
|
/**
|
|
* Advance the Session-list metadata projection by one committed event.
|
|
* @param state - metadata before the event.
|
|
* @param event - next committed Session event.
|
|
* @returns the original or advanced metadata value.
|
|
*/
|
|
export function applySessionListMetadata(
|
|
state: SessionListMetadata,
|
|
event: SessionEvent,
|
|
): SessionListMetadata {
|
|
const blank = state.blank && event.type !== 'turn/start'
|
|
const lastPromptAt = event.type === 'user/message' && event.data.source.kind === 'user'
|
|
? event.time
|
|
: state.lastPromptAt
|
|
return blank === state.blank && lastPromptAt === state.lastPromptAt
|
|
? state
|
|
: { blank, lastPromptAt }
|
|
}
|
|
|
|
/**
|
|
* Fold exact list metadata for an attached Session.
|
|
* @param events - complete attached Session event log.
|
|
* @returns metadata derived from the event prefix.
|
|
*/
|
|
export function sessionListMetadata(events: readonly SessionEvent[]): SessionListMetadata {
|
|
let state: SessionListMetadata = { blank: true, lastPromptAt: null }
|
|
for (const event of events) state = applySessionListMetadata(state, event)
|
|
return state
|
|
}
|
|
|
|
/**
|
|
* Return the longest prefix containing at most `maximum` Unicode code points.
|
|
* @param value - source text.
|
|
* @param maximum - maximum number of Unicode code points.
|
|
* @returns the source text or its longest allowed prefix.
|
|
*/
|
|
export function truncateUnicodeCodePoints(value: string, maximum: number): string {
|
|
let count = 0
|
|
let end = 0
|
|
for (const codePoint of value) {
|
|
if (count === maximum) return value.slice(0, end)
|
|
count++
|
|
end += codePoint.length
|
|
}
|
|
return value
|
|
}
|
|
|
|
/** Owns list projection registration, cold summaries, and authorized search. */
|
|
export class ApiSessionList {
|
|
/**
|
|
* @param ctx - Host context carrying Session, persistence, and projection services.
|
|
* @param coldBlankProbeMaxBytes - maximum physical artifact size read to verify cold blankness.
|
|
*/
|
|
constructor(
|
|
private readonly ctx: Context,
|
|
private readonly coldBlankProbeMaxBytes: number,
|
|
) {
|
|
ctx.inject(['sessionProjections'], (projectionCtx) => {
|
|
projectionCtx.sessionProjections.register<'sessionListMetadata', SessionListMetadata>({
|
|
key: 'sessionListMetadata',
|
|
stateSchema: sessionListMetadataSchema,
|
|
init: () => ({ blank: true, lastPromptAt: null }),
|
|
apply: applySessionListMetadata,
|
|
wire: { viewSchema: sessionListMetadataSchema, view: state => state },
|
|
stateVersion: 1,
|
|
})
|
|
})
|
|
ctx.inject(['sessionProjections', 'attachments'], (projectionCtx) => {
|
|
projectionCtx.sessionProjections.register<'imageLimits', null>({
|
|
key: 'imageLimits',
|
|
stateSchema: z.null(),
|
|
init: () => null,
|
|
apply: state => state,
|
|
wire: {
|
|
viewSchema: imageLimitsSchema,
|
|
view: () => projectionCtx.attachments.imageLimits,
|
|
},
|
|
stateVersion: 1,
|
|
})
|
|
})
|
|
}
|
|
|
|
/**
|
|
* Build one current attached-Session summary.
|
|
* @param session - attached Session to summarize.
|
|
* @returns current list metadata and available projections.
|
|
*/
|
|
summaryFor(session: Session): SessionSummary {
|
|
const metadata = sessionListMetadata(session.events)
|
|
const projections = this.projectionsFor(session.header, session)
|
|
return {
|
|
sessionId: session.id,
|
|
updatedAt: updatedAt(session.header, metadata),
|
|
running: this.ctx.agents.get(session.id)?.status === 'running',
|
|
blank: metadata.blank,
|
|
...listFields(session.header, session.events),
|
|
...(projections === undefined ? {} : { projections }),
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Read every visible attached and persisted Session without activating an Agent.
|
|
* @param signal - optional cancellation for persistence reads.
|
|
* @returns visible Session summaries ordered by activity.
|
|
*/
|
|
async list(signal?: AbortSignal): Promise<SessionSummary[]> {
|
|
signal?.throwIfAborted()
|
|
const items = this.ctx.sessions.list().map(session => this.summaryFor(session))
|
|
const attached = new Set(items.map(item => item.sessionId))
|
|
const persistence = this.ctx.get('sessionPersistence')
|
|
if (persistence !== undefined) {
|
|
const cold = (await persistence.list(signal))
|
|
.filter(meta => !attached.has(meta.id) && meta.cwd !== undefined)
|
|
signal?.throwIfAborted()
|
|
for (let offset = 0; offset < cold.length; offset += COLD_SUMMARY_BATCH_SIZE) {
|
|
const settled = await Promise.allSettled(cold.slice(offset, offset + COLD_SUMMARY_BATCH_SIZE)
|
|
.map(async (meta) => {
|
|
const projections = this.projectionsFor(meta, undefined)
|
|
const summary = await summarizeCold(
|
|
this.ctx,
|
|
persistence,
|
|
meta,
|
|
projections?.values.sessionListMetadata,
|
|
this.coldBlankProbeMaxBytes,
|
|
signal,
|
|
)
|
|
const raced = this.ctx.sessions.get(meta.id)
|
|
if (raced !== undefined) return this.summaryFor(raced)
|
|
return { ...summary, ...(projections === undefined ? {} : { projections }) }
|
|
}))
|
|
const summaries = settled.map((result) => {
|
|
if (result.status === 'rejected') throw result.reason
|
|
return result.value
|
|
})
|
|
signal?.throwIfAborted()
|
|
items.push(...summaries)
|
|
}
|
|
}
|
|
items.sort((left, right) => right.updatedAt - left.updatedAt)
|
|
return items
|
|
}
|
|
|
|
/**
|
|
* Search current visible message content without activating any matching Session.
|
|
* @param query - literal message-content query.
|
|
* @param signal - cancellation for list and search reads.
|
|
* @returns authorized bounded Session search results.
|
|
*/
|
|
async search(query: string, signal: AbortSignal): Promise<SessionSearchValue> {
|
|
const normalizedQuery = normalizeSearchQuery(query)
|
|
signal.throwIfAborted()
|
|
const provider = this.ctx.get('sessionQuery')
|
|
if (provider === undefined) {
|
|
reject(
|
|
'internal',
|
|
'session search is unavailable: this deployment does not mount @deepseek-ai/dsh-session-query',
|
|
{},
|
|
)
|
|
}
|
|
try {
|
|
const visible = await this.list(signal)
|
|
signal.throwIfAborted()
|
|
if (visible.length === 0) return { items: [], hasMore: false }
|
|
const visibleIds = new Set(visible.map(item => item.sessionId))
|
|
const authorized: SessionSearchItem[] = []
|
|
const acceptedIds = new Set<SessionId>()
|
|
const seenCursors = new Set<SessionSearchCursor>()
|
|
let cursor: SessionSearchCursor | undefined
|
|
let providerCalls = 0
|
|
let pageLimit = SESSION_SEARCH_RESULT_LIMIT
|
|
while (authorized.length <= SESSION_SEARCH_RESULT_LIMIT) {
|
|
signal.throwIfAborted()
|
|
if (providerCalls >= SEARCH_PROVIDER_CALL_LIMIT) {
|
|
throw new Error(`session search provider exceeded the ${SEARCH_PROVIDER_CALL_LIMIT}-call work budget`)
|
|
}
|
|
providerCalls++
|
|
const requestedCursor = cursor
|
|
const requestedLimit = pageLimit
|
|
let page
|
|
try {
|
|
page = await provider.searchSessions({
|
|
query: normalizedQuery,
|
|
eventFilters: [
|
|
{ kind: 'type', values: ['user/message', 'assistant/message'] },
|
|
{ kind: 'surface', values: ['current'] },
|
|
],
|
|
limit: requestedLimit,
|
|
...(requestedCursor === undefined ? {} : { cursor: requestedCursor }),
|
|
}, { signal })
|
|
signal.throwIfAborted()
|
|
} catch (error: unknown) {
|
|
signal.throwIfAborted()
|
|
if (requestedCursor === undefined
|
|
&& error instanceof SessionQueryError
|
|
&& error.code === 'SESSION_QUERY_INVALID_LIMIT'
|
|
&& requestedLimit > 1) {
|
|
pageLimit = Math.max(1, Math.floor(requestedLimit / 2))
|
|
continue
|
|
}
|
|
if (requestedCursor !== undefined
|
|
&& error instanceof SessionQueryError
|
|
&& error.code === 'SESSION_QUERY_STALE_CURSOR') {
|
|
authorized.length = 0
|
|
acceptedIds.clear()
|
|
seenCursors.clear()
|
|
cursor = undefined
|
|
continue
|
|
}
|
|
throw error
|
|
}
|
|
if (page.items.length > requestedLimit) {
|
|
throw new Error(`session search provider returned ${String(page.items.length)} items; maximum is ${String(requestedLimit)}`)
|
|
}
|
|
for (const hit of page.items) {
|
|
if (authorized.length > SESSION_SEARCH_RESULT_LIMIT) continue
|
|
if (!visibleIds.has(hit.header.id)
|
|
|| hit.bestMatch.sessionId !== hit.header.id
|
|
|| hit.bestMatch.surface !== 'current'
|
|
|| !MESSAGE_TYPES.has(hit.bestMatch.type)
|
|
|| acceptedIds.has(hit.header.id)) continue
|
|
acceptedIds.add(hit.header.id)
|
|
authorized.push({
|
|
sessionId: hit.header.id,
|
|
snippet: truncateUnicodeCodePoints(hit.bestMatch.snippet, SESSION_SEARCH_SNIPPET_MAX_CODE_POINTS),
|
|
})
|
|
}
|
|
if (page.nextCursor !== undefined) {
|
|
if (seenCursors.has(page.nextCursor)) {
|
|
throw new Error('session search provider repeated a continuation cursor')
|
|
}
|
|
seenCursors.add(page.nextCursor)
|
|
}
|
|
if (authorized.length > SESSION_SEARCH_RESULT_LIMIT || page.nextCursor === undefined) break
|
|
cursor = page.nextCursor
|
|
}
|
|
return {
|
|
items: authorized.slice(0, SESSION_SEARCH_RESULT_LIMIT),
|
|
hasMore: authorized.length > SESSION_SEARCH_RESULT_LIMIT,
|
|
}
|
|
} catch (error: unknown) {
|
|
signal.throwIfAborted()
|
|
if (error instanceof SessionQueryError && error.code === 'SESSION_QUERY_ABORTED') {
|
|
reject('cancelled', 'session search was aborted', {})
|
|
}
|
|
reject('internal', `session search failed: ${String(error)}`, {})
|
|
}
|
|
}
|
|
|
|
private projectionsFor(
|
|
header: SessionHeader,
|
|
session: Session | undefined,
|
|
): SessionProjectionsBlock | undefined {
|
|
try {
|
|
const block = session === undefined
|
|
? this.ctx.get('sessionProjectionCache')?.cachedSnapshot(header)
|
|
: this.ctx.get('sessionProjections')?.snapshot(session)
|
|
return block !== undefined && Object.keys(block.values).length > 0
|
|
? {
|
|
asOfSeq: block.asOfSeq,
|
|
// Projection definitions validate whole JSON values before snapshot publication.
|
|
values: block.values as SessionProjectionValues,
|
|
}
|
|
: undefined
|
|
} catch (error) {
|
|
this.ctx.logger.warn(
|
|
`api-session.list: projection column for "${header.id}" failed; serving the row without it: ${String(error)}`,
|
|
)
|
|
return undefined
|
|
}
|
|
}
|
|
}
|
|
|
|
function normalizeSearchQuery(query: string): string {
|
|
const normalized = query.trim()
|
|
if (normalized.length === 0) {
|
|
reject('bad-request', 'session search query must not be empty', {})
|
|
}
|
|
if (normalized.length > SESSION_SEARCH_QUERY_MAX_CHARS) {
|
|
reject(
|
|
'bad-request',
|
|
`session search query must contain at most ${SESSION_SEARCH_QUERY_MAX_CHARS} UTF-16 code units`,
|
|
{},
|
|
)
|
|
}
|
|
if (normalized.includes('\0')) {
|
|
reject('bad-request', 'session search query must not contain NUL', {})
|
|
}
|
|
return normalized
|
|
}
|
|
|
|
function reject(code: string, message: string, details: object): never {
|
|
throw new TypertRemoteFailure({ code, message, details })
|
|
}
|
|
|
|
function updatedAt(header: SessionHeader, metadata: SessionListMetadata | undefined): number {
|
|
return Math.max(header.createdAt, metadata?.lastPromptAt ?? 0)
|
|
}
|
|
|
|
function listFields(header: SessionHeader, events: readonly SessionEvent[] = []): {
|
|
readonly parentSessionId?: SessionId
|
|
readonly origin?: 'subagent'
|
|
readonly cwd?: string
|
|
readonly agentPreset?: string
|
|
} {
|
|
const agentPreset = resolveSessionPreset({ header, events })
|
|
return {
|
|
...(header.parentSession === undefined ? {} : { parentSessionId: header.parentSession }),
|
|
...(header.origin === undefined ? {} : { origin: header.origin }),
|
|
...(header.cwd === undefined ? {} : { cwd: header.cwd }),
|
|
...(agentPreset === undefined ? {} : { agentPreset }),
|
|
}
|
|
}
|
|
|
|
async function summarizeCold(
|
|
ctx: Context,
|
|
persistence: SessionPersistence,
|
|
header: SessionHeader,
|
|
metadata: SessionListMetadata | undefined,
|
|
blankProbeMaxBytes: number,
|
|
signal?: AbortSignal,
|
|
): Promise<SessionSummary> {
|
|
const probed = metadata?.blank === false
|
|
? undefined
|
|
: await probeColdMetadata(ctx, persistence, header, blankProbeMaxBytes, signal)
|
|
return {
|
|
sessionId: header.id,
|
|
updatedAt: updatedAt(header, probed ?? metadata),
|
|
running: false,
|
|
blank: metadata?.blank === false ? false : probed?.blank ?? false,
|
|
...listFields(header),
|
|
}
|
|
}
|
|
|
|
async function probeColdMetadata(
|
|
ctx: Context,
|
|
persistence: SessionPersistence,
|
|
header: SessionHeader,
|
|
maxBytes: number,
|
|
signal?: AbortSignal,
|
|
): Promise<SessionListMetadata | undefined> {
|
|
if (maxBytes === 0) return undefined
|
|
signal?.throwIfAborted()
|
|
const location = persistence.locate(header)
|
|
if (location === undefined) return undefined
|
|
let size: number
|
|
try {
|
|
size = (await stat(location.path)).size
|
|
} catch {
|
|
signal?.throwIfAborted()
|
|
return undefined
|
|
}
|
|
if (size > maxBytes) return undefined
|
|
try {
|
|
const { events } = await persistence.readFrom(header.id, 0, signal)
|
|
signal?.throwIfAborted()
|
|
return sessionListMetadata(events)
|
|
} catch (error) {
|
|
signal?.throwIfAborted()
|
|
ctx.logger.warn(
|
|
`api-session.list: blank probe for "${header.id}" failed; serving it as visible: ${String(error)}`,
|
|
)
|
|
return undefined
|
|
}
|
|
}
|