Merge remote-tracking branch 'origin/master' into worktree/semantic-session-checkpoints
# Conflicts: # docs/architecture.md # docs/event-producer-consumer.md # examples/acp-agent/tests/snapshots/cancel-tool-calls/session.jsonl # packages/core/agent-loop/README.md # packages/core/agent-loop/src/loop.ts # packages/core/agent-loop/tests/cancel.spec.ts
This commit is contained in:
Yichen Jiang
@@ -102,7 +102,7 @@ describe('SessionStore.fork', () => {
|
||||
const { ctx, sessions } = await setup()
|
||||
const reasons: TurnEndReason[] = [
|
||||
{ kind: 'completed' },
|
||||
{ kind: 'aborted', reason: 'cancelled by user' },
|
||||
{ kind: 'aborted' },
|
||||
{ kind: 'error', step: 1, message: 'model failed', code: 'MODEL' },
|
||||
{ kind: 'disposed' },
|
||||
{ kind: 'max-tokens' },
|
||||
|
||||
226
packages/core/session/tests/out-of-band.spec.ts
Normal file
226
packages/core/session/tests/out-of-band.spec.ts
Normal file
@@ -0,0 +1,226 @@
|
||||
import { Context } from 'cordis'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
|
||||
declare module '@deepseek-ai/dsh-session' {
|
||||
interface SessionEventMap {
|
||||
'test/log-only': { value: string }
|
||||
}
|
||||
|
||||
interface OutOfBandSessionEventMap {
|
||||
'test/log-only': true
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
const updateTrigger = { kind: 'injection', source: { kind: 'plugin', plugin: 'test' } } as const
|
||||
|
||||
describe('SessionStore.appendOutOfBand', () => {
|
||||
it('joins an open turn without adding a boundary or flushing it', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const session = ctx.sessions.create(SessionId('open'))
|
||||
let flushes = 0
|
||||
ctx.on('session/flush', () => { flushes += 1 })
|
||||
session.append('turn/start', {
|
||||
turn: 1,
|
||||
trigger: { kind: 'message', source: { kind: 'user' } },
|
||||
})
|
||||
|
||||
const event = await ctx.sessions.appendOutOfBand(
|
||||
session,
|
||||
'test/log-only',
|
||||
{ value: 'inside' },
|
||||
updateTrigger,
|
||||
)
|
||||
|
||||
expect(event).toMatchObject({ type: 'test/log-only', seq: 1, data: { value: 'inside' } })
|
||||
expect(session.events.map(item => item.type)).toEqual(['turn/start', 'test/log-only'])
|
||||
expect(flushes).toBe(0)
|
||||
})
|
||||
|
||||
it('wraps a closed log in one zero-step turn and flushes the balanced update', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const session = ctx.sessions.create(SessionId('closed'))
|
||||
const flushedTypes: string[][] = []
|
||||
ctx.on('session/flush', (flushed) => {
|
||||
flushedTypes.push(flushed.events.map(event => event.type))
|
||||
})
|
||||
|
||||
const first = await ctx.sessions.appendOutOfBand(
|
||||
session,
|
||||
'test/log-only',
|
||||
{ value: 'first' },
|
||||
updateTrigger,
|
||||
)
|
||||
const second = await ctx.sessions.appendOutOfBand(
|
||||
session,
|
||||
'test/log-only',
|
||||
{ value: 'second' },
|
||||
updateTrigger,
|
||||
)
|
||||
|
||||
expect(first.seq).toBe(1)
|
||||
expect(second.seq).toBe(4)
|
||||
expect(session.events).toMatchObject([
|
||||
{ type: 'turn/start', seq: 0, data: { turn: 1, trigger: updateTrigger } },
|
||||
{ type: 'test/log-only', seq: 1, data: { value: 'first' } },
|
||||
{ type: 'turn/end', seq: 2, data: { turn: 1, reason: { kind: 'completed' } } },
|
||||
{ type: 'turn/start', seq: 3, data: { turn: 2, trigger: updateTrigger } },
|
||||
{ type: 'test/log-only', seq: 4, data: { value: 'second' } },
|
||||
{ type: 'turn/end', seq: 5, data: { turn: 2, reason: { kind: 'completed' } } },
|
||||
])
|
||||
expect(flushedTypes).toEqual([
|
||||
['turn/start', 'test/log-only', 'turn/end'],
|
||||
['turn/start', 'test/log-only', 'turn/end', 'turn/start', 'test/log-only', 'turn/end'],
|
||||
])
|
||||
})
|
||||
|
||||
it('closes and flushes a zero-step turn when the target event is rejected', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const session = ctx.sessions.create(SessionId('rejected'))
|
||||
let flushes = 0
|
||||
ctx.on('session/flush', () => { flushes += 1 })
|
||||
|
||||
await expect(ctx.sessions.appendOutOfBand(
|
||||
session,
|
||||
'test/log-only',
|
||||
{ value: 1n } as never,
|
||||
updateTrigger,
|
||||
)).rejects.toThrow(/non-JSON-serializable/)
|
||||
|
||||
expect(session.events).toMatchObject([
|
||||
{ type: 'turn/start', data: { turn: 1 } },
|
||||
{ type: 'turn/end', data: { turn: 1, reason: { kind: 'completed' } } },
|
||||
])
|
||||
expect(flushes).toBe(1)
|
||||
})
|
||||
|
||||
it('does not flush when the synthetic turn cannot open', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const session = ctx.sessions.create(SessionId('start-failure'))
|
||||
let flushes = 0
|
||||
ctx.on('session/flush', () => { flushes += 1 })
|
||||
|
||||
await expect(ctx.sessions.appendOutOfBand(
|
||||
session,
|
||||
'test/log-only',
|
||||
{ value: 'unreachable' },
|
||||
{ ...updateTrigger, invalid: 1n } as never,
|
||||
)).rejects.toThrow(/non-JSON-serializable/)
|
||||
|
||||
expect(session.events).toEqual([])
|
||||
expect(flushes).toBe(0)
|
||||
})
|
||||
|
||||
it('preserves a target rejection when the balancing flush also rejects', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const session = ctx.sessions.create(SessionId('target-and-flush-failure'))
|
||||
ctx.on('session/flush', () => { throw new Error('disk failed') })
|
||||
|
||||
await expect(ctx.sessions.appendOutOfBand(
|
||||
session,
|
||||
'test/log-only',
|
||||
{ value: 1n } as never,
|
||||
updateTrigger,
|
||||
)).rejects.toThrow(/non-JSON-serializable/)
|
||||
|
||||
expect(session.events.map(event => event.type)).toEqual([
|
||||
'turn/start',
|
||||
'turn/end',
|
||||
])
|
||||
})
|
||||
|
||||
it('keeps the session attached through publication and its flush', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const session = ctx.sessions.prepare(SessionId('dispose'))
|
||||
const detach = ctx.sessions.enter(session)
|
||||
ctx.sessions.announce(session)
|
||||
let liveDuringFlush = false
|
||||
ctx.on('session/event', (_observed, event) => {
|
||||
if (event.type === 'turn/start') detach()
|
||||
})
|
||||
ctx.on('session/flush', () => {
|
||||
liveDuringFlush = ctx.sessions.get(session.id) === session
|
||||
})
|
||||
|
||||
await ctx.sessions.appendOutOfBand(
|
||||
session,
|
||||
'test/log-only',
|
||||
{ value: 'last' },
|
||||
updateTrigger,
|
||||
)
|
||||
|
||||
expect(session.events.map(event => event.type)).toEqual([
|
||||
'turn/start',
|
||||
'test/log-only',
|
||||
'turn/end',
|
||||
])
|
||||
expect(liveDuringFlush).toBe(true)
|
||||
expect(ctx.sessions.get(session.id)).toBeUndefined()
|
||||
})
|
||||
|
||||
it('rejects detached sessions before opening a turn', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const session = ctx.sessions.prepare(SessionId('detached'))
|
||||
|
||||
await expect(ctx.sessions.appendOutOfBand(
|
||||
session,
|
||||
'test/log-only',
|
||||
{ value: 'nope' },
|
||||
updateTrigger,
|
||||
)).rejects.toThrow('session "detached" is not live in this store')
|
||||
expect(session.events).toEqual([])
|
||||
})
|
||||
|
||||
it('leaves a balanced log when the durability checkpoint rejects', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const session = ctx.sessions.create(SessionId('flush-failure'))
|
||||
ctx.on('session/flush', () => { throw new Error('disk failed') })
|
||||
|
||||
await expect(ctx.sessions.appendOutOfBand(
|
||||
session,
|
||||
'test/log-only',
|
||||
{ value: 'accepted' },
|
||||
updateTrigger,
|
||||
)).rejects.toThrow('disk failed')
|
||||
expect(session.events.map(event => event.type)).toEqual([
|
||||
'turn/start',
|
||||
'test/log-only',
|
||||
'turn/end',
|
||||
])
|
||||
})
|
||||
|
||||
it('rejects overlapping updates while the first append is still settling', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const session = ctx.sessions.create(SessionId('overlap'))
|
||||
let release!: () => void
|
||||
const checkpoint = new Promise<void>((resolve) => {
|
||||
release = resolve
|
||||
})
|
||||
ctx.on('session/flush', () => checkpoint)
|
||||
|
||||
const first = ctx.sessions.appendOutOfBand(
|
||||
session,
|
||||
'test/log-only',
|
||||
{ value: 'first' },
|
||||
updateTrigger,
|
||||
)
|
||||
await expect(ctx.sessions.appendOutOfBand(
|
||||
session,
|
||||
'test/log-only',
|
||||
{ value: 'overlap' },
|
||||
updateTrigger,
|
||||
)).rejects.toThrow(/out-of-band append in progress/)
|
||||
release()
|
||||
await expect(first).resolves.toMatchObject({ data: { value: 'first' } })
|
||||
})
|
||||
})
|
||||
@@ -1,7 +1,13 @@
|
||||
import { describe, expect, expectTypeOf, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { CallId } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { SESSION_FORMAT_VERSION, Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import SessionStore, {
|
||||
findLastMessageTurnEnd,
|
||||
SESSION_FORMAT_VERSION,
|
||||
Session,
|
||||
SessionEvent,
|
||||
SessionId,
|
||||
} from '@deepseek-ai/dsh-session'
|
||||
import type { CreateSessionOptions, SessionEventType, SessionHeader, SessionSurface, TodoItem } from '@deepseek-ai/dsh-session'
|
||||
|
||||
describe('Session', () => {
|
||||
@@ -48,6 +54,68 @@ describe('Session', () => {
|
||||
expect(structuredClone(turnEnd.data.reason)).toEqual({ kind: 'max-tokens' })
|
||||
})
|
||||
|
||||
it('finds the latest message-turn outcome past later non-message turns', () => {
|
||||
const session = new Session(SessionId('message-turn-outcome'))
|
||||
expect(findLastMessageTurnEnd(session.events)).toBeUndefined()
|
||||
session.append('turn/start', {
|
||||
turn: 1,
|
||||
trigger: { kind: 'injection', source: { kind: 'plugin', plugin: 'before' } },
|
||||
})
|
||||
session.append('context/message', {
|
||||
content: [{ type: 'text', text: 'before' }],
|
||||
source: { kind: 'plugin', plugin: 'before' },
|
||||
}, { surfaceOp: 'append' })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
expect(findLastMessageTurnEnd(session.events)).toBeUndefined()
|
||||
|
||||
session.append('turn/start', {
|
||||
turn: 2,
|
||||
trigger: { kind: 'message', source: { kind: 'user' } },
|
||||
})
|
||||
session.append('user/message', {
|
||||
content: [{ type: 'text', text: 'bounded prompt' }],
|
||||
source: { kind: 'user' },
|
||||
}, { surfaceOp: 'append' })
|
||||
const messageEnd = session.append('turn/end', { turn: 2, reason: { kind: 'max-tokens' } })
|
||||
session.append('turn/start', {
|
||||
turn: 3,
|
||||
trigger: { kind: 'injection', source: { kind: 'plugin', plugin: 'after' } },
|
||||
})
|
||||
session.append('context/message', {
|
||||
content: [{ type: 'text', text: 'after' }],
|
||||
source: { kind: 'plugin', plugin: 'after' },
|
||||
}, { surfaceOp: 'append' })
|
||||
session.append('turn/end', { turn: 3, reason: { kind: 'completed' } })
|
||||
|
||||
expect(findLastMessageTurnEnd(session.events)).toBe(messageEnd)
|
||||
})
|
||||
|
||||
it('round-trips the coarse aborted turn outcome', () => {
|
||||
const session = new Session(SessionId('aborted'))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'aborted' } })
|
||||
const replayed = new Session(SessionId('aborted-replay'), structuredClone(session.events))
|
||||
expect(replayed.events).toEqual(session.events)
|
||||
const turnEnd = replayed.events.findLast(event => event.type === 'turn/end')
|
||||
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted' })
|
||||
})
|
||||
|
||||
it('rejects legacy reason-bearing aborted outcomes at the seed/load boundary', () => {
|
||||
const legacy = [
|
||||
{
|
||||
type: 'turn/start', seq: 0, time: 1,
|
||||
data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } },
|
||||
},
|
||||
{
|
||||
type: 'turn/end', seq: 1, time: 2,
|
||||
data: { turn: 1, reason: { kind: 'aborted', reason: 'legacy cancellation detail' } },
|
||||
},
|
||||
] as unknown as SessionEvent[]
|
||||
|
||||
expect(() => new Session(SessionId('legacy-aborted'), legacy))
|
||||
.toThrow('seed turn/end at index 1 uses unsupported reason-bearing aborted format')
|
||||
})
|
||||
|
||||
it('renders context and steering messages as plain user content', () => {
|
||||
const session = new Session(SessionId('s2'))
|
||||
session.append('context/message', {
|
||||
|
||||
Reference in New Issue
Block a user