stream-executor.js
5.51 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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
/**
* 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=1800000] - 空闲超时 (默认30分钟，由上层心跳监控负责真正的超时管理)
* @returns {Promise<string>} stdout 文本
*/
async function runPromptInSession(session, prompt, options = {}) {
const run = async () => {
const { onData, onAcpEvent, logSession, idleTimeoutMs = 1_800_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}` });
// ── 事件监听器（协议无关） ──
let currentToolInput = '';
const onAcpEventWrapped = async (event) => {
if (logSession) {
const direction = event.type === 'request' ? 'out' : 'in';
await logSession.logProtocolEvent(session.agentName, direction, event);
}
if (event.method === 'claude/tool_use_start') {
currentToolInput = '';
const toolName = event.params.name;
const toolMsg = `\n🔧 [TOOL_CALL] ${toolName} (ID: ${event.params.tool_use_id || 'unknown'})\n`;
stdout += toolMsg;
if (onData) onData(toolMsg);
} else if (event.method === 'claude/stream/content_block_stop') {
if (currentToolInput) {
const inputLog = ` 输入: ${currentToolInput.slice(0, 500)}${currentToolInput.length > 500 ? '...' : ''}\n`;
stdout += inputLog;
if (onData) onData(inputLog);
currentToolInput = '';
}
} else if (event.method?.startsWith('claude/')) {
const innerEvent = event.params;
if (innerEvent?.delta?.type === 'input_json_delta' && innerEvent.delta.partial_json) {
currentToolInput += innerEvent.delta.partial_json;
}
if (innerEvent?.type === 'tool_result' || innerEvent?.content_block?.type === 'tool_result') {
const result = innerEvent.content || innerEvent.content_block?.content || '';
const resultStr = typeof result === 'string' ? result : JSON.stringify(result);
const truncated = resultStr.slice(0, 1000) + (resultStr.length > 1000 ? `...(${resultStr.length} chars)` : '');
const resultLog = `\n 结果: ${truncated}\n`;
stdout += resultLog;
if (onData) onData(resultLog);
}
}
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 };