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() const immediate = createPromptQueue({ ...options, file: join(root, 'immediate.json'), blocked: async () => true, deliverNow: (_payload, id) => new Promise((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 }) }