#Plan 03 — contacts-analyzer Caller Migration Plan

// ── 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<typeof TaskDecision>;

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.'),
});

// 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
});

      // ── ④ 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<string>(
        taskDecisions
          .filter((d): d is Extract<TaskDecision, { action: 'close' }> => 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.)
       */

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,
        },
      };
  }
}

// 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

      // ── ②ʹ 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))),
      );

      // ── 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,
            });
          }
        }
      }

      // ── 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',
          }),
        );
      }

      // ── 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',
        }),
      );

   * 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,
  });

  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<void> {
    const {
      phone,
      franchiseId,
      accountId,
      analysis,
      contactAnalysisRunId,
      aiRunStartedAt,
      // ...rest
    } = params;
    // ...

        const baseCtx = {
          storeId,
          actor: { ... },
          aiRunStartedAt, // ← threaded from handler.ts
          db: client,
        };

// 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);
    }
  });
});