#Unified Pipeline Phase 1 — Implementation Plan

Type: Implementation plan(基于 unified-pipeline-final.md 高层设计 + PR #368 收口的 5 条 contract 决策) Date: 2026-06-02 Status: Draft — 等 Max review Test env only: 所有改动在测试环境,无 prod migration 压力 依赖:

• PR #368 merge(unified-pipeline-final.md + tasks-schema.md + contacts-schema.md 最新版作为 spec SoT)
• prompt engineer 交付拆分后的 Contact Profile + Task Decision prompt(action 命名:create_open / create_closed / close / update / record_progress / reopen)
• studio-api Drizzle 迁移 issue #449不阻塞 Phase 1(本 plan 先走 raw SQL helper,Drizzle 迁移后续独立 PR 删除)

#§1. Phase 1 Goals

把现有散落在 4 个 Lambda + studio-api 的 task / contact / timeline mutation 收口到 callytics-common shared modules,达成 3 个 outcome:

1. task_progress_events 表上线 —— no_answer / left_voicemail / text_sent 进度从 closeResult 里拆出来。
2. applyTaskAction() 共享 module —— 6 个 action 由统一 state machine 处理,人 + AI 走同一接口,并发幂等由 SQL 层 enforce。
3. tasks.store_id NOT NULL —— tenant isolation 在 DB-level 强制。

明确 不在 Phase 1:拆 prompt(并行外包)/ tool calling runtime / SMS meaningful 分类 / studio-api Drizzle 迁移 / contacts.actionNeeded DROP COLUMN(Phase 2)/ Contact Writer 18 个 AI 画像字段统一(Phase 2)。

S4 — Phase 1 scope = backend only:POST /v2/tasks/:taskId/progress 端点 ship 后,Phase 1 不改前端 UI(Status Update 按钮 / Progress Update 区域 / API 调用)。前端改动作为 Phase 1.5(独立 PR + 团队),依赖 Phase 1 后端 stable 1 周后开工。Phase 1 期间 staff 通过 close API 用 closeResult ∈ progress_set 触发的旧行为已被禁(返回 400)—— 在前端没 ship 前,临时 fallback:close API 收到 progress closeResult 时返回 400 + 提示"用 progress endpoint",staff 需手动 curl 或等前端。test env 接受这个临时体验。

#§2. Final Schema Delta(Phase 1 end state)

本节描述 Phase 1 完成后 的 final schema。PR 数量不是设计约束;实现可以用一个大 PR 或少量 PR,但只要存在旧代码和新代码同时运行的窗口,就必须先走 transition-compatible migration,再收缩到本节 final 状态。

过渡期间保留旧写法兼容:TASK_STATUS = ['pending','open','closed'],TASK_CLOSE_RESULT 同时含旧 18 值 + 新 unable_to_reach,tasks.action_needed / tasks.task_type 字段仍存在。最终 cutover 后才收缩到本节 final 状态。

#2.1 新建 task_progress_events 表

CREATE TABLE task_progress_events (
  event_id          uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  task_id           uuid NOT NULL REFERENCES tasks(task_id),
  store_id          text NOT NULL,                    -- 冗余,避免 join 查 store guard
  contact_phone     text NOT NULL,                    -- 冗余,task-progress query 用
  progress_type     text NOT NULL CHECK (progress_type IN (
                      'no_answer', 'left_voicemail', 'text_sent',
                      'callback_requested', 'follow_up_scheduled', 'customer_considering'
                    )),
  channel           text NOT NULL CHECK (channel IN ('phone', 'sms', 'voicemail', 'email')),
  actor_type        text NOT NULL CHECK (actor_type IN ('staff', 'system', 'ai_agent', 'contact_analysis')),
  actor_id          uuid NULL,
  call_id           text NULL,                        -- evidence reference,no FK(calls 表 PK 是 telephony_session_id,call 可能延迟写入)
  message_id        bigint NULL,                      -- evidence reference,no FK(messages.id bigint,且 SMS event 可能在 message 行落库前到达 — 不引入时序耦合;upstream tasks-schema.md 列了 FK→messages.id,本 spec 故意去掉以避免 race)
  note              text NULL,
  next_due_at       timestamptz NULL,
  occurred_at       timestamptz NOT NULL,
  idempotency_key   text NOT NULL UNIQUE,             -- 唯一去重键(见 §3.2 生成规则)
  created_at        timestamptz NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_tpe_task_occurred ON task_progress_events (task_id, occurred_at DESC);
CREATE INDEX idx_tpe_store_occurred ON task_progress_events (store_id, occurred_at DESC);
CREATE INDEX idx_tpe_actor ON task_progress_events (actor_type, actor_id, occurred_at DESC);
CREATE INDEX idx_tpe_type_channel ON task_progress_events (progress_type, channel, occurred_at DESC);

#2.2 tasks 表字段改动

改名 + DEFAULT:

ALTER TABLE tasks ALTER COLUMN status DROP DEFAULT;
UPDATE tasks SET status='open' WHERE status='pending';
ALTER TABLE tasks ALTER COLUMN status SET DEFAULT 'open';

-- CHECK constraint 更新
ALTER TABLE tasks DROP CONSTRAINT IF EXISTS chk_tasks_closed_integrity;
ALTER TABLE tasks ADD CONSTRAINT chk_tasks_closed_integrity CHECK (
  (status = 'open' AND close_type IS NULL AND closed_at IS NULL)
  OR
  (status = 'closed' AND close_type IS NOT NULL AND closed_at IS NOT NULL)
);

-- partial unique index 同步
DROP INDEX IF EXISTS uq_tasks_pending_contact_category;
CREATE UNIQUE INDEX uq_tasks_pending_contact_category 
  ON tasks (contact_phone, store_id, type_category)
  WHERE status = 'open' AND store_id IS NOT NULL;

新增字段:

ALTER TABLE tasks ADD COLUMN executor_type text NULL CHECK (
  executor_type IS NULL OR executor_type IN ('human', 'ai_agent', 'system')
);
ALTER TABLE tasks ADD COLUMN attempt_count integer NOT NULL DEFAULT 0;
ALTER TABLE tasks ADD COLUMN source_call_id text NULL;

-- create_closed 防重 partial unique
CREATE UNIQUE INDEX uq_tasks_create_closed_evidence 
  ON tasks (contact_phone, store_id, type_category, source_call_id)
  WHERE status = 'closed' AND source_call_id IS NOT NULL;

closeType + closeResult enum 扩展(schema.ts const 改;DB 用 text 不是 enum type,无 migration SQL):

• TASK_CLOSE_TYPE:加 create_closed
• TASK_CLOSE_RESULT:18 → 15,移出 no_answer / left_voicemail / callback_later / attempted,新增 unable_to_reach
• CHECK (close_result IS NULL OR close_result IN (...)) 同步更新

tasks.store_id NOT NULL:

-- 测试环境直接 DELETE NULL 老 task(无价值数据,不 backfill)
DELETE FROM tasks WHERE store_id IS NULL;
ALTER TABLE tasks ALTER COLUMN store_id SET NOT NULL;

-- partial unique 删 store_id IS NOT NULL 条件
DROP INDEX IF EXISTS idx_tasks_store_id;
CREATE INDEX idx_tasks_store_id ON tasks (store_id);

字段退役(DROP COLUMN):

ALTER TABLE tasks DROP COLUMN action_needed;
ALTER TABLE tasks DROP COLUMN task_type;
DROP INDEX IF EXISTS idx_tasks_action_needed;
-- chk_tasks_task_type_category_consistency CHECK 一起删
ALTER TABLE tasks DROP CONSTRAINT IF EXISTS chk_tasks_task_type_category_consistency;

#2.3 contacts 表字段改动

contacts.lifecycleState 保留(撤销退役,2026-06-02 engineer feedback):无 SQL 改动。churned re-engage 等需 AI 语义判断,不可纯派生;lead_declined close result 设计依赖该字段。

contacts.actionNeeded Phase 1 改读路径(字段保留,Phase 2 才 DROP):

只改 query 不改 schema。详见 §4 caller migration 5e。

#2.4 callytics-common schema 文件更新

#§3. Module Contracts

#3.1 applyTaskAction() + closeAllOpenForContact() —— Task Orchestrator(Drizzle 版)

位置:callytics-common/src/domain/task-orchestrator.ts

Action 类型(6 action + 1 bulk helper):

import type { SQL } from 'drizzle-orm';

export type TaskAction =
  | { action: 'create_open'; payload: CreateOpenPayload }
  | { action: 'create_closed'; payload: CreateClosedPayload }    // 必带 source_call_id
  | { action: 'close'; payload: ClosePayload }                    // 必带 taskId
  | { action: 'update'; payload: UpdatePayload }                  // 必带 taskId
  | { action: 'record_progress'; payload: RecordProgressPayload } // 必带 taskId
  | { action: 'reopen'; payload: ReopenPayload };                 // 必带 taskId

export type ApplyResult =
  | {
      status: 'allow';
      statements: SQL[];       // caller 放进 db.batch() 执行(Drizzle 自动事务)
      resultChecks?: Array<{
        statementIndex: number;
        zeroRowsReason: RejectReason;
      }>;                      // executor 必须把 RETURNING 0 rows 映射为 reject
      generatedIds?: { taskId?: string; idempotencyKey?: string };
    }
  | { status: 'reject'; reason: RejectReason; details: string }
  | { status: 'needs_review'; details: string };  // Phase 1 不返回这个,Phase 2 预留

export type RejectReason =
  | 'dnc'                       // contacts.do_not_contact = true
  | 'store_mismatch'            // taskId 引用的 task 跨 store
  | 'low_confidence'            // AI proposal confidence < threshold
  | 'stale_proposal'            // SQL conditional WHERE updated_at <= $aiRunStartedAt 0 rows(见 §3.6)
  | 'duplicate'                 // partial unique 冲突
  | 'task_not_open'             // close / update / record_progress 对非 open task
  | 'task_not_closed'           // reopen 对非 closed task
  | 'create_closed_with_open_task'  // S3: create_closed 时已有 open task(应改用 close)
  | 'invalid_state_transition'
  | 'hallucinated_task_id'      // taskId 不存在 / 不属于该 contact
  | 'invalid_type_category'     // AI 选了不在 allowed set 的 typeCategory(见 §3.6 S2)
  | 'invalid_progress_type'
  | 'invalid_close_result'
  | 'close_note_required';      // closeResult='other' 时 closeNote 必填(codex §3)

ActionContext 拆 2 type(S6 fix —— contactPhone 必传规则统一):

type BaseActionContext = {
  storeId: string;              // 必传,Policy Guard enforce
  actor: { type: 'staff' | 'system' | 'ai_agent' | 'contact_analysis'; id?: string; name?: string };
  aiRunStartedAt?: Date;        // AI 提案才传,human authority guard 用
  db: DrizzleClient;            // caller 注入
};

// create_open / create_closed —— caller 必传 contactPhone(新建 task)
export type CreateActionContext = BaseActionContext & {
  contactPhone: string;
};

// close / update / record_progress / reopen —— caller 只传 taskId,Orchestrator 从 DB 反查 contactPhone
export type TaskIdActionContext = BaseActionContext;

export type ActionContext = CreateActionContext | TaskIdActionContext;
// Orchestrator 内部按 action.action 决定 narrow type

Function signatures:

// 单 action(6 种)
export async function applyTaskAction(
  action: TaskAction,
  ctx: ActionContext,
): Promise<ApplyResult>;

// B7: bulk close,DNC cascade / 重复用例
// 关闭某 contact 所有 open tasks,统一 closeType='auto_closed' + 同一 closeResult
export async function closeAllOpenForContact(
  params: {
    contactPhone: string;
    storeId: string;
    closeResult: TaskCloseResult;
    closeNote: string;
    actor: BaseActionContext['actor'];
  },
  ctx: { db: DrizzleClient },
): Promise<{
  status: 'allow';
  statements: SQL[];       // 1 个 UPDATE 关多 task + N 个 timeline INSERT
  closedTaskIds: string[];
} | { status: 'reject'; reason: RejectReason; details: string }>;

Policy Guard 检查顺序(每个 action 进 SQL 前 —— 详细见 §3.6):

1. storeId 非空
2. contacts.doNotContact 查询 —— true 时(B6 fix):reject create_open / create_closed / update / record_progress / reopen;只允许 close 单 action 和 closeAllOpenForContact() bulk。DNC 语义是"停止主动触达 + 关闭 open tasks":record_progress 等于继续触达;reopen 把 closed task 变回 open work 等于恢复触达 —— 两者 DNC 下都不应触发
3. taskId 引用合法 —— close / update / record_progress / reopen 提案的 taskId 必须存在 + 属于当前 (contactPhone, storeId),不接受跨 store 的 taskId(hallucination guard)
4. AI proposal confidence 检查(如有)—— 低于 threshold 直接 reject low_confidence + log + CloudWatch metric
5. State transition 检查 —— close / update / record_progress 对 open task;reopen 对 closed task
6. S3 — create_closed 时:SELECT 1 FROM tasks WHERE contact_phone=? AND store_id=? AND type_category=? AND status='open' 找到则 reject create_closed_with_open_task(应改用 close 那个 task,不该新建 closed,见 codex §9)
7. S2 — typeCategory allowed set(create_open / create_closed):仅约束 AI proposal(actor.type='contact_analysis' | 'ai_agent')。AI 选的 typeCategory 必须在 computeAllowedTypeCategories(contact) 返回值内(见 §3.6)。deterministic caller(如 lead-processor 创建 lead_outreach)不走这个 AI allowed-set,但仍走 store / DNC / duplicate guard
8. closeNote 检查:closeResult='other' 时 closeNote 必填,缺则 reject close_note_required(codex §3)
9. SQL 层 race guard 由 statements 自带(partial unique / conditional WHERE / CTE RETURNING),ApplyResult.resultChecks 声明哪些 statement 的 0 rows 有业务含义。执行层必须检查 rows count 并映射 reject,例如 close/update 的 AI stale path → stale_proposal,create duplicate → duplicate。不能把 0 rows 当 silent success。

S1 — tasks.priority 派生位置:create_open / create_closed payload 收 suggestedActions: SuggestedAction[],Orchestrator 内部算 priority = max(suggestedActions[].priority)(顺序 high > medium > low)→ 写入 tasks.priority 列。AI 不再输出 task 级 priority,只输出每个 suggestedAction 的 priority。computeDueAt(priority) 用算出来的 task priority(现有共享 helper)。

6 action SQL contract(详见 unified-pipeline-final §Step 2 表)。

#3.2 buildTaskActionSQL() + buildCloseAllOpenSQL() —— raw SQL 版给 studio-api

位置:callytics-common/src/domain/task-orchestrator-sql.ts

Signature:

export function buildTaskActionSQL(
  action: TaskAction,
  ctx: ActionContext,
): Promise<ApplyResultSQL>;

export function buildCloseAllOpenSQL(
  params: {...},  // 同 closeAllOpenForContact
  ctx: { sql: NeonHttpClient },
): Promise<ApplyResultSQLBulk>;

type ApplyResultSQL =
  | {
      status: 'allow';
      statements: { sql: string; params: unknown[] }[];
      resultChecks?: Array<{ statementIndex: number; zeroRowsReason: RejectReason }>;
      generatedIds?: { taskId?: string; idempotencyKey?: string };
    }
  | { status: 'reject'; reason: RejectReason; details: string }
  | { status: 'needs_review'; details: string };

B2 fix — caller 必须用 sql.transaction() 包裹 statements(原子事务):

// studio-api close.ts 调用模式(B2 强制)
import { neon } from '@neondatabase/serverless';
const sql = getNeonClient();

const result = await buildTaskActionSQL({ action: 'close', payload: {...} }, ctx);
if (result.status === 'reject') { ... }

// ❌ 错(plan 旧版):多次 sql.query 不是事务,progress 写了 timeline 写一半失败 = 数据不一致
// for (const stmt of result.statements) await sql.query(stmt.sql, stmt.params);

// ✅ 对:用 sql.transaction(neon-http 支持)
await sql.transaction(
  result.statements.map(stmt => sql(stmt.sql, stmt.params))
);

neon-http transaction 行为:@neondatabase/serverless 的 sql.transaction() 把多 statement 包成单个 HTTP request,服务端用真 Postgres BEGIN/COMMIT。多 statement 要么全成功要么全 rollback。

实现策略:沿用现有 buildContactTimelineInsertSQL() 风格:raw SQL helper 显式返回 { sql, params },共享 Policy Guard / idempotency_key / due-date 等纯 helper,只在最终 SQL rendering 处和 Drizzle path 分开。禁止用 Drizzle SQL.toQuery() / as any 把 Drizzle SQL 对象转成 raw SQL string;这不是稳定 public API,失败时可能产生空 SQL。

idempotency_key 生成规则(两个 helper 共享):

source_call_id 生成规则(create_closed 专属):

• AI 提案:payload.sourceCallId(prompt engineer 在 Task Decision 输出里包含)
• staff 直接 create_closed via API:payload.sourceCallId(API 参数)
• lead-processor 不用 create_closed,不涉及

B8 fix — record_progress SQL 用 CTE 防 attempt_count 重复 increment:

-- ❌ 错(plan 旧版):INSERT ON CONFLICT DO NOTHING 不插也不报错,UPDATE 仍跑 → attempt_count + 1
-- 重复 progress = attempt_count 翻倍
INSERT INTO task_progress_events (..., idempotency_key) VALUES (...) ON CONFLICT (idempotency_key) DO NOTHING;
UPDATE tasks SET attempt_count = attempt_count + 1, due_at = $newDueAt WHERE task_id = $1;

-- ✅ 对:CTE 让 UPDATE 只在真插入时跑
WITH inserted AS (
  INSERT INTO task_progress_events (..., idempotency_key)
  VALUES (...)
  ON CONFLICT (idempotency_key) DO NOTHING
  RETURNING 1
)
UPDATE tasks
   SET attempt_count = attempt_count + 1,
       due_at        = $newDueAt,
       updated_at    = NOW()
 WHERE task_id = $1
   AND EXISTS (SELECT 1 FROM inserted);

#3.3 record_progress 的 nextDueAt 算法

代码内 lookup table(per progressType 固定间隔),不依赖 AI hint。

const PROGRESS_NEXT_DUE_INTERVAL_MINUTES: Record<TaskProgressType, number | null> = {
  no_answer: 60,                  // 1 小时后再试
  left_voicemail: 60 * 24,        // 1 天后
  text_sent: 60 * 24 * 2,         // 2 天后(等回复)
  callback_requested: null,       // 不动 dueAt,需要 caller 手动设(API 参数 nextDueAt)
  follow_up_scheduled: null,      // 同上
  customer_considering: 60 * 24 * 3, // 3 天后
};

function computeNextDueAt(
  progressType: TaskProgressType,
  currentDueAt: Date | null,
  callerOverride?: Date,         // staff API 手动设的 nextDueAt
): Date | null {
  if (callerOverride) return callerOverride;
  const interval = PROGRESS_NEXT_DUE_INTERVAL_MINUTES[progressType];
  if (interval === null) return currentDueAt;  // 不动
  return new Date(Date.now() + interval * 60_000);
}

#3.4 Contact Writer(Phase 1 最小版)

位置:callytics-common/src/domain/contact-writer.ts

Phase 1 只封装 identity / DNC / lastActivityAt 三个 SQL fragment(不统一 18 个 AI 画像字段 —— 只有 contacts-analyzer 写,无多 writer 冲突)。

// B5 fix —— contacts.franchise_id / account_id 是 NOT NULL,upsert 时新建 contact 必须提供
export function upsertIdentity(params: {
  phone: string;
  storeId: string;
  franchiseId: string;          // B5: contacts NOT NULL
  accountId: string;             // B5: contacts NOT NULL
  firstName?: string;
  lastName?: string;
  trustScore: NameTrustScore;
  activityAt: Date;
}): SQL;  // 返回 Drizzle SQL,使用现有 NAME_TRUST + CASE WHEN > COALESCE pattern。
          // ON CONFLICT (phone, store_id) DO UPDATE SET ... 时 franchise/account 不更新(只用于 INSERT 路径)

export function setDNC(params: {
  phone: string;
  storeId: string;
  updatedBy: 'staff' | 'ai' | 'system';
}): SQL;  // sticky:contacts.do_not_contact = true,不接受 false

export function touchActivity(params: {
  phone: string;
  storeId: string;
  activityAt: Date;
}): SQL;  // forward-only:GREATEST(last_activity_at, $activityAt)

caller 怎么拿到 franchiseId / accountId:

• contacts-analyzer:从 SQS message body 拿(已有)
• lead-processor:从 lead row 拿(已有)
• message-processor:从 PhoneStoreAssignments lookup 时一起拿(resolvePhoneIdentity() 已返回)
• ai-analysis-processor:同上
• studio-api(staff 操作):从 getAuthorizedStoreNeon() context 拿(已有)

#3.5 Timeline Writer

位置:callytics-common/src/domain/timeline-writer.ts

封装现有 buildTimelineValues() + buildContactTimelineInsertSQL(),加 Event Catalog Zod schema。

export const TIMELINE_EVENT_TYPES = [
  'task.created', 'task.status_changed', 'task.updated', 
  'task.progress_recorded',                                     // ← Phase 1 新增
  'task.note_updated',
  'contact.lifecycle_changed', 'contact.lead_status_changed', 'contact.dnc_changed',
  'contact.complaint_opened', 'contact.complaint_resolved',
  'contact_analysis.completed',
  'transcribe.completed', 'call_analysis.completed',
  'call.status_changed', 'message.created',
  'lead.created',
] as const;

export function writeTimelineEvent(params: {
  eventType: TimelineEventType;
  entity: { type: string; id: string };
  contactPhone: string;
  storeId: string;
  actor: ActorIdentity;
  payload: unknown;          // Zod schema 按 eventType discriminate
  idempotencyKey: string;
}): SQL | { sql: string; params: unknown[] };  // 两套(Drizzle / raw SQL)

task.closed / task.reopened 不新增 event type;沿用现有 task.status_changed,payload 用 close/reopen discriminator。call.completed / message.received 也不新增;source event 名称按 live SoT 保持 call_analysis.completed / message.created。

#3.6 Policy Guard

位置:callytics-common/src/domain/policy-guard.ts

不是独立 service,是 Task Orchestrator + Contact Writer 内部调的 validation function 集合。Phase 1 列出的检查:

S2 — computeAllowedTypeCategories(contact) 实现:

type ContactLifecycle = {
  lifecycleStage: 'lead' | 'member' | 'churned' | 'unknown';
  lifecycleState: 'active' | 'paused' | 'terminal';
  leadStatus?: LeadStatus;
  doNotContact: boolean;
};

export function computeAllowedTypeCategories(
  contact: ContactLifecycle,
): TaskTypeCategory[] {
  // DNC:不允许任何 outreach create
  if (contact.doNotContact) return [];
  
  // terminal:不允许创建 outreach task;churned re-engagement 必须先由 AI 判成 active
  if (contact.lifecycleState === 'terminal') {
    return [];
  }
  
  // lead active / paused
  if (contact.lifecycleStage === 'lead') {
    // AI 不能创 lead_outreach(那是 lead-tracking 入口专属)
    return ['lead_follow_up', 'booked_not_converted'];
  }
  
  // member active / paused
  if (contact.lifecycleStage === 'member') {
    return ['cancellation_risk', 'retention', 'upgrade', 'renewal', 'referral'];
  }
  
  // churned re-engage(active)
  if (contact.lifecycleStage === 'churned') {
    return ['win_back'];
  }
  
  // unknown:不允许
  return [];
}

测试覆盖(必须):

#§4. Caller Migration

按 unified-pipeline-final §Step 5 表 顺序。每个 caller 列具体 file 改动。

#4a. studio-api close(改 raw SQL helper + B2 transaction wrap)

File: studio-website-monorepo/apps/api/src/routes/tasks/close.ts

#4b. studio-api 新端点 POST /v2/tasks/:taskId/progress

File: studio-website-monorepo/apps/api/src/routes/tasks/progress.ts(新建)

import { buildTaskActionSQL } from '@retaintive/common/domain';

const ProgressRequestSchema = z.object({
  progressType: z.enum(['no_answer', 'left_voicemail', 'text_sent', 'callback_requested', 'follow_up_scheduled', 'customer_considering']),
  channel: z.enum(['phone', 'sms', 'voicemail', 'email']),
  callId: z.string().optional(),
  messageId: z.string().optional(),
  note: z.string().optional(),
  nextDueAt: z.string().datetime().optional(),  // staff 手动设;callback_requested / follow_up_scheduled 必填
});

app.post('/v2/tasks/:taskId/progress', async c => {
  const taskId = c.req.param('taskId');
  const body = ProgressRequestSchema.parse(await c.req.json());
  const { storeId, userId, staffName } = await getAuthorizedStoreNeon(c);
  
  const result = await buildTaskActionSQL({
    action: 'record_progress',
    payload: { taskId, ...body },
  }, {
    storeId,
    contactPhone: '', // Orchestrator 内部从 taskId 查
    actor: { type: 'staff', id: userId, name: staffName },
    db: getNeonClient(),
  });
  
  if (result.status === 'reject') {
    if (result.reason === 'task_not_open') throw new ConflictError('Task not open');
    if (result.reason === 'duplicate') return c.json({ ok: true, deduped: true });  // 幂等
    // ... 其他 reason
  }
  
  const sql = getNeonClient();
  // B2 — 必须 sql.transaction 包裹,不是 for + sql.query(后者非原子)
  await sql.transaction(result.statements.map(s => sql(s.sql, s.params)));
  return c.json({ ok: true });
});

#4c. studio-api reopen + postpone(改 raw SQL helper + B2 transaction wrap)

Files: routes/tasks/reopen.ts / routes/tasks/postpone.ts

同 4a pattern:

• reopen 走 action: 'reopen',postpone 走 action: 'update'(只改 dueAt)
• 调用 buildTaskActionSQL() 返回 statements,必须 sql.transaction(statements.map(s => sql(s.sql, s.params))) 包裹

#4d. contacts-analyzer caller migration

File: callytics-infrastructure/lambda/contacts-analyzer/src/infrastructure/neon-repository.ts

TaskDecision Zod schema 改动(src/core/models.ts):

• 6 action union:create_open / create_closed / close / update / record_progress / reopen
• 加 .transform() 兜底兼容 create → create_open(prompt engineer 如果用 create,自动 normalize)
• create_closed action payload 必带 sourceCallId(S5 evidence reference)
• S1 — 删 task 级 priority 输出(从 prompt JSON 删该字段);Orchestrator 在 applyTaskAction({ action: 'create_open' | 'create_closed' }) 内部计算 max(suggestedActions[].priority) 写入 tasks.priority
• 删 actionNeeded boolean 输出(派生)
• 删 lifecycleState?不删(撤销 — engineer feedback,见 contacts-schema §3.2)

#4e. lead-processor caller migration

File: callytics-infrastructure/lambda/lead-processor/src/core/persist-downstream.ts

#4f. message-processor STOP cascade

File: callytics-infrastructure/lambda/message-processor/src/core/message-processing.ts

#4g. studio-api contacts.actionNeeded 读路径迁移(Phase 1)

Files: routes/v3/contacts.ts:112 + routes/v3/leads.ts:209

#4h. studio-api 其他 routes status 字符串 sweep + dashboard metric 口径迁移(B3)

Files: routes/tasks/list.ts / routes/tasks/events.ts / routes/v3/dashboard-*.ts(dashboard 保留 raw SQL 不切 Drizzle)

status 字符串改名:

B3 — dashboard metric 口径迁移(Phase 1 移除 closeResult='attempted' 之后必须配套改):

口径细节(per codex §11):

-- Phase 1 后正确的 attempt workload query
SELECT 
  COUNT(*) AS attempts,
  COUNT(DISTINCT task_id) AS tasks_touched,
  COUNT(DISTINCT actor_id) FILTER (WHERE actor_type = 'staff') AS active_staff
FROM task_progress_events
WHERE store_id = $1
  AND occurred_at >= NOW() - INTERVAL '7 days';

dashboard 不要求迁移到 ORM(见 issue #449),但口径必须改。

#4j. Source event 写入路径(S5 — 不经 Policy Guard 的 timeline event)

unified §1 架构图明确画了 source event 直连 Timeline Writer,不过 Policy Guard。Phase 1 这 3 个 source event 的写入路径:

实施:Phase 1 这 3 个 caller 不必改 source event 写入逻辑(现状已在写)。Phase 1 改动:

• call_analysis.completed / message.created / lead.created 跟 Phase 1 mutation event(task.created / task.progress_recorded / 等)用同一份 TIMELINE_EVENT_TYPES Zod schema enforce
• caller 仍直接调 TimelineWriter.writeTimelineEvent()(不经 Orchestrator / Policy Guard)
• actor type 保持现状(call_analysis / system / lead_webhook 等),source event 来源 = pipeline 自动,不是 staff / AI agent decision

#4i. studio-api callytics-infra task_type 字段消费端清理

Files:

• routes/tasks/types.ts:16 task_type: string 字段定义 → 删
• routes/tasks/list.ts:164 t.task_type SELECT → 删
• routes/tasks/close.ts:212 INSERT ... task_type ... → 删
• lambda/lead-processor/src/core/persist-downstream.ts:193 caller-side taskType: 'lead_outreach' → 删;过渡期 legacy shim 放 Orchestrator 内部
• lambda/contacts-analyzer/src/core/prompt-builder.ts:26 derive 注释 → 删
• lambda/contacts-analyzer/src/infrastructure/neon-repository.ts:692-772 caller-side derive + INSERT 6 处 → 删;过渡期 legacy shim 放 Orchestrator 内部

#§5. End-State Test Plan(normative)

本节是 Phase 1 的真正验收核心。PR 数量和拆分方式可以变,但以下 tests / invariants 必须在最终 cutover 前通过。

结构:§5.0 列 invariants(must hold 的事实);§5.1-§5.5 是 implements §5.0 的具体 test 方法,按 unit / integration / idempotency / replay / migration 分层。reviewer 优先看 §5.0,implementer 按 §5.1-§5.5 落地。

#5.0 End-state invariants(must hold after cutover)

#5.1 Unit tests(per module)

#5.2 Integration tests(per caller migration)

每个 caller 独立 test 文件,跑真实 Neon(integration-test:neon script):

#5.3 Idempotency tests

#5.4 Replay tests

跑历史 prompt output JSON(从 CloudWatch 抽 sample),verify Phase 1 module 行为不破坏既有数据。

#5.5 Migration tests

#§6. Execution Strategy(PR count is not the design)

PR 数量不是 Phase 1 的核心设计。可以按 repo 拆成少量 PR,也可以一个 PR 里多做事;真正必须守住的是 end-state contract 和 test gates。实现者不要为了 8 个 PR 的形状牺牲一致性。

#6.1 Minimum execution stages

如果能在测试环境一次性停写、apply final migration、部署所有 caller,中间 compatibility window 可以很短。但只要有旧 Lambda/studio-api 和新 schema/code 共存的窗口,必须按 compatibility window 处理。

#6.2 Transitional schema requirements(if old/new coexist)

-- open-state transitional CHECK:旧 pending + 新 open 都是 open work
ALTER TABLE tasks DROP CONSTRAINT IF EXISTS chk_tasks_closed_integrity;
ALTER TABLE tasks ADD CONSTRAINT chk_tasks_closed_integrity CHECK (
  (status IN ('pending', 'open') AND close_type IS NULL AND closed_at IS NULL)
  OR
  (status = 'closed' AND close_type IS NOT NULL AND closed_at IS NOT NULL)
);

-- duplicate guard covers both old and new open-state names
DROP INDEX IF EXISTS uq_tasks_pending_contact_category;
CREATE UNIQUE INDEX uq_tasks_pending_contact_category
  ON tasks (contact_phone, store_id, type_category)
  WHERE status IN ('pending', 'open') AND store_id IS NOT NULL;

过渡期间所有读路径必须用 status IN ('pending','open'),覆盖:

• closeAllOpenForContact() / DNC cascade
• close / update / record_progress 的 state 校验
• contacts.actionNeeded 的 EXISTS 子查询
• dashboard open workload 统计
• duplicate-open-task 去重判断

task_type 列在 DROP 之前不能写 NULL(列仍 NOT NULL)。过渡期 Orchestrator 内部写 legacy shim:

legacyTaskType = typeCategory === 'lead_outreach' ? 'lead_outreach' : 'follow_up';

Final cutover 后 DROP COLUMN。

#6.3 Cutover preflight 检查项

进 destructive cutover 之前,逐项 verify:

• 所有 Phase 1 caller 代码已部署到 test env
• contacts-analyzer / lead-processor / message-processor STOP / studio-api close/progress/reopen 的 smoke test 全 pass
• repo grep 无任何 writer 写 status='pending'
• CloudWatch 近期 logs 无 old-code path 写 task 出错
• dashboard 新旧 attempt workload query 对比无未解释 drift

通过后跑 final migration:

• UPDATE tasks SET status='open' WHERE status='pending'
• DELETE FROM tasks WHERE store_id IS NULL;ALTER TABLE tasks ALTER COLUMN store_id SET NOT NULL
• DROP tasks.action_needed / tasks.task_type
• 收缩 TASK_STATUS=['open','closed']
• 收缩 TASK_CLOSE_RESULT 至 15 values
• final CHECK / unique index 用 status='open'

#6.4 可延后的 cleanup

不阻塞 Phase 1 ship,Phase 1 后任何时候做:

• 删 buildContactTimelineInsertSQL(等 studio-api Drizzle 迁移 issue #449 落地后)
• Phase 2:停 contact_analyzer 写 contacts.actionNeeded,然后 DROP COLUMN + 删 idx_contacts_action_needed

#§7. Acceptance Criteria

#7.1 功能性

• task_progress_events 表创建并写入,任意 progressType 进 6 enum
• Phase 1 ship 后,studio-api close API 不接受 closeResult ∈ {no_answer, left_voicemail, callback_later, attempted},返回 400
• Phase 1 ship 后,POST /v2/tasks/:taskId/progress 工作正常,phone / sms / manual 三种 source 都能写
• tasks.status 全 schema + code + UI 用 'open' / 'closed',grep 'pending' 0 hit
• tasks.store_id 在 schema 是 NOT NULL,DB 验证无 NULL 行
• tasks.action_needed 字段不存在(DROP COLUMN);contacts.action_needed 字段保留但 studio-api routes/v3/contacts.ts + routes/v3/leads.ts 已改用 EXISTS
• tasks.task_type 字段不存在

#7.2 并发 / 幂等

• 跑 race condition test 套件 100 次,所有 invariant 不破(同 contact 同 typeCategory 最多 1 open;同 callId 同 progressType 最多 1 progress event;AI close stale propose 必 reject)
• 验证 PostgreSQL NULL ≠ NULL 不影响 record_progress 去重(idempotency_key NOT NULL UNIQUE 覆盖)

#7.3 Tenant isolation

• 验证 applyTaskAction() storeId 缺失时 reject store_mismatch,不让漏过
• 验证 taskId 跨 store 的提案 reject hallucinated_task_id
• Audit log:所有 mutation 在 contact_timeline 有 row(actor_type + store_id 完整)

#7.4 Backward compat(测试环境最低要求)

• studio-api 端点对外 request shape 不变(close / reopen / postpone 仍接受相同 request body,只在 closeResult ∈ progress_set 时返 400 + 提示用 progress endpoint)
• contacts-analyzer SQS message format 不变(handler 内部 normalize 旧 action 名为新 action 名)

#7.5 Phase 2 readiness

• CloudWatch metric mutation_rejected_total{reason} 上线,Phase 2 可根据数据决定 needs_review queue 设计
• contacts.actionNeeded Phase 1 后 DROP COLUMN(任何时候开 PR 都可,Phase 1 cutover 完成即解锁)

#§8. Phase 1 SoP(saas-tech-fundamentals 落地)

参考 local reference /Users/maxwsy/workspace/claude-plugins/plugins/core-tools/shared-references/saas-tech-fundamentals.md,Phase 1 落地以下 mental model:

#8.1 Tenant Isolation(OWASP API1: BOLA)

• 每个 DB query 都带 WHERE store_id = $X —— Policy Guard 在 application 层强制,SQL 层通过 tasks.store_id NOT NULL enforce
• taskId 引用必须 verify 跨 store —— AI hallucination guard 在 Policy Guard 内 SELECT verify
• Phase 1 不引入 Postgres RLS —— 2-3 人团队,application-layer Policy Guard 够。Phase 2 评估 RLS(评估项已 list 在 Phase 2 backlog)

#8.2 LLM01: Prompt Injection

Contact Analyzer prompt 读 transcripts / SMS body,这些是 user-controlled content。

Phase 1 mitigations:

• Output validation:Zod schema 强制 TaskDecision 输出 6 action 内、taskId 必须真实存在 + 属于 contact、typeCategory 必须在 allowed set 内 —— prompt injection 即使绕过 prompt 防御,输出阶段也被 Policy Guard reject
• Tenant 隔离覆盖 prompt injection 后果:即使 attacker 让 AI 输出错误 taskDecisions,storeId check 让 mutation 不能跨 store
• Phase 1 不做 input sanitization(transcripts / SMS body 不过滤,因为业务需要原始内容做 AI 分析)
• CloudWatch metric:mutation_rejected_total{reason='hallucinated_task_id'} 监控 prompt injection 可能信号

#8.3 LLM02: Sensitive Information Disclosure

LLM provider 是 DeepSeek V4 via OpenRouter(@retaintive/common ai-client.ts:8)。

Phase 1 action items:

• Verify OpenRouter data retention policy —— 如果 retention 长,加 transforms: ['middle-out'] 或考虑 --data-no-train flag
• PII 在 prompt context 里有(contact 全名 / 电话 / 通话内容)—— 不能避免(业务需要),但要 acknowledge 风险 写进 SoP

#8.4 LLM06: Excessive Agency

Phase 1 设计原则:

• AI 只输出 proposal,代码执行 mutation —— Policy Guard applyTaskAction() 是唯一 mutation 入口
• 低置信度 proposal 直接 reject —— Phase 1 module contract 支持 low_confidence;contacts-analyzer AI path 因 prompt 未输出 per-decision confidence 暂不触发
• AI 提案 confidence threshold env var MIN_AI_CONFIDENCE 默认 0.7,ops 可调;Phase 2 再把 confidence 字段接入 prompt / TaskDecision adapter

#8.5 LLM10: Unbounded Consumption

Contact Analyzer 触发频率优化路线见 unified-pipeline-final §3 触发机制与优化路线。

Phase 1 不做触发层优化(per_call cooldown / cron 增量),但:

• CloudWatch metric contacts_analyzer_invocations_per_store_per_day 上线 —— 监控成本异常
• OpenRouter billing dashboard 配 alarm(monthly cost > threshold)

#8.6 Multi-tenant Audit checklist 自检

#Appendix A — Open Items / TBD

#Appendix B — 不在 Phase 1 scope

• 拆 prompt(并行外包)
• tool calling runtime
• SMS meaningful 分类
• voice agent
• prod legacy 数据迁移
• 触发层优化(per_call cooldown / cron 增量)
• dashboard query 整体重做