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
测试只 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 编译保证的事"。
不写其他 test。
Step 2: Implementation — bump catalog pin
--- 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
--- 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
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
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 表):
为什么不 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)。
// lambda/lead-processor/src/core/persist-downstream.ts (block line 91-255 之后改造)
export async function persistLeadDownstream(client: DrizzleClient, row: LeadRow): Promise<void> {
// (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:
- DNC reject 仍写 timeline
lead.created —— audit trail 不能漏(spec §5.0 invariant "timeline audit"),但跳过 contact UPSERT 不刷新 lastActivityAt(避免"DNC 用户仍活跃" 假象)。
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 语义。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)。- 删
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。
- 删
actionNeeded: true —— 字段已退役(spec §2.2 line 119)。
- status 不再 caller 写
'pending' —— Plan 02 line 779 status: 'open' 由 Orchestrator 写入 final value(测试环境直接走 final 字符串,不留过渡)。
onConflictDoNothing 不再 caller 显式调 —— Plan 02 Task 3 Step 3 line 787 .onConflictDoNothing() 已在 Orchestrator builder 内,partial unique 行为 1:1 保留。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:
// 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<typeof createTestDrizzleClient>;
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
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
"
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
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 调用:
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 是否成立:
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
// (在 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
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
不 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 已写)
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:
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 词表:
grep -A 5 "actorType" /Users/maxwsy/workspace/callytics-common/src/db/schema/contact-timeline.ts | head -20
'lead_webhook',降级到 'system'(并在 timeline payload 里加 originatedFrom: 'lead_webhook' 字段保留 audit context)。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
// 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
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
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
Step 3: PR creation via pr skill
# 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
git push -u origin <feature-branch-name>
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):
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
vs Plan 03 估计 LoC:Plan 04 简单一半 —— lead-processor 只 1 个 caller path,无 TaskDecision 6 action union loop,无 derive logic;Plan 03 是 ~1500-2000。
Summary
关键设计选择:
- DNC reject 写 timeline 不写 contact —— audit trail 不能漏,但不刷新 lastActivityAt 误标"DNC 用户活跃"
actor.type='system'(Policy Guard 词表)+ actor.type='lead_webhook'(timeline 词表)分离 —— Policy Guard 用 system 触发 SKIP_TYPE_CHECK_ACTORS 短路;timeline 保留 lead_webhook 原 audit 语义- 不再 caller 写 lifecycleStage / leadStatus / actionNeeded / taskType / 'pending' 字符串 —— 全删,Orchestrator 内部写 final values(过渡期 legacy shim 由 Plan 02 处理,Plan 04 caller 无感知)
- 3 个 frame uncertainties 记录到 Appendix A,落地 verify 后再选最终路径 —— 不假装这些细节已 settled
