Agent-Native/runtime/loop.go

package runtime

import (
	"context"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"sync"
	"time"

	"agent-native/runtime/hooks"
	"agent-native/runtime/llm"
	"agent-native/runtime/memory"
	"agent-native/runtime/reminders"
	"agent-native/runtime/session"
	"agent-native/runtime/telemetry"
	"agent-native/runtime/tools"

	"logging-system/logsys"
)

const (
	ExecSequential = "sequential"
	ExecParallel   = "parallel"

	maxStopContinuations   = 3
	maxLengthContinuations = 3
	maxRepeatReminders     = 3
)

const repeatReminderText = "你已连续 3 次以相同参数调用同一工具。如果结果不符合预期，请更换参数或做法，不要再重复这一调用。"
const lengthContinuePrompt = "上一段回复因为达到长度上限被截断。请从断点接着写，不要重复已经写出的内容。"

// Event is a lifecycle notice for logs and UI.
type Event struct {
	Type    string
	Message llm.Message
	Tool    string
	Text    string
	// ToolIndex preserves the provider's call slot for tool_call_delta.
	ToolIndex int
	// CallID 在 tool_execution_start / tool_execution_end 时携带工具调用的
	// ID；tool_call_delta 也携带同一 ID。消费方按 ID 配对而非按顺序。
	CallID string
	// Detail carries tool arguments/results or reasoning deltas for
	// process-visualization consumers; log writers may ignore it.
	Detail string
	// PromptTokens / CompletionTokens / TotalTokens 在 type=usage 或
	// type=message_end 时有意义，供调用方累计用量。
	PromptTokens     int
	CompletionTokens int
	TotalTokens      int
	// 以下在 type=usage 时有意义（WP6：上一请求与缓存命中拆分）。
	LastPromptTokens     int
	LastCompletionTokens int
	LastTotalTokens      int
	CacheReadTokens      int
	CacheWriteTokens     int
}

// Config drives one agent session.
type Config struct {
	Model        string
	SystemPrompt string
	Tools        *tools.Registry
	Stream       llm.StreamFunc
	Session      *session.Store
	// Trace 开始一个 span，返回的函数结束它。nil 表示不采集。
	// 属性里不要放提示词正文。
	Trace         func(name string, attrs map[string]string) func()
	MaxTurns      int
	ToolExecution string
	// SerialTools 是并行执行下的顺序敏感工具名集合：这些调用按模型给出的
	// 原顺序在调用方 goroutine 逐个执行（互斥且保序），其余调用并发执行。
	SerialTools map[string]bool
	GetSteering func() []llm.Message
	GetFollowUp func() []llm.Message
	// OnEvent 观察循环事件。同一次 Run 内不会并发调用
	// （与 BeforeTool、AfterTool 共用一把锁串行）。
	OnEvent func(Event)
	// ThinkingLevel is forwarded on each provider request.
	ThinkingLevel string
	// Resolve, when set, overrides Model and ThinkingLevel at the start of each turn.
	Resolve func() (model, thinking string)
	// Log 可选。nil 时丢弃日志，不改变循环结果。
	Log *logsys.Logger
	// ExecutionID 只用于日志字段，不参与控制流。
	ExecutionID string
	// Budget 可选。任一上限命中即停止并返回 budget_exceeded 错误（StopError）；
	// 零值字段不设限。截止时刻与外层 ctx 先到者生效。
	Budget *Budget
	// History 是会话转录为空时的初始转录。以 system 开头时，不再前置
	// SystemPrompt 和工具清单。不回写进 Session。
	History []llm.Message
	// Stop 在模型不再调用工具、且本轮没有 follow-up 时询问是否续跑。
	// nil 表示不询问。内核最多接受 3 次 Continue。
	Stop func(last llm.Message, info StopInfo) StopDecision
	// BeforeTool 返回错误时不执行工具，但仍写一条 IsError 的 tool 结果。
	// 同一次 Run 内与 OnEvent、AfterTool 串行，不会并发调用。
	BeforeTool func(call llm.ToolCall) error
	// UnknownTool 处理注册表里没有的调用。nil 时仍返回 unknown tool。
	UnknownTool func(call llm.ToolCall) tools.Result
	// AfterTool 只观察工具结果，不能改写。
	// 同一次 Run 内与 OnEvent、BeforeTool 串行，不会并发调用。
	AfterTool func(call llm.ToolCall, res llm.Message)
	// Reminders 在每次调用模型前提供提醒。正文包在 system-reminder 里，
	// 并入最后一条工具结果或用户消息，只进入本次请求，不写进 Session。
	Reminders func(turn int) []string
	// RepeatGuard 为真时，同名且参数按键排序后相同的调用连续 3 次，注入一条提醒。
	// 每个 run 最多 3 条，不中断循环。
	RepeatGuard bool
	// ContinueOnLength 为真时，FinishReason=length 且没有工具调用就续写。
	// 续写提示不写进 Session。次数由 MaxLengthContinuations 决定。
	ContinueOnLength bool
	// MaxLengthContinuations 是 length 续写的次数上限。
	// 0 表示默认 3 次；负数表示不限制。
	MaxLengthContinuations int
	// ContextWindow 是模型窗口的字符估算。0 表示用保守默认值。
	// 超过窗口减去 OutputReserve 与 ContextBuffer 时，清理最近若干条之前的旧工具结果。
	ContextWindow int
	// OutputReserve 是留给模型输出的字符数。0 表示默认预留。
	OutputReserve int
	// ContextBuffer 是窗口估算的余量。0 表示默认余量。
	ContextBuffer int
	// KeepToolResults 是清理时保留的最近工具结果条数。0 表示 5。
	KeepToolResults int
	// ResultsDir 非空时，被清理的旧工具结果完整内容落盘到该目录
	// （占位写路径）；空则占位指向会话记录（旧行为）。
	ResultsDir string
	// JournalOnly 为真且 Session 非空时：Session 只作意图日志与恢复配对
	// （不恢复的完整权威上下文），请求上下文始终由 History 组装；
	// 本次恢复写入的配对摘要以系统消息注入（不持久化，已在 journal 里）。
	// 为假（默认）保持既有语义：Session 为权威上下文。
	JournalOnly bool
	// AgentName 是 invoke_agent span 的名字后缀（OTel GenAI 约定
	// "invoke_agent <agent>"）。空 = "agent-native"。
	AgentName string
	// ProviderName 是 gen_ai.system 属性值（如 openai / anthropic）。
	// 空 = "openai"。
	ProviderName string
	// Memory 是记忆端口（N3 WP5）：nil = 不启用。启用时每次请求前
	// Recall 一次并以系统消息注入请求副本（不落盘、不累计）。
	Memory memory.Store
	// MemoryScope 是 Recall 的 scope 过滤（"" = 不限，含 user/project；
	// AgentSpec.MemoryScope 的消费点）。
	MemoryScope string
	// RetryAttempts 是流式错误的默认重试次数（WP6 下沉为库默认）：
	// 0 值 = 2 次（默认）；负值 = 关闭；正数 = 次数。
	// 只在还没向调用方发出增量时重试（WithRetry 既有语义）。
	RetryAttempts int
	// traceCtx 是本次 Run 的 span 关联（Run 启动时建立；指针随 Config
	// 副本传播到各轮与各工具）。
	traceCtx *traceIDs
	// Hooks 是进程内钩子。空总线不改变停止、提醒和拒绝次数。
	Hooks *hooks.Bus
	// PermissionAsk 在钩子对一次工具调用表态 ask 时向宿主请求裁决。
	// nil 时保持旧行为：ask 与 deny 一样按拒绝处理。
	// 返回 Allow 时工具照常执行；宿主需要「本会话始终允许」时自行记录，
	// 内核不替宿主保存放行状态。
	PermissionAsk func(ctx context.Context, call llm.ToolCall) AskVerdict
}

// AskVerdict 是宿主对一次 ask 决策的裁决。
type AskVerdict struct {
	Allow  bool
	Reason string // 拒绝时写进配对 tool 结果的理由；放行时忽略
}

// StopInfo 是 Stop 钩子看到的本 run 进度。
type StopInfo struct {
	Continuations int
	ToolCalls     int
}

// StopDecision 决定模型停下之后是否再跑一轮。
type StopDecision struct {
	Continue    bool
	Replacement string
	Reason      string
	Notice      string
}

// Budget 是循环层的执行预算；与 exec.Budget 字段同构，由适配层换算。
type Budget struct {
	MaxPromptTokens     int
	MaxCompletionTokens int
	MaxToolCalls        int
	Deadline            time.Time
}

// Result is the final assistant text and the transcript produced by one run.
type Result struct {
	Text       string
	Messages   []llm.Message
	StopReason llm.StopReason
	Usage      llm.Usage
	Notice     string
	Err        error
	Retryable  bool
}

// Run executes the turn loop until the model stops, follow-ups drain, or the turn cap is hit.
func Run(ctx context.Context, cfg Config, user llm.Message) (Result, error) {
	if cfg.Stream == nil {
		return Result{StopReason: llm.StopError, Err: errString("stream func is required")}, nil
	}
	if cfg.MaxTurns <= 0 {
		cfg.MaxTurns = 12
	}
	if cfg.ToolExecution == "" {
		cfg.ToolExecution = ExecSequential
	}
	serializeHostCallbacks(&cfg)
	if cfg.Memory != nil && cfg.Tools != nil {
		memory.RegisterWrite(cfg.Tools, cfg.Memory, cfg.MemoryScope)
	}
	// 流式错误默认重试（WP6）：库默认 2 次、负值关闭；重试事件进事件面。
	if cfg.Stream != nil {
		attempts := cfg.RetryAttempts
		if attempts == 0 {
			attempts = 2
		}
		if attempts > 0 {
			cfg.Stream = llm.WithRetry(cfg.Stream, attempts, func(kind, detail string) {
				emit(cfg, Event{Type: kind, Text: detail})
			})
		}
	}
	cfg.traceCtx = newTraceContext()
	endRun := startSpan(cfg, "invoke_agent "+agentName(cfg), map[string]string{
		"gen_ai.system":        providerName(cfg),
		"gen_ai.agent.name":    agentName(cfg),
		"gen_ai.request.model": cfg.Model,
	}, true)
	defer endRun()
	var messages []llm.Message
	if cfg.Session != nil && !cfg.JournalOnly {
		loaded, err := cfg.Session.Load()
		if err != nil {
			return Result{}, err
		}
		messages = loaded
	}
	recovered, err := recoverPending(ctx, cfg)
	if err != nil {
		return Result{}, err
	}
	// 尾部配对兜底：无论进程死在 persist(assistant) 与 commitToolIntent
	// 之间的哪个窗口，补全尾部 assistant 的无结果调用，转录配对恒成立。
	repaired, err := repairOrphanTail(cfg)
	if err != nil {
		return Result{}, err
	}
	recovered = append(recovered, repaired...)
	if cfg.Session != nil && !cfg.JournalOnly {
		loaded, err := cfg.Session.Load()
		if err != nil {
			return Result{}, err
		}
		messages = loaded
	}
	seeded := false
	if len(messages) == 0 && len(cfg.History) > 0 {
		messages = append([]llm.Message{}, cfg.History...)
		seeded = true
	}
	prependedSys := false
	if len(messages) == 0 || messages[0].Role != llm.RoleSystem {
		sys := llm.Message{Role: llm.RoleSystem, Content: systemText(cfg)}
		messages = append([]llm.Message{sys}, messages...)
		if err := persist(cfg, sys); err != nil {
			return Result{}, err
		}
		prependedSys = true
	}
	// 种子落盘（N3 WP2）：转录为权威上下文时，History 只作一次性种子——
	// 不落盘则第二跑起种子静默丢失。已落盘的前置系统消息不重复写；
	// JournalOnly 时种子不进 journal（journal 只收本轮起的消息）。
	if seeded && !cfg.JournalOnly {
		for i, m := range messages {
			if i == 0 && prependedSys {
				continue
			}
			if err := persist(cfg, m); err != nil {
				return Result{}, err
			}
		}
	}
	if cfg.JournalOnly && len(recovered) > 0 {
		// 恢复摘要以系统消息注入上下文（不持久化——内容已在 journal 里）。
		messages = append(messages, llm.Message{Role: llm.RoleSystem, Content: recoveredNoteText(recovered)})
	}
	if err := persist(cfg, user); err != nil {
		return Result{}, err
	}
	messages = append(messages, user)
	emit(cfg, Event{Type: "agent_start"})
	cfg.Hooks.Emit(hooks.Call{Event: hooks.SessionStart})
	if out := cfg.Hooks.Emit(hooks.Call{Event: hooks.UserPromptSubmit, LastText: user.Content}); out.Decision != "" {
		switch out.Decision {
		case hooks.Deny, hooks.Block:
			// 输入护栏拒绝：不进入循环，以理由直接收尾（原文已持久化留痕）。
			why := out.Reason
			if why == "" {
				why = "输入被护栏拒绝"
			}
			return Result{Text: why, Messages: messages, StopReason: llm.StopEnd, Notice: why}, nil
		case hooks.Modify:
			// 改写后的输入追加为新的用户消息（原文保留，可审计）。
			if out.Replacement != "" {
				note := llm.Message{Role: llm.RoleUser, Content: out.Replacement}
				if err := persist(cfg, note); err != nil {
					return Result{}, err
				}
				messages = append(messages, note)
				user = note
			}
		}
	}

	var last llm.Message
	var usage llm.Usage
	toolCalls := 0
	stopContinues := 0
	lengthContinues := 0
	lengthNudge := false
	var repeats repeatState
	defer func() {
		cfg.Hooks.Emit(hooks.Call{Event: hooks.SessionEnd, LastText: last.Content, ToolCalls: toolCalls})
	}()
	for turn := 0; turn < cfg.MaxTurns; turn++ {
		if ctx.Err() != nil {
			cfg.logger().Error("runtime turn aborted", ctx.Err())
			return Result{Text: last.Content, Messages: messages, StopReason: llm.StopAborted, Err: ctx.Err()}, nil
		}
		if reason := budgetExceeded(cfg.Budget, usage, toolCalls, time.Now()); reason != "" {
			return budgetStop(cfg, last, messages, usage, reason), nil
		}
		emit(cfg, Event{Type: "turn_start"})
		emit(cfg, Event{Type: "message_start"})
		request := prepareRequest(cfg, messages, turn, &repeats, &lengthNudge)
		end, errEv := consume(ctx, cfg, request)
		if errEv != nil {
			cfg.logger().Error("runtime turn failed", errEv, "retryable", end.Retryable)
			emitUsage(cfg, usage)
			return Result{Text: end.Message.Content, Messages: messages, StopReason: end.Message.StopReason, Err: errEv, Retryable: end.Retryable}, nil
		}
		last = end.Message
		usage.PromptTokens += end.Usage.PromptTokens
		usage.CompletionTokens += end.Usage.CompletionTokens
		usage.TotalTokens += end.Usage.TotalTokens
		// cache 只入累计位（WP6 决策：不加 Last* cache 字段，避免字段膨胀）。
		usage.CacheReadTokens += end.Usage.CacheReadTokens
		usage.CacheWriteTokens += end.Usage.CacheWriteTokens
		usage.LastPromptTokens = end.Usage.PromptTokens
		usage.LastCompletionTokens = end.Usage.CompletionTokens
		usage.LastTotalTokens = end.Usage.TotalTokens
		if end.Err != nil {
			cfg.logger().Error("runtime terminal stream failed", end.Err)
			emitUsage(cfg, usage)
			return Result{Text: last.Content, Messages: messages, StopReason: llm.StopError,
				Usage: usage, Err: end.Err, Retryable: false}, nil
		}
		if err := persist(cfg, last); err != nil {
			return Result{}, err
		}
		messages = append(messages, last)
		emit(cfg, Event{Type: "message_end", Message: last, PromptTokens: end.Usage.PromptTokens, CompletionTokens: end.Usage.CompletionTokens, TotalTokens: end.Usage.TotalTokens, CacheReadTokens: end.Usage.CacheReadTokens, CacheWriteTokens: end.Usage.CacheWriteTokens})

		if last.StopReason == llm.StopToolCalls && len(last.ToolCalls) > 0 {
			if reason := budgetExceeded(cfg.Budget, usage, toolCalls, time.Now()); reason != "" {
				var err error
				messages, err = appendSkippedToolResults(cfg, last.ToolCalls, messages)
				if err != nil {
					return Result{}, err
				}
				return budgetStop(cfg, last, messages, usage, reason), nil
			}
			if cfg.Budget != nil && cfg.Budget.MaxToolCalls > 0 && toolCalls+len(last.ToolCalls) > cfg.Budget.MaxToolCalls {
				reason := fmt.Sprintf("budget_exceeded: tool calls %d+%d > max %d", toolCalls, len(last.ToolCalls), cfg.Budget.MaxToolCalls)
				var err error
				messages, err = appendSkippedToolResults(cfg, last.ToolCalls, messages)
				if err != nil {
					return Result{}, err
				}
				return budgetStop(cfg, last, messages, usage, reason), nil
			}
			toolCalls += len(last.ToolCalls)
			if cfg.RepeatGuard {
				repeats.observe(last.ToolCalls)
			}
			results := executeTools(ctx, cfg, last)
			for _, res := range results {
				if err := persist(cfg, res); err != nil {
					return Result{}, err
				}
				messages = append(messages, res)
			}
			if steered := drain(cfg.GetSteering); len(steered) > 0 {
				for _, m := range steered {
					if err := persist(cfg, m); err != nil {
						return Result{}, err
					}
					messages = append(messages, m)
				}
			}
			cfg.Hooks.Emit(hooks.Call{Event: hooks.PostToolBatch, Turn: turn, ToolCalls: len(last.ToolCalls)})
			emit(cfg, Event{Type: "turn_end", Message: last})
			continue
		}
		emit(cfg, Event{Type: "turn_end", Message: last})
		if cfg.ContinueOnLength && end.FinishReason == "length" && len(last.ToolCalls) == 0 {
			lengthCap := cfg.MaxLengthContinuations
			if lengthCap == 0 {
				lengthCap = maxLengthContinuations
			}
			if lengthCap > 0 && lengthContinues >= lengthCap {
				errCap := errString("length continuation exceeded")
				cfg.logger().Error("runtime length continuation exceeded", errCap)
				return Result{Text: last.Content, Messages: messages, StopReason: llm.StopError, Usage: usage, Err: errCap}, nil
			}
			lengthContinues++
			lengthNudge = true
			continue
		}
		if follow := drain(cfg.GetFollowUp); len(follow) > 0 {
			for _, m := range follow {
				if err := persist(cfg, m); err != nil {
					return Result{}, err
				}
				messages = append(messages, m)
			}
			continue
		}
		if decision, hooked := stopDecision(cfg, last, stopContinues, toolCalls); hooked {
			if decision.Continue && stopContinues < maxStopContinuations {
				stopContinues++
				if decision.Replacement != "" && len(messages) > 0 {
					// 替换文只以 user 消息追加（N3 WP1/N6：不再改写内存里的
					// assistant——内存与转录保持一致；模型经该消息看到更正）。
					note := llm.Message{Role: llm.RoleUser, Content: decision.Replacement}
					if err := persist(cfg, note); err != nil {
						return Result{}, err
					}
					messages = append(messages, note)
				}
				if decision.Reason != "" {
					reasonMsg := llm.Message{Role: llm.RoleUser, Content: decision.Reason}
					if err := persist(cfg, reasonMsg); err != nil {
						return Result{}, err
					}
					messages = append(messages, reasonMsg)
				}
				continue
			}
			return settle(cfg, last, messages, usage, decision.Notice), nil
		}
		return settle(cfg, last, messages, usage, ""), nil
	}
	errCap := errString("max turns exceeded")
	cfg.logger().Error("runtime turn cap hit", errCap)
	return Result{
		Text:       last.Content,
		Messages:   messages,
		StopReason: llm.StopError,
		Usage:      usage,
		Err:        errCap,
	}, nil
}

func (cfg Config) logger() *logsys.Logger {
	if cfg.Log == nil {
		return logsys.Nop()
	}
	log := cfg.Log.Module("runtime")
	if cfg.ExecutionID == "" {
		return log
	}
	return log.With("execution_id", cfg.ExecutionID)
}

func systemText(cfg Config) string {
	var b strings.Builder
	b.WriteString(cfg.SystemPrompt)
	if cfg.Tools == nil {
		return b.String()
	}
	decls := cfg.Tools.Decls()
	if len(decls) == 0 {
		return b.String()
	}
	if b.Len() > 0 {
		b.WriteString("\n\n")
	}
	b.WriteString("Tools available in this workspace:")
	for _, d := range decls {
		b.WriteString("\n- ")
		b.WriteString(d.Name)
		b.WriteString(": ")
		b.WriteString(d.Description)
	}
	return b.String()
}

func consume(ctx context.Context, cfg Config, messages []llm.Message) (llm.Event, error) {
	spanAttrs := map[string]string{
		"gen_ai.system":        providerName(cfg),
		"gen_ai.request.model": cfg.Model,
	}
	endSpan := startSpan(cfg, "chat "+cfg.Model, spanAttrs, false)
	defer endSpan()
	var decls []llm.ToolDecl
	if cfg.Tools != nil {
		for _, d := range cfg.Tools.Decls() {
			decls = append(decls, llm.ToolDecl{Name: d.Name, Description: d.Description, Parameters: d.Parameters})
		}
	}
	model := cfg.Model
	opts := llm.StreamOptions{ThinkingLevel: cfg.ThinkingLevel}
	if cfg.Resolve != nil {
		if m, th := cfg.Resolve(); m != "" || th != "" {
			if m != "" {
				model = m
			}
			if th != "" {
				opts.ThinkingLevel = th
			}
		}
	}
	var end llm.Event
	sawEnd := false
	for ev := range cfg.Stream(ctx, model, messages, decls, opts) {
		switch ev.Kind {
		case llm.KindTextDelta:
			emit(cfg, Event{Type: "text_delta", Text: ev.Text})
		case llm.KindReasoningDelta:
			emit(cfg, Event{Type: "reasoning_delta", Text: ev.ReasoningDelta})
		case llm.KindToolCallDelta:
			emit(cfg, Event{Type: "tool_call_delta", ToolIndex: ev.ToolIndex,
				CallID: ev.ToolID, Tool: ev.ToolName, Detail: ev.ArgumentsDelta})
		case llm.KindError:
			return ev, ev.Err
		case llm.KindMessageEnd:
			end = ev
			sawEnd = true
		}
	}
	// usage 在 span 结束前补进属性（结束时过滤导出，见 telemetry 工厂）。
	if sawEnd {
		spanAttrs["gen_ai.usage.input_tokens"] = itoa(end.Usage.PromptTokens)
		spanAttrs["gen_ai.usage.output_tokens"] = itoa(end.Usage.CompletionTokens)
	}
	if !sawEnd {
		return llm.Event{Message: llm.Message{StopReason: llm.StopError}}, errString("stream ended without a message")
	}
	return end, nil
}

const toolBudgetSkipContent = "not executed: tool budget exceeded"

// appendSkippedToolResults 为未执行的 tool_call 各写一条错误结果，保持调用与结果成对。
// 不发工具事件：这些调用没有进入执行。
func appendSkippedToolResults(cfg Config, calls []llm.ToolCall, messages []llm.Message) ([]llm.Message, error) {
	for _, call := range calls {
		msg := llm.Message{
			Role:       llm.RoleTool,
			Name:       call.Name,
			ToolCallID: call.ID,
			Content:    toolBudgetSkipContent,
			IsError:    true,
		}
		if err := persist(cfg, msg); err != nil {
			return nil, err
		}
		messages = append(messages, msg)
	}
	return messages, nil
}

// serializeHostCallbacks 让同一次 Run 的宿主回调不会并发进入。
func serializeHostCallbacks(cfg *Config) {
	var mu sync.Mutex
	if cfg.OnEvent != nil {
		inner := cfg.OnEvent
		cfg.OnEvent = func(ev Event) {
			mu.Lock()
			defer mu.Unlock()
			inner(ev)
		}
	}
	if cfg.BeforeTool != nil {
		inner := cfg.BeforeTool
		cfg.BeforeTool = func(call llm.ToolCall) error {
			mu.Lock()
			defer mu.Unlock()
			return inner(call)
		}
	}
	if cfg.AfterTool != nil {
		inner := cfg.AfterTool
		cfg.AfterTool = func(call llm.ToolCall, res llm.Message) {
			mu.Lock()
			defer mu.Unlock()
			inner(call, res)
		}
	}
}

func executeTools(ctx context.Context, cfg Config, assistant llm.Message) []llm.Message {
	if cfg.ToolExecution == ExecParallel && len(assistant.ToolCalls) > 1 {
		if len(cfg.SerialTools) == 0 {
			return executeParallel(ctx, cfg, assistant)
		}
		return executeMixed(ctx, cfg, assistant)
	}
	out := make([]llm.Message, 0, len(assistant.ToolCalls))
	for _, call := range assistant.ToolCalls {
		if err := ctx.Err(); err != nil {
			out = append(out, cancelledToolMessage(call, err))
			continue
		}
		out = append(out, runOne(ctx, cfg, call))
	}
	return out
}

// executeMixed 并行执行非串行调用，串行名（SerialTools）按模型给出的原顺序
// 在调用方 goroutine 逐个执行——互斥且保序（卡片流等有先后依赖的工具不能只
// 靠互斥：并发派发下先抢到锁的不一定是先声明的调用）。
func executeMixed(ctx context.Context, cfg Config, assistant llm.Message) []llm.Message {
	out := make([]llm.Message, len(assistant.ToolCalls))
	var wg sync.WaitGroup
	serialIdx := make([]int, 0, len(assistant.ToolCalls))
	for i, call := range assistant.ToolCalls {
		if cfg.SerialTools[call.Name] {
			serialIdx = append(serialIdx, i)
			continue
		}
		wg.Add(1)
		go func(i int, call llm.ToolCall) {
			defer wg.Done()
			out[i] = runOne(ctx, cfg, call)
		}(i, call)
	}
	for _, i := range serialIdx {
		if err := ctx.Err(); err != nil {
			out[i] = cancelledToolMessage(assistant.ToolCalls[i], err)
			continue
		}
		out[i] = runOne(ctx, cfg, assistant.ToolCalls[i])
	}
	wg.Wait()
	return out
}

func executeParallel(ctx context.Context, cfg Config, assistant llm.Message) []llm.Message {
	out := make([]llm.Message, len(assistant.ToolCalls))
	var wg sync.WaitGroup
	for i, call := range assistant.ToolCalls {
		wg.Add(1)
		go func(i int, call llm.ToolCall) {
			defer wg.Done()
			out[i] = runOne(ctx, cfg, call)
		}(i, call)
	}
	wg.Wait()
	return out
}

func cancelledToolMessage(call llm.ToolCall, err error) llm.Message {
	return llm.Message{
		Role:       llm.RoleTool,
		Name:       call.Name,
		ToolCallID: call.ID,
		Content:    "not executed: " + err.Error(),
		IsError:    true,
	}
}

func rejectIncompleteToolCall(call llm.ToolCall) string {
	if strings.TrimSpace(call.Name) == "" {
		return "工具调用未执行：名称不完整"
	}
	args := strings.TrimSpace(call.Arguments)
	if !json.Valid([]byte(args)) || !strings.HasPrefix(args, "{") {
		return "工具调用未执行：参数不是完整的 JSON 对象"
	}
	return ""
}

func stopDecision(cfg Config, last llm.Message, continuations, toolCalls int) (StopDecision, bool) {
	if cfg.Hooks != nil {
		out := cfg.Hooks.Emit(hooks.Call{
			Event: hooks.Stop, LastText: last.Content, Continuations: continuations, ToolCalls: toolCalls,
		})
		if out.Decision != "" {
			return StopDecision{
				Continue: out.Decision == hooks.Block, Reason: out.Reason,
				Replacement: out.Replacement, Notice: out.Notice,
			}, true
		}
	}
	if cfg.Stop != nil {
		return cfg.Stop(last, StopInfo{Continuations: continuations, ToolCalls: toolCalls}), true
	}
	return StopDecision{}, false
}

func gateTool(ctx context.Context, cfg Config, call llm.ToolCall) (llm.ToolCall, llm.Message, bool) {
	if cfg.Hooks == nil {
		return call, llm.Message{}, false
	}
	out := cfg.Hooks.Emit(hooks.Call{Event: hooks.PreToolUse, Tool: call.Name, Args: call.Arguments})
	switch out.Decision {
	case hooks.Ask:
		// ask 先问宿主；没有裁决端口（或宿主拒绝）时保持旧行为：按拒绝处理。
		cfg.Hooks.Emit(hooks.Call{Event: hooks.PermissionRequest, Tool: call.Name, Args: call.Arguments})
		why := out.Reason
		if cfg.PermissionAsk != nil {
			if verdict := cfg.PermissionAsk(ctx, call); verdict.Allow {
				return call, llm.Message{}, false
			} else if verdict.Reason != "" {
				why = verdict.Reason
			}
		}
		return call, rejectToolMessage(call, why), true
	case hooks.Deny, hooks.Block:
		return call, rejectToolMessage(call, out.Reason), true
	case hooks.Modify:
		if out.Args != "" {
			call.Arguments = out.Args
		}
	}
	return call, llm.Message{}, false
}

func rejectToolMessage(call llm.ToolCall, why string) llm.Message {
	if why == "" {
		why = "工具调用被拒绝"
	}
	return llm.Message{Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: why, IsError: true}
}

func noteToolHook(cfg Config, call llm.ToolCall, msg llm.Message) {
	if cfg.Hooks == nil {
		return
	}
	ev := hooks.PostToolUse
	if msg.IsError {
		ev = hooks.PostToolUseFailure
	}
	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, 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, 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, CallID: call.ID, Text: msg.Content})
		noteToolHook(cfg, call, msg)
		if cfg.AfterTool != nil {
			cfg.AfterTool(call, msg)
		}
		return msg
	} else {
		call = gated
	}
	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, CallID: call.ID, Text: msg.Content})
			noteToolHook(cfg, call, msg)
			if cfg.AfterTool != nil {
				cfg.AfterTool(call, msg)
			}
			return msg
		}
	}
	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, CallID: call.ID, Text: msg.Content})
		return msg
	}
	endSpan := startSpan(cfg, "execute_tool "+call.Name, map[string]string{
		"gen_ai.system":    providerName(cfg),
		"gen_ai.tool.name": call.Name,
	}, false)
	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))
	case cfg.UnknownTool != nil:
		res = cfg.UnknownTool(call)
	case cfg.Tools != nil:
		res = cfg.Tools.Execute(ctx, call.Name, json.RawMessage(call.Arguments))
	}
	msg := llm.Message{
		Role:       llm.RoleTool,
		Name:       call.Name,
		ToolCallID: call.ID,
		Content:    res.Content,
		IsError:    res.IsError,
	}
	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)
	}
	return msg
}

func settle(cfg Config, last llm.Message, messages []llm.Message, usage llm.Usage, notice string) Result {
	emit(cfg, Event{Type: "agent_end", Message: last})
	emit(cfg, Event{Type: "agent_settled", Message: last})
	emitUsage(cfg, usage)
	return Result{Text: last.Content, Messages: messages, StopReason: last.StopReason, Usage: usage, Notice: notice}
}

func prepareRequest(cfg Config, messages []llm.Message, turn int, repeats *repeatState, lengthNudge *bool) []llm.Message {
	request := clearOldResults(cfg, persistedProjection(cfg, messages))
	request = injectMemoryBlock(cfg, request)
	return reminders.Attach(request, turnReminderTexts(cfg, turn, repeats, lengthNudge))
}

// injectMemoryBlock 把记忆召回块插到请求副本的系统段之后（request-scoped：
// 不落盘、不累计；Recall 失败静默跳过——记忆是增强面，不阻断对话）。
func injectMemoryBlock(cfg Config, request []llm.Message) []llm.Message {
	if cfg.Memory == nil {
		return request
	}
	entries, err := cfg.Memory.Recall(context.Background(), memory.RecallQuery{Scope: cfg.MemoryScope})
	if err != nil {
		return request
	}
	block := memory.Block(entries, time.Now())
	if block == "" {
		return request
	}
	// 系统段之后插入（保持历史与尾部不动）。
	i := 0
	for i < len(request) && request[i].Role == llm.RoleSystem {
		i++
	}
	out := make([]llm.Message, 0, len(request)+1)
	out = append(out, request[:i]...)
	out = append(out, llm.Message{Role: llm.RoleSystem, Content: block})
	return append(out, request[i:]...)
}

// persistedProjection 在有会话存储时只采用已经落盘的消息。
// steer 与 follow-up 必须先写入存储，下一轮请求才会看见它们。
func persistedProjection(cfg Config, messages []llm.Message) []llm.Message {
	// JournalOnly：journal 不作权威上下文，每轮请求都用内存态（History 组装）。
	if cfg.Session == nil || cfg.JournalOnly {
		return messages
	}
	loaded, err := cfg.Session.Load()
	if err != nil {
		return messages
	}
	return loaded
}

const unknownSideEffect = "副作用未知，未重试"

// crashInterrupted 是尾部兜底配对的正文（进程死在 persist(assistant) 与
// commitToolIntent 之间的窗口时使用）。
const crashInterrupted = "进程中断，该调用未执行"

// recoverPending 把未配对的意图补齐（safe 双声明重放，否则「副作用未知」
// 配对），返回本次写入的配对消息（JournalOnly 时供恢复摘要使用）。
func recoverPending(ctx context.Context, cfg Config) ([]llm.Message, error) {
	if cfg.Session == nil {
		return nil, nil
	}
	pending, err := cfg.Session.PendingIntents()
	if err != nil {
		return nil, err
	}
	var out []llm.Message
	for _, in := range pending {
		current := tools.ReplayNever
		if cfg.Tools != nil {
			current = cfg.Tools.Replay(in.Name)
		}
		msg := llm.Message{
			Role: llm.RoleTool, Name: in.Name, ToolCallID: in.CallID,
			Content: unknownSideEffect, IsError: true,
		}
		if in.Replay == tools.ReplaySafe && current == tools.ReplaySafe && cfg.Tools != nil {
			res := cfg.Tools.Execute(ctx, in.Name, json.RawMessage(in.Arguments))
			msg.Content = res.Content
			msg.IsError = res.IsError
		}
		if err := persist(cfg, msg); err != nil {
			return nil, err
		}
		out = append(out, msg)
	}
	return out, nil
}

// repairOrphanTail 兜底配对：找到转录里最后一个带 tool_calls 的 assistant，
// 凡没有对应 tool 结果的调用，补一条「进程中断，该调用未执行」的 IsError
// 结果并落盘。与意图日志正交（按 CallID 判重，已配对的跳过；意图路径
// 已处理的执行中死亡不受影响）。历史深处、其后已有新消息的孤立调用不在
// 此处修补（追加只落在尾部，位置语义不破）——那种形态由毒化轮转覆盖。
// repairOrphanTail 返回本次补写的配对消息（JournalOnly 时供恢复摘要使用）。
func repairOrphanTail(cfg Config) ([]llm.Message, error) {
	if cfg.Session == nil {
		return nil, nil
	}
	msgs, err := cfg.Session.Load()
	if err != nil {
		return nil, err
	}
	tail := -1
	for i := len(msgs) - 1; i >= 0; i-- {
		m := msgs[i]
		if m.Role == llm.RoleAssistant && len(m.ToolCalls) > 0 {
			tail = i
			break
		}
		if m.Role == llm.RoleTool {
			continue // 尾部连续 tool 结果属于同一条 assistant，继续上溯
		}
		break
	}
	if tail < 0 {
		return nil, nil
	}
	results := map[string]bool{}
	for _, m := range msgs {
		if m.Role == llm.RoleTool && m.ToolCallID != "" {
			results[m.ToolCallID] = true
		}
	}
	var out []llm.Message
	for _, call := range msgs[tail].ToolCalls {
		if results[call.ID] {
			continue
		}
		pair := llm.Message{
			Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID,
			Content: crashInterrupted, IsError: true,
		}
		if err := persist(cfg, pair); err != nil {
			return nil, err
		}
		results[call.ID] = true
		out = append(out, pair)
	}
	return out, nil
}

// recoveredNoteText 把本次恢复写入的配对摘成一段系统说明（注入上下文用；
// 内容截断防爆，工具名与调用 ID 全列）。
func recoveredNoteText(recovered []llm.Message) string {
	var b strings.Builder
	b.WriteString("上次会话中断的调用已完成恢复（这是系统说明，不是用户发言）：")
	for _, m := range recovered {
		content := m.Content
		if len([]rune(content)) > 120 {
			content = string([]rune(content)[:120]) + "…"
		}
		b.WriteString("\n- ")
		b.WriteString(m.Name)
		b.WriteString("（")
		b.WriteString(m.ToolCallID)
		b.WriteString("）：")
		b.WriteString(content)
	}
	return b.String()
}

func commitToolIntent(cfg Config, call llm.ToolCall) error {
	if cfg.Session == nil {
		return nil
	}
	replay := tools.ReplayNever
	if cfg.Tools != nil {
		replay = cfg.Tools.Replay(call.Name)
	}
	return cfg.Session.AppendToolIntent(session.ToolIntent{
		CallID: call.ID, Name: call.Name, Arguments: call.Arguments, Replay: replay,
	})
}

func itoa(n int) string { return strconv.Itoa(n) }

// traceIDs 是一次 Run 的 span 关联：根 trace + invoke_agent 根 span。
type traceIDs struct {
	traceID string
	rootID  string
}

// newTraceContext 建立一次 Run 的关联（Trace 为 nil 时也给空壳，调用方免判空）。
func newTraceContext() *traceIDs {
	return &traceIDs{traceID: telemetry.NewTraceID(), rootID: telemetry.NewSpanID()}
}

// startSpan 开一个 OTel GenAI span：root=true 时作 invoke_agent 根（无 parent），
// 否则作子 span（parent = 根）。attrs 由调用方持有引用——存续期间可补
// （usage 在结束时随过滤导出）。
func startSpan(cfg Config, name string, attrs map[string]string, root bool) func() {
	if cfg.Trace == nil {
		return func() {}
	}
	if attrs == nil {
		attrs = map[string]string{}
	}
	if cfg.traceCtx != nil {
		attrs[telemetry.AttrTraceID] = cfg.traceCtx.traceID
		if root {
			attrs[telemetry.AttrSpanID] = cfg.traceCtx.rootID
		} else {
			attrs[telemetry.AttrSpanID] = telemetry.NewSpanID()
			attrs[telemetry.AttrParentSpanID] = cfg.traceCtx.rootID
		}
	}
	end := cfg.Trace(name, attrs)
	if end == nil {
		return func() {}
	}
	return end
}

// agentName / providerName 是 OTel 属性的默认值解析。
func agentName(cfg Config) string {
	if strings.TrimSpace(cfg.AgentName) != "" {
		return cfg.AgentName
	}
	return "agent-native"
}

func providerName(cfg Config) string {
	if strings.TrimSpace(cfg.ProviderName) != "" {
		return cfg.ProviderName
	}
	return "openai"
}

func clearOldResults(cfg Config, messages []llm.Message) []llm.Message {
	window := cfg.ContextWindow
	if window <= 0 {
		window = session.DefaultContextWindow
	}
	reserve := cfg.OutputReserve
	if reserve <= 0 {
		reserve = session.DefaultOutputReserve
	}
	buffer := cfg.ContextBuffer
	if buffer <= 0 {
		buffer = session.DefaultContextBuffer
	}
	if !session.NeedsClear(session.EstimateChars(messages), window, reserve, buffer) {
		return messages
	}
	return session.ClearOldToolResultsTo(messages, cfg.KeepToolResults, cfg.ResultsDir)
}

func turnReminderTexts(cfg Config, turn int, repeats *repeatState, lengthNudge *bool) []string {
	var extra []string
	if cfg.Reminders != nil {
		for _, text := range cfg.Reminders(turn) {
			if strings.TrimSpace(text) == "" {
				continue
			}
			extra = append(extra, text)
		}
	}
	if lengthNudge != nil && *lengthNudge {
		extra = append(extra, lengthContinuePrompt)
		*lengthNudge = false
	}
	if cfg.RepeatGuard && repeats != nil {
		due := repeats.streak/3 - repeats.reminders
		room := maxRepeatReminders - repeats.reminders
		if due > room {
			due = room
		}
		for i := 0; i < due; i++ {
			extra = append(extra, repeatReminderText)
			repeats.reminders++
		}
	}
	return extra
}

type repeatState struct {
	sig       string
	streak    int
	reminders int
}

func (s *repeatState) observe(calls []llm.ToolCall) {
	for _, call := range calls {
		sig := call.Name + "\n" + canonicalArgs(call.Arguments)
		if sig == s.sig {
			s.streak++
			continue
		}
		s.sig = sig
		s.streak = 1
	}
}

func canonicalArgs(raw string) string {
	var v any
	if err := json.Unmarshal([]byte(raw), &v); err != nil {
		return raw
	}
	b, err := json.Marshal(v)
	if err != nil {
		return raw
	}
	return string(b)
}

func persist(cfg Config, msg llm.Message) error {
	if cfg.Session == nil {
		return nil
	}
	return cfg.Session.Append(msg)
}

func drain(fn func() []llm.Message) []llm.Message {
	if fn == nil {
		return nil
	}
	return fn()
}

func emitUsage(cfg Config, usage llm.Usage) {
	emit(cfg, Event{
		Type:         "usage",
		PromptTokens: usage.PromptTokens, CompletionTokens: usage.CompletionTokens, TotalTokens: usage.TotalTokens,
		LastPromptTokens: usage.LastPromptTokens, LastCompletionTokens: usage.LastCompletionTokens, LastTotalTokens: usage.LastTotalTokens,
		CacheReadTokens: usage.CacheReadTokens, CacheWriteTokens: usage.CacheWriteTokens,
	})
}

func emit(cfg Config, ev Event) {
	if cfg.OnEvent != nil {
		cfg.OnEvent(ev)
	}
}

// budgetExceeded 返回命中的预算说明；空串表示预算仍充足（或未设预算）。
// 只在「即将消耗更多预算」的节点调用（轮次顶端 / 工具批执行前）。
func budgetExceeded(b *Budget, usage llm.Usage, toolCalls int, now time.Time) string {
	if b == nil {
		return ""
	}
	if !b.Deadline.IsZero() && now.After(b.Deadline) {
		return fmt.Sprintf("budget_exceeded: deadline %s passed", b.Deadline.Format(time.RFC3339))
	}
	if b.MaxPromptTokens > 0 && usage.LastPromptTokens > b.MaxPromptTokens {
		return fmt.Sprintf("budget_exceeded: prompt tokens %d > max %d", usage.LastPromptTokens, b.MaxPromptTokens)
	}
	if b.MaxCompletionTokens > 0 && usage.CompletionTokens > b.MaxCompletionTokens {
		return fmt.Sprintf("budget_exceeded: completion tokens %d > max %d", usage.CompletionTokens, b.MaxCompletionTokens)
	}
	if b.MaxToolCalls > 0 && toolCalls > b.MaxToolCalls {
		return fmt.Sprintf("budget_exceeded: tool calls %d > max %d", toolCalls, b.MaxToolCalls)
	}
	return ""
}

// budgetStop 以预算超限收尾：发事件、留日志、按 StopError 返回（不可重试）。
func budgetStop(cfg Config, last llm.Message, messages []llm.Message, usage llm.Usage, reason string) Result {
	emit(cfg, Event{Type: "budget_exceeded", Text: reason})
	emitUsage(cfg, usage)
	err := errString(reason)
	cfg.logger().Error("runtime budget exceeded", err)
	return Result{Text: last.Content, Messages: messages, StopReason: llm.StopError, Usage: usage, Err: err}
}

type errString string

func (e errString) Error() string { return string(e) }