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

# Unified Pipeline Phase 1 (Codex)

> **当前状态**: Phase 1 implementation design。基于 [Current State Research](/product-design/v2/unified-pipeline/research/unified-pipeline-current-state-codex.md)，定义第一阶段应该怎么落地。
> **日期**: 2026-05-31
> **Canonical brief**: [Unified Pipeline Architecture 设计 Brief](/product-design/v2/unified-pipeline/research/unified-pipeline-design-brief.md)
> **理想终态**: [Unified Pipeline Target State (Codex)](/product-design/v2/unified-pipeline/original/unified-pipeline-target-state-codex.md)

***

## Final Design First

Phase 1 最终应该长这样：

```mermaid
flowchart LR
  subgraph Runtime["Existing runtime kept"]
    Studio["Studio API<br/>task routes"]
    Analyzer["contacts-analyzer<br/>taskDecisions[]"]
    Lead["lead-processor<br/>downstream write"]
    SmsStop["message-processor<br/>exact STOP only"]
  end

  subgraph SourceRecords["Source records stay owned by each pipeline"]
    Messages[("messages")]
    Leads[("leads")]
  end

  SmsStop -. "persist source message" .-> Messages
  Lead -. "persist source lead" .-> Leads

  subgraph Requests["Typed mutation requests"]
    TaskAction["TaskAction<br/>create / update / close / record_progress / reopen"]
    ContactAction["ContactAction<br/>identity / DNC / trust"]
  end

  Studio --> TaskAction
  Analyzer --> TaskAction
  Lead --> TaskAction
  Lead --> ContactAction
  SmsStop --> ContactAction
  SmsStop --> TaskAction

  subgraph SharedBoundary["Phase 1 shared mutation boundary"]
    Guard["Policy Guard<br/>store / DNC / state / idempotency"]
    Task["Task Orchestrator"]
    Contact["Contact Writer<br/>minimal: identity / DNC / trust"]
    Timeline["Timeline Writer<br/>audit / contact feed projection"]
  end

  TaskAction --> Guard
  ContactAction --> Guard
  Guard --> Task
  Guard --> Contact
  Task --> Timeline
  Contact --> Timeline

  subgraph SharedState["Shared state written through shared modules"]
    Tasks[("tasks")]
    Progress[("task_progress_events")]
    Contacts[("contacts")]
    Feed[("contact_timeline")]
  end

  Task --> Tasks
  Task --> Progress
  Contact --> Contacts
  Timeline --> Feed
```

一句话：

**Phase 1 不是重画所有 pipeline，而是先把已经会改 `tasks` / `contacts` / `contact_timeline` 的高风险写入收口到 shared mutation boundary。**

具体说：

- `Studio API`、`contacts-analyzer`、`lead-processor`、`message-processor` exact STOP 继续作为现有入口存在。
- 这些入口不再自己拼 task/contact mutation SQL，而是产出 `TaskAction` / `ContactAction`。
- `Policy Guard` 决定 action 能不能执行。
- `Task Orchestrator` 负责 task lifecycle 和 `task_progress_events`。
- `Contact Writer` 第一版只管 identity / DNC / trust，不试图覆盖 contacts 表所有 AI 字段。
- `Timeline Writer` 负责被 shared modules 触发的 audit / contact feed projection。
- `messages` / `leads` 这类 source records 仍由原 pipeline 写入；source event timeline 的全面迁移可以放到后续 target-state cleanup。

Phase 1 同时定义但不完整迁移这些 invocation contracts：

| Contract                   | Phase 1 stance                                                          |
| -------------------------- | ----------------------------------------------------------------------- |
| Call Analysis Invocation   | 只定义 retry/backfill/reconciliation contract，不重写 call analysis runtime    |
| Meaningful SMS Invocation  | 只定义 meaningful SMS 如何进入后续 workflow；Phase 1 只做 exact STOP                |
| Contact Analysis Reprocess | 收敛 `source` / `reprocessMode` / `intentKind` 为 `ProcessingIntent`       |
| Lead Downstream Contract   | lead 仍是 deterministic flow，但 downstream task/contact 写入走 shared modules |

### Phase 1 Decision Summary

| Decision              | Phase 1 answer                                                                       |
| --------------------- | ------------------------------------------------------------------------------------ |
| Runtime               | 保留现有 SQS/Lambda/API，不引入 workflow engine                                              |
| AI mode               | 继续 one-time structured output，不做 tool calling runtime                                |
| First shared module   | `Task Orchestrator`                                                                  |
| First schema addition | `task_progress_events`                                                               |
| First API correction  | `record_progress`，避免 `no_answer` / `left_voicemail` 被当成 close                        |
| Timeline              | 通过 `Timeline Writer` 做 Phase 1 business mutation audit；source event timeline 可后续统一迁移 |
| Policy                | 先做 shared functions，不做独立 policy service                                              |
| Call/SMS/Lead         | 定义 capability invocation contract；写入仍走 shared mutation modules                       |
| Prod                  | 不做 destructive migration；prod legacy 需要单独 audit                                      |

如果只读这一节，读者应该已经知道 Phase 1 的终局形态。下面才是原因、contract 和 migration order。

***

## Phase 1 Goal

Phase 1 的目标不是重写 pipeline runtime，也不是做 tool calling。

Phase 1 的目标是：

**保留现有 SQS/Lambda/API 入口，把 Phase 1 涉及的 deterministic mutation、policy guard、business mutation audit 写入统一成 shared modules。**

换句话说：

```text
Before:
  each pipeline owns trigger + AI + DB mutation + audit

After Phase 1:
  each pipeline owns trigger + context + AI judgment
  shared modules own mutation + state transition + policy + audit
```

***

## Phase 1 Non-goals

- 不切换到 tool calling runtime。
- 不引入 LangGraph / Temporal / workflow engine。
- 不把所有 Lambda 合并成一个 pipeline。
- 不一次性迁移 prod legacy。
- 不把 Call/SMS/Lead 变成新的 table writer。
- 不让 AI 直接执行 DB mutation。

***

## Recommended Slice

### Phase 1A: Contracts first

先定义 shared module contracts，不急着迁移所有 caller。

建议放在 common/domain 层，而不是某个 Lambda 内部：

```text
callytics-common/src/domain/pipeline/
  task-orchestrator/
  contact-writer/
  timeline-writer/
  policy-guard/
  processing-intent/
```

如果 repo ownership 不适合，也可以先放在 infrastructure shared package，但原则是：不能归属于 `contacts-analyzer` 或 `studio-api`。

### Phase 1B: Task Orchestrator first

Task 是最需要优先统一的，因为当前已有 4 类 caller：

- AI `contacts-analyzer`
- deterministic `lead-processor`
- human `studio-api`
- system `message-processor` STOP cascade

核心 contract：

```ts
applyTaskAction({
  action: 'create' | 'update' | 'close' | 'record_progress' | 'reopen',
  identity: {
    contactPhone: string;
    storeId: string;
    franchiseId: string;
    accountId: string;
  },
  actor: {
    actorType: 'staff' | 'contact_analysis' | 'lead_webhook' | 'system' | 'ai_agent';
    actorSubjectId?: string;
    actorName?: string;
    actorSourceType: 'human_ui' | 'human_api' | 'service' | 'integration' | 'import' | 'unknown';
    actorSourceSystem?: string;
  },
  payload: TaskActionPayload,
  idempotencyKey: string,
  policyContext?: PolicyContext,
})
```

返回：

```ts
{
  status: 'applied' | 'noop' | 'rejected' | 'needs_review';
  taskId?: string;
  reason?: string;
  timelineEventIds?: string[];
}
```

### Phase 1C: Add `task_progress_events`

新增 task-level progress source of truth：

```text
task_progress_events
  id
  task_id
  contact_phone
  store_id
  franchise_id
  account_id
  progress_type
  progress_result
  note
  actor_type
  actor_subject_id
  actor_source_type
  actor_source_system
  source_type
  source_entity_type
  source_entity_id
  ai_run_id
  ai_confidence
  idempotency_key
  occurred_at
  created_at
```

第一版 `progress_type` 可以很保守：

```text
attempted_call
left_voicemail
no_answer
callback_requested
sms_sent
note_added
other
```

关键语义：

- `record_progress` 不关闭 task。
- `close` 才表达 objective 的最终结果。
- `contact_timeline` 是 contact-level projection / audit feed，不是 task progress source of truth。

### Phase 1D: Timeline Writer / Event Catalog

不要让每个 caller 手写 event payload。抽一个薄的 `Timeline Writer`：

```ts
writeTimelineEvent({
  contactIdentity,
  eventType,
  eventCategory,
  entityType,
  entityId,
  oldValue,
  newValue,
  metadata,
  actor,
  aiForensics,
  idempotencyKey,
})
```

Phase 1 只需要覆盖 task 相关事件：

- `task.created`
- `task.updated`
- `task.status_changed`
- `task.due_at_changed`
- `task.note_updated`
- `task.progress_recorded`（如果新增 event type）

后续再扩展到 contact / message / call / lead。

### Phase 1E: Policy Guard as functions, not service

Phase 1 不需要独立 policy service。先用 shared functions：

```ts
validateTaskActionPolicy({
  action,
  currentTask,
  contact,
  actor,
  payload,
  idempotencyKey,
})
```

必须覆盖：

| Policy                 | Rule                                                                      |
| ---------------------- | ------------------------------------------------------------------------- |
| Store isolation        | action identity 必须包含 `storeId`，mutation WHERE 必须 scope 到 `storeId`        |
| DNC hard stop          | DNC contact 不能 create/update outreach task；只能 close/noop                  |
| State transition       | pending → closed；closed → pending 只能 reopen；record\_progress 只能用于 pending |
| AI hallucination guard | AI close/update 必须引用 current contact/store 下的 existing pending task       |
| Idempotency            | every mutation must accept deterministic `idempotencyKey`                 |
| Human authority        | staff mutation should not be silently overwritten by stale AI proposal    |

***

## Caller Migration Order

### 1. Studio API task routes

先迁移 human route，因为语义最容易确认：

| Current route                         | Phase 1 target                                 |
| ------------------------------------- | ---------------------------------------------- |
| `PATCH /v2/tasks/close`               | calls `TaskOrchestrator.closeTask()`           |
| `PATCH /v2/tasks/reopen`              | calls `TaskOrchestrator.reopenTask()`          |
| `PATCH /v2/tasks/postpone`            | calls `TaskOrchestrator.updateTask({ dueAt })` |
| new `POST /v2/tasks/:taskId/progress` | calls `TaskOrchestrator.recordProgress()`      |

关键行为改动：

- `closeResult = no_answer` / `left_voicemail` 不应该再 close + auto-create follow-up。
- 这些应该转成 `record_progress` + optional `dueAt` update。

### 2. Contacts-analyzer `taskDecisions[]`

当前 `taskDecisions[]`：

```text
create | update | close
```

Phase 1 target：

```text
create | update | close | record_progress
```

迁移方式：

1. 先让 repository 内部调用 `TaskOrchestrator`，不改 prompt。
2. Contract 稳定后再改 prompt schema，加入 `record_progress`。
3. 保持旧 `close` 行为兼容，避免一次改动 prompt + DB + API。

### 3. Lead downstream

`lead-processor` 目前 deterministic 写 `contacts + tasks + timeline`。

Phase 1 target：

```text
persist lead source record
  -> ContactWriter.upsertFromLead()
  -> TaskOrchestrator.createTask(typeCategory='lead_outreach')
  -> TimelineWriter.write('lead.created')
```

注意：lead 不经过 AI，仍然是 code-only deterministic flow。

### 4. SMS exact STOP

`message-processor` 现在已经有 exact STOP DNC cascade。

Phase 1 target：

```text
exact STOP
  -> ContactWriter.setDnc(source='system', reason='sms_stop')
  -> TaskOrchestrator.closeAllOpenOutreachTasks(reason='do_not_contact')
  -> TimelineWriter.write(...)
```

不要把 meaningful SMS AI analysis 塞进这个 migration。exact STOP 是 Phase 1；meaningful SMS 是 Phase 2 capability。

### 5. Contact Writer extraction

最小可行 Contact Writer：

```ts
upsertContactIdentity({
  phone,
  storeId,
  franchiseId,
  accountId,
  firstName,
  lastName,
  nameTrustScore,
  lastActivityAt,
  source,
})

setDoNotContact({
  phone,
  storeId,
  source: 'staff' | 'system' | 'ai',
  reason,
  idempotencyKey,
})
```

不要第一版就试图覆盖 contacts 表所有 AI 字段。先抽最容易重复和最危险的部分：identity、name trust、DNC、lastActivityAt。

***

## Processing Capability Contracts

Phase 1 可以先定义 contract，不一定全部实现。

### Call Analysis Invocation

```ts
analyzeCall({
  callId,
  mode: 'normal' | 'retry' | 'backfill' | 'reconciliation',
  requestedBy,
  idempotencyKey,
})
```

必须决定：

- retry 是否清 `aiAnalysisCompletedAt`
- retry 是否覆盖 `s3AnalysisPath`
- retry 是否重新触发 contacts-analyzer
- retry timeline 是覆盖 canonical row 还是追加 attempt event

### SMS Signal Invocation

```ts
evaluateSms({
  messageId,
  mode: 'normal' | 'backfill',
  idempotencyKey,
})
```

Phase 1 只实现 exact STOP；meaningful SMS 留给 Phase 2。

### Contact Analysis Reprocess

```ts
reanalyzeContact({
  phone,
  storeId,
  reason: 'cron' | 'per_call' | 'on_demand' | 'reprocess' | 'task_close',
  force,
  idempotencyKey,
})
```

当前 `source` / `reprocessMode` / `intentKind` 已经有雏形。Phase 1 应该把它收敛成 `ProcessingIntent` 类型。

***

## Schema Changes

### Required

1. Add `task_progress_events`.
2. Add event type if needed: `task.progress_recorded`.
3. Add shared TypeScript contracts for:
   - `TaskAction`
   - `TaskActor`
   - `TaskProgressType`
   - `ProcessingIntent`
   - `TimelineEventPayload`

### Optional / Later

- Add `processing_runs` table if retry/backfill needs first-class ledger.
- Add `schema_version` to `contact_timeline.newValue` payloads.
- Add contact-level projection fields only after source-of-truth decision.

***

## Current vs Phase 1 Workflow

```mermaid
flowchart TD
  Pipeline[Existing Lambda/API caller] --> Judgment[Prompt or deterministic code]
  Judgment --> Action[Typed action request]
  Action --> Guard[Policy Guard]
  Guard --> Orchestrator[Task Orchestrator / Contact Writer]
  Orchestrator --> DB[(tasks / task_progress_events / contacts)]
  Orchestrator --> Timeline[Timeline Writer]
  Timeline --> CT[(contact_timeline)]
```

Important: pipeline runtime does not change. Only mutation ownership changes.

***

## Test Plan

Phase 1 must have contract tests, because shared modules raise blast radius.

| Test type                     | Coverage                                                                   |
| ----------------------------- | -------------------------------------------------------------------------- |
| Unit tests                    | each TaskAction path: create/update/close/record\_progress/reopen          |
| Policy tests                  | DNC blocks create/update; storeId required; staff override not overwritten |
| Idempotency tests             | same idempotency key does not duplicate task/timeline/progress             |
| Migration compatibility tests | old close route still works during transition                              |
| Prompt golden tests           | contacts-analyzer output with `record_progress` validates                  |
| Replay tests                  | apply same action twice; retry after partial timeline failure              |

***

## Phase 1 Acceptance Criteria

Phase 1 is done when:

- At least Studio API task routes and contacts-analyzer task decisions use `Task Orchestrator`.
- `record_progress` exists as API + backend contract.
- `task_progress_events` exists and is source of truth for attempts/progress.
- `contact_timeline` receives projection/audit events through `Timeline Writer`.
- DNC task close behavior goes through the same task contract for AI DNC and SMS STOP.
- Lead `lead_outreach` creation uses the same create-task contract or has an explicit, documented exception.
- No AI prompt can directly introduce a DB mutation shape not represented in typed action schema.

***

## What To Defer

- Full tool calling.
- Voice agent.
- Meaningful SMS AI analysis.
- Full Contact Writer migration across every field.
- Prod migration / data backfill plan.
- Replacing SQS/Lambda with workflow engine.

Those are target-state or Phase 2+ topics.