llm.registerAdapter, agents.register, sessions.create, systemPrompt.section, systemPrompt.tools, and tools.register each mutated state, emitted a change event, then returned the disposer. In Cordis a synchronous throw before the effect returns its disposer leaves nothing for the fiber to collect, so a throwing change-listener leaked the registry entry permanently — HMR/dispose could not clean it, and the duplicate-name/already-exists check stayed wedged until restart. Convert each to the generator-effect pattern already proven in AgentLoop.create: mutate state, `yield` the disposer that undoes it (collected before the next step runs, so it is torn down if a later step throws), THEN emit the change event. The existing duplicate-name throws are unchanged — they fire before any mutation, so they correctly leak nothing. No public API change: generator effects are still synchronous SyncEffects and register() keeps returning its fire-and-forget disposer wrapper. Tests: a listener-throw rollback test for all six methods — register with a change-listener that throws, assert the call throws AND the registry is clean (entry absent; a subsequent listener-free register of the same name succeeds and contributes exactly once). For systemPrompt (no duplicate-name check) the two tests assert assembly is clean. Verified each fails against the pre-fix emit-before-return-disposer form.
159 lines
7.0 KiB
TypeScript
159 lines
7.0 KiB
TypeScript
import { describe, expect, it } from 'vitest'
|
|
import { Context } from 'cordis'
|
|
import { CallId } from '@deepseek-ai/dsh-llm'
|
|
import SessionStore, { Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
|
|
|
|
describe('Session', () => {
|
|
it('derives message history from the event log', () => {
|
|
const session = new Session(SessionId('s1'))
|
|
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
|
session.append('user/message', { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } })
|
|
session.append('assistant/chunk', { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'hi' } })
|
|
session.append('assistant/message', {
|
|
turn: 1, step: 1,
|
|
content: [
|
|
{ type: 'text', text: 'let me check' },
|
|
{ type: 'tool-call', id: CallId('c1'), name: 'echo', arguments: '{}' },
|
|
],
|
|
})
|
|
session.append('tool/result', { turn: 1, step: 1, callId: CallId('c1'), content: [{ type: 'text', text: 'ok' }], isError: false })
|
|
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
|
|
|
const messages = session.deriveMessages()
|
|
expect(messages.map(m => m.role)).toEqual(['user', 'assistant', 'user'])
|
|
// raw chunks must NOT appear in derived history
|
|
expect(messages[1]!.content).toHaveLength(2)
|
|
expect(messages[2]!.content[0]).toMatchObject({ type: 'tool-result', toolCallId: CallId('c1') })
|
|
})
|
|
|
|
it('renders context and steering messages as tagged synthetic user content', () => {
|
|
const session = new Session(SessionId('s2'))
|
|
session.append('context/message', {
|
|
content: [{ type: 'text', text: 'file changed: a.ts' }],
|
|
source: { kind: 'plugin', plugin: 'watcher' },
|
|
})
|
|
session.append('steering/message', {
|
|
turn: 1,
|
|
content: [{ type: 'text', text: 'focus on tests' }],
|
|
source: { kind: 'user' },
|
|
})
|
|
|
|
const [contextMessage, steeringMessage] = session.deriveMessages()
|
|
expect(contextMessage!.role).toBe('user')
|
|
expect(contextMessage!.content[0]).toMatchObject({ type: 'text', text: '<context source="plugin">' })
|
|
expect(contextMessage!.content.at(-1)).toMatchObject({ type: 'text', text: '</context>' })
|
|
expect(steeringMessage!.content[0]).toMatchObject({ type: 'text', text: '<steering source="user">' })
|
|
})
|
|
|
|
it('replays identically from a seeded event log', () => {
|
|
const original = new Session(SessionId('s3'))
|
|
original.append('user/message', { content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } })
|
|
original.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'a' }] })
|
|
|
|
const replayed = new Session(SessionId('s3-replay'), [...original.events])
|
|
expect(replayed.deriveMessages()).toEqual(original.deriveMessages())
|
|
expect(replayed.seq).toBe(original.seq)
|
|
})
|
|
|
|
it('isolates the log from mutation through a derived message (append-only contract)', () => {
|
|
const session = new Session(SessionId('s4'))
|
|
session.append('user/message', { content: [{ type: 'text', text: 'original' }], source: { kind: 'user' } })
|
|
session.append('tool/result', {
|
|
turn: 1, step: 1, callId: CallId('c1'),
|
|
content: [{ type: 'text', text: 'tool out' }], isError: false,
|
|
})
|
|
const before = structuredClone(session.events)
|
|
|
|
// A request middleware / adapter mutates the messages it was handed.
|
|
const messages = session.deriveMessages()
|
|
const userBlock = messages[0]!.content[0]!
|
|
if (userBlock.type === 'text') userBlock.text = 'HACKED'
|
|
const toolBlock = messages[1]!.content[0]!
|
|
if (toolBlock.type === 'tool-result') {
|
|
toolBlock.content.push({ type: 'text', text: 'injected' })
|
|
}
|
|
messages[0]!.content.push({ type: 'text', text: 'extra' })
|
|
|
|
// The log is unchanged: deep-equal to the snapshot taken before mutation.
|
|
expect(session.events).toEqual(before)
|
|
// And a fresh derivation still reflects the original content.
|
|
expect(session.deriveMessages()[0]!.content).toEqual([{ type: 'text', text: 'original' }])
|
|
})
|
|
})
|
|
|
|
describe('SessionStore', () => {
|
|
it('creates sessions, emits session/created and session/event', async () => {
|
|
const ctx = new Context()
|
|
await ctx.plugin(SessionStore)
|
|
|
|
const created: Session[] = []
|
|
const events: [Session, SessionEvent][] = []
|
|
ctx.on('session/created', session => void created.push(session))
|
|
ctx.on('session/event', (session, event) => void events.push([session, event]))
|
|
|
|
const session = ctx.sessions.create()
|
|
expect(created).toEqual([session])
|
|
|
|
session.append('user/message', { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' } })
|
|
expect(events).toHaveLength(1)
|
|
expect(events[0]![0]).toBe(session)
|
|
expect(events[0]![1].type).toBe('user/message')
|
|
|
|
expect(ctx.sessions.get(session.id)).toBe(session)
|
|
expect(ctx.sessions.list()).toEqual([session])
|
|
})
|
|
|
|
it('rejects duplicate ids and supports seeding', async () => {
|
|
const ctx = new Context()
|
|
await ctx.plugin(SessionStore)
|
|
const a = ctx.sessions.create('fixed')
|
|
expect(() => ctx.sessions.create('fixed')).toThrow('already exists')
|
|
|
|
a.append('user/message', { content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } })
|
|
const forked = ctx.sessions.create('fork', [...a.events])
|
|
expect(forked.deriveMessages()).toEqual(a.deriveMessages())
|
|
})
|
|
|
|
it('detaches sessions when the creating fiber is disposed (HMR safety)', async () => {
|
|
const ctx = new Context()
|
|
await ctx.plugin(SessionStore)
|
|
|
|
let session!: Session
|
|
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
|
|
session = inner.sessions.create('scoped')
|
|
}, { inject: ['sessions'] }))
|
|
expect(ctx.sessions.get('scoped')).toBe(session)
|
|
|
|
let observed = 0
|
|
ctx.on('session/event', () => void observed++)
|
|
|
|
await fiber.dispose()
|
|
expect(ctx.sessions.get('scoped')).toBeUndefined()
|
|
session.append('user/message', { content: [{ type: 'text', text: 'late' }], source: { kind: 'user' } })
|
|
expect(observed).toBe(0)
|
|
})
|
|
|
|
it('rolls back the session (and onAppend) when a session/created listener throws (P1-1)', async () => {
|
|
const ctx = new Context()
|
|
await ctx.plugin(SessionStore)
|
|
|
|
let threw = false
|
|
ctx.on('session/created', () => {
|
|
if (!threw) { threw = true; throw new Error('boom created listener') }
|
|
})
|
|
|
|
// The throwing emit must roll the store entry back, not leak it.
|
|
expect(() => ctx.sessions.create('fixed')).toThrow('boom created listener')
|
|
expect(ctx.sessions.get('fixed')).toBeUndefined() // rolled back, not leaked
|
|
|
|
// A subsequent create of the SAME id succeeds (the already-exists check is
|
|
// not wedged) and its onAppend is correctly wired (events observable).
|
|
const events: SessionEvent[] = []
|
|
ctx.on('session/event', (_session, event) => void events.push(event))
|
|
const session = ctx.sessions.create('fixed')
|
|
expect(ctx.sessions.get('fixed')).toBe(session)
|
|
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } })
|
|
expect(events).toHaveLength(1)
|
|
})
|
|
})
|