# 架构重构设计文档：引入 Workflow Orchestrator

## 1. 背景与目标
当前系统存在 `api-gateway` 职责过载、业务逻辑分散、状态机隐式且脆弱、前后端状态不同步等核心问题。为了彻底解决这些架构痛点，本设计提出引入 **Workflow Orchestrator Service**，作为系统的“大脑”，负责集中管理业务流程、状态流转与事件协调。

### 核心目标
1.  **解耦 (Decoupling)**: 将业务协调逻辑从 `api-gateway` 剥离，Gateway 回归纯粹的流量入口和连接管理职责。
2.  **状态一致性 (Consistency)**: 建立单一事实来源 (Single Source of Truth)，所有业务状态由 Orchestrator 统一维护并广播。
3.  **细粒度任务编排 (Fine-Grained Orchestration)**: 废除粗粒度的“阶段”概念，转向基于 DAG (有向无环图) 的任务编排。后端只负责执行任务和广播每个任务的状态，前端根据任务状态自由决定呈现逻辑。

## 2. 架构全景图 (Architecture Overview)

### 2.1 服务角色重定义

| 服务 | 现有职责 | **新职责** |
| :--- | :--- | :--- |
| **API Gateway** | 路由, 鉴权, 注册发现, 业务聚合, 流程触发 | 路由, 鉴权, 注册发现, **SSE/WS 代理 (Frontend Proxy)** |
| **Workflow Orchestrator** | *(新服务)* | **DAG 调度**, **任务依赖管理**, **事件广播**, **状态快照** |
| **Data Providers** | 数据抓取, 存库, 发 NATS 消息 | (保持不变) 接收指令 -> 干活 -> 发结果事件 |
| **Report Generator** | 报告生成, 发 NATS 消息 | (保持不变) 接收指令 -> 干活 -> 发进度/结果事件 |
| **Data Processors** | *(新服务类型)* | **数据清洗/转换** (接收上下文 -> 转换 -> 更新上下文) |

### 2.2 数据流向 (Data Flow)

1.  **启动**: 前端 -> Gateway (`POST /start`) -> **Orchestrator** (NATS: `StartWorkflow`)
2.  **调度**: **Orchestrator** 解析模板构建 DAG -> NATS: 触发无依赖的 Tasks (如 Data Fetching)
3.  **反馈**: Executors (Providers/ReportGen/Processors) -> NATS: `TaskCompleted` -> **Orchestrator**
4.  **流转**: **Orchestrator** 检查依赖 -> NATS: 触发下一层 Tasks
5.  **广播**: **Orchestrator** -> NATS: `WorkflowEvent` (Task Status Updates) -> Gateway -> 前端 (SSE)

## 3. 接口与协议定义 (Contracts & Schemas)

需在 `services/common-contracts` 中进行以下调整：

### 3.1 新增 Commands (NATS Subject: `workflow.commands.*`)

```rust
// Topic: workflow.commands.start
#[derive(Serialize, Deserialize, Debug)]
pub struct StartWorkflowCommand {
    pub request_id: Uuid,
    pub symbol: CanonicalSymbol,
    pub market: String,
    pub template_id: String,
}

// 新增：用于手动请求状态对齐 (Reconnect Scenario)
// Topic: workflow.commands.sync_state
#[derive(Serialize, Deserialize, Debug)]
pub struct SyncStateCommand {
    pub request_id: Uuid,
}
```

### 3.2 新增 Events (NATS Subject: `events.workflow.{request_id}`)

这是前端唯一需要订阅的流。

```rust
// Topic: events.workflow.{request_id}
#[derive(Serialize, Deserialize, Debug)]
#[serde(tag = "type", content = "payload")] 
pub enum WorkflowEvent {
    // 1. 流程初始化 (携带完整的任务依赖图)
    WorkflowStarted { 
        timestamp: i64, 
        // 定义所有任务及其依赖关系，前端可据此绘制流程图或进度条
        task_graph: WorkflowDag 
    },
    
    // 2. 任务状态变更 (核心事件)
    TaskStateChanged {
        task_id: String, // e.g., "fetch:tushare", "process:clean_financials", "module:swot_analysis"
        task_type: TaskType, // DataFetch | DataProcessing | Analysis
        status: TaskStatus, // Pending, Scheduled, Running, Completed, Failed, Skipped
        message: Option<String>,
        timestamp: i64
    },
    
    // 3. 任务流式输出 (用于 LLM 打字机效果)
    TaskStreamUpdate {
        task_id: String,
        content_delta: String,
        index: u32 
    },
    
    // 4. 流程整体结束
    WorkflowCompleted {
        result_summary: serde_json::Value,
        end_timestamp: i64
    },
    
    WorkflowFailed {
        reason: String,
        is_fatal: bool,
        end_timestamp: i64
    },
    
    // 5. 状态快照 (用于重连/丢包恢复)
    // 当前端重连或显式发送 SyncStateCommand 时，Orchestrator 发送此事件
    WorkflowStateSnapshot {
        timestamp: i64,
        task_graph: WorkflowDag,
        tasks_status: HashMap<String, TaskStatus>, // 当前所有任务的最新状态
        tasks_output: HashMap<String, Option<String>> // (可选) 已完成任务的关键输出摘要
    }
}

#[derive(Serialize, Deserialize, Debug)]
pub struct WorkflowDag {
    pub nodes: Vec<TaskNode>,
    pub edges: Vec<TaskDependency> // from -> to
}

#[derive(Serialize, Deserialize, Debug)]
pub struct TaskNode {
    pub id: String,
    pub name: String,
    pub type: TaskType,
    pub initial_status: TaskStatus
}

#[derive(Serialize, Deserialize, Debug, PartialEq)]
pub enum TaskType {
    DataFetch,      // 创造原始上下文
    DataProcessing, // 消耗并转换上下文 (New)
    Analysis        // 读取上下文生成新内容
}

#[derive(Serialize, Deserialize, Debug, PartialEq)]
pub enum TaskStatus {
    Pending,    // 等待依赖
    Scheduled,  // 依赖满足，已下发给 Worker
    Running,    // Worker 正在执行
    Completed,  // 执行成功
    Failed,     // 执行失败
    Skipped     // 因上游失败或策略原因被跳过
}
```

### 3.3 调整现有 Messages

*   **`FetchCompanyDataCommand`**: Publisher 变更为 `Workflow Orchestrator`。
*   **`GenerateReportCommand`**: Publisher 变更为 `Workflow Orchestrator`。

## 4. Workflow Orchestrator 内部设计

### 4.1 DAG 调度器 (DAG Scheduler)
每个 `request_id` 对应一个 DAG 实例。

1.  **初始化**: 根据 `TemplateID` 读取配置。
    *   创建 Data Fetch Tasks (作为 DAG 的 Root Nodes)。
    *   创建 Analysis Module Tasks (根据 `dependencies` 配置连接边)。
2.  **依赖检查**:
    *   监听 Task 状态变更。
    *   当 Task A 变成 `Completed` -> 检查依赖 A 的 Task B。
    *   如果 Task B 的所有依赖都 `Completed` -> 触发 Task B。
    *   如果 Task A `Failed` -> 将依赖 A 的 Task B 标记为 `Skipped` (除非有容错策略)。

### 4.2 状态对齐机制 (State Alignment / Snapshot)
为了解决前端刷新或网络丢包导致的状态不一致：

1.  **主动推送快照 (On Connect)**: 
    *   Gateway 在前端建立 SSE 连接时，向 Orchestrator 发送 `SyncStateCommand`。
    *   Orchestrator 收到命令后，将当前内存中的完整 DAG 状态打包成 `WorkflowStateSnapshot` 事件发送。
2.  **前端合并逻辑**:
    *   前端收到 Snapshot 后，全量替换本地的任务状态树。
    *   如果 Snapshot 显示某任务 `Running`，前端恢复 Loading 动画。
    *   如果 Snapshot 显示某任务 `Completed`，前端渲染结果。

### 4.3 容错策略 (Policy)
Orchestrator 需要内置策略来处理非二元结果。
*   **Data Fetch Policy**: 并非所有 Data Fetch 必须成功。可以配置 "At least one data source" 策略。如果满足策略，Orchestrator 将下游的 Analysis Task 依赖视为满足。

## 5. 实施步骤 (Implementation Checklist)

### Phase 1: Contract & Interface
- [x] **Update common-contracts**:
    - [x] Add `StartWorkflowCommand` and `SyncStateCommand`.
    - [x] Add `WorkflowEvent` enum (incl. Started, StateChanged, StreamUpdate, Completed, Failed, Snapshot).
    - [x] Add `WorkflowDag`, `TaskNode`, `TaskType`, `TaskStatus` structs.
    - [x] Update publishers for `FetchCompanyDataCommand` and `GenerateReportCommand`.
    - [x] Bump version and publish crate.

### Phase 2: Workflow Orchestrator Service (New)
- [x] **Scaffold Service**:
    - [x] Create new Rust service `services/workflow-orchestrator-service`.
    - [x] Setup `Dockerfile`, `Cargo.toml`, and `main.rs`.
    - [x] Implement NATS connection and multi-topic subscription.
- [x] **Core Logic - State Machine**:
    - [x] Implement `WorkflowState` struct (InMemory + Redis/DB persistence optional for MVP).
    - [x] Implement `DagScheduler`: Logic to parse template and build dependency graph.
- [x] **Core Logic - Handlers**:
    - [x] Handle `StartWorkflowCommand`: Init DAG, fire initial tasks.
    - [x] Handle `TaskCompleted` events (from Providers/ReportGen): Update DAG, trigger next tasks.
    - [x] Handle `SyncStateCommand`: Serialize current state and emit `WorkflowStateSnapshot`.
- [x] **Policy Engine**:
    - [x] Implement "At least one provider" policy for data fetching.

### Phase 3: API Gateway Refactoring
- [x] **Remove Legacy Logic**:
    - [x] Delete `aggregator.rs` completely.
    - [x] Remove `trigger_data_fetch` aggregation logic.
    - [x] Remove `/api/tasks` polling endpoint.
- [x] **Implement Proxy Logic**:
    - [x] Add `POST /api/v2/workflow/start` -> Publishes `StartWorkflowCommand`.
    - [x] Add `GET /api/v2/workflow/events/{id}` -> Subscribes to NATS, sends `SyncStateCommand` on open, proxies events to SSE.

### Phase 4: Integration & Frontend
- [x] **Docker Compose**: Add `workflow-orchestrator-service` to stack.
- [x] **Frontend Adapter**:
    - [x] **Type Definitions**: Define `WorkflowEvent`, `WorkflowDag`, `TaskStatus` in `src/types/workflow.ts`.
    - [x] **API Proxy**: Implement Next.js Route Handlers for `POST /workflow/start` and `GET /workflow/events/{id}` (SSE).
    - [x] **Core Logic (`useWorkflow`)**:
        - [x] Implement SSE connection management with auto-reconnect.
        - [x] Handle `WorkflowStarted`, `TaskStreamUpdate`, `WorkflowCompleted`.
        - [x] Implement state restoration via `WorkflowStateSnapshot`.
    - [x] **UI Components**:
        - [x] `WorkflowVisualizer`: Task list and status tracking.
        - [x] `TaskOutputViewer`: Markdown-rendered stream output.
        - [x] `WorkflowReportLayout`: Integrated analysis page layout.
    - [x] **Page Integration**: Refactor `app/report/[symbol]/page.tsx` to use the new workflow engine.

---
*Updated: 2025-11-20 - Added Implementation Checklist*
