eightbf-analysis-go/internal/worker/live_runner.go

package worker

import (
	"fmt"
	"strings"
	"sync"
	"time"
	"unicode/utf8"

	"git.gozeros.com/wsk/eightbf-analysis-go/internal/deerflow"
	"git.gozeros.com/wsk/eightbf-analysis-go/internal/domain"
	"git.gozeros.com/wsk/eightbf-analysis-go/internal/logging"
	"git.gozeros.com/wsk/eightbf-analysis-go/internal/repo"
)

// LiveChainRunner 真模型按链条顺序执行（被 LiveWorker 异步调用）。
type LiveChainRunner struct {
	DF    *deerflow.Client
	Repo  *repo.JobRepo
	Coord string

	mu        sync.Mutex
	lastFlush time.Time
}

// Run 跑完一场真模型会商。调用方已经把状态改成 running，这里不再抢队列。
//
// 顺序：席位不够直接失败 → 校验模型（禁止网关静默换模）→ 没有 thread 就 init →
// 总控派活、席位发言，人工干预开着则每席结束后暂停 → 收口写成 done。
// 边界：任一步 panic 记失败，不当成功；取消看库里的 cancel_flag，不靠内存标记。
func (r *LiveChainRunner) Run(job *domain.AnalysisJob) {
	jobID := job.ID
	defer func() {
		if rec := recover(); rec != nil {
			logging.L.Errorw("真模型会商 panic", logging.KeyJobID, jobID, "err", rec)
			r.fail(job, fmt.Sprint(rec))
		}
	}()

	agents := job.Agents
	if len(agents) < 2 {
		r.fail(job, "有效席位不足 2 个，无法启动真模型会商")
		return
	}
	var agentNames []string
	for _, a := range agents {
		id := asStr(a["id"])
		if id != "" {
			agentNames = append(agentNames, id)
		}
	}
	if len(agentNames) < 2 {
		r.fail(job, "席位 agentId 无效（需要 DeerFlow 的 hyphen-case id）")
		return
	}

	plan := ensurePlan(job)
	r.pullModelFromDB(job)
	if err := r.DF.AssertModelAvailable(job.Model); err != nil {
		r.fail(job, err.Error())
		return
	}
	job.Status = domain.StatusRunning
	job.Phase = "running"
	r.flush(job)

	if r.cancelled(jobID) {
		r.markCancelled(job)
		return
	}

	threadIDs := loadThreadIDs(job)
	if len(threadIDs) == 0 {
		putProgress(job, 2, "正在初始化 DeerFlow 席位线程…")
		r.flush(job)
		r.pullModelFromDB(job)
		if strings.TrimSpace(job.DraftID) == "" {
			r.fail(job, "缺少会商标识（draftId）。请从页面重新「开始研讨」以先落盘草稿。")
			return
		}
		roomSessionID := asStr(plan["roomSessionId"])
		if roomSessionID == "" {
			roomSessionID = "rs_" + jobID
			plan["roomSessionId"] = roomSessionID
			job.Plan = plan
		}
		dfTaskID := strings.TrimSpace(job.DraftKey)
		if dfTaskID == "" {
			dfTaskID = job.TaskID
		}
		putProgress(job, 3, "正在占用会商房间…")
		r.flush(job)
		if err := r.DF.AcquireRoundtableRoom(job.DraftID, dfTaskID, roomSessionID); err != nil {
			r.fail(job, err.Error())
			return
		}
		tids, err := r.DF.InitMultiAgent(agentNames, r.Coord, job.Model, deerflow.InitMultiAgentOpts{
			DraftID:       job.DraftID,
			TaskID:        dfTaskID,
			RoomSessionID: roomSessionID,
		})
		if err != nil {
			r.fail(job, err.Error())
			return
		}
		threadIDs = tids
		store := map[string]interface{}{}
		for k, v := range tids {
			store[k] = v
		}
		job.ThreadIDs = store
		putProgress(job, 8, "席位线程已就绪，开始链条研讨")
		r.flush(job)
	} else {
		pct := toInt(job.Progress["percent"], 8)
		if pct < 8 {
			pct = 8
		}
		putProgress(job, pct, "从检查点继续会商")
		r.flush(job)
	}

	seed := asStr(plan["seedMessage"])
	if seed == "" {
		seed = BuildSeed(job.Progress)
		plan["seedMessage"] = seed
	}
	if inj := consumePendingInject(plan); inj != "" {
		seed = seed + "\n\n" + inj
		plan["seedMessage"] = seed
	}

	total := len(agentNames)
	start := toInt(plan["seatIndex"], 0)
	if start < 0 {
		start = 0
	}
	seatDeliveries := loadDeliveries(job)
	lastLeader := ""

	for i := start; i < total; i++ {
		if r.cancelled(jobID) {
			r.markCancelled(job)
			return
		}
		r.pullManualFromDB(job, plan)
		r.pullInjectFromDB(job, plan)
		r.pullModelFromDB(job)
		model := job.Model
		if mid := consumePendingInject(plan); mid != "" {
			seed = seed + "\n\n" + mid
			plan["seedMessage"] = seed
		}
		plan["seatIndex"] = i
		job.Plan = plan

		seat := agents[i]
		seatID := asStr(seat["id"])
		seatName := firstNonEmpty(asStr(seat["name"]), seatID)
		job.ActiveAgentID = r.Coord
		job.Cycle = i + 1
		applyAgentSpeaking(agents, seatID, false)
		job.Agents = agents

		leaderBubbleID := openBubble(job, "leader", "总控协调", r.Coord, "链条派活中")
		r.flush(job)
		leaderRes, err := r.DF.RunStream("leader", threadIDs, r.Coord, "总控协调",
			BuildLeaderTask(seed, i, total, seatID, seatName, i == 0),
			model, false,
			r.deltaHandler(job, leaderBubbleID),
			func() bool { return r.cancelled(jobID) },
		)
		if err != nil {
			finishBubble(job, leaderBubbleID, err.Error(), "error")
			r.fail(job, buildModelHint("总控派活失败", model, err.Error()))
			return
		}
		if r.cancelled(jobID) || leaderRes.Aborted {
			finishBubble(job, leaderBubbleID, leaderRes.Content, "stopped")
			r.markCancelled(job)
			return
		}
		if leaderRes.Clarification {
			finishBubble(job, leaderBubbleID, leaderRes.Content, "clarification")
			detail := leaderRes.ClarificationDetail
			if detail == "" {
				detail = leaderRes.Content
			}
			r.pauseForInput(job, plan, i, "clarification", detail)
			return
		}
		leaderBad := leaderRes.InvalidDelivery || strings.TrimSpace(leaderRes.Content) == "" || looksLikeModelFailure(leaderRes.Content)
		if leaderBad {
			finishBubble(job, leaderBubbleID, leaderRes.Content, "error")
			r.fail(job, buildModelHint("总控派活失败", model, leaderRes.Content))
			return
		}
		finishBubble(job, leaderBubbleID, leaderRes.Content, "done")
		lastLeader = leaderRes.Content
		r.flush(job)

		task := PickChainTask(leaderRes.Dispatched, leaderRes.Content, seatID, seatName)
		job.ActiveAgentID = seatID
		applyAgentSpeaking(agents, seatID, true)
		job.Agents = agents
		seatBubbleID := openBubble(job, "special", seatName, seatID, "席位输出中…")
		r.flush(job)
		seatRes, err := r.DF.RunStream("special", threadIDs, seatID, seatName, task, model, false,
			r.deltaHandler(job, seatBubbleID),
			func() bool { return r.cancelled(jobID) },
		)
		if err != nil {
			finishBubble(job, seatBubbleID, err.Error(), "error")
			r.fail(job, buildModelHint("席位「"+seatName+"」输出失败", model, err.Error()))
			return
		}
		if r.cancelled(jobID) || seatRes.Aborted {
			finishBubble(job, seatBubbleID, seatRes.Content, "stopped")
			r.markCancelled(job)
			return
		}
		bad := seatRes.InvalidDelivery || strings.TrimSpace(seatRes.Content) == "" || looksLikeModelFailure(seatRes.Content)
		status := "done"
		if bad {
			status = "error"
		}
		finishBubble(job, seatBubbleID, seatRes.Content, status)
		if bad {
			r.fail(job, buildModelHint("席位「"+seatName+"」输出失败", model, seatRes.Content))
			return
		}
		seatDeliveries = append(seatDeliveries, map[string]interface{}{
			"sender":  seatName,
			"agentId": seatID,
			"content": seatRes.Content,
		})
		saveDeliveries(job, seatDeliveries)
		pct := (i + 1) * 85 / total
		if pct > 90 {
			pct = 90
		}
		job.ConsensusPercentage = pct
		putProgress(job, pct, fmt.Sprintf("已完成 %d/%d 席", i+1, total))
		plan["seatIndex"] = i + 1
		job.Plan = plan
		r.flush(job)

		r.pullManualFromDB(job, plan)
		if readManual(plan) && i+1 < total {
			r.pauseForInput(job, plan, i+1, "manual_continue",
				fmt.Sprintf("人工干预：第 %d/%d 席已完成，请确认后继续下一席。", i+1, total))
			return
		}
	}

	if r.cancelled(jobID) {
		r.markCancelled(job)
		return
	}
	job.ActiveAgentID = r.Coord
	r.pullModelFromDB(job)
	closeModel := job.Model
	closeID := openBubble(job, "leader", "总控协调", r.Coord, "全案盖印")
	r.flush(job)
	closeRes, err := r.DF.RunStream("leader", threadIDs, r.Coord, "总控协调", Closeout, closeModel, true,
		r.deltaHandler(job, closeID),
		func() bool { return r.cancelled(jobID) },
	)
	if err != nil {
		finishBubble(job, closeID, err.Error(), "error")
		r.fail(job, buildModelHint("全案收口失败", closeModel, err.Error()))
		return
	}
	if r.cancelled(jobID) || closeRes.Aborted {
		finishBubble(job, closeID, closeRes.Content, "stopped")
		r.markCancelled(job)
		return
	}
	if closeRes.InvalidDelivery || looksLikeModelFailure(closeRes.Content) ||
		(strings.TrimSpace(closeRes.Content) == "" && strings.TrimSpace(lastLeader) == "") {
		finishBubble(job, closeID, closeRes.Content, "error")
		r.fail(job, buildModelHint("全案收口失败", closeModel, closeRes.Content))
		return
	}
	if len(seatDeliveries) < total {
		finishBubble(job, closeID, closeRes.Content, "error")
		r.fail(job, fmt.Sprintf("会商未完整交付（成功 %d/%d 席），请检查模型后重新开始。", len(seatDeliveries), total))
		return
	}
	if dialoguesHaveError(job) {
		finishBubble(job, closeID, closeRes.Content, "error")
		r.fail(job, "会商过程中已有失败发言，不能进入收口/下一阶段。请换可用模型后重新开始。")
		return
	}
	summary := closeRes.Content
	if strings.TrimSpace(summary) == "" {
		summary = lastLeader
	}
	finishBubble(job, closeID, summary, "done")
	for _, a := range agents {
		a["status"] = "done"
	}
	job.Agents = agents
	job.ActiveAgentID = ""
	job.Status = domain.StatusDone
	job.Phase = "done"
	job.ConsensusPercentage = 100
	putProgress(job, 100, "completed")
	job.Result = map[string]interface{}{
		"summary":             summary,
		"consensusPercentage": 100,
		"seatDeliveries":      seatDeliveries,
	}
	plan["awaitReason"] = nil
	job.Plan = plan
	r.flush(job)
	logging.L.Infow("真模型会商完成",
		logging.KeyJobID, jobID,
		logging.KeyStatus, domain.StatusDone,
		logging.KeyModel, job.Model,
	)
}

// pauseForInput 把作业停在 awaiting_input。seatIndex 记下一席，等人或等确认后再从这里继续。
// reason 只有两类约定：澄清（用户要补一句）和 manual_continue（人工确认再往下走）。
func (r *LiveChainRunner) pauseForInput(job *domain.AnalysisJob, plan map[string]interface{}, next int, reason, detail string) {
	plan["seatIndex"] = next
	plan["awaitReason"] = reason
	job.Plan = plan
	job.Status = domain.StatusAwaitingInput
	job.Phase = reason
	job.PendingClarification = detail
	msg := "等待用户澄清"
	if reason == "manual_continue" {
		msg = "等待人工确认继续"
	}
	putProgress(job, job.ConsensusPercentage, msg)
	r.flush(job)
	logging.L.Infow("作业暂停等待输入",
		logging.KeyJobID, job.ID,
		logging.KeyStatus, domain.StatusAwaitingInput,
		logging.KeyPhase, reason,
	)
}

func (r *LiveChainRunner) deltaHandler(job *domain.AnalysisJob, bubbleID string) deerflow.TextDeltaHandler {
	return func(delta string) error {
		appendBubble(job, bubbleID, delta)
		r.throttleFlush(job)
		return nil
	}
}

func (r *LiveChainRunner) throttleFlush(job *domain.AnalysisJob) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if time.Since(r.lastFlush) >= 150*time.Millisecond {
		r.flushLocked(job)
	}
}

func (r *LiveChainRunner) flush(job *domain.AnalysisJob) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.flushLocked(job)
}

// flushLocked 把内存里的作业写回表。写之前从库里拉模型、人工开关和尚未消费的注入，
// 这样页面上改设置不会被本轮旧内存盖掉。调用方必须已持有 r.mu。
func (r *LiveChainRunner) flushLocked(job *domain.AnalysisJob) {
	plan := ensurePlan(job)
	r.pullManualFromDB(job, plan)
	r.pullModelFromDB(job)
	injectConsumed := plan["_injectConsumed"] == true
	if !injectConsumed {
		if strings.TrimSpace(asStr(plan["pendingInject"])) == "" {
			r.pullInjectFromDB(job, plan)
		}
	}
	delete(plan, "_injectConsumed")
	job.Plan = plan
	_ = r.Repo.UpdateProgress(job)
	r.lastFlush = time.Now()
}

func (r *LiveChainRunner) cancelled(jobID string) bool {
	ok, err := r.Repo.IsCancelFlag(jobID)
	return err == nil && ok
}

func (r *LiveChainRunner) markCancelled(job *domain.AnalysisJob) {
	job.Status = domain.StatusCancelled
	job.Phase = "cancelled"
	job.Error = "cancelled"
	putProgress(job, job.ConsensusPercentage, "cancelled")
	r.flush(job)
	logging.L.Infow("真模型会商已取消", logging.KeyJobID, job.ID, logging.KeyStatus, domain.StatusCancelled)
}

func (r *LiveChainRunner) fail(job *domain.AnalysisJob, message string) {
	job.Status = domain.StatusError
	job.Phase = "error"
	job.Error = message
	job.ConsensusPercentage = 0
	if job.Result != nil {
		delete(job.Result, "summary")
		job.Result["consensusPercentage"] = 0
		job.Result["failed"] = true
	}
	putProgress(job, 0, "error")
	r.flush(job)
	// 详情截断，避免把整段模型正文打进日志
	detail := message
	if rs := []rune(detail); len(rs) > 200 {
		detail = string(rs[:200]) + "…"
	}
	logging.L.Warnw("真模型会商失败",
		logging.KeyJobID, job.ID,
		logging.KeyStatus, domain.StatusError,
		logging.KeyModel, job.Model,
		"detail", detail,
	)
}

func (r *LiveChainRunner) pullModelFromDB(job *domain.AnalysisJob) {
	fresh, err := r.Repo.FindByID(job.ID)
	if err != nil || fresh == nil {
		return
	}
	job.Model = fresh.Model
}

func (r *LiveChainRunner) pullManualFromDB(job *domain.AnalysisJob, plan map[string]interface{}) {
	fresh, err := r.Repo.FindByID(job.ID)
	if err != nil || fresh == nil || fresh.Plan == nil {
		return
	}
	if v, ok := fresh.Plan["manualIntervention"]; ok {
		plan["manualIntervention"] = v
	}
}

func (r *LiveChainRunner) pullInjectFromDB(job *domain.AnalysisJob, plan map[string]interface{}) {
	fresh, err := r.Repo.FindByID(job.ID)
	if err != nil || fresh == nil || fresh.Plan == nil {
		return
	}
	dbText := strings.TrimSpace(asStr(fresh.Plan["pendingInject"]))
	if dbText == "" {
		return
	}
	local := strings.TrimSpace(asStr(plan["pendingInject"]))
	if local == "" {
		plan["pendingInject"] = dbText
	} else if !strings.Contains(local, dbText) {
		plan["pendingInject"] = local + "\n\n" + dbText
	}
}

func consumePendingInject(plan map[string]interface{}) string {
	s := strings.TrimSpace(asStr(plan["pendingInject"]))
	plan["pendingInject"] = nil
	plan["_injectConsumed"] = true
	return s
}

func ensurePlan(job *domain.AnalysisJob) map[string]interface{} {
	p := job.EnsurePlan()
	if _, ok := p["seatIndex"]; !ok {
		p["seatIndex"] = 0
	}
	if _, ok := p["manualIntervention"]; !ok {
		p["manualIntervention"] = true
	}
	return p
}

func putProgress(job *domain.AnalysisJob, percent int, message string) {
	p := job.EnsureProgress()
	p["percent"] = percent
	p["message"] = message
}

func loadThreadIDs(job *domain.AnalysisJob) map[string]string {
	out := map[string]string{}
	for k, v := range job.ThreadIDs {
		if v != nil {
			out[k] = fmt.Sprint(v)
		}
	}
	return out
}

func loadDeliveries(job *domain.AnalysisJob) []map[string]interface{} {
	if job.Result == nil {
		return nil
	}
	raw, ok := job.Result["seatDeliveries"]
	if !ok {
		return nil
	}
	switch t := raw.(type) {
	case []map[string]interface{}:
		return t
	case []interface{}:
		var out []map[string]interface{}
		for _, o := range t {
			if m, ok := o.(map[string]interface{}); ok {
				out = append(out, m)
			}
		}
		return out
	default:
		return nil
	}
}

func saveDeliveries(job *domain.AnalysisJob, list []map[string]interface{}) {
	if job.Result == nil {
		job.Result = map[string]interface{}{}
	}
	job.Result["seatDeliveries"] = list
}

func readManual(plan map[string]interface{}) bool {
	if plan == nil {
		return true
	}
	v, ok := plan["manualIntervention"]
	if !ok || v == nil {
		return true
	}
	switch t := v.(type) {
	case bool:
		return t
	default:
		return !strings.EqualFold(strings.TrimSpace(fmt.Sprint(t)), "false")
	}
}

func openBubble(job *domain.AnalysisJob, role, sender, agentID, phase string) string {
	list := job.EnsureDialogues()
	id := fmt.Sprintf("dlg_%s_%d", job.ID, len(list)+1)
	job.Dialogues = append(list, map[string]interface{}{
		"id": id, "role": role, "sender": sender, "agentId": agentID,
		"content": "", "status": "streaming", "phaseLabel": phase,
	})
	return id
}

func appendBubble(job *domain.AnalysisJob, bubbleID, delta string) {
	if delta == "" {
		return
	}
	for _, b := range job.Dialogues {
		if asStr(b["id"]) == bubbleID {
			b["content"] = asStr(b["content"]) + delta
			return
		}
	}
}

func finishBubble(job *domain.AnalysisJob, bubbleID, finalContent, status string) {
	for _, b := range job.Dialogues {
		if asStr(b["id"]) != bubbleID {
			continue
		}
		if strings.TrimSpace(finalContent) != "" {
			prev := asStr(b["content"])
			b["content"] = mergeKeepThinking(prev, finalContent)
		}
		if status == "" {
			status = "done"
		}
		b["status"] = status
		b["phaseLabel"] = ""
		return
	}
}

func mergeKeepThinking(streamed, finalContent string) string {
	if strings.Contains(finalContent, "<think>") || strings.Contains(finalContent, "</think>") {
		return finalContent
	}
	open := strings.Index(streamed, "<think>")
	if open < 0 {
		return finalContent
	}
	close := strings.LastIndex(streamed, "</think>")
	if close >= open {
		return streamed[:close+len("</think>")] + "\n" + finalContent
	}
	return streamed + "</think>\n" + finalContent
}

func applyAgentSpeaking(agents []map[string]interface{}, speakingID string, seatTurn bool) {
	for _, a := range agents {
		id := asStr(a["id"])
		if seatTurn && id == speakingID {
			a["status"] = "speaking"
		} else {
			a["status"] = "listening"
		}
	}
}

func dialoguesHaveError(job *domain.AnalysisJob) bool {
	for _, b := range job.Dialogues {
		if strings.EqualFold(asStr(b["status"]), "error") {
			return true
		}
	}
	return false
}

func looksLikeModelFailure(content string) bool {
	t := strings.TrimSpace(content)
	if t == "" {
		return false
	}
	lower := strings.ToLower(t)
	needles := []string{"model_not_found", "invalid model", "does not exist", "no such model",
		"litellm", "aperror", "apierror", "authenticationerror", "rate_limit", "context_length"}
	for _, n := range needles {
		if strings.Contains(lower, n) {
			return true
		}
	}
	cn := []string{"大模型本次调用未能成功完成", "请稍后再试或更换模型", "大模型因鉴权失败",
		"大模型服务暂时繁忙", "大模型响应暂时中断", "大模型当前响应异常", "已被熔断保护",
		"已拒绝自动切换", "模型不存在", "无可用模型", "模型不可用", "鉴权失败"}
	for _, n := range cn {
		if strings.Contains(t, n) {
			return true
		}
	}
	if utf8.RuneCountInString(t) < 400 && len(t) >= 5 && t[0] == '[' && t[4] == ']' {
		return true
	}
	return false
}

func buildModelHint(head, model, detail string) string {
	sb := head
	if strings.TrimSpace(model) != "" {
		sb += "（当前模型：" + strings.TrimSpace(model) + "）"
	}
	sb += "。请换可用模型后重新开始会商。"
	d := strings.TrimSpace(detail)
	if d != "" {
		if utf8.RuneCountInString(d) > 180 {
			rs := []rune(d)
			d = string(rs[:180]) + "…"
		}
		sb += " 详情：" + d
	}
	return sb
}

func toInt(v interface{}, def int) int {
	switch t := v.(type) {
	case int:
		return t
	case int64:
		return int(t)
	case float64:
		return int(t)
	case string:
		var n int
		_, err := fmt.Sscanf(t, "%d", &n)
		if err == nil {
			return n
		}
	}
	return def
}