> For AI agents: the complete documentation index is available at /llms.txt, the full documentation bundle is available at /llms-full.txt. # Plan 05 — message-processor STOP cascade 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/message-processor` 的 SMS STOP keyword DNC cascade 从自拼 SQL + shared `dnc-cascade.ts` helper,迁移到 `@retaintive/common@1.2.0/domain` 的 `ContactWriter.setDNC()` + `TaskOrchestrator.closeAllOpenForContact()`,同时把 task status filter 从 `'pending'` 改成 `IN ('pending','open')` 兼容 transitional 期间。 **Architecture:** message-processor 处理 inbound SMS 时,如果 body exact-match STOP keyword,触发两件事:(1) 把 `contacts.do_not_contact` 设成 true(sticky);(2) 关闭该 contact 所有 open task(`closeResult='do_not_contact'`,每个 closed task 写一条 `task.status_changed` timeline)。Phase 1 之前 caller 自己拼 SQL + 调 `lambda/shared/utils/dnc-cascade.ts`;Phase 1 改用 `@retaintive/common/domain` 共享 module — caller 只 call `setDNC()` + `closeAllOpenForContact()`,timeline 由 Orchestrator 内部 build,不再 caller-side 拼。`dnc-cascade.ts` helper Phase 1 仍存在(被 `closeAllOpenForContact` 内部调用),caller code 不再直接 import — Phase 2 等所有 caller migrate 完后删 helper。 **Tech Stack:** TypeScript 5 / Drizzle ORM / Vitest / `@retaintive/common` 1.2.0 / Neon PostgreSQL / AWS Lambda Node.js 20 **Spec source:** `docs/product-design/v2/unified-pipeline/implementation-plan/normative-spec.md` §4f(line 601-609) + §3.1 `closeAllOpenForContact()` signature(line 218-232) + §3.4 `setDNC()`(line 377-381) + §5.0 invariants + §5.2 integration test row(line 726) **Dependency:** Plan 02 merged + `@retaintive/common@1.2.0` published with `./domain` export(含 `applyTaskAction` / `closeAllOpenForContact` / `setDNC`) **Scope boundary(明确不做)**: - 不做 SMS meaningful 分类(spec Appendix B)—— message-processor 只处理 exact STOP,meaningful SMS 走 Phase 2 - 不删 `lambda/shared/utils/dnc-cascade.ts`(Phase 1 由 `closeAllOpenForContact` 内部 import;Phase 2 cutover 完成后删) - 不改 message UPSERT / contacts identity UPSERT / timeline `message.created` INSERT 三段主写 path(只改 STOP cascade 分支 + status filter sweep) - 不改 `isInboundStopMessage()` 检测逻辑(已 case-insensitive + word-boundary 正确) *** ## File Structure | File | Responsibility | Action | | ------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------- | | `lambda/message-processor/package.json` | bump `@retaintive/common` catalog 引用 → 1.2.0 | **Modify** | | `pnpm-workspace.yaml`(repo root catalog) | 如 catalog 未 bump,先 bump | **Modify (if needed)** | | `lambda/message-processor/src/infrastructure/neon-repository.ts` | (1) 删 `import { closeOpenTasksForDnc } from '../../../shared/utils/dnc-cascade'`; (2) 加 `import { setDNC, closeAllOpenForContact } from '@retaintive/common/domain'`; (3) STOP cascade 改用 module; (4) 删 inline `doNotContact` UPSERT 片段(由 `setDNC()` 接管) | **Modify** | | `lambda/message-processor/tests/unit/persist-message-batch.test.ts` | 加 6 个 STOP scenario test(覆盖 case map 1-6) | **Modify** | | `lambda/message-processor/tests/integration/stop-cascade.test.ts` | 新建 integration test 连真 Neon,verify Module 集成行为 | **Create** | 测试文件位置遵循现有结构: - Unit: `tests/unit/*.test.ts`(已有 13 个 unit test,加 case 到 `persist-message-batch.test.ts`) - Integration: `tests/integration/*.test.ts`(新建子目录;不存在的话先 mkdir) *** ## Task 1: Bump `@retaintive/common` dependency to 1.2.0 + verify domain export **Files:** - Modify: `lambda/message-processor/package.json`(if catalog ref;否则直接 bump 版本) - Modify(可能): repo root `pnpm-workspace.yaml`(catalog `@retaintive/common` entry) **Why this task first:** module import 要先有版本兼容才能 typecheck;否则 Task 2 改 code 时 import 报错。 ### Step 1: Inspect current dependency - [ ] **Step 1.1: Read `lambda/message-processor/package.json` 现状** 预期看到 `"@retaintive/common": "catalog:"` —— message-processor 用 catalog 引用,真版本在 repo root `pnpm-workspace.yaml`。 ```bash cat /Users/maxwsy/workspace/callytics-infrastructure/lambda/message-processor/package.json | grep retaintive cat /Users/maxwsy/workspace/callytics-infrastructure/pnpm-workspace.yaml | grep -A2 'retaintive/common' ``` 如果 catalog 已经是 1.2.0(Plan 02 完成后由 Plan 02 的 cleanup task bump),则无需改 package.json。 ### Step 2: Bump catalog 到 1.2.0(如未 bump) - [ ] **Step 2.1: Modify `pnpm-workspace.yaml` catalog entry** 找到 catalog 中 `@retaintive/common` 行,改成: ```yaml catalog: "@retaintive/common": ^1.2.0 # bumped from ^1.1.0 — Plan 05 needs ./domain export ``` - [ ] **Step 2.2: 跑 `pnpm install` 更新 lockfile** ```bash cd /Users/maxwsy/workspace/callytics-infrastructure && pnpm install ``` Expected: `pnpm-lock.yaml` 更新,`@retaintive/common@1.2.0` 已 resolved。 ### Step 3: Verify `./domain` subpath import works - [ ] **Step 3.1: 临时写一行 import 跑 typecheck** 在 `neon-repository.ts` 顶部临时加(下一 task 会正式加): ```typescript import { setDNC, closeAllOpenForContact } from '@retaintive/common/domain'; ``` 跑: ```bash cd /Users/maxwsy/workspace/callytics-infrastructure/lambda/message-processor && pnpm typecheck ``` Expected: PASS — 表明 `./domain` subpath export 在 `@retaintive/common@1.2.0` 中可见,types 也 resolve。如果 fail("Cannot find module '@retaintive/common/domain'")→ Plan 02 的 `package.json` `exports` 没 ship `./domain` path,回退 Plan 02 修正。 - [ ] **Step 3.2: 删除临时 import line**(Task 2 会正式加在正确位置 + 用法) ### Step 4: Commit ```bash git add pnpm-workspace.yaml pnpm-lock.yaml git commit -m "chore(message-processor): bump @retaintive/common catalog to 1.2.0 Phase 1 Plan 05 Task 1. Plan 05 needs setDNC + closeAllOpenForContact from @retaintive/common/domain (introduced in Plan 02). Verified import path resolves via typecheck. Spec: normative-spec.md §4f " ``` *** ## Task 2: Replace STOP cascade with `setDNC()` + `closeAllOpenForContact()` **Files:** - Modify: `lambda/message-processor/src/infrastructure/neon-repository.ts` - Test: `lambda/message-processor/tests/unit/persist-message-batch.test.ts`(加 case map 1-6) **Why:** 这是 Plan 05 的核心 — caller 不再自拼 DNC SQL + 不再直接 import shared `dnc-cascade.ts`。 ### Step 1: Map STOP cascade test scenarios(case map 表) | # | Scenario | Inbound message body | Pre-state(contact) | Pre-state(open tasks) | Expected after | Verify | | - | ------------------------------ | -------------------------------- | -------------------------- | ------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------ | | 1 | STOP + contact 有 2 open task | `'STOP'` | exists, doNotContact=false | 2 open task(`type_category` 任意) | (a) `contacts.do_not_contact=true` (b) 2 tasks `status='closed'`, `close_result='do_not_contact'`, `close_note='Auto-closed by DNC hard stop (SMS STOP keyword detected)'`, `close_type='auto_closed'` (c) 2 条 `task.status_changed` timeline rows (d) 1 条 `message.created` timeline row(已有) | `closeAllOpenForContact()` 返回 `closedTaskIds.length===2` + module 内 build 2 timeline statements + DNC set via `setDNC()` | | 2 | STOP + contact 不存在 | `'STOP'` | 不存在 | N/A | (a) `setDNC()` 是 UPDATE WHERE phone+storeId,no row matched → no-op (b) `closeAllOpenForContact()` 返回 `closedTaskIds=[]` + `statements=[]` (c) STOP cascade 整体 no-op,不报错 | logger 记录 closedCount=0 | | 3 | STOP + contact 已 DNC=true | `'STOP'` | exists, doNotContact=true | 0 open task(之前都关了) | (a) `setDNC()` idempotent — DNC 仍 true,`updated_at` 推进(行为可接受) (b) `closeAllOpenForContact()` SELECT 0 rows → 返回 `closedTaskIds=[]` (c) 不报错 | 不抛 exception;logger 记录 closedCount=0 | | 4 | STOP lowercase | `'stop'` | exists, dnc=false | 1 open task | 同 case 1(1 task)— `isInboundStopMessage` 已 normalize to lowercase 比较,小写 hit | `isStop===true`,走 cascade 路径 | | 5 | "just kidding STOP not really" | `'just kidding STOP not really'` | exists, dnc=false | 1 open task | (a) `isInboundStopMessage===false`(因为 exact-token match,不是 substring)(b) 不触发 cascade (c) DNC 保持 false,task 仍 open (d) message UPSERT + `message.created` timeline 正常写 | `isStop===false`,不调 `setDNC` / `closeAllOpenForContact` | | 6 | 普通 inbound SMS `'hello'` | `'hello'` | exists, dnc=false | 1 open task | 同 case 5 — 不触发 cascade,只 UPSERT message + contact.lastActivityAt forward + `message.created` timeline | `isStop===false` | **关键 invariant**(必 test): - I1:STOP cascade **不应阻塞** message+contact 主写 batch — caller 现状里 cascade 在 main batch commit 之后跑,且包 try/catch。Plan 05 保持这个边界:`setDNC` / `closeAllOpenForContact` 失败不应 fail 整个 entry。 - I2:`closeAllOpenForContact()` 返回的 `statements[]` 通过 `client.batch(statements)` 原子提交 —— DNC set 和 task close 必须**一起成功或一起失败**(spec §3.1 line 218 "1 个 UPDATE 关多 task + N 个 timeline INSERT")。 - I3:`closeAllOpenForContact()` 内部 status filter 是 `IN ('pending','open')`(由 Plan 02 实现),caller 不需要传。 ### Step 2: Sample test code(implementer 参考 style) Implementer 不需要写完整 6 test,以下 2 个 sample 表明期望的 test style: **Sample 1 — case 1(STOP + 2 open task → 全关)**: ```typescript // tests/unit/persist-message-batch.test.ts(在现有 describe 块内 append) import { setDNC, closeAllOpenForContact } from '@retaintive/common/domain'; vi.mock('@retaintive/common/domain', () => ({ setDNC: vi.fn(() => ({ __fakeSQL: 'setDNC' })), closeAllOpenForContact: vi.fn(async () => ({ status: 'allow' as const, statements: [{ __fakeSQL: 'close1' }, { __fakeSQL: 'close2' }, { __fakeSQL: 't1' }, { __fakeSQL: 't2' }], closedTaskIds: ['task-uuid-1', 'task-uuid-2'], })), })); it('STOP message closes all open tasks via closeAllOpenForContact', async () => { const repo = createNeonRepository('ssm-param'); const mockClient = createMockDrizzleClient(); // ... seed message entry with body='STOP', resolved customerPhone='+15551234567', storeId='STORE_A' // ... seed mockClient.batch() to capture statements await repo.persistMessageBatch([entry]); // Verify setDNC called once with correct args expect(setDNC).toHaveBeenCalledWith({ phone: '+15551234567', storeId: 'STORE_A', updatedBy: 'system', }); // Verify closeAllOpenForContact called once with correct args expect(closeAllOpenForContact).toHaveBeenCalledWith( { contactPhone: '+15551234567', storeId: 'STORE_A', closeResult: 'do_not_contact', closeNote: 'Auto-closed by DNC hard stop (SMS STOP keyword detected)', actor: { type: 'system', id: 'message-processor' }, }, { db: mockClient }, ); // Verify client.batch called with DNC stmt + 4 close-cascade statements (1 stmt setDNC + 4 from closeAllOpen) expect(mockClient.batch).toHaveBeenCalled(); const lastBatchCall = mockClient.batch.mock.calls.at(-1)![0]; expect(lastBatchCall).toHaveLength(5); }); ``` **Sample 2 — case 5(substring not match)**: ```typescript it('inbound SMS containing "STOP" as substring does NOT trigger DNC cascade', async () => { const repo = createNeonRepository('ssm-param'); const mockClient = createMockDrizzleClient(); const entry = makeEntry({ subject: 'just kidding STOP not really', direction: 'Inbound', type: 'SMS' }); await repo.persistMessageBatch([entry]); // Cascade module NOT called expect(setDNC).not.toHaveBeenCalled(); expect(closeAllOpenForContact).not.toHaveBeenCalled(); // Main batch (message + contact + timeline) still ran expect(mockClient.batch).toHaveBeenCalledTimes(1); }); ``` 剩 4 case implementer 自己写。 ### Step 3: Implementation — modify `neon-repository.ts` - [ ] **Step 3.1: Update imports(top of file,\~line 12-33)** Before: ```typescript import { createDrizzleClient, messages, contacts, contactTimeline, sql, buildTimelineValues, buildIdentityFields, lastActivityAtForward, NAME_TRUST, } from '@retaintive/common/db'; import type { DrizzleClient, MessageParty, MessageAttachment } from '@retaintive/common/db'; import { resolveContactIdentity } from '@retaintive/common/phone-identity'; import { closeOpenTasksForDnc } from '../../../shared/utils/dnc-cascade'; import { createLogger } from '../../../shared/utils/logger'; ``` After: ```typescript import { createDrizzleClient, messages, contacts, contactTimeline, sql, buildTimelineValues, buildIdentityFields, lastActivityAtForward, NAME_TRUST, } from '@retaintive/common/db'; import type { DrizzleClient, MessageParty, MessageAttachment } from '@retaintive/common/db'; import { resolveContactIdentity } from '@retaintive/common/phone-identity'; import { setDNC, closeAllOpenForContact } from '@retaintive/common/domain'; // REMOVED Plan 05: `closeOpenTasksForDnc` 不再直接 import — Phase 1 由 `closeAllOpenForContact` // 内部调用 dnc-cascade.ts helper(Plan 02 内部 wire 好)。Phase 2 删 helper file。 import { createLogger } from '../../../shared/utils/logger'; ``` - [ ] **Step 3.2: 删除 inline `doNotContact` set in contacts UPSERT(\~line 313-314 + 327-335)** `setDNC()` 现在专门负责 DNC 写入,contacts UPSERT 不再 inline 处理 DNC。 Before(contacts INSERT values,\~line 297-315): ```typescript statements.push( client .insert(contacts) .values({ ...buildIdentityFields({ phone: customerPhone, franchiseId: entry.franchiseId, accountId: entry.accountId, storeId: resolvedStoreId, }), lastActivityAt: eventTime, ...(formattedName != null && { firstName: formattedName, firstNameTrustScore: trustScore, firstNameUpdatedAt: eventTime, }), /* SMS STOP (TCPA opt-out): mark DNC on the new contact row. */ ...(isStop && { doNotContact: true, doNotContactUpdatedBy: 'system' }), }) .onConflictDoUpdate({ target: [contacts.phone, contacts.storeId], set: { lastActivityAt: lastActivityAtForward(eventTime), updatedAt: sql`NOW()`, ...(formattedName != null && { firstName: sql`CASE WHEN ${trustScore} >= COALESCE(${contacts.firstNameTrustScore}, 0) THEN ${formattedName} ELSE ${contacts.firstName} END`, firstNameTrustScore: sql`CASE WHEN ${trustScore} >= COALESCE(${contacts.firstNameTrustScore}, 0) THEN ${trustScore} ELSE ${contacts.firstNameTrustScore} END`, firstNameUpdatedAt: sql`CASE WHEN ${trustScore} >= COALESCE(${contacts.firstNameTrustScore}, 0) THEN ${eventTime} ELSE ${contacts.firstNameUpdatedAt} END`, }), /* * SMS STOP (TCPA opt-out): set DNC. doNotContact is set true; the * updatedBy attribution only stamps 'system' on a false→true transition * so a prior 'staff' marker is preserved (mirrors contacts-analyzer's * sticky DNC write). */ ...(isStop && { doNotContact: true, doNotContactUpdatedBy: sql`CASE WHEN ${contacts.doNotContact} = true THEN ${contacts.doNotContactUpdatedBy} ELSE 'system' END`, }), }, }), ); ``` After(只删去 `isStop` 两个 inline spread,**保留** contacts identity / lastActivityAt / name 部分不变): ```typescript statements.push( client .insert(contacts) .values({ ...buildIdentityFields({ phone: customerPhone, franchiseId: entry.franchiseId, accountId: entry.accountId, storeId: resolvedStoreId, }), lastActivityAt: eventTime, ...(formattedName != null && { firstName: formattedName, firstNameTrustScore: trustScore, firstNameUpdatedAt: eventTime, }), // Plan 05: 删去 isStop spread — DNC 写入由 setDNC() 接管(Task 2 Step 3.3) }) .onConflictDoUpdate({ target: [contacts.phone, contacts.storeId], set: { lastActivityAt: lastActivityAtForward(eventTime), updatedAt: sql`NOW()`, ...(formattedName != null && { firstName: sql`CASE WHEN ${trustScore} >= COALESCE(${contacts.firstNameTrustScore}, 0) THEN ${formattedName} ELSE ${contacts.firstName} END`, firstNameTrustScore: sql`CASE WHEN ${trustScore} >= COALESCE(${contacts.firstNameTrustScore}, 0) THEN ${trustScore} ELSE ${contacts.firstNameTrustScore} END`, firstNameUpdatedAt: sql`CASE WHEN ${trustScore} >= COALESCE(${contacts.firstNameTrustScore}, 0) THEN ${eventTime} ELSE ${contacts.firstNameUpdatedAt} END`, }), // Plan 05: 删去 isStop spread — DNC 写入由 setDNC() 接管 }, }), ); ``` - [ ] **Step 3.3: Replace STOP cascade block(\~line 384-416)** Before: ```typescript /* * SMS STOP cascade: after the message+DNC batch commits, close every open task * for this contact (shared lambda/shared/utils/dnc-cascade — same writes as the * AI-detected DNC path in contacts-analyzer). Deliberately AFTER the message * batch and in its own try: the opt-out (doNotContact=true) is already durably * recorded; a task-close failure must not fail the message write. Idempotent — * a repeat STOP finds no pending tasks. Only when contact identity fully resolved. */ if (isStop && customerPhone && storeId != null && entry.franchiseId && entry.accountId) { try { const dnc = await closeOpenTasksForDnc(client, { phone: customerPhone, storeId, franchiseId: entry.franchiseId, accountId: entry.accountId, actor: { actorType: 'system', actorSubjectId: ACTOR_SUBJECT_ID, actorSourceType: ACTOR_SOURCE_TYPE, actorSourceSystem: ACTOR_SOURCE_SYSTEM, closeInitiator: 'system', }, idempotencyKeyFor: (taskId) => `task.status_changed:dnc_stop:${String(msg.id)}:${taskId}`, }); logger.info('SMS STOP — set DNC and ran task-close cascade', { customerPhone, storeId, closedCount: dnc.closedCount, taskIds: dnc.closedTaskIds, messageId: String(msg.id), }); } catch (err) { logger.error('SMS STOP — DNC flag set but task-close cascade failed', { customerPhone, storeId, messageId: String(msg.id), error: err, }); } } ``` After: ```typescript /* * Plan 05: SMS STOP cascade now uses @retaintive/common/domain shared modules. * * Two SQL units run in ONE atomic batch (spec §3.1 line 218 — DNC set + task close * must succeed or fail together): * ① setDNC() — 1 UPDATE contacts SET do_not_contact=true * ② closeAllOpenForContact() — 1 UPDATE tasks (bulk close) + N timeline INSERT * * Module signatures: * - setDNC: sticky, only writes true (spec §3.4 line 377-381) * - closeAllOpenForContact: SELECT open tasks INSIDE module, returns SQL[] * + closedTaskIds; internally uses status IN ('pending','open') filter * (transitional — spec §6.2 line 785-790). Timeline rows are built INSIDE * the module — caller does NOT prepend its own timeline statements. * * Try/catch boundary unchanged: cascade failure logs error but does NOT fail * the main message persist batch (already committed above). Idempotent — a repeat * STOP on an already-DNC contact finds no open tasks, closeAllOpenForContact() * returns closedTaskIds=[], no-op. */ if (isStop && customerPhone && storeId != null && entry.franchiseId && entry.accountId) { try { const dncStmt = setDNC({ phone: customerPhone, storeId, updatedBy: 'system', }); const closeResult = await closeAllOpenForContact( { contactPhone: customerPhone, storeId, closeResult: 'do_not_contact', closeNote: 'Auto-closed by DNC hard stop (SMS STOP keyword detected)', actor: { type: 'system', id: ACTOR_SUBJECT_ID }, }, { db: client }, ); if (closeResult.status === 'reject') { // Module-level reject is unexpected for this caller (no DNC contact / hallucinated taskId // possible here — caller only triggers on inbound SMS STOP). Log and continue. logger.error('SMS STOP — closeAllOpenForContact returned reject', { customerPhone, storeId, reason: closeResult.reason, details: closeResult.details, messageId: String(msg.id), }); } else { // Atomic batch: DNC set + bulk close + N timeline. Drizzle batch supports 1+ stmts. // eslint-disable-next-line @typescript-eslint/no-explicit-any -- same Drizzle batch tuple workaround as main batch above await client.batch([dncStmt, ...closeResult.statements] as any); logger.info('SMS STOP — DNC set + task-close cascade committed', { customerPhone, storeId, closedCount: closeResult.closedTaskIds.length, taskIds: closeResult.closedTaskIds, messageId: String(msg.id), }); } } catch (err) { logger.error('SMS STOP — DNC flag set but task-close cascade failed', { customerPhone, storeId, messageId: String(msg.id), error: err, }); } } ``` **关键对齐点**: - `setDNC()` 返回单一 SQL,放进 `client.batch([dncStmt, ...closeResult.statements])` 一起原子提交 - `closeAllOpenForContact()` 已经把 timeline INSERT 包在 `statements[]` 里(spec §3.1 line 229 注释)—— caller 不需要再 build timeline - `closeResult` 的 status 应该总是 `'allow'`(message-processor 不传 confidence / aiRunStartedAt / taskId,不会触发 reject reasons),但为防御性写 reject branch + log - `ACTOR_SUBJECT_ID = 'message_processor'`(已 const 在 \~line 40)用作 `actor.id` ### Step 4: Run tests + commit - [ ] **Step 4.1: 加 6 个 case 到 `persist-message-batch.test.ts` 并跑** ```bash cd /Users/maxwsy/workspace/callytics-infrastructure/lambda/message-processor && \ pnpm test tests/unit/persist-message-batch.test.ts ``` Expected: 6 个新 case 全 PASS + 现有 test 不退化。 - [ ] **Step 4.2: 跑 typecheck** ```bash cd /Users/maxwsy/workspace/callytics-infrastructure/lambda/message-processor && pnpm typecheck ``` Expected: no errors。 - [ ] **Step 4.3: Commit** ```bash git add lambda/message-processor/src/infrastructure/neon-repository.ts \ lambda/message-processor/tests/unit/persist-message-batch.test.ts git commit -m "feat(message-processor): STOP cascade — use shared setDNC + closeAllOpenForContact Phase 1 Plan 05 Task 2. Before: - inline doNotContact in contacts UPSERT - direct import closeOpenTasksForDnc from lambda/shared/utils/dnc-cascade.ts - caller-side actor / idempotencyKey wiring After: - setDNC({phone, storeId, updatedBy:'system'}) — sticky DNC SQL - closeAllOpenForContact({contactPhone, storeId, closeResult:'do_not_contact', closeNote:'Auto-closed by DNC hard stop (SMS STOP keyword detected)', actor:{type:'system', id:'message_processor'}}, {db: client}) - atomic batch [dncStmt, ...closeResult.statements] commits DNC + bulk close + N timeline rows together - shared dnc-cascade.ts helper Phase 1 仍存在(closeAllOpenForContact 内部调用); Phase 2 删 helper 6 STOP scenarios covered via unit test: 1. STOP + 2 open tasks → 2 closed + 2 timeline + DNC set 2. STOP + contact missing → no-op (setDNC UPDATE 0 rows) 3. STOP + DNC already true → idempotent no-op 4. lowercase 'stop' → still hits (case-insensitive) 5. substring 'kidding STOP not really' → NOT triggered (exact-token match) 6. 'hello' → no cascade, only message UPSERT + message.created timeline Spec: normative-spec.md §4f + §3.1 + §3.4 " ``` *** ## Task 3: Sweep `status='pending'` read path → `status IN ('pending','open')` **Files:** - Modify(任意):`lambda/message-processor/src/infrastructure/neon-repository.ts`(grep "pending" 看是否还有遗留) - Modify(其他源文件):`lambda/message-processor/src/core/*.ts` / `src/handler*.ts`(grep 全 message-processor source) **Why:** Plan 03/04 caller 可能同时在写 'pending'(transitional)和 'open'(post-Plan 02);Phase 1 transitional window 期间所有读路径必须用 `IN ('pending','open')` 兼容两种值(spec §6.2 line 785-790)。message-processor 自己不直接 SELECT tasks(STOP cascade 走 `closeAllOpenForContact`,module 内部已 sweep),但如果 source 里有任何 `status='pending'` SELECT/WHERE 需要 sweep。 ### Step 1: Map test scenarios 无 test scenario — 这是 search-and-replace task。verify 通过 grep + typecheck。 但 Task 2 改完后,**重新 review 一次** `neon-repository.ts` 看是否还有 `'pending'` 字符串残留 — Task 2 已经把 STOP cascade 主要的 `status='pending'` 引用删了(它在 `closeOpenTasksForDnc` 内部,不在 caller code)。 ### Step 2: Implementation — grep + sweep - [ ] **Step 2.1: 全 message-processor source grep `'pending'` / `"pending"`** ```bash cd /Users/maxwsy/workspace/callytics-infrastructure/lambda/message-processor && \ grep -rn --include='*.ts' "['\"]pending['\"]" src/ ``` Expected outcome: - **0 hit in `src/`** —— message-processor 现状里 `'pending'` 都通过 `closeOpenTasksForDnc` import 进来,Task 2 已删 import。caller code 没直接 SELECT tasks。 - 如果 grep 有 hit,说明有遗漏的 `tasks.status` 引用,逐个改成 `sql\`$\{tasks.status} IN ('pending', 'open')\``(Drizzle) 或 raw SQL `status IN ('pending','open')\`。 - [ ] **Step 2.2: grep test files 是否在 mock task status 用 `'pending'`** ```bash cd /Users/maxwsy/workspace/callytics-infrastructure/lambda/message-processor && \ grep -rn --include='*.ts' "['\"]pending['\"]" tests/ ``` Expected: - Task 2 加的新 STOP cascade test fixtures 用 `status: 'open'`(新写入)或 `status: 'pending'`(测试 transitional)都可接受 — 关键是 module 内部 SELECT 用 `IN`,test 不依赖具体 status 值。 - 如果 test 显式 assert `WHERE status='pending'` SQL string,改成 assert `IN ('pending','open')`。 ### Step 3: Commit(如有改动) 如果 Step 2.1 grep 有 hit 并改动,commit: ```bash git add lambda/message-processor/src/ git commit -m "chore(message-processor): sweep status='pending' read path → IN ('pending','open') Phase 1 Plan 05 Task 3. Transitional window 期间(Plan 02 → final cutover)两种 status 值共存。 所有读路径用 IN clause 兼容(spec §6.2 line 785-790)。 message-processor 现状 caller code 不直接 SELECT tasks(STOP cascade 走 closeAllOpenForContact module,内部已 sweep)。本 task 主要是 verify grep 无遗漏。 Spec: normative-spec.md §6.2 " ``` 如果 grep 0 hit,跳过 commit,记录 task 完成。 *** ## Task 4: Integration test + PR **Files:** - Create: `lambda/message-processor/tests/integration/stop-cascade.test.ts` - Modify(可能):`lambda/message-processor/package.json`(加 `test:integration` script if 不存在) **Why:** Unit test 用 mock `@retaintive/common/domain`;integration test 跑真 Neon verify module 集成行为 — DNC set 真生效 + open task 真被关 + timeline 真写入。spec §5.2(line 726)明确列了 message-processor STOP 作为 integration test scenario。 ### Step 1: Map integration test scenarios 跑真实 Neon test branch(用 `NEON_TEST_DATABASE_URL` env var),scenarios: | # | Scenario | Setup(direct SQL seed) | Action | Expected DB state | | -------- | ------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | INT1 | STOP + 2 open task → 全关 + DNC | INSERT contact (phone='+15550001', store\_id='STORE\_X', dnc=false); INSERT 2 tasks (status='open', type\_category='lead\_follow\_up' / 'cancellation\_risk') | call `repo.persistMessageBatch([entry])` with body='STOP' | SELECT contact → dnc=true; SELECT tasks WHERE phone+store → 2 rows status='closed', close\_result='do\_not\_contact', close\_note='Auto-closed by DNC hard stop (SMS STOP keyword detected)', close\_type='auto\_closed'; SELECT contact\_timeline → 2 rows event\_type='task.status\_changed' + 1 row event\_type='message.created' | | INT2 | STOP + 0 open task → DNC set,no task closed | INSERT contact (dnc=false); 不 INSERT tasks | call persist with 'STOP' | contact dnc=true; tasks 0 rows; timeline: 0 task.status\_changed + 1 message.created | | INT3 | 同一 contact 第二次 STOP(idempotent) | INT1 已跑过 → state 是 dnc=true + 2 closed tasks | 再 call persist with 'STOP'(new messageId) | state 不变 — dnc 仍 true,2 tasks 仍 closed,**没新增** task.status\_changed timeline(因为 closeAllOpenForContact SELECT 0 open tasks),但**有新** message.created(messageId 不同),`do_not_contact_updated_by` 仍为 INT1 的初始值 | | INT4 | 'hello' message(control) | INSERT contact (dnc=false); 1 open task | call persist with 'hello' | contact dnc=false; task 仍 open; timeline 0 task.status\_changed + 1 message.created | | **INT5** | **Sticky DNC audit trail — staff 先 set,system 再 STOP** | INSERT contact (dnc=true, **do\_not\_contact\_updated\_by='staff'**); 0 open tasks | call persist with 'STOP'(messageId='msg5') | contact dnc 仍 true; **`do_not_contact_updated_by` 仍 'staff' 不被 'system' 覆盖**;timeline +1 message.created | **关键 invariant**: - I-INT-1:**Atomic batch wiring** — caller test 不模拟 Postgres/Drizzle rollback。Verify `setDNC()` 返回的 SQL 和 `closeAllOpenForContact()` 返回的所有 statements 进入同一次 `client.batch([dncStmt, ...closeResult.statements])`;`client.batch.mock.calls` 中不允许出现 DNC 和 close 分属两次 batch。Postgres rollback 行为由 common/module integration test 覆盖。 - I-INT-2:**Status filter 用 IN** — seed 用 `status='pending'`(旧值),expect 仍被 close(verify `closeAllOpenForContact` 用 `IN ('pending','open')` 而不是 `='open'`)。 - **I-INT-3:TCPA sticky audit trail** —— INT5 守这个 invariant。即使 Plan 02 §3.4 module 内部已实现 sticky CASE WHEN,caller 层 integration test 必须 cover real DB roundtrip — `do_not_contact_updated_by` 一旦被 'staff' stamped 永远不被 'ai'/'system' 覆盖。这是 TCPA 合规 audit trail 要求(see [stickyDncFragments](../../../callytics-common/src/db/helpers/contact-writer.ts)、message-processor:332-335 verbatim pattern source)。漏这条测试 = silent regression risk:Plan 02 module 改动 / DB CHECK 调整 / migration 任一处出错都直接 silent 破合规。 ### Step 2: Sample integration test code ```typescript // tests/integration/stop-cascade.test.ts import { describe, it, expect, beforeAll, afterEach } from 'vitest'; import { createNeonRepository } from '../../src/infrastructure/neon-repository'; import { createDrizzleClient, tasks, contacts, contactTimeline } from '@retaintive/common/db'; import { and, eq } from 'drizzle-orm'; const NEON_URL = process.env['NEON_TEST_DATABASE_URL']; const SSM_PARAM = process.env['NEON_DATABASE_URL_SSM_PARAM'] ?? '/test/neon/url'; const STORE_X = 'STORE_X_PLAN05_TEST'; const CONTACT_PHONE = '+15555550001'; describe.skipIf(!NEON_URL)('STOP cascade integration', () => { let db: ReturnType; beforeAll(() => { db = createDrizzleClient({ databaseUrl: NEON_URL! }); }); afterEach(async () => { // cleanup test rows await db.delete(contactTimeline).where(eq(contactTimeline.storeId, STORE_X)); await db.delete(tasks).where(eq(tasks.storeId, STORE_X)); await db.delete(contacts).where(and(eq(contacts.phone, CONTACT_PHONE), eq(contacts.storeId, STORE_X))); }); it('INT1: STOP closes 2 open tasks + sets DNC + writes 2 timeline rows', async () => { // seed (omitted for brevity — INSERT contact + 2 tasks) // ... const repo = createNeonRepository(SSM_PARAM); const entry = makeEntry({ subject: 'STOP', direction: 'Inbound', type: 'SMS', customerPhone: CONTACT_PHONE }); await repo.persistMessageBatch([entry]); // Assert contact.do_not_contact = true const [c] = await db.select().from(contacts).where(and(eq(contacts.phone, CONTACT_PHONE), eq(contacts.storeId, STORE_X))); expect(c.doNotContact).toBe(true); // Assert 2 tasks closed const closedTasks = await db.select().from(tasks).where(and(eq(tasks.contactPhone, CONTACT_PHONE), eq(tasks.storeId, STORE_X))); expect(closedTasks).toHaveLength(2); closedTasks.forEach(t => { expect(t.status).toBe('closed'); expect(t.closeResult).toBe('do_not_contact'); expect(t.closeNote).toBe('Auto-closed by DNC hard stop (SMS STOP keyword detected)'); expect(t.closeType).toBe('auto_closed'); }); // Assert timeline: 2 task.status_changed + 1 message.created const timeline = await db.select().from(contactTimeline).where(eq(contactTimeline.storeId, STORE_X)); expect(timeline.filter(t => t.eventType === 'task.status_changed')).toHaveLength(2); expect(timeline.filter(t => t.eventType === 'message.created')).toHaveLength(1); }); // INT2 / INT3 / INT4 implementer 自己按 case map 写 }); ``` 剩 3 个 case + 2 个 invariant test implementer 写。 ### Step 3: Run integration test(localhost,可选 — CI skip if env not set) - [ ] **Step 3.1: Set env var + run** ```bash export NEON_TEST_DATABASE_URL='postgres://...test-branch...' cd /Users/maxwsy/workspace/callytics-infrastructure/lambda/message-processor && \ pnpm test tests/integration/stop-cascade.test.ts ``` Expected: 4 case + 2 invariant PASS。 CI 跑这个 test 需要在 GH Actions 配 secret + Neon test branch — Phase 1 spec 没强制 CI 跑 integration,本地手跑 verify 即可。 ### Step 4: Commit + verification + PR - [ ] **Step 4.1: Commit integration test** ```bash git add lambda/message-processor/tests/integration/stop-cascade.test.ts git commit -m "test(message-processor): STOP cascade integration test on real Neon Phase 1 Plan 05 Task 4. Covers 4 scenarios + 2 invariants per spec §5.2 line 726: - INT1: STOP + 2 open tasks → all closed + DNC + timeline - INT2: STOP + 0 open tasks → DNC only - INT3: idempotent (repeat STOP no-op) - INT4: 'hello' control — no cascade - I-INT-1: atomic batch wiring — DNC stmt + close statements in one client.batch call - I-INT-2: status='pending' seed still closed (IN clause verified) Skipped when NEON_TEST_DATABASE_URL env not set (local-only by default). Spec: normative-spec.md §5.2 + §5.0 invariants " ``` - [ ] **Step 4.2: Verify full test suite + typecheck pass** ```bash cd /Users/maxwsy/workspace/callytics-infrastructure/lambda/message-processor && \ pnpm test && pnpm typecheck ``` Expected: 全 PASS。 - [ ] **Step 4.3: Push + Open PR** ```bash git push -u origin gh pr create --title "feat(message-processor): Phase 1 Plan 05 — STOP cascade via shared modules" \ --body-file <(cat <<'EOF' ## Plan 05: message-processor STOP cascade migration ### 背景 unified-pipeline Phase 1 把 task / contact mutation 收口到 `@retaintive/common/domain` 共享 module。message-processor 处理 SMS STOP keyword 触发的 DNC cascade 是其中一个 caller。 ### 改了什么 1. **STOP cascade 迁移到共享 module** - 删 `import { closeOpenTasksForDnc } from '../../../shared/utils/dnc-cascade'` - 加 `import { setDNC, closeAllOpenForContact } from '@retaintive/common/domain'` - DNC set 由 `setDNC()` 接管(原 inline contacts UPSERT 里的 `doNotContact: true` spread 删除) - 关闭 open task 由 `closeAllOpenForContact()` 接管,timeline 由 module 内部 build 2. **原子性**:DNC SQL + 关 task SQL + N timeline INSERT 通过 `client.batch([dncStmt, ...closeResult.statements])` 一起提交(spec §3.1 line 218 B7 helper 要求) 3. **status filter sweep**:caller code 现状不直接 SELECT tasks,`closeAllOpenForContact` 内部已用 `IN ('pending','open')`(transitional spec §6.2) 4. **依赖 bump**:`@retaintive/common` catalog 1.1.0 → 1.2.0 ### 改前 vs 改后 | 维度 | 改前 | 改后 | |---|---|---| | DNC 写入 | inline contacts UPSERT 的 `isStop && {doNotContact:true}` spread | `setDNC({phone, storeId, updatedBy:'system'})` 单一 SQL 接管 | | 关 task | 调 `closeOpenTasksForDnc()` shared helper(直接 SELECT + 拼 SQL + batch) | 调 `closeAllOpenForContact()` module,返回 `{statements[], closedTaskIds}` | | Timeline | helper 内部拼 `task.status_changed` event | module 内部 build,caller 不再 import buildTimelineValues for cascade | | Atomic 边界 | DNC 在主 batch + close 在 cascade batch(两批不原子)| DNC + close + timeline 在**同一** cascade batch(原子)| | Caller LoC | ~35 行(import + cascade block + inline DNC) | ~30 行(import + cascade block,无 inline) | | Shared helper 依赖 | 直接 import `dnc-cascade.ts` | 不再直接 import(`closeAllOpenForContact` 内部 wrap helper,Phase 2 删 helper)| ### 使用方式 caller 入口不变 — SQS handler 仍 call `repo.persistMessageBatch(entries)`,STOP 检测 + cascade 内部自动跑。 ### 改动文件 - `pnpm-workspace.yaml`(catalog bump) - `lambda/message-processor/src/infrastructure/neon-repository.ts`(imports + DNC UPSERT 删 + cascade 重写) - `lambda/message-processor/tests/unit/persist-message-batch.test.ts`(加 6 case) - `lambda/message-processor/tests/integration/stop-cascade.test.ts`(新建) ### 部署 CDK stack 不变 — Lambda runtime / IAM / SQS binding 全无关。`pnpm install` 拉新 `@retaintive/common@1.2.0`,bundle 后部署。 ### 关联 - normative-spec.md §4f / §3.1 / §3.4 / §5.0 / §5.2 - Plan 02 PR(`@retaintive/common@1.2.0` 含 `./domain` export)— 必须先 merge - Phase 2 跟进:删 `lambda/shared/utils/dnc-cascade.ts`(所有 caller 都 migrate 完之后) ### Test plan - [x] `pnpm test tests/unit/persist-message-batch.test.ts` — 6 个新 case PASS - [x] `pnpm typecheck` — 无 type 错 - [x] `pnpm test tests/unit/stop-keyword.test.ts` — 已有 isInboundStopMessage test 不退化 - [ ] **本地跑 integration**:`NEON_TEST_DATABASE_URL=... pnpm test tests/integration/stop-cascade.test.ts` — 4 case + 2 invariant - [ ] **Test env smoke**:部署到测试环境后发一条真 inbound 'STOP' SMS,verify 测试 Neon 里 DNC set + open task 关闭 + timeline 2 条 task.status_changed + 1 条 message.created EOF ) ``` *** ## Self-Review Checklist ### Spec coverage - ✅ §4f line 605 "DNC cascade SQL" → Task 2 Step 3.3 `setDNC` + `closeAllOpenForContact` 接管 - ✅ §4f line 608 "status filter pending→open" → Task 3 sweep(message-processor caller code 不直接 SELECT,module 内部已用 `IN`) - ✅ §4f line 609 "现有 dnc-cascade.ts shared helper 仍存在,caller 不直接 import" → Task 2 Step 3.1 删 import,helper file 不动 - ✅ §3.1 line 218-232 `closeAllOpenForContact` signature(`contactPhone` / `storeId` / `closeResult` / `closeNote` / `actor`,返回 `{status, statements, closedTaskIds}`)→ Task 2 Step 3.3 完整调用 - ✅ §3.1 line 229 "1 个 UPDATE 关多 task + N 个 timeline INSERT" → Task 2 Step 3.3 atomic batch + Task 4 INT1 timeline 2 行 assertion - ✅ §3.4 line 377-381 `setDNC` sticky 行为 → Task 2 Step 3.3 调用 + case 3 idempotent verify - ✅ §5.0 invariant "DNC hard stop" → Task 4 INT1 + INT3 - ✅ §5.2 line 726 "message-processor STOP" integration test → Task 4 整体 - ✅ §6.2 transitional `IN ('pending','open')` → Task 3 sweep(caller 0 hit + module 内部已实现) - ✅ Appendix B "不在 Phase 1 scope:SMS meaningful 分类" → Plan 文头 scope boundary 明确不做 ### Placeholder scan - ✅ 6 case map + 4 integration scenario + 2 invariant 全列详细 input/expected - ✅ Implementation 关键 SQL pattern(`setDNC` 单语句 / `closeAllOpenForContact` 返回 statements / atomic batch)代码完整 - ✅ Test sample 2 个完整 code block(case 1 + case 5)+ INT1 完整代码示例 - ✅ 每 task 列具体 file path + line range(neon-repository.ts \~line 12-33 imports / \~line 297-315 UPSERT / \~line 384-416 cascade block) - ✅ Commit messages 全具体,引用 spec section ### Type consistency - ✅ `setDNC()` signature 与 Plan 02 Task 4 定义一致:`{phone, storeId, updatedBy: 'staff' | 'ai' | 'system'}` → caller 传 `updatedBy: 'system'` 满足 union - ✅ `closeAllOpenForContact()` signature 与 Plan 02 Task 3 定义一致: - params: `{contactPhone, storeId, closeResult: TaskCloseResult, closeNote, actor: BaseActionContext['actor']}` → caller 全部传齐 - ctx: `{db: DrizzleClient}` → caller 传 `{db: client}` - 返回: `{status:'allow', statements: SQL[], closedTaskIds: string[]}` 或 `{status:'reject', reason, details}` → caller branch 分别处理 - ✅ `actor.type` 用 `'system'`(在 Plan 02 BaseActionContext.actor union `'staff' | 'system' | 'ai_agent' | 'contact_analysis'` 内) - ✅ `actor.id` 传 `ACTOR_SUBJECT_ID = 'message_processor'`(已 const 定义)— `BaseActionContext.actor.id?: string` 接受 string - ✅ `closeResult: 'do_not_contact'` —— Plan 02 normative-spec §2.2 `TASK_CLOSE_RESULT` 15 values 中包含 `do_not_contact`(原 18 values 中保留,见 schema) - ✅ `closeNote` 字符串 `'Auto-closed by DNC hard stop (SMS STOP keyword detected)'` 与原 `dnc-cascade.ts` 的 `DNC_CLOSE_NOTE` 略不同(原是 `'Auto-closed by DNC hard stop (code-derived cascade).'`)— 新 note 文案更明确指出 SMS STOP 来源,prompt 里也明确要求这个文案 - ✅ Plan 02 Task 5 `TIMELINE_EVENT_TYPES` 含 `'task.status_changed'`(line 1271)→ closeAllOpenForContact 内部 build 这个 event 时 Zod schema 验证 pass ### Outstanding 无。Plan 05 完成 = message-processor STOP cascade 走 `@retaintive/common/domain` 共享 module,caller code 不再直接 import shared `dnc-cascade.ts`,DNC + 关 task + timeline 在同一 atomic batch 提交。 *** ## LoC estimate | 改动 | LoC | | ------------------------------------------------ | ----------------- | | `package.json` / `pnpm-workspace.yaml` 改 catalog | \~2 | | `neon-repository.ts` imports 删 1 + 加 1 | \~3 | | `neon-repository.ts` 删 inline DNC UPSERT 片段(2 处) | \~-8 | | `neon-repository.ts` 重写 STOP cascade block | \~+50 | | 6 unit test cases | \~250 | | 4 integration test scenarios + 2 invariants | \~350 | | Self-doc comments(`Plan 05` 标注 / inline 解释) | \~30 | | **合计** | **\~700-900 LoC** | *** ## Execution Handoff **Plan 05 完成 = message-processor STOP cascade migration 通过测试 Neon verify + PR merged。** **Option 1: Subagent-Driven**(推荐)— 每 Task 一 fresh subagent,review 之间 user 看 unit test 覆盖度 + integration result。Task 2 是 critical task(改 production 写入 path),建议 user 在 Task 2 完成后亲自跑 INT1 verify atomicity。 **Option 2: Inline** — `superpowers:executing-plans` 跑全 4 task,checkpoint 在 Task 2 + Task 4 之前停。 依赖明确:Plan 02 PR 必须先 merge + `@retaintive/common@1.2.0` 已 publish。如 Plan 02 还在 review,本 Plan 05 可以并行**写代码**(local branch + 暂用 link 临时 package)但**不能 push PR** —— 等 catalog 真 bump 到 1.2.0 才 push。