mirror of
https://github.com/deepseek-ai/deepseek-harness
synced 2026-08-15 21:04:50 +00:00
7e445c3a67
git mv the 12 packages from session-persistence/, session-projection/, session-title/, and telemetry/ into one session/ group per the regrouping RFC; merge the four group READMEs into one bilingual triplet; rewrite the group segment in tsconfig references (intra-group references shorten to ../<pkg>), tsconfig.base.json paths/globs, knip.json keys, vitest include, gate scripts, and authored doc/note citations; regenerate module graph, doc graphs, catalogs, and the lockfile importer keys. No npm names change. Full unit suite: 8779 passed; the 18 reported failures reproduce as env flakes (ambient-proxy IPv6 tunneling, watched-dir inotify timeouts under parallel load) — each passes in isolation with NO_PROXY set, matching their known pre-existing behavior on master.
276 lines
8.6 KiB
TypeScript
276 lines
8.6 KiB
TypeScript
import { afterEach, describe, expect, it, vi } from 'vitest'
|
|
import type { SessionEvent } from '@deepseek-ai/dsh-session'
|
|
import { SessionWriteBehind } from '../src/write-behind.ts'
|
|
|
|
/** Minimal ordered event fixture; batching does not interpret event vocabulary. */
|
|
function event(seq: number): SessionEvent<'turn/start'> {
|
|
return {
|
|
type: 'turn/start',
|
|
seq,
|
|
time: seq,
|
|
data: { turn: seq + 1 },
|
|
}
|
|
}
|
|
|
|
afterEach(() => {
|
|
vi.useRealTimers()
|
|
})
|
|
|
|
describe('SessionWriteBehind', () => {
|
|
it('uses one fixed window from the first queued event and owns its copy', async () => {
|
|
vi.useFakeTimers()
|
|
const batches: SessionEvent[][] = []
|
|
const controller = new SessionWriteBehind({
|
|
maxDelayMs: 200,
|
|
write: async (events) => { batches.push(structuredClone(events) as SessionEvent[]) },
|
|
reportBackgroundFailure: vi.fn(),
|
|
})
|
|
const first = event(0)
|
|
|
|
controller.enqueue(first)
|
|
first.data.turn = 99
|
|
await vi.advanceTimersByTimeAsync(150)
|
|
controller.enqueue(event(1))
|
|
await vi.advanceTimersByTimeAsync(49)
|
|
expect(batches).toEqual([])
|
|
|
|
await vi.advanceTimersByTimeAsync(1)
|
|
expect(batches).toEqual([[
|
|
expect.objectContaining({ seq: 0, data: { turn: 1 } }),
|
|
expect.objectContaining({ seq: 1 }),
|
|
]])
|
|
expect(controller.hasWork).toBe(false)
|
|
})
|
|
|
|
it('coalesces twenty events admitted ten milliseconds apart into one 200 ms batch', async () => {
|
|
vi.useFakeTimers()
|
|
const batches: number[][] = []
|
|
const controller = new SessionWriteBehind({
|
|
maxDelayMs: 200,
|
|
write: async (events) => { batches.push(events.map(item => item.seq)) },
|
|
reportBackgroundFailure: vi.fn(),
|
|
})
|
|
|
|
controller.enqueue(event(0))
|
|
for (let seq = 1; seq < 20; seq += 1) {
|
|
await vi.advanceTimersByTimeAsync(10)
|
|
controller.enqueue(event(seq))
|
|
}
|
|
expect(batches).toEqual([])
|
|
|
|
await vi.advanceTimersByTimeAsync(10)
|
|
expect(batches).toEqual([Array.from({ length: 20 }, (_, seq) => seq)])
|
|
await controller.flush()
|
|
})
|
|
|
|
it('makes concurrent flushes one immediate barrier that drains admitted tails', async () => {
|
|
vi.useFakeTimers()
|
|
const gate = Promise.withResolvers<boolean>()
|
|
const batches: number[][] = []
|
|
const controller = new SessionWriteBehind({
|
|
maxDelayMs: 200,
|
|
write: async (events) => {
|
|
batches.push(events.map(item => item.seq))
|
|
if (batches.length === 1) await gate.promise
|
|
},
|
|
reportBackgroundFailure: vi.fn(),
|
|
})
|
|
|
|
controller.enqueue(event(0))
|
|
const first = controller.flush()
|
|
const second = controller.flush()
|
|
expect(second).toBe(first)
|
|
await Promise.resolve()
|
|
expect(batches).toEqual([[0]])
|
|
|
|
controller.enqueue(event(1))
|
|
gate.resolve(true)
|
|
await first
|
|
expect(batches).toEqual([[0], [1]])
|
|
expect(controller.hasWork).toBe(false)
|
|
expect(vi.getTimerCount()).toBe(0)
|
|
})
|
|
|
|
it('starts a new window for work admitted after an already-quiescent barrier', async () => {
|
|
vi.useFakeTimers()
|
|
const batches: number[][] = []
|
|
const controller = new SessionWriteBehind({
|
|
maxDelayMs: 200,
|
|
write: async (events) => { batches.push(events.map(item => item.seq)) },
|
|
reportBackgroundFailure: vi.fn(),
|
|
})
|
|
|
|
const barrier = controller.flush()
|
|
controller.enqueue(event(0))
|
|
await barrier
|
|
expect(batches).toEqual([])
|
|
expect(vi.getTimerCount()).toBe(1)
|
|
|
|
await vi.advanceTimersByTimeAsync(200)
|
|
expect(batches).toEqual([[0]])
|
|
expect(controller.hasWork).toBe(false)
|
|
})
|
|
|
|
it('starts an over-budget tail immediately after the active write', async () => {
|
|
vi.useFakeTimers()
|
|
const gate = Promise.withResolvers<boolean>()
|
|
const batches: number[][] = []
|
|
const controller = new SessionWriteBehind({
|
|
maxDelayMs: 200,
|
|
write: async (events) => {
|
|
batches.push(events.map(item => item.seq))
|
|
if (batches.length === 1) await gate.promise
|
|
},
|
|
reportBackgroundFailure: vi.fn(),
|
|
})
|
|
|
|
controller.enqueue(event(0))
|
|
await vi.advanceTimersByTimeAsync(200)
|
|
expect(batches).toEqual([[0]])
|
|
controller.enqueue(event(1))
|
|
await vi.advanceTimersByTimeAsync(200)
|
|
expect(batches).toEqual([[0]])
|
|
|
|
gate.resolve(true)
|
|
await vi.advanceTimersByTimeAsync(0)
|
|
expect(batches).toEqual([[0], [1]])
|
|
await controller.flush()
|
|
})
|
|
|
|
it('keeps a tail deadline that has not expired when the active write finishes', async () => {
|
|
vi.useFakeTimers()
|
|
const gate = Promise.withResolvers<boolean>()
|
|
const batches: number[][] = []
|
|
const controller = new SessionWriteBehind({
|
|
maxDelayMs: 200,
|
|
write: async (events) => {
|
|
batches.push(events.map(item => item.seq))
|
|
if (batches.length === 1) await gate.promise
|
|
},
|
|
reportBackgroundFailure: vi.fn(),
|
|
})
|
|
|
|
controller.enqueue(event(0))
|
|
await vi.advanceTimersByTimeAsync(200)
|
|
controller.enqueue(event(1))
|
|
await vi.advanceTimersByTimeAsync(50)
|
|
gate.resolve(true)
|
|
await vi.advanceTimersByTimeAsync(0)
|
|
expect(batches).toEqual([[0]])
|
|
|
|
await vi.advanceTimersByTimeAsync(149)
|
|
expect(batches).toEqual([[0]])
|
|
await vi.advanceTimersByTimeAsync(1)
|
|
expect(batches).toEqual([[0], [1]])
|
|
await controller.flush()
|
|
})
|
|
|
|
it('pauses automatic retries after failure and preserves order for new work', async () => {
|
|
vi.useFakeTimers()
|
|
const failure = new Error('storage unavailable')
|
|
const report = vi.fn()
|
|
const batches: number[][] = []
|
|
let attempt = 0
|
|
const controller = new SessionWriteBehind({
|
|
maxDelayMs: 200,
|
|
write: async (events) => {
|
|
batches.push(events.map(item => item.seq))
|
|
if (++attempt === 1) throw failure
|
|
},
|
|
reportBackgroundFailure: report,
|
|
})
|
|
|
|
controller.enqueue(event(0))
|
|
await vi.advanceTimersByTimeAsync(200)
|
|
expect(report).toHaveBeenCalledWith(failure)
|
|
expect(controller.hasWork).toBe(true)
|
|
await vi.advanceTimersByTimeAsync(1_000)
|
|
expect(batches).toEqual([[0]])
|
|
|
|
controller.enqueue(event(1))
|
|
await vi.advanceTimersByTimeAsync(199)
|
|
expect(batches).toEqual([[0]])
|
|
await vi.advanceTimersByTimeAsync(1)
|
|
expect(batches).toEqual([[0], [0, 1]])
|
|
await controller.flush()
|
|
})
|
|
|
|
it('observes an overlapping background failure and retries it inside flush', async () => {
|
|
vi.useFakeTimers()
|
|
const gate = Promise.withResolvers<boolean>()
|
|
const report = vi.fn()
|
|
const batches: number[][] = []
|
|
const controller = new SessionWriteBehind({
|
|
maxDelayMs: 200,
|
|
write: async (events) => {
|
|
batches.push(events.map(item => item.seq))
|
|
if (batches.length === 1) {
|
|
await gate.promise
|
|
throw new Error('transient')
|
|
}
|
|
},
|
|
reportBackgroundFailure: report,
|
|
})
|
|
|
|
controller.enqueue(event(0))
|
|
await vi.advanceTimersByTimeAsync(200)
|
|
const first = controller.flush()
|
|
const second = controller.flush()
|
|
gate.resolve(true)
|
|
|
|
await expect(Promise.all([first, second])).resolves.toEqual([undefined, undefined])
|
|
expect(batches).toEqual([[0], [0]])
|
|
expect(report).toHaveBeenCalledOnce()
|
|
expect(controller.hasWork).toBe(false)
|
|
})
|
|
|
|
it('surfaces a barrier failure without detached logging and retains its batch', async () => {
|
|
vi.useFakeTimers()
|
|
const failure = new Error('durability failed')
|
|
const report = vi.fn()
|
|
const batches: number[][] = []
|
|
let attempt = 0
|
|
const controller = new SessionWriteBehind({
|
|
maxDelayMs: 200,
|
|
write: async (events) => {
|
|
batches.push(events.map(item => item.seq))
|
|
if (++attempt === 1) throw failure
|
|
},
|
|
reportBackgroundFailure: report,
|
|
})
|
|
|
|
controller.enqueue(event(0))
|
|
await expect(controller.flush()).rejects.toBe(failure)
|
|
expect(report).not.toHaveBeenCalled()
|
|
expect(controller.hasWork).toBe(true)
|
|
|
|
controller.enqueue(event(1))
|
|
await vi.advanceTimersByTimeAsync(200)
|
|
expect(batches).toEqual([[0], [0, 1]])
|
|
await controller.flush()
|
|
})
|
|
|
|
it('retains a failed batch larger than the engine call-argument limit', async () => {
|
|
const failure = new Error('durability failed')
|
|
const batchSize = 150_000
|
|
const sizes: number[] = []
|
|
let attempt = 0
|
|
const controller = new SessionWriteBehind({
|
|
maxDelayMs: 200,
|
|
write: async (events) => {
|
|
sizes.push(events.length)
|
|
if (++attempt === 1) throw failure
|
|
},
|
|
reportBackgroundFailure: vi.fn(),
|
|
})
|
|
|
|
for (let seq = 0; seq < batchSize; seq += 1) controller.enqueue(event(seq))
|
|
await expect(controller.flush()).rejects.toBe(failure)
|
|
expect(controller.hasWork).toBe(true)
|
|
|
|
await controller.flush()
|
|
expect(sizes).toEqual([batchSize, batchSize])
|
|
expect(controller.hasWork).toBe(false)
|
|
})
|
|
})
|