> For AI agents: the complete documentation index is available at /llms.txt, the full documentation bundle is available at /llms-full.txt.

# Pipeline 全景分析

> **Historical code audit / Non-normative direction（2026-07-15）**：本文是 2026-05-31 的 pipeline snapshot，保留问题发现与 shared mutation 方向；其中旧 `Task Pipeline Deliverable`、`task_progress_events`、global `closeResult`、`create_closed` 和独立 `lead_outreach` Task 不再是 Target。Task 设计见 [Task System Design V3](/product-design/v3/tasks-feature/task-domain-lifecycle.md)；任何 Current 实现结论都必须重新以 live code 核查。
>
> **读者**：产品/工程团队成员、AI Agent — 理解系统所有数据 pipeline 的完整流转、交叉依赖、code/prompt 职责边界，以及从 Task Pipeline 设计出发的统一架构方向。
>
> **数据来源**：2026-05-31 从 `callytics-infrastructure`、`callytics-common`、`lead-tracking` 代码直接验证。**以代码为准**，文档描述如与代码不符已标注。
>
> **与其他文档的关系**：
>
> - [系统架构总览](/architecture/0-system-overview.md) — repo 职责、runtime 部署
> - [Backend 数据管道](/architecture/2-backend.md) — 7 条数据线路的入口/出口/存储描述
> - [Backend 架构 Pattern](/system-design/backend-patterns.md) — Layered vs Hexagonal 选型
> - [Task System Design V3](/product-design/v3/tasks-feature/task-domain-lifecycle.md) — 现行 Task / AI runtime contract
> - [Task Pipeline 设计 Deliverable](/product-design/v2/tasks-feature/design/task-pipeline-deliverable-codex.md) — historical Task Orchestrator 设计依据，不是现行 contract

***

## 一、Pipeline 全景图

系统当前有 **5 条活跃数据 pipeline** + 2 条辅助 pipeline。每条 pipeline 的触发、处理、写入、下游消费路径如下。

```mermaid
%%{init: {'theme': 'base', 'themeVariables': {'primaryColor': '#e8eaf6', 'lineColor': '#666'}}}%%
flowchart TB
    subgraph ingress["Ingress 层"]
        rc["RC Webhook<br/>(ringcentralSubscriptionService)"]
        imap["IMAP 轮询<br/>(lead-tracking)"]
    end

    subgraph queues["SQS 队列层"]
        tq["transcribe-queue<br/>(5min 延迟)"]
        mq["message-processing-queue"]
        aiq["ai-analysis-queue"]
        dbq["dailyBatchQueue.fifo<br/>(batchSize=1, maxConcurrency=2)"]
        lpq["lead-processor-queue<br/>(EventBridge → SQS)"]
    end

    subgraph p1["Call Analysis"]
        tp["transcribe-processor<br/>(concurrency=2)"]
        aip["ai-analysis-processor<br/>(concurrency=20)"]
    end

    subgraph p2["Contact Analysis"]
        ca["contacts-analyzer<br/>(dispatch + SQS worker)"]
    end

    subgraph p3["SMS Processing"]
        mp["message-processor"]
    end

    subgraph p4["Lead Processing"]
        lp["lead-processor"]
    end

    subgraph storage["Storage 层"]
        neon["Neon PostgreSQL<br/>calls · contacts · messages<br/>leads · tasks · contact_timeline"]
        ddb["DynamoDB<br/>call-analysis · LeadTracking-v2"]
        s3["S3<br/>录音 · MMS · config"]
    end

    rc -->|call event| tq
    rc -->|SMS event| mq
    imap -->|LeadCreated| lpq

    tq --> tp
    tp -->|S3 录音 + Deepgram 转录| s3
    tp --> aiq
    aiq --> aip
    aip -->|"3-stage AI:<br/>triage → classify → coaching"| neon
    aip --> ddb
    aip -->|"per-call fan-out<br/>(follow_up_needed=yes)"| dbq

    mq --> mp
    mp -->|"db.batch() 原子写"| neon

    lpq --> lp
    lp -->|"db.batch() 原子写<br/>contacts + tasks + timeline"| neon

    dbq --> ca
    ca -->|"contact-analyzer AI<br/>taskDecisions[]"| neon

    style p1 fill:#e3f2fd
    style p2 fill:#fff3e0
    style p3 fill:#fce4ec
    style p4 fill:#e8f5e9
```

***

## 二、逐条 Pipeline 详解

### Call Analysis（通话分析）

```mermaid
%%{init: {'theme': 'base', 'themeVariables': {'primaryColor': '#e3f2fd', 'lineColor': '#666'}}}%%
flowchart LR
    rc["RC Webhook"] --> tq["SQS transcribe\n(5min delay)"]
    tq --> tp["transcribe-processor\n(concurrency=2)"]
    tp --> s3["S3 录音"]
    tp --> aiq["SQS ai-analysis"]
    aiq --> aip["ai-analysis-processor\n(concurrency=20)"]
    aip --> |"Pre-Triage → Triage\n→ Classify → Coaching"| neon["Neon\ncalls · contacts\ntimeline"]
    aip --> ddb["DDB\ncall-analysis"]
    aip -->|"follow_up\n=yes"| dbq["dailyBatchQueue\n→ Contact Analysis"]

    style tp fill:#e3f2fd
    style aip fill:#e3f2fd
```

**触发**：RC Webhook → `ringcentralSubscriptionService` → SQS `transcribe-queue`（5 分钟延迟等录音就绪）

**阶段链**：

| 阶段    | Lambda                                   | 输入                                | 处理                          | 输出/写入                                                                         |
| ----- | ---------------------------------------- | --------------------------------- | --------------------------- | ----------------------------------------------------------------------------- |
| 转录    | `transcribe-processor` (concurrency=2)   | SQS message（callSessionId + 租户信息） | 下载 RC 录音 → S3 → Deepgram 转录 | S3 录音文件、DDB call-events、SQS → ai-analysis-queue                               |
| AI 分析 | `ai-analysis-processor` (concurrency=20) | SQS 转录结果                          | 3-stage prompt pipeline（见下） | Neon calls/contacts/contact\_timeline、DDB call-analysis、SQS → dailyBatchQueue |

**3-Stage AI Prompt Pipeline**（`ai-analysis-processor/src/core/stages/`）：

| Stage          | Prompt                   | 输入                                | 输出                                                                                                               | 职责边界                                                                                                   |
| -------------- | ------------------------ | --------------------------------- | ---------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------ |
| Pre-Triage     | 无（纯代码规则）                 | 原始 transcript 文本                  | `PreTriageResult`（callState, staffName）或 null                                                                    | **代码闸门** — 关键词匹配（"mailbox is full"、"please leave your message" 等），节省 \~40% triage AI 调用。0 tokens、\<1ms |
| Triage         | `TRIAGE_SYSTEM_PROMPT`   | 通话前 500 字符 + 员工名单 + 通话时长          | `call_state`（human\_conversation/voicemail/no\_answer/system\_error/busy\_signal）、`staff_name`、`worth_analyzing` | **AI 闸门** — 决定这通电话要不要继续分析。不触发 task、不改 lifecycle                                                        |
| Classification | `CLASSIFY_SYSTEM_PROMPT` | 完整 transcript + business context  | `category`/`subcategory`、`customer_type`、`outcome`、`follow_up.needed` + reason 码                                 | **单通分析** — 不创建/关闭 task；`follow_up.needed` 是下游 contact-analyzer 的触发信号之一                                 |
| Coaching       | `COACHING_SYSTEM_PROMPT` | 完整 transcript + classification 结果 | 员工话术反馈（coaching notes）                                                                                           | **话术评估** — 不碰 task、不改 lifecycle、不改 lead status                                                         |

**写入表（代码验证）**：

| 表                       | 写入方式                                                                  | 关键字段                                                                         |
| ----------------------- | --------------------------------------------------------------------- | ---------------------------------------------------------------------------- |
| Neon `calls`            | UPSERT（transcribe-processor 写基础字段，ai-analysis-processor 补 33 个 AI 字段） | `store_id`、`account_id`、`call_state`、`category`、`outcome`、`follow_up_needed` |
| Neon `contacts`         | UPSERT（更新 `lastActivityAt`、AI 字段）                                     | `store_id`、`lifecycle_stage`、`lead_status`                                   |
| Neon `contact_timeline` | INSERT                                                                | 通话事件审计记录                                                                     |
| DDB `call-analysis`     | PUT（legacy dual-write）                                                | 完整 AI 分析结果                                                                   |

**下游 fan-out**：当 `follow_up_needed=yes` 时，向 `dailyBatchQueue.fifo` 发 SQS 消息，触发 Contact Analysis 对该联系人重新做跨通话分析。

**架构 pattern**：**Hexagonal** — `handler.ts` 是 composition root，`core/stages/` 含 pipeline use case，`core/protocols.ts` 定义 ports，`infrastructure/` 含 AI/Neon/DDB/S3/SQS adapters。这是系统中最成熟的架构实现。

***

### Contact Analysis（联系人跨通话分析 + Task 生命周期）

```mermaid
%%{init: {'theme': 'base', 'themeVariables': {'primaryColor': '#fff3e0', 'lineColor': '#666'}}}%%
flowchart LR
    cron["EventBridge\n06:00 cron"] --> dbq["dailyBatchQueue\n(FIFO)"]
    p1["Call Analysis\nfollow_up=yes"] --> dbq
    ui["studio-api\non-demand"] --> dbq
    dbq --> ca["contacts-analyzer"]
    ca --> |"聚合 calls+msgs\n+leads+tasks"| ai["AI 分析\n18 字段 +\ntaskDecisions[]"]
    ai --> neon["Neon\ncontacts · tasks\ntimeline"]

    style ca fill:#fff3e0
```

**触发方式**（3 种）：

| 触发源                    | 事件路径                                                                                  | SQS message body                                                                       |
| ---------------------- | ------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------- |
| 每日 cron                | EventBridge 06:00 UTC → contacts-analyzer dispatch path → 分页扫描活跃联系人 → dailyBatchQueue | `{phone, storeId, source: 'cron'}`                                                     |
| Call Analysis per-call | ai-analysis-processor `follow_up_needed=yes` → SQS dailyBatchQueue                    | `{phone, storeId, source: 'per_call_analysis', telephonySessionId, webhookReceivedAt}` |
| 员工 On-Demand Refresh   | studio-api endpoint → SQS dailyBatchQueue                                             | `{phone, storeId, source: 'on_demand'}`                                                |

**处理流程**（`handler.ts` 第 154-331 行）：

```
1. 从 Neon 读联系人行（phone + storeId PK）
2. 并行聚合：getRecentCalls() + getRecentMessages() + getRecentLeads()
3. 从 Neon 读 pending + closed tasks（按 phone + storeId 隔离）
4. buildUserMessage() 组装 prompt 输入
5. AI 调用（OneRouter/Grok 4.1）→ 输出 ContactsAnalysisSchema
6. writeAnalysisWithTasks() 原子写入
```

**AI Prompt 输出**（`core/prompt-builder.ts`）：

| 输出区域                                | 内容                                                                                                                       | 下游消费                             |
| ----------------------------------- | ------------------------------------------------------------------------------------------------------------------------ | -------------------------------- |
| Contacts 字段                         | `lifecycleStage`、`leadStatus`、`customerSummary`、`riskSignals` 等 18 个字段                                                   | Neon contacts 表                  |
| `taskDecisions[]`                   | 每个 decision 含 `action`（CREATE/UPDATE/CLOSE/NO\_CHANGE）+ `typeCategory` + `priority` + `closeResult` + `suggestedActions` | Neon tasks + contact\_timeline 表 |
| `actionNeeded` + `suggestedActions` | 是否需要人工跟进 + 具体建议                                                                                                          | Tasks UI 展示                      |

**Task 生命周期管理**（代码路径：`infrastructure/neon-repository.ts:writeAnalysisWithTasks()`）：

| 操作        | 条件                             | 代码行为                                                                                             |
| --------- | ------------------------------ | ------------------------------------------------------------------------------------------------ |
| CREATE    | 有新的未解决目标 + 无同类 pending task    | INSERT tasks + INSERT contact\_timeline                                                          |
| UPDATE    | 同一目标仍开放 + 新证据改变 priority/dueAt | UPDATE tasks (priority, suggestedActions, dueAt)                                                 |
| CLOSE     | 目标完成/失效/DNC/wrong-number       | UPDATE tasks (status='closed', closeResult, closeType='auto\_closed') + INSERT contact\_timeline |
| NO CHANGE | 新事件无有用 task 信息                 | 不操作                                                                                              |

**写入表**：

| 表                       | 写入内容                                                     |
| ----------------------- | -------------------------------------------------------- |
| Neon `contacts`         | 18 个 AI 分析字段 + `lastContactAnalysisAt`                   |
| Neon `tasks`            | CREATE/UPDATE/CLOSE 操作                                   |
| Neon `contact_timeline` | task 生命周期事件（`task.created`、`task.closed`、`task.updated`） |

**架构 pattern**：**Semi-hexagonal** — 有 `infrastructure/protocols.ts` 定义 ports（AIClient、ContactsReader、ContactsWriter），有 `infrastructure/neon-repository.ts` 和 `infrastructure/ai-client.ts` 作 adapters。但 `core/` 只有 `models.ts` + `prompt-builder.ts`，没有显式的 pipeline/use-case 层。业务逻辑主要在 `handler.ts:analyzeContact()` 和 `neon-repository.ts:writeAnalysisWithTasks()` 里。

***

### SMS Processing（短信处理）

```mermaid
%%{init: {'theme': 'base', 'themeVariables': {'primaryColor': '#fce4ec', 'lineColor': '#666'}}}%%
flowchart LR
    rc["RC Webhook"] --> mq["SQS message\nprocessing"]
    mq --> mp["message-processor"]
    mp --> s3["S3 MMS 附件"]
    mp --> |"db.batch()\n原子写"| neon["Neon\nmessages · contacts\ntimeline"]

    style mp fill:#fce4ec
```

**触发**：RC Webhook → `ringcentralSubscriptionService` → SQS `message-processing-queue`

**处理**（`message-processor/src/core/message-processing.ts`）：

1. 从 RC API fetch message 详情
2. S3 保存 MMS 附件
3. 原子 `db.batch()` 写入 Neon

**写入表**：

| 表                       | 写入内容                                       |
| ----------------------- | ------------------------------------------ |
| Neon `messages`         | SMS/VM 消息记录（direction、content、attachments） |
| Neon `contacts`         | UPSERT（更新 `lastActivityAt`）                |
| Neon `contact_timeline` | `message.received` / `message.sent` 事件     |

**关键特征**：

- **不调用 AI** — 纯数据持久化管道
- **不触发下游 pipeline** — 不向 dailyBatchQueue 发消息
- SMS 内容要等到 Contact Analysis 的每日 cron 或 per-call 触发时才会被 AI 分析

**架构 pattern**：**Hexagonal（轻量）** — 有 `core/` + `infrastructure/` 分层，但没有 AI 相关 ports。

***

### Lead Processing（线索处理）

```mermaid
%%{init: {'theme': 'base', 'themeVariables': {'primaryColor': '#e8f5e9', 'lineColor': '#666'}}}%%
flowchart LR
    imap["IMAP 轮询\n(lead-tracking)"] --> eb["EventBridge\nLeadCreated"]
    eb --> lpq["SQS lead\nprocessor"]
    lpq --> lp["lead-processor"]
    lp --> |"db.batch()\ncontacts + task\n+ timeline"| neon["Neon"]

    style lp fill:#e8f5e9
```

**触发**：`lead-tracking` Lambda（IMAP 轮询邮箱）→ EventBridge `LeadCreated` → SQS → `lead-processor`

**处理**（`lead-processor/src/core/persist-downstream.ts`）：

单个原子 `db.batch()` 写入 3 个实体：

```typescript
await client.batch([
  // 1. contacts UPSERT — lifecycleStage='lead', leadStatus='new'
  client.insert(contacts).values({...}).onConflictDoUpdate({...}),
  // 2. tasks INSERT — lead_outreach, high priority, 5-min SLA
  client.insert(tasks).values({
    taskType: 'lead_outreach',
    typeCategory: 'lead_outreach',
    sourceType: 'lead',
    priority: 'high',
    dueAt: computeDueAt('high', now, { slaMinutes: 5 }),
    suggestedActions: [{ action: 'Call the lead back promptly', ... }],
  }).onConflictDoNothing(),
  // 3. contactTimeline INSERT — lead.created 事件
  client.insert(contactTimeline).values({...}).onConflictDoNothing(),
]);
```

**关键特征**：

- **不调用 AI** — Task 创建是硬编码的确定性规则（5 分钟 SLA、high priority）
- **直接写 tasks 表** — 不经过任何 Task Orchestrator 或 contacts-analyzer
- **`suggestedActions` 是模板** — 不是 AI 生成的，是代码里写死的默认建议
- `onConflictDoNothing` — 通过 `uq_tasks_source_lead` 唯一索引防重复

**架构 pattern**：**Thin handler + core function** — `handler.ts` 做 SQS 解析和 Neon client 管理，`core/persist-downstream.ts` 是纯业务逻辑（no AWS SDK import）。

***

### Analytics / Periodic（定时报告和监控）

| Lambda                        | 触发                                | 功能                                      | 技术栈             |
| ----------------------------- | --------------------------------- | --------------------------------------- | --------------- |
| `analytics-generator`         | EventBridge daily 7:00 AM ET      | 汇总 call-analysis 数据 → PDF/HTML 报告 → 邮件  | **Python 3.13** |
| `force-refresh`               | 手动触发                              | 手动重算分析数据                                | Python 3.12     |
| `storeid-coverage-monitor`    | EventBridge hourly（仅 test）        | UNION ALL 查 6 张 Neon 表 NULL storeId 覆盖率 | TypeScript      |
| `reconciliation-orchestrator` | EventBridge every 3h（当前 DISABLED） | 扫描 DDB 找遗漏/失败的通话 → reconciliation-queue | TypeScript      |
| `reconciliation-worker`       | SQS reconciliation-queue          | 逐条修复 → 重新注入 transcribe-queue            | TypeScript      |

***

## 三、跨 Pipeline 依赖矩阵

### 3.1 共享 Neon 表 — 多 Writer 冲突风险

| 表                      | Writer 1                                          | Writer 2                                          | Writer 3                                                                                                                           | 冲突风险                                                                                                                                                    |
| ---------------------- | ------------------------------------------------- | ------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **`contacts`**         | transcribe-processor（RC metadata + name trust=60） | ai-analysis-processor（AI 字段 + name trust=40/80）   | message-processor（`lastActivityAt`）、contacts-analyzer（18 个 AI 字段）、lead-processor（UPSERT, trust=80）、studio-api（staff 手动, trust=100） | **中** — 6 个 writer 各写不同字段集。name trust scoring（WEBHOOK=20 \< AI\_TRANSCRIPT=40 \< RC\_API=60 \< LEAD=80 \< STAFF=100）分散在 5 个 Lambda 代码库，没有共享 enforcement |
| **`contact_timeline`** | ai-analysis-processor                             | message-processor                                 | contacts-analyzer、lead-processor                                                                                                   | **低** — INSERT-only（append-only audit log），`onConflictDoNothing` 用 idempotencyKey 去重                                                                    |
| **`tasks`**            | lead-processor（lead\_outreach CREATE）             | contacts-analyzer（follow\_up CREATE/UPDATE/CLOSE） | studio-api（manual CREATE/CLOSE/REOPEN）                                                                                             | **中** — 三个独立入口，各自用不同的去重策略（`uq_tasks_source_lead` / `effectivePendingCategories` code check / 无去重）。没有统一的 mutation 入口                                     |
| **`calls`**            | transcribe-processor（基础字段）                        | ai-analysis-processor（33 个 AI 字段）                 | —                                                                                                                                  | **低** — 两步 UPSERT 是设计好的顺序（先转录再 AI），不会并发                                                                                                                 |

### 3.2 SQS 队列 — 生产/消费关系

| 队列                         | 生产者                                                                                           | 消费者                      | 类型       |
| -------------------------- | --------------------------------------------------------------------------------------------- | ------------------------ | -------- |
| `transcribe-queue`         | ringcentralSubscriptionService、reconciliation-worker                                          | transcribe-processor     | Standard |
| `ai-analysis-queue`        | transcribe-processor（via S3 Object Created → EventBridge）                                     | ai-analysis-processor    | Standard |
| `message-processing-queue` | ringcentralSubscriptionService                                                                | message-processor        | Standard |
| `dailyBatchQueue.fifo`     | ai-analysis-processor（per-call fan-out）、contacts-analyzer dispatch path、studio-api（on-demand） | contacts-analyzer worker | FIFO     |
| `lead-processor-queue`     | lead-tracking（via EventBridge）                                                                | lead-processor           | Standard |
| `reconciliation-queue`     | reconciliation-orchestrator                                                                   | reconciliation-worker    | Standard |

### 3.2.1 Onboarding historical backfill 的正确位置

新店 onboarding 需要“把过去 14 天数据导入系统”时，不应该新写一条绕过现有 pipeline 的 ETL。

```text
Calls:
operator / control-plane
  -> reconciliation-queue
  -> reconciliation-worker
  -> RingCentral call-log
  -> synthetic webhook
  -> transcribe-queue
  -> transcribe-processor
  -> ai-analysis-processor
  -> contacts-analyzer

Messages:
operator / control-plane
  -> message historical backfill job（待补）
  -> RingCentral message-store
  -> existing message-processor write path
```

Calls 已经有可复用的机制：`reconciliation-worker` 支持 `reconciliation_window.startTime/endTime`，检测 missing calls 后把 synthetic webhook 投进 `transcribe-queue`。Onboarding 只需要一个更清晰的 operator trigger。

Messages 目前还没有同级的 historical backfill job。`studio-api` 能按 `storeId/dateFrom/dateTo` 读取 RingCentral `message-store`，`message-processor` 能持久化 message-store webhook，但中间缺“按日期拉历史消息并复用 message-processor 写入”的 job（跟踪：retaintive/callytics-infrastructure#1156）。

### 3.3 Pipeline 交叉点

```mermaid
%%{init: {'theme': 'base', 'themeVariables': {'primaryColor': '#e8eaf6', 'lineColor': '#666'}}}%%
flowchart LR
    P1["Call Analysis"] -->|"follow_up_needed=yes<br/>→ dailyBatchQueue"| P2["Contact Analysis"]
    P2 -->|"读 P1 产出的<br/>calls + call-analysis"| P1
    P3["SMS Processing"] -->|"P2 cron 读<br/>messages 表"| P2
    P4["Lead Processing"] -->|"P2 cron 读<br/>leads 表"| P2

    P1 -->|"写 contacts"| shared["contacts 表<br/>(4 writer)"]
    P2 -->|"写 contacts"| shared
    P3 -->|"写 contacts"| shared
    P4 -->|"写 contacts"| shared

    P2 -->|"写 tasks"| tasks["tasks 表<br/>(2 writer)"]
    P4 -->|"写 tasks"| tasks
```

**核心交叉点**：

1. **P1 → P2 的 per-call fan-out** — ai-analysis-processor 通过 `follow_up_needed=yes` 触发 contacts-analyzer 重新分析
2. **P2 读取所有其他 pipeline 的产出** — contacts-analyzer 聚合 calls（P1 产出）+ messages（P3 产出）+ leads（P4 产出）做跨维度 AI 分析
3. **contacts 表是 4 个 pipeline 的共享写入点** — 是系统中 writer 最多的表
4. **tasks 表有 2 个独立写入入口** — lead-processor 直接创建 lead\_outreach，contacts-analyzer 管理 follow\_up 生命周期

***

## 四、Code vs Prompt 职责边界（现状）

### 4.1 各 Pipeline 的职责分配

| Pipeline                | Code 负责                                                                                   | AI/Prompt 负责                                                   | 边界清晰度                                                                                                                  |
| ----------------------- | ----------------------------------------------------------------------------------------- | -------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------- |
| **P1 Call Analysis**    | 转录调度、录音下载、S3 存储、DB 写入、SQS fan-out、重试/幂等                                                   | Triage（闸门）、Classification（语义分类）、Coaching（话术评估）                 | **清晰** — Hexagonal 架构，core/ports/adapters 分层明确                                                                         |
| **P2 Contact Analysis** | 联系人分页扫描、数据聚合（calls+messages+leads+tasks）、task mutation 执行（CREATE/UPDATE/CLOSE）、幂等/去重/并发控制 | 客户画像分析、task 决策提案（`taskDecisions[]`）、priority 判断、closeResult 判断 | **中等** — AI 提案 + code 执行的分工存在，但 `writeAnalysisWithTasks()` 里的 task mutation 逻辑散落在 neon-repository 里，不是独立的 Orchestrator |
| **P3 SMS**              | 全部（RC API fetch、MMS 存储、DB 原子写入）                                                           | 无 AI 介入                                                        | **N/A** — 纯 code pipeline                                                                                              |
| **P4 Lead**             | 全部（IMAP 解析、task 创建、contact UPSERT）                                                        | 无 AI 介入                                                        | **N/A** — 纯 code pipeline，task 创建是硬编码规则                                                                                |

### 4.2 Task Pipeline Deliverable 提出的理想模式

Deliverable 定义了一个 **8 步 responsibility chain**（第 2 节）：

```
Input event → Filter layer (code) → Task Orchestrator (code) → Read business context
→ AI semantic judgment → Code execution layer → DB transaction → Downstream consumption
```

核心原则：**"AI proposes, code executes."** AI 只输出 `taskDecisions[]` 提案，所有状态变更由 Orchestrator 代码验证并执行。

### 4.3 现状 vs 理想的差距

| 维度                      | Deliverable 理想                                                                          | 现状                                                                                                                               | Gap                                    |
| ----------------------- | --------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------- |
| **统一入口**                | 所有 task mutation 经过 Task Orchestrator                                                   | lead-processor 直接写 tasks 表，contacts-analyzer 通过 `writeAnalysisWithTasks()` 写                                                     | **Task 写入有 2 个独立入口，没有统一 Orchestrator** |
| **Filter layer**        | 代码层在 AI 之前做 dedup、merge、trivial 过滤                                                      | P1 有 Triage 闸门（但是 AI 做的不是代码做的）；P3 SMS 没有 AI 过滤                                                                                   | **Filter 逻辑混在 AI prompt 里而不是代码层**      |
| **progress vs outcome** | `no_answer`/`left_voicemail` 是 progress event（新表 `task_progress_events`），不是 closeResult | 当前 closeResult 枚举 18 个值混合了 progress（`no_answer`、`left_voicemail`、`callback_later`）和 business outcome（`converted`、`cancel_saved`） | **进展和结果没分离，closeResult 枚举过载**          |
| **create\_closed 路径**   | 支持 "当场解决" 的事（来电预约、来电投诉当场解决） → 创建后立即关闭                                                   | closeType 只有 `auto_closed` + `manual_closed`，没有 `create_closed`                                                                  | **无法记录 "来了就完成" 的工作**                   |

***

## 五、Schema 健康检查

### 5.1 Task 枚举对齐状态

| 枚举                   | Schema 定义                          | Prompt 实际使用                                                                              | V1 文档描述             | 状态                      |
| -------------------- | ---------------------------------- | ---------------------------------------------------------------------------------------- | ------------------- | ----------------------- |
| `TASK_TYPE_CATEGORY` | 9 个值                               | 8 个（排除 `lead_outreach`，由 lead-tracking 创建）                                               | 12 个场景标签（命名/粒度不一致）  | ⚠️ V1 文档 stale          |
| `TASK_CLOSE_RESULT`  | 18 个值（代码）                          | 13 个（prompt 不知道 `booked`/`cancelled`/`already_member`/`callback_later`/`left_voicemail`） | V1 文档写"11 个"或"13 个" | ⚠️ Prompt vs Schema 不同步 |
| `TASK_CLOSE_TYPE`    | 2 个（`auto_closed`/`manual_closed`） | Deliverable 提议加 `create_closed`                                                          | —                   | 🔲 待实现                  |
| `TASK_STATUS`        | 2 个（`pending`/`closed`）            | Deliverable 建议语义映射 `pending` → `open`                                                    | —                   | 设计决策                    |

### 5.2 关键字段覆盖

| 字段                                     | 定义位置                                   | 写入方                               | 问题                                       |
| -------------------------------------- | -------------------------------------- | --------------------------------- | ---------------------------------------- |
| `tasks.executorType`                   | Deliverable 提议（human/ai\_agent/system） | 尚未创建                              | 🔲 新字段                                   |
| `tasks.attemptCount`                   | Deliverable 提议（反规范化计数器）                | 尚未创建                              | 🔲 新字段（需 `task_progress_events` 表）       |
| `task_progress_events`                 | Deliverable 提议（新表）                     | 尚未创建                              | 🔲 新表 — 承载从 closeResult 中剥离的 progress 事件 |
| `contacts.lastContactAnalysisAt`       | Schema 已定义                             | contacts-analyzer 写入              | ✅ 正常                                     |
| `tasks.storePhone`                     | Schema 已定义                             | contacts-analyzer 写入（审计字段）        | ✅ 正常                                     |
| `closeResult` = `booked` / `cancelled` | Schema 已加入（infra #164）                 | ❌ 前端 dropdown 缺失、prompt 不知道这 2 个值 | ⚠️ Schema/UI/Prompt 三方不同步                |

***

## 六、Gap Analysis — 从 Task Pipeline 推广到全系统

### Gap 0(P0)~~:NeonRetryProcessor DLQ retry 不发 LeadCreated 事件~~ ✅ 已修复(2026-06-08,lead-tracking#201)

**原现状**:`lead-tracking` 的 Neon 写入失败时进 DLQ,`NeonRetryProcessor` 负责 retry。但 retry 成功后**不发 EventBridge `LeadCreated` 事件**。

**原后果**:DLQ retry 成功的 leads 在 Neon `leads` 表里存在,但**永远没有下游 contacts/tasks/timeline**。

**修复**(lead-tracking#201 PR):

- 抽 `publishLeadCreatedEvent` 到共享 module `src/lead-created-event.ts`,poller 和 NeonRetryProcessor 共用同一份事件契约
- NeonRetryProcessor `persistLeadPipeline` 成功后调用 `publishLeadCreatedEvent(row, tenantId)`,使用 persistLeadPipeline 返回的 tenantId 避免下游再查 control-plane
- 显式检查 `PutEventsCommand` 返回的 `FailedEntryCount > 0`(AWS SDK 在 200 OK + per-entry failure 时不抛,会静默丢失事件)
- 给 NeonRetryProcessor Lambda 加 `events:PutEvents` IAM grant(之前只有 poller 有)
- 区分 stage-tagged log:Neon 失败 vs EventBridge publish 失败,操作人员能清晰判断哪一步出问题
- 详细写入流程见 [lead-tracking 写入流程](/system-design/write-matrix/lead-tracking-writes.md)

### Gap 1：Task 写入缺乏统一 Orchestrator

**现状**（3 个独立入口）：

- **lead-processor** — 直接 INSERT tasks（lead\_outreach），hardcoded 5-min SLA、hardcoded suggestedActions，去重靠 `uq_tasks_source_lead` unique index
- **contacts-analyzer** — `writeAnalysisWithTasks()` 处理 AI 产出的 CREATE/UPDATE/CLOSE，去重靠 `effectivePendingCategories` code check + DB unique constraint
- **studio-api** — HTTP API 直接写（manual CREATE/CLOSE/REOPEN），无 AI 判断，无共享 state machine

**行为不一致**：

| 行为          | lead-processor                          | contacts-analyzer                      | studio-api                |
| ----------- | --------------------------------------- | -------------------------------------- | ------------------------- |
| 去重          | `uq_tasks_source_lead` index            | code check + DB constraint             | 无                         |
| closeResult | 不 close                                 | AI 输出, 代码校验                            | UI dropdown（含 progress 值） |
| dueAt 计算    | `computeDueAt(high, now, slaMinutes=5)` | `computeDueAt(priority)`               | 员工直接设                     |
| timeline 审计 | `lead.created`                          | `task.created` / `task.status_changed` | `task.status_changed`     |

**建议方向**：Deliverable 的 Task Orchestrator 应该成为 **唯一的 task mutation 入口**。lead-processor 调 Orchestrator 的 `createTask(type='lead_outreach', source='lead', ...)` 而不是直接 INSERT。lead-processor 的 deterministic create 可以作为 Orchestrator 的 **bypass-AI 模式**。

### Gap 2：SMS Pipeline 没有 AI 分析

**现状**：message-processor 只做数据持久化，不做任何语义分析。SMS 内容要等 Contact Analysis 的 cron（最多 24 小时后）才会被 AI 看到。

**问题**：

- 客户发 "STOP" → 应立即标 DNC + 关闭所有 pending tasks → 现在要等到次日凌晨
- 客户 SMS 回复表达高意向 → 应立即触发 contacts-analyzer → 现在没有 per-message fan-out

**建议方向**：在 message-processor 末尾加一步 **Code Filter**（不是 AI）：

- 检测 "STOP"/"UNSUBSCRIBE" → 立即写 DNC + 触发 task close
- 检测有意义回复（非自动回复、非单字） → 向 dailyBatchQueue 发 SQS 触发 Contact Analysis

### Gap 3：`contact_timeline` 缺乏统一写入标准

**现状**：4 个 pipeline 各自写 contact\_timeline，event\_type 命名、actor\_type 取值、newValue 结构都是各 pipeline 自行定义。

**问题**：没有中央的 "timeline event catalog"，新增 event type 时容易命名不一致或遗漏字段。

**建议方向**：在 `callytics-common` 里定义 timeline event schema（event\_type 枚举 + 每种 event 的 required fields），各 pipeline 的 timeline 写入通过 `buildTimelineValues()` helper 强制走 schema validation。

### Gap 4：contacts 表 name trust scoring 分散，storeId null 策略不统一

**Name Trust 现状**：contacts 表的 `firstName`/`lastName` 由 6 个 writer 写入，各自带不同的信任分数：

```
WEBHOOK=20 < AI_TRANSCRIPT=40 < RC_API=60 < LEAD=80 < STAFF=100
```

这个 trust scoring 逻辑（SQL `CASE WHEN` + `GREATEST`）**分散在 5 个 Lambda 代码库中**，没有共享的 enforcement layer。如果某个 pipeline 的 UPSERT 忘了带 trust guard，低信任来源可以覆盖高信任来源的姓名。

**storeId null 现状**：4 个 pipeline 对 null storeId 的行为不一致：

| Pipeline             | null storeId 行为                     |
| -------------------- | ----------------------------------- |
| transcribe-processor | 跳过整个 `persistCallBatch`             |
| message-processor    | 跳过 contacts + timeline（只写 messages） |
| contacts-analyzer    | 继续处理但 `logger.warn`                 |
| lead-processor       | 从 StoresV2 DDB 查 fallback           |

**建议**：写一份 `store-id-null-policy.md` 明确每个 pipeline 的策略，并在 `storeid-coverage-monitor` 增加 per-pipeline breakdown 维度。

### Gap 5：Prompt 输入不完整

**现状**：contacts-analyzer prompt 的 PENDING TASKS 输入只包含 taskId、typeCategory、priority、dueAt、first suggested action。**不包含**：

- 员工手动调整的 dueAt 变更历史（`dueAtChangelog`）
- task progress events（Deliverable 提议的新表，当前不存在）
- 来自其他 pipeline 的最新 SMS 内容（如果距离 cron 运行时刚收到）

**问题**：AI 做 task 决策时缺少关键上下文，导致可能重复创建已经在跟进的 task，或覆盖员工手动调整的 dueAt。

### Gap 6：Per-Call AI（P1）和 Contact-Level AI（P2）的职责边界需要重新审视

**现状**：

- P1 Classification 输出 `follow_up.needed` + reason 码 → 是 P2 的**触发信号**
- P1 Classification 输出 `outcome.result`（booked/cancelled/pending\_follow\_up）→ 是 P2 做 task 决策的**证据之一**
- P1 不直接操作 task → 这是设计意图（"单通信息不足以判断 task"）

**问题**：P1 的 `follow_up.needed=yes` 触发 P2 重新分析，但 P2 的 AI 还需要自己重新读一遍那通电话的内容来做判断。这意味着同一通电话被 AI 分析了 **两次**（P1 一次 + P2 一次），且 P2 的分析范围更大（全量 calls + messages + leads）。

**这不一定是问题** — P1 和 P2 的分析粒度不同（单通 vs 跨通话），两次分析的目的不同。但 **token 成本可以优化**：P2 可以直接消费 P1 的结构化输出（category/outcome/follow\_up）作为预处理过的事实，而不是从 transcript 重新推断。

***

## 七、架构一致性评估

| Pipeline            | 架构 Pattern              | Hexagonal 成熟度 | 有无 Ports         | 有无 Core                        | 有无 Adapters          |
| ------------------- | ----------------------- | ------------- | ---------------- | ------------------------------ | -------------------- |
| P1 Call Analysis    | **Hexagonal**           | ★★★★          | ✅ `protocols.ts` | ✅ `core/stages/`               | ✅ AI/Neon/DDB/S3/SQS |
| P2 Contact Analysis | **Semi-hexagonal**      | ★★★           | ✅ `protocols.ts` | ⚠️ 只有 models + prompt-builder  | ✅ AI/Neon            |
| P3 SMS              | **Hexagonal（轻量）**       | ★★☆           | —                | ✅ `core/`                      | ✅ `infrastructure/`  |
| P4 Lead             | **Thin handler + core** | ★★            | —                | ✅ `core/persist-downstream.ts` | —                    |
| P5 Analytics        | **Script**              | ★             | —                | —                              | —                    |

**建议**：不需要所有 pipeline 都达到 P1 的 Hexagonal 深度。判断标准是"业务规则是否值得从 runtime/vendor 细节里剥离出来"（引自 `backend-patterns.md` §6）。P3 和 P4 业务规则简单，thin handler 够用。P2 的 task mutation 逻辑值得抽成独立的 Orchestrator core。

***

## 八、Prompt 设计对 Pipeline 架构的影响

当你为每个 pipeline 设计 prompt 时，需要清楚 **prompt 能读到什么数据**：

| Prompt                        | 需要的数据                                                                    | 来源 pipeline            | 来源表                                                 | 当前是否可用                                     |
| ----------------------------- | ------------------------------------------------------------------------ | ---------------------- | --------------------------------------------------- | ------------------------------------------ |
| P1 Triage                     | 通话开头 transcript + 员工名单                                                   | 当前通话                   | `staff` + RC API                                    | ✅                                          |
| P1 Classification             | 完整 transcript + business context                                         | 当前通话                   | RC transcript + DDB config                          | ✅                                          |
| P1 Coaching                   | transcript + classification 结果                                           | P1 前一 stage            | 内存传递                                                | ✅                                          |
| P2 Contact Analyzer           | 全量 calls + messages + leads + pending tasks + closed tasks + contacts 快照 | P1 + P3 + P4 + P2 历史   | Neon calls/messages/leads/tasks/contacts            | ✅ 但缺 task progress events + dueAtChangelog |
| **Future**: Task Orchestrator | 当前 task 状态 + progress events + 最近活动 + allowed enum set + store policy    | P2 AI 输出 + code policy | Neon tasks + task\_progress\_events + store\_config | 🔲 需要 `task_progress_events` 表             |

**关键洞察**：prompt 的输入质量直接取决于上游 pipeline 写入的数据质量。如果 SMS pipeline 不做 AI 分析，那 P2 prompt 里的 RECENT MESSAGES 就只有原始文本，没有预处理的 intent/sentiment。如果 lead-processor 不记录 progress events，P2 prompt 就不知道员工已经打了几次电话给这个 lead。

***

## 九、实际运行架构全景（2026-06 code-verified）

从代码直接验证的端到端架构 — 从外部源到 Neon 写入，含 shared mutation layer 和尚未接入的 Control Plane / studio-api。

```
┌─────────────────────────────── 外部源 ──────────────────────────────────────────┐
│                                                                                 │
│  📞 RingCentral (call ended)     💬 RingCentral (SMS)        📧 健身房网站 (Lead) │
│           │ webhook                       │ webhook               │ IMAP poll   │
│           ▼                               ▼                       ▼             │
│  ┌────────────────────────────────────────────┐       ┌───────────────────────┐ │
│  │ ringcentralSubscriptionService (独立 repo)  │       │ lead-tracking (独立)  │ │
│  │ • call → callLogQueue (300s delay)          │       │ • 写 leads + DDB     │ │
│  │ • SMS → message queue                       │       │ • 发 LeadCreated EB  │ │
│  └──┬──────────────────────┬───────────────────┘       └──────────┬────────────┘│
│     │ SQS                  │ SQS                                  │ EventBridge │
└─────┼──────────────────────┼──────────────────────────────────────┼─────────────┘
      ▼                      │                                      ▼
═══ Stage 1: source ingest ══│══════════════════════════════════════════════════════

┌────────────────────────┐   │                          ┌──────────────────────────┐
│ transcribe-processor   │   │                          │ lead-processor           │
│ • 下载录音 → Deepgram   │   │                          │ • applyTaskAction(       │
│ • 写 calls 表           │   │                          │    create_open,          │
│ • timeline:            │   │                          │    lead_outreach)        │
│   transcribe.completed │   │                          └──────────┬───────────────┘
└────────┬───────────────┘   │                                     │
         │ S3 → EB → SQS    │                                     │
         ▼                   │                                     │
┌────────────────────────┐   │                                     │
│ ai-analysis-processor  │   │                                     │
│ • AI: Triage/Classify/ │   │                                     │
│   Verify/Coaching      │   │                                     │
│   (DeepSeek V4 Flash)  │   │                                     │
│ • 写 calls (33 字段)    │   │                                     │
│ • 发 SQS 触发下游       │   │                                     │
└────────┬───────────────┘   │                                     │
         │ SQS               │                                     │
         ▼                   ▼                                     │
   dailyBatchQueue ◄── cron 06:00      ┌───────────────────────┐   │
   (FIFO)          ◄── on_demand       │ message-processor     │   │
         │                             │ • 入库 messages 表     │   │
         ▼                             │ • S3 存 MMS 附件       │   │
═══ Stage 2: contact-level ════════════│ • STOP → setDNC()     │   │
                                       └───────────┬───────────┘   │
┌──────────────────────────────┐                   │               │
│ contacts-analyzer            │                   │               │
│ • 聚合 calls+msgs+leads+tasks│                   │               │
│ • AI: 18 画像字段 +           │                   │               │
│   taskDecisions[] (6 action) │                   │               │
│ • applyTaskAction() ×N       │                   │               │
│ • inline UPDATE contacts     │                   │               │
└──────────┬───────────────────┘                   │               │
           │                                       │               │
           ▼                                       ▼               ▼
═══ Stage 3: shared mutation layer (callytics-common/src/domain/) ══════════════════

    ┌──────────────────────────────────────────────────────────────────┐
    │           Policy Guard (8 checks, pure functions)                │
    │  storeId / DNC / hallucination / state / typeCategory / etc.    │
    └──────┬──────────────────────┬──────────────────────┬────────────┘
           ▼                      ▼                      ▼
    ┌──────────────┐    ┌────────────────┐    ┌──────────────────┐
    │ Task         │    │ Contact        │    │ Timeline         │
    │ Orchestrator │    │ Writer         │    │ Writer           │
    │ 6 actions    │    │ upsertIdentity │    │ 16 eventType     │
    └──────┬───────┘    └────────┬───────┘    └──────┬───────────┘
           ▼                     ▼                   ▼
═══ Stage 4: Neon PostgreSQL (db.batch() 原子事务) ═════════════════════════════════

    tasks          contacts         contact_timeline    task_progress_events
    (open/closed)  (画像 SoT)       (审计 + feed)       (progress SoT)

═══ 独立平面（尚未接入） ═══════════════════════════════════════════════════════════

    control-plane repo: tenants / tenant_members / tenant_store_mappings
    ⚠️ 尚无 callytics-infra / studio-website-monorepo 代码消费

    studio-api tasks routes: close/reopen/postpone 仍走老逻辑
    ⚠️ 已接 @retaintive/common 但只用 schema 不用 domain modules
```

***

## 十、建议的统一架构方向

基于 Task Pipeline Deliverable 的 "Task Orchestrator" 模式，推广到整个系统的统一架构：

```
每条 Pipeline 遵循同一个 pattern:

Input Event → Code Filter → Context Reader → AI Judgment (optional) → Orchestrator → DB Transaction → Downstream

其中：
- Code Filter: dedup、merge、trivial 过滤（不是 AI 做的）
- Context Reader: 聚合该决策需要的所有数据
- AI Judgment: 只输出结构化提案，不执行
- Orchestrator: 验证提案 + 执行状态变更 + 幂等/并发控制
- DB Transaction: 原子写入
- Downstream: SQS fan-out / event publish
```

这不是要重写所有 pipeline — 是一个**渐进式的统一方向**，按优先级排序：

| 优先级    | 项目                                                                                   | 理由                                                          |
| ------ | ------------------------------------------------------------------------------------ | ----------------------------------------------------------- |
| **P0** | 推进 issue #771 — studio-api 迁移到 Neon 读取 call-analysis                                 | DDB 双写期间无 compensating transaction，是数据一致性风险                 |
| **P1** | **Task Orchestrator** — 统一 lead-processor / contacts-analyzer / studio-api 的 task 写入 | Deliverable 已设计完，3 个入口的去重/closeResult/dueAt/审计全不一致          |
| **P1** | **SMS Filter Layer** — message-processor 末尾加 code-level 过滤，触发 P2                     | STOP → 立即 DNC；有意义回复 → 立即触发 contacts-analyzer；当前延迟 24h       |
| **P1** | 统一 storeId null handling policy                                                      | 4 个 pipeline 行为不一致，需明确策略文档                                  |
| **P2** | **Timeline Event Catalog** — callytics-common 统一 event\_type 定义                      | 6 个 writer 各自定义 event\_type，无中央 catalog                     |
| **P2** | CLAUDE.md Product Context 更新 AI 模型描述                                                 | 文档写 "Kimi K2/Claude"，代码实际是 DeepSeek V4 Flash via OpenRouter |
| **P2** | closeResult 前端 dropdown 补齐 `booked` / `cancelled`                                    | Schema 有 18 个值，UI 只有 16 个                                   |
| **P2** | **Prompt Input Enrichment** — task\_progress\_events 加入 P2 prompt                    | AI 缺少员工手动调整和进展历史                                            |
| **P3** | reconciliation-orchestrator 去除 orangeTheory hardcode                                 | 单租户假设嵌入多租户架构                                                |
| **P3** | contacts.hasCardOnFile 清理决策                                                          | 零 writer，永远 NULL                                            |
| **P3** | AI 输出字段增加 DB CHECK constraints                                                       | calls 表 AI 字段无 DB-level 校验                                  |

***

## 相关文档

- [系统架构总览](/architecture/0-system-overview.md)
- [Backend 数据管道](/architecture/2-backend.md)
- [Backend 架构 Pattern](/system-design/backend-patterns.md)
- [Task Pipeline 设计 Deliverable](/product-design/v2/tasks-feature/design/task-pipeline-deliverable-codex.md)
- [Task Pipeline 设计 Brainstorm](/product-design/v2/tasks-feature/design/research/task-pipeline-design-brief.md)
- [Task 枚举权威清单](../product-design/v2/tasks-feature/task-enums.md)
- [Task 生命周期（Prompt 视角）](/product-design/v1/tasks-feature/prompt-task-lifecycle.md)
- [Prompt 改进索引](../product-design/v1/prompt-improvement/index.md)