From 4df81ef1da30547745fcbdb4c04235098d2d9f7f Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 10 Oct 2026 09:02:10 +0000 Subject: [PATCH] Enable parallel tool execution, fix delegate truncation, sandbox deps layer Why: Real runs were fully serial: the direct path never set the kernel's parallel tool mode, delegate_task handed the parent only the child's 160-char summary (the full text was dropped), MiniMax streamed its reasoning inside content blocks, and the EXT-16 sandbox hid ~/.local so python-docx/openpyxl/pptx could not even import. All four were measured defects (F1/F2/F3/F5) in the delegation/routing proposal. Why: Serial-only mutual exclusion is not enough for order-sensitive tools - the card-stream test failed with out-of-order appends - so the ordering moved into the kernel (SerialTools partition, Agent-Native #44); process events pair by CallID (#42) and tool results write back through the injected call context (#43). Evidence: go test ./... green, including new regression anchors: TestDelegateToolReturnsFullText, TestToolEventsPairByCallID, TestToolExecModeAndSerialSet, TestOpenLLMClientReasoningSplitAuto, TestSandboxArgvDepsLayer, TestExecEnvJoinsPythonPath. Impact: Direct runs now execute independent same-turn calls (searches, reads, delegate_task, review_findings) concurrently by default (runtime.parallelTools=false restores sequential); card flow, plans, workflows, exec primitives and file edits keep model-declared order. delegate_task returns the capped full child text. MiniMax-* models get reasoning_split=true so thinking streams via reasoning_content. --exec-deps-dir/AGENT_SYSTEM_EXEC_DEPS_DIR mounts a read-only dependency layer into the sandbox and merges PYTHONPATH. Bumps deps/agent-native to c348441. Refs: proposal 2026-10-10-agent-delegation-routing-sandbox.md (E1, F1-F5); Agent-Native #42/#43/#44; Dev task feature/agent-e1-parallel-fixes. --- cmd/agent-system/main.go | 8 +- deps/agent-native | 2 +- internal/app/app.go | 7 ++ internal/config/config.go | 9 ++ internal/runtime/llm.go | 13 +- internal/runtime/llm_direct.go | 158 ++++++++++++++++++++----- internal/runtime/llm_direct_e1_test.go | 103 ++++++++++++++++ internal/runtime/llm_test.go | 10 ++ internal/runtime/squad_runner.go | 2 +- internal/runtime/sub_loop.go | 2 +- internal/runtime/verify.go | 2 +- internal/runtime/workflow_compiled.go | 2 +- internal/settings/settings.go | 1 + internal/tools/exec_backend.go | 8 ++ internal/tools/exec_backend_test.go | 42 +++++++ internal/tools/exec_interpreter.go | 21 +++- 16 files changed, 347 insertions(+), 43 deletions(-) create mode 100644 internal/runtime/llm_direct_e1_test.go diff --git a/cmd/agent-system/main.go b/cmd/agent-system/main.go index 84899fe..6e9cf18 100644 --- a/cmd/agent-system/main.go +++ b/cmd/agent-system/main.go @@ -46,6 +46,7 @@ func main() { allowedHosts := fs.String("allowed-hosts", "", "逗号分隔的额外主机名，供反向代理的 Host 与 Origin 校验") noExec := fs.Bool("disable-exec-tools", false, "硬关闭执行原语 run_shell/run_python/register_artifact（用户设置无法覆盖；多用户部署在执行隔离就位前建议开启）") execBackend := fs.String("exec-backend", "", "执行后端：direct（默认，现状直启）| sandbox（bwrap 隔离档：断网、只读系统根、仅会话目录可写、进程数限额）") + execDepsDir := fs.String("exec-deps-dir", "", "sandbox 档只读依赖层目录（site-packages 等；空=不挂载）") fs.Parse(os.Args[1:]) cliSet := visited(fs) @@ -53,7 +54,7 @@ func main() { Addr: *addr, DataDir: *data, PluginsDir: *pluginsDir, Mock: *mock, StorageDSN: *dsn, ArtifactsDir: *artifacts, AuthMode: *auth, Instance: *instance, InsecureListen: *insecure, AllowedHostsRaw: *allowedHosts, DisableExecTools: *noExec, - ExecBackend: *execBackend, + ExecBackend: *execBackend, ExecDepsDir: *execDepsDir, }, cliSet) if err != nil { fmt.Fprintln(os.Stderr, "agent-system:", err) @@ -63,7 +64,7 @@ func main() { Addr: resolved.Addr, DataDir: resolved.DataDir, PluginsDir: resolved.PluginsDir, ForceMock: resolved.Mock, StorageDSN: resolved.StorageDSN, ArtifactsDir: resolved.ArtifactsDir, AuthMode: resolved.AuthMode, InstanceID: resolved.Instance, InsecureListen: resolved.InsecureListen, AllowedHosts: resolved.AllowedHosts(), - DisableExecTools: resolved.DisableExecTools, ExecBackend: resolved.ExecBackend, + DisableExecTools: resolved.DisableExecTools, ExecBackend: resolved.ExecBackend, ExecDepsDir: resolved.ExecDepsDir, }); err != nil { fmt.Fprintln(os.Stderr, "agent-system:", err) os.Exit(1) @@ -108,12 +109,13 @@ func runConfig(args []string) int { allowedHosts := fs.String("allowed-hosts", "", "") noExec := fs.Bool("disable-exec-tools", false, "") execBackend := fs.String("exec-backend", "", "") + execDepsDir := fs.String("exec-deps-dir", "", "") fs.Parse(args[1:]) resolved, err := config.Load(*cfgFile, os.LookupEnv, config.Config{ Addr: *addr, DataDir: *data, PluginsDir: *pluginsDir, Mock: *mock, StorageDSN: *dsn, ArtifactsDir: *artifacts, AuthMode: *auth, Instance: *instance, InsecureListen: *insecure, AllowedHostsRaw: *allowedHosts, DisableExecTools: *noExec, - ExecBackend: *execBackend, + ExecBackend: *execBackend, ExecDepsDir: *execDepsDir, }, visited(fs)) if err != nil { fmt.Fprintln(os.Stderr, "agent-system:", err) diff --git a/deps/agent-native b/deps/agent-native index 3b92cb6..c348441 160000 --- a/deps/agent-native +++ b/deps/agent-native @@ -1 +1 @@ -Subproject commit 3b92cb63b41b5dd035371dada9be50ad66e39941 +Subproject commit c348441691e8bdab784a03aa9be64d53dacff341 diff --git a/internal/app/app.go b/internal/app/app.go index ca2bf58..9579cde 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -79,6 +79,9 @@ type Options struct { // ExecBackend 执行后端（EXT-16）：空/direct=现状直启；sandbox=bwrap 隔离档。 // 硬关闭 DisableExecTools 优先级高于后端选择。 ExecBackend string + // ExecDepsDir sandbox 档只读依赖层（site-packages 等，空=不挂）：修复 + // --tmpfs /home 遮蔽 ~/.local 导致交付物生成链路不可用（F3）。 + ExecDepsDir string } // Run 装配并阻塞服务直至 SIGINT/SIGTERM（优雅停机：中断在途 run、冲净事件缓冲）。 @@ -138,6 +141,10 @@ func Run(opt Options) error { if tools.ExecBackendName() == "sandbox" { log.Printf("exec tools: backend=sandbox (bwrap: no network, ro root, session-only writes, proc limit)") } + if opt.ExecDepsDir != "" { + tools.SetExecDepsDir(opt.ExecDepsDir) + log.Printf("exec tools: sandbox deps layer=%s", opt.ExecDepsDir) + } if err := tools.RegisterExecTools(reg); err != nil { return fmt.Errorf("exec tools: %w", err) } diff --git a/internal/config/config.go b/internal/config/config.go index b87afc6..4901cfb 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -31,6 +31,8 @@ type Config struct { // ExecBackend 执行后端（EXT-16）：direct（空 = 现状直启）| sandbox（bwrap // 隔离档）。DisableExecTools 优先级高于后端选择。 ExecBackend string `json:"execBackend"` + // ExecDepsDir sandbox 档只读依赖层目录（空 = 不挂载）。 + ExecDepsDir string `json:"execDepsDir"` } var defaults = Config{ @@ -51,6 +53,7 @@ var envKey = map[string]string{ "mock": "AGENT_SYSTEM_MOCK", "instance": "AGENT_SYSTEM_INSTANCE", "execBackend": "AGENT_SYSTEM_EXEC_BACKEND", + "execDepsDir": "AGENT_SYSTEM_EXEC_DEPS_DIR", "insecureListen": "AGENT_SYSTEM_INSECURE_LISTEN", "allowedHosts": "AGENT_SYSTEM_ALLOWED_HOSTS", "disableExecTools": "AGENT_SYSTEM_DISABLE_EXEC_TOOLS", @@ -176,6 +179,8 @@ func loadFile(path string) (Config, map[string]bool, error) { dst = &c.DisableExecTools case "execBackend": dst = &c.ExecBackend + case "execDepsDir": + dst = &c.ExecDepsDir default: return Config{}, nil, fmt.Errorf("config file %s: unknown key %q", path, k) } @@ -207,6 +212,8 @@ func stringField(c Config, f string) string { return c.AllowedHostsRaw case "execBackend": return c.ExecBackend + case "execDepsDir": + return c.ExecDepsDir } return "" } @@ -257,6 +264,8 @@ func setString(c *Config, f, v string) { c.AllowedHostsRaw = v case "execBackend": c.ExecBackend = v + case "execDepsDir": + c.ExecDepsDir = v } } diff --git a/internal/runtime/llm.go b/internal/runtime/llm.go index a3291f3..1d0deb6 100644 --- a/internal/runtime/llm.go +++ b/internal/runtime/llm.go @@ -242,7 +242,7 @@ func (l *LLMRunner) Execute(rc RunCtx) error { if stream == nil { // 重试经库默认（WP6：runtime.llmRetries → Config.RetryAttempts 映射）， // 此处不再显式包装（防双重）。 - client := openLLMClient(route.Base, route.Key) + client := openLLMClient(route.Base, route.Key, route.Model) stream = client.Stream } @@ -488,8 +488,15 @@ func agentNameOf(prof core.AgentProfile) string { } // openLLMClient 不设 HTTP 客户端超时。单次生成可能很长，停止按钮通过 run 的 context 取消。 -func openLLMClient(base, key string) *agentllm.Client { - return &agentllm.Client{BaseURL: base, APIKey: key, HTTPClient: &http.Client{}} +func openLLMClient(base, key, model string) *agentllm.Client { + c := &agentllm.Client{BaseURL: base, APIKey: key, HTTPClient: &http.Client{}} + // MiniMax 系（interleaved thinking）的 OpenAI 兼容接口默认把思考混在 + // content 的 块里；reasoning_split=true 让思考改走独立 + // reasoning_content 字段（SSE 捕获路径不变，正文流不再夹带思考）。 + if strings.HasPrefix(model, "MiniMax-M") { + c.ReasoningSplit = true + } + return c } // SubLLMRunner 子代理执行器：按内置 profile 跑一轮带工具的内核循环。 diff --git a/internal/runtime/llm_direct.go b/internal/runtime/llm_direct.go index 5dabe90..a560c3e 100644 --- a/internal/runtime/llm_direct.go +++ b/internal/runtime/llm_direct.go @@ -49,7 +49,8 @@ type directRun struct { responding bool acting bool queued []agentllm.ToolCall - inflight *toolSlot + inflight map[string]*toolSlot // 并行执行下按内核事件 CallID 配对 + toolMu sync.Mutex // 并行 onEvent 下 inflight/queued/finalParts 互斥 turnBody string evidence []toolEvidence @@ -97,7 +98,7 @@ func (t directTool) Execute(ctx context.Context, args json.RawMessage) (agenttoo } func (l *LLMRunner) runDirect(rc RunCtx, stream agentllm.StreamFunc, model string, msgs []agentllm.Message, decls []agentllm.ToolDecl, streamOpts agentllm.StreamOptions, cardflow bool, eff core.EffectiveSettings, maxTurns, maxTools int, refresh turnRefresh) error { - st := &directRun{l: l, rc: rc, eff: eff, cardflow: cardflow} + st := &directRun{l: l, rc: rc, eff: eff, cardflow: cardflow, inflight: map[string]*toolSlot{}} // 记忆端口（N3 WP5）：runCtx 暴露 store 时接线（注入由循环按请求完成）。 if ms, ok := rc.(interface{ StoreMemory() memoryStore }); ok { st.memoryStore = ms.StoreMemory() @@ -127,7 +128,7 @@ func (l *LLMRunner) runDirect(rc RunCtx, stream agentllm.StreamFunc, model strin for _, d := range decls { name := d.Name reg.Register(directTool{name: name, run: func(ctx context.Context, args json.RawMessage) agenttools.Result { - return st.runTool(name, string(args)) + return st.runToolCtx(ctx, name, string(args)) }}) } // 内核工作区工具接入直连路径（裁定 1：尽可能开放工具）：行号分页 read、 @@ -153,7 +154,7 @@ func (l *LLMRunner) runDirect(rc RunCtx, stream agentllm.StreamFunc, model strin wsReg := ws reg.Register(directTool{name: name, run: func(ctx context.Context, args json.RawMessage) agenttools.Result { res := wsReg.Execute(ctx, name, args) - st.noteKernelResult(res) + st.noteKernelResult(ctx, res) return res }}) } @@ -267,6 +268,8 @@ func (l *LLMRunner) runDirect(rc RunCtx, stream agentllm.StreamFunc, model strin }) res, err := nruntime.Run(rc.Context(), nruntime.Config{ Model: model, + ToolExecution: toolExecMode(eff), + SerialTools: serialTools, History: history, Session: sessStore, JournalOnly: sessStore == nil, // 有转录即为权威；无转录时不涉及 @@ -367,20 +370,41 @@ func (l *LLMRunner) runDirect(rc RunCtx, stream agentllm.StreamFunc, model strin // noteKernelResult 镜像 runTool 的结果登记：内核工作区工具不经过产品执行路径， // 摘要/状态/验证文本在这里补记，保证过程面板与终轮验证器看到同样的证据。 -func (st *directRun) noteKernelResult(res agenttools.Result) { - if st.inflight == nil { +func (st *directRun) noteKernelResult(ctx context.Context, res agenttools.Result) { + call, ok := nruntime.ToolCallFromContext(ctx) + slot := st.slotFor("") + if ok { + slot = st.slotFor(call.ID) + } + if slot == nil { return } summary := summarize(res.Content, nil) if res.IsError { summary = "error: " + summary } - st.inflight.summary = summary - st.inflight.status = statusOf(nil) + slot.summary = summary + slot.status = statusOf(nil) if res.IsError { - st.inflight.status = "error" + slot.status = "error" } - st.inflight.verifyText = tailRunes(res.Content, 600) + slot.verifyText = tailRunes(res.Content, 600) +} + +// slotFor 按调用 ID 取在途槽位；ID 为空且恰有一个槽位时取之（兼容无注入 +// 的调用路径）。返回的指针仅供本次工具执行内写回，不跨事件持有。 +func (st *directRun) slotFor(callID string) *toolSlot { + st.toolMu.Lock() + defer st.toolMu.Unlock() + if callID != "" { + return st.inflight[callID] + } + if len(st.inflight) == 1 { + for _, slot := range st.inflight { + return slot + } + } + return nil } // noteToolDelta 把模型正在编写的工具参数立刻转成会话事件。 @@ -451,7 +475,9 @@ func (st *directRun) onEvent(ev nruntime.Event) { st.usageTotal.LastCompletionTokens = ev.CompletionTokens st.usageTotal.CacheReadTokens += ev.CacheReadTokens st.usageTotal.CacheWriteTokens += ev.CacheWriteTokens + st.toolMu.Lock() st.queued = append([]agentllm.ToolCall(nil), ev.Message.ToolCalls...) + st.toolMu.Unlock() hasCalls := len(ev.Message.ToolCalls) > 0 if st.turnThink.Len() > 0 { st.finalParts = append(st.finalParts, core.Part{Type: core.PartThinking, Text: st.turnThink.String(), Process: true, DurationMs: durationMs(st.thinkStart, st.thinkEnd)}) @@ -484,13 +510,19 @@ func (st *directRun) onEvent(ev nruntime.Event) { if len(st.queued) == 0 { return } - call := st.queued[0] - st.queued = st.queued[1:] - st.inflight = &toolSlot{call: call, start: st.l.now()} + call, ok := st.takeQueued(ev.CallID) + if !ok { + return + } + st.toolMu.Lock() + st.inflight[call.ID] = &toolSlot{call: call, start: st.l.now()} + st.toolMu.Unlock() _ = st.rc.Emit(core.EvToolCall, core.ToolCallPayload{CallID: call.ID, Name: call.Name, Arguments: call.Arguments}) case "tool_execution_end": - slot := st.inflight + st.toolMu.Lock() + slot := st.takeInflight(ev.CallID) if slot == nil { + st.toolMu.Unlock() return } end := st.l.now() @@ -502,8 +534,8 @@ func (st *directRun) onEvent(ev nruntime.Event) { st.deliverables = append(st.deliverables, slot.artifact) } st.evidence = append(st.evidence, toolEvidence{Name: slot.call.Name, Status: slot.status, Summary: slot.summary, VerifyText: slot.verifyText}) + st.toolMu.Unlock() _ = st.rc.Emit(core.EvToolResult, core.ToolResultPayload{CallID: slot.call.ID, Status: slot.status, Summary: slot.summary, Truncated: slot.truncated, DurationMs: durationMs(slot.start, end)}) - st.inflight = nil } } @@ -658,10 +690,17 @@ func (st *directRun) continueOrNotice(info nruntime.StopInfo, reason string) nru } func (st *directRun) runTool(name, args string) agenttools.Result { - call := agentllm.ToolCall{Name: name, Arguments: args} - if st.inflight != nil { - call = st.inflight.call + return st.runToolCtx(context.Background(), name, args) +} + +// runToolCtx 执行一个工具调用；ctx 携带内核注入的当前调用（并行执行下按 +// CallID 定位在途槽位），无注入时回落唯一在途槽位（旧内核/测试直调）。 +func (st *directRun) runToolCtx(ctx context.Context, name, args string) agenttools.Result { + call, _ := nruntime.ToolCallFromContext(ctx) + if call.ID == "" { + call = agentllm.ToolCall{Name: name, Arguments: args} } + slot := st.slotFor(call.ID) var out string var err error switch call.Name { @@ -710,7 +749,10 @@ func (st *directRun) runTool(name, args string) agenttools.Result { if serr != nil { out = "子代理失败：" + serr.Error() } else { - out = orDefault(res.Summary, "子代理完成") + // F1 修复：回传子代理结论全文（capSubagentResult 已做 32k 截断与 + // 超长落盘指针），Summary 只是 160 字摘要——只回摘要会让父代理在 + // 摘要的摘要上凭空补细节。 + out = orDefault(res.Text, orDefault(res.Summary, "子代理完成")) st.pinSubagentArtifacts(res.Summary + "\n" + res.Text) } case toolReview: @@ -752,8 +794,8 @@ func (st *directRun) runTool(name, args string) agenttools.Result { } else { out = res.Output } - if res.Artifact != nil && st.inflight != nil { - st.inflight.artifact = &core.ArtifactRef{ArtifactID: res.Artifact.ID, Name: res.Artifact.Name, Kind: res.Artifact.Kind} + if res.Artifact != nil && slot != nil { + slot.artifact = &core.ArtifactRef{ArtifactID: res.Artifact.ID, Name: res.Artifact.Name, Kind: res.Artifact.Kind} if st.cardflow && res.Truncated && !st.explicitCard { fc, ferr := st.rc.SubmitCard(artifactFacadeCard(call.Name, *res.Artifact, summarize(out, nil))) if ferr == nil { @@ -761,8 +803,8 @@ func (st *directRun) runTool(name, args string) agenttools.Result { } } } - if st.inflight != nil { - st.inflight.truncated = res.Truncated + if slot != nil { + slot.truncated = res.Truncated } } @@ -772,10 +814,10 @@ func (st *directRun) runTool(name, args string) agenttools.Result { summary = "error: " + err.Error() out = summary } - if st.inflight != nil { - st.inflight.summary = summary - st.inflight.status = status - st.inflight.verifyText = tailRunes(out, 600) + if slot != nil { + slot.summary = summary + slot.status = status + slot.verifyText = tailRunes(out, 600) } if r := []rune(out); len(r) > 16000 { out = string(r[:16000]) + "…（超长截断）" @@ -966,3 +1008,65 @@ func errString(err error) string { } return err.Error() } + +// ---- 并行工具执行的辅助（E1/F5）---- + +// serialTools 顺序敏感或带会话级副作用的工具：作为内核 SerialTools 传入—— +// 并行执行下这些调用按模型声明顺序逐个执行（互斥且保序；只互斥不保序会让 +// 流式卡 card_begin→append→commit 乱序，实测踩过）。卡片流四步有先后依赖， +// 计划/待办是整表更新，工作流与小队编排派生 run，执行原语与文件编辑共享会话 +// 工作区。读检索类与 delegate_task/review_findings 并行。 +var serialTools = map[string]bool{ + toolPlan: true, toolTodos: true, + toolCard: true, toolCardBegin: true, toolCardAppend: true, toolCardCommit: true, + toolRunWorkflow: true, toolSquad: true, + "run_shell": true, "run_python": true, "register_artifact": true, + "edit": true, "multi_edit": true, +} + +// toolExecMode 同回合多工具调用的执行模式：runtime.parallelTools（缺省开） +// 为 parallel（顺序敏感工具经 Config.SerialTools 由内核保序），关闭时空串 +// 回落内核缺省 sequential。 +func toolExecMode(eff core.EffectiveSettings) string { + if eff.GetBool("runtime.parallelTools", true) { + return nruntime.ExecParallel + } + return "" +} + +// takeQueued 按内核事件 CallID 取出队列中对应调用（并行下 start 事件乱序到达）； +// CallID 为空（旧内核/未知来源）时按原先进先出。调用方须持有 toolMu。 +func (st *directRun) takeQueued(callID string) (agentllm.ToolCall, bool) { + if callID != "" { + for i, c := range st.queued { + if c.ID == callID { + st.queued = append(st.queued[:i], st.queued[i+1:]...) + return c, true + } + } + return agentllm.ToolCall{}, false + } + if len(st.queued) == 0 { + return agentllm.ToolCall{}, false + } + c := st.queued[0] + st.queued = st.queued[1:] + return c, true +} + +// takeInflight 按事件 CallID 取出并删除在途槽位；CallID 为空且恰有一个在途 +// 槽位时取之（兼容旧内核）。调用方须持有 toolMu。 +func (st *directRun) takeInflight(callID string) *toolSlot { + if callID != "" { + slot := st.inflight[callID] + delete(st.inflight, callID) + return slot + } + if len(st.inflight) == 1 { + for _, slot := range st.inflight { + clear(st.inflight) + return slot + } + } + return nil +} diff --git a/internal/runtime/llm_direct_e1_test.go b/internal/runtime/llm_direct_e1_test.go new file mode 100644 index 0000000..5d0d341 --- /dev/null +++ b/internal/runtime/llm_direct_e1_test.go @@ -0,0 +1,103 @@ +package runtime + +import ( + "strings" + "testing" + + nruntime "agent-native/runtime" + agentllm "agent-native/runtime/llm" + core "agent-system-core" +) + +// ---- F1：delegate_task 回传子代理全文，不再只有 160 字摘要 ---- + +func TestDelegateToolReturnsFullText(t *testing.T) { + full := strings.Repeat("子代理结论细节。", 60) // 480 字，远超 160 + f := &fakeRunCtx{spawnAgentFn: func(agentID, name, task, hint string) (core.AgentResultPayload, error) { + return core.AgentResultPayload{Status: core.RunCompleted, Text: full, Summary: "完成（160字内摘要）"}, nil + }} + st := &directRun{l: &LLMRunner{}, rc: f, eff: f.Settings(), inflight: map[string]*toolSlot{}} + res := st.runTool("delegate_task", `{"name":"调研","task":"查证X"}`) + if res.IsError { + t.Fatalf("delegate failed: %s", res.Content) + } + if !strings.Contains(res.Content, "子代理结论细节。子代理结论细节。子代理结论细节。") { + t.Fatalf("delegate result must carry the full child text, got: %.120s…", res.Content) + } + if strings.Contains(res.Content, "160字内摘要") { + t.Fatalf("summary must be fallback only, got: %.120s…", res.Content) + } +} + +// ---- F5：并行执行下工具事件按 CallID 配对 ---- + +func TestToolEventsPairByCallID(t *testing.T) { + f := &fakeRunCtx{} + st := &directRun{l: &LLMRunner{}, rc: f, eff: f.Settings(), inflight: map[string]*toolSlot{}} + st.queued = []agentllm.ToolCall{ + {ID: "c1", Name: "knowledge_search", Arguments: `{}`}, + {ID: "c2", Name: "web_search", Arguments: `{}`}, + } + // 乱序进出：c1 先开始，c2 先结束（并行下的真实时序）。 + st.onEvent(nruntime.Event{Type: "tool_execution_start", Tool: "knowledge_search", CallID: "c1"}) + st.onEvent(nruntime.Event{Type: "tool_execution_start", Tool: "web_search", CallID: "c2"}) + st.onEvent(nruntime.Event{Type: "tool_execution_end", Tool: "web_search", CallID: "c2"}) + st.onEvent(nruntime.Event{Type: "tool_execution_end", Tool: "knowledge_search", CallID: "c1"}) + + if len(st.inflight) != 0 { + t.Fatalf("inflight must be empty after both ends: %v", st.inflight) + } + if len(st.finalParts) != 4 { + t.Fatalf("finalParts = %d (want 4: call+result ×2)", len(st.finalParts)) + } + // 结束顺序决定部件顺序：先 c2 后 c1。 + if got := st.finalParts[0].ToolCall.ID; got != "c2" { + t.Fatalf("first tool part = %s (want c2)", got) + } + if got := st.finalParts[1].ToolResult.CallID; got != "c2" { + t.Fatalf("first result part = %s (want c2)", got) + } + if got := st.finalParts[2].ToolCall.ID; got != "c1" { + t.Fatalf("second tool part = %s (want c1)", got) + } + if got := st.finalParts[3].ToolResult.CallID; got != "c1" { + t.Fatalf("second result part = %s (want c1)", got) + } +} + +// ---- F5：执行模式与串行集合 ---- + +func TestToolExecModeAndSerialSet(t *testing.T) { + if got := toolExecMode(core.EffectiveSettings{}); got != nruntime.ExecParallel { + t.Fatalf("default must be parallel, got %q", got) + } + off := core.EffectiveSettings{"runtime": map[string]any{"parallelTools": false}} + if got := toolExecMode(off); got != "" { + t.Fatalf("parallelTools=false must fall back to sequential, got %q", got) + } + // 顺序敏感工具必须串行；可并行的读检索与委派不得误入。 + for _, name := range []string{"update_plan", "update_todos", "submit_card", "card_begin", "card_append", "card_commit", "run_workflow", "squad_orchestrate", "run_shell", "run_python", "register_artifact"} { + if !serialTools[name] { + t.Fatalf("serialTools missing %s", name) + } + } + for _, name := range []string{"delegate_task", "review_findings", "knowledge_search", "web_search", "fs_read_file", "write_artifact"} { + if serialTools[name] { + t.Fatalf("serialTools must not contain %s", name) + } + } +} + +// ---- F2：MiniMax 前缀自动置位 reasoning_split ---- + +func TestOpenLLMClientReasoningSplitAuto(t *testing.T) { + if c := openLLMClient("http://x", "k", "MiniMax-M2.7-highspeed"); !c.ReasoningSplit { + t.Fatal("MiniMax model must set ReasoningSplit") + } + if c := openLLMClient("http://x", "k", "MiniMax-M3"); !c.ReasoningSplit { + t.Fatal("MiniMax M3 must set ReasoningSplit") + } + if c := openLLMClient("http://x", "k", "claude-sonnet-5"); c.ReasoningSplit { + t.Fatal("non-MiniMax model must not set ReasoningSplit") + } +} diff --git a/internal/runtime/llm_test.go b/internal/runtime/llm_test.go index 13870d0..e946605 100644 --- a/internal/runtime/llm_test.go +++ b/internal/runtime/llm_test.go @@ -45,6 +45,7 @@ type fakeRunCtx struct { memEntries []memEntry // 非空时 StoreMemory 返回之（N3 WP5 记忆注入测试） usageTotal core.Usage // AddUsage 累计（N3 WP6 记账断言用） bgCalls int + spawnAgentFn func(agentID, name, task, hint string) (core.AgentResultPayload, error) } func (f *fakeRunCtx) Context() context.Context { return f.ctx } @@ -78,6 +79,15 @@ func (f *fakeRunCtx) AskHITL(card core.HITLCard) (core.HITLResponse, error) { func (f *fakeRunCtx) SpawnSubagent(name, task, hint string) (core.AgentResultPayload, error) { return core.AgentResultPayload{}, fmt.Errorf("not implemented") } + +// SpawnAgent 供 delegate_task 直连分支（E1/F1 测试注入）；未设置时与 +// SpawnSubagent 一致报错。 +func (f *fakeRunCtx) SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) { + if f.spawnAgentFn != nil { + return f.spawnAgentFn(agentID, name, task, hint) + } + return core.AgentResultPayload{}, fmt.Errorf("not implemented") +} func (f *fakeRunCtx) UpdatePlan(p core.PlanSnapshot) error { return nil } func (f *fakeRunCtx) UpdateTodos(l core.TodoList) error { return nil } func (f *fakeRunCtx) WriteArtifact(name, kind, mime string, content []byte) (core.Artifact, error) { diff --git a/internal/runtime/squad_runner.go b/internal/runtime/squad_runner.go index 5abb248..22775cb 100644 --- a/internal/runtime/squad_runner.go +++ b/internal/runtime/squad_runner.go @@ -139,7 +139,7 @@ func newModelRouter(rc RunCtx) modelRouter { if route.Model == "" || route.Base == "" { return modelRouter{} } - client := openLLMClient(route.Base, route.Key) + client := openLLMClient(route.Base, route.Key, route.Model) return modelRouter{stream: client.Stream, model: route.Model} } diff --git a/internal/runtime/sub_loop.go b/internal/runtime/sub_loop.go index 1d34ef8..41f1a2d 100644 --- a/internal/runtime/sub_loop.go +++ b/internal/runtime/sub_loop.go @@ -54,7 +54,7 @@ func (l *SubLLMRunner) Execute(rc RunCtx) error { if route.Model == "" || route.Base == "" { return fmt.Errorf("llm-subagent: 模型未配置（当前解析：model=%q base=%q）", route.Model, route.Base) } - client := openLLMClient(route.Base, route.Key) + client := openLLMClient(route.Base, route.Key, route.Model) stream = client.Stream } if spec.Model == "chat" { diff --git a/internal/runtime/verify.go b/internal/runtime/verify.go index 703acf2..3a9d1f2 100644 --- a/internal/runtime/verify.go +++ b/internal/runtime/verify.go @@ -41,7 +41,7 @@ func (l *LLMRunner) runVerifier(rc RunCtx, body string, evidence []toolEvidence, if route.Model == "" || route.Base == "" { return gates.Verdict{}, fmt.Errorf("验证器调用失败") } - client := openLLMClient(route.Base, route.Key) + client := openLLMClient(route.Base, route.Key, route.Model) // 验证器不经内核循环：显式接重试（WP6 同映射）与 5xx 回退链。 stream = agentllm.WithFallback( agentllm.WithRetry(client.Stream, retryAttemptsOf(rc.Settings()), nil), diff --git a/internal/runtime/workflow_compiled.go b/internal/runtime/workflow_compiled.go index b805674..ab7a3d3 100644 --- a/internal/runtime/workflow_compiled.go +++ b/internal/runtime/workflow_compiled.go @@ -79,7 +79,7 @@ func judgeStreamFor(rc RunCtx) agentllm.StreamFunc { if route.Model == "" || route.Base == "" { return nil } - client := openLLMClient(route.Base, route.Key) + client := openLLMClient(route.Base, route.Key, route.Model) model := route.Model return func(ctx context.Context, _ string, msgs []agentllm.Message, tools []agentllm.ToolDecl, opts agentllm.StreamOptions) <-chan agentllm.Event { return client.Stream(ctx, model, msgs, tools, opts) diff --git a/internal/settings/settings.go b/internal/settings/settings.go index 050a083..ed65426 100644 --- a/internal/settings/settings.go +++ b/internal/settings/settings.go @@ -91,6 +91,7 @@ func Schema() []core.SettingDesc { {Key: "runtime.mock", Group: "runtime", Label: "演示模式（mock runner）", Type: core.SettingBool, Default: false, Effect: core.EffectNewRun, Description: "不调用真实模型，按脚本回放事件流"}, {Key: "runtime.squad", Group: "runtime", Label: "小队编排（Agent-Squad）", Type: core.SettingBool, Default: true, Effect: core.EffectNewRun, Description: "启用 Agent-Squad 引擎编排（#squad 指令与 squad_orchestrate 工具）"}, {Key: "runtime.entry", Group: "runtime", Label: "对话入口", Type: core.SettingEnum, Default: "direct", Options: []string{"direct", "squad"}, Effect: core.EffectNewRun, Description: "direct 走现有对话循环。squad 由小队先分诊：简单任务单步完成，复杂任务由 leader 编排。灰度期默认 direct"}, + {Key: "runtime.parallelTools", Group: "runtime", Label: "并行工具执行", Type: core.SettingBool, Default: true, Effect: core.EffectNewRun, Description: "同一回合并行执行互不依赖的工具调用（检索、读文件、delegate_task 等）；卡片流、计划、执行原语等顺序敏感工具仍互斥串行"}, {Key: "runtime.rateLimitPerMin", Group: "runtime", Label: "租户限流（次/分）", Type: core.SettingNumber, Default: 0, Effect: core.EffectHot, Description: "0 = 不限；多用户建议 300–1200"}, {Key: "runtime.redisAddr", Group: "runtime", Label: "Redis 地址（限流共享）", Type: core.SettingString, Default: "", Effect: core.EffectRestart, Description: "原生 Redis（如 127.0.0.1:6379）；空 = 进程内限流（单实例）"}, {Key: "runtime.verify", Group: "runtime", Label: "终轮结果检查", Type: core.SettingBool, Default: true, Effect: core.EffectNewRun, Description: "终轮确定性检查与独立验证"}, diff --git a/internal/tools/exec_backend.go b/internal/tools/exec_backend.go index 70faf3d..3b6de8e 100644 --- a/internal/tools/exec_backend.go +++ b/internal/tools/exec_backend.go @@ -30,6 +30,10 @@ type ExecSpec struct { Dir string // 会话沙箱工作目录（sandbox 档唯一可写挂载） Env []string // 最小环境（execEnv 口径） SkillsDir string // 技能根（sandbox 档只读挂载；空 = 不挂） + // DepsDir 依赖层（sandbox 档只读挂载；空 = 不挂）：python-docx 等装在 + // ~/.local 下会被 --tmpfs /home 遮蔽，只读挂回其原路径并并入 PYTHONPATH， + // 交付物生成链路（F3）在隔离档下才可用。 + DepsDir string Timeout time.Duration // 硬超时（调用方 ctx.WithTimeout 之前传入，仅记录用） } @@ -168,6 +172,10 @@ func sandboxArgv(spec ExecSpec) []string { if spec.SkillsDir != "" { argv = append(argv, "--ro-bind", spec.SkillsDir, spec.SkillsDir) } + if spec.DepsDir != "" { + // 必须后于 --tmpfs /home：先遮蔽再只读挂回依赖层（挂载顺序敏感）。 + argv = append(argv, "--ro-bind", spec.DepsDir, spec.DepsDir) + } argv = append(argv, "--chdir", spec.Dir, "--clearenv") for _, kv := range spec.Env { if i := strings.IndexByte(kv, '='); i > 0 { diff --git a/internal/tools/exec_backend_test.go b/internal/tools/exec_backend_test.go index 03d0c85..41006f3 100644 --- a/internal/tools/exec_backend_test.go +++ b/internal/tools/exec_backend_test.go @@ -283,3 +283,45 @@ func TestDirectBackendRunInterpreterUnchanged(t *testing.T) { t.Fatalf("direct timeout regression: err=%v got=%q", err, got) } } + +// TestSandboxArgvDepsLayer F3：依赖层在 --tmpfs /home 之后只读挂回原路径 +// （顺序敏感——先遮蔽再挂回），且挂载点存在性由 bwrap 创建。 +func TestSandboxArgvDepsLayer(t *testing.T) { + argv := sandboxArgv(ExecSpec{ + Bin: "/usr/bin/python3", Args: []string{"/data/s/x.py"}, + Dir: "/data/s", DepsDir: "/home/u/.local/lib/python3.14/site-packages", + }) + joined := strings.Join(argv, "\x00") + deps := "--ro-bind\x00/home/u/.local/lib/python3.14/site-packages\x00/home/u/.local/lib/python3.14/site-packages" + if !strings.Contains(joined, deps) { + t.Fatalf("sandbox argv missing deps ro-bind: %v", argv) + } + tmpHome := strings.Index(joined, "--tmpfs\x00/home") + depsAt := strings.Index(joined, deps) + if tmpHome < 0 || depsAt < tmpHome { + t.Fatalf("deps bind must come after --tmpfs /home: %v", argv) + } +} + +// TestExecEnvJoinsPythonPath F3：依赖层与技能根合成一条 PYTHONPATH， +// 不允许重复键互相覆盖（否则只挂载了依赖层却 import 不到）。 +func TestExecEnvJoinsPythonPath(t *testing.T) { + oldSkills, oldDeps := skillsRoot, execDepsDir + defer func() { skillsRoot, execDepsDir = oldSkills, oldDeps }() + skillsRoot = "/data/skills" + execDepsDir = "/opt/pydeps" + env := execEnv() + count, value := 0, "" + for _, kv := range env { + if strings.HasPrefix(kv, "PYTHONPATH=") { + count++ + value = strings.TrimPrefix(kv, "PYTHONPATH=") + } + } + if count != 1 { + t.Fatalf("PYTHONPATH entries = %d（必须合成一条）: %v", count, env) + } + if !strings.Contains(value, "/opt/pydeps") || !strings.Contains(value, "/data/skills/house-style") { + t.Fatalf("PYTHONPATH = %q（应同时含依赖层与 house-style）", value) + } +} diff --git a/internal/tools/exec_interpreter.go b/internal/tools/exec_interpreter.go index d1dd945..7c877a3 100644 --- a/internal/tools/exec_interpreter.go +++ b/internal/tools/exec_interpreter.go @@ -55,6 +55,12 @@ var skillsRoot string // SetSkillsRoot 让 run_python / run_shell 能 import housestyle，并看到技能目录。 func SetSkillsRoot(dir string) { skillsRoot = dir } +// execDepsDir 是 sandbox 档的只读依赖层（site-packages 等）；组合根启动时注入。 +var execDepsDir string + +// SetExecDepsDir 配置 sandbox 档依赖层目录（空 = 不挂载，现状）。 +func SetExecDepsDir(dir string) { execDepsDir = dir } + // execEnv 最小环境：PATH/HOME/LANG/TZ/TERM 继承（pip --user、字体、中文 locale）， // 其余变量不带入子进程。技能根配置后追加 AGENT_SKILLS_DIR 与 PYTHONPATH。 func execEnv() []string { @@ -68,11 +74,16 @@ func execEnv() []string { } } out = append(out, "PYTHONDONTWRITEBYTECODE=1") + var pypaths []string + if execDepsDir != "" { + pypaths = append(pypaths, execDepsDir) + } if skillsRoot != "" { - out = append(out, - "AGENT_SKILLS_DIR="+skillsRoot, - "PYTHONPATH="+filepath.Join(skillsRoot, "house-style"), - ) + out = append(out, "AGENT_SKILLS_DIR="+skillsRoot) + pypaths = append(pypaths, filepath.Join(skillsRoot, "house-style")) + } + if len(pypaths) > 0 { // 合成一条，避免重复键互相覆盖 + out = append(out, "PYTHONPATH="+strings.Join(pypaths, string(os.PathListSeparator))) } return out } @@ -148,7 +159,7 @@ func runInterpreter(parent context.Context, interpreter, scriptPath, dir string, defer cancel() spec := ExecSpec{ Bin: bin, Args: []string{scriptPath}, Dir: dir, Env: execEnv(), - SkillsDir: skillsRoot, Timeout: time.Duration(timeoutS) * time.Second, + SkillsDir: skillsRoot, DepsDir: execDepsDir, Timeout: time.Duration(timeoutS) * time.Second, } pr, pw := io.Pipe() var buf bytes.Buffer