#Unified Pipeline Research Appendix

当前状态: Research appendix。Canonical 入口请读 Unified Pipeline Architecture 设计 Brief。本文件的代码调研结论仍然有用，但 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 做一轮快速代码调研，判断它是否会自然收敛到 Unified Pipeline Architecture 设计 Brief 里说的”统一 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：

#Current State Inventory

#1. Business objects

#2. Prompt surfaces

当前主要 prompt surface：

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

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

#3. Writers

多条 pipeline 直接写 shared tables：

#Key Findings

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

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

真正缺的是：

shared mutation / orchestration layer

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

所以：

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

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

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

这说明旧 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，但必须定义语义：

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。

建议：

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 只存储，不处理” 已经不准确。更准确是：

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 历史上只调用：

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 写入流程。

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

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：

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：

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 本身也支持这个原则：

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

Problem:

Every pipeline knows how to write shared business state.

#Target State

Goal:

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:

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:

shared modules + Code as Guardrail + future tool compatibility

But it should be reframed:

统一 pipeline

should become:

统一 shared mutation layer / pipeline write model

#Final judgment

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

The research confirms that the right direction is:

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?