#Unified Pipeline 现状调研 (Codex)

当前状态: Current-state research。只描述现在从本地代码和已有文档能验证到的事实，不把 target design 当作已落地实现。 日期: 2026-05-31 Canonical brief: Unified Pipeline Architecture 设计 Brief 下一步设计: Unified Pipeline Phase 1 (Codex) 理想终态: Unified Pipeline Target State (Codex)

#Read This First

这份文档不是按“我怎么查代码”的顺序写，而是按读者做判断的顺序写。

先看结论：

当前代码已经有不少 unified architecture 的局部基础，但这些基础散在不同 Lambda / API route 里，还没有变成稳定 shared modules。

最重要的 current-state picture 是：

Reader takeaway:

细节、evidence 和逐项 audit 放在后面。

#Scope / Source of Truth

这份 research 的事实来源是本机代码和现有 docs：

没有做的事：

• 没有查询 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

这次设计必须把三层分开：

后续任何 design doc 都应该标注：

Target intent:
Current test/code reality:
Prod reality:
Gap:
Decision:

#Current Object Model

#Current Pipeline Inventory

#Writer Matrix

#Current Code vs Target Design

#Current Workflow Diagram

#Current: Code / Prompt / Writer Mixed

#Missing Layer

#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:

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:

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:

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

#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.