AgentServer/apps/server/test/team-jobs.test.ts

// Bot 工作组二期（docs/design/13-bot-teams.md §5、§7、§12.2）：能力卡片、组长申请加入机器人、任务与子任务、限额、替身组员、逐步执行
import fs from 'node:fs'
import path from 'node:path'
import pino from 'pino'
import { afterAll, beforeAll, describe, expect, it } from 'vitest'
import type { JobDto, ProviderStatusDto, TaskDto, TeamDetailDto, TeamEvent, TeamMessage } from '@agent-server/shared'
import { createApp, type App } from '../src/app'
import { buildServer } from '../src/http/server'
import type { AgentProvider, RunContext } from '../src/providers/types'

type Call = (tool: string, args: Record<string, unknown>) => Promise<{ text: string; isError: boolean }>
/** 按机器人名称编排的假引擎：脚本可以调用协作工具（与引擎经 MCP Gateway 调用的是同一组工具） */
type Script = (input: string, call: Call, ctx: RunContext) => Promise<string>

class ScriptedProvider implements AgentProvider {
  readonly id = 'claude' as const
  scripts: Record<string, Script> = {}
  inputs: { bot: string; text: string; prompt: string }[] = []
  async run(ctx: RunContext) {
    ctx.setProviderSessionId(`fake-${ctx.sessionId}`)
    const bot = app.sessions.getRow(ctx.sessionId).title.split(' · ').at(-1)!
    this.inputs.push({ bot, text: ctx.input.text, prompt: ctx.systemPromptAppend })
    const gctx = app.gatewayTokens.get(ctx.gateway.token)!
    const call: Call = async (tool, args) => {
      const t = gctx.extraTools?.find(x => x.name === tool)
      if (!t) return { text: `没有工具 ${tool}`, isError: true }
      const r = await t.handler(t.schema.parse(args) as Record<string, unknown>, gctx)
      return { text: String(r.content[0]?.text ?? ''), isError: !!r.isError }
    }
    const reply = await (this.scripts[bot] ?? (async () => `${bot} 收到`))(ctx.input.text, call, ctx)
    ctx.emit('message.completed', { messageId: `m${this.inputs.length}`, text: reply })
    return { usage: { inputTokens: 10, outputTokens: 5, cacheReadTokens: 0, cacheWriteTokens: 0, costUsd: 0.01 } }
  }
  async status(): Promise<ProviderStatusDto> {
    return { id: 'claude', enabled: true, available: true, loggedIn: true, account: 'test', cliPath: '', version: 'x', models: ['m'], defaultModel: 'm', message: null }
  }
}

let app: App
let server: Awaited<ReturnType<typeof buildServer>>
const fake = new ScriptedProvider()
let adminCookie: Record<string, string>
let botKey: string
const ids: Record<string, string> = {}
const bk = () => ({ authorization: `Bearer ${botKey}` })
const human = { id: 'u1', name: '张三' }

async function waitFor<T>(fn: () => T | undefined | false | null, ms = 4000): Promise<T> {
  const end = Date.now() + ms
  while (Date.now() < end) {
    const v = fn()
    if (v) return v
    await new Promise(r => setTimeout(r, 20))
  }
  throw new Error('timeout')
}

const messages = (teamId: string) =>
  app.teams.events.list(teamId, 0, 5000, ['message.created']).map(e => (e as TeamEvent & { type: 'message.created' }).payload as TeamMessage)
const jobsOf = async (teamId: string, cookies?: Record<string, string>) =>
  (await server.inject({ method: 'GET', url: `/bot/v1/teams/${teamId}/jobs`, ...(cookies ? { cookies } : { headers: bk() }) })).json() as { jobs: JobDto[]; tasks: TaskDto[] }
const idle = (teamId: string) => app.teams.toDto(app.teams.getRow(teamId)).bots.every(b => b.status === 'idle')

beforeAll(async () => {
  const log = pino({ level: 'silent' })
  app = createApp(log, { providers: { claude: fake } })
  server = await buildServer(app, { logger: log })
  const login = await server.inject({ method: 'POST', url: '/admin/v1/auth/login', payload: { password: 'test-password' } })
  const c = login.cookies[0]!
  adminCookie = { [c.name]: c.value }
  const profileId = app.profiles.getBuiltin().id
  for (const [name, kind] of [['组长', 'lead'], ['数据员', 'worker'], ['文档员', 'worker'], ['禁用员', 'worker']] as const) {
    const res = await server.inject({
      method: 'POST', url: '/admin/v1/bots', cookies: adminCookie,
      payload: { name, avatar: 'kl:bot:blue', kind, description: `${name}：负责${name.slice(0, 2)}相关工作`, profileId, enabled: true },
    })
    expect(res.statusCode).toBe(201)
    ids[name] = res.json().id
  }
  const key = await server.inject({ method: 'POST', url: '/admin/v1/bot-keys', cookies: adminCookie, payload: { name: '平台A', allowedBotIds: [ids['组长'], ids['数据员'], ids['文档员']] } })
  botKey = key.json().key
})
afterAll(() => server.close())

describe('能力卡片', () => {
  it('GET /bots 返回能力卡片；组长的提示词中带有组员的卡片', async () => {
    const cards = (await server.inject({ method: 'GET', url: '/bot/v1/bots', headers: bk() })).json()
    const data = cards.find((b: { name: string }) => b.name === '数据员')
    expect(data).toMatchObject({ kind: 'worker', permissionMode: expect.any(String) })
    expect(Array.isArray(data.skills)).toBe(true)

    fake.scripts['组长'] = async () => '你好'
    const team = (await server.inject({ method: 'POST', url: '/bot/v1/teams', headers: bk(), payload: { human, leadBotId: ids['组长'], workerBotIds: [ids['数据员']] } })).json() as TeamDetailDto
    await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/messages`, headers: bk(), payload: { text: '在吗' } })
    await waitFor(() => messages(team.id).some(m => m.sender.kind === 'bot'))
    const prompt = fake.inputs.filter(i => i.bot === '组长').at(-1)!.prompt
    expect(prompt).toContain('当前组员的能力卡片')
    expect(prompt).toContain(`### 数据员（id: ${ids['数据员']}）`)
  })

  it('工作组的限额只能调低；Bot Key 不能设置调试选项', async () => {
    const tooHigh = await server.inject({ method: 'POST', url: '/bot/v1/teams', headers: bk(), payload: { human, leadBotId: ids['组长'], limits: { maxTasksPerJob: 999 } } })
    expect(tooHigh.statusCode).toBe(400)
    const debug = await server.inject({ method: 'POST', url: '/bot/v1/teams', headers: bk(), payload: { human, leadBotId: ids['组长'], debug: { stepMode: true } } })
    expect(debug.statusCode).toBe(403)
  })
})

describe('任务流程', () => {
  it('只有组长：申请加入两个机器人（经审批），拆成有依赖的子任务，组员交付文件，组长汇总交付', async () => {
    const team = (await server.inject({ method: 'POST', url: '/bot/v1/teams', headers: bk(), payload: { human, title: '周报', leadBotId: ids['组长'] } })).json() as TeamDetailDto
    let leadTurn = 0
    fake.scripts['组长'] = async (input, call) => {
      leadTurn++
      if (leadTurn === 1) {
        const catalog = await call('team_bot_catalog', {})
        expect(catalog.text).toContain('数据员')
        expect(catalog.text).not.toContain('禁用员') // 不在 Bot Key 的范围内
        for (const name of ['数据员', '文档员']) {
          const r = await call('team_request_bot', { bot_id: ids[name], reason: `需要${name}` })
          expect(r.text).toContain('已加入')
        }
        await call('job_start', { title: '周报', plan: '先统计再写报告' })
        await call('task_create', { bot_id: ids['数据员'], title: '统计', instructions: '统计销量，写到 outputs/data.csv' })
        const t2 = await call('task_create', { bot_id: ids['文档员'], title: '写报告', instructions: '根据 #1 写报告', depends_on: [1] })
        expect(t2.isError).toBe(false)
        return '已安排：数据员统计，文档员写报告'
      }
      if (input.includes('子任务 #2 已完成')) {
        const r = await call('job_complete', { summary: '周报已完成', files: ['outputs/report.md'] })
        expect(r.isError).toBe(false)
        return '交付完毕'
      }
      return '收到，等待其他子任务'
    }
    fake.scripts['数据员'] = async (input, call, ctx) => {
      if (!input.includes('【子任务')) return '我是数据员'
      fs.mkdirSync(path.join(ctx.workspaceDir, 'outputs'), { recursive: true })
      fs.writeFileSync(path.join(ctx.workspaceDir, 'outputs/data.csv'), 'a,b\n1,2\n')
      await call('task_complete', { result: '销量 100', files: ['outputs/data.csv'] })
      return '完成'
    }
    fake.scripts['文档员'] = async (input, call, ctx) => {
      // 依赖的子任务结果随输入一起交给它
      expect(input).toContain('【依赖的子任务 #')
      expect(input).toContain('销量 100')
      fs.writeFileSync(path.join(ctx.workspaceDir, 'outputs/report.md'), '# 周报')
      await call('task_complete', { result: '报告已写好', files: ['outputs/report.md'] })
      return '完成'
    }

    await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/messages`, headers: bk(), payload: { text: '帮我做本周周报' } })
    // 两次申请都显示审批卡片，由成员批准
    for (let i = 0; i < 2; i++) {
      const pending = await waitFor(() => app.teams.detail(app.teams.getRow(team.id)).pendingInteractions[0])
      expect(pending.request.kind === 'approval' && pending.request.title).toContain('申请让')
      const r = await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/interactions/${pending.request.interactionId}`, headers: bk(), payload: { decision: 'allow_once' } })
      expect(r.statusCode).toBe(200)
    }

    const done = await waitFor(() => {
      const j = app.teams.jobs.jobs(team.id)[0]
      return j?.status === 'completed' && idle(team.id) ? j : null
    })
    expect(done.summary).toBe('周报已完成')
    expect(done.files.map(f => f.path)).toEqual(['outputs/report.md'])
    const { jobs, tasks } = await jobsOf(team.id)
    expect(jobs[0]).toMatchObject({ status: 'completed', taskCount: 2 })
    expect(jobs[0]!.leadRuns).toBeGreaterThanOrEqual(2)
    expect(tasks.map(t => [t.no, t.status])).toEqual([[1, 'completed'], [2, 'completed']])
    expect(tasks[0]!.files.map(f => f.path)).toEqual(['outputs/data.csv'])
    expect(tasks[1]!.dependsOn).toEqual([1])

    const msgs = messages(team.id)
    expect(msgs.some(m => m.card?.kind === 'job' && m.card.event === 'started')).toBe(true)
    expect(msgs.filter(m => m.card?.kind === 'task' && m.card.event === 'assigned')).toHaveLength(2)
    // 组员的结果发给组长（唤醒组长）
    const result = msgs.find(m => m.card?.kind === 'task' && m.card.event === 'completed' && m.card.no === 1)!
    expect(result.targets).toEqual([ids['组长']])
    expect(result.files.map(f => f.path)).toEqual(['outputs/data.csv'])
    expect(msgs.some(m => m.card?.kind === 'job' && m.card.event === 'completed' && m.files.length === 1)).toBe(true)
    expect(msgs.filter(m => m.sender.kind === 'system').some(m => m.text.includes('成员已批准'))).toBe(true)

    // 用量按机器人、任务、子任务统计
    const usage = (await server.inject({ method: 'GET', url: `/bot/v1/teams/${team.id}/usage`, headers: bk() })).json()
    expect(usage.total.runs).toBeGreaterThanOrEqual(4)
    expect(usage.jobs[0].runs).toBe(usage.total.runs)
    expect(usage.tasks).toHaveLength(2)
    expect(usage.bots.map((b: { botId: string }) => b.botId).sort()).toEqual([ids['组长'], ids['数据员'], ids['文档员']].sort())
  })

  it('同一时间只有一个任务；组员不在子任务中时 task_complete 失败', async () => {
    const team = (await server.inject({ method: 'POST', url: '/bot/v1/teams', headers: bk(), payload: { human, leadBotId: ids['组长'], workerBotIds: [ids['数据员']] } })).json() as TeamDetailDto
    let second: { text: string; isError: boolean } | undefined
    fake.scripts['组长'] = async (_input, call) => {
      await call('job_start', { title: 'A', plan: '' })
      second = await call('job_start', { title: 'B', plan: '' })
      return 'ok'
    }
    let direct: { text: string; isError: boolean } | undefined
    fake.scripts['数据员'] = async (_input, call) => {
      direct = await call('task_complete', { result: 'x' })
      return '直接回复'
    }
    await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/messages`, headers: bk(), payload: { text: '开始' } })
    await waitFor(() => second)
    expect(second).toMatchObject({ isError: true })
    expect(second!.text).toContain('已有进行中的任务')
    await waitFor(() => idle(team.id) && messages(team.id).some(m => m.sender.kind === 'bot' && m.text === 'ok'))
    await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/messages`, headers: bk(), payload: { text: '单项工作', mentions: [ids['数据员']] } })
    await waitFor(() => direct)
    expect(direct).toMatchObject({ isError: true })
    // 成员取消任务
    const job = app.teams.jobs.active(team.id)!
    expect((await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/jobs/${job.id}/cancel`, headers: bk() })).statusCode).toBe(200)
    expect(app.teams.jobs.job(job.id).status).toBe('cancelled')
    expect((await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/jobs/${job.id}/cancel`, headers: bk() })).statusCode).toBe(409)
  })
})

describe('限额', () => {
  it('超出子任务数上限时任务暂停，成员选择继续后放宽一倍、组长被唤醒；也可以取消', async () => {
    const team = (await server.inject({
      method: 'POST', url: '/bot/v1/teams', headers: bk(),
      payload: { human, leadBotId: ids['组长'], workerBotIds: [ids['数据员']], limits: { maxTasksPerJob: 1 } },
    })).json() as TeamDetailDto
    expect(team.limits.maxTasksPerJob).toBe(1)
    const results: boolean[] = []
    let turn = 0
    fake.scripts['组长'] = async (_input, call) => {
      turn++
      if (turn === 1) {
        await call('job_start', { title: '大任务', plan: '' })
        results.push((await call('task_create', { bot_id: ids['数据员'], title: 't1', instructions: 'x' })).isError)
        results.push((await call('task_create', { bot_id: ids['数据员'], title: 't2', instructions: 'x' })).isError)
        return '分派中'
      }
      return `第 ${turn} 轮`
    }
    fake.scripts['数据员'] = async (_input, call) => {
      await call('task_complete', { result: 'ok' })
      return 'done'
    }
    await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/messages`, headers: bk(), payload: { text: '开始' } })
    const blocked = await waitFor(() => {
      const j = app.teams.jobs.active(team.id)
      return j?.status === 'blocked' && idle(team.id) ? j : null
    })
    expect(results).toEqual([false, true])
    expect(blocked.blockedBy).toBe('maxTasksPerJob')
    expect(messages(team.id).some(m => m.card?.kind === 'limit')).toBe(true)
    // 暂停期间组员的结果不会唤醒组长
    const leadTurns = turn
    await new Promise(r => setTimeout(r, 100))
    expect(turn).toBe(leadTurns)

    expect((await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/jobs/${blocked.id}/continue`, headers: bk() })).statusCode).toBe(200)
    await waitFor(() => turn > leadTurns && idle(team.id))
    const job = app.teams.jobs.job(blocked.id)
    expect(job).toMatchObject({ status: 'running', blockedBy: null })
    expect(job.limits.maxTasksPerJob).toBe(2)
    expect((await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/jobs/${blocked.id}/continue`, headers: bk() })).statusCode).toBe(409)
  })

  it('组长执行轮数达到上限时不再唤醒组长', async () => {
    const team = (await server.inject({
      method: 'POST', url: '/bot/v1/teams', headers: bk(),
      payload: { human, leadBotId: ids['组长'], limits: { maxLeadRunsPerJob: 1 } },
    })).json() as TeamDetailDto
    let turns = 0
    fake.scripts['组长'] = async (_input, call) => {
      turns++
      if (turns === 1) await call('job_start', { title: 'x', plan: '' })
      return 'ok'
    }
    await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/messages`, headers: bk(), payload: { text: '一' } })
    await waitFor(() => turns === 1 && idle(team.id))
    await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/messages`, headers: bk(), payload: { text: '二' } })
    const job = await waitFor(() => app.teams.jobs.active(team.id)?.status === 'blocked' && app.teams.jobs.active(team.id))
    expect(job.blockedBy).toBe('maxLeadRunsPerJob')
    expect(turns).toBe(1)
  })
})

describe('Bot Playground 的调试功能', () => {
  it('全部组员设为替身时，由管理员填写结果完成一项任务', async () => {
    const created = await server.inject({
      method: 'POST', url: '/bot/v1/teams', cookies: adminCookie,
      payload: { human, leadBotId: ids['组长'], workerBotIds: [ids['数据员']], debug: { stubBotIds: [ids['数据员']] } },
    })
    expect(created.statusCode).toBe(201)
    const team = created.json() as TeamDetailDto
    fake.scripts['组长'] = async (input, call) => {
      if (input.includes('子任务 #')) {
        expect(input).not.toContain('替身')
        await call('job_complete', { summary: `汇总：${input.includes('算出 42') ? '42' : '?'}` })
        return '完成'
      }
      await call('job_start', { title: '替身测试', plan: '' })
      await call('task_create', { bot_id: ids['数据员'], title: '算数', instructions: '算 6×7' })
      return '已分派'
    }
    const before = fake.inputs.filter(i => i.bot === '数据员').length
    await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/messages`, cookies: adminCookie, payload: { text: '开始' } })
    const tasks = await waitFor(() => {
      const t = app.teams.jobs.tasks(team.id, { status: ['stub'] })
      return t.length ? t : null
    })
    expect(messages(team.id).some(m => m.card?.kind === 'stub')).toBe(true)
    // Bot Key 不能填写替身结果
    expect((await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/tasks/${tasks[0]!.id}/stub`, headers: bk(), payload: { outcome: 'completed', result: 'x' } })).statusCode).toBe(404)
    const r = await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/tasks/${tasks[0]!.id}/stub`, cookies: adminCookie, payload: { outcome: 'completed', result: '算出 42' } })
    expect(r.statusCode).toBe(200)
    const job = await waitFor(() => app.teams.jobs.jobs(team.id)[0]?.status === 'completed' && app.teams.jobs.jobs(team.id)[0])
    expect(job.summary).toBe('汇总：42')
    // 替身的子任务没有真正执行
    expect(fake.inputs.filter(i => i.bot === '数据员').length).toBe(before)
  })

  it('逐步执行：每次唤醒前停下，点"继续"后才执行；关闭后放行', async () => {
    const team = (await server.inject({
      method: 'POST', url: '/bot/v1/teams', cookies: adminCookie,
      payload: { human, workerBotIds: [ids['数据员']], debug: { stepMode: true } },
    })).json() as TeamDetailDto
    fake.scripts['数据员'] = async input => `做了：${input.split('\n').at(-1)}`
    await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/messages`, cookies: adminCookie, payload: { text: '第一步' } })
    const step = await waitFor(() => app.teams.detail(app.teams.getRow(team.id)).pendingSteps[0])
    expect(step.reason).toContain('1 条新消息')
    await new Promise(r => setTimeout(r, 50))
    expect(messages(team.id).some(m => m.sender.kind === 'bot')).toBe(false)
    expect((await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/steps/${step.id}/continue`, cookies: adminCookie })).statusCode).toBe(200)
    await waitFor(() => messages(team.id).some(m => m.sender.kind === 'bot'))
    expect(app.teams.detail(app.teams.getRow(team.id)).pendingSteps).toEqual([])

    await waitFor(() => idle(team.id))
    await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/messages`, cookies: adminCookie, payload: { text: '第二步' } })
    await waitFor(() => app.teams.detail(app.teams.getRow(team.id)).pendingSteps[0])
    const patched = await server.inject({ method: 'PATCH', url: `/bot/v1/teams/${team.id}`, cookies: adminCookie, payload: { debug: { stepMode: false } } })
    expect(patched.json().debug.stepMode).toBe(false)
    await waitFor(() => messages(team.id).filter(m => m.sender.kind === 'bot').length === 2)
  })
})