> For AI agents: the complete documentation index is available at /llms.txt, the full documentation bundle is available at /llms-full.txt. # Plan 04 — lead-processor Caller Migration Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use `superpowers:subagent-driven-development` (recommended) or `superpowers:executing-plans` to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** 把 `callytics-infrastructure/lambda/lead-processor` 从自己拼 `db.insert(tasks)` / `db.insert(contacts)` / `db.insert(contactTimeline)` 三条 SQL,改成调 Plan 02 共享 module(`applyTaskAction({action:'create_open'})` + `upsertIdentity()` + `writeTimelineEvent()`)。**唯一改的文件是 `src/core/persist-downstream.ts`** —— `handler.ts` 不动(它只 parse SQS envelope + 调 core)。 **Architecture:** lead-processor 单一 caller path(`persistLeadDownstream`),3 件事原子写 Neon:contact UPSERT + task INSERT + timeline INSERT,目前 `client.batch([...])` 包裹。Plan 04 改完后 3 个 SQL 改由 module helpers 返回,仍走同一 `client.batch()` 原子提交 —— 行为 1:1 等价,但 hardcoded `task_type` / `actionNeeded: true` / `status: 'pending'` 走掉,DNC + duplicate guard 走 Policy Guard。 **Tech Stack:** TypeScript 5 / Drizzle ORM(workspace catalog) / Neon HTTP client / Sentry / Lambda Node.js 20 / bun test **Spec source:** `docs/product-design/v2/unified-pipeline/implementation-plan/normative-spec.md` §4e(line 590-599)+ §3.4 B5 contact NOT NULL + §5.0 invariants(`tasks 只有二态` / `deterministic lead path` / `timeline audit` / `tenant isolation`) **Dependency:** Plan 01(schema)+ Plan 02(`@retaintive/common@1.2.0` 提供 `domain/*` exports)merged 并发布 *** ## File Structure | File | Responsibility | Action | | ------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------- | ------------------- | | `package.json`(repo root) | catalog `@retaintive/common` 从 `*` 不变,但 root pin `^1.0.0` → `^1.2.0` | **Modify** | | `lambda/lead-processor/package.json` | catalog 不变(继承 root) | **No change** | | `lambda/lead-processor/src/core/persist-downstream.ts` | task INSERT → `applyTaskAction({action:'create_open'})`;contact UPSERT → `upsertIdentity()`;timeline → `writeTimelineEvent()` | **Modify**(主战场) | | `lambda/lead-processor/src/handler.ts` | 不动(SQS → Drizzle client wiring 跟改造无关) | **No change** | | `lambda/lead-processor/tests/core/persist-downstream.test.ts` | 新增 / 更新 integration test 覆盖 5 case map | **Create / Modify** | 测试只 mirror `core/persist-downstream.ts`,handler 的 SQS envelope parsing 已有现成 test 不动。 *** ## Task 1: Bump `@retaintive/common` to 1.2.0 + import `domain/*` **Files:** - Modify: `package.json`(repo root,workspace catalog pin) - Modify: `lambda/lead-processor/src/core/persist-downstream.ts`(import block 1 处) **Why this task first:** Plan 02 把 `applyTaskAction` / `upsertIdentity` / `writeTimelineEvent` 放在 `@retaintive/common@1.2.0/domain` export path 下;Task 2-4 都依赖这个 import。先 ship 让 typecheck pass。 ### Step 1: Map test scenarios **原则:** 改 catalog 版本号 + 改 import 是 mechanical,**没有真 invariant 可 test**。只验证 typecheck pass —— 这就是"不 test framework / TS 编译保证的事"。 | # | Scenario | 为什么 test | | - | --------------------------------------------------------------------------------- | -------------------------------------------------- | | 1 | `bun run typecheck`(lambda/lead-processor)pass | catch import path typo;catch 新版 type 跟旧 caller 不兼容 | | 2 | `pnpm install` 后 `node_modules/@retaintive/common/package.json` version = `1.2.0` | catch catalog 解析没生效(workspace 拉错版本) | 不写其他 test。 ### Step 2: Implementation — bump catalog pin ```diff --- a/package.json (repo root) +++ b/package.json (repo root) "workspaces": { "packages": [...], "catalog": { - "@retaintive/common": "*", + "@retaintive/common": "*", ... } }, "dependencies": { - "@retaintive/common": "^1.0.0", + "@retaintive/common": "^1.2.0", ... } ``` > Note:catalog 值仍是 `"*"`(workspace protocol),实际 resolve 走 root `dependencies` 的 `^1.2.0`。如果 callytics-infrastructure root 同时有其他 caller 还在用 1.0.x range 满足的 API,1.2.0 是向后兼容的 minor bump(Plan 02 只新增 `domain/*` export,没改老 export),所以全 repo 升 1.2.0 安全。 ### Step 3: Implementation — 改 import block ```diff --- a/lambda/lead-processor/src/core/persist-downstream.ts +++ b/lambda/lead-processor/src/core/persist-downstream.ts import { computeDueAt } from '@retaintive/common'; -import { contacts, tasks, contactTimeline, buildTimelineValues, sql } from '@retaintive/common/db'; +import { sql } from '@retaintive/common/db'; import type { DrizzleClient, SuggestedAction, - TaskType, - TaskTypeCategory, - TaskStatus, - TaskPriority, - TaskSourceType, } from '@retaintive/common/db'; +import { + applyTaskAction, + upsertIdentity, + writeTimelineEvent, + NAME_TRUST, +} from '@retaintive/common/domain'; import { createLogger } from '../../../shared/utils/logger'; ``` `NAME_TRUST` 是 Plan 02 已定义的常量(`NAME_TRUST.LEAD = 80`)。`sql` template 暂留,Task 3 删除最后一个用户(`COALESCE(...)` 改 module 内做)。 ### Step 4: Verify + commit ```bash cd /Users/maxwsy/workspace/callytics-infrastructure pnpm install # 拉 @retaintive/common@1.2.0 cd lambda/lead-processor bun run typecheck # 必须 pass — 验证 import path 正确 + 类型存在 ``` Expected: typecheck PASS。如果挂在 `applyTaskAction` not exported / `domain` not a path → Plan 02 没 ship,先 fix Plan 02。 ### Step 5: Commit ```bash git add package.json lambda/lead-processor/src/core/persist-downstream.ts git commit -m "chore(lead-processor): bump @retaintive/common 1.0.x → 1.2.0 + import domain/* Phase 1 Plan 04 Task 1 — caller migration scaffolding. - Catalog pin bumps to ^1.2.0 (forward-compat minor, adds domain/* exports) - Import applyTaskAction / upsertIdentity / writeTimelineEvent / NAME_TRUST - Drop now-unused TaskType / TaskStatus / TaskPriority / TaskSourceType / TaskTypeCategory type imports (literal values move into payload objects validated by Plan 02 Zod schemas) - contacts / tasks / contactTimeline table imports removed in Tasks 2-4 Spec: normative-spec.md §4e " ``` *** ## Task 2: Replace task INSERT with `applyTaskAction({action:'create_open'})` **Files:** - Modify: `lambda/lead-processor/src/core/persist-downstream.ts`(line 185-206 block) - Test: `lambda/lead-processor/tests/core/persist-downstream.test.ts`(add test 1-5) **Why this task second:** task INSERT 是 spec §4e 表第一行(also the biggest mental jump);先做这个,Tasks 3-4 是更轻的替换。 ### Step 1: Map test scenarios — full case map 5 个 case 全枚举(对应 spec §3.1 + §3.6 触发的 reject reasons + §5.2 integration test 表): | # | Scenario | Setup | Expected behavior | | - | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | 1 | **Happy path** — 新 lead,contact 不存在 | LeadRow: phone=+15551234, storeId=S1, franchiseId=F1, accountId=A1, leadType='trial' | `applyTaskAction` returns `{status:'allow', statements:[insertStmt]}` → batch commits → tasks row 存在 status='open' + 1 timeline row + contact UPSERT row | | 2 | **Duplicate lead** — 同 lead.id 再来一次 | 同样 LeadRow,seed `tasks` 表已有 row w/ sourceLeadId=row\.id | `applyTaskAction` 返回 allow + statement 但 `onConflictDoNothing` (uq\_tasks\_source\_lead) 触发 → tasks 行数仍 1。**Module 层不抛 reject `duplicate`**(因为 module 用 partial unique 在 SQL 层兜底,caller 看到的是 batch commit 成功 + 0 new rows) | | 3 | **DNC contact** — contact 已存在且 `doNotContact=true` | Seed contact w/ doNotContact=true,然后来同 phone 的新 lead | `applyTaskAction` returns `{status:'reject', reason:'dnc', details:...}` → lead-processor 不调用 `client.batch`,**timeline 仍要写 `lead.created`**(audit trail 不能丢),contact 不再 UPSERT(避免 lastActivityAt 被刷新触发外部观察"DNC 用户还在活动")。logger.warn + Sentry breadcrumb,**不抛错**(SQS 不重试) | | 4 | **Terminal lead but actor=system** — contact lifecycleState='terminal'(非 churned)→ Policy Guard `checkTypeCategoryAllowed` 应 skip(因为 actor.type='system'),deterministic path 仍 allow | Seed contact w/ lifecycleState='terminal', lifecycleStage='lead';新 lead webhook 进 | `applyTaskAction({sourceType:'lead'})` 用 `actor.type='system'` → Plan 02 policy-guard.ts `SKIP_TYPE_CHECK_ACTORS` 含 'system' → skip allowed-set check → allow + insert task | | 5 | **Missing franchiseId/accountId** — LeadRow w/ franchiseId=null 或 accountId=null | LeadRow\.franchiseId=null | 现有 guard(line 111-119)在 `applyTaskAction` 调用**之前**已 return,**不进入 module**。这是 lead-processor handler 层的早期 validation,Plan 04 保留不动 | **为什么不 test 这些**(避免 over-test): - ❌ "applyTaskAction 内部 Policy Guard 各 check 工作正常" —— Plan 02 Task 2 已 test - ❌ "task.priority 从 suggestedActions 派生" —— Plan 02 Task 3 已 test(spec §3.1 S1) - ❌ "Drizzle batch 是原子事务" —— framework 行为,不 test - ❌ "`onConflictDoNothing` 在 partial unique 触发时不抛错" —— Drizzle 行为,不 test ### Step 2: Implementation — restructure persist-downstream 把 `client.batch([contactStmt, taskStmt, timelineStmt])` 拆成"先收集 statements → 再 batch"(让 reject path 可以早 return)。 ```typescript // lambda/lead-processor/src/core/persist-downstream.ts (block line 91-255 之后改造) export async function persistLeadDownstream(client: DrizzleClient, row: LeadRow): Promise { // (line 92-119 的 guard 保持不变 —— storeId null / phone null / franchiseId null / accountId null // 全部 early return + logger,不进入 module) if (row.storeId == null) { /* unchanged */ return; } if (!row.phone || !row.franchiseId || !row.accountId) { /* unchanged */ return; } const now = new Date(); const dueAt = computeDueAt('high', now, { slaMinutes: 5 }); const suggestedActions: SuggestedAction[] = [ { action: 'Call the lead back promptly', reason: 'New lead requires immediate outreach within SLA', priority: 'high', priorityReason: 'Lead SLA is 5 minutes — speed to contact is critical for conversion', }, ]; /* * Build task statement via Plan 02 Task Orchestrator. * * NOTE: applyTaskActions(决策array, ctx)是 batch entry(Plan 02 §3.1 line 1255) * — 即使 lead-processor 只 1 个 decision,也 wrap array 保持 contract 一致。 * Orchestrator 内部 loadSnapshot 一次,然后 dispatch。 * * actor.type='system' skips AI allowed-typeCategory check (lead_outreach is * deterministic lead-tracking entry, not AI-proposed). Still goes through * storeId / DNC / duplicate (partial unique uq_tasks_source_lead) guards. */ const [taskResult] = await applyTaskActions( [{ action: 'create_open', payload: { typeCategory: 'lead_outreach', suggestedActions, dueAt, sourceType: 'lead', sourceLeadId: row.id, }, }], { storeId: row.storeId, contactPhone: row.phone, franchiseId: row.franchiseId, accountId: row.accountId, actor: { type: 'system', subjectId: 'lead-processor' }, db: client, }, ); /* * taskId 由 Orchestrator 生成并 expose 在 generatedIds —— 不再 caller-side * crypto.randomUUID()。Timeline payload 引用同一 taskId,避免 caller-side * UUID 跟 Orchestrator 内部 UUID 不一致(spec §3.1 line 203 contract)。 * * reject 路径(DNC / duplicate / typeCategory)taskResult.generatedIds 仍存在 * 但 taskId 未 INSERT — timeline 不引用(fallback null)。 */ const generatedTaskId = taskResult.status === 'allow' ? taskResult.generatedIds?.taskId : undefined; /* * Build timeline + contact statements regardless of taskResult status. * - Timeline: lead.created MUST be written for audit even if task is rejected * (DNC case: we still need to record the inbound lead event). * - Contact: skip UPSERT on DNC reject to avoid refreshing lastActivityAt for * a contact that explicitly opted out (would mislead "active contact" filters). * * writeTimelineEvent(client, params) — 双 arg(Plan 02 §3.5 line 1545)。 * entityType / entityId 是 flat;actor.subjectId 是 nested 但只这一层。 * payload conforms PAYLOAD_SCHEMAS['lead.created'] Zod(Plan 02 line 1502): * { leadId, leadType, source } * payload 不再含 firstName/lastName(那些 caller-side context 不进 timeline JSONB) * 也不再含 taskId(generatedTaskId 通过 entityId reference 即可) */ const timelineStmt = writeTimelineEvent(client, { eventType: 'lead.created', entityType: 'lead', entityId: row.id, contactPhone: row.phone, storeId: row.storeId, actor: { type: 'system', subjectId: 'lead-processor', name: 'lead-processor lambda', }, payload: { leadId: row.id, leadType: row.leadType, source: row.sourceSystem ?? 'lead-tracking', // Zod required }, occurredAt: now, idempotencyKey: `lead.created:${row.id}`, }); if (taskResult.status === 'reject') { // Audit trail still recorded (timeline only). No contact UPSERT under DNC. logger.warn('applyTaskActions rejected lead — writing audit timeline only', { leadId: row.id, phone: maskPhone(row.phone), reason: taskResult.reason, details: taskResult.details, }); await client.batch([timelineStmt] as any); return; } /* * Allow path: contact UPSERT + task INSERT statements + timeline, all atomic. * * upsertIdentity(client, params) — 双 arg(Plan 02 §3.4 line 1249)。 * NAME_TRUST.LEAD (80) reflects lead-form provenance; lower than STAFF (100) * but higher than AI_TRANSCRIPT (40). Winner-takes-name pattern in module。 */ const contactStmt = upsertIdentity(client, { phone: row.phone, storeId: row.storeId, franchiseId: row.franchiseId, accountId: row.accountId, firstName: row.firstName ?? undefined, lastName: row.lastName ?? undefined, trustScore: NAME_TRUST.LEAD, activityAt: now, }); const batchResults = await client.batch([ contactStmt, ...taskResult.statements, timelineStmt, ] as any); /* * resultChecks contract(normative-spec §3.1 line 251): * Orchestrator 用 0-row RETURNING 标记 silent business reject。 * 这里 statements layout 是: [contact, ...taskStmts, timeline] * resultChecks[i].statementIndex 是相对 taskResult.statements 的(从 0 起), * 我们 offset +1(contactStmt 之后开始)拿全局位置: */ const TASK_STMTS_OFFSET = 1; for (const check of taskResult.resultChecks ?? []) { const globalIdx = TASK_STMTS_OFFSET + check.statementIndex; const result = batchResults[globalIdx]; if (result?.rowCount === 0) { logger.warn('Lead task creation silent reject', { leadId: row.id, phone: maskPhone(row.phone), reason: check.zeroRowsReason, // 'duplicate' for sourceLeadId already used }); // duplicate 是预期 idempotent path(lead webhook retry),不算 error。 // Phase 2 加 metrics: mutation_rejected_total{reason='duplicate'} } } logger.info('Lead downstream persist complete', { leadId: row.id, taskId: generatedTaskId, phone: maskPhone(row.phone), storeId: row.storeId, leadType: row.leadType, }); } function maskPhone(phone: string): string { return phone.length > 4 ? `***${phone.slice(-4)}` : '***'; } ``` **关键选择 + rationale:** 1. **DNC reject 仍写 timeline `lead.created`** —— audit trail 不能漏(spec §5.0 invariant "timeline audit"),但跳过 contact UPSERT 不刷新 `lastActivityAt`(避免"DNC 用户仍活跃" 假象)。 2. **`actor.type='system'`(不是 `'lead_webhook'`)** —— Plan 02 `BaseActionContext.actor.type` 只接受 4 个值(`staff` / `system` / `ai_agent` / `contact_analysis`)。`'lead_webhook'` 是 **timeline `actor.type`**(spec/audit 命名空间),跟 Policy Guard 的 actor.type 是两套独立词表。Policy Guard 用 `'system'` 触发 `SKIP_TYPE_CHECK_ACTORS` 短路;timeline 仍记 `'lead_webhook'` 保留原 audit 语义。 3. **`taskId` 不再 caller 生成** —— Plan 02 `applyTaskAction` 内部生成并通过 `statements` 的 `RETURNING` 暴露。Plan 04 在 timeline payload 里仍用 caller 生成的 `taskId`(变量在作用域内),但这个 taskId 不再喂给 INSERT —— Orchestrator 自己 generate。**如果 Plan 02 设计是 Orchestrator 接受 caller 传 taskId**(看 Task 3 Step 3 line 771-789 `taskId` 字段未显式 set,Drizzle 用 schema default `gen_random_uuid()`),则 caller 这边的 `taskId` 只用于 timeline payload 标记,实际写入 DB 的 task\_id 是 default 生成的。**Implementer 落地时 verify** Plan 02 是否提供 way to pass caller-generated taskId;如否,timeline payload 的 `taskId` 字段语义就是"建议的关联 task,实际 DB id 看 task\_progress\_events / task\_id"。这个细节在 Plan 02 Task 3 Step 3 line 771 evident(`values()` 中无 taskId field)。 4. **删 `taskType: 'lead_outreach'`** —— hardcoded 不必由 caller 写,Plan 02 Task 3 Step 3 line 778 已在 Orchestrator 内做 legacy shim(`typeCategory === 'lead_outreach' ? 'lead_outreach' : 'follow_up'`),过渡期写入,final cutover DROP COLUMN。 5. **删 `actionNeeded: true`** —— 字段已退役(spec §2.2 line 119)。 6. **status 不再 caller 写 `'pending'`** —— Plan 02 line 779 `status: 'open'` 由 Orchestrator 写入 final value(测试环境直接走 final 字符串,不留过渡)。 7. **`onConflictDoNothing` 不再 caller 显式调** —— Plan 02 Task 3 Step 3 line 787 `.onConflictDoNothing()` 已在 Orchestrator builder 内,partial unique 行为 1:1 保留。 8. **`actionNeededReason` 字段** —— 原 hardcoded(`New ${row.leadType} lead requires outreach`),Plan 02 `CreateOpenPayload` 不暴露该字段(observe-only 写读分离,Phase 2 删 column,spec line 22)。**这条字段 lead-processor 写入断开** —— `contacts.actionNeeded` Phase 1 仍存在但读路径走 EXISTS(spec §4g),**tasks.action\_needed 已 DROP**。`actionNeededReason` 是 `tasks` 表已 drop 的字段(spec line 117-124 `DROP COLUMN action_needed`)or `contacts.actionNeededReason`?重读 spec: 老代码 line 200 写 `actionNeededReason` 是给 `tasks` row 的;Plan 02 Task 3 Step 3 line 771-789 `values()` 不含 `actionNeededReason` → drop column 时一起没了。lead-processor 不再写,**不补 Phase 2 内容**。 ### Step 3: Test write — sample 2 tests for style guidance Implementer 按 case map 1-5 写完整 5 个 test,以下 sample 1 + sample 3 展示 style: ```typescript // lambda/lead-processor/tests/core/persist-downstream.test.ts import { describe, it, expect, beforeEach } from 'bun:test'; import { persistLeadDownstream, type LeadRow } from '../../src/core/persist-downstream'; import { createTestDrizzleClient, seedContact, queryTasks, queryTimeline } from '../helpers/db'; const LEAD_BASE: LeadRow = { id: 'lead-001', leadEmail: 'a@b.com', leadType: 'trial', firstName: 'John', lastName: 'Doe', phone: '+15551234567', bookedDate: null, bookedTime: null, emailSubject: null, emailFrom: null, emailRecipient: null, extractedTrackingId: null, isForwarded: false, forwardedOriginalFrom: null, forwardedOriginalTo: null, forwardedOriginalDate: null, rawBody: null, processedBy: null, receivedAt: new Date(), franchiseId: 'F1', accountId: 'A1', storeId: 'S1', }; describe('persistLeadDownstream — Plan 04', () => { let db: ReturnType; beforeEach(async () => { db = createTestDrizzleClient(); await db.execute(`TRUNCATE tasks, contacts, contact_timeline CASCADE`); }); // Case 1: happy path it('writes contact + task + timeline atomically for a new lead', async () => { await persistLeadDownstream(db, { ...LEAD_BASE }); const tasks = await queryTasks(db, { contactPhone: '+15551234567', storeId: 'S1' }); expect(tasks).toHaveLength(1); expect(tasks[0]).toMatchObject({ typeCategory: 'lead_outreach', sourceType: 'lead', sourceLeadId: 'lead-001', status: 'open', priority: 'high', }); const timeline = await queryTimeline(db, { contactPhone: '+15551234567' }); expect(timeline.map(t => t.eventType)).toContain('lead.created'); }); // Case 3: DNC contact — task rejected, timeline still written, contact NOT touched it('rejects task but still writes lead.created timeline when contact is DNC', async () => { await seedContact(db, { phone: '+15551234567', storeId: 'S1', franchiseId: 'F1', accountId: 'A1', doNotContact: true, lastActivityAt: new Date('2026-01-01'), // stale, must NOT be refreshed }); await persistLeadDownstream(db, { ...LEAD_BASE }); const tasks = await queryTasks(db, { contactPhone: '+15551234567', storeId: 'S1' }); expect(tasks).toHaveLength(0); // task rejected const timeline = await queryTimeline(db, { contactPhone: '+15551234567' }); expect(timeline.map(t => t.eventType)).toContain('lead.created'); // audit preserved const [contact] = await db.execute( `SELECT last_activity_at FROM contacts WHERE phone='+15551234567' AND store_id='S1'`, ); expect(contact.last_activity_at).toEqual(new Date('2026-01-01')); // NOT refreshed }); // Cases 2, 4, 5 follow same shape — implementer fills in }); ``` ### Step 4: Commit ```bash cd /Users/maxwsy/workspace/callytics-infrastructure bun test lambda/lead-processor/tests/core/persist-downstream.test.ts # expect 5 PASS git add lambda/lead-processor/src/core/persist-downstream.ts \ lambda/lead-processor/tests/core/persist-downstream.test.ts git commit -m "feat(lead-processor): task INSERT → applyTaskAction({action:'create_open'}) Phase 1 Plan 04 Task 2 — caller migration core. - Replace inline client.insert(tasks).onConflictDoNothing() with applyTaskAction({action:'create_open', payload:{...}}) via Plan 02 Task Orchestrator - actor.type='system' triggers Policy Guard SKIP_TYPE_CHECK_ACTORS (lead_outreach is deterministic, not AI-proposed) - Drop hardcoded taskType='lead_outreach' (Orchestrator writes legacy shim in transition window, DROPped final cutover) - Drop actionNeeded:true (column retired, spec §2.2) - Drop actionNeededReason on task (column gone with action_needed DROP) - status writes 'open' (final value, not 'pending') - DNC reject path: write lead.created timeline only (audit trail intact), skip contact UPSERT (don't refresh lastActivityAt for opted-out contacts) - Integration tests: 5 case map (happy / duplicate / DNC / terminal-system / missing-identity early-return) Spec: normative-spec.md §4e + §3.6 + §5.0 invariants " ``` *** ## Task 3: Replace contact UPSERT with `upsertIdentity()` (B5: franchiseId + accountId) **Files:** - Modify: `lambda/lead-processor/src/core/persist-downstream.ts`(contact block 已在 Task 2 改造时引入 `upsertIdentity`,Task 3 处理边角 case + 验证 B5 NOT NULL 行为) - Test: `lambda/lead-processor/tests/core/persist-downstream.test.ts`(add 2 tests) **Why this task third:** Task 2 已用 `upsertIdentity` 替代了 contact UPSERT 主体;Task 3 单独立 task 是为了 **验证 B5 fix**(spec §3.4 line 364-374:`contacts.franchise_id` / `account_id` NOT NULL,upsert 时新建 contact 必须提供;ON CONFLICT DO UPDATE path 不更新 franchise/account)以及 **winner-takes-name 行为**(NAME\_TRUST.LEAD=80 vs 已存在 staff 输入 100 的优先级)。 ### Step 1: Map test scenarios | # | Scenario | Setup | Expected behavior | | - | ---------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------ | | 1 | **B5 — 新 contact 必须有 franchiseId+accountId** | LeadRow franchiseId=F1, accountId=A1, phone 不存在于 contacts | `upsertIdentity` INSERT path → contacts row franchise\_id=F1, account\_id=A1 | | 2 | **B5 — 已存在 contact,ON CONFLICT 不动 franchise/account** | Seed contact w/ franchise\_id=F\_OLD, account\_id=A\_OLD;LeadRow franchiseId=F\_NEW, accountId=A\_NEW(模拟 webhook 数据 drift) | 写入后 contact.franchise\_id 仍 = F\_OLD,account\_id 仍 = A\_OLD(spec §3.4 "ON CONFLICT 时 franchise/account 不更新") | | 3 | **Winner-takes-name** — staff 已写过 firstName='Robert' trustScore=100,新 lead 写 firstName='Bob' trustScore=80 | Seed contact w/ first\_name='Robert', first\_name\_trust\_score=100;LeadRow firstName='Bob' | 写入后 contact.first\_name = 'Robert'(保留高分),trust\_score = 100(不降级) | | 4 | **Forward-only activityAt** — 已存在 contact.last\_activity\_at=2026-06-01,新 lead 在 2026-01-01 来(乱序消息) | Seed contact lastActivityAt=2026-06-01;LeadRow receivedAt=2026-01-01,now=2026-01-01 | 写入后 contact.last\_activity\_at 仍 = 2026-06-01(GREATEST 保护) | case 3-4 只做 caller integration smoke:验证 lead-processor 传了 `NAME_TRUST.LEAD` 和 `activityAt` 后最终 DB outcome 合理。不要 assert `upsertIdentity()` 的 SQL fragment / CASE WHEN / GREATEST 细节;完整 NAME\_TRUST matrix 和 forward-only matrix 由 Plan 02 module tests 覆盖。 ### Step 2: Implementation — verify `upsertIdentity` call site(Task 2 已写) Task 2 写出的 `upsertIdentity` 调用: ```typescript const contactStmt = upsertIdentity({ phone: row.phone, storeId: row.storeId, franchiseId: row.franchiseId, // B5: NOT NULL, ALWAYS passed from LeadRow accountId: row.accountId, // B5: NOT NULL, ALWAYS passed from LeadRow firstName: row.firstName ?? undefined, lastName: row.lastName ?? undefined, trustScore: NAME_TRUST.LEAD, // 80 — lead-form provenance activityAt: now, // forward-only GREATEST in module }); ``` **guard 验证**:`row.franchiseId` / `row.accountId` 在 line 111-119 已被早期 return guard 拦住 null(`if (!row.phone || !row.franchiseId || !row.accountId) return`),所以传给 `upsertIdentity` 时 **类型上不可能是 null**。Task 3 不改动 implementation,只新增 test 验证这条 chain。 **为什么不传 `lifecycleStage: 'lead'` / `leadStatus: 'new'`** —— 原 code line 163-164 caller 显式设这两个字段。Plan 02 `upsertIdentity` signature(spec §3.4 line 365-375)**只接受 identity / DNC / activity 三种字段**,不接受 lifecycleStage / leadStatus。Phase 1 的设计选择是:**lifecycle/leadStatus 由 contacts-analyzer 在 AI 分析后写**,lead-processor 不再 caller-side 拍 `'lead'` / `'new'`(避免 AI 已经把这个 contact 推进到 `'qualified'` 后 lead-webhook 又来推回 `'new'` 的回退 bug)。INSERT path(新 contact)走 DB-level DEFAULT —— check `contacts.lifecycleStage` 默认值。 **Verify 这条 frame 是否成立:** ```bash grep -A 2 "lifecycle_stage" /Users/maxwsy/workspace/callytics-common/src/db/schema/contacts.ts | head -10 ``` 如果 `lifecycleStage` 无 DEFAULT(NOT NULL 缺 default)→ Plan 02 `upsertIdentity` INSERT 会 fail。**Implementer 落地前 verify**。如果 spec/schema 漂移,要么扩 `upsertIdentity` payload(spec change),要么在 `upsertIdentity` 内部硬编 `DEFAULT 'unknown'`(也 spec change)。这是 Plan 04 落地时的 known risk,记下来。 **降级方案(如果 verify 后 lifecycleStage 必须 caller 传):** - 短期:Plan 04 implementer 在调 `upsertIdentity` 前先 `client.insert(contacts).values({phone, storeId, franchiseId, accountId, lifecycleStage:'lead', leadStatus:'new'}).onConflictDoNothing()` 兜底建 row,再 `upsertIdentity` 走 ON CONFLICT path - 长期:Plan 02 v1.2.1 patch 扩 `upsertIdentity` 加可选 `lifecycleStage` / `leadStatus`(deferred,track 为 follow-up) ### Step 3: Test write — sample 1 test ```typescript // (在 Task 2 写的 describe block 内继续) // Case 6 (Task 3 Step 1 #2): ON CONFLICT 不更新 franchise/account it('does not update franchiseId/accountId on existing contact (B5 fix)', async () => { await seedContact(db, { phone: '+15551234567', storeId: 'S1', franchiseId: 'F_OLD', accountId: 'A_OLD', }); await persistLeadDownstream(db, { ...LEAD_BASE, franchiseId: 'F_NEW', accountId: 'A_NEW', }); const [contact] = await db.execute( `SELECT franchise_id, account_id FROM contacts WHERE phone='+15551234567' AND store_id='S1'`, ); expect(contact.franchise_id).toBe('F_OLD'); expect(contact.account_id).toBe('A_OLD'); }); // Cases 7, 8 (Task 3 Step 1 #3, #4) — winner-takes-name + forward-only activityAt — follow same shape ``` ### Step 4: Commit ```bash bun test lambda/lead-processor/tests/core/persist-downstream.test.ts # expect Task 2's 5 + Task 3's 3 = 8 PASS git add lambda/lead-processor/tests/core/persist-downstream.test.ts git commit -m "test(lead-processor): contact UPSERT — B5 franchise/account NOT NULL + winner-takes-name Phase 1 Plan 04 Task 3 — caller integration tests for upsertIdentity(). - Verify B5: INSERT path requires franchiseId + accountId; ON CONFLICT DO UPDATE preserves existing franchise/account (no overwrite) - Verify NAME_TRUST.LEAD=80 passed correctly (staff 100 wins over lead 80) - Verify forward-only activityAt via GREATEST (stale message order safe) Note: lifecycleStage / leadStatus no longer caller-set — relies on contacts schema DEFAULT or contacts-analyzer subsequent write. Verified schema has lifecycleStage DEFAULT 'unknown' [or equivalent]. If schema verify fails at implementation time, fall back to bridge INSERT before upsertIdentity (see Plan 04 Task 3 Step 2 'Verify this frame'). Spec: normative-spec.md §3.4 B5 " ``` *** ## Task 4: Replace timeline INSERT with `writeTimelineEvent()` **Files:** - Modify: `lambda/lead-processor/src/core/persist-downstream.ts`(timeline block 已在 Task 2 改造) - Test: `lambda/lead-processor/tests/core/persist-downstream.test.ts`(add 2 tests) **Why this task fourth:** Task 2 已用 `writeTimelineEvent` 替代了 timeline INSERT;Task 4 单独立是为了 **验证 spec §4j 的 source event 写入规则**(`lead.created` event 不经 Policy Guard,直接走 Timeline Writer)+ **idempotencyKey 防重**。 ### Step 1: Map test scenarios | # | Scenario | Setup | Expected behavior | | - | -------------------------------------------------------------------------------------------------------- | ----------------------------------- | ---------------------------------------------------------------------------------------------------------------- | | 1 | **`lead.created` event type 验证** — payload 必须 match `TIMELINE_EVENT_TYPES` 中 `lead.created` 的 Zod schema | LeadRow happy path | `writeTimelineEvent` 不 throw;timeline row eventType='lead.created' | | 2 | **Idempotency on SQS retry** — 同 leadId 再来一次(模拟 SQS at-least-once 重投递) | LeadRow leadId='lead-001' 第一次 + 第二次 | 第二次 ON CONFLICT idempotency\_key='lead.created:lead-001' DO NOTHING → timeline 表 row 数仍 = 1 | | 3 | **`actor.type='lead_webhook'`(timeline 语义,跟 Policy Guard `system` 区分)** | LeadRow happy path | timeline row actor\_type='lead\_webhook',actor\_source\_type='integration',actor\_source\_system='lead\_webhook' | **不 test 的(避免重复 Plan 02 module test)**: - ❌ Zod schema validation 各字段 happy/reject(Plan 02 Task 5 覆盖) - ❌ ON CONFLICT idempotency\_key 行为(DB constraint 行为)—— 只 test caller 传的 idempotencyKey 字符串格式对 ### Step 2: Implementation — verify `writeTimelineEvent` call site(Task 2 已写) ```typescript const timelineStmt = writeTimelineEvent({ eventType: 'lead.created', entity: { type: 'lead', id: row.id }, contactPhone: row.phone, storeId: row.storeId, actor: { type: 'lead_webhook', // 注意:timeline 词表,不是 Policy Guard 词表 id: ACTOR_SUBJECT_ID, // 'lead_webhook' sourceType: ACTOR_SOURCE_TYPE, // 'integration' sourceSystem: ACTOR_SOURCE_SYSTEM, // 'lead_webhook' }, payload: { leadType: row.leadType, firstName: row.firstName, lastName: row.lastName, taskId, // caller-known UUID(虽 task INSERT 内部生成,timeline 这里只是关联标记) }, occurredAt: now, idempotencyKey: `lead.created:${row.id}`, // 同原 code line 236 }); ``` **关键 frame:** 1. **`actor.type` 词表差异** —— Plan 02 `writeTimelineEvent` 的 `actor.type` 字段是 timeline `contact_timeline.actor_type` 列(不是 Policy Guard `BaseActionContext.actor.type`),接受值由 schema 决定(`contact_timeline` 历史一直是 'staff' / 'system' / 'ai\_agent' / 'contact\_analysis' / **'lead\_webhook'** / ...)。**Implementer verify** schema 词表: ```bash grep -A 5 "actorType" /Users/maxwsy/workspace/callytics-common/src/db/schema/contact-timeline.ts | head -20 ``` 如果 schema 不接受 `'lead_webhook'`,降级到 `'system'`(并在 timeline payload 里加 `originatedFrom: 'lead_webhook'` 字段保留 audit context)。 2. **`payload.taskId` 来源** —— caller 生成的 `crypto.randomUUID()`(原 code line 122),timeline 的 payload 里仍保留(便于审计串联),但 **Plan 02 Task Orchestrator 是否真的 INSERT 这个 taskId** 是 Plan 02 内部细节 —— 如果 Orchestrator 自己 generate,timeline payload 的 taskId 跟 tasks.task\_id 可能不 match。**Implementer 落地时 verify** Plan 02 是否提供 caller-passed taskId hook;如否,timeline payload 删 taskId(无法保证正确)或改为 Orchestrator 返回的 taskId(需要 module API 改 —— Plan 02 v1.2.1 follow-up)。**当前 Plan 04 保留 taskId 字段,落地时 fix。** ### Step 3: Test write — sample 1 test ```typescript // Case 9 (Task 4 Step 1 #2): idempotent on SQS retry it('writes only one lead.created timeline row on duplicate SQS delivery', async () => { await persistLeadDownstream(db, { ...LEAD_BASE }); await persistLeadDownstream(db, { ...LEAD_BASE }); // SQS retry const timeline = await queryTimeline(db, { contactPhone: '+15551234567', eventType: 'lead.created', }); expect(timeline).toHaveLength(1); expect(timeline[0].idempotencyKey).toBe('lead.created:lead-001'); }); // Cases 10, 11 (Task 4 Step 1 #1, #3) — eventType/Zod + actor.type='lead_webhook' — follow same shape ``` ### Step 4: Commit ```bash bun test lambda/lead-processor/tests/core/persist-downstream.test.ts # expect 5 + 3 + 3 = 11 PASS git add lambda/lead-processor/tests/core/persist-downstream.test.ts git commit -m "test(lead-processor): timeline writer — lead.created event + idempotency Phase 1 Plan 04 Task 4 — integration tests for writeTimelineEvent() at the lead.created source event. - spec §4j: source events (lead.created / call_analysis.completed / message.created) bypass Policy Guard, written directly by caller via Timeline Writer. lead-processor preserves this path. - Verify idempotencyKey='lead.created:\${leadId}' deduplicates on SQS retry (at-least-once delivery contract) - Verify timeline actor.type='lead_webhook' (audit semantic), distinct from Policy Guard actor.type='system' used in task orchestrator call Spec: normative-spec.md §4j + §3.5 + §5.0 'timeline audit' invariant " ``` *** ## Task 5: Final verification + commit + PR **Files:** - Verify all 3 module integrations work together - Run full test suite - Run end-to-end smoke against test Neon - Create PR with `pr` skill ### Step 1: Verification commands ```bash cd /Users/maxwsy/workspace/callytics-infrastructure # 1. Typecheck whole lambda cd lambda/lead-processor && bun run typecheck && cd ../.. # 2. Run all lead-processor tests bun test lambda/lead-processor/tests/ # 3. Static check: no remnant `task_type` / `actionNeeded` / `'pending'` literals in lead-processor grep -nE "(taskType|actionNeeded|'pending')" lambda/lead-processor/src/core/persist-downstream.ts \ && echo "FAIL: legacy literal remains" || echo "OK: no legacy literals" # 4. Static check: no direct contacts / tasks / contactTimeline table import grep -nE "from '@retaintive/common/db'" lambda/lead-processor/src/core/persist-downstream.ts \ | grep -E "(contacts|tasks|contactTimeline|buildTimelineValues)" \ && echo "FAIL: direct table import remains" || echo "OK: domain/* only" # 5. Integration test against real test Neon NEON_DATABASE_URL=$TEST_NEON_URL bun test lambda/lead-processor/tests/core/persist-downstream.test.ts ``` Expected: all PASS,2 static checks output "OK"。 ### Step 2: Self-Review checklist | Check | Method | Pass? | | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------ | ----- | | **Spec §4e coverage** — line 590-599 table 4 rows all addressed | grep `taskType` / `actionNeeded` / `'pending'` not in src | | | **Spec §3.4 B5** — franchiseId / accountId always passed to upsertIdentity | re-read Task 3 Step 2 + Task 3 test cases | | | **Spec §4j** — lead.created written via TimelineWriter, not Policy Guard | re-read Task 4 Step 2 | | | **Spec §5.0 invariants** — "deterministic lead path" / "timeline audit" / "tenant isolation" pass | run integration tests case 1, 4 | | | **Type consistency with Plan 02** — `applyTaskAction({action:'create_open'})` payload matches `CreateOpenPayload` in Plan 02 Task 1 line 102-113 | diff payload object field names against Plan 02 line 102-113 | | | **No placeholders** — every file path / type / commit message concrete | read every step | | | **5 case map covered in tests** — Task 2 cases 1-5 + Task 3 cases 6-8 + Task 4 cases 9-11 | count test `it(...)` blocks | | | **DNC behavior consistent with spec §3.6 line 438** — reject create\_open + timeline still written + contact NOT touched | re-read Task 2 implementation lines 50-70 + test case 3 | | | **Frame uncertainty acknowledged** — Plan 02 taskId hook + contacts.lifecycleStage DEFAULT + timeline actor.type vocab — all 3 'verify at implementation time' notes recorded | grep for "verify" / "落地时" in plan | | ### Step 3: PR creation via `pr` skill ```bash # Bring all 4 task commits together — feature branch ready git log --oneline main..HEAD # Expect 4 commits matching Task 1-4 messages # Invoke pr skill to generate Chinese PR body — 必须用 pr skill, 不凭 fallback template # Skill outputs 8-section structured PR body (背景 / 改了什么 / 改前 vs 改后 / 使用方式 / # 改动文件 / 部署 / 关联 / Test plan) ``` PR title example: `feat(lead-processor): caller migration → applyTaskAction + upsertIdentity + writeTimelineEvent (Phase 1 Plan 04)` PR body 必含: - 改前 vs 改后 SQL block diff(line 185-206 task INSERT 旧 → 新 module 调用) - Plan 02 dependency note(`@retaintive/common@1.2.0` 必须先 ship) - spec §4e + §5.0 invariant cross-reference - "落地时 verify" 3 个 known-risk items 列在 PR 描述里(taskId hook + lifecycleStage DEFAULT + actor.type vocab) ### Step 4: Push + create PR ```bash git push -u origin gh pr create --title "..." --body-file /tmp/lead-processor-migration-pr-body.md ``` (Main session 统一 push,subagent 不直接 push — Plan 04 不 push,只准备好 commits + PR draft body) *** ## Appendix A — Frame Uncertainties Recorded at Plan Time 落地 implementer 必须 verify 的 3 点(spec / Plan 02 现状不够明确,Plan 04 选最保守路径,但 verify 失败需 fix): | Uncertainty | 当前 plan 假设 | Verify 方法 | 失败时降级 | | ---------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | **Plan 02 是否允许 caller 传 taskId 给 `applyTaskAction({action:'create_open'})`** | Plan 02 Task 3 Step 3 line 771-789 `.values()` 无显式 taskId,假设 Orchestrator 内部 generate(走 schema default `gen_random_uuid()`) | 读 Plan 02 v1.2.0 published source `src/domain/task-orchestrator.ts` `handleCreateOpen` impl | 若 Orchestrator 内部 generate → timeline payload taskId 改用 Orchestrator 返回值;若需 caller 传 → Plan 04 加 `taskId: crypto.randomUUID()` 到 payload(API 已支持) | | **`contacts.lifecycleStage` 是否有 DB DEFAULT** | 假设有(`'unknown'` or 类似),允许 `upsertIdentity()` INSERT path 不传该字段 | `grep lifecycle_stage /Users/maxwsy/workspace/callytics-common/src/db/schema/contacts.ts` | 无 DEFAULT → Plan 04 Task 3 Step 2 降级方案:先 `client.insert(contacts).values({lifecycleStage:'lead', leadStatus:'new'}).onConflictDoNothing()`,再 `upsertIdentity()` 走 ON CONFLICT path | | **`contact_timeline.actor_type` 是否接受 `'lead_webhook'` 字面值** | 假设接受(原 code line 232 一直在写) | `grep -A 5 actorType /Users/maxwsy/workspace/callytics-common/src/db/schema/contact-timeline.ts` 或读 schema CHECK constraint | 不接受 → 降级到 `'system'` + 在 timeline payload 加 `originatedFrom: 'lead_webhook'` 保留 audit context | *** ## Appendix B — Not in Plan 04 scope 明确不做(避免 scope creep): - ❌ `handler.ts` 改造(SQS envelope 解析 + Drizzle client lazy init 跟 mutation 改造无关) - ❌ `lambda/lead-processor` 的 dependency 升级(Sentry / aws-sdk 等) - ❌ Phase 2 `contacts.actionNeeded` DROP COLUMN(spec §6.4 follow-up) - ❌ `buildContactTimelineInsertSQL` 删除(spec §6.4,等 studio-api Drizzle 迁移 #449 后) - ❌ Plan 03(contacts-analyzer caller)/ Plan 05(message-processor)/ Plan 06(studio-api)的改造 —— 独立 plan - ❌ schema migration(Plan 01 scope) - ❌ `applyTaskAction` 内部 logic 验证(Plan 02 test 覆盖) *** ## Estimated LoC | Category | LoC | | -------------------------------------------------------------------- | ------------------------------------------------------------- | | `package.json` catalog bump | \~2 | | `src/core/persist-downstream.ts` 改造(import block + body restructure) | \~120(净:-30 老 SQL,+90 module 调用 + DNC branch + masked helper) | | `tests/core/persist-downstream.test.ts` 新增 11 test | \~450(每个 test \~40 line w/ setup + assert) | | commit messages × 4 | \~80 | | `tests/helpers/db.ts` 复用(seedContact / queryTasks / queryTimeline) | \~150(若尚未存在;假设已在 plan 03 / 02 ship 时建好,Plan 04 不创建) | | **Total Plan 04** | **\~650-800**(不算 helpers reuse;算 helper 首次创建则 \~850-1000) | **vs Plan 03 估计 LoC**:Plan 04 简单一半 —— lead-processor 只 1 个 caller path,无 TaskDecision 6 action union loop,无 derive logic;Plan 03 是 \~1500-2000。 *** ## Summary | Task | 内容 | LoC | | ------ | ------------------------------------------------------------------------------------------- | --------- | | 1 | Bump catalog `@retaintive/common` 1.0.x → 1.2.0 + import `domain/*` | \~20 | | 2 | Task INSERT → `applyTaskAction({action:'create_open'})` + DNC branch | \~250 | | 3 | Contact UPSERT 已 Task 2 引入 `upsertIdentity()`;此 task 验证 B5 NOT NULL + winner-takes-name | \~150 | | 4 | Timeline INSERT 已 Task 2 引入 `writeTimelineEvent()`;此 task 验证 §4j source event + idempotency | \~150 | | 5 | 整体 verification + commit + PR(pr skill) | \~80 | | **总计** | 4 implementation tasks + 1 verification task | **\~650** | **关键设计选择**: 1. **DNC reject 写 timeline 不写 contact** —— audit trail 不能漏,但不刷新 lastActivityAt 误标"DNC 用户活跃" 2. **`actor.type='system'`(Policy Guard 词表)+ `actor.type='lead_webhook'`(timeline 词表)分离** —— Policy Guard 用 system 触发 SKIP\_TYPE\_CHECK\_ACTORS 短路;timeline 保留 lead\_webhook 原 audit 语义 3. **不再 caller 写 lifecycleStage / leadStatus / actionNeeded / taskType / 'pending' 字符串** —— 全删,Orchestrator 内部写 final values(过渡期 legacy shim 由 Plan 02 处理,Plan 04 caller 无感知) 4. **3 个 frame uncertainties 记录到 Appendix A,落地 verify 后再选最终路径** —— 不假装这些细节已 settled