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

# Unified Pipeline Target State (Codex)

> **当前状态**: Target state / ideal architecture。描述我们认为最合理的终态，不代表 current code 已经实现。
> **日期**: 2026-05-31
> **Current-state research**: [Unified Pipeline Current State Research](/product-design/v2/unified-pipeline/research/unified-pipeline-current-state-codex.md)
> **Phase 1 design**: [Unified Pipeline Phase 1 Implementation Design](/product-design/v2/unified-pipeline/original/unified-pipeline-phase-1-codex.md)

***

## Final Architecture First

读者先看这一张图。后面的 layer 说明、workflow examples、open questions 都是在解释这张图。

```mermaid
flowchart TD
  subgraph Entry["Entrypoints"]
    CallEvent["Call event / retry"]
    SmsEvent["SMS event"]
    LeadEvent["Lead event"]
    StaffUi["Human UI / Backend API"]
    AgentTool["Future AI tool call"]
  end

  subgraph Capability["Processing Capability Layer"]
    CallCap["Call Analysis"]
    SmsCap["SMS Signal"]
    ContactCap["Contact Analysis"]
    LeadCap["Lead Downstream"]
  end

  Proposal["Typed proposal / deterministic action"]
  SourceProjection["Source event projection<br/>call / message / lead facts"]

  subgraph Mutation["Shared Mutation Layer"]
    Guard["Policy Guard"]
    Task["Task Orchestrator"]
    Contact["Contact Writer"]
    Timeline["Timeline Writer"]
  end

  subgraph SourceData["Source records"]
    Calls[(calls)]
    Messages[(messages)]
    Leads[(leads)]
  end

  subgraph SharedData["Shared state / projections"]
    Contacts[(contacts)]
    Tasks[(tasks)]
    Progress[(task_progress_events)]
    Feed[(contact_timeline)]
  end

  CallEvent --> CallCap
  SmsEvent --> SmsCap
  LeadEvent --> LeadCap
  StaffUi --> Proposal
  AgentTool --> Proposal

  CallCap --> Calls
  SmsEvent --> Messages
  LeadEvent --> Leads

  CallCap --> SourceProjection
  SmsCap --> SourceProjection
  LeadCap --> SourceProjection
  SourceProjection --> Timeline

  CallCap --> ContactCap
  SmsCap --> ContactCap
  ContactCap --> Proposal
  LeadCap --> Proposal

  Proposal --> Guard
  Guard --> Task
  Guard --> Contact

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

最终状态的核心不是“AI 更聪明”，而是系统有稳定的 state authority：

| Layer                 | Final responsibility                               |
| --------------------- | -------------------------------------------------- |
| Entrypoints           | 发现事件或接收请求                                          |
| Processing Capability | 复用 call/SMS/contact/lead 处理能力                      |
| Judgment              | AI 或 code 产出 typed proposal / deterministic action |
| Timeline Event Intake | 把 call/message/lead 事实事件规范投影到 `contact_timeline`   |
| Policy Guard          | 决定 action 能不能执行                                    |
| Shared Mutation       | 统一写 `tasks` / `contacts` / `contact_timeline`      |
| Storage               | 明确 source of truth 和 projection                    |

一句话：

**未来无论是 Lambda、UI、retry job，还是 AI agent，都不直接随意写 shared state；它们都通过同一套 typed、audited、policy-guarded modules 改变系统状态。**

***

## One Sentence

最理想的状态不是“一个万能 AI agent 接管所有 pipeline”，也不是“把所有 Lambda 合并成一个 pipeline”。

最理想的状态是：

**多个 pipeline / API / future agent 共享同一套 domain schema、policy guard、mutation modules、audit event catalog；AI 只负责 semantic judgment，代码负责 state authority。**

***

## Detailed Architecture

上面的图是给 review 快速理解的；这里是更细的版本，展示每个 entrypoint、capability、judgment 和 storage 之间的关系。

```mermaid
flowchart TD
  subgraph Entrypoints
    CallEntry[Call webhook / retry / backfill]
    SmsEntry[SMS webhook / backfill]
    LeadEntry[Lead ingestion]
    StaffEntry[Staff UI / Backend API]
    AgentEntry[Future AI agent / tool call]
  end

  subgraph Capabilities["Processing Capability / Invocation Layer"]
    CallModule[Call Analysis Module]
    SmsModule[SMS Signal Module]
    ContactModule[Contact Analysis Module]
    LeadModule[Lead Downstream Module]
  end

  subgraph Judgment["Judgment Layer"]
    CodeRules[Code-only rules]
    AI[AI structured output]
  end

  subgraph TimelineIntake["Timeline Event Intake"]
    SourceProjection[Source event projection]
  end

  subgraph Mutation["Shared Mutation Layer"]
    Guard[Policy Guard]
    Task[Task Orchestrator]
    Contact[Contact Writer]
    Timeline[Timeline Writer]
  end

  subgraph Storage["Source of Truth / Projection"]
    Calls[(calls)]
    Messages[(messages)]
    Leads[(leads)]
    Contacts[(contacts)]
    Tasks[(tasks)]
    Progress[(task_progress_events)]
    Feed[(contact_timeline)]
  end

  CallEntry --> CallModule
  SmsEntry --> SmsModule
  LeadEntry --> LeadModule
  StaffEntry --> Guard
  AgentEntry --> Guard

  SmsEntry --> Messages
  LeadEntry --> Leads

  CallModule --> CodeRules
  CallModule --> AI
  CallModule --> Calls
  CallModule --> SourceProjection
  SmsModule --> CodeRules
  SmsModule --> AI
  SmsModule --> SourceProjection
  ContactModule --> AI
  LeadModule --> CodeRules
  LeadModule --> SourceProjection

  CodeRules --> Guard
  AI --> Guard
  Guard --> Task
  Guard --> Contact

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

这个图里有一个关键边界：

- `Call / SMS / Lead` 的 source record 仍然写入 `calls` / `messages` / `leads`。这些是事实记录，不经过 `Task Orchestrator` 或 `Contact Writer`。
- `call.created` / `call.analysis_completed` / `message.received` / `lead.received` 这类 source event 通过 `Timeline Writer` 写 `contact_timeline`，但不需要 full `Policy Guard`。
- `create_task` / `close_task` / `record_progress` / `mark_contact_dnc` / `update_contact_identity` 这类 business mutation 必须先过 `Policy Guard`，再由 `Task Orchestrator` / `Contact Writer` 执行，最后投影到 `contact_timeline`。
- `Timeline Writer` 自己也要做 validation：event catalog、payload schema、store/contact identity、idempotency、actor fields、`occurredAt`、`schema_version`。这叫 timeline event validation，不等同于 business `Policy Guard`。

***

## Layer Responsibilities

### 1. Entrypoints

Entrypoints only decide that something happened:

- RingCentral call event
- RingCentral SMS event
- lead email received
- staff clicked a button
- admin started reprocess
- future AI agent called a tool

Entrypoints should not own business mutation rules.

### 2. Processing Capability / Invocation Layer

This layer owns reusable processing capabilities:

| Module                  | Contract                                    | What it owns                                      | What it does not own    |
| ----------------------- | ------------------------------------------- | ------------------------------------------------- | ----------------------- |
| Call Analysis Module    | `analyze_call(callId, mode)`                | transcript / per-call AI / call classification    | task creation           |
| SMS Signal Module       | `evaluate_sms(messageId, mode)`             | exact STOP, trivial filter, meaningful SMS signal | direct task write       |
| Contact Analysis Module | `reanalyze_contact(phone, storeId, reason)` | aggregate context + contact-level AI judgment     | raw DB mutation         |
| Lead Downstream Module  | `process_lead_downstream(leadId)`           | deterministic lead-to-contact/task intent         | custom task persistence |

These are capability modules, not writer modules.

### 3. Judgment Layer

Judgment can be code-only, AI-only, or hybrid:

| Decision                              | Owner                       |
| ------------------------------------- | --------------------------- |
| Exact STOP keyword                    | Code                        |
| DNC natural-language intent           | AI proposes, code validates |
| Task lifecycle transition             | Code                        |
| Customer intent / objection / summary | AI                          |
| Store isolation / permission          | Code                        |
| Retry/backfill mode behavior          | Code contract               |
| Suggested next action                 | AI proposes                 |

### 4. Shared Mutation Layer

This is the stable architecture core:

| Module            | Owns                                                                                  |
| ----------------- | ------------------------------------------------------------------------------------- |
| Policy Guard      | allow/reject/needs\_review, DNC, RBAC, store isolation, state transition, idempotency |
| Task Orchestrator | create/update/close/reopen/record\_progress, task dedup, task source attribution      |
| Contact Writer    | contact identity, trust score, DNC sticky semantics, aggregate state updates          |
| Timeline Writer   | event catalog, payload schema, idempotency, actor/AI forensic fields                  |

`Timeline Writer` 有两种入口：source event projection 可以直接进入 `Timeline Writer`；business mutation audit 必须先经过 `Policy Guard` 和对应 writer/orchestrator。不要把所有 timeline event 都画成 `Policy Guard` 后面的副作用，也不要让 pipeline 绕过 `Timeline Writer` 自己拼 `contact_timeline` payload。

### 5. Storage / Projection

Target source-of-truth split:

| Table                  | Role                                             |
| ---------------------- | ------------------------------------------------ |
| `calls`                | per-call source record + AI call analysis result |
| `messages`             | per-message source record                        |
| `leads`                | per-lead source record                           |
| `contacts`             | customer aggregate profile and current summary   |
| `tasks`                | current work object snapshot                     |
| `task_progress_events` | task progress / attempt source of truth          |
| `contact_timeline`     | contact-level audit/feed projection              |

***

## Ideal Business Object Semantics

### Contact

Contact is not “what staff must do next”. Contact is the customer aggregate:

```text
identity + lifecycle + DNC + current profile + summary projections
```

`contacts.actionNeeded` can exist as a read-model / ranking summary, but the source of truth for human work should be `tasks`.

### Task

Task is a work objective:

```text
call this lead
save this cancellation
follow up after no-show
win back this former member
```

Task should not be closed just because one attempt failed.

### Task Progress

Progress records attempts:

```text
called_no_answer
left_voicemail
sent_sms
customer_requested_callback
staff_added_note
```

Progress is append-only; it does not replace task lifecycle.

### Timeline

Timeline is contact-level audit / feed projection:

```text
what happened to this customer across calls/messages/tasks/leads/AI
```

Timeline is not the canonical task progress table, but can mirror important progress for UI feed.

***

## Target AI Pattern

All AI actions follow this sequence:

```text
AI observes context
AI returns structured proposal
Code validates proposal against schema
Policy Guard accepts/rejects/needs_review
Shared module executes mutation
Timeline records what happened
```

AI never owns:

- final task state
- idempotency
- permission
- DNC hard stop
- store isolation
- transaction boundary
- audit trail

***

## One-time AI Call vs Tool Calling

Target state supports both.

### Batch / pipeline mode

Use one-time structured output:

```text
code gathers context
AI returns contact analysis + task action proposals
shared modules execute
```

Best for:

- daily contact analysis
- per-call follow-up trigger
- batch reprocess

### Interactive / agent mode

Use tool calling:

```text
AI asks for data via tools
AI proposes or calls allowed action tools
shared modules execute
```

Best for:

- staff asks “why is this task open?”
- AI assistant investigates a customer
- voice agent handles a live call
- admin asks to retry a specific processing run

Important: tool implementation should be thin wrappers around shared modules.

| Tool                   | Implementation                             |
| ---------------------- | ------------------------------------------ |
| `create_task`          | `TaskOrchestrator.create()`                |
| `record_task_progress` | `TaskOrchestrator.recordProgress()`        |
| `close_task`           | `TaskOrchestrator.close()`                 |
| `update_contact`       | `ContactWriter.update()`                   |
| `analyze_call`         | `CallAnalysisModule.analyzeCall()`         |
| `reanalyze_contact`    | `ContactAnalysisModule.reanalyzeContact()` |

***

## Target Workflow Examples

### Call completed

```text
Call event
  -> Call Analysis Module
  -> per-call AI result written to calls
  -> Timeline Writer records call source/analysis event
  -> Contact Analysis Module invoked
  -> AI proposes task/contact changes
  -> Policy Guard
  -> Contact Writer / Task Orchestrator
  -> Timeline Writer records business mutation audit
```

### SMS received

```text
SMS event
  -> message stored
  -> Timeline Writer records message.received
  -> SMS Signal Module
      exact STOP -> Policy Guard -> Contact Writer.setDnc + Task Orchestrator.closeAllOpenOutreach
      trivial -> no-op
      meaningful -> invoke Contact Analysis Module
      natural-language DNC -> AI proposes DNC, code validates
```

### Staff logs no answer

```text
Staff UI
  -> POST /tasks/:id/progress
  -> Policy Guard
  -> Task Orchestrator.recordProgress(no_answer)
  -> task_progress_events append
  -> optional dueAt update
  -> contact_timeline projection
```

### Future voice agent closes task

```text
Voice agent tool call close_task(...)
  -> Policy Guard checks actor capability + DNC + store + state
  -> Task Orchestrator.close
  -> timeline records actorType='ai_agent'
  -> low confidence path returns needs_review
```

***

## What “Adding a New Workflow” Should Feel Like

Target experience:

1. Define the event / trigger.
2. Decide if the decision is code-only, AI proposal, or hybrid.
3. Add or reuse a structured action schema.
4. Call shared modules.
5. Add tests and event catalog payload.

It should not require:

- designing a new ad hoc DB writer
- inventing new timeline payload conventions
- re-deciding DNC behavior
- re-deciding task dedup
- re-deciding store isolation
- teaching every prompt how to mutate database state

***

## Target Design Principles

- **Schema first**: stable business objects before prompts.
- **API first**: all objects readable and writable through typed contracts.
- **State machine first**: lifecycle transitions are deterministic.
- **Prompt last**: prompt output is proposal, not authority.
- **Shared mutation**: every writer of shared state goes through the same module.
- **Capability modules**: Call/SMS/Lead/contact analysis are reusable invocation contracts.
- **Audit by default**: every meaningful mutation emits timeline/progress event.
- **Agent-ready, not agent-first**: build stable tools by building stable modules first.

***

## Open Questions For Target State

| Question                                                            | Why it matters                                             |
| ------------------------------------------------------------------- | ---------------------------------------------------------- |
| Should `contacts.actionNeeded` be deprecated or kept as projection? | Prevents contact/task ownership confusion.                 |
| Should `processing_runs` become a first-class table?                | Retry/backfill/reconciliation need durable run history.    |
| How strict should AI agent autonomy be?                             | Some actions can auto-execute; others need human review.   |
| Should timeline payloads be versioned?                              | Event schema will evolve; old events must remain readable. |
| What is the prod migration sequence?                                | Prod legacy may differ materially from test/current code.  |
| How should cost budgets work for SMS and tool calling?              | Prevents unbounded multi-turn AI cost.                     |

***

## North Star

The north star is simple:

**Every workflow, whether triggered by Lambda, UI, API, retry job, or future AI agent, should mutate shared business state through the same typed, audited, policy-guarded modules.**