sundynix-agentix/sundynix-dispatcher/internal/eino/orchestrator.go

// Package eino 封装基于 CloudWeGo Eino 的 Agent 图编排引擎。
package eino

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"log/slog"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/cloudwego/eino/components/model"
	"github.com/cloudwego/eino/schema"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"go.opentelemetry.io/otel/trace"

	"github.com/sundynix/sundynix-dispatcher/internal/dsl"
	"github.com/sundynix/sundynix-dispatcher/internal/harness"
	"github.com/sundynix/sundynix-dispatcher/internal/llm"
	"github.com/sundynix/sundynix-shared/contract"
	"github.com/sundynix/sundynix-shared/otelx"
	"github.com/sundynix/sundynix-shared/prompts"
)

// TokenSink 是 Token 流回流出口（由 NATS bus 实现）。
type TokenSink interface {
	PublishToken(taskID string, token []byte) error
	CompleteStream(taskID string) error
}

// ToolCaller 经 NATS 调起第 5 层 MCP 工具（由 NATS bus 实现）。
type ToolCaller interface {
	CallTool(ctx context.Context, subject string, call *contract.ToolCall) (*contract.ToolResult, error)
}

// StatusSink 回写任务生命周期状态（由 NATS bus 实现；可为 nil → 不回写）。
type StatusSink interface {
	PublishTaskStatus(taskID, status, detail string) error
}

// ApprovalWaiter 阻塞等待审批节点的人工决定（由 NATS bus 实现；可为 nil → 审批节点自动放行）。
type ApprovalWaiter interface {
	WaitApproval(ctx context.Context, taskID string, timeout time.Duration) (*contract.ApprovalDecision, error)
}

// EvalSink 回写自动化评测结果供网关落库（由 NATS bus 实现；可为 nil → 仅打日志不落库）。
type EvalSink interface {
	PublishEval(ev *contract.EvalEvent) error
}

// UsageSink 回写任务 token 用量供网关累计/计费（由 NATS bus 实现；可为 nil → 不回写）。
type UsageSink interface {
	PublishUsage(ev *contract.UsageEvent) error
}

// errBudget 是单任务 token 预算触顶时的哨兵错误：任务以 failed 收尾并附明确原因（防失控成本）。
var errBudget = errors.New("token 预算超限，已中止")

// errRejected 是审批节点拒绝（或超时）时图执行返回的哨兵错误：它是合法终态而非故障，
// Handle 据此判 rejected 并优雅收尾（不计熔断失败）。
var errRejected = errors.New("approval rejected")

// errInterrupted 是 HITL 审批中断（compose checkpoint）时图执行返回的哨兵错误：任务停在
// waiting、checkpoint 已落、goroutine 即可释放；非故障非终态，等人工决定到达后 resume 续跑。
var errInterrupted = errors.New("approval interrupted (checkpointed)")

// LLM 是编排所需的语言模型能力（生产由 *llm.Pool 实现）。抽成接口便于测试注入假模型。
type LLM interface {
	Ready() bool
	ChatStream(ctx context.Context, msgs []llm.ChatMessage, onToken func(string), onReasoning func(string)) error
	StreamText(ctx context.Context, text string, onToken func([]byte)) error
	Chat(ctx context.Context, msgs []llm.ChatMessage) (string, error)
	// ToolCallingModel 返回支持函数调用的模型（ReAct agent 用）；不支持则返回 nil。
	ToolCallingModel() model.ToolCallingChatModel
	// ChatModel 返回 Eino ChatModel 组件（compose.Graph 编排用）；未就绪则 nil。
	ChatModel() model.BaseChatModel
}

// 工具调用超时；超时即降级（不带工具上下文继续推理）。
const toolCallTimeout = 3 * time.Second

// taskExecTimeout 是单个任务整体执行上限；超时即判 timeout（状态机），避免无限期"运行中"。
// 含 HITL 审批等待预算（approvalTimeout）+ 常规图执行；须 < bus 消费者 AckWait(15min) 以免重投。
const taskExecTimeout = 10 * time.Minute

// approvalTimeout 是单个审批节点等待人工决定的上限；超时安全默认拒绝（fail-safe）。
const approvalTimeout = 5 * time.Minute

// specialistTimeout 是多智能体单个专家派发的执行上限；超时即作为"观察"跳过该专家
// （不中断整个协调），避免单专家卡死拖垮 lead 无限等待。专家可能多轮 react+工具调用，故给较宽。
const specialistTimeout = 3 * time.Minute

// Orchestrator 把每个 DSL 任务动态编译为 Eino 图并执行（记忆召回 → 工具节点 → 注入 → 流式）。
type Orchestrator struct {
	pool        LLM
	breaker     *harness.CircuitBreaker
	eval        *harness.Evaluator
	sink        TokenSink
	tools       ToolCaller
	exec        ExecSink
	status      StatusSink          // 任务生命周期状态回写（可为 nil）
	approval    ApprovalWaiter      // HITL 审批等待（可为 nil → 审批节点自动放行）
	evalSink    EvalSink            // 评测结果回写落库（可为 nil → 仅打日志）
	guard       *harness.Classifier // 输入护栏 Tier2：对网关标记的灰区任务做 LLM 裁决（可为 nil → 不做）
	usageSink   UsageSink           // token 用量回写（可为 nil → 不回写）
	checkpoints CheckpointKV        // HITL 持久化中断的 checkpoint 后端（可为 nil → 审批走阻塞模型）

	turnMu sync.Mutex     // 保护 turns（攒批计数，多任务 goroutine 共享）
	turns  map[string]int // sessionID → 累计轮次，用于每 N 轮触发 consolidate
}

// NewOrchestrator 持有依赖；图按任务的 DSL 在 Handle 内动态编译。
// exec 为执行可视化事件出口（可为 nil）；eval 为自动化评测（可为 nil）；status 为状态回写（可为 nil）；
// approval 为 HITL 审批等待（可为 nil）；evalSink 为评测结果落库出口（可为 nil）。
func NewOrchestrator(pool LLM, breaker *harness.CircuitBreaker, eval *harness.Evaluator, sink TokenSink, tools ToolCaller, exec ExecSink, status StatusSink, approval ApprovalWaiter, evalSink EvalSink) (*Orchestrator, error) {
	return &Orchestrator{pool: pool, breaker: breaker, eval: eval, sink: sink, tools: tools, exec: exec, status: status, approval: approval, evalSink: evalSink}, nil
}

// SetGuardian 注入输入护栏 Tier2 的 LLM 分类器（可选；不注入则灰区任务直接放行执行）。
func (o *Orchestrator) SetGuardian(c *harness.Classifier) { o.guard = c }

// SetUsageSink 注入 token 用量回写出口（可选；不注入则不上报用量）。
func (o *Orchestrator) SetUsageSink(s UsageSink) { o.usageSink = s }

// SetCheckpoints 注入 HITL 持久化中断的 checkpoint 后端（可选）。注入后审批节点改走
// compose.Interrupt 中断模型（落 checkpoint、释放 goroutine、抗重启）；不注入则维持阻塞模型。
func (o *Orchestrator) SetCheckpoints(kv CheckpointKV) { o.checkpoints = kv }

// taskBudget 取本任务的 token 预算上限：优先 Meta（网关按用户/套餐下发），否则 env TASK_TOKEN_BUDGET（默认 20 万）。
func (o *Orchestrator) taskBudget(t *contract.Task) int {
	switch n := t.Meta[contract.MetaTokenBudget].(type) {
	case float64:
		if n > 0 {
			return int(n)
		}
	case int:
		if n > 0 {
			return n
		}
	}
	return envInt("TASK_TOKEN_BUDGET", 200000)
}

// emitUsage 任务收尾回写本轮 token 用量（用量为 0 或无出口则跳过）。
func (o *Orchestrator) emitUsage(t *contract.Task, b *harness.Budget) {
	if o.usageSink == nil {
		return
	}
	p, c, total := b.Snapshot()
	if total == 0 {
		return
	}
	uid, _ := t.Meta[contract.MetaUserID].(string)
	tid, _ := t.Meta[contract.MetaTenantID].(string)
	if err := o.usageSink.PublishUsage(&contract.UsageEvent{
		TaskID: t.ID, UserID: uid, TenantID: tid, PromptTok: p, CompTok: c, TotalTok: total,
		Exceeded: b.Exceeded(), TS: time.Now().UnixMilli(),
	}); err != nil {
		log.Printf("[usage] 回写用量失败 task=%s: %v", t.ID, err)
	}
}

// envInt 读正整数环境变量，缺省回退 def。
func envInt(key string, def int) int {
	if v := os.Getenv(key); v != "" {
		if n, err := strconv.Atoi(v); err == nil && n > 0 {
			return n
		}
	}
	return def
}

// setStatus 回写一次任务状态流转（status 为 nil 时静默跳过）。
func (o *Orchestrator) setStatus(taskID, status, detail string) {
	if o.status == nil {
		return
	}
	if err := o.status.PublishTaskStatus(taskID, status, detail); err != nil {
		log.Printf("[eino] 回写任务状态 %s=%s 失败: %v", taskID, status, err)
	}
}

// finishStatus 据收尾错误把任务置为 done / timeout / failed。
func (o *Orchestrator) finishStatus(taskID string, err error) {
	switch {
	case err == nil:
		o.setStatus(taskID, contract.TaskDone, "")
	case errors.Is(err, context.DeadlineExceeded):
		o.setStatus(taskID, contract.TaskTimeout, "执行超时")
	default:
		o.setStatus(taskID, contract.TaskFailed, truncate(err.Error(), 200))
	}
}

// Handle 消费一个任务：按 DSL 编译 Eino 图并执行，把 Token 流回流到 sundynix.streams.<id>。
func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error {
	// 护栏：丢弃空任务（无 id），避免误投/历史脏数据被当真任务处理并触发状态回写放大。
	if t.ID == "" {
		log.Printf("[eino] 跳过空任务（无 id）")
		return nil
	}

	// 链路根（dispatcher 侧）：续上 gateway 经 NATS 传来的 trace，覆盖整个图执行。
	ctx, span := otelx.Tracer().Start(ctx, "task.execute",
		trace.WithAttributes(attribute.String("sundynix.task_id", t.ID)))
	defer span.End()

	tr := o.tracer(t.ID)
	defer tr.done()

	// 熔断开启：快速拒绝，但要让客户端解阻（回流提示 + 收尾流），不静默丢弃。
	if !o.breaker.Allow() {
		log.Printf("[eino] 熔断开启，拒绝任务 %s", t.ID)
		tr.info("task", "system", "服务熔断", "后端连续失败，暂时拒绝新任务，请稍后重试")
		_ = o.sink.PublishToken(t.ID, []byte("⚠️ 服务繁忙（已触发熔断保护），请稍后重试。"))
		_ = o.sink.CompleteStream(t.ID)
		o.setStatus(t.ID, contract.TaskFailed, "服务熔断")
		return nil
	}

	// 输入护栏 Tier2：仅对网关 Tier1 标记的「灰区」任务做 LLM 越狱裁决（明确干净/恶意的已在网关处理，
	// 不付 LLM 成本）。命中即拒绝执行（合法终态，非故障，不计熔断）。
	if o.guard != nil {
		if safety, _ := t.Meta[contract.MetaSafetyCheck].(bool); safety {
			query := dsl.Compile(t.Graph).Query
			gctx, gcancel := context.WithTimeout(ctx, 10*time.Second)
			block, sev, reason := o.guard.Classify(gctx, query)
			gcancel()
			if block {
				slog.WarnContext(ctx, "input guardrail blocked", "task_id", t.ID, "severity", sev, "reason", reason)
				tr.info("guardrail", "system", "输入护栏拦截", reason)
				_ = o.sink.PublishToken(t.ID, []byte("⚠️ 输入被安全护栏拦截："+reason))
				_ = o.sink.CompleteStream(t.ID)
				o.breaker.Report(true) // 安全拦截是策略决策，非后端故障
				o.setStatus(t.ID, contract.TaskRejected, truncate("输入护栏："+reason, 200))
				return nil
			}
			if sev > 0 {
				slog.InfoContext(ctx, "input guardrail suspect passed", "task_id", t.ID, "severity", sev)
			}
		}
	}

	// 任务状态机：进入执行 → running；整体加超时上限，超时判 timeout（杜绝无限期"运行中"）。
	o.setStatus(t.ID, contract.TaskRunning, "")
	tctx, cancel := context.WithTimeout(ctx, taskExecTimeout)
	defer cancel()

	// 成本护栏：单任务 token 预算挂到 ctx，沿图执行各 LLM 节点计量+封顶；收尾回写用量（计费/日预算）。
	budget := harness.NewBudget(o.taskBudget(t))
	tctx = harness.WithBudget(tctx, budget)
	defer o.emitUsage(t, budget)

	// 报告生成走专用多步编排（规划→分章并行检索撰写→汇聚→渲染 Word），而非通用对话图。
	if intent, _ := t.Meta[contract.MetaIntent].(string); intent == contract.IntentReport {
		err := o.handleReport(tctx, t, tr)
		o.finishStatus(t.ID, err)
		return err
	}
	// ctx 携带 task.execute span → 这些任务生命周期日志自动带 trace_id，可与 Jaeger 链路互跳。
	slog.InfoContext(ctx, "task received", "task_id", t.ID, "graph_bytes", len(t.Graph))
	tr.info("task", "system", "任务受理", fmt.Sprintf("DSL %d 字节，按图执行", len(t.Graph)))

	// 按 DSL 图执行：编译为 Eino compose.Graph 并跑（唯一编排引擎）；agent 节点流式回流 token。
	answer, refs, err := o.executeGraph(tctx, t, tr)
	if errors.Is(err, errInterrupted) {
		// HITL 持久化中断：checkpoint 已落、任务停在 waiting（状态在审批节点内已置）。
		// 释放 goroutine 但不收尾——不 CompleteStream、不评测、不判 done、不计熔断失败；
		// token 流与 exec 流都保持打开（suspended），resume 时续录，否则复盘里审批节点永停在等待中。
		tr.suspended = true
		slog.InfoContext(ctx, "task interrupted for approval (checkpointed)", "task_id", t.ID)
		o.breaker.Report(true) // 中断是正常暂停，非后端故障
		return nil
	}
	if errors.Is(err, errRejected) {
		// HITL 拒绝：合法终态，非故障。收尾流 + 置 rejected，不计熔断、不重投。
		slog.InfoContext(ctx, "task rejected by approval", "task_id", t.ID)
		if answer != "" {
			_ = o.sink.PublishToken(t.ID, []byte(answer))
		}
		_ = o.sink.CompleteStream(t.ID)
		o.breaker.Report(true) // 拒绝是人为决策，不算后端失败
		o.setStatus(t.ID, contract.TaskRejected, truncate(answer, 120))
		return nil
	}
	if errors.Is(err, errBudget) {
		// token 预算触顶：策略性中止，非后端故障。收尾流 + 置 failed（附明确原因），不计熔断、不重投。
		_, _, total := budget.Snapshot()
		slog.WarnContext(ctx, "task aborted: token budget exceeded", "task_id", t.ID, "tokens", total)
		if answer != "" {
			_ = o.sink.PublishToken(t.ID, []byte(answer))
		}
		_ = o.sink.PublishToken(t.ID, []byte("\n\n⚠️ 已达单任务 token 预算上限，自动中止。"))
		_ = o.sink.CompleteStream(t.ID)
		o.breaker.Report(true)
		o.setStatus(t.ID, contract.TaskFailed, fmt.Sprintf("token 预算超限（已用约 %d）", total))
		return nil
	}
	if err != nil {
		span.RecordError(err)
		span.SetStatus(codes.Error, err.Error())
		slog.ErrorContext(ctx, "task graph error", "task_id", t.ID, "err", err.Error())
		_ = o.sink.CompleteStream(t.ID)
		o.breaker.Report(false)
		o.finishStatus(t.ID, err)
		return err
	}

	if cerr := o.sink.CompleteStream(t.ID); cerr != nil {
		log.Printf("[eino] complete stream failed: %v", cerr)
	}
	slog.InfoContext(ctx, "task done", "task_id", t.ID, "answer_runes", len([]rune(answer)))
	o.breaker.Report(true)
	o.finishStatus(t.ID, nil)

	// 写回阶段（离热路径，单 goroutine 串起评测与落历史）：
	// 先评测（低分则自动纠偏重生成、取更优者），再把最终采纳的答案落会话历史 —— 保证
	// 多轮上下文用的是纠偏后的好答案，而非被判低分的初版。
	go func() {
		query := dsl.Compile(t.Graph).Query
		final := o.evaluate(t, query, answer, refs)
		o.memorize(t, final)
	}()
	return nil
}

// maxRefineRounds 是低分自动纠偏的最大重生成轮数（防成本失控/死循环）。
// 每轮 = 1 次重写 + 1 次重评；poor 本就稀少，1 轮足以覆盖大多数偶发劣化。
const maxRefineRounds = 1

// evaluate 异步对一次输出做自动化评测并记录评分（off 热路径，不影响响应）。
// 低分（poor）时触发"评语驱动"的自动纠偏：重生成更优答案、重评、取更优者（恒温器闭环）。
// sources 为本轮检索来源：非空时额外评忠实度（幻觉检测）。
// 返回最终采纳的答案（纠偏成功则为修订版，否则为原文）——供调用方落会话历史。
func (o *Orchestrator) evaluate(t *contract.Task, input, output string, sources []string) string {
	if o.eval == nil {
		return output
	}
	ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
	defer cancel()
	r := o.eval.Score(ctx, input, output, sources)
	level := evalLevel(r)
	log.Printf("[eval] task %s 综合 %.2f（规则 %.2f / LLM %.2f / 忠实 %.2f，来源 %d）level=%s flags=%v %s",
		t.ID, r.Overall, r.Rule, r.LLM, r.Faithful, len(sources), level, r.Flags, r.Reason)

	// 低分自动纠偏：把"问题 + 初版回答 + 评审短板"交给模型重写，重评后仅采纳更优者（不退步）。
	corrected := false
	for round := 0; level == contract.EvalPoor && round < maxRefineRounds && o.canRefine(); round++ {
		newOut, newR, ok := o.refine(ctx, t, input, output, sources, r)
		if !ok {
			break // 重写失败或未改好：保留当前最优，停止
		}
		prevOverall := r.Overall
		output, r, level, corrected = newOut, newR, evalLevel(newR), true
		log.Printf("[eval] task %s 自动纠偏第 %d 轮采纳：综合 %.2f→%.2f level=%s", t.ID, round+1, prevOverall, r.Overall, level)
	}

	if level == contract.EvalPoor { // 纠偏后仍低质量：告警（可观测 + 后续可挂报警）
		slog.Warn("eval poor quality", "task_id", t.ID, "overall", r.Overall, "faithful", r.Faithful, "flags", r.Flags, "corrected", corrected)
	}
	// 闭环：评测结果（纠偏后终值）落库（供 UI 查询 / 质量趋势 / 门控）。
	if o.evalSink != nil {
		if err := o.evalSink.PublishEval(&contract.EvalEvent{
			TaskID: t.ID, Overall: r.Overall, Rule: r.Rule, LLM: r.LLM, Faithful: r.Faithful,
			Level: level, Flags: r.Flags, Reason: r.Reason, Sources: len(sources), Corrected: corrected,
			TS: time.Now().UnixMilli(),
		}); err != nil {
			log.Printf("[eval] 回写评测结果失败 task=%s: %v", t.ID, err)
		}
	}
	return output
}

// canRefine 报告是否具备自动纠偏条件：模型就绪且熔断未开（重写要再打一次模型，避雪上加霜）。
func (o *Orchestrator) canRefine() bool {
	return o.pool != nil && o.pool.Ready() && (o.breaker == nil || o.breaker.Allow())
}

// refine 对低分输出做一次"评语驱动"的重生成：把原问题 + 初版回答 + 评测短板（flags/评语）
// 交给模型重写，再重新评分。仅当新分严格更高时采纳（ok=true）；否则保留原文（不退步）。
// 有检索来源时把来源一并喂回，要求严格基于来源作答（同时压低幻觉）。
func (o *Orchestrator) refine(ctx context.Context, t *contract.Task, input, output string, sources []string, prev harness.Result) (string, harness.Result, bool) {
	tr := o.tracer(t.ID)
	end := tr.span("refine", "system", "低分自动纠偏")

	critique := strings.Join(prev.Flags, "；")
	if prev.Reason != "" {
		if critique != "" {
			critique += "；"
		}
		critique += "评语：" + prev.Reason
	}
	if critique == "" {
		critique = "整体质量不足（相关性/准确性/完整性欠佳）"
	}
	sys := prompts.Get(prompts.EvalRefine)
	srcBlock := ""
	if len(sources) > 0 {
		srcBlock = "\n\n【可参考的检索资料（严格基于它作答，不要编造资料外的内容）】\n" +
			truncate(strings.Join(sources, "\n---\n"), 3000)
	}
	user := fmt.Sprintf("【用户问题】%s\n\n【初版回答（质量不足）】%s\n\n【评审指出的问题】%s%s\n\n请直接输出修订后的回答正文，不要解释你做了哪些修改。",
		truncate(input, 1000), truncate(output, 2000), critique, srcBlock)

	revised, err := o.pool.Chat(ctx, []llm.ChatMessage{
		{Role: "system", Content: sys},
		{Role: "user", Content: user},
	})
	if err != nil || strings.TrimSpace(revised) == "" {
		end("重写失败", errOrEmpty(err))
		return output, prev, false
	}
	newR := o.eval.Score(ctx, input, revised, sources)
	if newR.Overall <= prev.Overall {
		end(fmt.Sprintf("重写未提升（%.2f≤%.2f），保留原版", newR.Overall, prev.Overall), nil)
		return output, prev, false
	}
	end(fmt.Sprintf("已采纳修订：综合 %.2f→%.2f", prev.Overall, newR.Overall), nil)
	return revised, newR, true
}

// errOrEmpty 把空输出也表达成一个错误，供 exec error 事件着色。
func errOrEmpty(err error) error {
	if err != nil {
		return err
	}
	return errors.New("空输出")
}

// evalLevel 据综合分 + 忠实度把评测分级（闭环门控/告警用）。
func evalLevel(r harness.Result) string {
	switch {
	case r.Overall < 0.5:
		return contract.EvalPoor
	case r.Overall < 0.75 || (r.Faithful > 0 && r.Faithful < 0.6):
		return contract.EvalWarn
	default:
		return contract.EvalOK
	}
}

// fetchMemory 经 MCP memory_get 工具召回用户常驻画像。
// 工具不可用/超时/无 user_id 时返回空串，降级为无记忆推理（不阻断主流程）。
func (o *Orchestrator) fetchMemory(ctx context.Context, userID, _ string) string {
	if o.tools == nil || userID == "" {
		return ""
	}
	cctx, cancel := context.WithTimeout(ctx, toolCallTimeout)
	defer cancel()
	res, err := o.tools.CallTool(cctx, contract.ToolSubjectGo("memory_get"), &contract.ToolCall{
		Tool: "memory_get",
		Args: map[string]any{"user_id": userID},
	})
	if err != nil {
		log.Printf("[eino] memory_get unavailable for %s, degrade: %v", userID, err)
		return ""
	}
	if !res.OK {
		log.Printf("[eino] memory_get error for %s: %s", userID, res.Error)
		return ""
	}
	log.Printf("[eino] memory_get ok for %s: %s", userID, res.Content)
	return res.Content
}

// fetchHistory 经 MCP history_get 工具召回会话短期多轮历史，转为 Eino 消息。
// 工具不可用/无 session 时返回空，降级为无历史（不阻断主流程）。
func (o *Orchestrator) fetchHistory(ctx context.Context, sessionID string) []*schema.Message {
	if o.tools == nil || sessionID == "" {
		return nil
	}
	cctx, cancel := context.WithTimeout(ctx, toolCallTimeout)
	defer cancel()
	res, err := o.tools.CallTool(cctx, contract.ToolSubjectGo("history_get"), &contract.ToolCall{
		Tool: "history_get",
		Args: map[string]any{"session_id": sessionID},
	})
	if err != nil || res == nil || !res.OK || res.Content == "" {
		return nil
	}
	var turns []struct {
		Role    string `json:"role"`
		Content string `json:"content"`
	}
	if json.Unmarshal([]byte(res.Content), &turns) != nil {
		return nil
	}
	msgs := make([]*schema.Message, 0, len(turns))
	for _, tn := range turns {
		if tn.Role == "assistant" {
			msgs = append(msgs, schema.AssistantMessage(tn.Content, nil))
		} else {
			msgs = append(msgs, schema.UserMessage(tn.Content))
		}
	}
	if len(msgs) > 0 {
		log.Printf("[eino] history_get ok for %s: %d 条历史", sessionID, len(msgs))
	}
	return msgs
}

// consolidateEveryTurns 控制记忆对账的攒批节奏：每 N 轮 consolidate 一次（不逐轮，省成本）。
const consolidateEveryTurns = 3

// memorize 写回阶段（异步、离热路径）：落短期历史；每 N 轮做一次记忆对账（consolidate）。
func (o *Orchestrator) memorize(t *contract.Task, answer string) {
	// 空答复（LLM 失败/降级/拒绝）不落历史：空 assistant 消息会被 LLM API 拒绝（400），
	// 一旦写入会毒化该会话后续所有请求。失败的一轮干脆不留痕。
	if strings.TrimSpace(answer) == "" {
		return
	}
	uid, _ := t.Meta[contract.MetaUserID].(string)
	sid, _ := t.Meta[contract.MetaSessionID].(string)
	if sid != "" && o.tools != nil {
		o.appendHistory(sid, "user", dsl.Compile(t.Graph).Query) // 落真实用户输入，而非 DSL 原文
		o.appendHistory(sid, "assistant", answer)
		log.Printf("[eino] (writeback) task %s 已落会话历史 session=%s", t.ID, sid)
	}
	if uid == "" || sid == "" || o.tools == nil {
		return
	}
	// 攒批：累计轮次，每 N 轮才把近期对话与已有画像交给 LLM 对账一次。
	o.turnMu.Lock()
	if o.turns == nil {
		o.turns = map[string]int{}
	}
	o.turns[sid]++
	n := o.turns[sid]
	o.turnMu.Unlock()
	if n%consolidateEveryTurns == 0 {
		ctx := context.Background()
		o.consolidateMemory(ctx, uid, o.fetchHistory(ctx, sid))
	}
}

func (o *Orchestrator) appendHistory(sessionID, role, content string) {
	cctx, cancel := context.WithTimeout(context.Background(), toolCallTimeout)
	defer cancel()
	if _, err := o.tools.CallTool(cctx, contract.ToolSubjectGo("history_append"), &contract.ToolCall{
		Tool: "history_append",
		Args: map[string]any{"session_id": sessionID, "role": role, "content": content},
	}); err != nil {
		log.Printf("[eino] history_append failed: %v", err)
	}
}