> For AI agents: the complete documentation index is available at /llms.txt, the full documentation bundle is available at /llms-full.txt.

# Unified Pipeline Phase 1 — Implementation Plan

> **Type**: Implementation plan(基于 [unified-pipeline-final.md](/product-design/v2/unified-pipeline/unified-pipeline-final.md) 高层设计 + [PR #368](https://github.com/retaintive/docs/pull/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](https://github.com/retaintive/studio-website-monorepo/issues/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` 表

```sql
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**:

```sql
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;
```

**新增字段**:

```sql
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**:

```sql
-- 测试环境直接 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):

```sql
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 文件更新

| 文件                                      | 改动                                                                                                                                                                                                                                                         |
| --------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `src/db/schema/tasks.ts`                | **Phase 1 final**:`TASK_STATUS = ['open', 'closed']`;`TASK_CLOSE_TYPE` 加 `create_closed`;`TASK_CLOSE_RESULT` 15 values;加 `executorType` / `attemptCount` / `sourceCallId` column;删 `taskType` / `actionNeeded` column;CHECK constraint 同步;index 同步。过渡态见 §6 |
| `src/db/schema/task-progress-events.ts` | **新建文件** —— `taskProgressEvents` pgTable + `TASK_PROGRESS_TYPE` / `TASK_CHANNEL` / `TASK_ACTOR_TYPE` const                                                                                                                                                 |
| `src/db/schema/task-ui.ts`              | `CLOSE_RESULT_OPTIONS` 15 values;`STATUS_OPTIONS` value `open`(UI label 仍 `Open`);新增 `PROGRESS_OPTIONS` const                                                                                                                                              |
| `src/db/schema/contacts.ts`             | 无 column 改动(`lifecycleState` 撤销退役保留;`actionNeeded` 留 Phase 2 删)                                                                                                                                                                                            |
| `src/db/schema/index.ts`                | export `taskProgressEvents`                                                                                                                                                                                                                                |
| `drizzle/` migration files              | 用 `drizzle-kit generate` 生成 migration SQL,人工 review 后 apply 到测试 Neon                                                                                                                                                                                       |

***

## §3. Module Contracts

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

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

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

```typescript
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 必传规则统一):

```typescript
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**:

```typescript
// 单 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 表](/product-design/v2/unified-pipeline/unified-pipeline-final.md))。

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

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

**Signature**:

```typescript
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**(原子事务):

```typescript
// 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 共享):

| Progress 来源               | idempotency\_key 格式                                                            |
| ------------------------- | ------------------------------------------------------------------------------ |
| 电话(callId 非空)             | `progress:{taskId}:call:{callId}:{progressType}`                               |
| SMS(messageId 非空)         | `progress:{taskId}:message:{messageId}:{progressType}`                         |
| 手动(staff 触发)              | `progress:{taskId}:manual:{actorId}:{occurredAt.toISOString()}:{progressType}` |
| 系统(cron / reconciliation) | `progress:{taskId}:system:{runId}:{progressType}`                              |

**`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**:

```sql
-- ❌ 错(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。

```typescript
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 冲突)。

```typescript
// 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。

```typescript
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 列出的检查:

| Check                               | 实现                                                                                                                                                                                                                                                                                                       |                                                                                                                                                                          |
| ----------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| storeId 非空                          | runtime check + Zod                                                                                                                                                                                                                                                                                      |                                                                                                                                                                          |
| DNC hard stop                       | `SELECT do_not_contact FROM contacts WHERE phone=? AND store_id=?` —— **true 时**(B6):reject `create_open` / `create_closed` / `update` / `record_progress` / `reopen`;**只允许** `close` / `closeAllOpenForContact`                                                                                         |                                                                                                                                                                          |
| AI hallucination guard              | `SELECT task_id FROM tasks WHERE task_id=? AND contact_phone=? AND store_id=?` —— 0 rows = reject `hallucinated_task_id`                                                                                                                                                                                 |                                                                                                                                                                          |
| Human authority guard(S8)           | **纯 SQL 层** conditional WHERE `updated_at <= $aiRunStartedAt`;UPDATE RETURNING 0 rows → module 层 map 成 `stale_proposal` reason。**module 层不提前 SELECT 比较**(避免 TOCTOU race)                                                                                                                                 |                                                                                                                                                                          |
| State transition                    | runtime check(close / update / record\_progress 对 open;reopen 对 closed)                                                                                                                                                                                                                                  |                                                                                                                                                                          |
| `create_closed` 已有 open task 守卫(S3) | `SELECT 1 FROM tasks WHERE contact_phone=? AND store_id=? AND type_category=? AND status='open' LIMIT 1` —— 找到则 reject `create_closed_with_open_task`(caller 应改用 `close` 现有 task)                                                                                                                        |                                                                                                                                                                          |
| typeCategory allowed set(S2)        | 仅用于 AI proposal(\`actor.type='contact\_analysis'                                                                                                                                                                                                                                                         | 'ai\_agent'`);`computeAllowedTypeCategories(contact)`函数见下方表。deterministic caller(如 lead-processor`lead\_outreach\`)不走这个 AI allowed-set,但仍走 store / DNC / duplicate guard |
| `closeNote` 必填(closeResult='other') | runtime check —— 缺则 reject `close_note_required`(codex §3)                                                                                                                                                                                                                                               |                                                                                                                                                                          |
| Confidence threshold                | env var `MIN_AI_CONFIDENCE`(default 0.7),当 action payload 显式携带 `confidence` 时低于阈值直接 reject `low_confidence` + CloudWatch metric `mutation_rejected_total{reason='low_confidence'}`。contacts-analyzer 当前 prompt 不输出 per-decision confidence,所以 Phase 1 只做 module-level test;AI path activation 留 Phase 2。 |                                                                                                                                                                          |

**S2 — `computeAllowedTypeCategories(contact)` 实现**:

```typescript
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 [];
}
```

**测试覆盖**(必须):

| Contact 状态                                   | 允许 typeCategory                                           | 测试场景                                                                     |
| -------------------------------------------- | --------------------------------------------------------- | ------------------------------------------------------------------------ |
| `doNotContact=true`                          | \[]                                                       | DNC 任何 create 都 reject                                                   |
| `lifecycleStage='lead', state='active'`      | lead\_follow\_up, booked\_not\_converted                  | AI 选 `lead_outreach` reject `invalid_type_category`(那是 lead-tracking 专属) |
| `lifecycleStage='member', state='active'`    | cancellation\_risk, retention, upgrade, renewal, referral | 普通 member                                                                |
| `lifecycleStage='churned', state='active'`   | win\_back                                                 | re-engage 客户                                                             |
| `lifecycleStage='churned', state='terminal'` | \[]                                                       | terminal 不允许 create;churned re-engage 必须先被判成 active                      |
| `lifecycleState='terminal'`(non-churned)     | \[]                                                       | terminal lead 不允许 outreach                                               |

***

## §4. Caller Migration

按 [unified-pipeline-final §Step 5 表](/product-design/v2/unified-pipeline/unified-pipeline-final.md) 顺序。每个 caller 列具体 file 改动。

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

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

| 改动                                                     | Before                                                                                             | After                                                                                                                                                                                        |
| ------------------------------------------------------ | -------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| close 流程                                               | 直接 `sql.query('UPDATE tasks ... WHERE status=\'pending\'')` + 条件 `INSERT INTO tasks ... follow-up` | `const result = await buildTaskActionSQL({ action: 'close', payload: {...} }, ctx); await sql.transaction(result.statements.map(s => sql(s.sql, s.params)));` —— **必须 transaction wrap**(B2) |
| no\_answer / left\_voicemail                           | 走 close + 自动建 follow-up                                                                            | 改走新 `POST /v2/tasks/:taskId/progress` endpoint(见 4b);**close API 拒绝 closeResult ∈ progress\_set** + 返回 400                                                                                   |
| status 字符串                                             | `'pending'`                                                                                        | `'open'`                                                                                                                                                                                     |
| timeline                                               | 自己拼 `buildContactTimelineInsertSQL`                                                                | Orchestrator 内部自动 build,跟 UPDATE 同 transaction 提交                                                                                                                                            |
| FOLLOW\_UP\_TASK\_TYPE 常量(line 23-27)                  | hardcode                                                                                           | 删除(task\_type 字段退役)                                                                                                                                                                          |
| INTERVAL hardcode `'${daysToFollowUp} days'`(line 213) | hardcode                                                                                           | 删除(no\_answer / left\_voicemail 不再走 close)                                                                                                                                                   |

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

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

```typescript
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`

| 改动                                  | Before                                                         | After                                                                                                                                                                  |
| ----------------------------------- | -------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| TaskDecision 遍历                     | for each decision: `db.insert(tasks)` / `db.update(tasks)` 自己拼 | `for (const d of taskDecisions) { const r = await applyTaskAction(d, ctx); if (r.status === 'allow') statements.push(...r.statements); }` 最后 `db.batch(statements)` 原子 |
| status                              | `'pending'`                                                    | `'open'`                                                                                                                                                               |
| derivedTaskType(line 692-772)       | caller 自己 derive `task_type` from `type_category`              | caller 删除该逻辑;过渡期由 Orchestrator 写 legacy `task_type = typeCategory === 'lead_outreach' ? 'lead_outreach' : 'follow_up'`,final cutover 后 DROP COLUMN                     |
| trust score CASE WHEN(line 396-398) | 自己拼 SQL                                                        | 调 `ContactWriter.upsertIdentity()`                                                                                                                                     |
| timeline                            | 自己拼 `buildTimelineValues`                                      | Orchestrator + Timeline Writer 内部                                                                                                                                      |

**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`

| 改动                                    | Before                                                                                     | After                                                                                                                                                                                                                                           |
| ------------------------------------- | ------------------------------------------------------------------------------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Task INSERT(line 185-206)             | `client.insert(tasks).values({ taskType, actionNeeded: true, ... }).onConflictDoNothing()` | `const r = await applyTaskAction({ action: 'create_open', payload: { contactPhone, storeId, typeCategory: 'lead_outreach', priority: 'high', suggestedActions, ... } }, ctx);` + `db.batch([...contactStmt, ...r.statements, ...timelineStmt])` |
| `actionNeeded: true`(line 199)        | hardcode                                                                                   | 删除(字段退役)                                                                                                                                                                                                                                        |
| `taskType: 'lead_outreach'`(line 193) | hardcode                                                                                   | caller 删除 hardcode;过渡期由 Orchestrator 写 legacy shim,final cutover 后 DROP COLUMN                                                                                                                                                                  |
| status                                | `'pending'`                                                                                | `'open'`                                                                                                                                                                                                                                        |

### 4f. message-processor STOP cascade

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

| 改动                                | Before                                 | After                                                                                                                                                                                                                                  |
| --------------------------------- | -------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| DNC cascade SQL                   | 自己拼 + 调 `dnc-cascade.ts` shared helper | 调 `ContactWriter.setDNC()` + `TaskOrchestrator.closeAllOpenForContact({ contactPhone, storeId, closeResult: 'do_not_contact', closeNote: 'Auto-closed by DNC hard stop', actor: {type:'system'} })` —— **B7** Orchestrator bulk helper |
| status filter                     | `status='pending'`                     | `status='open'`                                                                                                                                                                                                                        |
| 现有 `dnc-cascade.ts` shared helper | 仍存在                                    | Phase 1 由 `closeAllOpenForContact()` 内部调用,caller 不直接 import dnc-cascade(Phase 2 删 helper)                                                                                                                                              |

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

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

| 改动                                 | Before                                                                     | After                                                                                                                                                                     |
| ---------------------------------- | -------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| contacts list query                | `SELECT c.action_needed, c.action_needed_reason ...`                       | 加 `LATERAL JOIN (SELECT EXISTS(SELECT 1 FROM tasks WHERE contact_phone=c.phone AND store_id=c.store_id AND status='open') AS has_open_task)` 替代,但**字段保留**(Phase 2 才 DROP) |
| leads KPI                          | `COUNT(*) FILTER (WHERE c.action_needed = true)::integer AS action_needed` | `COUNT(DISTINCT c.phone) FILTER (WHERE EXISTS(SELECT 1 FROM tasks t WHERE t.contact_phone=c.phone AND t.store_id=c.store_id AND t.status='open'))`                        |
| contact\_analyzer 写 `actionNeeded` | 仍写(保险,observe-only)                                                        | 不动,Phase 2 PR 一起删                                                                                                                                                         |

### 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 字符串改名**:

| 改动                  | Before                                      | After                                    |
| ------------------- | ------------------------------------------- | ---------------------------------------- |
| status filter       | `WHERE status = 'pending'`                  | `WHERE status = 'open'`                  |
| status range filter | `status = 'pending' AND due_at BETWEEN ...` | `status = 'open' AND due_at BETWEEN ...` |

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

| Metric                  | Before(Phase 1 之前)                                                                                                                          | After(Phase 1)                                                                                                                                  |
| ----------------------- | ------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------- |
| Attempt workload        | `COUNT(*) WHERE close_result = 'attempted'`(在 tasks 表)                                                                                      | `COUNT(*) FROM task_progress_events WHERE store_id = $X AND occurred_at BETWEEN ...` — 算 progress events 数,按 actor\_type / progress\_type group |
| Completed objectives    | `COUNT(*) WHERE status = 'closed'`                                                                                                          | 不变(但 status 字符串 `'pending'→'open'` 影响 open task 反查路径)                                                                                           |
| Business wins           | `COUNT(*) WHERE close_result IN ('converted','booked','cancel_saved','issue_resolved','win_back','renewed','upgraded','referral_obtained')` | 不变(close\_result 15 values 这 8 个都在 final set 内)                                                                                                 |
| `attempted` 历史 backfill | `attempted` 是 row 值                                                                                                                         | `tasks.close_result = 'attempted'` 老行 final cutover 之前仍可能存在;dashboard query 加 fallback 兼容(`COALESCE` 算两边相加)或等 final cutover 后 backfill 清掉       |
| Open workload           | `WHERE status = 'pending'`                                                                                                                  | `WHERE status = 'open'`                                                                                                                         |

**口径细节**(per codex §11):

```sql
-- 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](https://github.com/retaintive/studio-website-monorepo/issues/449)),但**口径必须改**。

***

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

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

| Event                     | 谁写                    | 何时                                                     | Payload                                                 | idempotencyKey                     |
| ------------------------- | --------------------- | ------------------------------------------------------ | ------------------------------------------------------- | ---------------------------------- |
| `call_analysis.completed` | ai-analysis-processor | call 分析完成后,跟 `calls` 表写入同事务                            | `{ callId, primaryOutcomeResult, followUpNeeded, ... }` | `call_analysis.completed:{callId}` |
| `message.created`         | message-processor     | message-processor 处理 inbound SMS 时,跟 `messages` 表写入同事务 | `{ messageId, direction, messageType, ... }`            | `message.created:{messageId}`      |
| `lead.created`            | lead-processor        | lead-processor `persistLeadDownstream` 已经在写(现有逻辑保留)    | `{ leadId, leadType, source, ... }`                     | `lead.created:{leadId}`            |

**实施**: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)

| Invariant               | 怎么 verify                                                                                                                                 | 由哪节实测                                                                        |
| ----------------------- | ----------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------- |
| tasks 只有二态              | `SELECT status, COUNT(*) FROM tasks GROUP BY status` 只返回 `open` / `closed`                                                                | §5.5 Migration test                                                          |
| open task 唯一            | 同 `(contact_phone, store_id, type_category)` 最多 1 个 `status='open'` task                                                                  | §5.3 Idempotency test 第 1 条                                                  |
| progress 不 close task   | `record_progress(no_answer/left_voicemail/text_sent)` 后 task 仍 `open`,只新增 `task_progress_events` row                                      | §5.2 Integration `studio-api progress`                                       |
| closeResult 纯 outcome   | close API 拒绝 `no_answer` / `left_voicemail` / `callback_later` / `attempted`;这些只能走 progress endpoint                                      | §5.1 Unit `applyTaskAction()` reject path + §5.2 `studio-api close`          |
| DNC hard stop           | DNC contact reject `create_open` / `create_closed` / `update` / `record_progress` / `reopen`;允许 `close` / `closeAllOpenForContact`        | §5.1 Unit Policy Guard DNC check                                             |
| tenant isolation        | 任意 `taskId` mutation 必须 verify `storeId`;跨 store taskId reject `hallucinated_task_id` / `store_mismatch`                                  | §5.1 Unit Policy Guard + §5.3 Idempotency cross-store test                   |
| deterministic lead path | lead-processor 可通过 Orchestrator 创建 `lead_outreach`;AI proposal 不能创建 `lead_outreach`                                                       | §5.2 Integration `lead-processor` + §5.1 Unit `computeAllowedTypeCategories` |
| timeline audit          | 每个 task mutation 同事务写 `contact_timeline`;source events 使用 live event names `call_analysis.completed` / `message.created` / `lead.created` | §5.1 Unit `TimelineWriter` + §5.2 Integration 各 caller                       |
| legacy columns gone     | final schema 无 `tasks.action_needed` / `tasks.task_type`;`contacts.action_needed` 字段仍存在但读路径不再依赖                                           | §5.5 Migration test                                                          |

### 5.1 Unit tests(per module)

| Module                           | Test                                                                                                                                 |
| -------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------ |
| `applyTaskAction()`              | 6 个 action × allow / reject 路径 = 12+ test;每个 reject reason 单独 test;Policy Guard 5 个 check 单独 test                                    |
| `buildTaskActionSQL()`           | 同上,verify 输出 SQL string + params 跟 Drizzle 版语义一致                                                                                     |
| `computeNextDueAt()`             | 6 个 progressType × callerOverride 有/无 = 12 test                                                                                      |
| `ContactWriter.upsertIdentity()` | NAME\_TRUST 6 levels × old higher / new higher / same score = 18 test                                                                |
| `ContactWriter.setDNC()`         | sticky(true → false 不允许)+ 三种 updatedBy                                                                                               |
| `TimelineWriter`                 | 所有 `TIMELINE_EVENT_TYPES` × Zod schema validation;verify 不新增 `task.closed` / `task.reopened` / `call.completed` / `message.received` |

### 5.2 Integration tests(per caller migration)

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

| Caller                          | Test scenarios                                                                                                                                                            |
| ------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| studio-api close                | close open task → close + timeline written;close already-closed → ConflictError;close stale AI proposal → reject `stale_proposal`                                         |
| studio-api progress             | record\_progress with callId → progress event + tasks.attempt\_count + 1;same callId twice → second 幂等返回 deduped;record\_progress on closed task → reject `task_not_open` |
| studio-api reopen               | reopen closed → open;reopen open → reject `task_not_closed`                                                                                                               |
| contacts-analyzer taskDecisions | AI taskDecisions\[] with create\_open / close / record\_progress 混合 → 全成功 atomic;AI hallucinate taskId → reject `hallucinated_task_id` skip 不影响其他 decision                |
| lead-processor                  | New lead → create\_open task + contact UPSERT + timeline 原子;duplicate lead → onConflictDoNothing 行为不变                                                                     |
| message-processor STOP          | inbound STOP SMS → DNC set + close all open tasks + timeline `task.status_changed` rows                                                                                   |

### 5.3 Idempotency tests

| Scenario                                    | Expectation                                                           |
| ------------------------------------------- | --------------------------------------------------------------------- |
| 同 contact 同 typeCategory 两次 create\_open    | 第二次 reject `duplicate`,只一行                                            |
| 同 callId 同 progressType 两次 record\_progress | 第二次 ON CONFLICT idempotency\_key DO NOTHING,attempt\_count 只 + 1      |
| 同 source\_call\_id 两次 create\_closed        | 第二次 partial unique 冲突,reject `duplicate`                              |
| AI close + staff close 同 task 并发            | conditional WHERE `status='open'` 保证只 1 成功,另一个 reject `task_not_open` |
| AI propose 过期(updated\_at > aiRunStartedAt) | reject `stale_proposal`,task 状态不变                                     |

### 5.4 Replay tests

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

### 5.5 Migration tests

| Test                                | Method                                                                                                   |
| ----------------------------------- | -------------------------------------------------------------------------------------------------------- |
| status `pending → open` migration   | 测试环境跑 migration SQL → verify 所有 tasks rows status ∈ \{'open', 'closed'}                                  |
| `tasks.store_id` SET NOT NULL       | 先 SELECT NULL 行 count → DELETE → ALTER → verify SET NOT NULL 成功                                          |
| `tasks.action_needed` DROP COLUMN   | drizzle-kit generate migration → verify no orphan index / CHECK constraint                               |
| `task_progress_events` CREATE TABLE | drizzle-kit generate → apply → verify schema 跟 callytics-common/src/db/schema/task-progress-events.ts 一致 |

***

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

| Stage                           | 目的                   | 完成判据                                                                            |
| ------------------------------- | -------------------- | ------------------------------------------------------------------------------- |
| Build final code + tests        | 把 final contract 写出来 | §2 schema delta、§3 module contracts、§4 caller migration 全部落地;§5 test suite pass |
| Compatibility window(if needed) | 旧代码和新代码可能同时跑时不破      | §6.2 transitional schema 已 apply;读路径含旧值;legacy shim 已就位                         |
| Contract cutover                | 收缩到 final schema     | §6.3 preflight 全过;final migration SQL 跑完;§5.0 invariant tests 全 pass            |

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

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

```sql
-- 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:

```typescript
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](https://github.com/retaintive/studio-website-monorepo/issues/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 触发机制与优化路线](/product-design/v2/unified-pipeline/unified-pipeline-final.md)。

**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 自检

| Check                                                           | Phase 1 状态                                                                         |
| --------------------------------------------------------------- | ---------------------------------------------------------------------------------- |
| 每个 DB query 包含 tenant ID in WHERE clause                        | ✅ `applyTaskAction()` ctx 强制 storeId                                               |
| Tenant ID 来自 authenticated session,不是 request body              | ✅ studio-api `getAuthorizedStoreNeon()` 已有                                         |
| 跨 tenant 数据访问不可能即使有 valid auth token                            | ✅ Policy Guard `store_mismatch` reject + DB partial unique 含 storeId               |
| Background jobs / cron / batch 也 enforce tenant scope           | ✅ contacts-analyzer SQS message 含 storeId,handler 用 `(phone, storeId)` 作 PK lookup |
| Admin / super-user 显式 scoped + audited                          | ⚠️ Phase 1 无 admin endpoint,Phase 2+ 评估                                            |
| Database-level enforcement(RLS, separate schemas, separate DBs) | ⚠️ Phase 1 不引入 RLS,application-layer Policy Guard;Phase 2 评估                       |

***

## Appendix A — Open Items / TBD

| Item                                                                      | 处理                                                                                              |
| ------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------- |
| prompt engineer 用 `create` 还是 `create_open`                               | TaskDecision Zod `.transform()` 兜底兼容;ping engineer 1 分钟 confirm 后删 transform                    |
| `contacts.actionNeededReason` / `suggestedActions` 归属 contacts vs task 实体 | Phase 2 decide(observe period 后)                                                                |
| needs\_review queue / table / UI / SLA / ownership                        | Phase 2 decide(看 CloudWatch metric 数据后)                                                         |
| Postgres RLS 评估                                                           | Phase 2+ tech debt                                                                              |
| studio-api 切 Drizzle                                                      | [issue #449](https://github.com/retaintive/studio-website-monorepo/issues/449),Phase 1 完成后独立 PR |

## Appendix B — 不在 Phase 1 scope

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