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

# Unified Pipeline Research Appendix

> **当前状态**: Research appendix。Canonical 入口请读 [Unified Pipeline Architecture 设计 Brief](/product-design/v2/unified-pipeline/research/unified-pipeline-design-brief.md)。本文件的代码调研结论仍然有用，但 Phase 1 的架构表述已经进一步细化为 `Shared Mutation Modules` + `Processing Capability / Invocation Modules` 两层。
> **使用方式**: 用这里的 facts 校准设计，不要把这里的 early Phase 1 wording 当成最终分层。如果和 canonical brief 不一致，以 canonical brief 为准。
> **日期**: 2026-05-31\
> **目的**: 按 [Unified Pipeline Reference Checklist](/product-design/v2/unified-pipeline/research/unified-pipeline-reference-checklist.md) 做一轮快速代码调研，判断它是否会自然收敛到 [Unified Pipeline Architecture 设计 Brief](/product-design/v2/unified-pipeline/research/unified-pipeline-design-brief.md) 里说的”统一 pipeline / shared modules”方向。\
> **结论先行**: 两个方向不冲突。新的 brief 更适合作为调研入口；旧 brainstorm 的“统一 pipeline”是正确终点，但应该精确表述为 **统一 mutation/orchestration layer**，不是把所有 Lambda/SQS pipeline 合并成一个 runtime。

***

## Executive Summary

调研后的判断：

**Phase 1 不应该做 tool calling，也不应该重写成一个新的 pipeline framework。Phase 1 应该做的是：保留现有 SQS/Lambda pipeline，但把跨 pipeline 共享的写入行为统一到一层 shared mutation layer。**

这层 shared mutation layer 包括：

1. **Task Orchestrator**\
   统一 `create_task`、`update_task`、`close_task`、`record_task_progress`。

2. **Contact Writer**\
   统一 `contacts` UPSERT、name trust、DNC、lastActivityAt、lifecycle guard。

3. **Timeline Writer / Event Catalog**\
   统一 `contact_timeline` 写入、payload schema、idempotency、actor/source/AI forensic fields。

4. **Policy Guard**\
   统一 DNC、tenant isolation、storeId null、permission、idempotency、state transition。

旧 brainstorm 抓住了 “shared modules” 这个主要架构方向。新的 brief 抓住了“先调研 schema/prompt/API/workflow，再决定 shared modules 怎么做”的方法论。

所以裁判结论是：

**用新的 brief 当入口，用旧 brainstorm 的 shared module 方向当 Phase 1 endpoint。**

***

## 这次调研覆盖了什么

调研的 source of truth：

| Area             | 读了什么                                                                                                                                                                 |
| ---------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| DB schema        | `callytics-common/src/db/schema/tasks.ts`、`contacts.ts`、`contact-timeline.ts`、`calls.ts`、`messages.ts`、`leads.ts`、`name-trust.ts`                                    |
| Contact analysis | `callytics-infrastructure/lambda/contacts-analyzer/src/core/models.ts`、`prompt-builder.ts`、`infrastructure/neon-repository.ts`                                       |
| Call analysis    | `callytics-infrastructure/lambda/ai-analysis-processor/src/core/stages/*`、`infrastructure/neon-repository.ts`                                                        |
| SMS              | `callytics-infrastructure/lambda/message-processor/src/core/stop-keyword.ts`、`infrastructure/neon-repository.ts`                                                     |
| Lead             | `lead-tracking/src/neon-repository.ts`、`src/poller.ts`、`src/neon-retry-processor.ts`、`callytics-infrastructure/lambda/lead-processor/src/core/persist-downstream.ts` |
| Studio API       | `studio-website-monorepo/apps/api/src/routes/tasks/close.ts`、`list.ts`、`events.ts`                                                                                   |
| External refs    | OpenRouter tool calling docs、Pi.dev homepage                                                                                                                         |

***

## Current State Inventory

### 1. Business objects

| Object             | 当前职责                                            | 调研判断                                                          |
| ------------------ | ----------------------------------------------- | ------------------------------------------------------------- |
| `contacts`         | customer profile / aggregate state              | 职责偏宽，既有 profile，也有 action summary，还被多个 pipeline 写入            |
| `tasks`            | human-work objective / action item              | 是实际 work object，但 progress 和 final outcome 混在 `closeResult` 里 |
| `contact_timeline` | audit/activity feed                             | 结构已经很强，但 payload catalog 还没有完全产品化                             |
| `calls`            | per-call source record + AI call classification | 职责清楚，不应该直接创建 task                                             |
| `messages`         | SMS/voicemail source record                     | source record 清楚；meaningful SMS 是否触发 AI 还未形成 workflow         |
| `leads`            | inbound lead source record                      | source record 清楚；downstream contact/task/timeline 由事件触发       |

### 2. Prompt surfaces

当前主要 prompt surface：

| Prompt surface    | 类型                                  | 是否写 shared state                        |
| ----------------- | ----------------------------------- | --------------------------------------- |
| pre-triage        | code-only                           | 不写                                      |
| triage            | AI structured output                | 影响 call analysis flow，不直接写 task         |
| classify          | AI structured output                | 写 calls，间接影响 contacts-analyzer context  |
| verify            | AI structured output / observe-only | 当前不落库改变分类结果                             |
| coaching          | AI structured output                | 写 coaching fields，不创建 task              |
| contacts-analyzer | AI structured output                | 直接产生 contact fields + `taskDecisions[]` |

重点不是“几个 prompt”，而是：

**只有 contacts-analyzer 现在真正输出 task mutation proposal。per-call AI 不应该直接 own task lifecycle。**

### 3. Writers

多条 pipeline 直接写 shared tables：

| Table              | 当前 writers                                                            | 状态                                                            |
| ------------------ | --------------------------------------------------------------------- | ------------------------------------------------------------- |
| `contacts`         | transcribe、ai-analysis、message、lead、contacts-analyzer、studio/API 相关路径 | 有 common helpers，但还没有统一 Contact Writer                        |
| `tasks`            | lead-processor、contacts-analyzer、studio API、DNC cascade               | 有局部 guard，但还没有统一 Task Orchestrator                            |
| `contact_timeline` | transcribe、ai-analysis、message、lead、contacts-analyzer、studio API      | 有 `buildTimelineValues`，但还没有完整 event catalog/payload contract |

***

## Key Findings

### Finding 1: 旧 brainstorm 的 “统一 pipeline” 方向是对的，但名字容易误导

代码已经说明：现有系统不是缺一个 Temporal/LangGraph 类型的 runtime。SQS/Lambda/EventBridge 的基础设施已经可以支撑 async pipeline。

真正缺的是：

```text
shared mutation / orchestration layer
```

也就是每条 pipeline 仍然保留自己的 trigger/context/AI logic，但写 shared state 时必须走同一套模块：

```mermaid
flowchart TD
  Trigger["Pipeline trigger: SQS / cron / EventBridge / API"] --> Context["Pipeline-specific context builder"]
  Context --> Judgment["AI or code judgment"]
  Judgment --> Proposal["Business action proposal"]
  Proposal --> Guard["Shared policy + schema validation"]
  Guard --> ContactWriter["Contact Writer"]
  Guard --> TaskOrchestrator["Task Orchestrator"]
  Guard --> TimelineWriter["Timeline Writer"]
```

所以：

- 不建议说“统一成一个 pipeline”。
- 建议说“统一 pipeline write model / shared mutation layer”。

### Finding 2: 代码里已经有 shared abstraction 的雏形

这不是纯理论。代码已经有几个局部证明：

| Existing abstraction                                      | 说明                                                          |
| --------------------------------------------------------- | ----------------------------------------------------------- |
| `lambda/shared/utils/dnc-cascade.ts`                      | DNC 事实由 AI/SMS 断言，关闭 tasks 由 shared code 执行                 |
| `buildTimelineValues`                                     | timeline row 的 actor/source/AI forensic 字段已有统一 builder      |
| `buildIdentityFields` / `buildContactPhoneIdentityFields` | phone/store/franchise/account identity 已部分集中                |
| `lastActivityAtForward`                                   | contacts lastActivityAt forward-only 更新逻辑已集中                |
| common enums                                              | `TASK_TYPE_CATEGORY`、`TASK_CLOSE_RESULT`、timeline enums 已集中 |

这说明旧 brainstorm 的 shared module 方向不是过度设计。系统已经在往这个方向长，只是还没有形成完整边界。

### Finding 3: `tasks.closeResult` 确认存在 progress/outcome 混淆

`TASK_CLOSE_RESULT` 里同时包含：

Outcome-like:

- `converted`
- `win_back`
- `issue_resolved`
- `cancel_saved`
- `renewed`
- `upgraded`
- `booked`
- `cancelled`
- `do_not_contact`
- `wrong_number`

Progress-like:

- `attempted`
- `no_answer`
- `left_voicemail`
- `callback_later`

Studio API 也印证了这个混淆：`PATCH /v2/tasks/close` 在 `closeResult = no_answer / left_voicemail` 时，会关闭当前 task，然后自动创建 follow-up task。

这说明系统现在用“关闭 task + 新建 task”来表达“记录一次 attempt/progress”。这不是理想模型。

Phase 1 应该考虑：

- 新增 `task_progress_events`，或至少先定义 `record_task_progress` contract。
- `no_answer`、`left_voicemail`、`callback_later`、`text_sent` 不应该作为 final close outcome。
- task close 应该表达 objective completed / invalid / abandoned / blocked，而不是每次操作尝试。

### Finding 4: Contact-level action 和 Task-level action 确实重叠

`contacts` 里有：

- `actionNeeded`
- `actionNeededReason`
- `suggestedActions`

`tasks` 里也有：

- `actionNeeded`
- `actionNeededReason`
- `suggestedActions`

调研判断：

**task 应该是 actual work object source of truth；contact-level action 更适合作为 summary/projection。**

否则会出现：

- prompt 同时更新 contact action 和 task action，不知道谁赢。
- UI 不知道展示 contact action 还是 task action。
- API 不知道“需要跟进”到底是一个 task，还是一个 profile flag。

Phase 1 不一定要删除 contact fields，但必须定义语义：

```text
tasks = source of truth for human work
contacts.actionNeeded/suggestedActions = derived summary / compatibility projection
```

### Finding 5: `contact_timeline` 已经很接近统一 audit layer，但还不是 task progress SoT

`contact_timeline` schema 已经有：

- typed `eventType`
- `eventCategory`
- polymorphic entity
- actor identity
- actor source type/system
- AI forensic columns
- idempotency key

这是很好的基础。

但它仍然不应该直接替代 `task_progress_events`。原因：

- timeline 是 contact-level feed。
- task progress 是 task lifecycle 内部事实。
- timeline payload 是 JSONB，适合 audit/projection，但不适合承担所有 task attempt/count 的 source of truth。

现在 `tasks/list.ts` 的 `attempt_count` 是通过 counting same contact/store closed tasks 推出来的。这说明缺一个更准确的 progress source。

建议：

```text
task_progress_events = task-level SoT
contact_timeline = user-visible/audit projection
```

### Finding 6: SMS STOP 处理比旧文档更先进，但 meaningful SMS workflow 还没完成

当前 `message-processor` 已经有 deterministic STOP handling：

- inbound SMS exact body equals `stop`
- code sets `contacts.doNotContact = true`
- code writes `message.created`
- code calls shared `closeOpenTasksForDnc`

这已经是 `Code as Guardrail, AI as Judgment` 的很好例子。

还没完成的是：

- meaningful inbound SMS 是否应触发 contacts-analyzer。
- 如何过滤 `ok` / `thanks` / emoji / auto-reply。
- natural-language DNC 由哪个 path 进入 AI。

所以旧 brainstorm 里如果说 “SMS 只存储，不处理” 已经不准确。更准确是：

```text
SMS exact STOP 已 code-only 处理；
meaningful SMS → contact analysis trigger 还未设计。
```

### Finding 7: Lead pipeline happy path 拆得合理，但 retry path 有 downstream gap 风险

happy path：

1. lead-tracking writes `leads`
2. lead-tracking publishes EventBridge `LeadCreated`
3. lead-processor writes `contacts + lead_outreach task + contact_timeline`

这个拆分是合理的。

但 `neon-retry-processor.ts` 历史上只调用：

```ts
persistLeadPipeline(row)
```

它没有重新 publish `LeadCreated` event。

如果原始 Neon write 失败进入 DLQ，retry success 后可能只补上 `leads`，但不触发 downstream contact/task/timeline。

✅ **已修复(lead-tracking#201, 2026-06-08)** — 抽出共享 `publishLeadCreatedEvent`,retry path 也发同一份事件,详见 [lead-tracking 写入流程](/system-design/write-matrix/lead-tracking-writes.md)。

这就是 unified architecture 要检查的典型问题：

```text
happy path 和 retry/backfill path 是否走同一个 workflow contract？
```

### Finding 8: Contact Writer 不是空想，当前最大重复点就是 contact UPSERT

多个 writer 都在做类似逻辑：

- resolve identity
- `storeId` guard
- insert/update contacts
- name trust
- `lastActivityAt` forward-only
- timeline

虽然已经有 `NAME_TRUST` 和 helper，但 enforcement 仍然散在不同 repository 里。

Contact Writer 应该定义成一个 shared contract：

```ts
upsertContactEvent({
  identity,
  source,
  actor,
  trust,
  observedAt,
  patch,
  policy,
  timeline,
})
```

它不一定第一版覆盖所有字段，但至少应该先统一：

- identity fields
- name trust
- lastActivityAt
- DNC sticky rule
- lifecycle/leadStatus guard strategy
- timeline side effect

### Finding 9: Task Orchestrator 是 Phase 1 里最应该先做的 shared module

原因：

1. task deliverable 已经基本通过。
2. task 当前有 3 个 mutation 入口：lead、contacts-analyzer、studio API。
3. task 里 progress/outcome 混淆最直接影响 UI 和 analytics。
4. task 是未来 AI tool 最容易暴露的 API。

建议 Task Orchestrator 的 Phase 1 contract：

```ts
applyTaskAction({
  action: 'create' | 'update' | 'close' | 'record_progress',
  actor,
  identity,
  taskRef,
  payload,
  idempotencyKey,
})
```

先不要追求大而全，但要保证：

- tenant/store guard
- pending dedup
- close integrity
- timeline projection
- future tool API 可以直接复用

### Finding 10: Tool calling 现在不是 blocker，也不是 Phase 1 的解法

OpenRouter 文档说明的 tool calling 是：

1. backend 把 tool schema 给 model。
2. model 返回 `tool_calls`。
3. application/backend 执行 tool。
4. backend 把 tool result 返回 model。

也就是说，tool calling 本身也支持这个原则：

```text
AI requests/proposes; backend executes.
```

它不是“AI 随便操作数据库”。

对 Retaintive 当前阶段来说，tool calling 的价值主要在：

- interactive diagnosis
- voice agent
- user asks a question and AI decides what to fetch
- unknown query path

当前主要 pipeline 是 batch/async：

- call analysis
- contact analysis
- lead processing
- SMS event processing

这些更适合 one-time AI call + code orchestration。

结论：

**Phase 1 做 agent-ready，不做 agent runtime。**

Pi.dev 目前更像 coding-agent / dev harness 方向，不应该直接影响 backend architecture 结论。

***

## Current vs Target Workflow

### Current State

```mermaid
flowchart TD
  Call["Call webhook / transcribe"] --> CallAI["AI call analysis prompts"]
  CallAI --> CallWrite["writes calls + contacts + timeline"]
  CallAI --> ContactQueue["queues contacts-analyzer"]

  ContactQueue --> ContactPrompt["contacts-analyzer prompt"]
  ContactPrompt --> ContactWrite["writes contacts + tasks + timeline"]

  SMS["SMS webhook"] --> SMSCode["message-processor code"]
  SMSCode --> SMSWrite["writes messages + contacts + timeline"]
  SMSCode --> DNCShared["shared DNC cascade only for exact STOP"]

  Lead["Lead email"] --> LeadEvent["LeadCreated EventBridge"]
  LeadEvent --> LeadProcessor["lead-processor"]
  LeadProcessor --> LeadWrite["writes contacts + lead_outreach task + timeline"]

  Studio["Studio API"] --> ManualTaskWrite["direct task update + timeline + auto follow-up"]
```

Problem:

```text
Every pipeline knows how to write shared business state.
```

### Target State

```mermaid
flowchart TD
  Call["Call pipeline"] --> Proposal["Business facts / AI structured output"]
  ContactAnalyzer["Contact analyzer"] --> Proposal
  SMS["SMS pipeline"] --> Proposal
  Lead["Lead pipeline"] --> Proposal
  Studio["Studio API"] --> Proposal

  Proposal --> Guard["Policy Guard: schema + tenant + DNC + idempotency"]
  Guard --> ContactWriter["Contact Writer"]
  Guard --> TaskOrchestrator["Task Orchestrator"]
  Guard --> TimelineWriter["Timeline Writer"]

  FutureTool["Future AI tool call"] --> Guard
```

Goal:

```text
Every pipeline owns context and judgment.
Shared modules own state mutation.
```

***

## Phase 1 Recommendation

### Phase 1A: Task Orchestrator first

Do this first because task is where product semantics are most mixed.

Scope:

- Add/define `record_progress` contract.
- Decide whether to add `task_progress_events`.
- Move `no_answer` / `left_voicemail` / `callback_later` out of final outcome semantics.
- Refactor `studio-api` close behavior conceptually: no-answer should record attempt/progress, not close+recreate.
- Route contacts-analyzer `taskDecisions[]` through the orchestrator.
- Route lead-created `lead_outreach` task creation through the orchestrator.

### Phase 1B: Timeline catalog

Scope:

- Keep `contact_timeline` as audit/feed.
- Define payload schema per `eventType`.
- Ensure every task mutation produces a timeline projection.
- Add tests for event payload shape.

### Phase 1C: Contact Writer

Scope:

- Start with identity/name/lastActivityAt/DNC, not every contact field.
- Move repeated UPSERT logic into one shared path.
- Preserve existing name trust semantics.
- Define lifecycle/leadStatus guard as observe-only first, then enforce when product rules are clear.

### Phase 1D: Trigger/re-entry contract

Scope:

- Daily batch, per-call trigger, on-demand, reprocess, retry, onboarding, manual API should all call the same business operation where possible.
- Fix lead retry downstream event gap.
- Define meaningful SMS trigger policy.

***

## What This Means For The Two Briefs

### New `unified-pipeline-architecture-brief.md`

Better as the main entry point because it forces the right order:

```text
schema → pipeline/prompt inventory → workflow → API check → Phase 1
```

This is the safer way to avoid over-design.

### Old `unified-pipeline-design-brainstorm.md`

Still valuable because it explains the architectural endpoint:

```text
shared modules + Code as Guardrail + future tool compatibility
```

But it should be reframed:

```text
统一 pipeline
```

should become:

```text
统一 shared mutation layer / pipeline write model
```

### Final judgment

Both documents are useful, but they sit at different layers:

| Document                                 | Best role                         |
| ---------------------------------------- | --------------------------------- |
| `unified-pipeline-architecture-brief.md` | research/execution entry point    |
| `unified-pipeline-design-brainstorm.md`  | architecture narrative / endpoint |

The research confirms that the right direction is:

```text
Do not build a new agent runtime.
Do not merge all pipelines.
Do build shared modules that every pipeline and future tool API must call.
```

***

## Open Questions For Max Review

1. Should `task_progress_events` be a real new table in Phase 1, or first be expressed as a contract and projected into `contact_timeline`?
2. Should `contacts.actionNeeded/suggestedActions` remain as compatibility summary fields, or should we plan to deprecate them after task UI stabilizes?
3. Should `no_answer` / `left_voicemail` be removed from `TASK_CLOSE_RESULT`, or kept temporarily but no longer used by new APIs?
4. Should meaningful inbound SMS trigger contacts-analyzer immediately, or go through a code filter / debounce queue first?
5. Should lead retry success republish `LeadCreated`, or should downstream processing be made idempotently callable from retry/backfill directly?
6. Should Phase 1 prioritize Task Orchestrator only, or do Task Orchestrator + Timeline Writer together as one slice?