agent-manager-session-scope.test.js
5.52 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
/**
* Unit Tests — `options.sessionScope` 真的从 `AgentManager.callAgent` 一路传到
* `SessionPool.getKey`（27-P4 §4.2 补的那一维）
*
* ── 这条测试还的是哪笔债 ──────────────────────────────────────────────
* [27-P4](../../../docs/todo/system-design/27-p4-parallel-and-artifacts.md) §4.2
* 设计了「图引擎并发调用同一 system agent 时按 `gr:{nodeRunId}` 传 scope，避免撞会话键」，
* `SessionPool.getKey()` 的 `scope` 参数也确实早就在（`session-pool-scope-key.test.js` 锁的
* 正是 `SessionPool` 自己的存取一致性），但**从 `AgentManager._runViaProtocol` 到
* `SessionPool.getOrCreate` 这一段接线此前从未写过**——`callAgent` 的 options 解构里
* `sessionScope` 只是摆设，`_runViaProtocol` 两处 `getOrCreate` 调用都固定只传 5 个参数。
*
* 后果：`executor-fanout.json` 这类骨架里 `{{orgNode}}-impl-a` / `{{orgNode}}-impl-b`
* 引用同一个 `systemAgentId`，若真的被 `advanceBatch` 并发拾取（两者同一个 run，
* `taskId` 都等于 `ctx.runId`），`sessionPool.getKey` 算出来的键完全相同——第二次调用会
* 复用第一次那个还在跑的 CLI 会话，两个"并行"分支实际上在同一个会话队列里排队执行，
* 且不报错、不留痕迹。
*
* 本测试只钉「接线通不通」，不追真实 CLI（用注册制协议工厂换一个内存桩，同
* `session-pool-scope-key.test.js` 的手法）：真实 agent 目录 + 真实 `AgentManager.callAgent`
* 全量跑一遍，`_runViaProtocol` 不做任何桩替换。
*/
import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import { EventEmitter } from 'events';
import fs from 'fs';
import path from 'path';
const AgentManager = require('../../src/core/agent-manager');
const { registerProtocol } = require('../../src/core/protocol-factory');
const locator = require('../../services/agent-locator');
/** 只满足 SessionPool.create() 用到的方法 + runPromptInSession 要的 sendPrompt。 */
let stubCounter = 0;
class StubProtocol extends EventEmitter {
constructor(agentInfo) {
super();
this.agentInfo = agentInfo;
this.isConnected = false;
this.sessionId = null;
}
async connect() { this.isConnected = true; }
async initialize() {}
async newSession() { this.sessionId = `stub-session-${++stubCounter}`; }
async sendPrompt() { return undefined; } // 立即"跑完"，不 emit 任何流式事件
disconnect() { this.isConnected = false; }
}
beforeAll(() => registerProtocol('claude-cli', () => StubProtocol));
afterAll(() => registerProtocol('claude-cli', () => require('@mas/acp-core').ClaudeCliProtocol));
describe('AgentManager.callAgent — sessionScope 接线（27-P4 §4.2）', () => {
const AGENT_ID = 'zz-scope-agent'; // zz- 前缀 = 测试建的注册表条目
const BODY_DIR = locator.bodyDir(AGENT_ID);
const CO = 'zz-scope-co';
const ADDR = `${CO}:${AGENT_ID}`;
const newManager = () => {
const m = new AgentManager({}, locator.REPO_ROOT, {});
m._resolveLogSession = async () => null;
m.hookRunner = { run: async () => {} };
return m;
};
beforeAll(() => {
fs.mkdirSync(BODY_DIR, { recursive: true });
fs.writeFileSync(path.join(BODY_DIR, 'agent.md'), '# 会话作用域测试本体\n', 'utf-8');
fs.writeFileSync(path.join(BODY_DIR, 'config.json'),
JSON.stringify({ id: AGENT_ID, name: AGENT_ID, type: 'skill', status: 'active' }, null, 2), 'utf-8');
});
afterAll(() => {
fs.rmSync(locator.companyDir(AGENT_ID, CO), { recursive: true, force: true });
});
it('不传 sessionScope：同一 run（同 taskId）两次调用撞上同一个会话（这是被 27-P4 §4.2 指出的问题本身）', async () => {
const manager = newManager();
await manager.callAgent(ADDR, 'T1', { metadata: { taskId: 'run-1' } });
await manager.callAgent(ADDR, 'T2', { metadata: { taskId: 'run-1' } });
// 两次都落在同一个 taskId 下、没有 scope 区分 —— 会话池里只会有 1 个会话
expect(manager.sessionPool.sessions.size).toBe(1);
});
it('传 sessionScope=gr:{nodeRunId}：同一 run 内两个并发节点各自拿到独立会话，互不覆盖', async () => {
const manager = newManager();
const [a, b] = await Promise.all([
manager.callAgent(ADDR, 'impl-a', {
metadata: { taskId: 'run-2' }, sessionScope: 'gr:node-a',
}),
manager.callAgent(ADDR, 'impl-b', {
metadata: { taskId: 'run-2' }, sessionScope: 'gr:node-b',
}),
]);
expect(a.status).toBe('success');
expect(b.status).toBe('success');
// 同一个 taskId、不同 scope —— 两个会话都要留在池子里，不能有一个把另一个顶掉
expect(manager.sessionPool.sessions.size).toBe(2);
const keys = [...manager.sessionPool.sessions.keys()];
expect(keys.some((k) => k.endsWith(':gr:node-a'))).toBe(true);
expect(keys.some((k) => k.endsWith(':gr:node-b'))).toBe(true);
});
it('断链复现：把 sessionScope 从 options 里去掉，上一条用例会退回到 1 个会话（反向验证这条测试真的在盯接线）', async () => {
const manager = newManager();
await Promise.all([
manager.callAgent(ADDR, 'impl-a', { metadata: { taskId: 'run-3' } }),
manager.callAgent(ADDR, 'impl-b', { metadata: { taskId: 'run-3' } }),
]);
// 没有 scope 区分：并发的两次调用会挤在同一个会话键上
expect(manager.sessionPool.sessions.size).toBe(1);
});
});