stream-executor.js
3.99 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
/**
* Stream Executor — 在复用会话中执行 prompt
*
* 职责：
* - 绑定流式事件监听器（acp_event / session_update / error）
* - IdleTimer 空闲超时保护
* - 串行化队列（同一 session 内按序执行）
* - 收集 stdout 文本
*
* 不负责：
* - 会话创建/销毁（由 SessionPool 决定）
* - 重试策略（由 AgentManager._runViaProtocol 决定）
*
* 设计原则：
* 事件处理完全协议无关。所有 Protocol 统一 emit:
* - session_update { update: { sessionUpdate, content: { text } } }
* - acp_event { type, method, params, timestamp }
* 前端根据 method 前缀（session/、codex/）自行解析渲染。
*/
const { IdleTimer } = require('../engine');
/**
* 在复用会话中执行 prompt
*
* @param {object} session - SessionPool 返回的 session 对象
* @param {string} prompt - 完整 prompt 文本
* @param {object} [options]
* @param {function} [options.onData] - 流式文本回调
* @param {function} [options.onAcpEvent] - 原始协议事件回调（透传）
* @param {object} [options.logSession] - 日志会话
* @param {number} [options.idleTimeoutMs=120000] - 空闲超时
* @returns {Promise<string>} stdout 文本
*/
async function runPromptInSession(session, prompt, options = {}) {
const run = async () => {
const { onData, onAcpEvent, logSession, idleTimeoutMs = 120_000 } = options;
const protocol = session.protocol;
let stdout = '';
// ── IdleTimer 空闲超时保护 ──
let idleReject = null;
const idleTimer = new IdleTimer(idleTimeoutMs, () => {
const err = new Error(
`Agent ${session.agentName} idle timeout: no stream data for ${idleTimeoutMs}ms`
);
err.code = 'IDLE_TIMEOUT';
if (idleReject) idleReject(err);
}, { label: `acp-idle:${session.agentName}` });
// ── 事件监听器（协议无关） ──
const onAcpEventWrapped = async (event) => {
if (logSession) {
const direction = event.type === 'request' ? 'out' : 'in';
await logSession.logProtocolEvent(session.agentName, direction, event);
}
if (onAcpEvent) onAcpEvent(event);
};
const onSessionUpdateWrapped = async (params) => {
idleTimer.touch();
const update = params.update;
const text = update?.content?.text || '';
// 协议无关：所有 Protocol 统一 emit sessionUpdate 格式
if (update.sessionUpdate === 'agent_thought_chunk') {
// thought chunks（目前仅 ACP 会 emit），同样透传
if (logSession) await logSession.logStreamChunk(session.agentName, text, 'thought');
if (onData) onData(text);
} else if (update.sessionUpdate === 'agent_message_chunk') {
stdout += text;
if (logSession) await logSession.logStreamChunk(session.agentName, text, 'message');
if (onData) onData(text);
}
};
const onErrorWrapped = async (err) => {
if (logSession) {
await logSession.logAgentError(session.agentName, err, {
type: `${session.protocolName}_error`,
});
}
};
// ── 绑定 → 执行 → 解绑 ──
protocol.on('acp_event', onAcpEventWrapped);
protocol.on('session_update', onSessionUpdateWrapped);
protocol.on('error', onErrorWrapped);
try {
const result = await new Promise((resolve, reject) => {
idleReject = reject;
idleTimer.start();
protocol.sendPrompt(prompt).then(
() => resolve(stdout),
(err) => reject(err)
);
});
return result;
} finally {
idleTimer.cancel();
protocol.off('acp_event', onAcpEventWrapped);
protocol.off('session_update', onSessionUpdateWrapped);
protocol.off('error', onErrorWrapped);
}
};
// 串行化：同一 session 内的调用按队列顺序执行
const chained = session.callQueue.then(run, run);
session.callQueue = chained.then(() => undefined, () => undefined);
return chained;
}
module.exports = { runPromptInSession };