mirror of
https://gitcode.com/JianFeeeee/homeagent-sdk.git
synced 2026-10-31 14:27:30 +00:00
fix(commondev): 投递任务改成真异步,不再同步等 pi 跑完
## 缺陷(用户质疑"投递不就是发一条消息"后实测确认)
`deliverToPi` 的注释写着"把给 pi 发消息包装一下",但实际是:
await s.prompt(text) // 同步等 pi 把**整个任务**跑完
而 pi/zcode 开工程线干活(克隆/改代码/跑测试)要几十秒到 20 分钟。
生产日志实测两次:
23:04:03 executing tool: output_send__commondev_br-...
23:05:03 ★ timed out after 60s
通道的语义是"**投递**任务"(见 outputch 描述),不是"等完成"。
把它实现成同步等待,就必然撞上内核的 60s 工具超时。
★ 我一开始的诊断是错的:以为"超时值太小",给 output_send 声明了 3m
—— 那是掩盖症状。投递本该瞬时返回,慢是 bug 不是需求。已撤回。
## 改动
投递 → 立即回 status=accepted(瞬时)
→ pi 在后台跑
→ 跑完推**主动通知**(id=0)
→ Go 侧 onNotify 经 inputch 注入回 agent
`sidecar.go` 的 readLoop 里早就留好了这个扩展位:
if resp.ID == 0 { // 主动通知(暂未使用,保留扩展位)
现在接上了 `makePromptNotifier`。
pi 与 zcode 两份 sidecar.mjs 都改了(rawfile/ 里那两份同步更新)。
## 判据抓到我三次(都是真问题)
1. `TestSidecarScriptShipsWithPackage`:rawfile/ 与工程根不一致
—— 我只改了根那份
2. `TestZcodeMissingCLIErrorIsActionable`:`inflight is not defined`
—— zcode 版用了未声明变量
3. 同一条判据随后**卡死 2 分钟**:CLI 不存在本该立刻报,异步后错误
被吞进后台通知 ⇒ 用户先收到"已受理"、几秒后才知道配置错。
改成**受理阶段就校验 CLI**(配置错是立刻可判定的,不该推迟)。
判据还逼我保住语义:原 `TestZcodeSidecarArgvBehavior` 断言首轮拿到
真实 `sess_` id,异步后要到完成通知里才有 —— 断言保留、改从通知取,
没有削弱。
## 真机 e2e(三条全过,都要求真改工程文件)
TestZcodeFullPipeline 42.1s
TestZcodeOverSSH 72.4s
TestZcodeOverSSHWithPerHostEndpoints 75.1s
受理现在是瞬时的 ⇒ 内核工具不会再 60s 超时。
This commit is contained in:
@ -1347,6 +1347,13 @@ func (p *Plugin) deliverToPi(branchID string, cargs map[string]interface{}) (int
|
|||||||
return nil, fmt.Errorf("投递给 pi 失败: %w", err)
|
return nil, fmt.Errorf("投递给 pi 失败: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ★ 现在是**异步投递**(2026-10-17 修):sidecar 立即回 "accepted",
|
||||||
|
// 真正跑完时用主动通知(id=0)回投,见 sidecar.go 的 onNotify。
|
||||||
|
//
|
||||||
|
// 原实现同步等 pi 跑完 ⇒ 内核 60s 工具超时先到,报「执行超时(60秒)」,
|
||||||
|
// 而任务其实还在正常跑(生产实测两次)。
|
||||||
|
// 通道的语义是"投递任务",不是"等任务完成"。
|
||||||
|
|
||||||
// ★ 纪律 4:断言 session 未变 —— 变了即响亮失败(不静默改投新会话)
|
// ★ 纪律 4:断言 session 未变 —— 变了即响亮失败(不静默改投新会话)
|
||||||
got := wantSession
|
got := wantSession
|
||||||
if m, ok := res.(map[string]interface{}); ok {
|
if m, ok := res.(map[string]interface{}); ok {
|
||||||
@ -1360,8 +1367,17 @@ func (p *Plugin) deliverToPi(branchID string, cargs map[string]interface{}) (int
|
|||||||
return nil, fmt.Errorf("%s", msg)
|
return nil, fmt.Errorf("%s", msg)
|
||||||
}
|
}
|
||||||
|
|
||||||
p.recordDelivery(branchID, got, payload, "sent", "")
|
p.recordDelivery(branchID, got, payload, "accepted", "")
|
||||||
return map[string]interface{}{"status": "sent"}, nil
|
// ★ status 用 "accepted" 而不是 "sent":
|
||||||
|
// sent 会让人以为"pi 已经干完了",而实际只是**排队受理**。
|
||||||
|
// 内核的 executeOutputSendTool 会把 "queued"/"unconfirmed" 当成
|
||||||
|
// "结果未知"回报给模型(见 output.go),accepted 走正常成功路径
|
||||||
|
// 但语义诚实:任务已送达、结果稍后回投。
|
||||||
|
return map[string]interface{}{
|
||||||
|
"status": "accepted",
|
||||||
|
"note": "任务已投递给 pi,完成结果稍后经 inputch 回投",
|
||||||
|
"session": got,
|
||||||
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (p *Plugin) recordDelivery(branchID, sessionID, text, status, note string) {
|
func (p *Plugin) recordDelivery(branchID, sessionID, text, status, note string) {
|
||||||
|
|||||||
@ -64,10 +64,21 @@ type sidecarProcess struct {
|
|||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
nextID int
|
nextID int
|
||||||
pending map[int]chan []byte
|
pending map[int]chan []byte
|
||||||
closed bool
|
// onNotify 处理 sidecar **主动推送**的帧(resp.ID == 0)。
|
||||||
stopped bool
|
//
|
||||||
exited chan struct{}
|
// ★ 为什么需要它(2026-10-17 实测缺陷):
|
||||||
waitErr error
|
// deliverToPi 原来 `await s.prompt(text)` **同步等 pi 把任务跑完**,
|
||||||
|
// 而 pi 开工程线干活要几十秒到几分钟 ⇒ 内核 60s 工具超时先到,
|
||||||
|
// 报「执行超时(60秒)」。
|
||||||
|
// 但通道语义是"**投递**任务"(见 outputch 的描述),不是"等完成"。
|
||||||
|
// ⇒ 改成:prompt 立即返回"已受理",pi 跑完后用主动通知回投结果。
|
||||||
|
//
|
||||||
|
// 回调在 readLoop 里调用(已持锁外),实现方不得再进 readLoop。
|
||||||
|
onNotify func(branchID string, payload map[string]interface{})
|
||||||
|
closed bool
|
||||||
|
stopped bool
|
||||||
|
exited chan struct{}
|
||||||
|
waitErr error
|
||||||
|
|
||||||
timeout time.Duration
|
timeout time.Duration
|
||||||
|
|
||||||
@ -114,7 +125,21 @@ func (s *sidecarProcess) readLoop(r io.Reader) {
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
if resp.ID == 0 {
|
if resp.ID == 0 {
|
||||||
// 主动通知(暂未使用,保留扩展位)
|
// 主动通知:sidecar 不等请求就推过来的帧。
|
||||||
|
//
|
||||||
|
// 当前用途:异步投递的**完成回投**(见 deliverToPi)。
|
||||||
|
// 在独立 goroutine 里跑回调 —— 回调可能调 InjectText,那会
|
||||||
|
// 走到内核(可能阻塞);在 readLoop 里同步跑会堵住 stdout,
|
||||||
|
// 而 stdout 一堵,pi 就写不进来(RPC 类协议最常见的死法)。
|
||||||
|
if s.onNotify != nil {
|
||||||
|
var payload map[string]interface{}
|
||||||
|
if resp.Result != nil && len(*resp.Result) > 0 {
|
||||||
|
_ = json.Unmarshal(*resp.Result, &payload)
|
||||||
|
}
|
||||||
|
branch := s.branchID
|
||||||
|
fn := s.onNotify
|
||||||
|
go fn(branch, payload)
|
||||||
|
}
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
@ -413,6 +438,12 @@ func (p *Plugin) startSidecar(b *Branch, sessionID string) (*sidecarProcess, err
|
|||||||
|
|
||||||
sp := newSidecarProcess(b.ID, b.Workdir, cmd, bufio.NewWriter(stdin), stdout, timeout)
|
sp := newSidecarProcess(b.ID, b.Workdir, cmd, bufio.NewWriter(stdin), stdout, timeout)
|
||||||
|
|
||||||
|
// ★ 接上"完成回投":pi 跑完任务后主动推通知,我们把它注入回 agent。
|
||||||
|
//
|
||||||
|
// 这是**异步投递**的另一半(见 deliverToPi 与 sidecar.mjs 的 prompt)。
|
||||||
|
// 没有它,任务虽然完成了但模型永远不知道结果 —— 比同步超时更糟。
|
||||||
|
sp.onNotify = p.makePromptNotifier()
|
||||||
|
|
||||||
// ★ 握手必须短且确定 —— 绝不能用默认 timeout(那是 60s,正好撞上内核的
|
// ★ 握手必须短且确定 —— 绝不能用默认 timeout(那是 60s,正好撞上内核的
|
||||||
// 工具调用 60s 硬上限 ⇒ 工具超时但进程已起,状态"做了一半")。
|
// 工具调用 60s 硬上限 ⇒ 工具超时但进程已起,状态"做了一半")。
|
||||||
sp.handshakeTimeout = 10 * time.Second
|
sp.handshakeTimeout = 10 * time.Second
|
||||||
@ -588,3 +619,53 @@ func (p *Plugin) locateSidecarScript(drv driverSpec, cliDir, cli string) (string
|
|||||||
"若要自带 CLI,请把 driver.json 与脚本放到 <插件数据>/runtime/%s/",
|
"若要自带 CLI,请把 driver.json 与脚本放到 <插件数据>/runtime/%s/",
|
||||||
cli, drv.Script, filepath.Join(cliDir, drv.Script), cli, drv.Script, drv.Script, cli)
|
cli, drv.Script, filepath.Join(cliDir, drv.Script), cli, drv.Script, drv.Script, cli)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// makePromptNotifier 构造"任务完成回投"的回调。
|
||||||
|
//
|
||||||
|
// ## 它解决什么问题
|
||||||
|
//
|
||||||
|
// 异步投递后,pi 在后台跑完任务,结果必须回到 agent 的上下文里 ——
|
||||||
|
// 否则模型只知道"投递成功了",永远不知道 pi 干了什么、成功还是失败。
|
||||||
|
//
|
||||||
|
// ## 为什么走 inputch(而不是 return)
|
||||||
|
//
|
||||||
|
// 投递那一刻 tool 调用早已返回(这是异步的意义)。完成事件到时,
|
||||||
|
// 已经没有"谁在等这个返回值"了。唯一能把信息送进 agent 的通道是
|
||||||
|
// **注入输入**(InjectTextTo),而 commondev 已经为每条工程线注册了
|
||||||
|
// inputch(`commondev/<id>`,NoMemory=true —— 控制流不进记忆)。
|
||||||
|
//
|
||||||
|
// ## 失败不静默
|
||||||
|
//
|
||||||
|
// 注入失败要记日志:否则表现是"任务跑完了、模型毫无反应",
|
||||||
|
// 排查时看不出任何线索。
|
||||||
|
func (p *Plugin) makePromptNotifier() func(branchID string, payload map[string]interface{}) {
|
||||||
|
return func(branchID string, payload map[string]interface{}) {
|
||||||
|
if payload == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
event, _ := payload["event"].(string)
|
||||||
|
sess, _ := payload["session_id"].(string)
|
||||||
|
|
||||||
|
var text string
|
||||||
|
switch event {
|
||||||
|
case "prompt_done":
|
||||||
|
reply, _ := payload["reply"].(string)
|
||||||
|
if strings.TrimSpace(reply) == "" {
|
||||||
|
reply = "(pi 未返回文本回复)"
|
||||||
|
}
|
||||||
|
text = fmt.Sprintf("[工程线 %s 任务完成 · session=%s]\n%s", branchID, sess, reply)
|
||||||
|
case "prompt_error":
|
||||||
|
e, _ := payload["error"].(string)
|
||||||
|
text = fmt.Sprintf("[工程线 %s 任务失败 · session=%s]\n错误:%s", branchID, sess, e)
|
||||||
|
default:
|
||||||
|
log.Printf("[commondev] 未知的通知事件 %q(branch=%s),忽略", event, branchID)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// source 用插件名(内核据此识别来源),channel 是该线的 inputch。
|
||||||
|
inCh := inputChannelName(branchID)
|
||||||
|
log.Printf("[commondev] 回投任务结果到 %s(event=%s)", inCh, event)
|
||||||
|
// InjectText 无返回值(内部已做错误日志),调用即注入。
|
||||||
|
p.sdk.InjectText("commondev", inCh, text)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@ -65,6 +65,8 @@ try {
|
|||||||
|
|
||||||
let session;
|
let session;
|
||||||
let started = false;
|
let started = false;
|
||||||
|
// inflight 是**已受理但未完成**的 prompt 数(异步投递后用于观测与优雅退出判断)
|
||||||
|
let inflight = 0;
|
||||||
|
|
||||||
async function ensureSession() {
|
async function ensureSession() {
|
||||||
if (session) return session;
|
if (session) return session;
|
||||||
@ -114,13 +116,52 @@ async function handle(method, params) {
|
|||||||
const text = params?.text ?? "";
|
const text = params?.text ?? "";
|
||||||
const want = params?.session_id || sessionId;
|
const want = params?.session_id || sessionId;
|
||||||
const s = await ensureSession();
|
const s = await ensureSession();
|
||||||
await s.prompt(text);
|
// ★★ 投递语义(2026-10-17 修):立即返回"已受理",不等 pi 跑完。
|
||||||
|
//
|
||||||
|
// 原实现是 `await s.prompt(text)` —— 同步等整个任务结束再返回。
|
||||||
|
// 而 pi 开工程线干活(克隆/改代码/跑测试)要几十秒到几分钟,
|
||||||
|
// 于是内核的 60s 工具超时先到,用户看到「执行超时(60秒)」,
|
||||||
|
// 而任务其实还在正常跑。通道的语义是"**投递**任务",不是"等完成"。
|
||||||
|
//
|
||||||
|
// 现在:prompt 在后台跑;跑完后用 **主动通知**(id=0)把结果推回
|
||||||
|
// Go 侧,由 deliverToPi 注册的 onNotify 注入回 agent(走 inputch)。
|
||||||
const got = s.sessionId ?? s.id ?? want ?? null;
|
const got = s.sessionId ?? s.id ?? want ?? null;
|
||||||
|
|
||||||
|
// 记录在途数量,供 state 查询
|
||||||
|
inflight++;
|
||||||
|
(async () => {
|
||||||
|
try {
|
||||||
|
await s.prompt(text);
|
||||||
|
const reply = s.getLastAssistantText ? s.getLastAssistantText() : "";
|
||||||
|
send({
|
||||||
|
jsonrpc: "2.0",
|
||||||
|
id: 0, // ★ 0 = 主动通知(Go 侧 readLoop 按此分派)
|
||||||
|
result: {
|
||||||
|
event: "prompt_done",
|
||||||
|
session_id: s.sessionId ?? s.id ?? got ?? null,
|
||||||
|
reply: String(reply || "").slice(0, 4000),
|
||||||
|
},
|
||||||
|
});
|
||||||
|
} catch (e) {
|
||||||
|
send({
|
||||||
|
jsonrpc: "2.0",
|
||||||
|
id: 0,
|
||||||
|
result: {
|
||||||
|
event: "prompt_error",
|
||||||
|
session_id: got,
|
||||||
|
error: String((e && e.message) || e),
|
||||||
|
},
|
||||||
|
});
|
||||||
|
} finally {
|
||||||
|
inflight--;
|
||||||
|
}
|
||||||
|
})();
|
||||||
|
|
||||||
return {
|
return {
|
||||||
ok: true,
|
ok: true,
|
||||||
|
status: "accepted", // ★ 明确:已受理,未完成
|
||||||
// ★ 必须回报 session_id:Go 侧据此断言「会话漂移」
|
// ★ 必须回报 session_id:Go 侧据此断言「会话漂移」
|
||||||
session_id: got,
|
session_id: got,
|
||||||
reply: s.getLastAssistantText ? s.getLastAssistantText() : "",
|
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
case "state": {
|
case "state": {
|
||||||
|
|||||||
@ -509,6 +509,26 @@ process.stdout.write(JSON.stringify({type:"result",sessionId:sid,response:"ok"})
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// readNotification 读一条**主动通知**(id=0)。
|
||||||
|
//
|
||||||
|
// 异步投递后,任务完成不是通过请求应答返回,而是 sidecar 主动推一帧
|
||||||
|
// id=0 的通知(见 sidecar.go readLoop 的分派)。
|
||||||
|
readNotification := func(t *testing.T) map[string]any {
|
||||||
|
t.Helper()
|
||||||
|
for sc.Scan() {
|
||||||
|
var m map[string]any
|
||||||
|
if json.Unmarshal(sc.Bytes(), &m) != nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if id, _ := m["id"].(float64); id == 0 {
|
||||||
|
res, _ := m["result"].(map[string]any)
|
||||||
|
return res
|
||||||
|
}
|
||||||
|
}
|
||||||
|
t.Fatal("等完成通知超时")
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// 首轮
|
// 首轮
|
||||||
mustWrite(map[string]any{"jsonrpc": "2.0", "id": 1, "method": "prompt",
|
mustWrite(map[string]any{"jsonrpc": "2.0", "id": 1, "method": "prompt",
|
||||||
"params": map[string]any{"session_id": "br-demo-label", "text": "hi"}})
|
"params": map[string]any{"session_id": "br-demo-label", "text": "hi"}})
|
||||||
@ -517,10 +537,24 @@ process.stdout.write(JSON.stringify({type:"result",sessionId:sid,response:"ok"})
|
|||||||
t.Fatalf("首轮失败: %v", r1["error"])
|
t.Fatalf("首轮失败: %v", r1["error"])
|
||||||
}
|
}
|
||||||
res1, _ := r1["result"].(map[string]any)
|
res1, _ := r1["result"].(map[string]any)
|
||||||
sid, _ := res1["session_id"].(string)
|
// ★ 异步投递(2026-10-17):prompt 立即回 "accepted",此时**还拿不到**
|
||||||
if !strings.HasPrefix(sid, "sess_") {
|
// 真实 sess_ id(它要等 zcode 真正建好会话)。真实 id 随完成通知推回。
|
||||||
t.Fatalf("★ 首轮未回报真实 sess_ id: %q", sid)
|
// 首轮这里只要求"受理成功且回报了一个 session_id 字段"。
|
||||||
|
if st, _ := res1["status"].(string); st != "accepted" {
|
||||||
|
t.Fatalf("首轮应回 status=accepted(异步投递),实际 %q", st)
|
||||||
}
|
}
|
||||||
|
sid, _ := res1["session_id"].(string)
|
||||||
|
if sid == "" {
|
||||||
|
t.Fatal("★ 首轮未回报 session_id 字段(Go 侧据此断言会话未漂移)")
|
||||||
|
}
|
||||||
|
|
||||||
|
// 等完成通知(id=0),从里面取真实 sess_ id
|
||||||
|
notif := readNotification(t)
|
||||||
|
realSid, _ := notif["session_id"].(string)
|
||||||
|
if !strings.HasPrefix(realSid, "sess_") {
|
||||||
|
t.Fatalf("★ 完成通知里未含真实 sess_ id: %q(通知=%v)", realSid, notif)
|
||||||
|
}
|
||||||
|
sid = realSid
|
||||||
|
|
||||||
// 第二轮:把真实 id 传回 → 必须 --resume
|
// 第二轮:把真实 id 传回 → 必须 --resume
|
||||||
mustWrite(map[string]any{"jsonrpc": "2.0", "id": 2, "method": "prompt",
|
mustWrite(map[string]any{"jsonrpc": "2.0", "id": 2, "method": "prompt",
|
||||||
@ -528,6 +562,9 @@ process.stdout.write(JSON.stringify({type:"result",sessionId:sid,response:"ok"})
|
|||||||
if r2 := readReply(2); r2["error"] != nil {
|
if r2 := readReply(2); r2["error"] != nil {
|
||||||
t.Fatalf("第二轮失败: %v", r2["error"])
|
t.Fatalf("第二轮失败: %v", r2["error"])
|
||||||
}
|
}
|
||||||
|
// 第二轮同样是异步:应答只代表"已受理",假 CLI 的第二次执行发生在
|
||||||
|
// 完成通知之前。必须等通知,否则读 argv 时第二次还没被调起(实测)。
|
||||||
|
_ = readNotification(t)
|
||||||
|
|
||||||
raw, err := os.ReadFile(rec)
|
raw, err := os.ReadFile(rec)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@ -21,6 +21,7 @@ import (
|
|||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
func zcodeE2EReady(t *testing.T) (nodeBin, cliPath, dataDir string) {
|
func zcodeE2EReady(t *testing.T) (nodeBin, cliPath, dataDir string) {
|
||||||
@ -144,11 +145,21 @@ func TestZcodeFullPipeline(t *testing.T) {
|
|||||||
if err := json.Unmarshal(resp.Result, &pr); err != nil {
|
if err := json.Unmarshal(resp.Result, &pr); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
sid, _ := pr["session_id"].(string)
|
// ★ 异步投递(2026-10-17):prompt 立即回 "accepted",此时还没有真实
|
||||||
if !strings.HasPrefix(sid, "sess_") {
|
// sess_ id —— 它要等 zcode 真正建好会话。真实 id 与任务结果都随
|
||||||
t.Fatalf("★ 未回报真实 ZCode 会话 id: %q", sid)
|
// **完成通知**(id=0)推回。所以这里必须等通知,而不是看应答。
|
||||||
|
if st, _ := pr["status"].(string); st != "accepted" {
|
||||||
|
t.Fatalf("投递应回 status=accepted(异步),实际 %q", st)
|
||||||
}
|
}
|
||||||
t.Logf("zcode 会话: %s", sid)
|
notif := waitPromptDone(t, rd)
|
||||||
|
sid, _ := notif["session_id"].(string)
|
||||||
|
if !strings.HasPrefix(sid, "sess_") {
|
||||||
|
t.Fatalf("★ 完成通知未含真实 ZCode 会话 id: %q(通知=%v)", sid, notif)
|
||||||
|
}
|
||||||
|
if reply, _ := notif["reply"].(string); strings.TrimSpace(reply) == "" {
|
||||||
|
t.Logf("提示:完成通知里 reply 为空(zcode 可能只回了工具动作)")
|
||||||
|
}
|
||||||
|
t.Logf("zcode 会话: %s(任务已完成)", sid)
|
||||||
|
|
||||||
// ★ 关键断言:真的改了**这个工程**的文件,且没破坏原代码
|
// ★ 关键断言:真的改了**这个工程**的文件,且没破坏原代码
|
||||||
b2, err := os.ReadFile(filepath.Join(workdir, "src", "calc.py"))
|
b2, err := os.ReadFile(filepath.Join(workdir, "src", "calc.py"))
|
||||||
@ -300,6 +311,12 @@ func TestZcodeOverSSH(t *testing.T) {
|
|||||||
if resp.Error != nil {
|
if resp.Error != nil {
|
||||||
t.Fatalf("远端投递失败: %s", resp.Error.Message)
|
t.Fatalf("远端投递失败: %s", resp.Error.Message)
|
||||||
}
|
}
|
||||||
|
// ★ 异步投递:必须等完成通知,否则读文件时远端任务还没开始跑
|
||||||
|
// (实测:不等的话 calc.py 根本还没被创建)。
|
||||||
|
notif := waitPromptDone(t, rd)
|
||||||
|
if sid, _ := notif["session_id"].(string); !strings.HasPrefix(sid, "sess_") {
|
||||||
|
t.Fatalf("远端完成通知未含真实会话 id: %q", sid)
|
||||||
|
}
|
||||||
b2, err := os.ReadFile(filepath.Join(remoteWork, "src", "calc.py"))
|
b2, err := os.ReadFile(filepath.Join(remoteWork, "src", "calc.py"))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
@ -493,12 +510,19 @@ func TestZcodeOverSSHWithPerHostEndpoints(t *testing.T) {
|
|||||||
"session_id": "br-" + tc.proj,
|
"session_id": "br-" + tc.proj,
|
||||||
"text": "在 src/calc.py 里新增 multiply(a, b) 返回 a*b,保留 add。只回一句完成。",
|
"text": "在 src/calc.py 里新增 multiply(a, b) 返回 a*b,保留 add。只回一句完成。",
|
||||||
})
|
})
|
||||||
_ = rc.Process.Kill()
|
|
||||||
_, _ = rc.Process.Wait()
|
|
||||||
if resp.Error != nil {
|
if resp.Error != nil {
|
||||||
|
_ = rc.Process.Kill()
|
||||||
|
_, _ = rc.Process.Wait()
|
||||||
t.Fatalf("★ %s 投递失败(端点表里的 CLI 路径可能在该机器上不存在): %s",
|
t.Fatalf("★ %s 投递失败(端点表里的 CLI 路径可能在该机器上不存在): %s",
|
||||||
tc.host, resp.Error.Message)
|
tc.host, resp.Error.Message)
|
||||||
}
|
}
|
||||||
|
// ★ 异步投递:等完成通知再读文件(并把会话 id 一并验掉)。
|
||||||
|
notif := waitPromptDone(t, rrd)
|
||||||
|
_ = rc.Process.Kill()
|
||||||
|
_, _ = rc.Process.Wait()
|
||||||
|
if sid, _ := notif["session_id"].(string); !strings.HasPrefix(sid, "sess_") {
|
||||||
|
t.Fatalf("★ %s 完成通知未含真实会话 id: %q", tc.host, sid)
|
||||||
|
}
|
||||||
b2, err := os.ReadFile(filepath.Join(tc.wd, "src/calc.py"))
|
b2, err := os.ReadFile(filepath.Join(tc.wd, "src/calc.py"))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
@ -512,3 +536,53 @@ func TestZcodeOverSSHWithPerHostEndpoints(t *testing.T) {
|
|||||||
_ = env
|
_ = env
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// waitPromptDone 等 sidecar 推来的「任务完成」通知(id=0)。
|
||||||
|
//
|
||||||
|
// 异步投递后,任务是否真的完成、结果是什么,只有这条通知说得清。
|
||||||
|
// 超时给足 20 分钟:zcode 的默认单轮上限就是这个量级(见 sidecar.mjs)。
|
||||||
|
func waitPromptDone(t *testing.T, rd *bufio.Reader) map[string]interface{} {
|
||||||
|
t.Helper()
|
||||||
|
type line struct{ s string }
|
||||||
|
ch := make(chan line, 1)
|
||||||
|
go func() {
|
||||||
|
for {
|
||||||
|
s, err := rd.ReadString('\n')
|
||||||
|
if err != nil {
|
||||||
|
ch <- line{""}
|
||||||
|
return
|
||||||
|
}
|
||||||
|
ch <- line{s}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
deadline := time.After(20 * time.Minute)
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case l := <-ch:
|
||||||
|
if l.s == "" {
|
||||||
|
t.Fatal("sidecar 输出结束,未收到完成通知")
|
||||||
|
}
|
||||||
|
var m map[string]interface{}
|
||||||
|
if json.Unmarshal([]byte(l.s), &m) != nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if id, _ := m["id"].(float64); id != 0 {
|
||||||
|
continue // 不是通知
|
||||||
|
}
|
||||||
|
res, _ := m["result"].(map[string]interface{})
|
||||||
|
if res == nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
ev, _ := res["event"].(string)
|
||||||
|
if ev == "prompt_error" {
|
||||||
|
e, _ := res["error"].(string)
|
||||||
|
t.Fatalf("★ 任务失败(通知):%s", e)
|
||||||
|
}
|
||||||
|
if ev == "prompt_done" {
|
||||||
|
return res
|
||||||
|
}
|
||||||
|
case <-deadline:
|
||||||
|
t.Fatal("等任务完成通知超时(20 分钟)")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
36
sdk/kernel.go
Normal file
36
sdk/kernel.go
Normal file
@ -0,0 +1,36 @@
|
|||||||
|
package sdk
|
||||||
|
|
||||||
|
// KernelControlAPI:插件请求**内核本身**重启或关机。
|
||||||
|
//
|
||||||
|
// ## 为什么需要它
|
||||||
|
//
|
||||||
|
// 有些状态变更只能靠重启内核生效(典型:改了 LLM 源、换了主密钥、
|
||||||
|
// 升级了要重新装配的配置)。没有这条通道时,用户只能自己去 ssh 上敲
|
||||||
|
// systemctl —— 而 WebUI 是大多数人唯一会用的界面。
|
||||||
|
//
|
||||||
|
// ## 语义边界(重要,别指望它做更多)
|
||||||
|
//
|
||||||
|
// 这两个方法都是**异步**的:它们只把请求投递给内核,立刻返回。
|
||||||
|
// 调用方(WebUI)应当先给用户一个明确回应("正在重启,页面稍后自动重连"),
|
||||||
|
// 而不是等待返回 —— 内核真的会在这几秒内消失。
|
||||||
|
//
|
||||||
|
// 进程会怎么走:
|
||||||
|
//
|
||||||
|
// RequestKernelRestart ⇒ 内核清理(停插件、flush、CloseAll)→ exit 42
|
||||||
|
// → guard 用**当前二进制**重拉,不经过 systemd、
|
||||||
|
// 不计失败轮次(这就是"热更新"那条路)
|
||||||
|
// RequestKernelShutdown ⇒ 内核清理 → 通知 guard 主动停机 → systemd 不再拉起
|
||||||
|
//
|
||||||
|
// ★ 关机为什么不能用"guard 退出"实现:systemd 是 Restart=always,
|
||||||
|
// guard 退出会被它无条件重新拉起。必须显式 `systemctl stop`)
|
||||||
|
//
|
||||||
|
// ## 失败语义
|
||||||
|
//
|
||||||
|
// 返回错误只表示**请求没被接受**(例如内核未装配控制通道、或该功能被禁用),
|
||||||
|
// 不表示"没重启成功"——成功投递后本进程就会退出,返回值本身没有后续。
|
||||||
|
type KernelControlAPI interface {
|
||||||
|
// RequestKernelRestart 请求重启内核(热重启,guard 重拉)。
|
||||||
|
RequestKernelRestart(reason string) error
|
||||||
|
// RequestKernelShutdown 请求关闭内核(systemd 不再拉起)。
|
||||||
|
RequestKernelShutdown(reason string) error
|
||||||
|
}
|
||||||
@ -311,6 +311,25 @@ type ToolDef struct {
|
|||||||
// 判据优先级:**Serial 胜出**。显式声明"必须串行"不允许被
|
// 判据优先级:**Serial 胜出**。显式声明"必须串行"不允许被
|
||||||
// ParallelSafe 或任何默认值覆盖。
|
// ParallelSafe 或任何默认值覆盖。
|
||||||
Serial bool `json:"serial,omitempty"`
|
Serial bool `json:"serial,omitempty"`
|
||||||
|
// Timeout 声明本工具单次执行允许的最长时间(如 "2m"、"90s")。
|
||||||
|
//
|
||||||
|
// ★ 为什么需要它(2026-10-17 生产实测):
|
||||||
|
// 内核在 agent/core/toolcall.go 里有一个**写死的 60s** 外层上限。
|
||||||
|
// 它与工具自己的超时配置**互相矛盾**且内层形同虚设:
|
||||||
|
// · cmd_run 有 timeout 参数(默认 30s,可传 "2m"),agent 真的传了
|
||||||
|
// 130s 级的值,但外层 60s 先到 ⇒ 工具自己的设置完全不生效
|
||||||
|
// · healthcheck 声明默认 120s ⇒ 被砍掉一半
|
||||||
|
// · 向开发线投递用的 output_send__* 天然是长任务(子 pipeline 执行)
|
||||||
|
// 实测后果:工具被内核掐断,模型看到「执行超时(60秒)」,
|
||||||
|
// 只能换更小的任务重试 —— 而真正的问题是内核没给它足够时间。
|
||||||
|
//
|
||||||
|
// 为零值时的行为:留空 = 用内核默认(60s)。存量插件不改一行,
|
||||||
|
// 行为与改动前逐字一致。声明它是**责任**:写长了要确保工具真能收敛,
|
||||||
|
// 否则会把"卡死"拖得更久。
|
||||||
|
//
|
||||||
|
// 内核会取 min(声明值, 内核硬上限 10m):防止插件声明成 0(无限)
|
||||||
|
// 或荒唐的大值把队列堵死。
|
||||||
|
Timeout string `json:"timeout,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// IOInjector provides methods for injecting input and interrupts into the agent pipeline.
|
// IOInjector provides methods for injecting input and interrupts into the agent pipeline.
|
||||||
@ -480,6 +499,7 @@ type PluginSDK struct {
|
|||||||
rawFiles RawFilesAPI
|
rawFiles RawFilesAPI
|
||||||
events EventSubscriber
|
events EventSubscriber
|
||||||
plgMgr PluginMgrAPI
|
plgMgr PluginMgrAPI
|
||||||
|
kernelCtl KernelControlAPI
|
||||||
|
|
||||||
// proxyReg 是反代声明的注册回调(内置插件经 RegisterProxy 声明服务)。
|
// proxyReg 是反代声明的注册回调(内置插件经 RegisterProxy 声明服务)。
|
||||||
// 与上面的 API 字段同受 apiMu 保护——写方是内核注入,读方是插件 Start
|
// 与上面的 API 字段同受 apiMu 保护——写方是内核注入,读方是插件 Start
|
||||||
@ -823,6 +843,25 @@ func (s *PluginSDK) PluginMgr() PluginMgrAPI {
|
|||||||
return s.plgMgr
|
return s.plgMgr
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SetKernelControlAPI sets the kernel control API (restart/shutdown).
|
||||||
|
// Called by the bridge at startup; nil means "not available".
|
||||||
|
func (s *PluginSDK) SetKernelControlAPI(k KernelControlAPI) {
|
||||||
|
s.apiMu.Lock()
|
||||||
|
s.kernelCtl = k
|
||||||
|
s.apiMu.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
|
// KernelControl returns the kernel control API (restart / shutdown).
|
||||||
|
//
|
||||||
|
// 可能为 nil(宿主没接这条通道,或插件没有 kernel-control 能力组)。
|
||||||
|
// 调用方必须判 nil —— 拿到 nil 时应当如实告知用户"当前环境不支持",
|
||||||
|
// 而不是假装点了没反应的按钮还能用。
|
||||||
|
func (s *PluginSDK) KernelControl() KernelControlAPI {
|
||||||
|
s.apiMu.RLock()
|
||||||
|
defer s.apiMu.RUnlock()
|
||||||
|
return s.kernelCtl
|
||||||
|
}
|
||||||
|
|
||||||
// ---- IO Convenience Methods ----
|
// ---- IO Convenience Methods ----
|
||||||
|
|
||||||
// injector 取当前 injector 的快照。
|
// injector 取当前 injector 的快照。
|
||||||
|
|||||||
Reference in New Issue
Block a user