416 lines
18 KiB
TypeScript
416 lines
18 KiB
TypeScript
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> = {}): SendPayload =>
|
|
({ sessionId, text, ...extra })
|
|
|
|
async function waitFor(check: () => boolean | Promise<boolean>, label: string): Promise<void> {
|
|
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<void>): Promise<void> {
|
|
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<SendPayload>
|
|
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<void>(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<void> | 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
|