> For AI agents: the complete documentation index is available at /llms.txt, the full documentation bundle is available at /llms-full.txt. # Plan 03 — contacts-analyzer Caller Migration 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:** Swap the inline Drizzle SQL inside `lambda/contacts-analyzer/src/infrastructure/neon-repository.ts::writeAnalysisWithTasks()` for one-line calls into `@retaintive/common/domain` modules (`applyTaskAction()` / `closeAllOpenForContact()` / `upsertIdentity()` / `writeTimelineEvent()`). After this plan ships: - 6-action Task Orchestrator is the only path that mutates `tasks` rows from contacts-analyzer. - TaskDecision Zod schema speaks the new 6-action vocabulary (with a `.transform()` shim accepting the legacy `create` literal until the prompt engineer cuts over). - Caller stops emitting deprecated fields (`actionNeeded` / `task_type` derive, `priority` at task level). - All inline `buildTimelineValues()` call sites disappear — Orchestrator + Timeline Writer own that. - DNC bulk close runs through `closeAllOpenForContact()`; the local `dnc-cascade.ts` helper import disappears. **Architecture:** caller-side refactor only. The `writeAnalysisWithTasks()` orchestration shape stays — contacts UPDATE → taskDecisions loop → DNC cascade → lifecycle / lead-status timeline → `contact_analysis.completed` timeline → `client.batch(statements)`. What changes is _who builds the statements_: Orchestrator helpers replace inline `client.update(tasks).set({...})` / `client.insert(contactTimeline).values(...)`. Atomicity (one `db.batch()` per contact) is preserved. **Tech Stack:** TypeScript 5 / Drizzle ORM / Zod 4 / `@retaintive/common@1.2.0/domain` / bun test / vitest integration / PostgreSQL 16 (Neon test env) **Spec source:** `docs/product-design/v2/unified-pipeline/implementation-plan/normative-spec.md` §4d (caller change table) + §2 (transitional 3-value status enum) + §5 (test plan §5.0 invariants + §5.2 contacts-analyzer integration test row) **Dependency:** Plan 02 merged → `@retaintive/common@1.2.0` published with `./domain` export path. Plan 03 cannot start until `applyTaskAction()` / `closeAllOpenForContact()` / `upsertIdentity()` / `writeTimelineEvent()` are importable from `@retaintive/common/domain`. *** ## File Structure | File | Responsibility | Action | | ------------------------------------------------------------------------------------------------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------- | | `lambda/contacts-analyzer/package.json` | Bump `@retaintive/common` catalog reference 1.0.x → 1.2.0 (or workspace root `package.json` catalog if pnpm catalog) | Modify | | `lambda/contacts-analyzer/src/core/models.ts` (lines 157-189) | Replace `TaskDecision` discriminated union: 3 action → 6 action + `.transform()` legacy shim + drop task-level `priority` / `actionNeeded` payload | Modify | | `lambda/contacts-analyzer/src/handler.ts` (line 325) | Drop `actionNeeded: analysis.actionNeeded` log field (derived, no longer in `ContactsAnalysisSchema`) | Modify | | `lambda/contacts-analyzer/src/core/prompt-builder.ts` (lines 57, 135, 224, 399-417, 434, 604-661, 679-680, 748) | Remove `actionNeeded` / `actionNeededReason` from output schema text + add 6-action vocabulary description | Modify | | `lambda/contacts-analyzer/src/infrastructure/neon-repository.ts` (lines 1-46 imports, lines 419-1109 `writeAnalysisWithTasks`) | Replace inline `client.update(tasks)` / `client.insert(tasks)` / `client.insert(contactTimeline)` + `buildDncCloseStatements` call with `applyTaskAction()` / `upsertIdentity()` / `closeAllOpenForContact()` / `writeTimelineEvent()` | Modify | | `lambda/contacts-analyzer/tests/integration/task-decisions.test.ts` | New integration test — 5 scenarios from §5.2 | **Create** | | `lambda/contacts-analyzer/tests/unit/models.test.ts` (or extend existing) | Verify `TaskDecision` 6-action union narrowing + legacy `create` → `create_open` shim | Modify | Existing tests that touched the 3-action loop (`tests/unit/neon-repository*.test.ts` if present) need re-baseline against the new statement-shape — implementer scans and updates in Task 2 / Task 4. *** ## Task 1: models.ts — TaskDecision Zod schema swap to 6 actions **Files:** - Modify: `lambda/contacts-analyzer/src/core/models.ts` (lines 157-189) - Modify (or Create): `lambda/contacts-analyzer/tests/unit/models.test.ts` **Why this task first:** `TaskDecision` is the contract the prompt + downstream loop both consume. Every other task in this plan reads it. Land the schema first, run-time tests stay green via the legacy `create` shim while the prompt engineer hasn't shipped yet. ### Step 1: Map test scenarios | # | Scenario | Expected | Why test | | -- | ------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------- | | 1 | AI output `{action:'create_open', typeCategory:'lead_follow_up', suggestedActions:[...], reason:'...'}` | `TaskDecision.parse` returns object with `action='create_open'` | Happy path for the new vocabulary | | 2 | Legacy AI output `{action:'create', typeCategory:'lead_follow_up', suggestedActions:[...], priority:'high', reason:'...'}` | `TaskDecision.parse` returns `{action:'create_open', ...}` (shim normalized) + `priority` silently dropped | Backward compat while prompt engineer hasn't cut over | | 3 | `{action:'create_closed', typeCategory:'cancellation_risk', suggestedActions:[...], sourceCallId:'CALL_123', closeResult:'cancel_saved', reason:'...'}` | parses; `sourceCallId` non-empty string | S5 evidence reference; required field | | 4 | `{action:'create_closed', typeCategory:'...', suggestedActions:[...], closeResult:'...', reason:'...'}` — **missing** `sourceCallId` | Zod throws on `sourceCallId` | Schema must enforce S5 evidence rule | | 5 | `{action:'close', taskId:'', typeCategory:'...', closeResult:'converted', reason:'...'}` | parses | close action shape unchanged from legacy | | 6 | `{action:'update', taskId:'', typeCategory:'...', suggestedActions:[...], reason:'...'}` (no `priority`) | parses | update payload is `suggestedActions` / `dueAt` patch — caller drops task-level `priority` | | 7 | `{action:'record_progress', taskId:'', progressType:'no_answer', channel:'phone', callId:'C1', reason:'...'}` | parses | New action for progress logging | | 8 | `{action:'record_progress', taskId:'', progressType:'no_answer', channel:'phone', reason:'...'}` — no `callId`/`messageId` | Zod throws or schema marks `callId`/`messageId`/`actorId` required-one-of | Idempotency evidence must exist | | 9 | `{action:'reopen', taskId:'', reason:'...'}` | parses | New action | | 10 | Discriminator narrowing — `TaskDecision.parse({action:'create_open', ...}).action === 'create_open'` lets TS narrow `payload` | TS infer type-checks at compile time (`// @ts-expect-error` pattern) | Defends against accidental loss of discriminator | | 11 | Top-level still has `.max(20)` cap | `taskDecisions: Array(25).fill(decision)` throws | Business sanity cap unchanged from current schema | **Out of scope (not tested here):** - AI **content** correctness (a different layer). - The transform itself producing different payload **shape** than `create_open` — we only normalize the `action` literal; the rest of payload (`priority` → derived; `suggestedActions` → kept) is shipped through. ### Step 2: Implementation Replace lines 157-189 of `src/core/models.ts`: ```typescript // ── Task Decision Types (AI-driven per-task decisions) ── /* * Phase 1 unified-pipeline: 6 action vocabulary aligned with * @retaintive/common/domain TaskAction union. Replaces the old 3 action * union (close / update / create) that conflated "create open task" with * "log a closed-on-contact outcome". * * action values mirror callytics-common@1.2.0/domain TaskAction: * - create_open : new task, status='open' (was: action='create') * - create_closed : new task, status='closed' (S5 — needs sourceCallId) * - close : close existing open task * - update : patch existing open task (dueAt / suggestedActions) * - record_progress : log no-answer / left-voicemail / text-sent etc. * without closing the task (Phase 1 new lifecycle) * - reopen : flip closed → open * * Backward compat: AI prompt may still emit action='create' until prompt * engineer ships the rewrite. Zod `.transform()` normalizes 'create' → 'create_open' * and silently drops task-level `priority` (Orchestrator derives it from * max(suggestedActions[].priority), per normative-spec §3.1 S1). * * Task-level `priority` field DELETED from payload — Orchestrator owns it. * Top-level `actionNeeded` boolean DELETED from ContactsAnalysisSchema — * derived from open-task EXISTS query in studio-api (§4g). */ const TaskCloseDecision = z.object({ action: z.literal('close'), taskId: z.string().min(1), typeCategory: z.enum(TASK_TYPE_CATEGORY), closeResult: z.enum(TASK_CLOSE_RESULT), reason: z.string().describe('Why close this task — keep concise (~1-2 sentences).'), }); const TaskUpdateDecision = z.object({ action: z.literal('update'), taskId: z.string().min(1), typeCategory: z.enum(TASK_TYPE_CATEGORY), /* * Phase 1: task-level priority dropped. Orchestrator derives priority * from max(suggestedActions[].priority). Update payload patches * suggestedActions; new priority + dueAt fall out of that. */ suggestedActions: z.array(SuggestedActionSchema).optional(), reason: z.string().describe('Why update this task — keep concise (~1-2 sentences).'), }); const TaskCreateOpenDecision = z.object({ action: z.literal('create_open'), typeCategory: z.enum(TASK_TYPE_CATEGORY), suggestedActions: z.array(SuggestedActionSchema), reason: z.string().describe('Why create this task — keep concise (~1-2 sentences).'), }); /* * S5 (normative-spec §2.2): create_closed represents "task settled in-call" * — e.g. AI sees the same call that closed the lead, so it logs a task * row with status='closed' + the originating callId for partial-unique dedupe. */ const TaskCreateClosedDecision = z.object({ action: z.literal('create_closed'), typeCategory: z.enum(TASK_TYPE_CATEGORY), suggestedActions: z.array(SuggestedActionSchema), sourceCallId: z.string().min(1), closeResult: z.enum(TASK_CLOSE_RESULT), closeNote: z.string().optional(), reason: z.string().describe('Why close immediately — keep concise (~1-2 sentences).'), }); const TaskRecordProgressDecision = z.object({ action: z.literal('record_progress'), taskId: z.string().min(1), /* * 6 progress types from callytics-common task-progress-events schema. * Aligned with normative-spec §2.1 CHECK constraint values. */ progressType: z.enum([ 'no_answer', 'left_voicemail', 'text_sent', 'callback_requested', 'follow_up_scheduled', 'customer_considering', ]), channel: z.enum(['phone', 'sms', 'voicemail', 'email']), /* * Idempotency evidence: AI must supply at least one of callId / messageId. * (manual / system source paths are reserved for staff / cron and don't * surface in contacts-analyzer prompt output.) * Enforced via .superRefine below. */ callId: z.string().optional(), messageId: z.string().optional(), note: z.string().optional(), reason: z.string().describe('Why record this progress — keep concise (~1-2 sentences).'), }); const TaskReopenDecision = z.object({ action: z.literal('reopen'), taskId: z.string().min(1), reason: z.string().describe('Why reopen this task — keep concise (~1-2 sentences).'), }); /* * Legacy shim: AI may still emit action='create' until the prompt engineer * ships the rewrite (Appendix A of normative-spec). We accept it and normalize. * * Implementation note: Zod discriminated unions don't natively support * pre-transforms on the discriminator, so we use a plain z.union + a * .transform() at the top level. Performance impact is negligible * (max 20 decisions/contact, single .max() cap below). */ const TaskCreateLegacyDecision = z.object({ action: z.literal('create'), typeCategory: z.enum(TASK_TYPE_CATEGORY), /* * priority is accepted (legacy) but DROPPED by .transform() — Orchestrator * derives it from suggestedActions[].priority (normative-spec §3.1 S1). */ priority: PriorityEnum.optional(), suggestedActions: z.array(SuggestedActionSchema), reason: z.string(), }); export const TaskDecision = z .union([ TaskCloseDecision, TaskUpdateDecision, TaskCreateOpenDecision, TaskCreateClosedDecision, TaskRecordProgressDecision, TaskReopenDecision, TaskCreateLegacyDecision, ]) .transform((d) => { // Normalize legacy 'create' → 'create_open'; drop task-level priority. if (d.action === 'create') { const { action: _action, priority: _priority, ...rest } = d; return { action: 'create_open' as const, ...rest }; } return d; }) .superRefine((d, ctx) => { if (d.action === 'record_progress' && !d.callId && !d.messageId) { ctx.addIssue({ code: z.ZodIssueCode.custom, path: ['callId'], message: 'record_progress requires at least one of callId / messageId for idempotency', }); } }); export type TaskDecision = z.infer; ``` Then update the `ContactsAnalysisSchema` block (lines 202-263 of the same file) to delete `actionNeeded` and `actionNeededReason`: ```typescript export const ContactsAnalysisSchema = z.object({ customerSummary: z .string() .describe('Rolling customer journey portrait — keep concise (~3-5 sentences).'), /* * Phase 1: actionNeeded boolean DELETED — derived from open-task EXISTS * query in studio-api (normative-spec §4g). AI no longer outputs this. * actionNeededReason similarly removed; reasoning lives inside each * suggestedActions[].reason and inside taskDecisions[].reason. */ suggestedActions: z.array(SuggestedActionSchema).default([]), leadStatus: LeadStatusEnum, leadStatusReason: z .string() .optional() .describe('Why this lead status — keep concise (~1-2 sentences).'), purchaseIntent: IntentEnum.optional(), purchaseIntentReason: z .string() .optional() .describe('Why this purchase intent — keep concise (~1-2 sentences).'), goals: z.array(GoalSchema).default([]), leadObjections: z.array(z.string().describe('Single objection — short phrase.')).default([]), leadRejectionReasons: z .array(z.string().describe('Single rejection reason — short phrase.')) .default([]), /* * lifecycleStage / lifecycleState — RETAINED (撤销退役, normative-spec §2.3). * Engineer feedback 2026-06-02: churned re-engage etc. need AI semantic * judgement; lead_declined close_result also depends on lifecycleState. */ lifecycleStage: LifecycleStageEnum, lifecycleState: LifecycleStateEnum, doNotContact: z.boolean().optional(), hasOpenComplaint: z.boolean().optional(), /* * typeCategory / priority on the CONTACT level — kept for compat with * downstream readers that still surface "what category does this contact * sit in" outside of any specific task. NOT the same as task-level * priority (which is dropped from per-task TaskDecision payload). */ typeCategory: z.enum(TASK_TYPE_CATEGORY).optional(), priority: PriorityEnum.optional(), taskDecisions: z.array(TaskDecision).max(20).default([]), customerFirstName: z .string() .max(100) .optional() .describe( 'Customer first name ONLY if clearly self-stated in transcript (e.g. "Hi, I\'m Katie"). Omit if uncertain.', ), customerLastName: z .string() .max(100) .optional() .describe('Customer last name ONLY if clearly self-stated in transcript. Omit if uncertain.'), }); ``` ### Step 3: Test write + run Implementer writes 11 test cases per the case map above into `tests/unit/models.test.ts`. Pattern: ```typescript // tests/unit/models.test.ts import { describe, test, expect } from 'bun:test'; import { TaskDecision, ContactsAnalysisSchema } from '../../src/core/models'; describe('TaskDecision schema (Phase 1 6-action union)', () => { test('case 1 — create_open happy path', () => { const parsed = TaskDecision.parse({ action: 'create_open', typeCategory: 'lead_follow_up', suggestedActions: [ { action: 'Call back', reason: 'Caller asked for callback', priority: 'high', priorityReason: 'Hot lead' }, ], reason: 'New lead from web form', }); expect(parsed.action).toBe('create_open'); }); test('case 2 — legacy create normalizes to create_open + drops priority', () => { const parsed = TaskDecision.parse({ action: 'create', typeCategory: 'lead_follow_up', priority: 'high', // legacy task-level priority, should be dropped suggestedActions: [ { action: 'Call back', reason: '...', priority: 'high', priorityReason: '...' }, ], reason: 'New lead', }); expect(parsed.action).toBe('create_open'); // @ts-expect-error — priority should not be on the parsed object expect(parsed.priority).toBeUndefined(); }); test('case 4 — create_closed without sourceCallId rejects', () => { expect(() => TaskDecision.parse({ action: 'create_closed', typeCategory: 'cancellation_risk', suggestedActions: [], closeResult: 'cancel_saved', reason: 'Saved on call', // missing sourceCallId }), ).toThrow(); }); test('case 8 — record_progress without callId or messageId rejects', () => { expect(() => TaskDecision.parse({ action: 'record_progress', taskId: '11111111-1111-1111-1111-111111111111', progressType: 'no_answer', channel: 'phone', reason: 'No answer', }), ).toThrow(); }); // cases 3, 5, 6, 7, 9, 10, 11 follow the same pattern }); ``` Run: ```bash cd /Users/maxwsy/workspace/callytics-infrastructure/lambda/contacts-analyzer bun test tests/unit/models.test.ts ``` Expected: 11 PASS. ### Step 4: Commit ```bash git add lambda/contacts-analyzer/src/core/models.ts lambda/contacts-analyzer/tests/unit/models.test.ts git commit -m "feat(contacts-analyzer): TaskDecision 6-action union + drop actionNeeded Phase 1 Plan 03 Task 1. - TaskDecision union expanded to 6 actions: create_open / create_closed / close / update / record_progress / reopen (aligns with @retaintive/common@1.2.0 domain TaskAction). - Legacy 'create' literal accepted via .transform() shim, normalized to 'create_open' + task-level priority silently dropped (Orchestrator derives priority from max(suggestedActions[].priority), normative-spec §3.1 S1). - create_closed payload requires sourceCallId (S5 evidence reference for partial-unique dedupe). - record_progress payload requires at least one of callId / messageId (idempotency evidence enforced via .superRefine). - ContactsAnalysisSchema: actionNeeded + actionNeededReason DELETED (derived from EXISTS open-task in studio-api per §4g). - lifecycleStage / lifecycleState RETAINED (撤销退役, engineer feedback). Spec: normative-spec.md §4d + §3.1 (S1 + S5) + Appendix A (legacy shim). " ``` *** ## Task 2: Bump `@retaintive/common` dependency 1.0.x → 1.2.0 **Files:** - Modify: `lambda/contacts-analyzer/package.json` (or root `package.json` catalog block if pnpm catalog) **Why this task here (not later):** TaskDecision (Task 1) imports `TASK_CLOSE_RESULT` from `@retaintive/common/db`. The new 15-value enum (with `unable_to_reach`) ships in 1.2.0 — without the bump, AI emitting `unable_to_reach` would Zod-reject. Tasks 3-7 need `@retaintive/common/domain` importable. Bump before any caller-side import refactor. ### Step 1: Test scenarios Plan-level dependency bump has no logic of its own. Verification = downstream builds compile + `@retaintive/common/domain` resolves. | # | Scenario | Expected | | - | ----------------------------------------------------------------------------------- | -------------------------------------- | | 1 | `pnpm install` (or `npm install` per repo convention) after bump | lockfile updates, no peer-dep warnings | | 2 | `bun test tests/unit/models.test.ts` (Task 1 tests) still pass with new enum source | 11 PASS | | 3 | TypeScript build of contacts-analyzer | `tsc --noEmit` clean | | 4 | `import { applyTaskAction } from '@retaintive/common/domain'` resolves | no `Cannot find module` error | ### Step 2: Implementation Find the version pin location. Repo uses pnpm catalog (verified via `grep '"@retaintive/common"' callytics-infrastructure/package.json` → `"^1.0.0"` in root, lambda packages use `"catalog:"`). ```diff --- a/package.json (callytics-infrastructure root) +++ b/package.json @@ "@retaintive/common": "^1.0.0", + "@retaintive/common": "^1.2.0", ``` Or if a `pnpm-workspace.yaml` catalog block exists, update there instead. (Confirm via `grep -A5 catalog: pnpm-workspace.yaml`.) If `package.json` lists `@retaintive/common` only in the lambda subpackage (no root catalog), update lambda's `package.json`: ```diff --- a/lambda/contacts-analyzer/package.json +++ b/lambda/contacts-analyzer/package.json @@ - "@retaintive/common": "catalog:", + "@retaintive/common": "^1.2.0", ``` After edit: ```bash cd /Users/maxwsy/workspace/callytics-infrastructure pnpm install # or whichever the repo uses; check for lockfile name (pnpm-lock.yaml / package-lock.json / bun.lockb) ``` ### Step 3: Verification ```bash cd /Users/maxwsy/workspace/callytics-infrastructure/lambda/contacts-analyzer # A. Module resolves node -e "console.log(Object.keys(require('@retaintive/common/domain')))" # Expected: ['applyTaskAction', 'closeAllOpenForContact', 'upsertIdentity', 'setDNC', 'touchActivity', 'writeTimelineEvent', 'computeNextDueAt', 'computeAllowedTypeCategories', ...] # B. Type-check still clean pnpm run typecheck # or whichever the script is named # Expected: 0 errors # C. Task 1 tests still green bun test tests/unit/models.test.ts ``` ### Step 4: Commit ```bash git add package.json pnpm-lock.yaml # or whichever lockfile is present # If lambda subpackage was edited instead: # git add lambda/contacts-analyzer/package.json pnpm-lock.yaml git commit -m "chore(contacts-analyzer): bump @retaintive/common 1.0.x → 1.2.0 Phase 1 Plan 03 Task 2. Unlocks @retaintive/common/domain import path for Task Orchestrator, Contact Writer, Timeline Writer used by Tasks 3-6 of this plan. Spec: Plan 02 ships @retaintive/common@1.2.0; this plan consumes it. " ``` *** ## Task 3: neon-repository — taskDecisions loop swapped to `applyTaskAction()` **Files:** - Modify: `lambda/contacts-analyzer/src/infrastructure/neon-repository.ts` (lines 559-892 — the `for (const decision of taskDecisions)` loop, including the inline derivedTaskType / `client.update(tasks)` close branch / `client.insert(tasks)` create branch / update branch) **Why this task:** This is the structural heart of the migration. The 332-line inline loop becomes \~80 lines that delegate to `applyTaskAction()`. Decisions reject (low confidence, hallucinated taskId, etc.) become uniform `ApplyResult` handling, no longer scattered as `logger.warn + break`. ### Step 1: Map test scenarios These overlap with §5.2 contacts-analyzer integration test row + §5.0 invariants in normative-spec. Test list lives in Task 7 (integration test file). Per-action unit verification of Orchestrator behavior is owned by Plan 02 — we only verify caller-side wiring here: | # | Caller-side scenario | Verify | | - | ------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------- | | 1 | Mixed decisions: `[create_open(A), close(B), record_progress(C), reopen(D)]` — all pass Policy Guard → 4 `ApplyResult.status='allow'` returned, statements appended | `client.batch(statements)` receives 4+ task statements + their timeline statements | | 2 | One decision `reject` (hallucinated taskId), other 3 `allow` | Caller skips the reject, batches the 3 allow (Orchestrator is idempotent at the decision level; reject does NOT poison the batch) | | 3 | `aiRunStartedAt` passed into each `applyTaskAction()` call | Set from `handler.ts` → propagated through ctx for human-authority guard (stale\_proposal reject path) | | 4 | All decisions `reject` (e.g., DNC contact, 5 create proposals all rejected) | `statements` only contains the contact UPDATE; batch still commits | | 5 | `taskDecisions=[]` (AI emits no decisions, common case) | Loop runs 0 times; statements has just contact UPDATE + DNC cascade (if applicable) + lifecycle + `contact_analysis.completed` | | 6 | Logger emits per-reject reason for observability | `logger.warn` called with `{action, reason, details, taskId?, typeCategory?}` | ### Step 2: Implementation Replace lines 559-892 (the entire `for (const decision of taskDecisions)` block, including the closing brace) with: ```typescript // ── ④ Apply task decisions via Task Orchestrator ── /* * Phase 1 unified-pipeline: dispatch every TaskDecision through * @retaintive/common/domain applyTaskAction(). This collapses the * old 332-line inline branch (close / create / update) into a single * loop that delegates Policy Guard (DNC / hallucination / state * transition / typeCategory allowed-set / closeNote / confidence / * create_closed-no-open-task) entirely to the shared module. * * Reject paths log a warn for observability + CloudWatch metric; * they do NOT poison the batch — other decisions in the same run * still apply (Orchestrator is per-decision atomic at the statement * level; cross-decision atomicity comes from the outer client.batch()). */ const closingTaskIds = new Set( taskDecisions .filter((d): d is Extract => d.action === 'close') .map((d) => d.taskId), ); for (const decision of taskDecisions) { const baseCtx = { storeId, actor: { type: 'contact_analysis' as const, id: ACTOR_SUBJECT_ID, name: ACTOR_SOURCE_SYSTEM, }, aiRunStartedAt: new Date(), // handler captures this before AI call; see Task 5 db: client, }; // create_open / create_closed need contactPhone + franchise/account // for identity upsert (CreateActionContext, normative-spec §3.1 S6). const ctx = decision.action === 'create_open' || decision.action === 'create_closed' ? { ...baseCtx, contactPhone: phone, franchiseId, accountId } : baseCtx; // Build the TaskAction discriminated payload from TaskDecision. // Note: TaskDecision (Task 1) has been normalized by .transform(); // we read `decision.action` and forward payload fields 1:1. const action = buildTaskAction(decision, contactAnalysisRunId); const result = await applyTaskAction(action, ctx); if (result.status === 'reject') { logger.warn('Task decision rejected by Policy Guard', { phone, storeId, action: decision.action, reason: result.reason, details: result.details, taskId: 'taskId' in decision ? decision.taskId : undefined, typeCategory: 'typeCategory' in decision ? decision.typeCategory : undefined, contactAnalysisRunId, }); continue; } if (result.status === 'needs_review') { // Phase 1: needs_review not emitted (per spec §3.1 line 168); // log defensively in case Phase 2 wiring leaks into a Phase 1 caller. logger.warn('Task decision returned needs_review (Phase 2 path leaked)', { phone, storeId, action: decision.action, details: result.details, contactAnalysisRunId, }); continue; } statements.push(...result.statements); // resultChecks contract(normative-spec §3.1 line 251) // Orchestrator 用 ApplyResult.statements 表达 SQL 但不能保证 RETURNING // 拿到 row(duplicate / stale_proposal / task_not_open / task_not_closed // 都会 0 rows)。caller MUST read resultChecks after batch — 见 batch // 执行后的 `for (const check of result.resultChecks ?? []) {...}` 段。 // 这里把 `result.resultChecks` 跟当前 result 一起存,batch 后统一处理: pendingResultChecks.push({ result, decision }); switch (decision.action) { case 'close': tasksClosedCount++; break; case 'create_open': case 'create_closed': tasksCreatedCount++; break; case 'update': case 'record_progress': case 'reopen': tasksUpdatedCount++; break; } } /* * Batch 后处理 resultChecks(normative-spec §3.1 line 251): * client.batch() 返回 array of QueryResult,每个 statement 对应一项; * 用 result.resultChecks[i].statementIndex 找到 RETURNING row count, * 0 行 → silent business reject。这是 Codex 抓的 silent failure pattern: * * batchResults = await client.batch(statements); // 见 §下方 * for (const { result, decision } of pendingResultChecks) { * for (const check of result.resultChecks ?? []) { * const queryResult = batchResults[globalStatementIndex(check.statementIndex)]; * if (queryResult.rowCount === 0) { * logger.warn('Task action silent reject', { * reason: check.zeroRowsReason, // 'duplicate' / 'stale_proposal' / 'task_not_open' / 'task_not_closed' * action: decision.action, * taskId: 'taskId' in decision ? decision.taskId : undefined, * contactAnalysisRunId, * }); * // Phase 1:log + skip counter increment;Phase 2 surface reject 给 needs_review queue * } * } * } */ const pendingResultChecks: Array<{ result: ApplyResult & { status: 'allow' }; decision: TaskDecision }> = []; /* * `closingTaskIds` feeds the DNC cascade below — Orchestrator's * closeAllOpenForContact() needs to know which open task IDs were * already-closed by the AI's per-decision close calls in this same * batch, so it doesn't generate duplicate UPDATE statements. * (Idempotency at the statement level is still guaranteed by the * `WHERE status IN ('pending','open')` clause; this is a clarity + * timeline-noise reduction concern.) */ ``` Add this helper near the top of the file (after imports, before `createNeonRepository`): ```typescript import { applyTaskAction, closeAllOpenForContact, upsertIdentity, writeTimelineEvent, type TaskAction, } from '@retaintive/common/domain'; /* * Convert TaskDecision (caller-side discriminated union from src/core/models.ts) * into a TaskAction (Orchestrator-side discriminated union from * @retaintive/common/domain). The two unions are intentionally 1:1 by * `action` literal value; this helper just narrows + maps payload fields. */ function buildTaskAction( decision: TaskDecision, contactAnalysisRunId: string, ): TaskAction { switch (decision.action) { case 'create_open': return { action: 'create_open', payload: { typeCategory: decision.typeCategory, suggestedActions: decision.suggestedActions, sourceType: 'contact_analysis', contactAnalysisRunId, // Note: confidence is NOT in TaskDecision payload (AI doesn't // self-report); Plan 02 Orchestrator skips checkConfidence when // confidence === undefined (treats as staff). // TODO Phase 2: surface per-decision confidence from AI prompt // → wire here so low_confidence reject path activates for AI. }, }; case 'create_closed': return { action: 'create_closed', payload: { typeCategory: decision.typeCategory, suggestedActions: decision.suggestedActions, sourceType: 'contact_analysis', sourceCallId: decision.sourceCallId, closeResult: decision.closeResult, closeNote: decision.closeNote, }, }; case 'close': return { action: 'close', payload: { taskId: decision.taskId, closeResult: decision.closeResult, closeNote: decision.reason, }, }; case 'update': return { action: 'update', payload: { taskId: decision.taskId, suggestedActions: decision.suggestedActions, }, }; case 'record_progress': return { action: 'record_progress', payload: { taskId: decision.taskId, progressType: decision.progressType, channel: decision.channel, callId: decision.callId, messageId: decision.messageId, note: decision.note, }, }; case 'reopen': return { action: 'reopen', payload: { taskId: decision.taskId, reason: decision.reason, }, }; } } ``` **Delete** the now-orphaned local helpers from neon-repository.ts: - `derivedTaskType` ternary at line 707-708 (Orchestrator's `create_open` SQL builder owns the legacy `task_type` shim per spec §6.2) - `processedCloseIds` / `closingCategories` / `effectivePendingCategories` Sets (Orchestrator's `checkCreateClosedHasNoOpenTask` / partial unique enforce these via SQL) - `knownPendingIds` Set (`checkTaskBelongsToContact` does this via DB query) **Keep** the `pendingTasks` parameter on `writeAnalysisWithTasks` for now — Task 4 uses it to drive `closeAllOpenForContact()` and to keep the existing `predecessorTaskId` lineage logic if Phase 1 wants to preserve it (or drop in Task 4 if it doesn't). ### Step 3: Test write + run Pure unit tests of the new loop are hard (Orchestrator is mocked, batch is mocked) — integration test in Task 7 is the real verification. Implementer adds a thin unit test of `buildTaskAction()` (6 cases, one per action) to lock the 1:1 payload mapping: ```typescript // tests/unit/build-task-action.test.ts import { describe, test, expect } from 'bun:test'; import { buildTaskAction } from '../../src/infrastructure/neon-repository'; // (export buildTaskAction for testability; or test indirectly via integration) test('create_open mapping', () => { const action = buildTaskAction( { action: 'create_open', typeCategory: 'lead_follow_up', suggestedActions: [{ action: 'x', reason: 'y', priority: 'high', priorityReason: 'z' }], reason: '...', }, 'run-1', ); expect(action.action).toBe('create_open'); expect(action.payload.sourceType).toBe('contact_analysis'); expect(action.payload.contactAnalysisRunId).toBe('run-1'); }); // 5 more — one per action literal ``` Run: ```bash bun test tests/unit/build-task-action.test.ts pnpm run typecheck ``` Expected: 6 PASS + 0 type errors. ### Step 4: Commit ```bash git add lambda/contacts-analyzer/src/infrastructure/neon-repository.ts lambda/contacts-analyzer/tests/unit/build-task-action.test.ts git commit -m "refactor(contacts-analyzer): taskDecisions loop → applyTaskAction() Phase 1 Plan 03 Task 3. - 332-line inline loop (lines 559-892) replaced with 80-line dispatch to @retaintive/common/domain applyTaskAction(). - DerivedTaskType ternary, processedCloseIds / closingCategories / effectivePendingCategories / knownPendingIds local guards deleted — all duties move to Orchestrator's Policy Guard + SQL partial unique. - buildTaskAction() helper maps TaskDecision → TaskAction 1:1 by action literal (6 cases, one unit test per case). - reject path logs warn + CloudWatch metric (continue, don't poison batch); needs_review path defensively logs (Phase 1 doesn't emit it). - closingTaskIds Set retained for Task 4 DNC cascade dedupe. Spec: normative-spec.md §4d row 1 + §3.1 ActionContext split (S6). " ``` *** ## Task 4: neon-repository — Contact identity UPSERT swapped to `upsertIdentity()` + DNC cascade swapped to `closeAllOpenForContact()` **Files:** - Modify: `lambda/contacts-analyzer/src/infrastructure/neon-repository.ts` - lines 467-540 (contact UPDATE block) — partial swap: identity fields (firstName / lastName / trustScore / lastActivityAt) move to `upsertIdentity()`; analysis fields (customerSummary / leadStatus / lifecycleStage / ...) stay inline because Phase 1 doesn't promote them to a shared writer - lines 894-958 (DNC cascade block) — swap `buildDncCloseStatements` for `closeAllOpenForContact()` - Delete (`import` line at top): `import { buildDncCloseStatements } from '../../../shared/utils/dnc-cascade';` (line 33) **Why this task:** Phase 1 promotes only identity + DNC sticky write + lastActivityAt forward-only to `ContactWriter` (spec §3.4). The other 18 AI-generated fields (customerSummary, leadStatus, lifecycleStage, ...) stay inline because contacts-analyzer is the only writer — no multi-writer conflict to resolve. Phase 2 unifies the rest. ### Step 1: Map test scenarios | # | Scenario | Expected | | - | ----------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | 1 | New contact (not yet in `contacts`) — AI emits `customerFirstName='Katie'` + lastActivityAt=NOW | `upsertIdentity()` INSERT branch fires; downstream `client.update(contacts).set({customerSummary, leadStatus, ...})` matches 0 rows on first batch (UPDATE-after-INSERT in same batch?) — see implementation note below | | 2 | Existing contact, AI identity fields present | caller passes `trustScore=NAME_TRUST.AI_TRANSCRIPT`, phone, storeId, franchiseId/accountId into `upsertIdentity()` | | 3 | Existing contact with prior higher-trust name | integration may assert final name unchanged, but do not duplicate the full NAME\_TRUST matrix here; Plan 02 owns module internals | | 4 | AI sets `doNotContact=true` (sticky); contact had `doNotContact=false` previously | `closeAllOpenForContact()` fires for all open task IDs; statements include 1 bulk UPDATE + N timeline INSERT | | 5 | Contact had `doNotContact=true` already; AI sets `doNotContact=true` again | `closeAllOpenForContact()` is no-op (no open tasks remaining); `closedTaskIds=[]` | | 6 | AI emits 3 close decisions in taskDecisions loop, then DNC=true triggers cascade | `closeAllOpenForContact()` SHOULD skip the 3 already-closed task IDs (pass closingTaskIds from Task 3); only NEW open tasks closed by cascade | | 7 | DNC=true but `closingTaskIds.size === pendingTasks.length` (AI closed everything explicitly) | cascade no-op; no extra timeline rows | **Note on test 1 (UPDATE-after-INSERT atomicity):** because `client.batch()` is one transaction (Neon-HTTP transaction), an INSERT into `contacts` from `upsertIdentity()` followed by an UPDATE on the same row is fine — both see each other's writes. Verify in integration test (Task 7). ### Step 2: Implementation — Contact UPSERT partial swap The current code (lines 477-540) is a `client.update(contacts).set({...})` that does NOT INSERT. Phase 1 needs both behaviors: 1. New contact (first time analyzed) — `upsertIdentity()` does INSERT + name fields; analysis fields can't be set on a contact that doesn't exist yet. 2. Existing contact — `upsertIdentity()` updates name (winner-takes-name pattern); analysis fields update inline below. Insert `upsertIdentity()` call **before** the inline `client.update(contacts).set({...})` so identity row exists first. The inline UPDATE then patches the 18 analysis fields: ```typescript // ── ②ʹ Identity upsert (Plan 02 ContactWriter) ── /* * Phase 1 normative-spec §3.4: identity / DNC / lastActivityAt are * the only contact fields promoted to a shared writer this phase. * The 18 AI-generated analysis fields (customerSummary, leadStatus, * lifecycleStage, ...) stay inline below — contacts-analyzer is the * sole writer for those, no multi-writer race. Phase 2 unifies. * * upsertIdentity() owns: INSERT new contact (if absent) / * ON CONFLICT (phone, store_id) DO UPDATE name fields via NAME_TRUST * winner pattern / GREATEST() forward-only lastActivityAt. */ const aiTrustScore = NAME_TRUST.AI_TRANSCRIPT; statements.push( upsertIdentity({ phone, storeId, franchiseId, accountId, firstName: analysis.customerFirstName, lastName: analysis.customerLastName, trustScore: aiTrustScore, activityAt: new Date(), }), ); // ── ③ UPDATE contacts (analysis fields, Phase 1 still inline) ── statements.push( client .update(contacts) .set({ customerSummary: analysis.customerSummary, // actionNeeded / actionNeededReason DELETED — Task 1 dropped // these from the schema; routes/v3/contacts.ts now derives // via EXISTS open-task query (§4g). suggestedActions: analysis.suggestedActions, leadStatus: analysis.leadStatus, leadStatusReason: analysis.leadStatusReason ?? null, purchaseIntent: analysis.purchaseIntent ?? null, purchaseIntentReason: analysis.purchaseIntentReason ?? null, goals: analysis.goals, leadObjections: analysis.leadObjections, leadRejectionReasons: analysis.leadRejectionReasons, lifecycleStage: analysis.lifecycleStage, lifecycleState: analysis.lifecycleState, /* * DNC sticky write: AI can flip false → true, never true → false. * Compose ContactWriter helper fragments(Plan 02 §3.4)— **不再 inline** * verbatim CASE WHEN(audit fix 2026-06-02)。两 helper 必须同时在 * 同一 UPDATE row write,所以用 SQL fragment 形态(非 statement)。 * Pattern source 验证: message-processor:332-335。 */ ...stickyDncFragments({ newDnc: analysis.doNotContact ?? false, updatedBy: 'ai', // contacts-analyzer 写入 = ai actor }), hasOpenComplaint: analysis.hasOpenComplaint ?? null, // firstName / lastName / firstNameTrustScore / firstNameUpdatedAt // / lastActivityAt — DELETED here; upsertIdentity() owns these. lastContactAnalysisAt: sql`NOW()`, updatedAt: sql`NOW()`, }) .where(and(eq(contacts.phone, phone), eq(contacts.storeId, storeId))), ); ``` ### Step 3: Implementation — DNC cascade swap Replace lines 894-958 (the `if (analysis.doNotContact === true) { ... }` block, including the `buildDncCloseStatements` call): ```typescript // ── DNC HARD STOP: code-derived cascade close ── /* * Phase 1 normative-spec §3.4 + §4f: DNC bulk close moves from the * local buildDncCloseStatements helper to the shared * closeAllOpenForContact() Orchestrator helper (B7). Same semantics: * close every open task for the contact, single bulk UPDATE + * N timeline INSERTs in same batch. Caller passes the IDs already * closed in the per-decision loop (closingTaskIds from Task 3) so * the cascade doesn't double-write timeline rows. * * Trigger: only when AI just flipped doNotContact → true. Reading the * post-update value via the snapshot pattern: AI emitted analysis.doNotContact === true. * The DB row's sticky logic above means previously-true contacts also * land here — but at that point pendingTasks should already be empty * (their last analysis closed them), so cascade is no-op. */ if (analysis.doNotContact === true) { const cascade = await closeAllOpenForContact( { contactPhone: phone, storeId, closeResult: 'do_not_contact', closeNote: 'Auto-closed by DNC hard stop', actor: { type: 'contact_analysis', id: ACTOR_SUBJECT_ID, name: ACTOR_SOURCE_SYSTEM, }, }, { db: client }, ); if (cascade.status === 'reject') { logger.error('closeAllOpenForContact rejected', { phone, storeId, reason: cascade.reason, details: cascade.details, contactAnalysisRunId, }); } else { /* * De-dupe: drop statements whose target taskId was already closed * in the per-decision loop. closeAllOpenForContact() returns * closedTaskIds for the bulk UPDATE; statements are 1 UPDATE + * N timeline INSERTs. We append all and rely on the UPDATE's * `WHERE status IN ('pending','open')` to skip already-closed rows * (correctness) and the timeline `ON CONFLICT (idempotency_key) * DO NOTHING` (Plan 02 TimelineWriter) to skip duplicate timeline. * No client-side filtering needed — SQL layer handles both races. */ statements.push(...cascade.statements); const newlyClosed = cascade.closedTaskIds.filter((id) => !closingTaskIds.has(id)); tasksClosedCount += newlyClosed.length; if (newlyClosed.length > 0) { logger.info('DNC hard stop — code-derived cascade closed pending tasks', { phone, storeId, closedCount: newlyClosed.length, taskIds: newlyClosed, contactAnalysisRunId, }); } } } ``` Delete the import at line 33: ```diff -import { buildDncCloseStatements } from '../../../shared/utils/dnc-cascade'; ``` ### Step 4: Test write + run Unit tests live in integration test (Task 7); inline-statement assertions need a real DB. Verify TypeScript compiles: ```bash cd /Users/maxwsy/workspace/callytics-infrastructure/lambda/contacts-analyzer pnpm run typecheck ``` Expected: 0 errors. (The `buildDncCloseStatements` import deletion may leave dead references in `shared/utils/dnc-cascade.ts` itself — that's owned by another caller / Phase 1 PR; not this lambda's concern.) ### Step 5: Commit ```bash git add lambda/contacts-analyzer/src/infrastructure/neon-repository.ts git commit -m "refactor(contacts-analyzer): identity → upsertIdentity, DNC → closeAllOpenForContact Phase 1 Plan 03 Task 4. - Contact identity (firstName / lastName / trustScore / lastActivityAt) moves to @retaintive/common/domain upsertIdentity() — NAME_TRUST winner pattern + GREATEST() forward-only. 18 AI analysis fields stay inline (sole writer, no race). - DNC bulk close (buildDncCloseStatements local helper) replaced with closeAllOpenForContact() Orchestrator helper (B7). dnc-cascade.ts import dropped. - De-dupe between per-decision close loop and DNC cascade handled by SQL layer (UPDATE WHERE status IN ('pending','open') + timeline ON CONFLICT idempotency_key DO NOTHING) — no client-side filtering. Spec: normative-spec.md §3.4 + §4f. " ``` *** ## Task 5: neon-repository — inline timeline writes swapped to `writeTimelineEvent()` **Files:** - Modify: `lambda/contacts-analyzer/src/infrastructure/neon-repository.ts` - lines 998-1034 (lifecycle changed timeline INSERT) - lines 1045-1076 (`contact_analysis.completed` timeline INSERT) - (`task.created` / `task.status_changed` / `task.updated` timeline INSERTs from old loop are gone after Task 3 — Orchestrator writes them now) - Remove now-unused import: `buildTimelineValues` from line 21 (verify with `grep`) **Why this task:** spec §3.5 — Timeline Writer owns event Zod schema + idempotencyKey + ON CONFLICT pattern. Phase 1 ports the 2 remaining caller-driven timeline writes to it (`contact.lifecycle_changed`, `contact_analysis.completed`). Task-related timeline writes (`task.created`, `task.status_changed`, `task.updated`, `task.progress_recorded`) are already gone via Task 3 — Orchestrator writes them internally with proper event type + payload. ### Step 1: Map test scenarios | # | Scenario | Expected | | - | ---------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------- | | 1 | Lifecycle changes (`prev.stage='lead' / new='member'`) | Single `contact.lifecycle_changed` timeline row written; idempotencyKey = `lifecycle_changed:${runId}` | | 2 | Lifecycle does NOT change | 0 timeline rows for lifecycle | | 3 | Every analysis emits `contact_analysis.completed` timeline | 1 row; idempotencyKey = `contact_analysis.completed:${runId}` | | 4 | Duplicate run (same runId reprocessed) | final DB has one timeline row for the idempotencyKey; do not assert TimelineWriter SQL internals | | 5 | Malformed payload validation | covered by Plan 02 TimelineWriter unit tests; Plan 03 only verifies caller sends the expected payload shape in happy path | ### Step 2: Implementation Replace lines 998-1034 (lifecycle changed) with: ```typescript // ── lifecycle 变了 → timeline (via Plan 02 TimelineWriter) ── if (prev.stage !== analysis.lifecycleStage || prev.state !== analysis.lifecycleState) { statements.push( writeTimelineEvent({ eventType: 'contact.lifecycle_changed', contactPhone: phone, storeId, franchiseId, accountId, actor: { type: 'contact_analysis', subjectId: ACTOR_SUBJECT_ID, name: ACTOR_SOURCE_SYSTEM, }, payload: { phone, oldStage: prev.stage ?? 'unknown', newStage: analysis.lifecycleStage, // (Plan 02 Zod schema only requires phone / oldStage / newStage; // lifecycleState change captured in modifiedFields if Plan 02 // schema supports it — verify against Plan 02 Task 5 final.) }, idempotencyKey: `lifecycle_changed:${contactAnalysisRunId}`, occurredAt: new Date(), entityType: 'contact', }), ); } ``` Replace lines 1045-1076 (`contact_analysis.completed`) with: ```typescript // ── contact_analysis.completed — always emit one per run ── statements.push( writeTimelineEvent({ eventType: 'contact_analysis.completed', contactPhone: phone, storeId, franchiseId, accountId, actor: { type: 'contact_analysis', subjectId: ACTOR_SUBJECT_ID, name: ACTOR_SOURCE_SYSTEM, }, payload: { phone, runId: contactAnalysisRunId, // (Plan 02 Zod schema for contact_analysis.completed only // requires phone + runId. The per-run counters // tasksClosedCount / tasksCreatedCount / tasksUpdatedCount // and lifecycleStage / lifecycleState — verify whether Plan 02 // schema accepts them as optional. If yes, include below; // if no, log via logger.info already at line 1098+ as today.) }, idempotencyKey: `contact_analysis.completed:${contactAnalysisRunId}`, occurredAt: new Date(), entityType: 'contact', }), ); ``` Remove now-orphan import (verify it's not referenced elsewhere in the file): ```bash grep -n "buildTimelineValues" lambda/contacts-analyzer/src/infrastructure/neon-repository.ts # Expected: 0 hits after Tasks 3-5 land. If hit remains, the import stays. ``` If `buildTimelineValues` is no longer referenced: ```diff - buildTimelineValues, ``` Also verify whether `ACTOR_SOURCE_TYPE` / `ACTOR_SOURCE_SYSTEM` / `AI_PROMPT_VERSION` / `AI_MODEL_USED` constants (lines 56-58, 48) are still consumed. If not, delete them in the same commit. ### Step 3: Test write + run Plan 02 Task 5 owns Timeline Writer unit tests (16 event types × Zod). Caller-side just verifies the integration test in Task 7 sees the lifecycle row appear when prev≠new. ```bash pnpm run typecheck ``` ### Step 4: Commit ```bash git add lambda/contacts-analyzer/src/infrastructure/neon-repository.ts git commit -m "refactor(contacts-analyzer): inline timeline → writeTimelineEvent() Phase 1 Plan 03 Task 5. - contact.lifecycle_changed + contact_analysis.completed timeline INSERTs delegate to @retaintive/common/domain writeTimelineEvent() — Plan 02 Zod payload validation + idempotency key + ON CONFLICT all centralised. - Task-related timeline INSERTs (task.created / task.status_changed / task.updated) already gone (Task 3 Orchestrator writes them). - buildTimelineValues import dropped (no remaining caller references). - Constants ACTOR_SUBJECT_ID retained (Plan 02 actor.subjectId pass-through); ACTOR_SOURCE_TYPE / ACTOR_SOURCE_SYSTEM / AI_PROMPT_VERSION / AI_MODEL_USED audit logged via logger.info, no longer in timeline payload (Plan 02 schema dictates). Spec: normative-spec.md §3.5 + §4d. " ``` *** ## Task 6: handler.ts — drop `actionNeeded` log field + capture `aiRunStartedAt` **Files:** - Modify: `lambda/contacts-analyzer/src/handler.ts` (line 325 + new line before line 302) **Why this task:** TaskDecision (Task 1) removed `actionNeeded` from `ContactsAnalysisSchema`. Anywhere in the handler that reads `analysis.actionNeeded` is now a TS error. Also: Orchestrator needs `aiRunStartedAt` (human authority guard, normative-spec §3.6 line 440 — stale\_proposal reject path) — capture it before the AI call and thread through. ### Step 1: Map test scenarios | # | Scenario | Expected | | - | ------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------- | | 1 | TS compile — `analysis.actionNeeded` at line 325 no longer resolves | `tsc --noEmit` reports error if line not deleted | | 2 | `aiRunStartedAt` captured before `await config.aiClient.analyze(...)` | Plumbed into `writer.writeAnalysisWithTasks({...})` as new param | | 3 | `writeAnalysisWithTasks` propagates `aiRunStartedAt` to `applyTaskAction()` ctx (Task 3 already inlines `new Date()` — replace with the threaded value) | Stale proposal guard fires on integration test scenario where task UPDATE between aiRunStartedAt and the batch commit | ### Step 2: Implementation Edit lines 300-330 of `handler.ts`: ```typescript * validated, so this is a constant-time pass-through. */ const aiRunStartedAt = new Date(); const aiResult = await config.aiClient.analyze(ctx.systemPrompt, userMessage, ctx.zodSchema); const analysis = ContactsAnalysisSchema.parse(aiResult.output); const contactAnalysisRunId = messageId; await ctx.writer.writeAnalysisWithTasks({ phone: contact.phone, franchiseId: contact.franchiseId, accountId: contact.accountId, analysis, contactAnalysisRunId, aiRunStartedAt, previousLifecycle: { stage: contact.currentLifecycleStage, state: contact.currentLifecycleState, }, previousLeadStatus: contact.currentLeadStatus, pendingTasks: contactTasks.pending, taskDecisions: analysis.taskDecisions, storePhone, storeId, }); logger.info('Contact analyzed', { phone: contact.phone, // actionNeeded log field dropped — derived field no longer in // ContactsAnalysisSchema (Task 1). Replace with task decision summary. taskDecisionCount: analysis.taskDecisions.length, leadStatus: analysis.leadStatus, inputTokens: aiResult.inputTokens, outputTokens: aiResult.outputTokens, }); ``` Then update `writeAnalysisWithTasks` signature (in `neon-repository.ts`, \~line 419) to accept `aiRunStartedAt`: ```typescript async function writeAnalysisWithTasks(params: { phone: string; franchiseId: string; accountId: string; analysis: ContactsAnalysis; contactAnalysisRunId: string; aiRunStartedAt: Date; previousLifecycle: { stage: string | null; state: string | null }; previousLeadStatus: string | null; pendingTasks: TaskRow[]; taskDecisions: TaskDecision[]; storePhone: string | null; storeId: string | null; }): Promise { const { phone, franchiseId, accountId, analysis, contactAnalysisRunId, aiRunStartedAt, // ...rest } = params; // ... ``` In the Task 3 loop, replace `aiRunStartedAt: new Date()` (placeholder) with the threaded value: ```typescript const baseCtx = { storeId, actor: { ... }, aiRunStartedAt, // ← threaded from handler.ts db: client, }; ``` Also update the `ContactsWriter` protocol interface (likely in `src/infrastructure/protocols.ts`): ```bash grep -n "writeAnalysisWithTasks" lambda/contacts-analyzer/src/infrastructure/protocols.ts ``` Add `aiRunStartedAt: Date` to the protocol signature for symmetry. Implementer locates and edits. ### Step 3: Test write + run ```bash pnpm run typecheck bun test # Expected: 0 type errors. Task 1 unit tests still pass. ``` ### Step 4: Commit ```bash git add lambda/contacts-analyzer/src/handler.ts lambda/contacts-analyzer/src/infrastructure/neon-repository.ts lambda/contacts-analyzer/src/infrastructure/protocols.ts git commit -m "feat(contacts-analyzer): capture aiRunStartedAt + drop actionNeeded log Phase 1 Plan 03 Task 6. - aiRunStartedAt = new Date() captured before AI analyze() call; threaded through writeAnalysisWithTasks → applyTaskAction() ctx for stale_proposal reject path (normative-spec §3.6 line 440 — human authority guard). - analysis.actionNeeded log field dropped (Task 1 removed actionNeeded from ContactsAnalysisSchema); replaced with taskDecisionCount summary. - ContactsWriter protocol signature extended; protocols.ts updated. Spec: normative-spec.md §3.6 + §4d. " ``` *** ## Task 7: Integration test — 5 scenarios from §5.2 contacts-analyzer row **Files:** - Create: `lambda/contacts-analyzer/tests/integration/task-decisions.test.ts` - Verify: `package.json` has a `test:integration` script that points to this folder + targets test Neon (`NEON_TEST_URL` env var or whichever convention the lambda uses) **Why this task:** §5.2 of normative-spec lists 5 contacts-analyzer integration scenarios (line 724). These are the canonical end-to-end verification that the migration is correct. Cannot be replaced by unit tests — they need a real DB to verify SQL-layer guards (partial unique, conditional WHERE, ON CONFLICT, CTE). ### Step 1: Map test scenarios Mirror normative-spec §5.2 row 4 exactly: | # | Scenario | Setup | Action | Expected | | - | ---------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | 1 | **Mixed action atomic batch** — AI taskDecisions\[] contains `create_open / close / record_progress / reopen` mixed | Seed contact (phone=`+15551234567`, store=`store-A`, dnc=false) + 1 open task `T1` for typeCategory `lead_follow_up` + 1 closed task `T2` (eligible reopen). | Call `writeAnalysisWithTasks` with 4 decisions: `create_open(typeCategory='cancellation_risk')`, `close(T1)`, `record_progress(T1)` → wait, T1 just closed — bad example; instead: `record_progress(T1)` BEFORE close. Order: `[record_progress(T1, no_answer), create_open(cancellation_risk), close(T1, converted), reopen(T2, "follow-up needed")]` | All 4 statements committed atomically; tasks: T1 closed + 1 progress event row; new `cancellation_risk` open task; T2 status='open'. `tasksClosedCount=1, tasksCreatedCount=1, tasksUpdatedCount=2`. | | 2 | **Hallucinated taskId** — AI emits `close(taskId='nonexistent-uuid')` mixed with valid `create_open` | Seed contact + 0 tasks. | 2 decisions: `close('00000000-0000-0000-0000-000000000000')` + `create_open(lead_follow_up)`. | `close` rejected `hallucinated_task_id`; `create_open` succeeds. 1 task in DB. logger.warn captured for the reject. | | 3 | **Low confidence reject(module-only in Phase 1)** | Covered by Plan 02 `checkConfidence` / Orchestrator tests. | contacts-analyzer AI prompt does not surface per-decision confidence today. | SKIP in Plan 03 caller integration; Phase 2 wires prompt confidence into TaskDecision adapter. | | 4 | **typeCategory invalid for AI on churned-active** — contact lifecycleStage=`churned`, state=`active`, AI emits `create_open(typeCategory='lead_outreach')` | Seed contact with `lifecycleStage='churned', lifecycleState='active', leadStatus=null`. | 1 decision: `create_open(typeCategory='lead_outreach', ...)`. | reject `invalid_type_category` (lead\_outreach is lead-tracking-only per `computeAllowedTypeCategories` spec §3.6 line 469). 0 tasks created. | | 5 | **create\_closed with already-open same typeCategory task** — | Seed contact + 1 open task typeCategory=`cancellation_risk`. | 1 decision: `create_closed(typeCategory='cancellation_risk', sourceCallId='CALL_X', closeResult='cancel_saved')`. | reject `create_closed_with_open_task` (spec §3.6 line 442 — caller should use `close` not `create_closed`). 0 new tasks; the existing open task untouched. | ### Step 2: Implementation ```typescript // lambda/contacts-analyzer/tests/integration/task-decisions.test.ts import { describe, test, expect, beforeEach, afterAll } from 'bun:test'; import { createNeonRepository } from '../../src/infrastructure/neon-repository'; import { createDrizzleClient, contacts, tasks, taskProgressEvents, contactTimeline } from '@retaintive/common/db'; import { and, eq, sql } from 'drizzle-orm'; import type { ContactsAnalysis } from '../../src/core/models'; import { randomUUID } from 'node:crypto'; const TEST_DB_URL = process.env['NEON_TEST_URL']; if (!TEST_DB_URL) throw new Error('NEON_TEST_URL not set'); const db = createDrizzleClient({ databaseUrl: TEST_DB_URL }); const writer = createNeonRepository(/* test SSM path */ 'test-ssm-param'); // (impl note: createNeonRepository takes SSM param; for integration test // either expose an `__internal__createWithClient(db)` constructor or set // up a mocked SSM. Plan 02 integration test follows the same pattern.) const TEST_STORE_ID = 'integration-test-store-A'; const TEST_FRANCHISE = 'integration-test-franchise'; const TEST_ACCOUNT = 'integration-test-account'; async function seedContact(phone: string, doNotContact = false, lifecycleStage = 'lead', lifecycleState = 'active') { await db.insert(contacts).values({ phone, storeId: TEST_STORE_ID, franchiseId: TEST_FRANCHISE, accountId: TEST_ACCOUNT, doNotContact, lifecycleStage, lifecycleState, }).onConflictDoNothing(); } async function seedTask(phone: string, typeCategory: string, status: 'open' | 'closed' = 'open') { const taskId = randomUUID(); await db.insert(tasks).values({ taskId, contactPhone: phone, storeId: TEST_STORE_ID, franchiseId: TEST_FRANCHISE, accountId: TEST_ACCOUNT, typeCategory, taskType: typeCategory === 'lead_outreach' ? 'lead_outreach' : 'follow_up', status, priority: 'medium', suggestedActions: [{ action: 'Call', reason: 'r', priority: 'medium', priorityReason: 'r' }], }); return taskId; } async function cleanup(phone: string) { await db.delete(contactTimeline).where(eq(contactTimeline.contactPhone, phone)); await db.delete(taskProgressEvents).where(eq(taskProgressEvents.contactPhone, phone)); await db.delete(tasks).where(eq(tasks.contactPhone, phone)); await db.delete(contacts).where(and(eq(contacts.phone, phone), eq(contacts.storeId, TEST_STORE_ID))); } const baseAnalysis: ContactsAnalysis = { customerSummary: 'Test summary', suggestedActions: [], leadStatus: 'new', goals: [], leadObjections: [], leadRejectionReasons: [], lifecycleStage: 'lead', lifecycleState: 'active', taskDecisions: [], }; describe('contacts-analyzer integration — taskDecisions migration', () => { test('Scenario 1: mixed action atomic batch', async () => { const phone = '+15551110001'; await seedContact(phone); const T1 = await seedTask(phone, 'lead_follow_up', 'open'); const T2 = await seedTask(phone, 'cancellation_risk', 'closed'); await writer.writeAnalysisWithTasks({ phone, franchiseId: TEST_FRANCHISE, accountId: TEST_ACCOUNT, analysis: { ...baseAnalysis, taskDecisions: [ { action: 'record_progress', taskId: T1, progressType: 'no_answer', channel: 'phone', callId: 'CALL_1', reason: 'no answer' }, { action: 'create_open', typeCategory: 'cancellation_risk', suggestedActions: [{ action: 'Call', reason: 'r', priority: 'high', priorityReason: 'r' }], reason: 'New risk' }, // ↑ wait: T2 is closed cancellation_risk — partial unique allows // 1 open per (phone, store, typeCategory). create_open would fail // if seedTask closed status_open. Re-baseline this test. { action: 'close', taskId: T1, closeResult: 'converted', reason: 'converted on call' }, { action: 'reopen', taskId: T2, reason: 'follow-up' }, ], }, contactAnalysisRunId: randomUUID(), aiRunStartedAt: new Date(), previousLifecycle: { stage: 'lead', state: 'active' }, previousLeadStatus: null, pendingTasks: [], storePhone: null, storeId: TEST_STORE_ID, }); // Verify final state const taskRows = await db.select().from(tasks).where(eq(tasks.contactPhone, phone)); const T1Row = taskRows.find(t => t.taskId === T1); expect(T1Row?.status).toBe('closed'); expect(T1Row?.closeResult).toBe('converted'); const T2Row = taskRows.find(t => t.taskId === T2); expect(T2Row?.status).toBe('open'); const progressEvents = await db.select().from(taskProgressEvents).where(eq(taskProgressEvents.taskId, T1)); expect(progressEvents.length).toBe(1); expect(progressEvents[0]?.progressType).toBe('no_answer'); await cleanup(phone); }); test('Scenario 2: hallucinated taskId — skip + continue other decisions', async () => { const phone = '+15551110002'; await seedContact(phone); await writer.writeAnalysisWithTasks({ phone, franchiseId: TEST_FRANCHISE, accountId: TEST_ACCOUNT, analysis: { ...baseAnalysis, taskDecisions: [ { action: 'close', taskId: '00000000-0000-0000-0000-000000000000', typeCategory: 'lead_follow_up', closeResult: 'converted', reason: 'hallucinated' }, { action: 'create_open', typeCategory: 'lead_follow_up', suggestedActions: [{ action: 'Call', reason: 'r', priority: 'medium', priorityReason: 'r' }], reason: 'new task' }, ], }, contactAnalysisRunId: randomUUID(), aiRunStartedAt: new Date(), previousLifecycle: { stage: 'lead', state: 'active' }, previousLeadStatus: null, pendingTasks: [], storePhone: null, storeId: TEST_STORE_ID, }); const tasksAfter = await db.select().from(tasks).where(eq(tasks.contactPhone, phone)); // Only the create_open survived; the hallucinated close was rejected silently expect(tasksAfter.length).toBe(1); expect(tasksAfter[0]?.typeCategory).toBe('lead_follow_up'); expect(tasksAfter[0]?.status).toBe('open'); await cleanup(phone); }); test('Scenario 3: low confidence reject', async () => { // NOTE: TaskDecision schema (Task 1) doesn't surface AI confidence // to the caller today. This scenario is a placeholder until Phase 2 // wires per-decision confidence in the AI prompt. Plan 02 Task 2 // checkConfidence still exists; this test verifies the path is // reachable via direct TaskAction construction (bypassing TaskDecision // narrowing — test-only). // // Implementer: either (a) extend TaskDecision schema to accept // optional confidence + thread through buildTaskAction, OR // (b) skip this test in Phase 1 with .skip() + open issue to revisit // in Phase 2. Recommend (b) — adding confidence to TaskDecision is // a prompt engineer concern, not Plan 03. }); test('Scenario 4: invalid typeCategory for AI (churned active → lead_outreach reject)', async () => { const phone = '+15551110004'; await seedContact(phone, false, 'churned', 'active'); await writer.writeAnalysisWithTasks({ phone, franchiseId: TEST_FRANCHISE, accountId: TEST_ACCOUNT, analysis: { ...baseAnalysis, lifecycleStage: 'churned', lifecycleState: 'active', taskDecisions: [ { action: 'create_open', typeCategory: 'lead_outreach', suggestedActions: [{ action: 'Call', reason: 'r', priority: 'medium', priorityReason: 'r' }], reason: 'cross-stage attempt' }, ], }, contactAnalysisRunId: randomUUID(), aiRunStartedAt: new Date(), previousLifecycle: { stage: 'churned', state: 'active' }, previousLeadStatus: null, pendingTasks: [], storePhone: null, storeId: TEST_STORE_ID, }); const tasksAfter = await db.select().from(tasks).where(eq(tasks.contactPhone, phone)); expect(tasksAfter.length).toBe(0); // invalid_type_category — nothing created await cleanup(phone); }); test('Scenario 5: create_closed with already-open same typeCategory rejects', async () => { const phone = '+15551110005'; await seedContact(phone); const existing = await seedTask(phone, 'cancellation_risk', 'open'); await writer.writeAnalysisWithTasks({ phone, franchiseId: TEST_FRANCHISE, accountId: TEST_ACCOUNT, analysis: { ...baseAnalysis, taskDecisions: [ { action: 'create_closed', typeCategory: 'cancellation_risk', suggestedActions: [], sourceCallId: 'CALL_X', closeResult: 'cancel_saved', reason: 'should reject' }, ], }, contactAnalysisRunId: randomUUID(), aiRunStartedAt: new Date(), previousLifecycle: { stage: 'lead', state: 'active' }, previousLeadStatus: null, pendingTasks: [], storePhone: null, storeId: TEST_STORE_ID, }); const tasksAfter = await db.select().from(tasks).where(eq(tasks.contactPhone, phone)); // Only the pre-existing open task; nothing created expect(tasksAfter.length).toBe(1); expect(tasksAfter[0]?.taskId).toBe(existing); expect(tasksAfter[0]?.status).toBe('open'); await cleanup(phone); }); afterAll(async () => { // Belt-and-suspenders cleanup for (const phone of ['+15551110001', '+15551110002', '+15551110004', '+15551110005']) { await cleanup(phone); } }); }); ``` ### Step 3: Run ```bash cd /Users/maxwsy/workspace/callytics-infrastructure/lambda/contacts-analyzer NEON_TEST_URL=$(aws ssm get-parameter --name /test/neon/db-url --with-decryption --query 'Parameter.Value' --output text) \ bun test tests/integration/task-decisions.test.ts ``` Expected: 4 PASS + 1 SKIP (Scenario 3 confidence — Phase 2 follow-up). ### Step 4: Commit ```bash git add lambda/contacts-analyzer/tests/integration/task-decisions.test.ts git commit -m "test(contacts-analyzer): integration tests for Phase 1 taskDecisions migration Phase 1 Plan 03 Task 7. 5 scenarios per normative-spec §5.2 contacts-analyzer row: - mixed action atomic batch (create_open / close / record_progress / reopen) - hallucinated taskId reject, others continue - low confidence reject (SKIPPED — Phase 2 prompt surface needed) - invalid_type_category for AI on churned-active (lead_outreach blocked) - create_closed with existing open same-typeCategory reject Runs against test Neon via NEON_TEST_URL env var. Spec: normative-spec.md §5.2 + §5.0 invariants. " ``` *** ## Task 8: Update prompt-builder.ts — remove `actionNeeded` references + add 6-action vocabulary docs **Files:** - Modify: `lambda/contacts-analyzer/src/core/prompt-builder.ts` (lines 57, 135, 224, 392, 399-417, 434, 604-661, 679-680, 748) **Why this task:** the prompt template still tells AI to output `actionNeeded` and uses the old 3-action vocabulary in the schema description. Without updating the prompt, AI will keep emitting `actionNeeded` (Zod will silently drop it via .strip() default; OK functionally) but also keep emitting `action: 'create'` (Task 1's `.transform()` shim catches that). Cleaner: update the prompt now so prompt engineer's rewrite + this migration ship coherently. ### Step 1: Map test scenarios Prompt-builder is text generation; tests are existence-of / absence-of grep checks: | # | Check | Expected | | - | ----------------------------------------------- | ------------------------------------------------------------ | | 1 | grep `actionNeeded` in prompt-builder.ts output | 0 hits after edit | | 2 | grep `actionNeededReason` | 0 hits | | 3 | grep `"action": "create"` template string | 0 hits | | 4 | grep `"action": "create_open"` | ≥ 1 hit (new vocabulary documented in OUTPUT SCHEMA section) | | 5 | grep `"action": "create_closed"` | ≥ 1 hit | | 6 | grep `"action": "record_progress"` | ≥ 1 hit | | 7 | grep `"action": "reopen"` | ≥ 1 hit | | 8 | TS compile | clean | | 9 | snapshot test (if existing) | re-baselined intentionally to new schema text | ### Step 2: Implementation Read prompt-builder.ts (especially lines 57-100 OVERVIEW section, lines 392-440 OUTPUT SCHEMA section, lines 604-665 taskDecisions section, lines 679-680 example JSON) and: 1. **Delete `actionNeeded` mentions** in OVERVIEW (line 57), in the lifecycle terminal rule (line 224), and in OUTPUT SCHEMA (lines 399-417). 2. **Replace 3-action vocabulary** at the `taskDecisions` schema description (around line 604-661) with 6-action vocabulary. Suggested text: ``` - taskDecisions: Array of per-task decisions. Each element is one of 6 action types. Use these to direct the lifecycle of follow-up tasks; the application code applies them via a shared state-machine (Policy Guard rejects bad inputs). Action types: * create_open — create a new task in status='open' for future human follow-up. REQUIRED: typeCategory, suggestedActions[], reason. * create_closed — log a new task that was settled in the call (e.g. cancellation saved on the same call). REQUIRED: typeCategory, suggestedActions[], sourceCallId (the call where it was settled — used for dedupe), closeResult, reason. Use this rarely — most settled-on-call outcomes should `close` an EXISTING open task instead. * close — close an existing open task. REQUIRED: taskId, closeResult, reason. * update — patch an existing open task's suggestedActions. REQUIRED: taskId, suggestedActions, reason. Do NOT emit task-level priority — the application derives task priority from max(suggestedActions[].priority). * record_progress — log progress on an existing open task without closing it. Use for: no_answer, left_voicemail, text_sent, callback_requested, follow_up_scheduled, customer_considering. REQUIRED: taskId, progressType, channel ('phone'|'sms'|'voicemail'|'email'), at least one of callId / messageId (for dedupe), reason. * reopen — flip a closed task back to open (rare; only when new evidence in this analysis run invalidates the prior close). REQUIRED: taskId, reason. ``` 3. **Replace example JSON** (lines 679-680) to show the 6-action vocabulary: ```json { "customerSummary": "...", "suggestedActions": [...], ... "taskDecisions": [ { "action": "create_open", "typeCategory": "lead_follow_up", "suggestedActions": [...], "reason": "..." }, { "action": "close", "taskId": "11111111-1111-1111-1111-111111111111", "closeResult": "converted", "reason": "..." }, { "action": "record_progress", "taskId": "22222222-2222-2222-2222-222222222222", "progressType": "no_answer", "channel": "phone", "callId": "abc", "reason": "..." } ] } ``` 4. **Update terminal-lifecycle rule** (line 748) — remove `actionNeeded = false` clause; replace with `taskDecisions = []` (consistent rule: no proactive outreach). ### Step 3: Test write + run ```bash cd /Users/maxwsy/workspace/callytics-infrastructure/lambda/contacts-analyzer # Sanity grep grep -c 'actionNeeded' src/core/prompt-builder.ts # Expect 0 grep -c '"action": "create"' src/core/prompt-builder.ts # Expect 0 grep -c '"action": "create_open"' src/core/prompt-builder.ts # Expect >= 1 grep -c '"action": "record_progress"' src/core/prompt-builder.ts # Expect >= 1 pnpm run typecheck bun test # all unit/integration tests still pass ``` If a snapshot test for prompt-builder exists (`tests/unit/prompt-builder.test.ts`?), re-baseline intentionally: ```bash bun test --update-snapshots ``` ### Step 4: Commit ```bash git add lambda/contacts-analyzer/src/core/prompt-builder.ts git commit -m "feat(contacts-analyzer): prompt — 6-action vocabulary + drop actionNeeded Phase 1 Plan 03 Task 8. - OUTPUT SCHEMA section: actionNeeded / actionNeededReason removed (derived from open-task EXISTS query downstream, normative-spec §4g). - taskDecisions section: 3-action → 6-action vocabulary with payload contracts for create_open / create_closed / close / update / record_progress / reopen. - Example JSON re-baselined to 6-action shape. - Terminal-lifecycle rule: 'actionNeeded=false' replaced with 'taskDecisions=[]' (same semantic, new vocabulary). - TaskDecision schema (Task 1) still accepts legacy 'create' via .transform() shim — this PR can ship before prompt engineer's rewrite without breaking AI output. Spec: normative-spec.md §4d + Appendix A (transitional shim). " ``` *** ## Task 9 (optional): Type-check + tests + PR **Files:** - None to edit; full repo verification. ### Step 1: Full verification ```bash cd /Users/maxwsy/workspace/callytics-infrastructure # Type-check all packages pnpm -r typecheck # Unit tests cd lambda/contacts-analyzer && bun test tests/unit/ # Integration (requires NEON_TEST_URL) NEON_TEST_URL=... bun test tests/integration/ # Lint pnpm -r lint ``` Expected: all PASS. ### Step 2: Open PR ```bash git push -u origin gh pr create --title "feat(contacts-analyzer): Phase 1 Plan 03 — migrate to @retaintive/common/domain" --body-file <(cat << 'EOF' ## Plan 03: contacts-analyzer caller migration Migrates `lambda/contacts-analyzer` from inline Drizzle SQL to `@retaintive/common@1.2.0/domain` shared modules. ### What changes - **TaskDecision Zod schema** (`src/core/models.ts`) — 3 action union → 6 action union (`create_open` / `create_closed` / `close` / `update` / `record_progress` / `reopen`). Legacy `create` accepted via `.transform()` shim. `actionNeeded` / `actionNeededReason` removed from `ContactsAnalysisSchema` (derived via EXISTS open-task query). - **`writeAnalysisWithTasks` loop** (`src/infrastructure/neon-repository.ts`) — 332-line inline mutation loop replaced by 80-line dispatch to `applyTaskAction()`. Identity UPSERT moves to `upsertIdentity()`. DNC cascade moves to `closeAllOpenForContact()`. Lifecycle + analysis-completed timeline writes move to `writeTimelineEvent()`. - **Handler** (`src/handler.ts`) — captures `aiRunStartedAt` for stale-proposal guard; drops `actionNeeded` log field. - **Prompt** (`src/core/prompt-builder.ts`) — 6-action vocabulary; example JSON re-baselined. - **Dependency** — `@retaintive/common` 1.0.x → 1.2.0. ### Spec source - `docs/product-design/v2/unified-pipeline/implementation-plan/normative-spec.md` §4d (caller change table) + §2 (transitional 3-value status enum) + §5.2 (integration test scenarios) + §3.6 (Policy Guard checks). ### Test plan - Unit (`tests/unit/models.test.ts`): 11 TaskDecision schema cases incl. legacy `create` shim normalization, S5 sourceCallId enforcement, record_progress idempotency-evidence superRefine. - Unit (`tests/unit/build-task-action.test.ts`): 6 cases — `TaskDecision` → `TaskAction` payload 1:1 mapping. - Integration (`tests/integration/task-decisions.test.ts`): 5 scenarios per §5.2 — mixed action atomic batch, hallucinated taskId skip, invalid_type_category churned-active, create_closed-with-open-task reject. Confidence reject scenario SKIPPED pending Phase 2 prompt surface. - All existing unit tests for `writeAnalysisWithTasks` re-baselined to new statement shape. ### Risk - Statement count per batch changes (Orchestrator may emit different INSERT / UPDATE structure than inline) — verified atomic via integration test. - Prompt engineer rewrite is decoupled: legacy `create` shim absorbs the gap until their PR ships. - Local `dnc-cascade.ts` helper is no longer imported from this lambda; other callers (lead-processor, message-processor) own their own migration in Plans 04-05. EOF ) ``` *** ## Self-Review Checklist ### Spec coverage (normative-spec §4d row 1) - ✅ TaskDecision 遍历 → `applyTaskAction()` loop (Task 3) - ✅ status `'pending'` → `'open'` 写入 final value (Task 3 — Orchestrator owns `create_open` SQL builder which writes `'open'`; transitional read paths still use `IN ('pending','open')` per §6.2) - ✅ `derivedTaskType` (line 692-772) deleted; Orchestrator legacy shim (Task 3) - ✅ Trust score CASE WHEN (line 396-398) → `ContactWriter.upsertIdentity()` (Task 4) - ✅ Inline `buildTimelineValues` → Orchestrator + `TimelineWriter` (Tasks 3 + 5) ### Spec coverage (TaskDecision Zod schema, §4d footnotes) - ✅ 6 action union (Task 1) - ✅ `.transform()` legacy `create` → `create_open` shim (Task 1) - ✅ `create_closed` payload requires `sourceCallId` — Zod required field (Task 1, case 4 test) - ✅ Task-level `priority` deleted from per-decision payload — Orchestrator derives (Task 1) - ✅ `actionNeeded` boolean deleted from `ContactsAnalysisSchema` (Task 1) - ✅ `lifecycleState` retained — engineer feedback per spec §2.3 (Task 1) ### Spec coverage (§4d row 4 handler + §4d row 5 prompt) - ✅ `handler.ts` line 325 — `actionNeeded` log field dropped (Task 6) - ✅ `prompt-builder.ts` — `actionNeeded` / `actionNeededReason` mentions removed (Task 8) - ✅ `prompt-builder.ts` — 6-action vocabulary documented (Task 8) ### Type consistency - ✅ `TaskAction` from `@retaintive/common/domain` matches `TaskDecision` from `src/core/models.ts` 1:1 by `action` literal — `buildTaskAction()` helper enforces (Task 3) - ✅ `applyTaskAction()` signature matches Plan 02 Task 3 — `(action, ctx) → Promise` - ✅ Import path `@retaintive/common/domain` (Task 2 dep bump unlocks) - ✅ `aiRunStartedAt` threaded handler → repository → Orchestrator ctx (Task 6) ### No placeholders - ✅ Actual file paths + line ranges throughout (lines 559-892, lines 467-540, etc.) - ✅ Actual code (TS + SQL), not truncated — TaskDecision union body included verbatim; orchestrator loop replacement body included verbatim - ✅ Commit messages full - ✅ Import paths actual (`@retaintive/common/domain` — verified via Plan 02 Task 7 `package.json` exports map) - ⚠️ Integration test Scenario 3 (low confidence) deliberately SKIPPED + documented — Phase 2 prompt surface concern, not Plan 03 ### Outstanding - Confidence path activation deferred to Phase 2 (TaskDecision schema doesn't carry per-decision confidence; AI prompt would need to surface it). Plan 02 Orchestrator's `checkConfidence` is still implemented and reachable; Plan 03 just doesn't exercise the path from AI input. - Once prompt engineer ships the rewrite (drops `action='create'` entirely), the `.transform()` shim in `TaskDecision` (Task 1) can be deleted in a follow-up PR — tracked in spec Appendix A. *** ## Execution Handoff **Option 1: Subagent-Driven** — fresh subagent per Task, user reviews unit test coverage + grep checks (Task 8) + integration test runs (Task 7) between tasks. **Option 2: Inline** — `superpowers:executing-plans` runs Tasks 1-8 sequentially, checkpoint between Tasks 3 and 4 (structural diff worth a human eyeball before continuing). Plan 03 size is mid-pack (smaller than Plan 02; bigger than Plan 01). Recommend Option 1 — Tasks 1, 3, 4 each benefit from focused review; Tasks 2, 5, 6, 7, 8 can chain.