import assert from 'node:assert/strict' import { promises as fs } from 'node:fs' import { hostname } from 'node:os' import { join, resolve } from 'node:path' import { createPromptQueue } from '../src/main/prompt-queue' import { resolveQueuedRequestSettings } from '../src/main/agent/queued-request-settings' import type { SendPayload } from '../src/shared/ipc' 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-adversarial-')) const results: Array<{ name: string; pass: boolean; error?: string }> = [] const tick = (ms = 20) => new Promise(resolve => setTimeout(resolve, ms)) const payload = (sessionId: string, text: string, extra: Partial = {}): SendPayload => ({ sessionId, text, ...extra }) async function waitFor(check: () => boolean | Promise, label: string): Promise { for (let attempt = 0; attempt < 150; attempt++) { if (await check()) return await tick(10) } throw new Error('Timeout: ' + label) } async function verify(name: string, run: () => Promise): Promise { try { await run() results.push({ name, pass: true }) } catch (error) { results.push({ name, pass: false, error: error instanceof Error ? error.message : String(error) }) } } function options(file: string, input: { blocked?: (sessionId: string) => boolean resolveDispatch?: (payload: SendPayload, id: string) => SendPayload | Promise deliver?: (payload: SendPayload, id: string) => Promise<{ completion: Promise<{ status: 'done' | 'cancelled' | 'error' }> }> changed?: (snapshot: PromptQueueSnapshot) => void } = {}) { return { file, validate: async () => undefined, blocked: async sessionId => input.blocked?.(sessionId) ?? true, ...(input.resolveDispatch ? { resolveDispatch: input.resolveDispatch } : {}), deliver: input.deliver ?? (async () => ({ completion: new Promise<{ status: 'done' }>(() => undefined) })), changed: input.changed ?? (() => undefined) } } try { await verify('parallel same-instance enqueues preserve exactly 19 items', async () => { const file = join(root, 'cap.json') const queue = createPromptQueue(options(file)) const outcomes = await Promise.allSettled(Array.from({ length: 25 }, (_, index) => queue.enqueue(payload('cap', String(index))))) assert.equal(outcomes.filter(item => item.status === 'fulfilled').length, 19) assert.equal(outcomes.filter(item => item.status === 'rejected').length, 6) const snapshot = await queue.get('cap') assert.equal(snapshot.items.length, 19) assert.equal(new Set(snapshot.items.map(item => item.id)).size, 19) }) await verify('duplicate take is serialized to one winner', async () => { const file = join(root, 'take.json') const queue = createPromptQueue(options(file)) await queue.enqueue(payload('take', 'only')) const id = (await queue.get('take')).items[0].id const outcomes = await Promise.allSettled([queue.take('take', id), queue.take('take', id)]) assert.equal(outcomes.filter(item => item.status === 'fulfilled').length, 1) assert.equal(outcomes.filter(item => item.status === 'rejected').length, 1) assert.equal((await queue.get('take')).items.length, 0) }) await verify('delivering item is immutable and notify does not duplicate delivery', async () => { const file = join(root, 'delivery-lock.json') let releaseDelivery!: () => void let releaseCompletion!: (value: { status: 'done' }) => void const deliveryGate = new Promise(resolve => { releaseDelivery = resolve }) const completion = new Promise<{ status: 'done' }>(resolve => { releaseCompletion = resolve }) const delivered: string[] = [] const queue = createPromptQueue(options(file, { blocked: () => false, deliver: async (_value, id) => { delivered.push(id) await deliveryGate return { completion } } })) await queue.enqueue(payload('delivery', 'first')) await queue.enqueue(payload('delivery', 'second')) await waitFor(() => delivered.length === 1, 'first delivery reserve') const delivering = await queue.get('delivery') assert.equal(delivering.items[0].status, 'delivering') const observer = createPromptQueue(options(file, { blocked: () => true })) assert.equal((await observer.get('delivery')).items[0].status, 'delivering', 'fresh reads in the same process must preserve an owned active delivery') await assert.rejects(queue.clear('delivery'), /正在交付/) const first = delivering.items[0].id const second = delivering.items[1].id await queue.remove('delivery', second) await assert.rejects(queue.take('delivery', first), /交付/) await assert.rejects(queue.reorder('delivery', [first]), /排序/) queue.notify('delivery') queue.notify('delivery') queue.notify('delivery') await tick(50) assert.equal(delivered.length, 1) releaseDelivery() releaseCompletion({ status: 'done' }) await waitFor(async () => (await queue.get('delivery')).items.length === 0, 'delivery cleanup') }) await verify('clear removes a non-delivering session queue and is idempotent', async () => { const file = join(root, 'clear.json') const queue = createPromptQueue(options(file)) await queue.enqueue(payload('clear', 'one')) await queue.enqueue(payload('clear', 'two')) const cleared = await queue.clear('clear') assert.deepEqual(cleared.items, []) assert.equal(cleared.paused, false) assert.deepEqual((await createPromptQueue(options(file)).get('clear')).items, []) assert.deepEqual((await queue.clear('clear')).items, []) }) await verify('first enqueue after an empty queue was paused resumes a new run without unpausing existing work', async () => { const file = join(root, 'empty-pause.json') let blocked = true let finish!: (value: { status: 'done' }) => void const delivered: string[] = [] const queue = createPromptQueue(options(file, { blocked: () => blocked, deliver: async value => { delivered.push(value.text) return { completion: new Promise<{ status: 'done' }>(resolve => { finish = resolve }) } } })) await queue.pause('lifecycle', '上一条运行已停止') assert.equal((await queue.get('lifecycle')).paused, true) blocked = false await queue.enqueue(payload('lifecycle', 'new run')) await waitFor(() => delivered.length === 1, 'new run first delivery') const active = await queue.get('lifecycle') assert.equal(active.paused, false, 'a new run must not inherit a pause left on an empty queue') assert.equal(active.reason, undefined) finish({ status: 'done' }) await waitFor(async () => (await queue.get('lifecycle')).items.length === 0, 'first delivery cleanup') blocked = true await queue.enqueue(payload('lifecycle', 'queued while busy')) await queue.pause('lifecycle', '用户停止当前队列') await queue.enqueue(payload('lifecycle', 'later item')) const retained = await queue.get('lifecycle') assert.equal(retained.paused, true, 'a non-empty queue must retain its pause') assert.equal(retained.reason, '用户停止当前队列') assert.deepEqual(retained.items.map(item => item.payload.text), ['queued while busy', 'later item']) assert.deepEqual(delivered, ['new run']) }) await verify('incoming payload is validated before host side effects and persistence', async () => { const file = join(root, 'incoming-validation.json') let hostCalls = 0 const queue = createPromptQueue({ ...options(file), validate: async () => { hostCalls++ } }) await assert.rejects( queue.enqueue({ sessionId: 'incoming', text: '/compact', command: 'compact' } as SendPayload), /追加消息无效.*命令/ ) assert.equal(hostCalls, 0, 'invalid transport payload must not reach host validation') await assert.rejects(fs.access(file), (error: NodeJS.ErrnoException) => error.code === 'ENOENT') await assert.rejects( queue.enqueue({ sessionId: 'incoming', text: 'label', skillIds: ['skill:a'], capabilityLabels: { 'skill:a': { name: 'not-a-string' } } } as unknown as SendPayload), /追加消息无效.*能力标签/ ) assert.equal(hostCalls, 0, 'malformed display labels must not reach host validation') await assert.rejects(fs.access(file), (error: NodeJS.ErrnoException) => error.code === 'ENOENT') }) await verify('host-rebuilt capability labels are validated before persistence', async () => { const file = join(root, 'capability-labels.json') const queue = createPromptQueue({ ...options(file), validate: async value => { value.capabilityLabels = { 'skill:alpha': 'Alpha skill', 'mcp:server': 'Server' } } }) await queue.enqueue(payload('labels', 'use labels', { skillIds: ['alpha'], mcpServerIds: ['server'], capabilityLabels: { 'skill:alpha': 'untrusted renderer label' } })) const stored = (await queue.get('labels')).items[0].payload.capabilityLabels assert.deepEqual(stored, { 'skill:alpha': 'Alpha skill', 'mcp:server': 'Server' }) assert.deepEqual( (await createPromptQueue(options(file)).get('labels')).items[0].payload.capabilityLabels, stored, 'validated labels must survive a fresh disk read' ) }) await verify('two queue instances cannot overwrite one another', async () => { const file = join(root, 'two-instances.json') const a = createPromptQueue(options(file)) const b = createPromptQueue(options(file)) await Promise.all([a.get('shared'), b.get('shared')]) await Promise.all([ a.enqueue(payload('shared', 'from-a')), b.enqueue(payload('shared', 'from-b')) ]) await tick(80) const fresh = createPromptQueue(options(file)) const snapshot = await fresh.get('shared') assert.deepEqual(snapshot.items.map(item => item.payload.text).sort(), ['from-a', 'from-b']) }) await verify('corrupt queue cannot redirect or duplicate item identity', async () => { const file = join(root, 'corrupt.json') await fs.writeFile(file, JSON.stringify({ version: 1, queues: [{ sessionId: 'owner', revision: 1, paused: false, items: [ { id: 'same', createdAt: 1, status: 'pending', payload: payload('other', 'redirected') }, { id: 'same', createdAt: 2, status: 'pending', payload: payload('other', 'duplicate') } ] }] })) const queue = createPromptQueue(options(file)) await assert.rejects(queue.get('owner'), /损坏|重复|会话/) }) await verify('invalid revision, status and payload permission are rejected', async () => { const base = { version: 1, queues: [{ sessionId: 'invalid', revision: 1, paused: false, items: [{ id: 'valid-id', createdAt: 1, status: 'pending', payload: payload('invalid', 'text') }] }] } for (const [name, mutate] of [ ['revision', (value: any) => { value.queues[0].revision = -1 }], ['status', (value: any) => { value.queues[0].items[0].status = 'mystery' }], ['permission', (value: any) => { value.queues[0].items[0].payload.permissionMode = 'admin' }], ['capability-label', (value: any) => { value.queues[0].items[0].payload.capabilityLabels = { 'skill:a': { name: 'bad' } } }] ] as const) { const file = join(root, `invalid-${name}.json`) const value = structuredClone(base) mutate(value) await fs.writeFile(file, JSON.stringify(value)) await assert.rejects(createPromptQueue(options(file)).get('invalid'), /损坏/) } }) await verify('post-delivery removal failure becomes paused unknown instead of stuck delivering', async () => { const file = join(root, 'removal-failure.json') let sabotage: Promise | undefined const queue = createPromptQueue(options(file, { blocked: sessionId => sessionId !== 'removal', changed: snapshot => { if (!sabotage && snapshot.sessionId === 'removal' && snapshot.items[0]?.status === 'delivering') { sabotage = (async () => { await fs.rm(file, { force: true }) await fs.mkdir(file) })() } }, deliver: async () => { await sabotage return { completion: Promise.resolve({ status: 'done' }) } } })) await queue.enqueue(payload('bystander', 'must survive fallback')) await queue.enqueue(payload('removal', 'already persisted elsewhere')) await waitFor(() => !!sabotage, 'removal failure sabotage') await waitFor(async () => { try { const current = await queue.get('removal') return current.paused && current.items[0]?.status === 'unknown' } catch { return false } }, 'removal uncertainty fallback') const snapshot = await queue.get('removal') assert.equal(snapshot.items[0].status, 'unknown', 'acknowledged delivery must become unknown when queue cleanup cannot be persisted') assert.equal(snapshot.paused, true, 'an unconfirmed queue cleanup must pause automatic dispatch') assert.equal((await queue.get('bystander')).items[0].payload.text, 'must survive fallback') const unknown = snapshot.items[0] await fs.rm(file, { recursive: true, force: true }) const taken = await queue.take('removal', unknown.id) assert.equal(taken.status, 'unknown') assert.equal(taken.payload.text, 'already persisted elsewhere') assert.deepEqual((await queue.get('removal')).items, []) assert.equal((await queue.get('bystander')).items[0].payload.text, 'must survive fallback') }) await verify('stale owner nonce is recovered without trusting a reused PID', async () => { const file = join(root, 'owner-nonce.json') await fs.writeFile(file, JSON.stringify({ version: 1, queues: [{ sessionId: 'owner', revision: 2, paused: false, items: [{ id: 'owned-item', createdAt: 1, status: 'delivering', deliveryOwner: { pid: process.pid, nonce: 'old-process-instance', hostname: hostname(), startedAt: 1 }, payload: payload('owner', 'recover me') }] }] })) const snapshot = await createPromptQueue(options(file)).get('owner') assert.equal(snapshot.items[0].status, 'unknown') assert.equal(snapshot.paused, true) }) await verify('stale and expired lock files are reclaimed', async () => { const file = join(root, 'lock-reclaim.json') const lock = `${file}.lock` await fs.writeFile(lock, JSON.stringify({ pid: 99999999, nonce: 'dead-owner', hostname: hostname(), createdAt: Date.now(), expiresAt: Date.now() + 60_000 })) assert.deepEqual((await createPromptQueue(options(file)).get('lock')).items, []) await fs.writeFile(lock, JSON.stringify({ pid: process.pid, nonce: 'expired-owner', hostname: 'remote-test-host', createdAt: Date.now() - 60_000, expiresAt: Date.now() - 1 })) assert.deepEqual((await createPromptQueue(options(file)).get('lock')).items, []) }) await verify('live foreign lock times out with an explicit error', async () => { const file = join(root, 'lock-timeout.json') const lock = `${file}.lock` await fs.writeFile(lock, JSON.stringify({ pid: process.pid, nonce: 'live-foreign-owner', hostname: hostname(), createdAt: Date.now(), expiresAt: Date.now() + 60_000 })) try { await assert.rejects(createPromptQueue(options(file)).get('lock'), /文件锁超时/) } finally { await fs.rm(lock, { force: true }) } }) await verify('blocked notify does not create a state revision', async () => { const file = join(root, 'blocked-revision.json') const queue = createPromptQueue(options(file, { blocked: () => true })) await queue.enqueue(payload('revision', 'pending')) await tick(50) const before = await queue.get('revision') queue.notify('revision') await tick(50) const after = await queue.get('revision') assert.equal(after.revision, before.revision, 'a blocked no-op must not persist or notify a new revision') }) await verify('host resolver applies dispatch-time permission downgrade without mutating stored payload', async () => { const file = join(root, 'permission.json') let blocked = true let deliveredPermission: SendPayload['permissionMode'] assert.deepEqual(resolveQueuedRequestSettings({ queued: false, payloadPermissionMode: 'full-access', sessionPermissionMode: 'default-sandbox' }), { permissionMode: 'full-access', persistSessionPermission: true }) assert.deepEqual(resolveQueuedRequestSettings({ queued: true, payloadPermissionMode: 'full-access', sessionPermissionMode: 'default-sandbox' }), { permissionMode: 'default-sandbox', persistSessionPermission: false }) const queue = createPromptQueue(options(file, { blocked: () => blocked, resolveDispatch: value => ({ ...value, permissionMode: resolveQueuedRequestSettings({ queued: true, payloadPermissionMode: value.permissionMode, sessionPermissionMode: 'default-sandbox' }).permissionMode }), deliver: async value => { deliveredPermission = value.permissionMode return { completion: Promise.resolve({ status: 'done' }) } } })) await queue.enqueue(payload('permission', 'run', { permissionMode: 'full-access' })) assert.equal((await queue.get('permission')).items[0].payload.permissionMode, 'full-access') blocked = false queue.notify('permission') await waitFor(() => deliveredPermission !== undefined, 'permission delivery') assert.equal(deliveredPermission, 'default-sandbox') await waitFor(async () => (await queue.get('permission')).items.length === 0, 'permission delivery cleanup') }) } finally { if (!resolve(root).startsWith(workspace + '\\') && !resolve(root).startsWith(workspace + '/')) { throw new Error('Unsafe adversarial test cleanup path') } await fs.rm(root, { recursive: true, force: true }) } const failures = results.filter(result => !result.pass) console.log(JSON.stringify({ result: failures.length ? 'FAIL' : 'PASS', checks: results }, null, 2)) if (failures.length) process.exitCode = 1