package worker
import (
"fmt"
"log"
"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/repo"
)
// LiveChainRunner 真模型按链条顺序执行（被 LiveWorker 异步调用）。
type LiveChainRunner struct {
DF *deerflow.Client
Repo *repo.JobRepo
Coord string
mu sync.Mutex
lastFlush time.Time
}
func (r *LiveChainRunner) Run(job *domain.AnalysisJob) {
jobID := job.ID
defer func() {
if rec := recover(); rec != nil {
log.Printf("真模型会商 panic jobId=%s: %v", jobID, 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)
tids, err := r.DF.InitMultiAgent(agentNames, r.Coord, job.Model)
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)
log.Printf("真模型会商完成 jobId=%s", jobID)
}
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)
log.Printf("作业暂停等待输入 jobId=%s reason=%s", job.ID, 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)
}
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)
}
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)
}
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, "") || strings.Contains(finalContent, "") {
return finalContent
}
open := strings.Index(streamed, "")
if open < 0 {
return finalContent
}
close := strings.LastIndex(streamed, "")
if close >= open {
return streamed[:close+len("")] + "\n" + finalContent
}
return streamed + "\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
}