ProcessTransport.js
3.12 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
/**
* ProcessTransport - 子进程传输层实现
*
* 通过 child_process.spawn 创建子进程，使用 stdio 管道进行通信。
* 支持 NDJSON (Newline Delimited JSON) 格式的消息。
*/
const { spawn } = require('child_process');
const { Transport } = require('./Transport');
/**
* 子进程传输层
*
* 负责：
* - 启动子进程
* - 管理 stdio 管道通信
* - 处理数据缓冲和分包
* - 解析 NDJSON 消息
*/
class ProcessTransport extends Transport {
constructor() {
super();
this.child = null;
}
/**
* 检查是否已连接
*/
get isConnected() {
return this.child !== null && !this.child.killed;
}
/**
* 启动子进程并建立连接
* @param {string} command - 可执行命令
* @param {string[]} args - 参数列表
* @param {object} options - spawn 选项
*/
async connect(command, args, options = {}) {
return new Promise((resolve, reject) => {
try {
console.log(`[ProcessTransport] 启动进程: ${command} ${args.join(' ')}`);
// 打印关键环境变量
if (options.env) {
const criticalVars = ['ANTHROPIC_AUTH_TOKEN', 'ANTHROPIC_BASE_URL'];
console.log('[ProcessTransport] 子进程环境变量:');
criticalVars.forEach(key => {
if (options.env[key]) {
const masked = key.includes('TOKEN')
? options.env[key].substring(0, 12) + '...'
: options.env[key];
console.log(` ✅ ${key}: ${masked}`);
} else {
console.log(` ❌ ${key}: 未设置`);
}
});
} else {
console.log('[ProcessTransport] ⚠️ 未传递 env 参数，子进程将继承父进程环境');
}
this.child = spawn(command, args, {
stdio: ['pipe', 'pipe', 'pipe'],
...options,
});
// 监听进程错误
this.child.on('error', (err) => {
this.emit('error', err);
reject(err);
});
// 监听进程退出
this.child.on('exit', (code, signal) => {
this.emit('close', code, signal);
});
// 监听标准错误
this.child.stderr?.on('data', (data) => {
this.emit('stderr', data.toString());
});
// 监听标准输出 — 使用基类的 _handleData
this.child.stdout?.on('data', (data) => {
this._handleData(data);
});
// 等待进程启动
setTimeout(() => {
if (this.isConnected) {
resolve();
} else {
resolve();
}
}, 500);
} catch (err) {
reject(err);
}
});
}
/**
* 发送消息
* @param {object} msg - 消息对象
*/
send(msg) {
if (!this.isConnected || this.child.stdin.destroyed) {
throw new Error('Transport is closed');
}
const str = JSON.stringify(msg);
this.child.stdin.write(str + '\n');
}
/**
* 关闭连接
*/
close() {
if (this.child) {
this.child.kill();
this.child = null;
}
this.buffer = '';
}
}
module.exports = { ProcessTransport };