> For AI agents: the complete documentation index is available at /llms.txt, the full documentation bundle is available at /llms-full.txt. # Unified Pipeline 现状调研 (Codex) > **当前状态**: Current-state research。只描述现在从本地代码和已有文档能验证到的事实，不把 target design 当作已落地实现。 > **日期**: 2026-05-31 > **Canonical brief**: [Unified Pipeline Architecture 设计 Brief](/product-design/v2/unified-pipeline/research/unified-pipeline-design-brief.md) > **下一步设计**: [Unified Pipeline Phase 1 (Codex)](/product-design/v2/unified-pipeline/original/unified-pipeline-phase-1-codex.md) > **理想终态**: [Unified Pipeline Target State (Codex)](/product-design/v2/unified-pipeline/original/unified-pipeline-target-state-codex.md) *** ## Read This First 这份文档不是按“我怎么查代码”的顺序写，而是按读者做判断的顺序写。 先看结论： **当前代码已经有不少 unified architecture 的局部基础，但这些基础散在不同 Lambda / API route 里，还没有变成稳定 shared modules。** 最重要的 current-state picture 是： ```mermaid flowchart TD Call[Call Analysis] --> C1[Inline contacts + timeline write] Contact[Contact Analysis] --> C2[Inline contacts + tasks + timeline write] SMS[SMS Processor] --> C3[Inline messages + contacts + timeline + STOP cascade] Lead[Lead Processor] --> C4[Inline contacts + tasks + timeline write] Staff[Studio API] --> C5[Inline tasks + timeline write] C1 -. should converge .-> Shared[Shared Mutation Modules] C2 -. should converge .-> Shared C3 -. should converge .-> Shared C4 -. should converge .-> Shared C5 -. should converge .-> Shared Shared --> Task[Task Orchestrator] Shared --> ContactWriter[Contact Writer] Shared --> Timeline[Timeline Writer] Shared --> Guard[Policy Guard] ``` Reader takeaway: | Question | Answer | | -------------------------- | ------------------------------------------------------------------------------ | | 是否需要重写 pipeline runtime？ | 不需要。现有 SQS/Lambda 继续用。 | | 是否应该现在做 tool calling？ | 不应该。先把 shared modules 做出来。 | | 最大的 current gap 是什么？ | task/contact/timeline mutation logic 散在多个 writer 里。 | | current code 有没有证明这个方向是对的？ | 有。`contacts-analyzer`、SMS STOP、timeline idempotency、storeId guard 已经在局部实践这个模式。 | | 这份 research 的作用是什么？ | 给 Phase 1 design 提供事实依据，不是最终架构方案。 | 细节、evidence 和逐项 audit 放在后面。 *** ## Scope / Source of Truth 这份 research 的事实来源是本机代码和现有 docs： | Layer | Source | | ------------------------- | -------------------------------------------------------------------------------------------------------------------------- | | DB schema | `callytics-common/src/db/schema/*.ts` | | Call pipeline | `callytics-infrastructure/lambda/ai-analysis-processor/src/**` | | Contact analysis pipeline | `callytics-infrastructure/lambda/contacts-analyzer/src/**` | | SMS pipeline | `callytics-infrastructure/lambda/message-processor/src/**` | | Lead downstream | `callytics-infrastructure/lambda/lead-processor/src/core/persist-downstream.ts` and `lead-tracking/src/neon-repository.ts` | | Studio API | `studio-website-monorepo/apps/api/src/routes/tasks/**` | | Target design docs | `docs/product-design/v2/**` | **没有做的事**： - 没有查询 live prod DB。 - 没有查询 live test DB。 - 没有查 CloudWatch / AWS runtime logs。 所以本文里的 `Current code` 指本地 workspace 的当前实现；`Prod legacy` 只作为风险层存在，不能从本文推断 prod 已经长这样。 *** ## Executive Summary 当前代码已经部分朝 unified architecture 靠近，但还没有形成真正的 shared module 层。 已经有的好东西： - `contacts` 已以 `(phone, storeId)` 为 identity，`name trust score` 也进入 common schema。 - `tasks` 已有 `uq_tasks_pending_contact_category`，能防同一 `(contactPhone, storeId, typeCategory)` 下重复 pending task。 - `contact_timeline` 已是跨 call/message/lead/task/contact/AI 的统一 audit/feed 表，带 `idempotencyKey` 和 AI forensic fields。 - `contacts-analyzer` 已经是 `AI proposes, code executes`：AI 输出 `taskDecisions[]`，代码验证 taskId、storeId、DNC、pending status 后执行。 - `message-processor` 已有 exact inbound STOP 的 code-only DNC + task close cascade。 - `ai-analysis-processor` 已有 `tryClaimAnalysis()`，用 DB timestamp 做 per-call concurrency claim。 还没有统一的部分： - 没有独立的 `Task Orchestrator`。task mutation 逻辑散在 `contacts-analyzer`、`lead-processor`、`studio-api`、`message-processor` DNC cascade。 - 没有独立的 `Contact Writer`。trust score / DNC / lastActivityAt / storeId guard 在多个 repo 里重复实现。 - 没有独立的 `Timeline Writer`。event payload / idempotency key / actor fields 各 writer 自己组。 - 没有 productized `Policy Guard`。DNC、store isolation、AI hallucination guard、staff override guard 是局部代码规则。 - `task_progress_events` 还没有 current schema；现在 progress-like values 仍混在 `TASK_CLOSE_RESULT` 和 task close/reopen/postpone flow 里。 - Call/SMS/Lead 已经有 processing capability 的雏形，但还没有统一 invocation contract。 当前最准确的判断： **Phase 1 不应该重写 SQS/Lambda runtime，也不应该直接做 tool calling。Phase 1 应该把已经散落在代码里的 deterministic mutation / guard / audit 逻辑提炼成 shared modules，让现有 pipelines 逐步接入。** *** ## Environment Reality 这次设计必须把三层分开： | Layer | 怎么看 | | ----------------- | ------------------------------------------------------------ | | Target docs | 未来想设计成什么样。可以作为方向，但不是 implementation fact | | Test/current code | 新代码主要落在 test 环境，代表正在建设的新世界，但不等于完成态 | | Prod legacy | 生产是老东西，可能 schema、pipeline、prompt、API 都不同；未查 live prod 前不能下结论 | 后续任何 design doc 都应该标注： ```text Target intent: Current test/code reality: Prod reality: Gap: Decision: ``` *** ## Current Object Model | Object | Current code reality | Target pressure / gap | | ------------------ | ------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------- | | `contacts` | Aggregate profile。PK 是 `(phone, storeId)`；含 lifecycle、DNC、lead status、action summary、suggested actions、name trust。 | 职责偏宽：既是 profile，又有 AI action summary。未来应定义哪些字段是 source of truth，哪些是 task/read-model projection。 | | `tasks` | Work object。状态只有 `pending` / `closed`；`closeResult` 18 值里混有 final outcome 和 progress-like values。 | 需要 `record_progress` / `task_progress_events`，避免 `no_answer` / `left_voicemail` 关闭任务再新建 follow-up。 | | `contact_timeline` | Contact-level audit/feed。已有 16+ event types、actor identity、AI forensic fields、idempotency unique index。 | 需要 payload catalog 和 writer abstraction，否则 event payload 会继续各写各的。 | | `calls` | Per-call source record + AI call classification。`callState`、`followUpNeeded`、`primaryOutcomeResult`、AI cost、pipeline timestamps 已在表内。 | 不应该直接创建 tasks；但需要 formal `analyze_call(callId, mode)` invocation contract 支持 retry/backfill。 | | `messages` | RC message-store source record。SMS body 在 `subject`；voicemail transcription 有独立字段。 | exact STOP 已处理；meaningful SMS / natural-language DNC 还没有 AI workflow。 | | `leads` | Inbound lead source record，dedup key 是 primary key。 | Downstream contact/task/timeline 是 deterministic，但当前直接写表；应接入 shared mutation contract。 | *** ## Current Pipeline Inventory | Pipeline | Trigger | AI? | Current writes | Current guard / idempotency | | ----------------- | -------------------------------------- | --------------------------------------------- | -------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------- | | Call Analysis | S3 transcription / SQS | pre-triage code + triage/classify/coaching AI | `calls`, `contacts`, `contact_timeline`; then triggers contacts-analyzer | `tryClaimAnalysis()` prevents concurrent same-call analysis; `call_analysis.completed:${callId}` timeline idempotency | | Contact Analysis | EventBridge cron fan-out + SQS sources | 1 contact-level prompt | `contacts`, `tasks`, `contact_timeline`, `calls.contactsCompletedAt` | Zod output schema; taskId/pending/storeId guards; DNC code cascade; timeline idempotency | | SMS | RingCentral message SQS | No SMS AI | `messages`, `contacts`, `contact_timeline`; exact STOP also closes pending tasks | message UPSERT; `message.created:${messageId}` timeline idempotency; exact STOP code filter | | Lead | IMAP / lead event / EventBridge | No | `leads`, then downstream `contacts`, `tasks`, `contact_timeline` | lead primary key dedup; downstream uses `sourceLeadId` and timeline idempotency | | Studio API | Human UI | No | `tasks`, `contact_timeline` | store authorization; VIEWER cannot mutate; store filter by `store_phone` or `store_id` | | Contact reprocess | reprocess Lambda / SQS | Reuses contact-analyzer | `reprocess_run_items` plus normal contact analysis writes | `source='reprocess'` requires `reprocessMode` + `reprocessRunId`; cooldown logic exists | *** ## Writer Matrix | Shared table | Current writers | What is already good | What is not unified | | ------------------ | ------------------------------------------------------------------------------------------------------------------------------ | ----------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------ | | `contacts` | `ai-analysis-processor`, `contacts-analyzer`, `message-processor`, `lead-processor`, legacy `lead-tracking`, staff edit routes | Most writers now understand `(phone, storeId)` and trust score. DNC sticky behavior exists in contact analysis and STOP path. | No shared `Contact Writer`; trust / DNC / lastActivityAt / storeId unresolved policy still duplicated. | | `tasks` | `contacts-analyzer`, `lead-processor`, `studio-api`, `message-processor` DNC cascade | DB has pending dedup by `(contactPhone, storeId, typeCategory)` and lead idempotency by `sourceLeadId`. | No `Task Orchestrator`; close/update/create semantics differ by caller; progress is not first-class. | | `contact_timeline` | Call, contact analysis, SMS, lead, studio-api | Unified table is strong; idempotency index exists; actor and AI forensic fields exist. | No `Timeline Writer`; event payload schema/catalog is not centralized. | *** ## Current Code vs Target Design | Target design concept | Current code status | Research judgment | | ----------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------- | | Task Orchestrator | Not implemented as a module. Logic exists inside multiple writers. | Needed for Phase 1. This is not premature because code already has 4 task mutation callers. | | Contact Writer | Not implemented as a module. Trust score constants are centralized, but write logic is repeated. | Needed, but can land after Task Orchestrator or as a narrow wrapper first. | | Timeline Writer | Not implemented as a module. `buildTimelineValues` / `buildContactTimelineInsertSQL` helpers exist, but each caller chooses payload/idempotency. | Needed early because task/contact writers need audit consistency. | | Policy Guard | Not implemented as a named layer. Some guards exist locally. | Should be introduced as shared validation/policy functions, not a separate service. | | Processing Capability Modules | Partial. Call analysis, contact reprocess, SMS STOP, lead downstream exist as capabilities. | Need invocation contracts, not new tables/writers. | | Tool calling | Not implemented. | Not Phase 1. Shared modules should make future tools thin wrappers. | | `task_progress_events` | Not in common schema from local search. | Key Phase 1 schema addition if we want `record_progress`. | *** ## Current Workflow Diagram ### Current: Code / Prompt / Writer Mixed ```mermaid flowchart TD RC[RingCentral event] --> Transcribe[Code: transcribe / call row] Transcribe --> CallAI[Prompt: per-call triage/classify/coaching] CallAI --> CallWriter[Code: ai-analysis Neon repo writes calls + contacts + timeline] CallWriter --> ContactSQS[Code: enqueue contacts-analyzer] ContactSQS --> ContactAI[Prompt: contact analyzer returns contact fields + taskDecisions] ContactAI --> ContactWriterInline[Code: contacts-analyzer repo updates contacts] ContactAI --> TaskWriterInline[Code: contacts-analyzer repo creates/updates/closes tasks] ContactAI --> TimelineInline[Code: contacts-analyzer repo writes timeline] SMS[SMS event] --> MessageWriter[Code: message-processor writes messages + contacts + timeline] SMS --> StopFilter[Code: exact STOP] StopFilter --> DncCascade[Code: shared DNC cascade closes tasks] Lead[Lead event] --> LeadWriter[Code: lead-processor writes contacts + tasks + timeline] Staff[Staff UI] --> StudioTaskRoutes[Code: studio-api close/reopen/postpone writes tasks + timeline] ``` ### Missing Layer ```mermaid flowchart TD CallAI[Call AI] --> InlineA[Inline contact/timeline write] ContactAI[Contact AI] --> InlineB[Inline contact/task/timeline write] SMS[SMS STOP] --> InlineC[Inline DNC + task close] Lead[Lead] --> InlineD[Inline contact/task/timeline write] Staff[Staff UI] --> InlineE[Inline task/timeline write] InlineA -. should call .-> Shared[Shared Mutation Modules] InlineB -. should call .-> Shared InlineC -. should call .-> Shared InlineD -. should call .-> Shared InlineE -. should call .-> Shared ``` *** ## Specific Findings ### 1. Task is already central, but task operations are scattered Task creation / update / close currently happens through: - `contacts-analyzer` AI `taskDecisions[]` - `lead-processor` deterministic `lead_outreach` - `studio-api` manual close/reopen/postpone - `message-processor` exact STOP DNC cascade This validates the architecture premise: `Task Orchestrator` is not a task page helper. It is a shared mutation module. ### 2. Current close API conflicts with the target progress model `PATCH /v2/tasks/close` closes the current task for `no_answer` / `left_voicemail`, then auto-creates a follow-up task. That works in current code, but conflicts with the target model: ```text no_answer / left_voicemail should be record_progress task remains open attempt count comes from task_progress_events dueAt may be moved ``` This should be one of the first Phase 1 contract decisions. ### 3. SMS is not only storage anymore The old shorthand "message-processor only stores messages" is stale. Current code: - stores `messages` - UPSERTs `contacts` - writes `message.created` timeline - exact inbound `STOP` sets DNC - exact STOP closes pending tasks through DNC cascade Still missing: - meaningful SMS detection - natural-language DNC beyond exact STOP - immediate contact analysis trigger for meaningful replies ### 4. Contact analysis already demonstrates the right AI pattern `contacts-analyzer` is already close to the desired pattern: ```text AI proposes taskDecisions[] Code validates schema with Zod Code validates taskId belongs to current contact/store Code blocks create/update under DNC Code performs task mutation + timeline insert ``` The issue is not conceptual. The issue is that this pattern lives inside one Lambda repository instead of a reusable module. ### 5. Call analysis needs invocation abstraction, not writer unification first `calls` is mostly written by the call analysis pipeline itself. The problem is not "many writers fighting over calls"; the problem is "the call analysis capability should be callable by normal flow, retry, backfill, reconciliation". So Call Analysis Module should be: ```text analyze_call(callId, mode) ``` not a `Call Writer`. ### 6. Store identity is the sharpest current boundary Current code has strong movement toward `storeId` as identity, but unresolved `storeId` behavior still differs: - `contacts-analyzer`: skip entire write if `storeId` is missing. - `message-processor`: can degrade to messages-only if no contact identity, but skips full batch when contact write is expected and `storeId` unresolved. - `lead-processor`: skips downstream if `storeId` is null. - `ai-analysis-processor`: skips entire call write pipeline when contact was resolved but `storeId` null. This should become an explicit `StoreIdentityPolicy`, not repeated comments in each repo. *** ## Current Gaps That Should Feed Phase 1 | Gap | Phase 1 implication | | ------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------- | | No `Task Orchestrator` | Build shared `applyTaskAction()` contract first. | | No `task_progress_events` | Add schema and route before changing close semantics. | | No central Timeline Writer | Create payload/idempotency helpers before migrating more writers. | | No central Contact Writer | Start with trust-score upsert + DNC sticky helper; migrate writer by writer. | | Mixed `source` / `reprocessMode` / future `intentKind` | Define `ProcessingIntent` contract even if it is initially only TypeScript. | | Contact reprocess table not in common schema | Decide whether `reprocess_run_items` remains implementation-private or becomes a common processing ledger. | | Prod not verified | Do not design destructive migration until prod schema/runtime is explicitly audited. | *** ## What This Research Does Not Decide - Whether prod should be migrated immediately. - Whether to implement tool calling now. - Whether every writer must migrate in one release. - Whether `contacts.actionNeeded` should be removed or kept as projection. Those belong in Phase 1 design and target architecture, not in current-state research.