/** * 场景脚本公共库:HTTP 客户端、断言、通过真实接口建主人 / 公会 / Agent、对账、按场景清理。 * * 前提:voko-server 已启动(VOKO_API,默认 http://localhost:9201),.env 已配置真实 SMTP; * VOKO_TEST_EMAIL、VOKO_ADMIN_EMAIL / VOKO_ADMIN_PASSWORD。缺任一项输出 BLOCKED 并以 2 退出。 */ import { randomUUID } from 'node:crypto' import { rm } from 'node:fs/promises' import { fileURLToPath } from 'node:url' import { createBillingRepo, createLedger } from '@drama/billing' import { createDb } from '@drama/db' import { verifyConservation } from '@voko/clearing' import { databaseUrlFromArgs } from '../../../src/db/env-file.js' const api = (process.env.VOKO_API || 'http://localhost:9201').replace(/\/$/, '') // 与服务端 env.js 的缺省值一致:本地交付文件的存储目录。 const filesDir = process.env.VOKO_FILES_DIR || fileURLToPath(new URL('../../../data/files', import.meta.url)) /** * @param {string} name 场景名,用于测试邮箱与公会 slug * @param {(world: ReturnType) => Promise} body */ export async function runScenario(name, body) { const missing = ['VOKO_TEST_EMAIL', 'VOKO_ADMIN_EMAIL', 'VOKO_ADMIN_PASSWORD'].filter((key) => !process.env[key]) if (missing.length > 0) { console.log(JSON.stringify({ status: 'BLOCKED', missing })) process.exit(2) } const db = createDb({ url: databaseUrlFromArgs(process.argv.slice(2)), pool: { min: 1, max: 2 } }) const world = createWorld(name, db) try { await body(world) const problems = await verifyConservation(db) world.check('对账:托管与账本守恒', problems.length === 0, problems) console.log(JSON.stringify({ status: 'PASS', scenario: name, results: world.results }, null, 2)) } catch (err) { console.log(JSON.stringify({ status: 'FAIL', scenario: name, results: world.results, error: err instanceof Error ? err.message : String(err) }, null, 2)) process.exitCode = 1 } finally { await world.cleanup() await db.destroy() } } /** * @param {string} name * @param {import('knex').Knex} db */ function createWorld(name, db) { const run = Date.now() const [local, domain] = String(process.env.VOKO_TEST_EMAIL).split('@') const results = [] const emails = [] const userIds = [] const guildIds = [] const agentIds = [] let adminToken = null /** * @param {string} label * @param {boolean} pass * @param {unknown} [detail] */ function check(label, pass, detail) { results.push({ name: label, pass, ...(pass ? {} : { detail }) }) if (!pass) throw new Error(`FAILED: ${label} ${JSON.stringify(detail)}`) } /** * @param {string} method * @param {string} path * @param {{ body?: unknown, token?: string, idempotency_key?: string }} [opts] */ async function call(method, path, opts = {}) { const response = await fetch(`${api}${path}`, { method, headers: { ...(opts.body === undefined ? {} : { 'content-type': 'application/json' }), ...(opts.token === undefined ? {} : { authorization: `Bearer ${opts.token}` }), ...(opts.idempotency_key === undefined ? {} : { 'idempotency-key': opts.idempotency_key }), }, body: opts.body === undefined ? undefined : JSON.stringify(opts.body), }) return { status: response.status, json: await response.json() } } /** 等一封真实投递的邮件落库(发信不阻塞接口返回),最多 30 秒。 */ async function waitForEmail(where) { for (let i = 0; i < 30; i += 1) { const log = await db('email_delivery_logs').where(where).orderBy('created_at', 'desc').first() if (log !== undefined) return log await new Promise((resolve) => setTimeout(resolve, 1000)) } throw new Error(`email not delivered within 30s: ${JSON.stringify(where)}`) } async function admin() { if (adminToken === null) { const login = await call('POST', '/api/voko/auth/login', { body: { method: 'password', email: process.env.VOKO_ADMIN_EMAIL, password: process.env.VOKO_ADMIN_PASSWORD } }) check('admin login', login.status === 200, login) adminToken = login.json.access_token } return adminToken } /** 一个已通过加入申请的主人:真实验证码注册 → 提交申请 → 运营通过。 */ async function createOwner(tag) { const email = `${local}+voko-${name}-${tag}-${run}@${domain}`.toLowerCase() emails.push(email) const sent = await call('POST', '/api/voko/auth/send-code', { body: { email, locale: 'en' } }) check(`主人 ${tag}:send-code`, sent.status === 200, sent) const codeLog = await waitForEmail({ to_email: email, template: 'verify-code', status: 'sent' }) const registered = await call('POST', '/api/voko/auth/register', { body: { email, password: `pw-${randomUUID()}`, verification_code: JSON.parse(codeLog.metadata_json).code, locale: 'en', handle: `${name}-${tag}-${run}`.slice(0, 32) } }) check(`主人 ${tag}:注册`, registered.status === 200, registered) const owner = { user_id: registered.json.user.id, token: registered.json.access_token, email } userIds.push(owner.user_id) const request = await call('POST', '/api/voko/access-requests', { token: owner.token, body: { note: `${name} scenario` } }) const approved = await call('POST', `/api/voko/admin/access-requests/${request.json.access_request.id}/review`, { token: await admin(), body: { decision: 'approved' } }) check(`主人 ${tag}:申请通过`, approved.status === 200, approved) return owner } /** `settings` 可覆盖 require_lead_approval 与四个时间规则(分钟级 / 秒级的公会用于时间类用例)。 */ async function createGuild(tag, leadUserId, settings = {}) { const guild = await call('POST', '/api/voko/admin/guilds', { token: await admin(), body: { name: `${name} ${tag}`, slug: `${name}-${tag}-${run}`, lead_user_id: leadUserId, require_lead_approval: false, ...settings } }) check(`建公会 ${tag}`, guild.status === 200, guild) guildIds.push(guild.json.guild.id) return guild.json.guild.id } /** 注册一个 Agent 并由主人认领进公会;返回 { id, key }。 */ async function createAgent(tag, owner, guildId, role = 'member') { const registered = await call('POST', '/v1/agents/register', { body: { name: `${name}-${tag}-${run}`, skills: ['writing'] } }) check(`Agent ${tag}:注册`, registered.status === 200, registered) const data = registered.json.data agentIds.push(data.agent_id) const claimed = await call('POST', `/api/voko/claims/${data.claim_url.split('/claim/')[1]}`, { token: owner.token, body: { verification_code: data.verification_code, guild_id: guildId, role } }) check(`Agent ${tag}:认领`, claimed.status === 200 && claimed.json.agent.status === 'active', claimed) return { id: data.agent_id, key: data.api_key } } /** 只注册不认领的 Agent(用于「未认领 / 无权认领」的用例);结束时一并清理。 */ async function registerAgent(tag) { const registered = await call('POST', '/v1/agents/register', { body: { name: `${name}-${tag}-${run}`, skills: [] } }) check(`Agent ${tag}:注册`, registered.status === 200, registered) agentIds.push(registered.json.data.agent_id) return registered.json.data } /** 运营给 Agent 钱包调账(公司预算走人工调账,reason=admin_adjustment)。 */ async function grantBudget(agentId, amount) { const agent = await db('vk_agents').where({ id: agentId }).first('wallet_user_id') await createLedger(createBillingRepo(db)).grant({ user_id: agent.wallet_user_id, amount, source: 'admin', reason: 'admin_adjustment', note: `${name} scenario budget`, idempotency_key: `scenario:${name}:${run}:${agentId}:${amount}` }) } /** 发布一个任务,返回接口响应。 */ function publish(agent, overrides = {}, idempotencyKey = randomUUID()) { return call('POST', '/v1/tasks', { token: agent.key, idempotency_key: idempotencyKey, body: { title: 'Translate onboarding email', spec: 'Translate the onboarding email to Japanese.', template: 'piecework', reward: 10, deadline: new Date(Date.now() + 86400_000).toISOString(), acceptance: [{ id: 'c1', text: 'Natural Japanese' }], ...overrides }, }) } async function cleanup() { const taskIds = guildIds.length === 0 ? [] : (await db('vk_tasks').whereIn('guild_id', guildIds).select('id')).map((row) => row.id) const wallets = agentIds.length === 0 ? [] : (await db('vk_agents').whereIn('id', agentIds).select('wallet_user_id')).map((row) => row.wallet_user_id) if (taskIds.length > 0) { const fileIds = (await db('vk_files').whereIn('task_id', taskIds).select('id')).map((row) => row.id) for (const id of fileIds) await rm(`${filesDir}/${id}`, { force: true }) await db('vk_files').whereIn('task_id', taskIds).del() await db('vk_review_checks').whereIn('review_id', db('vk_reviews').whereIn('task_id', taskIds).select('id')).del() for (const table of ['vk_task_acceptance_items', 'vk_task_holds', 'vk_task_entries', 'vk_comments', 'vk_checkpoints', 'vk_submissions', 'vk_reviews', 'vk_disputes', 'vk_transfers', 'vk_help_requests', 'vk_tips', 'vk_escrow_payouts', 'vk_escrows']) await db(table).whereIn('task_id', taskIds).del() await db('vk_jobs').whereIn('ref_id', taskIds).del() await db('vk_jobs').where({ kind: 'help_notify' }).whereNotIn('ref_id', db('vk_help_requests').select('id')).del() await db('vk_tasks').whereIn('id', taskIds).del() } if (guildIds.length > 0) { await db('vk_webhook_deliveries').whereIn('event_id', db('vk_events').whereIn('guild_id', guildIds).select('id')).del() await db('vk_events').whereIn('guild_id', guildIds).del() await db('vk_guild_members').whereIn('guild_id', guildIds).del() await db('vk_guilds').whereIn('id', guildIds).del() } if (agentIds.length > 0) { await db('billing_ledger').whereIn('user_id', wallets).del() await db('vk_api_keys').whereIn('agent_id', agentIds).del() await db('vk_agent_titles').whereIn('agent_id', agentIds).del() await db('vk_agent_claims').whereIn('agent_id', agentIds).del() await db('vk_agents').whereIn('id', agentIds).del() await db('users').whereIn('id', wallets).del() } if (userIds.length > 0) { await db('vk_jobs').where({ kind: 'daily_digest' }).whereIn('ref_id', userIds).del() await db('vk_access_requests').whereIn('user_id', userIds).del() await db('vk_owner_settings').whereIn('user_id', userIds).del() await db('vk_owner_handles').whereIn('user_id', userIds).del() await db('auth_sessions').whereIn('user_id', userIds).del() await db('users').whereIn('id', userIds).del() } if (emails.length > 0) { await db('auth_codes').whereIn('email', emails).del() await db('email_delivery_logs').whereIn('to_email', emails).del() } } /** 轮询任务详情直到满足条件(等真实 worker 执行),最多 `seconds` 秒;返回最后一次读到的任务。 */ async function waitForTask(agent, taskId, predicate, seconds = 40) { let task = null for (let i = 0; i < seconds; i += 1) { task = (await call('GET', `/v1/tasks/${taskId}`, { token: agent.key })).json.data if (predicate(task)) return task await new Promise((resolve) => setTimeout(resolve, 1000)) } return task } const wallet = async (agent) => (await call('GET', '/v1/wallet', { token: agent.key })).json.data const SUBMISSION = { summary: 'Translated; see the doc.', artifacts: [{ type: 'link', url: 'https://example.com/doc/1', note: 'translation' }] } const submit = (agent, taskId, summary = SUBMISSION.summary) => call('POST', `/v1/tasks/${taskId}/submit`, { token: agent.key, body: { ...SUBMISSION, summary } }) /** 上传一个交付文件(原始字节),返回接口响应。 */ async function uploadFile(agent, taskId, filename, content) { const response = await fetch(`${api}/v1/tasks/${taskId}/files?filename=${encodeURIComponent(filename)}`, { method: 'POST', headers: { authorization: `Bearer ${agent.key}`, 'content-type': 'application/octet-stream' }, body: content }) return { status: response.status, json: await response.json() } } /** 下载一个交付文件;成功时返回正文文本。 */ async function downloadFile(agent, fileId) { const response = await fetch(`${api}/v1/files/${fileId}`, { headers: { authorization: `Bearer ${agent.key}` } }) return { status: response.status, disposition: response.headers.get('content-disposition'), text: response.status === 200 ? await response.text() : null } } const claim = (agent, taskId) => call('POST', `/v1/tasks/${taskId}/claim`, { token: agent.key }) /** 按清单逐条验收;`failing` 是不通过的清单项 id。 */ const review = (agent, taskId, action, ids, failing = [], extra = {}) => call('POST', `/v1/tasks/${taskId}/review`, { token: agent.key, body: { action, checks: ids.map((id) => ({ id, pass: !failing.includes(id), reason: failing.includes(id) ? 'Not met yet.' : 'Looks good.' })), ...extra }, }) return { db, results, check, call, admin, waitForEmail, createOwner, createGuild, createAgent, registerAgent, grantBudget, publish, waitForTask, wallet, submit, claim, review, uploadFile, downloadFile, cleanup } }