// @vitest-environment node import { afterEach, beforeEach, expect, it, vi } from 'vitest' import { mkdtempSync, rmSync } from 'node:fs' import { tmpdir } from 'node:os' import { join } from 'node:path' import { randomUUID } from 'node:crypto' import { WorkspaceStore } from '../../src/server/workspaceStore' import { WorkspaceRuntime } from '../../src/server/workspaceRuntime' import { UploadStore } from '../../src/server/uploads' import { discussionRounds } from '../../src/server/workspaceDiscussion' import type { Work } from '../../src/server/workspaceScheduler' import type { WorkspaceAgent as Agent, WorkspaceConversation as Conversation, WorkspaceMessage as Message, WorkspaceRun as Run } from '../../src/shared/workspace' import type { WorkspaceNodes } from '../../src/server/workspaceGateway' const owner = 'collaboration-owner' let home: string, store: WorkspaceStore, runtime: FakeRuntime, uploads: UploadStore class FakeRuntime extends WorkspaceRuntime { calls: Array<{ agentId: string; work: Work }> = [] reply: (agent: Agent, work: Work) => string = agent => `${agent.name} 的实质结果` interrupt: (work: Work) => Promise = work => { work.status = 'interrupted'; this.saveWork('collaboration-owner', work) return Promise.resolve() } protected override async performTurn(user: string, c: Conversation, agent: Agent, work: Work): Promise { this.calls.push({ agentId: agent.id, work }) const content = this.reply(agent, work) const message: Message = { id: randomUUID(), conversationId: c.id, conversationTaskId: work.conversationTaskId, seq: 0, role: 'assistant', agentId: agent.id, agentName: agent.name, content, reasoning: '', runId: work.runId, taskId: work.id, status: 'complete', attachments: [], tools: [], createdAt: Date.now() } this.store.saveMessage(user, message) work.status = 'complete'; work.currentMessageId = message.id; this.saveWork(user, work) return message } protected override async interruptTurn(user: string, work: Work): Promise { await this.interrupt(work) } } beforeEach(() => { home = mkdtempSync(join(tmpdir(), 'yaoyao-collaboration-')); store = new WorkspaceStore(home); uploads = new UploadStore(home) runtime = new FakeRuntime(store, { requireSource: () => {} } as unknown as WorkspaceNodes, uploads) }) afterEach(() => { runtime.close(); uploads.close(); store.close(); rmSync(home, { recursive: true, force: true }) }) const bot = (name: string) => store.createAgent(owner, { name, profile: 'default' }) const direct = (id: string) => store.list(owner, 'conversation').find(c => c.kind === 'direct' && c.memberIds[0] === id)! it('settles a stop locally while remote interruption is unavailable', async () => { vi.spyOn(runtime, 'wake').mockImplementation(() => {}) const a = bot('停止挂起') const run = runtime.send(owner, direct(a.id).id, { requestId: randomUUID(), content: '等待停止' }) const work = store.list(owner, 'turn').find(item => item.runId === run.id)! work.status = 'running' work.submitted = true store.put(owner, 'turn', work.id, work) let finishInterrupt!: () => void runtime.interrupt = () => new Promise(resolve => { finishInterrupt = resolve }) await Promise.race([ runtime.stop(owner, run.id), new Promise((_, reject) => setTimeout(() => reject(new Error('stop timed out')), 50)), ]) expect(store.require(owner, 'run', run.id)).toMatchObject({ status: 'interrupted', stopRequested: true }) expect(store.require(owner, 'turn', work.id)).toMatchObject({ status: 'interrupted', cleanupPending: true }) expect(store.require(owner, 'conversation', run.conversationId).activeRunId).toBeUndefined() finishInterrupt() await vi.waitFor(() => expect(store.require(owner, 'turn', work.id).cleanupPending).toBe(false)) }) it('projects real deliveries into compact direction metadata for live events and history without rewriting raw messages', () => { vi.spyOn(runtime, 'wake').mockImplementation(() => {}) const a = bot('竹儿'), b = bot('研究员') runtime.send(owner, direct(a.id).id, { requestId: randomUUID(), content: '请核对接口' }) const work = store.list(owner, 'turn')[0]! work.status = 'running'; store.put(owner, 'turn', work.id, work) const events: Message[] = [] const unsubscribe = store.observe((_owner, event) => { if (event.type === 'message.changed') events.push(event.data as Message) }) const input = { requestId: randomUUID(), agentId: b.id, content: '请核对接口版本,确认旧客户端是否兼容。', fileIds: [] } const peer = runtime.collaboration.send(owner, work.id, input) const deliveries = events.filter(m => m.peerMessageId === peer.id) expect(deliveries.map(m => m.communication?.direction)).toEqual(['incoming', 'outgoing']) expect(deliveries.every(m => m.role === 'system' && m.communication?.content === input.content)).toBe(true) expect(deliveries[0]?.content).toContain('来自 Bot') expect(deliveries[1]?.content).toContain('已向') expect(deliveries[0]?.communication).toMatchObject({ peerId: a.id, peerName: '竹儿' }) expect(deliveries[1]?.communication).toMatchObject({ peerId: b.id, peerName: '研究员' }) const history = store.messages(owner, direct(a.id).id).find(m => m.peerMessageId === peer.id)! expect(history.communication).toEqual(deliveries[1]?.communication) expect(store.require(owner, 'message', history.id).content).toContain('已向') const ordinary: Message = { ...history, id: randomUUID(), peerMessageId: undefined, communication: undefined, content: '已向 Bot 发送只是普通文本' } expect(store.messageForDisplay(owner, ordinary).communication).toBeUndefined() unsubscribe() }) it('honors explicit round counts and finishes all eight speakers in twelve rounds', async () => { expect(discussionRounds('请讨论一轮')).toBe(1); expect(discussionRounds('请讨论三轮')).toBe(3); expect(discussionRounds('讨论一下')).toBe(2) const members = Array.from({ length: 8 }, (_, i) => bot(`成员${i}`)) const c = store.createGroup(owner, { name: '平等讨论', memberIds: members.map(a => a.id), collaborationMode: 'discussion' }) const run = runtime.send(owner, c.id, { requestId: randomUUID(), content: '请讨论十二轮' }) await vi.waitFor(() => expect(store.require(owner, 'run', run.id).status).toBe('complete')) expect(runtime.calls).toHaveLength(96) expect(runtime.calls.slice(8, 16).map(c => c.agentId)).toEqual([...members.slice(1), members[0]!].map(a => a.id)) for (const member of members) expect(runtime.calls.filter(c => c.agentId === member.id)).toHaveLength(12) }) it('selects mentioned speakers and continues after an individual failure without starting a goal', async () => { const a = bot('甲'), b = bot('乙'), c = bot('丙') const group = store.createGroup(owner, { name: '评审', memberIds: [a.id, b.id, c.id], collaborationMode: 'discussion' }) runtime.reply = agent => { if (agent.id === b.id) throw new Error('成员暂不可用'); return '核对后的结果' } const run = runtime.send(owner, group.id, { requestId: randomUUID(), content: '回答一轮', mentionIds: [b.id, c.id] }) await vi.waitFor(() => expect(store.require(owner, 'run', run.id).status).toBe('complete')) expect(runtime.calls.map(c => c.agentId)).toEqual([b.id, c.id]) expect(store.list(owner, 'goal')).toEqual([]) expect(store.list(owner, 'message').some(m => m.agentId === b.id && m.status === 'failed')).toBe(true) }) it('delivers A → B → A asynchronously and returns the reply to the originating group topic', async () => { const a = bot('甲'), b = bot('乙') const group = store.createGroup(owner, { name: '原始话题', memberIds: [a.id, b.id], administratorId: a.id, mode: 'host' }) const task = store.tasks(owner, group.id)[0]! runtime.reply = (agent, work) => { const run = store.require(owner, 'run', work.runId) if (!run.peerMessageId) runtime.collaboration.send(owner, work.id, { requestId: randomUUID(), agentId: b.id, content: '请核对接口事实', fileIds: [] }) else if (agent.id === b.id) runtime.collaboration.send(owner, work.id, { requestId: randomUUID(), agentId: a.id, content: '已查明接口版本为二', fileIds: [], replyTo: run.peerMessageId }) return agent.id === b.id ? '已发送核对结果' : run.peerMessageId ? '向用户交付最终结果' : '已请乙协助' } const deviceHost=randomUUID() runtime.send(owner, group.id, { requestId: randomUUID(), taskId: task.id, content: '请找乙核对接口后告诉我',deviceHost }) await vi.waitFor(() => expect(runtime.calls).toHaveLength(3)) await vi.waitFor(() => expect(store.list(owner, 'run').every(r => r.status === 'complete')).toBe(true)) const reply = store.list(owner, 'message').find(m => m.content === '向用户交付最终结果')! expect(reply.conversationId).toBe(group.id); expect(reply.conversationTaskId).toBe(task.id) expect(runtime.collaboration.list(owner)).toHaveLength(2) expect(store.list(owner,'run').every(run=>run.deviceHost===deviceHost)).toBe(true) expect(store.list(owner, 'message').filter(m => m.peerMessageId).every(m => m.role !== 'user')).toBe(true) }) it('deduplicates sends, suppresses acknowledgements and stops pending peers after the source turn ends', async () => { vi.spyOn(runtime, 'wake').mockImplementation(() => {}) const a = bot('甲'), b = bot('乙'), run = runtime.send(owner, direct(a.id).id, { requestId: randomUUID(), content: '联系同伴' }) const work = store.list(owner, 'turn')[0]!; work.status = 'running'; store.put(owner, 'turn', work.id, work) const input = { requestId: randomUUID(), agentId: b.id, content: '请核对接口', fileIds: [] } const first = runtime.collaboration.send(owner, work.id, input) expect(runtime.collaboration.send(owner, work.id, input).id).toBe(first.id) expect(() => runtime.collaboration.send(owner, work.id, { ...input, requestId: randomUUID() })).toThrowError(expect.objectContaining({ code: 'peer_repeated' })) expect(() => runtime.collaboration.send(owner, work.id, { ...input, requestId: randomUUID(), content: '收到,谢谢' })).toThrowError(expect.objectContaining({ code: 'peer_acknowledgement' })) run.status = 'complete'; store.saveRun(owner, run) await runtime.stop(owner, run.id) expect(store.require(owner, 'peer-message', first.id).status).toBe('stopped') expect(store.require(owner, 'run', first.runId!).stopRequested).toBe(true) }) it('bounds a persisted asynchronous chain and rejects disabled recipients and unrelated files', () => { vi.spyOn(runtime, 'wake').mockImplementation(() => {}) const a = bot('甲'), b = bot('乙') runtime.send(owner, direct(a.id).id, { requestId: randomUUID(), content: '开始协作' }) let work = store.list(owner, 'turn')[0]! work.status = 'running'; store.put(owner, 'turn', work.id, work) expect(() => runtime.collaboration.send(owner, work.id, { requestId: randomUUID(), agentId: b.id, content: '文件请求', fileIds: [randomUUID()] })).toThrowError(expect.objectContaining({ code: 'peer_file_forbidden' })) const hiddenFile = randomUUID() store.saveMessage(owner, { id: randomUUID(), conversationId: work.conversationId, seq: 0, role: 'assistant', agentId: a.id, visible: false, content: '内部执行资料', reasoning: '', status: 'complete', attachments: [{ id: hiddenFile, name: 'internal.txt', mimeType: 'text/plain', size: 1, sender: 'user', createdAt: Date.now() }], tools: [], createdAt: Date.now() }) expect(() => runtime.collaboration.send(owner, work.id, { requestId: randomUUID(), agentId: b.id, content: '不能分享内部附件', fileIds: [hiddenFile] })).toThrowError(expect.objectContaining({ code: 'peer_file_forbidden' })) for (let depth = 1; depth <= 8; depth++) { const peer = runtime.collaboration.send(owner, work.id, { requestId: randomUUID(), agentId: work.agentId === a.id ? b.id : a.id, content: `第 ${depth} 次新的事实和请求`, fileIds: [] }) expect(peer.depth).toBe(depth) work = store.list(owner, 'turn').find(w => w.runId === peer.runId)! work.status = 'running'; store.put(owner, 'turn', work.id, work) } expect(() => runtime.collaboration.send(owner, work.id, { requestId: randomUUID(), agentId: work.agentId === a.id ? b.id : a.id, content: '第九次请求', fileIds: [] })).toThrowError(expect.objectContaining({ code: 'peer_chain_limit' })) expect(runtime.collaboration.list(owner)).toHaveLength(8) }) it('restores the original project on reply without granting that project to an outside peer', async () => { const a = bot('项目成员'), b = bot('项目外顾问') const project = runtime.knowledge.saveProject(owner, { requestId: randomUUID(), name: '私有项目', description: '', memberIds: [a.id], groupIds: [] }) const seen: Array<{ agentId: string; projectId?: string }> = [] runtime.reply = (agent, work) => { const run = store.require(owner, 'run', work.runId) seen.push({ agentId: agent.id, projectId: run.projectId }) if (!run.peerMessageId) runtime.collaboration.send(owner, work.id, { requestId: randomUUID(), agentId: b.id, content: '请对公开接口提供建议', fileIds: [] }) else if (agent.id === b.id) runtime.collaboration.send(owner, work.id, { requestId: randomUUID(), agentId: a.id, content: '建议保留旧版本兼容', fileIds: [], replyTo: run.peerMessageId }) return '完成当前阶段' } runtime.send(owner, direct(a.id).id, { requestId: randomUUID(), content: '请参考顾问建议推进当前项目', projectId: project.id }) await vi.waitFor(() => expect(seen).toHaveLength(3)) expect(seen).toEqual([{ agentId: a.id, projectId: project.id }, { agentId: b.id, projectId: undefined }, { agentId: a.id, projectId: project.id }]) expect(runtime.knowledge.memberships(owner, b.id)).toEqual([]) })