#Unified Pipeline Phase 1 (Codex)

当前状态: Phase 1 implementation design。基于 Current State Research，定义第一阶段应该怎么落地。 日期: 2026-05-31 Canonical brief: Unified Pipeline Architecture 设计 Brief 理想终态: Unified Pipeline Target State (Codex)

#Final Design First

Phase 1 最终应该长这样：

一句话：

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：

#Phase 1 Decision Summary

如果只读这一节，读者应该已经知道 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。

换句话说：

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 内部：

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：

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,
})

返回：

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

#Phase 1C: Add task_progress_events

新增 task-level progress source of truth：

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 可以很保守：

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：

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：

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

必须覆盖：

#Caller Migration Order

#1. Studio API task routes

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

关键行为改动：

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

#2. Contacts-analyzer taskDecisions[]

当前 taskDecisions[]：

create | update | close

Phase 1 target：

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：

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：

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：

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

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

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

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

#Contact Analysis Reprocess

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

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

#Test Plan

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

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