DEPTH_ROADMAP.md

@@ -183,8 +183,8 @@ RBAC 未做，暂以单管理员账号代理；概览口径必须是**系统级*
 - [ ] mcp-py 算法层去桩：文档解析接 PaddleOCR/magic-pdf（mineru.py:11 返回空块）| M
 - [ ] 报告原生 PDF（现仅 Word，office/unioffice.go）| M
 
-### [ ] T4.E 编排引擎边角
+### [~] T4.E 编排引擎边角
-- [ ] Map 节点错误传播 + 汇总（eino/graph.go:65 单子项失败无感知）| M
+- [x] Map 节点错误传播 + 汇总 ✅ —— writeSection 返回 (body,err) 仅真·LLM 失败计错(预算/无模型是主动降级不算)；writeSections 汇总失败数、失败项 Body 带可见标记；mapNode 全失败→置 b.fatalErr(判 failed 非静默 done-空)、部分失败→trace+流式告警。report handleReport 同步提示部分失败。3 单测(全失败/部分精确计数/mapNode 置 fatalErr)。
 - [ ] 熔断器接回 failover（harness/circuitbreaker.go 与 llm/failover.go 未接合，可能长卡备用）| M
 - [ ] Branch else 兜底 + coordinator 专家超时（专家卡死拖垮全局，coordinator.go:109）| M
 - [ ] DSL 拓扑/节点-工具映射校验（dsl/parser.go:23 TODO，现仅 JSON 格式校验）| M

sundynix-dispatcher/internal/eino/graph.go

@@ -68,7 +68,7 @@ func (o *Orchestrator) mapNode(ctx context.Context, taskID string, n dsl.Node, b
 	end(fmt.Sprintf("拆出 %d 项：%s", len(items), strings.Join(items, " / ")), nil)
 
 	o.emit(taskID, fmt.Sprintf("\n> 并行处理 %d 项…\n\n", len(items)))
-	secs := o.writeSections(ctx, b.query, b.kb, items, tr) // 有界并发，trace 出 section:i 各项
+	secs, failed := o.writeSections(ctx, b.query, b.kb, items, tr) // 有界并发，trace 出 section:i 各项
 	b.sections = secs
 	for _, s := range secs {
 		chunk := "## " + s.Heading + "\n\n" + s.Body + "\n\n"
@@ -76,6 +76,20 @@ func (o *Orchestrator) mapNode(ctx context.Context, taskID string, n dsl.Node, b
 		b.answer += chunk
 		b.refs = append(b.refs, s.Heading+"："+s.Body)
 	}
+	// 错误传播：并行子项不再静默丢失——汇总成败；全失败则置致命错，让任务判 failed 而非 done-空。
+	if failed > 0 {
+		endMap := tr.span("map:"+n.ID+":result", "plan", "并行结果汇总")
+		if failed >= len(items) && len(items) > 0 {
+			if b.fatalErr == nil {
+				b.fatalErr = fmt.Errorf("并行 fan-out 全部 %d 项撰写失败", failed)
+			}
+			endMap(fmt.Sprintf("全部 %d 项失败", failed), b.fatalErr)
+		} else {
+			note := fmt.Sprintf("%d/%d 项失败（已标注），其余成功", failed, len(items))
+			endMap(note, nil)
+			o.emit(taskID, fmt.Sprintf("\n> ⚠️ %s\n\n", note))
+		}
+	}
 }
 
 // execToolNode 执行工具节点：调 MCP 工具，产出累计进黑板；失败降级不阻断。

sundynix-dispatcher/internal/eino/map_error_test.go

@@ -0,0 +1,70 @@
+package eino
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/sundynix/sundynix-dispatcher/internal/dsl"
+	"github.com/sundynix/sundynix-dispatcher/internal/llm"
+)
+
+// writeSections：真·LLM 失败应逐项标记（Body 带可见失败标记）并汇总失败数，不静默丢失。
+func TestWriteSections_PropagatesFailures(t *testing.T) {
+	ll := &fakeLLM{ready: true, chat: func(_ []llm.ChatMessage) (string, error) {
+		return "", errors.New("boom")
+	}}
+	o := newOrch(ll, kbTool(nil), &fakeSink{}, &fakeExec{})
+
+	secs, failed := o.writeSections(context.Background(), "主题", "", []string{"A", "B", "C"}, o.tracer("t1"))
+	if failed != 3 {
+		t.Fatalf("全失败应 failed=3，got %d", failed)
+	}
+	if len(secs) != 3 {
+		t.Fatalf("应返回 3 个 section，got %d", len(secs))
+	}
+	for _, s := range secs {
+		if !strings.Contains(s.Body, "撰写失败") {
+			t.Errorf("失败项 Body 应含可见失败标记，got %q", s.Body)
+		}
+	}
+}
+
+// 部分失败：仅命中 heading B 的调用失败 → failed 精确为 1（非 all-or-nothing），A/C 正常。
+func TestWriteSections_PartialFailureCount(t *testing.T) {
+	ll := &fakeLLM{ready: true, chat: func(m []llm.ChatMessage) (string, error) {
+		for _, msg := range m {
+			if strings.Contains(msg.Content, "本章标题：B") {
+				return "", errors.New("boom-B")
+			}
+		}
+		return "正文", nil
+	}}
+	o := newOrch(ll, kbTool(nil), &fakeSink{}, &fakeExec{})
+
+	secs, failed := o.writeSections(context.Background(), "主题", "", []string{"A", "B", "C"}, o.tracer("t1"))
+	if failed != 1 {
+		t.Fatalf("仅 B 失败应 failed=1，got %d", failed)
+	}
+	if !strings.Contains(secs[1].Body, "撰写失败") {
+		t.Errorf("B 应标记失败，got %q", secs[1].Body)
+	}
+	if strings.Contains(secs[0].Body, "撰写失败") || strings.Contains(secs[2].Body, "撰写失败") {
+		t.Errorf("A/C 不应被误判失败")
+	}
+}
+
+// mapNode：并行 fan-out 全部子项失败 → 置 fatalErr，任务判 failed 而非静默 done-空。
+func TestMapNode_AllFailSetsFatal(t *testing.T) {
+	ll := &fakeLLM{ready: true, chat: func(_ []llm.ChatMessage) (string, error) {
+		return "", errors.New("boom")
+	}}
+	o := newOrch(ll, kbTool(nil), &fakeSink{}, &fakeExec{})
+	b := &board{query: "分析一下"}
+
+	o.mapNode(context.Background(), "t1", dsl.Node{ID: "m1"}, b, o.tracer("t1"))
+	if b.fatalErr == nil {
+		t.Fatalf("全部子项失败时应置 fatalErr（否则任务会静默判 done-空）")
+	}
+}

sundynix-dispatcher/internal/eino/rag_integration_test.go

@@ -96,7 +96,7 @@ func TestRAG_ReportSectionInjectsRefs(t *testing.T) {
 	}}
 	o := newOrch(ll, kbTool(nil), &fakeSink{}, &fakeExec{})
 
-	body := o.writeSection(context.Background(), "杭州旅游", "u1/travel", "西湖", o.tracer("t1"), "section:0")
+	body, _ := o.writeSection(context.Background(), "杭州旅游", "u1/travel", "西湖", o.tracer("t1"), "section:0")
 	if !strings.Contains(body, ragSnippet) {
 		t.Errorf("报告章节撰写 prompt 应含检索资料，got %q", body)
 	}

sundynix-dispatcher/internal/eino/report.go

@@ -7,6 +7,7 @@ import (
 	"log"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/sundynix/sundynix-dispatcher/internal/dsl"
@@ -75,7 +76,12 @@ func (o *Orchestrator) handleReport(ctx context.Context, t *contract.Task, tr *e
 		o.emit(t.ID, "\n> 正在并行撰写各章…\n\n")
 	}
 
-	sections := o.writeSections(ctx, topic, kb, outline.Sections, tr)
+	sections, failed := o.writeSections(ctx, topic, kb, outline.Sections, tr)
+	if failed > 0 {
+		note := fmt.Sprintf("⚠️ %d/%d 章撰写失败（下方已标注），其余照常成稿。", failed, len(sections))
+		tr.info("plan", "system", "部分章节失败", note)
+		o.emit(t.ID, "\n> "+note+"\n\n")
+	}
 
 	// 把完整报告正文流式呈现给客户端。
 	o.emit(t.ID, "\n---\n\n# "+firstNonEmpty(outline.Title, topic)+"\n\n")
@@ -162,11 +168,13 @@ func (o *Orchestrator) planItems(ctx context.Context, topic, splitBy string) []s
 	return items
 }
 
-// writeSections 各章节并行撰写（有界并发），结果按原顺序返回。
+// writeSections 各章节并行撰写（有界并发），结果按原顺序返回；第二返回值为失败项数。
-func (o *Orchestrator) writeSections(ctx context.Context, topic, kb string, headings []string, tr *execTracer) []reportSection {
+// 失败项的 Body 会带明确的失败标记（可见，不静默），失败计数供上游汇总/判定整体成败。
+func (o *Orchestrator) writeSections(ctx context.Context, topic, kb string, headings []string, tr *execTracer) ([]reportSection, int) {
 	out := make([]reportSection, len(headings))
 	sem := make(chan struct{}, reportFanout)
 	var wg sync.WaitGroup
+	var failed atomic.Int64
 	for i, h := range headings {
 		wg.Add(1)
 		go func(i int, h string) {
@@ -175,26 +183,33 @@ func (o *Orchestrator) writeSections(ctx context.Context, topic, kb string, head
 			defer func() { <-sem }()
 			node := fmt.Sprintf("section:%d", i)
 			end := tr.span(node, "section", fmt.Sprintf("第%d章 %s", i+1, h))
-			body := o.writeSection(ctx, topic, kb, h, tr, node)
+			body, err := o.writeSection(ctx, topic, kb, h, tr, node)
-			end(fmt.Sprintf("成稿 %d 字", len([]rune(body))), nil)
+			if err != nil {
+				failed.Add(1)
+				end("撰写失败："+err.Error(), err) // trace 记为该 section 失败（非静默）
+			} else {
+				end(fmt.Sprintf("成稿 %d 字", len([]rune(body))), nil)
+			}
 			out[i] = reportSection{Heading: h, Body: body}
 		}(i, h)
 	}
 	wg.Wait()
-	return out
+	return out, int(failed.Load())
 }
 
 // writeSection 撰写一章：先 RAG 检索参考资料（若挂了知识库），再让模型成稿。
-func (o *Orchestrator) writeSection(ctx context.Context, topic, kb, heading string, tr *execTracer, node string) string {
+// 返回 (正文, 失败错误)：仅「真·LLM 调用失败」返回非 nil err；预算触顶 / 模型未配置是
+// 主动降级（出可见的降级正文、err=nil），不计入失败。
+func (o *Orchestrator) writeSection(ctx context.Context, topic, kb, heading string, tr *execTracer, node string) (string, error) {
 	refs := o.retrieve(ctx, kb, topic+" "+heading)
 	if refs != "" {
 		tr.info(node, "section", "检索参考资料", truncate(strings.ReplaceAll(refs, "\n", " "), 120))
 	}
 	if !o.pool.Ready() {
 		if refs != "" {
-			return "（模型未配置，以下为检索到的参考资料）\n" + refs
+			return "（模型未配置，以下为检索到的参考资料）\n" + refs, nil
 		}
-		return "（模型未配置，无法撰写本章。）"
+		return "（模型未配置，无法撰写本章。）", nil
 	}
 	sys := "你是专业报告撰稿人，语言严谨、条理清晰，使用中文书面语。"
 	var ub strings.Builder
@@ -210,7 +225,7 @@ func (o *Orchestrator) writeSection(ctx context.Context, topic, kb, heading stri
 		bud.AddPrompt(sys + ub.String())
 		if bud.Exceeded() {
 			tr.info(node, "system", "token 预算", "已达预算上限，跳过本章")
-			return "（已达 token 预算上限，本章自动跳过。）"
+			return "（已达 token 预算上限，本章自动跳过。）", nil
 		}
 	}
 	cctx, cancel := llmCtx(ctx)
@@ -218,12 +233,12 @@ func (o *Orchestrator) writeSection(ctx context.Context, topic, kb, heading stri
 	txt, err := o.pool.Chat(cctx, []llm.ChatMessage{{Role: "system", Content: sys}, {Role: "user", Content: ub.String()}})
 	if err != nil {
 		log.Printf("[report] 撰写「%s」失败: %v", heading, err)
-		return "（本章撰写失败：" + err.Error() + "）"
+		return "（本章撰写失败：" + err.Error() + "）", err
 	}
 	if bud := harness.BudgetFrom(ctx); bud != nil {
 		bud.AddComplete(txt) // 成本护栏：计入输出
 	}
-	return strings.TrimSpace(txt)
+	return strings.TrimSpace(txt), nil
 }
 
 // retrieve 经 Eino Retriever 组件（包 mcp-go kb_search）检索知识库，整理为可读参考资料。