122 lines
7.6 KiB
TypeScript
122 lines
7.6 KiB
TypeScript
import assert from 'node:assert/strict'
|
|
import { promises as fs } from 'node:fs'
|
|
import { resolve, join } from 'node:path'
|
|
import { createPromptQueue } from '../src/main/prompt-queue'
|
|
import type { PromptQueueSnapshot } from '../src/shared/prompt-queue'
|
|
|
|
const workspace = resolve('.tmp')
|
|
await fs.mkdir(workspace, { recursive: true })
|
|
const root = await fs.mkdtemp(join(workspace, 'prompt-queue-'))
|
|
const tick = () => new Promise(resolve => setTimeout(resolve, 10))
|
|
try {
|
|
let blocked = true
|
|
const delivered: string[] = []
|
|
const changes: PromptQueueSnapshot[] = []
|
|
let finish: ((value: { status: 'done' | 'cancelled' }) => void) | undefined
|
|
const options = {
|
|
file: join(root, 'queues.json'),
|
|
validate: async () => {},
|
|
blocked: async () => blocked,
|
|
changed: (snapshot: PromptQueueSnapshot) => { changes.push(snapshot) },
|
|
deliver: async (_payload: unknown, id: string) => {
|
|
delivered.push(id)
|
|
return { completion: new Promise<{ status: 'done' | 'cancelled' }>(resolve => { finish = resolve }) }
|
|
}
|
|
}
|
|
const queue = createPromptQueue(options)
|
|
for (let index = 0; index < 19; index++) await queue.enqueue({ sessionId: 'a', text: `${index}`, resources: ['C:/workspace/a'], skillIds: ['skill'], mcpServerIds: ['server'], quotes: [], attachments: [] })
|
|
await assert.rejects(queue.enqueue({ sessionId: 'a', text: 'full' }), /19/)
|
|
assert.equal((await queue.get('a')).items.length, 19)
|
|
const order = (await queue.get('a')).items.map(item => item.id).reverse()
|
|
await queue.reorder('a', order)
|
|
await assert.rejects(queue.reorder('a', [...order.slice(1), order[1]]), /排序/)
|
|
const taken = await queue.take('a', order[0])
|
|
assert.deepEqual(taken.payload.skillIds, ['skill'])
|
|
assert.deepEqual(taken.payload.mcpServerIds, ['server'])
|
|
assert.deepEqual(taken.payload.resources, ['C:/workspace/a'])
|
|
assert.equal((await queue.get('b')).items.length, 0, 'sessions are isolated')
|
|
assert.equal(delivered.length, 0, 'active run blocks ordinary dispatch')
|
|
|
|
const recovered = createPromptQueue(options)
|
|
assert.equal((await recovered.get('a')).paused, true, 'restart requires explicit resume')
|
|
blocked = false
|
|
await recovered.resume('a')
|
|
for (let attempt = 0; attempt < 100 && !delivered.length; attempt++) await tick()
|
|
assert.equal(delivered.length, 1)
|
|
for (let attempt = 0; attempt < 100 && (await recovered.get('a')).items.length === 18; attempt++) await tick()
|
|
assert.equal((await recovered.get('a')).items.length, 17, 'only acknowledged delivery leaves the queue')
|
|
finish!({ status: 'cancelled' })
|
|
for (let attempt = 0; attempt < 100 && !(await recovered.get('a')).paused; attempt++) await tick()
|
|
assert.equal((await recovered.get('a')).paused, true)
|
|
assert.equal(delivered.length, 1, 'stop pauses all remaining ordinary sends')
|
|
|
|
const uncertain = createPromptQueue({ ...options, file: join(root, 'uncertain.json'),
|
|
deliver: async () => { throw new Error('lost acknowledgement') } })
|
|
await uncertain.enqueue({ sessionId: 'c', text: 'uncertain' })
|
|
for (let attempt = 0; attempt < 100 && !(await uncertain.get('c')).paused; attempt++) await tick()
|
|
assert.equal((await uncertain.get('c')).items[0].status, 'unknown')
|
|
await assert.rejects(uncertain.resume('c'), /结果不明/)
|
|
await assert.rejects(uncertain.sendNow('c', 'unknown'), /尚未接入/)
|
|
const disk = JSON.parse(await fs.readFile(options.file, 'utf8'))
|
|
const interruptedItem = disk.queues[0].items.find((item: { status: string }) => item.status !== 'editing')
|
|
interruptedItem.status = 'delivering'
|
|
delete interruptedItem.deliveryOwner
|
|
await fs.writeFile(options.file, JSON.stringify(disk))
|
|
const interrupted = createPromptQueue(options)
|
|
assert.equal((await interrupted.get('a')).items[0].status, 'unknown')
|
|
await assert.rejects(interrupted.resume('a'), /结果不明/)
|
|
assert.ok(changes.every(snapshot => snapshot.revision >= 1))
|
|
const editFile = join(root, 'editing.json')
|
|
const editing = createPromptQueue({ ...options, file: editFile, blocked: async () => true })
|
|
await editing.enqueue({ sessionId: 'edit', text: 'original', skillIds: ['skill'] })
|
|
const originalId = (await editing.get('edit')).items[0].id
|
|
const draft = await editing.take('edit', originalId)
|
|
assert.equal(draft.payload.queueEditId, originalId)
|
|
assert.equal((await editing.get('edit')).items.length, 0)
|
|
assert.equal((await editing.get('edit')).editingDraft?.payload.text, 'original', 'take remains recoverable on disk until submission')
|
|
await editing.restoreEdit('edit', originalId)
|
|
assert.equal((await editing.get('edit')).items[0].id, originalId, 'rollback preserves identity and order')
|
|
const retaken = await editing.take('edit', originalId)
|
|
await editing.enqueue({ ...retaken.payload, text: 'edited' })
|
|
assert.equal((await editing.get('edit')).items[0].payload.text, 'edited')
|
|
assert.equal((await editing.get('edit')).editingDraft, undefined)
|
|
await assert.rejects(editing.enqueue({ sessionId: 'edit', text: 'invalid handoff', queueEditId: '' }), /编辑取回标识/)
|
|
const ordered = createPromptQueue({ ...options, file: join(root, 'edit-order.json'), blocked: async () => true })
|
|
for (const text of ['first', 'second', 'third']) await ordered.enqueue({ sessionId: 'edit-order', text })
|
|
const middle = (await ordered.get('edit-order')).items[1]
|
|
const middleDraft = await ordered.take('edit-order', middle.id)
|
|
await ordered.enqueue({ ...middleDraft.payload, text: 'second edited' })
|
|
assert.deepEqual((await ordered.get('edit-order')).items.map(item => item.payload.text),
|
|
['first', 'second edited', 'third'], 'editing and resubmitting keeps the original queue position')
|
|
const toRecover = (await editing.get('edit')).items[0].id
|
|
await editing.take('edit', toRecover)
|
|
const editingDisk = JSON.parse(await fs.readFile(editFile, 'utf8'))
|
|
editingDisk.queues[0].items[0].deliveryOwner.nonce = 'previous-process'
|
|
await fs.writeFile(editFile, JSON.stringify(editingDisk))
|
|
assert.equal((await createPromptQueue({ ...options, file: editFile }).get('edit')).items[0].payload.text, 'edited', 'restart recovers an unsubmitted editor handoff')
|
|
|
|
const guideAcks = new Map<string, { resolve(): void; reject(error: Error): void }>()
|
|
const immediate = createPromptQueue({ ...options, file: join(root, 'immediate.json'), blocked: async () => true,
|
|
deliverNow: (_payload, id) => new Promise<void>((resolve, reject) => { guideAcks.set(id, { resolve, reject }) }) })
|
|
await immediate.enqueue({ sessionId: 'guide', text: 'a' })
|
|
await immediate.enqueue({ sessionId: 'guide', text: 'b' })
|
|
const guideIds = (await immediate.get('guide')).items.map(item => item.id)
|
|
const firstGuide = immediate.sendNow('guide', guideIds[0])
|
|
const secondGuide = immediate.sendNow('guide', guideIds[1])
|
|
const secondFailure = assert.rejects(secondGuide, /cancelled/)
|
|
for (let attempt = 0; attempt < 100 && guideAcks.size !== 2; attempt++) await tick()
|
|
assert.equal(guideAcks.size, 2)
|
|
assert.ok((await immediate.get('guide')).items.every(item => item.status === 'delivering'))
|
|
guideAcks.get(guideIds[0])!.resolve()
|
|
await firstGuide
|
|
assert.equal((await immediate.get('guide')).items[0].id, guideIds[1], 'one acknowledgement cannot consume another input')
|
|
guideAcks.get(guideIds[1])!.reject(Object.assign(new Error('cancelled'), { submitted: true }))
|
|
await secondFailure
|
|
assert.equal((await immediate.get('guide')).items[0].status, 'unknown')
|
|
assert.equal((await immediate.get('guide')).paused, true)
|
|
console.log('prompt-queue-regression: cap, payload, order, isolation, delivery, stop, recovery and unknown checks passed')
|
|
} finally {
|
|
if (!resolve(root).startsWith(workspace + '\\') && !resolve(root).startsWith(workspace + '/')) throw new Error('Unsafe test cleanup path')
|
|
await fs.rm(root, { recursive: true, force: true })
|
|
}
|