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

fix(session): harden cross-session references

Browse Source
This commit is contained in:
Yichen Jiang
2026-07-21 17:53:30 +08:00
parent 8394898ef5
commit ebb62c482c
19 changed files with 213 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
packages/context/session-reference/src/index.ts
View File

@@ -102,15 +102,22 @@ export class SessionReferenceService extends Service {
* @param agent - target agent; self is excluded and its cwd drives ranking.
* @param query - optional case-insensitive session-id/cwd substring.
* @param limit - optional positive result cap.
* @param signal - optional cancellation boundary for host autocomplete teardown.
* @returns candidate records in stable source creation order within each rank.
*/
async listCandidates(agent: Agent, query = '', limit = this.config.candidateLimit): Promise<SessionReferenceCandidate[]> {
async listCandidates(
agent: Agent,
query = '',
limit = this.config.candidateLimit,
signal?: AbortSignal,
): Promise<SessionReferenceCandidate[]> {
if (!Number.isSafeInteger(limit) || limit <= 0) {
throw new SessionReferenceError('candidate limit must be a positive safe integer', 'SESSION_REFERENCE_INVALID_REFERENCE')
}
const needle = query.toLocaleLowerCase()
const targetCwd = agent.session.header.cwd
const records = (await this.ctx.sessionQuery.listSessions())
assertNotCancelled(signal)
const records = (await settleWithCancellation(this.ctx.sessionQuery.listSessions(), signal))
.filter(record => record.header.id !== agent.id)
.filter((record) => {
if (needle === '') return true
@@ -149,10 +156,13 @@ export class SessionReferenceService extends Service {
assertNotCancelled(signal)
let prepared: PreparedSource[]
try {
prepared = await Promise.all(inputs.map(async input => ({
input,
snapshot: await this.ctx.sessionQuery.readSurface(input.sessionId),
})))
prepared = await settleWithCancellation(
Promise.all(inputs.map(async input => ({
input,
snapshot: await this.ctx.sessionQuery.readSurface(input.sessionId),
}))),
signal,
)
} catch (error: unknown) {
if (signal?.aborted === true) throw cancelled(signal)
throw new SessionReferenceError(
@@ -258,6 +268,25 @@ function assertNotCancelled(signal: AbortSignal | undefined): void {
if (signal?.aborted === true) throw cancelled(signal)
}
function settleWithCancellation<T>(work: Promise<T>, signal: AbortSignal | undefined): Promise<T> {
if (signal === undefined) return work
return new Promise<T>((resolve, reject) => {
const onAbort = (): void => { reject(cancelled(signal)) }
signal.addEventListener('abort', onAbort, { once: true })
void work.then(
(value) => {
signal.removeEventListener('abort', onAbort)
resolve(value)
},
(error: unknown) => {
signal.removeEventListener('abort', onAbort)
reject(error instanceof Error ? error : new Error(String(error)))
},
)
if (signal.aborted) onAbort()
})
}
function cancelled(signal: AbortSignal): SessionReferenceError {
return new SessionReferenceError('session reference preparation was cancelled', 'SESSION_REFERENCE_CANCELLED', { cause: signal.reason })
}

33
packages/context/session-reference/tests/session-reference.spec.ts
View File

@@ -190,6 +190,21 @@ describe('session reference discovery and preparation', () => {
])
await expect(ctx.sessionReferences.listCandidates(fakeAgent(target), '', 0))
.rejects.toThrow(expectCode('SESSION_REFERENCE_INVALID_REFERENCE'))
let releaseList: (() => void) | undefined
const listSessions = vi.spyOn(ctx.sessionQuery, 'listSessions').mockImplementationOnce(async () => {
await new Promise<void>((resolve) => { releaseList = resolve })
return []
})
const controller = new AbortController()
const pending = ctx.sessionReferences.listCandidates(fakeAgent(target), '', undefined, controller.signal)
await vi.waitFor(() => { expect(releaseList).toBeTypeOf('function') })
const cancelledList = expect(pending).rejects.toThrow(expectCode('SESSION_REFERENCE_CANCELLED'))
controller.abort('autocomplete superseded')
await cancelledList
releaseList?.()
await Promise.resolve()
listSessions.mockRestore()
})
it('projects only the current user/assistant surface and records snapshot metadata', async () => {
@@ -309,6 +324,9 @@ describe('session reference discovery and preparation', () => {
readSurface.mockRejectedValueOnce('non-error read failure')
await expect(ctx.sessionReferences.prepare(agent, content, [{ sessionId: one.id }]))
.rejects.toThrow(/non-error read failure/)
readSurface.mockRejectedValueOnce('non-error signalled read failure')
await expect(ctx.sessionReferences.prepare(agent, content, [{ sessionId: one.id }], new AbortController().signal))
.rejects.toThrow(/non-error signalled read failure/)
const duringRead = new AbortController()
readSurface.mockImplementationOnce(async () => {
@@ -317,6 +335,21 @@ describe('session reference discovery and preparation', () => {
})
await expect(ctx.sessionReferences.prepare(agent, content, [{ sessionId: one.id }], duringRead.signal))
.rejects.toThrow(expectCode('SESSION_REFERENCE_CANCELLED'))
const snapshot = await ctx.sessionQuery.readSurface(one.id)
let releaseRead: (() => void) | undefined
readSurface.mockImplementationOnce(async () => {
await new Promise<void>((resolve) => { releaseRead = resolve })
return snapshot
})
const hangingRead = new AbortController()
const pending = ctx.sessionReferences.prepare(agent, content, [{ sessionId: one.id }], hangingRead.signal)
await vi.waitFor(() => { expect(releaseRead).toBeTypeOf('function') })
const cancelledRead = expect(pending).rejects.toThrow(expectCode('SESSION_REFERENCE_CANCELLED'))
hangingRead.abort('cancelled while storage remained pending')
await cancelledRead
releaseRead?.()
await Promise.resolve()
readSurface.mockRestore()
const abort = new AbortController()

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

@@ -434,8 +434,8 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
summary: 'Exact-read consumer that prepares immutable cross-session message context.',
methods: [
{
signature: 'async listCandidates(agent: Agent, query = \'\', limit = this.config.candidateLimit): Promise<SessionReferenceCandidate[]>',
jsDoc: '/**\n * List metadata-only reference candidates, ranked by working-directory affinity.\n * @param agent - target agent; self is excluded and its cwd drives ranking.\n * @param query - optional case-insensitive session-id/cwd substring.\n * @param limit - optional positive result cap.\n * @returns candidate records in stable source creation order within each rank.\n */',
signature: 'async listCandidates( agent: Agent, query = \'\', limit = this.config.candidateLimit, signal?: AbortSignal, ): Promise<SessionReferenceCandidate[]>',
jsDoc: '/**\n * List metadata-only reference candidates, ranked by working-directory affinity.\n * @param agent - target agent; self is excluded and its cwd drives ranking.\n * @param query - optional case-insensitive session-id/cwd substring.\n * @param limit - optional positive result cap.\n * @param signal - optional cancellation boundary for host autocomplete teardown.\n * @returns candidate records in stable source creation order within each rank.\n */',
},
{
signature: 'async prepare( agent: Agent, content: ContentBlock[], references: SessionReferenceInput[], signal?: AbortSignal, ): Promise<PreparedReferencedMessage>',
@@ -1352,7 +1352,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [
},
{
name: 'InjectOptions',
declaration: 'export interface InjectOptions extends SendOptions {\n meta?: JsonValue;\n}',
declaration: 'export interface InjectOptions extends Omit<SendOptions, \'contexts\'> {\n meta?: JsonValue;\n}',
},
{
name: 'JsonValue',

2
packages/core/agent/src/types.ts
View File

@@ -40,7 +40,7 @@ export interface SendOptions {
}
/** Options specific to durable synthetic context injection. */
export interface InjectOptions extends SendOptions {
export interface InjectOptions extends Omit<SendOptions, 'contexts'> {
/** Opaque JSON state retained in the session event but hidden from the model. */
meta?: JsonValue
}

11
packages/examples/acp-demo/src/index.ts
View File

@@ -23,7 +23,7 @@ import SessionPersistenceJsonl, {
} from '@deepseek-ai/dsh-session-persistence-jsonl'
import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
import SessionQueryService from '@deepseek-ai/dsh-session-query'
import SessionReferenceService from '@deepseek-ai/dsh-session-reference'
import SessionReferenceService, { type Config as SessionReferenceConfig } from '@deepseek-ai/dsh-session-reference'
export const name = 'acp-demo'
const DEFAULT_PERSISTENCE_ROOT = './.sessions'
@@ -56,6 +56,8 @@ export interface Config {
persistenceRoot?: string
/** JSONL artifact encoding; defaults to checksummed Zstandard frames. */
persistenceCompression?: JsonlCompression
/** Cross-session reference discovery and snapshot byte budgets. */
sessionReferences?: SessionReferenceConfig
/** Controls automatic AGENTS.md/CLAUDE.md loading; configure a byte budget or set `false`. */
workspaceContext: agentCore.Config['workspaceContext']
/** Skill registry, local-provider, and model-facing consumer config forwarded to agent-spine-demo. */
@@ -86,6 +88,7 @@ export const Config: z<Config> = z.object({
dshHome: z.string(),
persistenceRoot: z.string().default(DEFAULT_PERSISTENCE_ROOT),
persistenceCompression: JsonlCompressionSchema,
sessionReferences: SessionReferenceService.Config,
workspaceContext: z.union([z.const(false), workspaceContext.Config]).required(),
skills: agentCore.SkillConfigSchema,
toolBash: agentCore.ToolBashConfigSchema,
@@ -107,12 +110,16 @@ export function apply(ctx: Context, config: Config): void {
ctx.plugin(CommandService)
if (goals !== false) ctx.plugin(commandGoal)
ctx.plugin(agentCore, { ...agentCore.pickSpineConfig(config), goals })
// This front door owns the same persistence/reference cluster as the TUI;
// extracting these few calls would introduce a shared app-composition facade.
/* jscpd:ignore-start */
ctx.plugin(UserInteractionService)
ctx.plugin(SessionPersistenceJsonl, {
root: config.persistenceRoot ?? DEFAULT_PERSISTENCE_ROOT,
...(config.persistenceCompression === undefined ? {} : { compression: config.persistenceCompression }),
})
ctx.plugin(SessionQueryService)
ctx.plugin(SessionReferenceService)
ctx.plugin(SessionReferenceService, config.sessionReferences ?? {})
/* jscpd:ignore-end */
ctx.plugin(acp, { provider: config.provider, model: config.model })
}

7
packages/examples/acp-demo/tests/acp-agent.spec.ts
View File

@@ -5,6 +5,7 @@ import { tmpdir } from 'node:os'
import { Context } from 'cordis'
import Loader from '@cordisjs/plugin-loader'
import { agentEvents, type Agent } from '@deepseek-ai/dsh-agent'
import { SessionId } from '@deepseek-ai/dsh-session'
import { TOOL_ORDER_REST } from '@deepseek-ai/dsh-system-prompt'
import type { Message } from '@deepseek-ai/dsh-llm'
import * as acpAgent from '../src/index.ts'
@@ -76,6 +77,7 @@ describe('dsh-acp-demo composition', () => {
persona: 'hi',
persistenceRoot: '/tmp/dsh-acp-demo-test',
persistenceCompression: 'none',
sessionReferences: { candidateLimit: 1 },
skills: await isolatedSkillsConfig(),
workspaceContext: false,
})
@@ -90,6 +92,11 @@ describe('dsh-acp-demo composition', () => {
expect(ctx.get('tools')?.get('ask_user_question')).toBeUndefined()
expect(ctx.get('goals')).toBeDefined()
expect(ctx.get('tools')?.get('get_goal')).toBeDefined()
const target = ctx.sessions.create(SessionId('candidate-target'))
ctx.sessions.create(SessionId('candidate-one'))
ctx.sessions.create(SessionId('candidate-two'))
await expect(ctx.sessionReferences.listCandidates({ id: target.id, session: target } as Agent))
.resolves.toHaveLength(1)
// No pre-created agents — ACP session/new creates them on demand.
expect(ctx.get('agents')!.list()).toHaveLength(0)
await ctx.fiber.dispose()

7
packages/examples/tui-demo/src/index.ts
View File

@@ -23,7 +23,7 @@ import SessionPersistenceJsonl, {
} from '@deepseek-ai/dsh-session-persistence-jsonl'
import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
import SessionQueryService from '@deepseek-ai/dsh-session-query'
import SessionReferenceService from '@deepseek-ai/dsh-session-reference'
import SessionReferenceService, { type Config as SessionReferenceConfig } from '@deepseek-ai/dsh-session-reference'
import * as toolAskUser from '@deepseek-ai/dsh-tool-ask-user'
import * as uiTui from '@deepseek-ai/dsh-tui'
@@ -51,6 +51,8 @@ export interface Config {
persistenceRoot?: string
/** JSONL artifact encoding; defaults to checksummed Zstandard frames. */
persistenceCompression?: JsonlCompression
/** Cross-session reference discovery and snapshot byte budgets. */
sessionReferences?: SessionReferenceConfig
/** TUI subtitle rendered on start. Defaults to `ready.`. */
welcome?: string
/** Full-screen TUI presentation settings. */
@@ -83,6 +85,7 @@ export const Config: z<Config> = z.object({
dshHome: z.string(),
persistenceRoot: z.string().default(DEFAULT_PERSISTENCE_ROOT),
persistenceCompression: JsonlCompressionSchema,
sessionReferences: SessionReferenceService.Config,
welcome: z.string().default(DEFAULT_WELCOME),
ui: uiTui.TuiConfigSchema,
skills: agentCore.SkillConfigSchema,
@@ -112,7 +115,7 @@ export function composeTuiApp(ctx: Context, config: Config): void {
...(config.persistenceCompression === undefined ? {} : { compression: config.persistenceCompression }),
})
ctx.plugin(SessionQueryService)
ctx.plugin(SessionReferenceService)
ctx.plugin(SessionReferenceService, config.sessionReferences ?? {})
ctx.plugin(UserInteractionService)
ctx.plugin(uiTui, {
...config.ui,

13
packages/examples/tui-demo/tests/tui-agent.spec.ts
View File

@@ -32,6 +32,12 @@ describe('dsh-tui-demo app', () => {
dshHome: '/tmp/dsh-home',
persistenceRoot: '/tmp/tui-sessions',
persistenceCompression: 'none',
sessionReferences: {
maxReferences: 2,
candidateLimit: 7,
maxReferenceBytes: 1234,
maxTotalBytes: 2345,
},
welcome: 'TUI ready',
ui: { color: false, maxToolOutputLines: 3 },
skills: { tool: { catalogDescriptionMaxLength: 8 } },
@@ -53,6 +59,12 @@ describe('dsh-tui-demo app', () => {
])
expect(calls[0]?.config).toBeUndefined()
expect(calls[2]?.config).toEqual({ root: '/tmp/tui-sessions', compression: 'none' })
expect(calls[4]?.config).toEqual({
maxReferences: 2,
candidateLimit: 7,
maxReferenceBytes: 1234,
maxTotalBytes: 2345,
})
const tuiConfig = calls[6]?.config as { sessionId: string }
expect(tuiConfig).toMatchObject({ welcome: 'TUI ready', color: false, maxToolOutputLines: 3 })
expect(tuiConfig.sessionId).toMatch(/^main-session-[0-9a-f-]{36}$/)
@@ -90,6 +102,7 @@ describe('dsh-tui-demo app', () => {
})
expect(calls[2]?.config).toEqual({ root: './.sessions' })
expect(calls[4]?.config).toEqual({})
expect(calls[6]?.config).toEqual({ welcome: 'ready.', sessionId: 'persisted-session' })
expect((calls[7]?.config as { agents: Array<Record<string, unknown>> }).agents[0]).toMatchObject({
id: 'main',

20
packages/ui/acp/tests/bridge.spec.ts
View File

@@ -397,25 +397,25 @@ describe('acp bridge', () => {
it('cancels reference preparation before a turn is created', async () => {
harness = await makeBridgeHarness({ storageDir, withSessionReferences: true, script: [] })
const source = harness.ctx.sessions.create(SessionId('source'))
const snapshot = await harness.ctx.sessionQuery.readSurface(source.id)
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
const { sessionId } = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })
const prepare = vi.spyOn(harness.ctx.sessionReferences, 'prepare').mockImplementation(
(_agent, _content, _references, signal) => new Promise((_resolve, reject) => {
if (signal?.aborted === true) {
reject(new Error('already aborted'))
return
}
signal?.addEventListener('abort', () => { reject(new Error('aborted')) }, { once: true })
}),
)
let releaseRead: (() => void) | undefined
const readSurface = vi.spyOn(harness.ctx.sessionQuery, 'readSurface').mockImplementationOnce(async () => {
await new Promise<void>((resolve) => { releaseRead = resolve })
return snapshot
})
const pending = harness.client.prompt({
sessionId,
prompt: [{ type: 'resource_link', uri: encodeSessionReferenceUri(source.id), name: 'source' }],
})
await vi.waitFor(() => { expect(prepare).toHaveBeenCalledOnce() })
await vi.waitFor(() => { expect(releaseRead).toBeTypeOf('function') })
await harness.client.cancel({ sessionId })
await expect(pending).resolves.toEqual({ stopReason: 'cancelled' })
expect(harness.ctx.agents.get(SessionId(sessionId))?.session.events).toHaveLength(0)
releaseRead?.()
await Promise.resolve()
readSurface.mockRestore()
})
it('rejects a prompt for an unknown session', async () => {

20
packages/ui/tui/src/index.ts
View File

@@ -202,6 +202,11 @@ function displayText(text: string): string {
`\\x${control.charCodeAt(0).toString(16).padStart(2, '0')}`)
}
/** Escape external controls for terminal fields that must remain on one line. */
function displayInlineText(text: string): string {
return displayText(text).replaceAll('\n', '\\x0a')
}
/**
* Theme-agnostic palette built from the standard 16-color ANSI set plus SGR
* attributes, which every terminal remaps to its active color scheme. Body
@@ -827,17 +832,20 @@ class SessionAutocompleteProvider implements AutocompleteProvider {
if (token === undefined) return basePromise
let candidates
try {
candidates = await this.sessions.listCandidates(this.agent, token.slice(1))
candidates = await this.sessions.listCandidates(this.agent, token.slice(1), undefined, options.signal)
} catch {
return basePromise
}
const base = await basePromise
if (options.signal.aborted) return base
const items: AutocompleteItem[] = candidates.map(candidate => ({
value: formatSessionReferenceMention({ sessionId: candidate.sessionId, label: candidate.label }),
label: `Session · ${candidate.sessionId}`,
description: `${candidate.cwd ?? '(no cwd)'} · ${new Date(candidate.createdAt).toISOString()}`,
}))
const items: AutocompleteItem[] = candidates.map((candidate) => {
const mentionLabel = displayInlineText(candidate.label)
return {
value: formatSessionReferenceMention({ sessionId: candidate.sessionId, label: mentionLabel }),
label: `Session · ${displayInlineText(candidate.sessionId)}`,
description: `${candidate.cwd === undefined ? '(no cwd)' : displayInlineText(candidate.cwd)} · ${new Date(candidate.createdAt).toISOString()}`,
}
})
if (items.length === 0) return base
return { items: [...items, ...(base?.items ?? [])], prefix: token }
}

65
packages/ui/tui/tests/tui.spec.ts
View File

@@ -2,7 +2,7 @@ import { homedir } from 'node:os'
import { join } from 'node:path'
import { describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import type { Terminal } from '@earendil-works/pi-tui'
import { CombinedAutocompleteProvider, type Terminal } from '@earendil-works/pi-tui'
import AgentRegistry, { type Agent } from '@deepseek-ai/dsh-agent'
import CommandService, { type CommandInvocation } from '@deepseek-ai/dsh-commands'
import SessionStore, { SessionId, type JsonValue } from '@deepseek-ai/dsh-session'
@@ -545,6 +545,40 @@ describe('pi-tui chat lifecycle and transcript', () => {
await dispose(result)
})
it('escapes session autocomplete metadata while preserving the referenced session id', async () => {
const unsafeId = SessionId('evil\x1b\x07\u009b\ns')
const unsafeCwd = '/x/\x1b\x07\u009b\nf'
const result = await setup({
async configureContext(ctx) {
ctx.provide('tools', { get: () => undefined } as never)
await ctx.plugin(SessionQueryService)
await ctx.plugin(SessionReferenceService)
const source = ctx.sessions.create(unsafeId, { meta: { cwd: unsafeCwd, createdAt: 1 } })
appendUser(source, 'safe background')
},
})
result.terminal.send('@evil')
await vi.waitFor(() => {
expect(result.terminal.output).toContain('Session · evil\\x1b\\x07\\x9b\\x0a')
})
expect(result.terminal.output).toContain('/x/\\x1b\\x07\\x9b\\x0af')
expect(result.terminal.output).not.toContain('evil\x1b\x07')
expect(result.terminal.output).not.toContain('/x/\x1b\x07')
result.terminal.send('\t')
await tick()
result.terminal.send('\r')
await vi.waitFor(() => { expect(result.agent.sent).toHaveLength(1) })
expect(result.agent.sent).toEqual([[
{ type: 'text', text: '@evil\\x1b\\x07\\x9b\\x0as' },
]])
expect(result.agent.sentOptions[0]?.contexts).toMatchObject([{
meta: { references: [{ sessionId: unsafeId }] },
}])
await dispose(result)
})
it('falls back cleanly for non-session, empty, failed, and superseded autocomplete requests', async () => {
const result = await setup({
async configureContext(ctx) {
@@ -575,19 +609,38 @@ describe('pi-tui chat lifecycle and transcript', () => {
await tick()
result.terminal.send('\x03')
let releaseFirst: (() => void) | undefined
let releaseBase: (() => void) | undefined
const baseSuggestions = vi.spyOn(CombinedAutocompleteProvider.prototype, 'getSuggestions')
.mockImplementationOnce(async () => {
await new Promise<void>((resolve) => { releaseBase = resolve })
return null
})
listCandidates.mockResolvedValueOnce([])
result.terminal.send('@base-slow')
await vi.waitFor(() => { expect(releaseBase).toBeTypeOf('function') })
const baseWaitSignal = listCandidates.mock.calls.at(-1)?.[3]
result.terminal.send('x')
await vi.waitFor(() => { expect(baseWaitSignal?.aborted).toBe(true) })
releaseBase?.()
await tick()
baseSuggestions.mockRestore()
let delayedSignal: AbortSignal | undefined
let delayed = true
listCandidates.mockImplementation(async (...args) => {
if (!delayed) return originalListCandidates(...args)
delayed = false
await new Promise<void>((resolve) => { releaseFirst = resolve })
delayedSignal = args[3]
if (delayedSignal === undefined) throw new Error('expected autocomplete cancellation signal')
await new Promise<void>((_resolve, reject) => {
delayedSignal?.addEventListener('abort', () => { reject(new Error('superseded')) }, { once: true })
})
return []
})
result.terminal.send('@slow')
await vi.waitFor(() => { expect(releaseFirst).toBeTypeOf('function') })
await vi.waitFor(() => { expect(delayedSignal).toBeDefined() })
result.terminal.send('x')
releaseFirst?.()
await tick()
await vi.waitFor(() => { expect(delayedSignal?.aborted).toBe(true) })
await dispose(result)
})