feat(service): 建立多层级流式事件发射模型与Context穿透引擎

Co-authored-by: Cursor <cursoragent@cursor.com>

docs/superpowers/plans/2026-09-08-streaming-events-and-digestible-ux.md

@@ -0,0 +1,111 @@
+# 流式传输服务与按需呈现交互实现计划 (Streaming Events & Digestible UX Implementation Plan)
+
+本文档制定“方向三：HTTP API 与 Server-Sent Events (SSE) 流式对话服务”的详细工程实施计划。重点落实阶段性任务状态更新、三层信息分级（状态、摘要、结论）与底层推演按需折叠机制。
+
+## 一、 总体技术架构与流转路径
+
+```
+[前端 Web / CLI 终端 / API 客户端]
+         │
+         │  HTTP / POST /api/v1/chat/stream (SSE 长连接, include_details=false/true)
+         ▼
+ ┌───────────────────────────────────────────────────────────────┐
+ │ internal/server: HTTP RESTful API & SSE 流式长连接引擎          │
+ └───────┬───────────────────────────────────────────────────────┘
+         │ 携带 Streamer Context
+         ▼
+ ┌───────────────────────────────────────────────────────────────┐
+ │ internal/chat & coordinator: 协同对话层与意图流式识别           │
+ │  • 发射 status: 意图识别分支、参数槽位、当前阶段                │
+ └───────┬───────────────────────────────────────────────────────┘
+         │ 任务执行 / 阶段跃迁推进
+         ▼
+ ┌───────────────────────────────────────────────────────────────┐
+ │ internal/bizdev & squad: 智能体小队与知识检索调度流             │
+ │  • 发射 phase: "squad_assembly" (组建小队)                    │
+ │  • 发射 phase: "knowledge_retrieval" (法规专家筛选条款)        │
+ │  • 发射 phase: "step_executing" (撰写专员初稿中)              │
+ │  • 发射 phase: "quality_gate" (审核主管门禁把关中)             │
+ │  • 发射 detail: 步骤推演详细草稿 (标记 collapsible: true)       │
+ └───────┬───────────────────────────────────────────────────────┘
+         │ 汇总交付与 B 轨提炼
+         ▼
+ ┌───────────────────────────────────────────────────────────────┐
+ │ 结论提炼与分层交付                                             │
+ │  • 发射 summary: 核心进展摘要 (1-3 句话低认知负荷)             │
+ │  • 发射 briefing: B 轨参谋决策锦囊 (评级、假设、风险、选项)     │
+ │  • 发射 deliverable: A 轨正式合规文档生成通知                   │
+ │  • 发射 done: 流正常结束                                      │
+ └───────────────────────────────────────────────────────────────┘
+```
+
+---
+
+## 二、 任务拆分与渐进式提交规划 (Progressive Commits)
+
+### Task 1: 结构化流式事件体系与上下文穿透器 (internal/stream) (feat(service))
+- **目标**：建立统一的流式事件模型（`StreamEvent`）、事件类型（`phase`, `status`, `summary`, `briefing`, `detail`, `delta`, `deliverable`, `done`, `error`）与 Context 注入提取工具。
+- **涉及文件**：
+  - 新建: `internal/stream/types.go`
+  - 新建: `internal/stream/stream.go`
+  - 新建: `internal/stream/stream_test.go`
+- **渐进提交**：`feat(service): 建立多层级流式事件发射模型与Context穿透引擎`
+
+### Task 2: 任务全流程节点流式事件插桩打点 (feat(agent))
+- **目标**：在智能体小队编排、知识库检索、专员单步执行、质量门禁把关及 B 轨参谋提炼中，实时发射阶段性状态（组建小队、法规筛选、初稿撰写、质量核验）及按需折叠的详细推演。
+- **涉及文件**：
+  - 修改: `internal/squad/models.go` / `workflow.go` / `service.go`
+  - 修改: `internal/bizdev/types.go` / `service.go`
+  - 修改: `internal/coordinator/engine.go`
+  - 修改: `internal/chat/session.go`
+- **渐进提交**：`feat(agent): 在小队编排与任务生命周期中打通阶段性状态与摘要流式通知`
+
+### Task 3: HTTP RESTful API 服务与 SSE 流式端点实现 (feat(api))
+- **目标**：基于 Go 标准库 `net/http` 构建开发环境 8080 服务，实现 `/api/v1/health`、`/api/v1/projects/:id`、`/api/v1/chat` 及核心 `/api/v1/chat/stream`，支持 `include_details` 按需控制参数。
+- **涉及文件**：
+  - 新建: `internal/server/server.go`
+  - 新建: `internal/server/handlers.go`
+  - 新建: `internal/server/sse.go`
+  - 新建: `internal/server/server_test.go`
+  - 新建: `cmd/server/main.go`
+  - 修改: `internal/config/config.go`
+- **渐进提交**：`feat(api): 实现HTTP服务骨架与Server-Sent Events流式交互接口`
+
+### Task 4: CLI 终端流式体验与分层交互升级 (feat(cli))
+- **目标**：升级 `cmd/bizdev-cli`，在控制台默认展现清爽的阶段性转轮/状态提示（如 `[组建小队] -> [法规筛选中] -> [撰写初稿中]`），收敛中间海量日志，并呈现清晰的【核心摘要】与【决策锦囊】，支持 `--verbose` 按需展开底层专员推演。
+- **涉及文件**：
+  - 修改: `cmd/bizdev-cli/main.go`
+- **渐进提交**：`feat(cli): 终端支持动态阶段状态流式输出与低认知负荷摘要呈现`
+
+### Task 5: 全链路集成验证、真实 Agent 实测与测试报告归档 (test & docs)
+- **目标**：
+  1. 运行全量 `go test ./...` 确保单测与集成测试 100% 通过；
+  2. 使用真实 `pi-agent` + `MiniMax-M3` 容器后端运行真实流式对话，取样并验证 SSE 流中的阶段状态、摘要、结论与按需折叠细节；
+  3. 按照标准规范生成测试报告归档至 `.doc/test-reports/2026-09-08-streaming-api-and-digestible-ux-test-report.md`。
+- **涉及文件**：
+  - 新建: `internal/server/e2e_test.go`
+  - 新建: `.doc/test-reports/2026-09-08-streaming-api-and-digestible-ux-test-report.md`
+- **渐进提交**：
+  - `test(api): 补充SSE流式长连接端到端自动化测试套件`
+  - `docs(test): 归档流式传输服务与按需呈现交互规范测试报告`
+
+---
+
+## 三、 详细步骤与执行清单
+
+- [ ] **Step 1**: 实现 `internal/stream` 核心包，支持事件订阅、Context 注入、过滤与多接收端分发；
+- [ ] **Step 2**: 完善 `internal/stream/stream_test.go` 保证并发与边界测试通过；
+- [ ] **Step 3**: 在 `squad.WorkflowDriver` 与 `Service` 中插桩：
+  - 发射 `phase: squad_assembly`（组建小队中）
+  - 发射 `phase: knowledge_retrieval`（法规专家正在筛选条款）
+  - 发射 `phase: step_executing`（单步角色、目标与进度百分比）
+  - 发射 `phase: quality_gate`（审核主管门禁校验中）
+  - 发射 `detail`（专员推演草稿，携带 `collapsible: true`）
+  - 发射 `summary` 与 `briefing`（B轨参谋锦囊）
+- [ ] **Step 4**: 在 `internal/chat/session.go` 中提供 `SendMessageStream(ctx, input, listener)`；
+- [ ] **Step 5**: 实现 `internal/server` HTTP 服务与 SSE 协议转换处理器；
+- [ ] **Step 6**: 实现 `cmd/server/main.go`，绑定配置中端口（默认 8080）；
+- [ ] **Step 7**: 升级 `cmd/bizdev-cli/main.go` 提供动态流式终端交互；
+- [ ] **Step 8**: 运行全量测试套件并执行真实 Agent SSE 端到端实测；
+- [ ] **Step 9**: 按照 `.cursor/rules/test-reports-standard.mdc` 规范撰写并归档测试报告；
+- [ ] **Step 10**: 执行阶段性 Git 提交入库。

docs/superpowers/specs/2026-09-08-streaming-events-and-digestible-ux-design.md

@@ -0,0 +1,158 @@
+# 流式传输协议与按需呈现交互规范设计 (Streaming API & Digestible UX Design)
+
+## 一、 背景与业务痛点 (Background & Pain Points)
+
+在商务与技术协同推进体系中，智能体小队在真实 `pi-agent`（MiniMax-M3）驱动下，针对医疗器械开展需求接入（`intake`）、注册可行性研判（`feasibility_reg`）与初步方案设计（`prelim_design`）等任务，通常需要经历：
+1. 法规标准知识库定向检索与条文初筛；
+2. 跨学科专员角色（如法规专员、撰写专员、审核主管）自主多阶段推演；
+3. 自动化质量门禁（QualityGate）合规核验与自我修复；
+4. 交付物汇总与业务参谋决策锦囊提炼。
+
+全流程在真实多智能体环境下执行耗时通常在数分钟（如 3~6 分钟）。当前系统面临以下关键用户体验与交互瓶颈：
+1. **黑盒漫长等待与确定性预期缺失**：用户发出指令后，控制台或客户端无中间状态回传，用户不知道当前系统在检索法规、撰写初稿还是在做质量审核，极易产生“系统是否卡死挂起”的焦虑；
+2. **信息洪流造成的认知过载 (Cognitive Overload)**：若将 Agent 的长文本思考链、工具调用明细、数百行技术草稿一股脑倾泻给用户，用户在极短时间内无法提取核心要点，产生强烈的认知疲劳；
+3. **缺乏清晰的信息分层**：用户最关心的是“当前进度如何（状态）”、“达成了什么结论（摘要）”以及“后续该怎么走（参谋建议）”，而详细的技术报告与专员论证记录应作为可追溯的支撑材料“按需查看”，不应喧宾夺主。
+
+---
+
+## 二、 设计目标与核心原则 (Design Goals & Core Principles)
+
+### 2.1 核心目标
+1. **建立确定性心智预期**：通过毫秒级实时流式传输（Server-Sent Events），向用户清晰同步阶段性任务状态（如：正在组建小队、法规专家正在筛选条款、撰写专员正在生成初稿、审核主管正在进行门禁校验）；
+2. **分层按需呈现 (Digestible on Demand)**：
+   - 默认呈现：**状态信息 (Status)** + **核心摘要 (Summary)** + **参谋决策建议 (Briefing)**；
+   - 按需呈现：**详细推演过程 (Detail)** 默认折叠，用户可随时一键展开深入核查；
+3. **标准化 RESTful HTTP API 与 SSE 对话服务**：基于 Go 标准库 `net/http` 构建企业级服务端，支持开发环境 `8080` 端口常驻；
+4. **双轨交付与贴身参谋无缝融合**：流式推送不仅服务于对话回复（Token Delta），更原生承载 A 轨合规交付物与 B 轨业务参谋决策锦囊（成熟度、预设假设、风险雷达、决策选项、汇报话术）。
+
+### 2.2 核心原则
+- **不打扰原则 (Non-Intrusive)**：不给用户输出过多的未经提炼的原始日志，保证用户在 10 秒内能看懂核心结论；
+- **透明可追溯原则 (Transparent & Traceable)**：详细的思考与草稿不能丢失，全部打标为 `collapsible` 细节按需提供；
+- **全链路非阻塞流转 (End-to-End Streaming)**：从 L1 意图层、L2 协调层到 L3 真实小队执行层，事件穿透并实时推送。
+
+---
+
+## 三、 信息架构三层分层模型 (Three-Tier Information Architecture)
+
+| 信息层级 | 对应流式事件 | 面向用户的心智价值 | 展现形态与认知成本 |
+| :--- | :--- | :--- | :--- |
+| **L1: 阶段性任务状态 (Phase / Status)** | `phase`, `status` | **“当前在干嘛、进行到哪了”**<br>消除黑盒焦虑，建立确定性预期 | **高频刷新、单行即时动态指示**<br>例如：“正在组建智能体小队（法规专员、撰写专员、审核主管）...”<br>“法规专家正在进行法规筛选与标准研判...”<br>“审核主管正在进行质量门禁与引用核验...” |
+| **L2: 摘要与结论信息 (Summary & Briefing)** | `summary`, `briefing`, `delta` | **“核心发现了什么、该怎么决策”**<br>低认知负荷，快速掌握局面 | **精炼成文、突出核心结论**<br>• **摘要 (Summary)**：1-3 句话概括事实与核心研判<br>• **决策锦囊 (Briefing)**：B 轨成熟度等级、基准假设、风险雷达、选项 [1]/[2]、沟通话术<br>• **对话回答 (Delta)**：参谋自然语言解答流式逐句吐出 |
+| **L3: 底层执行与推演细节 (Detail / Raw Trace)** | `detail`, `deliverable` | **“具体推演依据、技术证据留底”**<br>满足专业审查，支持随时深挖 | **按需展开、默认折叠 (Collapsible)**<br>• 法规专员的细分研判草稿<br>• 审核专员的修改痕迹与门禁打分明细<br>• A 轨长达数百行的正式 Word/Markdown 报告文档路径与下载 |
+
+---
+
+## 四、 协议与事件规范 (SSE Protocol Specification)
+
+### 4.1 SSE 通信协议
+- 媒体类型：`Content-Type: text/event-stream`
+- 缓存控制：`Cache-Control: no-cache`
+- 连接机制：`Connection: keep-alive`
+- 字符编码：`UTF-8`
+
+### 4.2 事件类型定义
+
+```go
+type EventType string
+
+const (
+    EventPhase       EventType = "phase"       // 阶段性状态跃迁（组建小队、法规筛选、初稿撰写、门禁审核）
+    EventStatus      EventType = "status"      // 意图识别结果、成熟度状态机更新
+    EventSummary     EventType = "summary"     // 提炼的核心进展摘要（低认知负荷）
+    EventBriefing    EventType = "briefing"    // 贴身业务参谋决策锦囊（B轨赋能）
+    EventDelta       EventType = "delta"       // 自然语言逐字/逐句流式吐出
+    EventDetail      EventType = "detail"      // 详细推演草稿或思考链（按需显示，默认折叠）
+    EventDeliverable EventType = "deliverable" // A轨正式合规文档生成通知
+    EventError       EventType = "error"       // 运行异常警报
+    EventDone        EventType = "done"        // 会话流完成
+)
+```
+
+### 4.3 典型数据载荷示例
+
+#### 1. 阶段状态事件 (`event: phase`)
+```json
+{
+  "phase_id": "knowledge_retrieval",
+  "message": "法规专家正在检索权威法规知识库并筛选适用条款（匹配 GB 9706.1-2020、YY 0505）...",
+  "role": "regulatory",
+  "step_index": 1,
+  "total_steps": 3,
+  "progress": 25,
+  "timestamp": "2026-09-08T09:15:00Z"
+}
+```
+
+#### 2. 详细过程事件 (`event: detail`，按需折叠)
+```json
+{
+  "category": "step_output",
+  "step_id": "tasks/01_regulatory",
+  "title": "法规专员研判纪要",
+  "content": "### 01 法规专员研判纪要\n依据 2024 年分类界定结果第（十二）条，该设备结构归属于 06-15，建议管理类别判定为第二类医疗器械...",
+  "collapsible": true
+}
+```
+
+#### 3. 核心摘要事件 (`event: summary`)
+```json
+{
+  "title": "方案成型阶段进展摘要",
+  "content": "已完成医用电子内窥镜图像处理系统的初步架构与法规路径分析，推荐按 II 类注册（编码 06-15）走同品种比对主线。",
+  "maturity_score": 80
+}
+```
+
+#### 4. 业务参谋决策锦囊 (`event: briefing`)
+```json
+{
+  "advisor_brief": {
+    "maturity_score": 80,
+    "maturity_level": "V0.5",
+    "maturity_headline": "「医用电子内窥镜图像处理系统」的设计方案框架与法规路径已成型，已具备进行详细 WBS 拆解与报价的条件",
+    "baseline_assumptions": [
+      "系统架构采用模块化分层设计，软硬件接口保持独立兼容",
+      "电气安全与电磁兼容暂按 GB 9706.1-2020 / YY 0505 标准预留检测整改预算"
+    ],
+    "preemptive_risks": [
+      "关键元器件交付周期可能受供应链波动影响，建议首期备料周期预留 4 周冗余"
+    ],
+    "next_best_actions": [
+      "[1] 审阅初步系统架构图与法规分类论证说明",
+      "[2] 进入 V1.0 商业锁定阶段，生成明细 CDO 报价单与 Excel 核算模型"
+    ],
+    "communication_kit": "您好！关于「医用电子内窥镜图像处理系统」，我们已完成了技术路线比选与法规路径分析..."
+  }
+}
+```
+
+---
+
+## 五、 HTTP RESTful API 端点清单
+
+| 端点 | 方法 | 说明 | 关键参数 |
+| :--- | :--- | :--- | :--- |
+| `/api/v1/health` | GET | 服务健康检查与基础设施状态（包含知识库数据库连接） | 无 |
+| `/api/v1/projects/:id` | GET | 查询项目当前上下文、黑板事实、挂起预期与参谋简报 | `id` |
+| `/api/v1/chat` | POST | 基础非流式对话交互（返回完整 JSON） | `session_id`, `message`, `user_id` |
+| `/api/v1/chat/stream` | POST/GET | **核心 SSE 流式对话长连接服务** | `session_id`, `message`, `include_details` (bool, 默认 false) |
+| `/api/v1/tasks/run` | POST | 独立触发执行指定任务或成熟度阶段 | `task_key`, `project_name`, `device_type`, `auto_repair` |
+| `/api/v1/tasks/run/stream`| POST/GET | **任务执行全流程 SSE 监控流** | 任务参数 + `include_details` |
+
+---
+
+## 六、 客户端按需呈现策略 (Client Presentation Strategy)
+
+### 6.1 Web / UI 界面
+- **主对话视窗**：
+  - 顶部/卡片内渲染动态进度条与当前阶段（如 `[正在组建小队] -> [法规筛选中] -> [撰写初稿中] -> [门禁校验中]`）；
+  - 参谋自然语言回复逐句流畅显示；
+  - 底部浮现高亮【💡 参谋决策锦囊】卡片，包含成熟度徽章、掌舵选项快捷按钮；
+- **抽屉与折叠区**：
+  - 中间的 Agent 思考过程与各专员步骤草稿收拢在 `<details><summary>查看专家推演与合规论证过程 (3 步)</summary>...</details>`；
+  - 仅当用户主动点击时才展开，彻底避免初次打开时的信息过载。
+
+### 6.2 CLI 终端界面
+- 默认开启流式进度模式，使用动态转轮或单行覆盖更新当前阶段（`⏳ [组建小队] ...`）；
+- 完成后以清爽排版输出【核心摘要】与【决策锦囊】；
+- 增加 `--verbose` 开关，仅在用户显式指定时打印每步专员的完整中间推演。

internal/stream/stream.go

@@ -0,0 +1,238 @@
+package stream
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"chat-agent-system/internal/squad"
+)
+
+type contextKey struct{}
+
+var streamerCtxKey = contextKey{}
+
+// Listener 事件监听回调
+type Listener func(StreamEvent)
+
+// Streamer 流式事件发射器
+type Streamer struct {
+	mu             sync.RWMutex
+	listeners      []Listener
+	includeDetails bool
+	eventCount     int64
+}
+
+// NewStreamer 创建新的发射器
+func NewStreamer(includeDetails bool, listeners ...Listener) *Streamer {
+	return &Streamer{
+		listeners:      listeners,
+		includeDetails: includeDetails,
+	}
+}
+
+// AddListener 动态注册监听器
+func (s *Streamer) AddListener(l Listener) {
+	if s == nil || l == nil {
+		return
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.listeners = append(s.listeners, l)
+}
+
+// Emit 发送一个流式事件
+func (s *Streamer) Emit(ev StreamEvent) {
+	if s == nil {
+		return
+	}
+	// 详情按需过滤
+	if ev.Type == EventDetail && !s.includeDetails {
+		return
+	}
+
+	count := atomic.AddInt64(&s.eventCount, 1)
+	if ev.ID == "" {
+		ev.ID = fmt.Sprintf("evt_%d_%d", time.Now().UnixNano(), count)
+	}
+	if ev.Timestamp.IsZero() {
+		ev.Timestamp = time.Now()
+	}
+
+	s.mu.RLock()
+	listeners := make([]Listener, len(s.listeners))
+	copy(listeners, s.listeners)
+	s.mu.RUnlock()
+
+	for _, l := range listeners {
+		if l != nil {
+			l(ev)
+		}
+	}
+}
+
+// EmitPhase 发送阶段性任务状态跃迁事件
+func (s *Streamer) EmitPhase(phaseID, message, role string, stepIndex, totalSteps, progress int) {
+	if s == nil {
+		return
+	}
+	s.Emit(StreamEvent{
+		Type: EventPhase,
+		Phase: &PhasePayload{
+			PhaseID:    phaseID,
+			Message:    message,
+			Role:       role,
+			StepIndex:  stepIndex,
+			TotalSteps: totalSteps,
+			Progress:   progress,
+		},
+	})
+}
+
+// EmitStatus 发送意图与状态机事件
+func (s *Streamer) EmitStatus(intent, branch, currentStage, message string, patches map[string]string) {
+	if s == nil {
+		return
+	}
+	s.Emit(StreamEvent{
+		Type: EventStatus,
+		Status: &StatusPayload{
+			Intent:       intent,
+			Branch:       branch,
+			CurrentStage: currentStage,
+			Patches:      patches,
+			Message:      message,
+		},
+	})
+}
+
+// EmitSummary 发送核心进展摘要事件 (低认知负荷)
+func (s *Streamer) EmitSummary(title, content string, maturityScore int, maturityLevel string) {
+	if s == nil {
+		return
+	}
+	s.Emit(StreamEvent{
+		Type: EventSummary,
+		Summary: &SummaryPayload{
+			Title:         title,
+			Content:       content,
+			MaturityScore: maturityScore,
+			MaturityLevel: maturityLevel,
+		},
+	})
+}
+
+// EmitBriefing 发送 B 轨业务参谋决策锦囊
+func (s *Streamer) EmitBriefing(brief *squad.AdvisorBriefing) {
+	if s == nil || brief == nil {
+		return
+	}
+	s.Emit(StreamEvent{
+		Type:     EventBriefing,
+		Briefing: brief,
+	})
+}
+
+// EmitDetail 发送详细推演草稿或思考链 (按需折叠)
+func (s *Streamer) EmitDetail(category, stepID, role, title, content string, collapsible bool) {
+	if s == nil {
+		return
+	}
+	s.Emit(StreamEvent{
+		Type: EventDetail,
+		Detail: &DetailPayload{
+			Category:    category,
+			StepID:      stepID,
+			Role:        role,
+			Title:       title,
+			Content:     content,
+			Collapsible: collapsible,
+		},
+	})
+}
+
+// EmitDelta 发送对话自然语言流式文本
+func (s *Streamer) EmitDelta(delta string) {
+	if s == nil || delta == "" {
+		return
+	}
+	s.Emit(StreamEvent{
+		Type:  EventDelta,
+		Delta: delta,
+	})
+}
+
+// EmitDeliverable 发送 A 轨合规文档交付事件
+func (s *Streamer) EmitDeliverable(taskKey, title, filePath string, wordCount int, score float64) {
+	if s == nil {
+		return
+	}
+	s.Emit(StreamEvent{
+		Type: EventDeliverable,
+		Deliverable: &DeliverablePayload{
+			TaskKey:      taskKey,
+			Title:        title,
+			FilePath:     filePath,
+			WordCount:    wordCount,
+			QualityScore: score,
+		},
+	})
+}
+
+// EmitError 发送错误警报
+func (s *Streamer) EmitError(err error) {
+	if s == nil || err == nil {
+		return
+	}
+	s.Emit(StreamEvent{
+		Type:  EventError,
+		Error: err.Error(),
+	})
+}
+
+// EmitDone 发送流结束事件
+func (s *Streamer) EmitDone(sessionID string, duration time.Duration) {
+	if s == nil {
+		return
+	}
+	s.Emit(StreamEvent{
+		Type: EventDone,
+		Done: &DonePayload{
+			SessionID:   sessionID,
+			DurationMs:  duration.Milliseconds(),
+			TotalEvents: int(atomic.LoadInt64(&s.eventCount)),
+		},
+	})
+}
+
+// EventCount 获取已发射事件总数
+func (s *Streamer) EventCount() int {
+	if s == nil {
+		return 0
+	}
+	return int(atomic.LoadInt64(&s.eventCount))
+}
+
+// WithStreamer 将 Streamer 注入 Context
+func WithStreamer(ctx context.Context, s *Streamer) context.Context {
+	return context.WithValue(ctx, streamerCtxKey, s)
+}
+
+// WithListener 便捷方法：基于 Listener 创建 Streamer 并注入 Context
+func WithListener(ctx context.Context, includeDetails bool, l Listener) (context.Context, *Streamer) {
+	s := NewStreamer(includeDetails, l)
+	return WithStreamer(ctx, s), s
+}
+
+// FromContext 从 Context 提取 Streamer，若未设置则返回安全空对象，避免空指针崩溃
+func FromContext(ctx context.Context) *Streamer {
+	if ctx == nil {
+		return nil
+	}
+	if s, ok := ctx.Value(streamerCtxKey).(*Streamer); ok && s != nil {
+		return s
+	}
+	return nil
+}

internal/stream/stream_test.go

@@ -0,0 +1,142 @@
+package stream
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/squad"
+)
+
+func TestStreamer_BasicEvents(t *testing.T) {
+	var received []StreamEvent
+	var mu sync.Mutex
+
+	listener := func(ev StreamEvent) {
+		mu.Lock()
+		defer mu.Unlock()
+		received = append(received, ev)
+	}
+
+	streamer := NewStreamer(true, listener)
+
+	streamer.EmitPhase("squad_assembly", "正在组建小队...", "leader", 1, 3, 33)
+	streamer.EmitStatus("ADVANCE_STAGE", "BranchExpectationHit", "V0.5_feasibility", "推进阶段", map[string]string{"market": "CE"})
+	streamer.EmitSummary("初审结论", "产品符合第二类医疗器械判定", 80, "V0.5")
+	streamer.EmitBriefing(&squad.AdvisorBriefing{MaturityScore: 80, MaturityLevel: "V0.5"})
+	streamer.EmitDetail("step_output", "tasks/01_regulatory", "regulatory", "条款梳理", "GB 9706.1 适用", true)
+	streamer.EmitDelta("您好，已为您完成推进。")
+	streamer.EmitDeliverable("feasibility_reg", "注册可行性评估报告", "sandbox/output.md", 2500, 100.0)
+	streamer.EmitError(errors.New("模拟告警"))
+	streamer.EmitDone("sess_123", 500*time.Millisecond)
+
+	mu.Lock()
+	count := len(received)
+	mu.Unlock()
+
+	if count != 9 {
+		t.Fatalf("expected 9 events, got %d", count)
+	}
+
+	// 校验事件类型
+	expectedTypes := []EventType{
+		EventPhase, EventStatus, EventSummary, EventBriefing,
+		EventDetail, EventDelta, EventDeliverable, EventError, EventDone,
+	}
+
+	for i, exp := range expectedTypes {
+		if received[i].Type != exp {
+			t.Errorf("event %d: expected type %s, got %s", i, exp, received[i].Type)
+		}
+	}
+}
+
+func TestStreamer_FilterDetails(t *testing.T) {
+	var received []StreamEvent
+
+	listener := func(ev StreamEvent) {
+		received = append(received, ev)
+	}
+
+	// includeDetails = false: 必须过滤 Detail 事件
+	streamer := NewStreamer(false, listener)
+
+	streamer.EmitPhase("step_executing", "法规专家研判中...", "regulatory", 1, 2, 50)
+	streamer.EmitDetail("step_output", "tasks/01", "regulatory", "庞大推演草稿", "长篇草稿内容...", true)
+	streamer.EmitSummary("精炼结论", "完成研判", 85, "V0.5")
+
+	if len(received) != 2 {
+		t.Fatalf("expected 2 events (Detail filtered out), got %d", len(received))
+	}
+
+	if received[0].Type != EventPhase || received[1].Type != EventSummary {
+		t.Errorf("unexpected event types: %v, %v", received[0].Type, received[1].Type)
+	}
+}
+
+func TestStreamer_ContextPropagation(t *testing.T) {
+	ctx := context.Background()
+
+	// 1. 未注入时 FromContext 返回 nil，测试 nil-safe 调用
+	nilStreamer := FromContext(ctx)
+	if nilStreamer != nil {
+		t.Fatalf("expected nil streamer from empty context")
+	}
+	// nil 调用不崩溃
+	nilStreamer.EmitPhase("test", "msg", "role", 1, 1, 100)
+	nilStreamer.EmitDelta("test")
+
+	// 2. 注入后正常提取
+	var events []StreamEvent
+	ctxWithS, s := WithListener(ctx, true, func(ev StreamEvent) {
+		events = append(events, ev)
+	})
+
+	extracted := FromContext(ctxWithS)
+	if extracted != s {
+		t.Fatalf("extracted streamer did not match original")
+	}
+
+	extracted.EmitPhase("squad_assembly", "组建小队", "leader", 1, 3, 33)
+	if len(events) != 1 {
+		t.Fatalf("expected 1 event, got %d", len(events))
+	}
+}
+
+func TestStreamer_Concurrency(t *testing.T) {
+	var count int64
+	var mu sync.Mutex
+
+	s := NewStreamer(true, func(ev StreamEvent) {
+		mu.Lock()
+		count++
+		mu.Unlock()
+	})
+
+	var wg sync.WaitGroup
+	workers := 10
+	iterations := 50
+
+	for i := 0; i < workers; i++ {
+		wg.Add(1)
+		go func(wID int) {
+			defer wg.Done()
+			for j := 0; j < iterations; j++ {
+				s.EmitPhase("running", "executing step", "worker", j, iterations, 50)
+			}
+		}(i)
+	}
+
+	wg.Wait()
+
+	mu.Lock()
+	total := count
+	mu.Unlock()
+
+	expected := int64(workers * iterations)
+	if total != expected {
+		t.Fatalf("expected %d events, got %d", expected, total)
+	}
+}

internal/stream/types.go

@@ -0,0 +1,92 @@
+package stream
+
+import (
+	"time"
+
+	"chat-agent-system/internal/squad"
+)
+
+// EventType 定义流式事件类型
+type EventType string
+
+const (
+	EventPhase       EventType = "phase"       // 阶段性状态跃迁（组建小队、法规筛选、初稿撰写、门禁审核）
+	EventStatus      EventType = "status"      // 意图识别结果、成熟度状态机更新
+	EventSummary     EventType = "summary"     // 提炼的核心进展摘要（低认知负荷）
+	EventBriefing    EventType = "briefing"    // 贴身业务参谋决策锦囊（B轨赋能）
+	EventDelta       EventType = "delta"       // 对话自然语言逐字/逐句流式吐出
+	EventDetail      EventType = "detail"      // 详细推演草稿或思考链（按需显示，默认折叠）
+	EventDeliverable EventType = "deliverable" // A轨正式合规文档生成通知
+	EventError       EventType = "error"       // 运行异常警报
+	EventDone        EventType = "done"        // 会话流完成
+)
+
+// PhasePayload 阶段性任务状态更新载荷
+type PhasePayload struct {
+	PhaseID    string `json:"phase_id"`              // 如 squad_assembly, knowledge_retrieval, step_executing, quality_gate
+	Message    string `json:"message"`               // 用户友好的自然语言进度说明
+	Role       string `json:"role,omitempty"`        // 角色 (如 regulatory, writer, leader)
+	StepIndex  int    `json:"step_index,omitempty"`  // 当前步骤索引 (如 1)
+	TotalSteps int    `json:"total_steps,omitempty"` // 总步骤数 (如 3)
+	Progress   int    `json:"progress,omitempty"`    // 百分比估算 (0-100)
+}
+
+// StatusPayload 意图与状态机更新载荷
+type StatusPayload struct {
+	Intent       string            `json:"intent,omitempty"`
+	Branch       string            `json:"branch,omitempty"`
+	CurrentStage string            `json:"current_stage,omitempty"`
+	Patches      map[string]string `json:"patches,omitempty"`
+	Message      string            `json:"message,omitempty"`
+}
+
+// SummaryPayload 核心进展摘要载荷 (低认知负荷)
+type SummaryPayload struct {
+	Title         string `json:"title"`
+	Content       string `json:"content"`
+	MaturityScore int    `json:"maturity_score,omitempty"`
+	MaturityLevel string `json:"maturity_level,omitempty"`
+}
+
+// DetailPayload 详细执行与推演过程载荷 (按需显示，默认折叠)
+type DetailPayload struct {
+	Category    string `json:"category"`              // agent_thought, step_output, tool_call, log
+	StepID      string `json:"step_id,omitempty"`     // 步骤标识 (如 tasks/01_regulatory)
+	Role        string `json:"role,omitempty"`        // 对应角色
+	Title       string `json:"title"`                 // 详情标题
+	Content     string `json:"content"`               // 完整推演正文或技术草稿
+	Collapsible bool   `json:"collapsible"`           // 客户端是否应当默认折叠显示
+}
+
+// DeliverablePayload A轨正式合规文档交付载荷
+type DeliverablePayload struct {
+	TaskKey      string  `json:"task_key"`
+	Title        string  `json:"title"`
+	FilePath     string  `json:"file_path"`
+	WordCount    int     `json:"word_count,omitempty"`
+	QualityScore float64 `json:"quality_score,omitempty"`
+}
+
+// DonePayload 流结束载荷
+type DonePayload struct {
+	SessionID   string `json:"session_id,omitempty"`
+	DurationMs  int64  `json:"duration_ms"`
+	TotalEvents int    `json:"total_events"`
+}
+
+// StreamEvent 统一多层级流式事件
+type StreamEvent struct {
+	ID        string    `json:"id"`
+	Type      EventType `json:"type"`
+	Timestamp time.Time `json:"timestamp"`
+
+	Phase       *PhasePayload          `json:"phase,omitempty"`
+	Status      *StatusPayload         `json:"status,omitempty"`
+	Summary     *SummaryPayload        `json:"summary,omitempty"`
+	Briefing    *squad.AdvisorBriefing `json:"briefing,omitempty"`
+	Detail      *DetailPayload         `json:"detail,omitempty"`
+	Delta       string                 `json:"delta,omitempty"`
+	Deliverable *DeliverablePayload    `json:"deliverable,omitempty"`
+	Error       string                 `json:"error,omitempty"`
+	Done        *DonePayload           `json:"done,omitempty"`
+}