Compare commits

...
Author SHA1 Message Date
weichen a5da18b710 Inject the active tool call into the Execute context
ci / vet-and-test (pull_request) Successful in 50s
ci / docker-checkout-hostgateway (pull_request) Successful in 1s
Why: With parallel tool execution the host-side tool wrapper can no
longer infer which invocation it is serving from a single in-flight
slot; tools need the call identity (ID/name/arguments) to pair process
state and per-call bookkeeping. Context injection keeps the tools
registry signature unchanged.

Evidence: go test ./... green; TestLoopToolSeesOwnCallInContext runs
two parallel calls and asserts each reads its own ID from ctx.

Impact: Additive WithToolCall/ToolCallFromContext helpers; runOne now
wraps the ctx before Execute. Callers that never inject keep ok=false
and are unaffected.

Refs: Dev task feature/agent-e1-parallel-fixes (F5 parallel wiring).
2026-10-10 08:48:26 +00:00
weichen d6267fba15 Merge pull request 'Carry tool CallID in loop events and add reasoning_split to the chat client' (#42) from feature/e1-parallel-reasoning into main
ci / vet-and-test (push) Successful in 49s
ci / docker-checkout-hostgateway (push) Successful in 0s
2026-10-10 08:36:09 +00:00
weichen 1f25049492 Carry tool CallID in loop events and add reasoning_split to the chat client
ci / vet-and-test (pull_request) Successful in 50s
ci / docker-checkout-hostgateway (pull_request) Successful in 1s
Why: Parallel tool execution makes same-turn tool events interleave, so
consumers can no longer pair starts and ends by order; the kernel Event
needs the call ID. MiniMax interleaved-thinking models stream their
reasoning inside content <think> blocks unless the OpenAI-compatible
request opts in with reasoning_split=true, which forces reasoning into
the dedicated reasoning_content field the SSE layer already captures.

Evidence: go test ./... green; new tests TestLoopToolEventsCarryCallID
and TestOpenAIChatBodyReasoningSplit pin both behaviors (regression
anchors for the backend parallel-work change).

Impact: Additive Event.CallID (empty for non-tool events; existing
consumers unaffected) and an opt-in Client.ReasoningSplit flag that only
adds a request field when set. No default behavior change.

Refs: Agent-System-Dev proposal 2026-10-10-agent-delegation-routing-sandbox.md
(F2/F5), Dev task feature/agent-e1-parallel-fixes.
2026-10-10 08:33:53 +00:00
weichen 76b6171de5 Merge pull request 'Capture OpenAI-compatible reasoning and replay it in assistant history' (#41) from feature/quality-p2-i2 into main
ci / vet-and-test (push) Successful in 49s
ci / docker-checkout-hostgateway (push) Successful in 1s
2026-10-09 16:07:23 +00:00
5 changed files with 201 additions and 7 deletions
+9
View File
@@ -20,6 +20,12 @@ type Client struct {
HTTPClient *http.Client
// Protocol 空 = openai-chat(零值兼容所有直接构造点)。
Protocol Protocol
// ReasoningSplit 在 openai-chat 请求体加 "reasoning_split": true:
// MiniMax 等 interleaved-thinking 提供商据此把思考放进独立的
// reasoning_content 字段而不是混在 content 的 <think> 块里(分流后
// SSE 的 reasoning_content 捕获照常生效)。对不认识该字段的服务器
// 属未知字段,多数按忽略处理;宿主按模型名自行决定是否置位。
ReasoningSplit bool
}
// Stream implements StreamFunc. Transport and provider failures are encoded as
@@ -67,6 +73,9 @@ func (c *Client) run(ctx context.Context, model string, messages []Message, tool
if effort := OpenAIReasoningEffort(opts.ThinkingLevel); effort != "" {
body["reasoning_effort"] = effort
}
if c.ReasoningSplit {
body["reasoning_split"] = true
}
if len(tools) > 0 {
body["tools"] = toProviderTools(tools)
body["tool_choice"] = "auto"
+47
View File
@@ -0,0 +1,47 @@
package llm
import (
"context"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
)
// TestOpenAIChatBodyReasoningSplit 验证 Client.ReasoningSplit 置位时
// openai-chat 请求体带 "reasoning_split": true,未置位时不带(未知字段
// 不外溢给别的提供商)。
func TestOpenAIChatBodyReasoningSplit(t *testing.T) {
var bodies []string
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
raw, _ := io.ReadAll(r.Body)
bodies = append(bodies, string(raw))
w.Header().Set("Content-Type", "text/event-stream")
_, _ = io.WriteString(w, "data: [DONE]\n\n")
}))
defer srv.Close()
drain := func(c *Client) {
for range c.Stream(context.Background(), "MiniMax-M3",
[]Message{{Role: RoleUser, Content: "hi"}}, nil, StreamOptions{}) {
}
}
drain(&Client{BaseURL: srv.URL, APIKey: "k"})
drain(&Client{BaseURL: srv.URL, APIKey: "k", ReasoningSplit: true})
if len(bodies) != 2 {
t.Fatalf("requests = %d", len(bodies))
}
if strings.Contains(bodies[0], "reasoning_split") {
t.Fatalf("flag off must not leak the field: %s", bodies[0])
}
var got map[string]any
if err := json.Unmarshal([]byte(bodies[1]), &got); err != nil {
t.Fatal(err)
}
if v, ok := got["reasoning_split"].(bool); !ok || !v {
t.Fatalf("flag on must set reasoning_split=true: %s", bodies[1])
}
}
+27 -7
View File
@@ -38,6 +38,9 @@ type Event struct {
Message llm.Message
Tool string
Text string
// CallID 在 tool_execution_start / tool_execution_end 时携带工具调用的
// ID:并行执行下同回合多个调用并发进出,消费方按 ID 配对而非按顺序。
CallID string
// Detail carries tool arguments/results or reasoning deltas for
// process-visualization consumers; log writers may ignore it.
Detail string
@@ -706,22 +709,38 @@ func noteToolHook(cfg Config, call llm.ToolCall, msg llm.Message) {
cfg.Hooks.Emit(hooks.Call{Event: ev, Tool: call.Name, Args: call.Arguments, Result: msg.Content, IsError: msg.IsError})
}
// toolCallKey 是把当前工具调用注入 Execute ctx 的私有键:并行执行下工具实现
// 需要知道自己服务于哪次调用(配对过程事件、按调用记账),经
// ToolCallFromContext 读取;旧宿主/直接构造的 ctx 返回 ok=false。
type toolCallKey struct{}
// WithToolCall 把 call 挂进 ctx(内核 runOne 在执行前注入)。
func WithToolCall(ctx context.Context, call llm.ToolCall) context.Context {
return context.WithValue(ctx, toolCallKey{}, call)
}
// ToolCallFromContext 取出当前正在执行的调用;不存在时 ok=false。
func ToolCallFromContext(ctx context.Context) (llm.ToolCall, bool) {
call, ok := ctx.Value(toolCallKey{}).(llm.ToolCall)
return call, ok
}
func runOne(ctx context.Context, cfg Config, call llm.ToolCall) llm.Message {
if err := ctx.Err(); err != nil {
return cancelledToolMessage(call, err)
}
if why := rejectIncompleteToolCall(call); why != "" {
msg := llm.Message{Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: why, IsError: true}
emit(cfg, Event{Type: "tool_execution_start", Tool: call.Name, Detail: call.Arguments})
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: why})
emit(cfg, Event{Type: "tool_execution_start", Tool: call.Name, CallID: call.ID, Detail: call.Arguments})
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, CallID: call.ID, Text: why})
if cfg.AfterTool != nil {
cfg.AfterTool(call, msg)
}
return msg
}
emit(cfg, Event{Type: "tool_execution_start", Tool: call.Name, Detail: call.Arguments})
emit(cfg, Event{Type: "tool_execution_start", Tool: call.Name, CallID: call.ID, Detail: call.Arguments})
if gated, msg, blocked := gateTool(ctx, cfg, call); blocked {
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: msg.Content})
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, CallID: call.ID, Text: msg.Content})
noteToolHook(cfg, call, msg)
if cfg.AfterTool != nil {
cfg.AfterTool(call, msg)
@@ -733,7 +752,7 @@ func runOne(ctx context.Context, cfg Config, call llm.ToolCall) llm.Message {
if cfg.BeforeTool != nil {
if err := cfg.BeforeTool(call); err != nil {
msg := llm.Message{Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: err.Error(), IsError: true}
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: msg.Content})
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, CallID: call.ID, Text: msg.Content})
noteToolHook(cfg, call, msg)
if cfg.AfterTool != nil {
cfg.AfterTool(call, msg)
@@ -744,7 +763,7 @@ func runOne(ctx context.Context, cfg Config, call llm.ToolCall) llm.Message {
emit(cfg, Event{Type: "tool_execution_update", Tool: call.Name, Detail: call.Arguments})
if err := commitToolIntent(cfg, call); err != nil {
msg := llm.Message{Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: err.Error(), IsError: true}
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: msg.Content})
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, CallID: call.ID, Text: msg.Content})
return msg
}
endSpan := startSpan(cfg, "execute_tool "+call.Name, map[string]string{
@@ -754,6 +773,7 @@ func runOne(ctx context.Context, cfg Config, call llm.ToolCall) llm.Message {
defer endSpan()
res := tools.Result{Content: "no tools configured", IsError: true}
known := cfg.Tools != nil && cfg.Tools.Has(call.Name)
ctx = WithToolCall(ctx, call)
switch {
case known:
res = cfg.Tools.Execute(ctx, call.Name, json.RawMessage(call.Arguments))
@@ -769,7 +789,7 @@ func runOne(ctx context.Context, cfg Config, call llm.ToolCall) llm.Message {
Content: res.Content,
IsError: res.IsError,
}
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: res.Content})
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, CallID: call.ID, Text: res.Content})
noteToolHook(cfg, call, msg)
if cfg.AfterTool != nil {
cfg.AfterTool(call, msg)
+69
View File
@@ -0,0 +1,69 @@
package runtime
import (
"context"
"encoding/json"
"sync"
"testing"
"agent-native/runtime/llm"
"agent-native/runtime/tools"
)
// ctxEchoTool 把自己从 ctx 读到的调用 ID 记下来(并发安全),验证内核
// 在 Execute 前注入了当前调用。
type ctxEchoTool struct {
mu sync.Mutex
ids []string
bad int
}
func (c *ctxEchoTool) Name() string { return "echo" }
func (c *ctxEchoTool) Description() string { return "echo" }
func (c *ctxEchoTool) Parameters() json.RawMessage { return json.RawMessage(`{"type":"object"}`) }
func (c *ctxEchoTool) Execute(ctx context.Context, _ json.RawMessage) (tools.Result, error) {
call, ok := ToolCallFromContext(ctx)
c.mu.Lock()
defer c.mu.Unlock()
if !ok || call.ID == "" || call.Name != "echo" {
c.bad++
return tools.Result{Content: "call missing from context", IsError: true}, nil
}
c.ids = append(c.ids, call.ID)
return tools.Result{Content: "ok " + call.ID}, nil
}
// TestLoopToolSeesOwnCallInContext 并行执行两个调用,各自读到自己的 ID。
func TestLoopToolSeesOwnCallInContext(t *testing.T) {
tool := &ctxEchoTool{}
reg := tools.NewRegistry(t.TempDir())
reg.Register(tool)
res, err := Run(context.Background(), Config{
Model: "fake",
Tools: reg,
ToolExecution: ExecParallel,
Stream: script(
llm.Message{Role: llm.RoleAssistant, StopReason: llm.StopToolCalls, ToolCalls: []llm.ToolCall{
{ID: "pa", Name: "echo", Arguments: `{}`},
{ID: "pb", Name: "echo", Arguments: `{}`},
}},
llm.Message{Role: llm.RoleAssistant, Content: "done", StopReason: llm.StopEnd},
),
}, llm.Message{Role: llm.RoleUser, Content: "go"})
if err != nil {
t.Fatal(err)
}
tool.mu.Lock()
defer tool.mu.Unlock()
if tool.bad != 0 || len(tool.ids) != 2 {
t.Fatalf("bad=%d ids=%v", tool.bad, tool.ids)
}
seen := map[string]bool{}
for _, id := range tool.ids {
seen[id] = true
}
if !seen["pa"] || !seen["pb"] {
t.Fatalf("ids = %v", tool.ids)
}
_ = res
}
+49
View File
@@ -0,0 +1,49 @@
package runtime
import (
"context"
"testing"
"agent-native/runtime/llm"
"agent-native/runtime/tools"
)
// TestLoopToolEventsCarryCallID 验证 tool_execution_start/end 事件带工具调用
// ID:并行执行下消费方按 ID 配对(回归锚点,防止字段被删后顺序假设回归)。
func TestLoopToolEventsCarryCallID(t *testing.T) {
hits := 0
reg := tools.NewRegistry(t.TempDir())
reg.Register(hitTool{hits: &hits})
var starts, ends []Event
res, err := Run(context.Background(), Config{
Model: "fake",
Tools: reg,
Stream: script(
llm.Message{Role: llm.RoleAssistant, StopReason: llm.StopToolCalls, ToolCalls: []llm.ToolCall{{
ID: "call-7", Name: "echo", Arguments: `{}`,
}}},
llm.Message{Role: llm.RoleAssistant, Content: "done", StopReason: llm.StopEnd},
),
OnEvent: func(ev Event) {
switch ev.Type {
case "tool_execution_start":
starts = append(starts, ev)
case "tool_execution_end":
ends = append(ends, ev)
}
},
}, llm.Message{Role: llm.RoleUser, Content: "go"})
if err != nil {
t.Fatal(err)
}
if hits != 1 || len(starts) != 1 || len(ends) != 1 {
t.Fatalf("hits=%d starts=%d ends=%d", hits, len(starts), len(ends))
}
if starts[0].CallID != "call-7" || starts[0].Tool != "echo" {
t.Fatalf("start = %+v", starts[0])
}
if ends[0].CallID != "call-7" || ends[0].Tool != "echo" {
t.Fatalf("end = %+v", ends[0])
}
_ = res
}