diff --git a/docs/superpowers/plans/2026-09-08-streaming-events-and-digestible-ux.md b/docs/superpowers/plans/2026-09-08-streaming-events-and-digestible-ux.md new file mode 100644 index 00000000..b6d30c94 --- /dev/null +++ b/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 提交入库。 diff --git a/docs/superpowers/specs/2026-09-08-streaming-events-and-digestible-ux-design.md b/docs/superpowers/specs/2026-09-08-streaming-events-and-digestible-ux-design.md new file mode 100644 index 00000000..20a4b1a5 --- /dev/null +++ b/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` | **“当前在干嘛、进行到哪了”**
消除黑盒焦虑，建立确定性预期 | **高频刷新、单行即时动态指示**
例如：“正在组建智能体小队（法规专员、撰写专员、审核主管）...”
“法规专家正在进行法规筛选与标准研判...”
“审核主管正在进行质量门禁与引用核验...” | +| **L2: 摘要与结论信息 (Summary & Briefing)** | `summary`, `briefing`, `delta` | **“核心发现了什么、该怎么决策”**
低认知负荷，快速掌握局面 | **精炼成文、突出核心结论**
• **摘要 (Summary)**：1-3 句话概括事实与核心研判
• **决策锦囊 (Briefing)**：B 轨成熟度等级、基准假设、风险雷达、选项 [1]/[2]、沟通话术
• **对话回答 (Delta)**：参谋自然语言解答流式逐句吐出 | +| **L3: 底层执行与推演细节 (Detail / Raw Trace)** | `detail`, `deliverable` | **“具体推演依据、技术证据留底”**
满足专业审查，支持随时深挖 | **按需展开、默认折叠 (Collapsible)**
• 法规专员的细分研判草稿
• 审核专员的修改痕迹与门禁打分明细
• 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 思考过程与各专员步骤草稿收拢在 `
查看专家推演与合规论证过程 (3 步)...
`； + - 仅当用户主动点击时才展开，彻底避免初次打开时的信息过载。 + +### 6.2 CLI 终端界面 +- 默认开启流式进度模式，使用动态转轮或单行覆盖更新当前阶段（`⏳ [组建小队] ...`）； +- 完成后以清爽排版输出【核心摘要】与【决策锦囊】； +- 增加 `--verbose` 开关，仅在用户显式指定时打印每步专员的完整中间推演。 diff --git a/internal/stream/stream.go b/internal/stream/stream.go new file mode 100644 index 00000000..f193c175 --- /dev/null +++ b/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 +} diff --git a/internal/stream/stream_test.go b/internal/stream/stream_test.go new file mode 100644 index 00000000..ab117062 --- /dev/null +++ b/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) + } +} diff --git a/internal/stream/types.go b/internal/stream/types.go new file mode 100644 index 00000000..43788a15 --- /dev/null +++ b/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"` +}