Inject the active tool call into the Execute context

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).

runtime/llm/anthropic.go

@@ -284,7 +284,7 @@ func parseAnthropicSSE(ctx context.Context, r io.Reader, out chan<- Event) {
 				}
 			case "content_block_start":
 				var v struct {
-					Index       int `json:"index"`
+					Index        int `json:"index"`
 					ContentBlock struct {
 						Type string `json:"type"`
 						ID   string `json:"id"`

runtime/llm/client.go

@@ -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"
@@ -121,6 +130,18 @@ func (c *Client) run(ctx context.Context, model string, messages []Message, tool
 	}
 }
 
+// replayThink 拼接一条 assistant 消息捕获到的全部思考块。
+func replayThink(m Message) string {
+	var b strings.Builder
+	for _, t := range m.Thinking {
+		if t.Thinking != "" {
+			b.WriteString(t.Thinking)
+			b.WriteString("\n")
+		}
+	}
+	return strings.TrimRight(b.String(), "\n")
+}
+
 func toProviderMessages(messages []Message) []map[string]any {
 	out := make([]map[string]any, 0, len(messages))
 	for _, m := range messages {
@@ -132,7 +153,14 @@ func toProviderMessages(messages []Message) []map[string]any {
 				"content":      m.Content,
 			})
 		case RoleAssistant:
-			item := map[string]any{"role": "assistant", "content": m.Content}
+			// 回放捕获的思考（Message.Thinking 由 SSE 的 reasoning_content 捕获而来）：
+			// 交错思考模型要求历史 assistant 原样带回 <think> 块。没有思考块的
+			// 消息零改动——非思考模型的上下文不受影响。
+			content := m.Content
+			if think := replayThink(m); think != "" {
+				content = "<think>\n" + think + "\n</think>\n\n" + content
+			}
+			item := map[string]any{"role": "assistant", "content": content}
 			if len(m.ToolCalls) > 0 {
 				calls := make([]map[string]any, 0, len(m.ToolCalls))
 				for _, tc := range m.ToolCalls {

runtime/llm/reasoning_split_test.go

@@ -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])
+	}
+}

runtime/llm/sse.go

@@ -52,6 +52,7 @@ func ParseSSE(ctx context.Context, r io.Reader) <-chan Event {
 
 func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
 	var content strings.Builder
+	var reasoning strings.Builder
 	calls := map[int]*ToolCall{}
 	order := []int{}
 	var usage Usage
@@ -108,8 +109,12 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
 		}
 		// reasoning_content（国产 OpenAI 兼容模型的思考链）与正文分流：
 		// 只作为 reasoning_delta 事件外发，不计入 Message.Content（与 runner
-		// stripThinkBlocks"正文无 think"语义对齐）。
+		// stripThinkBlocks"正文无 think"语义对齐）。同时累计进 reasoning——
+		// 交错思考模型（MiniMax/DeepSeek R1 等）要求历史 assistant 原样带回
+		// 思考内容，丢弃会降低后续轮质量；捕获进 Message.Thinking 随会话
+		// 转录持久化，请求序列化侧按需回放。
 		if choice.Delta.Reasoning != "" {
+			reasoning.WriteString(choice.Delta.Reasoning)
 			select {
 			case <-ctx.Done():
 				emitAbort(out, content.String(), ctx.Err())
@@ -166,6 +171,9 @@ func emitParsed(ctx context.Context, r io.Reader, out chan<- Event) {
 		order = nil
 	}
 	msg := Message{Role: RoleAssistant, Content: content.String(), StopReason: mapFinish(finished, len(order) > 0)}
+	if reasoning.Len() > 0 {
+		msg.Thinking = []ThinkingBlock{{Thinking: reasoning.String()}}
+	}
 	for _, idx := range order {
 		msg.ToolCalls = append(msg.ToolCalls, *calls[idx])
 	}

runtime/llm/think_replay_test.go

@@ -0,0 +1,45 @@
+package llm
+
+import (
+	"strings"
+	"testing"
+)
+
+// 交错思考模型（MiniMax 等）要求历史 assistant 原样带回思考内容：
+// SSE 把 reasoning_content 捕获进 Message.Thinking；序列化把它作为
+// <think> 前缀回放。没有思考块的消息保持零改动。
+func TestReasoningCaptureAndReplay(t *testing.T) {
+	// 回放侧。
+	msgs := []Message{
+		{Role: RoleUser, Content: "总结"},
+		{Role: RoleAssistant, Content: "结论。", Thinking: []ThinkingBlock{{Thinking: "先读材料。"}}},
+		{Role: RoleAssistant, Content: "没有思考的轮次。"},
+	}
+	out := toProviderMessages(msgs)
+	if got := out[1]["content"].(string); !strings.HasPrefix(got, "<think>\n先读材料。\n</think>") || !strings.HasSuffix(got, "结论。") {
+		t.Fatalf("assistant replay malformed: %q", got)
+	}
+	if got := out[2]["content"].(string); got != "没有思考的轮次。" {
+		t.Fatalf("non-thinking assistant must be untouched: %q", got)
+	}
+	if _, ok := out[0]["tool_calls"]; ok {
+		t.Fatal("user messages carry no tool_calls")
+	}
+	// 捕获侧：直接构造 SSE 流。
+	sse := "data: {\"choices\":[{\"delta\":{\"reasoning_content\":\"思考\"}}]}\n\n" +
+		"data: {\"choices\":[{\"delta\":{\"content\":\"正文\"}}]}\n\n" +
+		"data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n" +
+		"data: [DONE]\n\n"
+	var end Event
+	for ev := range ParseSSE(t.Context(), strings.NewReader(sse)) {
+		if ev.Kind == KindMessageEnd {
+			end = ev
+		}
+	}
+	if end.Message.Content != "正文" {
+		t.Fatalf("content polluted by reasoning: %q", end.Message.Content)
+	}
+	if len(end.Message.Thinking) != 1 || end.Message.Thinking[0].Thinking != "思考" {
+		t.Fatalf("reasoning not captured: %+v", end.Message.Thinking)
+	}
+}

runtime/llm/types.go

@@ -52,9 +52,9 @@ type Message struct {
 	Images     []Image    `json:"images,omitempty"`
 	ToolCallID string     `json:"tool_call_id,omitempty"`
 	ToolCalls  []ToolCall `json:"tool_calls,omitempty"`
-	// Thinking 是原生 Anthropic 路径的思考块（含签名），随 assistant 消息
-	// 在请求里原样往返（Anthropic 要求思考块在内容块最前且签名不失）。
-	// OpenAI 路径不写这个字段（reasoning 不落盘，见 sse.go 注释）。
+	// Thinking 是思考块：Anthropic 原生路径含签名、在请求里原样往返；
+	// OpenAI 兼容路径由 SSE 的 reasoning_content 捕获（不含签名），随会话
+	// 转录持久化，请求序列化时按 <think> 块回放（见 client.go replayThink）。
 	Thinking   []ThinkingBlock `json:"thinking,omitempty"`
 	StopReason StopReason      `json:"stop_reason,omitempty"`
 	IsError    bool            `json:"is_error,omitempty"`

runtime/loop.go

@@ -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)

runtime/loop_callctx_test.go

@@ -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
+}

runtime/loop_callid_test.go

@@ -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
+}