PM-6643 - let's talk automated ai classification

docs/api.md

@@ -145,6 +145,15 @@ They do not grant form management privileges.
 | `GET /forms/:key/submissions?limit=25&after=uuid&startDate=2026-09-01&endDate=2026-09-30` | `{ form, columns, labels, data, total, nextCursor }` across all published/retired revisions. |
 | `GET /forms/:key/submissions/export?startDate=2026-09-01&endDate=2026-09-30` | Complete streaming CSV with one header, independently of the table page. Omit both dates for all data. |
 
+**AI classification.** For a form with an `ai-classify` processor flow (Let's talk),
+both routes append `classification`, `classification_status`,
+`classification_confidence`, `classification_needs_review` and
+`classification_run_id`, after the answer columns. A submission never classified
+reports `classification: "UNKNOWN"` with null details. Both accept an optional
+`classification` label filter (`UPPER_SNAKE`, e.g. `?classification=SALES_LEAD`);
+`UNKNOWN` also matches unclassified submissions. The filter is ignored for forms
+without classification. See forms-processor-v6 ADR 0001.
+
 Dates are optional **inclusive UTC calendar dates** in `YYYY-MM-DD`; an end date
 includes timestamps through 23:59:59.999. Invalid calendar dates and reversed
 ranges return 400. Existing version-specific JSON/CSV endpoints accept these same
@@ -158,7 +167,7 @@ then descending UUID for stable ties across cursor pages; `total` counts matchin
 before pagination. Decimal values remain strings and multi-select values remain
 arrays. A field reused with a different type retains each revision's original value.
 
-The complete export accepts only date filters, not `limit` or `after`. It fetches
+The complete export accepts only date and classification filters, not `limit` or `after`. It fetches
 1,000 rows per batch, applies backpressure, stops when the client disconnects, and
 uses the established CSV escaping/formula protection. Columns and an upper
 submission timestamp are captured at export start. This is not a repeatable-read

docs/data-model.md

@@ -18,6 +18,7 @@ erDiagram
     FormField ||--o{ FieldOption : choices
     FormVersion ||--o{ Submission : receives
     Submission ||--o| SubmissionEvent : delivery
+    Submission ||--o| SubmissionClassification : classification
     Submission ||--o{ Answer : contains
     FormField ||--o{ Answer : constrains
     Answer ||--o{ AnswerSelection : selects
@@ -34,6 +35,7 @@ erDiagram
 | `FieldOption`     | Named option key and label, owned by a field/version.                                                                                                               |
 | `Submission`      | Exact version, server timestamp, UUID idempotency key, canonical request hash, verified member ID, and optional source page path.                                   |
 | `SubmissionEvent` | Optional one-to-one delivery receipt with submission UUID and nullable publication timestamp; cascades on submission deletion. |
+| `SubmissionClassification` | Optional one-to-one AI classification written by forms-processor-v6: label (`UPPER_SNAKE` string, not an enum, because the class set is configured in tc-ai-api), status (`PENDING`/`CLASSIFIED`/`FAILED`), confidence 0–1, needs-review flag, language, rationale, tc-ai-api workflow run id, attempts and last error. CHECK constraints enforce the label format, status values, confidence range and complete `CLASSIFIED` rows. Cascades on submission deletion. No row reports as `UNKNOWN`. |
 | `Answer`          | One scalar value in a type-specific column, or a multi-select answer parent; unique per submission/field.                                                           |
 | `AnswerSelection` | One row per selected option, with composite ownership foreign keys and duplicate prevention.                                                                        |
 
@@ -126,6 +128,12 @@ Flow additions/rule edits apply to events not yet routed; replay requires explic
 clearing `routedAt` and never deletes existing delivery receipts. Bus API email-event acceptance
 and the receipt cannot be atomic; a crash in between can cause a duplicate email.
 
+Flows whose rules carry a `classification` label list (`{"all":[],"classification":["SALES_LEAD"]}`)
+are routed by the processor only after the submission is classified, and only while
+enabled. Migration `20261022010000_lets_talk_classification` seeds the Let's talk
+classify and label flows disabled; forms-processor-v6 ADR 0001 and its
+`examples/lets-talk-shadow-mode.sql` / `lets-talk-cutover.sql` describe the rollout.
+
 Processor events contain answers and must be included in personal-data retention
 and erasure procedures separately from `Submission`. Deleting a ProcessorEvent
 cascades its delivery records and removes deduplication protection, so retain it

docs/lets-talk.md

@@ -80,3 +80,13 @@ The website maintains its analytics adapter around the same rendering contract.
    `form.submitted` and `external.action.email`, and verify the received email.
    Processor `deliveredAt` means Bus API acceptance; verify provider/inbox delivery
    separately. No historical HubSpot data is imported.
+
+## AI classification (forms-processor-v6 ADR 0001)
+
+Migration `20261022010000_lets_talk_classification` adds
+`forms."SubmissionClassification"` and seven disabled processor flows: one classifies
+each submission with the tc-ai-api `lets-talk-classification` workflow, and the
+others reply or forward by label. It does not change `lets-talk-sales-email`. Once the
+processor enables classification, the reports UI and CSV show a `classification`
+column (`UNKNOWN` until classified) and accept a `classification` filter. The rollout
+order and SQL are in forms-processor-v6 `deploy/README.md`.

prisma/migrations/20261022010000_lets_talk_classification/migration.sql

@@ -0,0 +1,42 @@
+-- AI classification of form submissions, written by forms-processor-v6 (its ADR 0001).
+-- One row per classified submission; a submission without a row reports as UNKNOWN.
+-- label is VARCHAR, not an enum: the class set is configurable in tc-ai-api.
+CREATE TABLE forms."SubmissionClassification" (
+  "submissionId" UUID PRIMARY KEY REFERENCES forms."Submission"(id) ON DELETE CASCADE,
+  label VARCHAR(40) NOT NULL DEFAULT 'UNKNOWN',
+  status VARCHAR(20) NOT NULL DEFAULT 'PENDING',
+  confidence DECIMAL(5, 4),
+  "needsReview" BOOLEAN,
+  "detectedLanguage" VARCHAR(8),
+  rationale VARCHAR(500),
+  "workflowId" VARCHAR(100) NOT NULL,
+  "workflowRunId" UUID,
+  attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0),
+  "lastError" VARCHAR(100),
+  "classifiedAt" TIMESTAMPTZ(3),
+  "createdAt" TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  "updatedAt" TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  CONSTRAINT "SubmissionClassification_label_format" CHECK (label ~ '^[A-Z][A-Z0-9_]{0,39}$'),
+  CONSTRAINT "SubmissionClassification_status" CHECK (status IN ('PENDING', 'CLASSIFIED', 'FAILED')),
+  CONSTRAINT "SubmissionClassification_confidence_range" CHECK (confidence IS NULL OR (confidence >= 0 AND confidence <= 1)),
+  -- A CLASSIFIED row always carries the model's verdict.
+  CONSTRAINT "SubmissionClassification_classified_complete" CHECK (
+    status <> 'CLASSIFIED' OR (confidence IS NOT NULL AND "needsReview" IS NOT NULL AND "classifiedAt" IS NOT NULL)
+  )
+);
+CREATE INDEX "SubmissionClassification_label_idx" ON forms."SubmissionClassification" (label);
+
+-- Let's talk flows, all disabled. Recipients, templates and the AI endpoint come
+-- from processor SSM settings, so settings stay empty here. Flows with a
+-- "classification" rule are routed only after the submission is classified.
+-- lets-talk-sales-email is intentionally unchanged: it keeps emailing every
+-- submission until the reviewed cut-over (processor examples/lets-talk-cutover.sql).
+INSERT INTO forms."ProcessorFlow" (id, "formKey", action, rules) VALUES
+  ('lets-talk-classify', 'lets_talk', 'ai-classify', '{"all":[]}'),
+  ('lets-talk-community-opportunity-reply', 'lets_talk', 'sendgrid-email', '{"all":[],"classification":["COMMUNITY_OPPORTUNITY"]}'),
+  ('lets-talk-community-skills-reply', 'lets_talk', 'sendgrid-email', '{"all":[],"classification":["COMMUNITY_SKILLS"]}'),
+  ('lets-talk-community-general-reply', 'lets_talk', 'sendgrid-email', '{"all":[],"classification":["COMMUNITY_GENERAL"]}'),
+  ('lets-talk-wipro-queries-reply', 'lets_talk', 'sendgrid-email', '{"all":[],"classification":["WIPRO_QUERIES"]}'),
+  ('lets-talk-support-forward', 'lets_talk', 'sendgrid-email', '{"all":[],"classification":["COMMUNITY_SUPPORT"]}'),
+  ('lets-talk-sales-reply', 'lets_talk', 'sendgrid-email', '{"all":[],"classification":["SALES_LEAD"]}')
+ON CONFLICT (id) DO NOTHING;

prisma/schema.prisma

@@ -130,6 +130,7 @@ model Submission {
   version        FormVersion @relation(fields: [versionId], references: [id], onDelete: Restrict)
   answers        Answer[]
   event          SubmissionEvent?
+  classification SubmissionClassification?
 
   @@unique([versionId, idempotencyKey])
   @@unique([id, versionId])
@@ -188,6 +189,30 @@ model SubmissionEvent {
   @@schema("forms")
 }
 
+/// AI classification of one submission, written by forms-processor-v6 (its ADR 0001).
+/// No row means UNKNOWN. label is a string: the class set is configurable in tc-ai-api.
+model SubmissionClassification {
+  submissionId     String     @id @db.Uuid
+  label            String     @default("UNKNOWN") @db.VarChar(40)
+  /// PENDING | CLASSIFIED | FAILED (SQL CHECK)
+  status           String     @default("PENDING") @db.VarChar(20)
+  confidence       Decimal?   @db.Decimal(5, 4)
+  needsReview      Boolean?
+  detectedLanguage String?    @db.VarChar(8)
+  rationale        String?    @db.VarChar(500)
+  workflowId       String     @db.VarChar(100)
+  workflowRunId    String?    @db.Uuid
+  attempts         Int        @default(0)
+  lastError        String?    @db.VarChar(100)
+  classifiedAt     DateTime?  @db.Timestamptz(3)
+  createdAt        DateTime   @default(now()) @db.Timestamptz(3)
+  updatedAt        DateTime   @default(now()) @db.Timestamptz(3)
+  submission       Submission @relation(fields: [submissionId], references: [id], onDelete: Cascade, onUpdate: NoAction)
+
+  @@index([label])
+  @@schema("forms")
+}
+
 /// Externally managed content rules and action settings. Disabled flows retain pending deliveries.
 model ProcessorFlow {
   id         String              @id @db.VarChar(100)

src/forms/dto.ts

@@ -22,6 +22,8 @@ import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
 import { FieldType, FormAccess } from '../generated/prisma/enums';
 
 export const KEY_PATTERN = /^[a-z][a-z0-9_]{0,39}$/;
+/** AI classification labels (forms."SubmissionClassification".label). */
+export const CLASSIFICATION_LABEL_PATTERN = /^[A-Z][A-Z0-9_]{0,39}$/;
 export const DECIMAL_PATTERN = /^-?(?:0|[1-9]\d{0,13})(?:\.\d{1,6})?$/;
 export const RESERVED_KEYS = new Set([
   'submission_id',
@@ -172,6 +174,15 @@ export class ReportDatesDto {
   @IsOptional()
   @Matches(/^\d{4}-\d{2}-\d{2}$/)
   endDate?: string;
+
+  @ApiPropertyOptional({
+    example: 'SALES_LEAD',
+    description:
+      'Only submissions with this AI classification label. UNKNOWN matches unclassified submissions too. Ignored for forms without AI classification.',
+  })
+  @IsOptional()
+  @Matches(CLASSIFICATION_LABEL_PATTERN)
+  classification?: string;
 }
 
 /** Defines bounded keyset pagination and inclusive UTC dates for private reporting. */

src/forms/forms.service.ts

@@ -46,6 +46,71 @@ type Definition = Prisma.FormVersionGetPayload<{
   include: typeof definitionInclude;
 }>;
 
+/** Processor action that classifies submissions (forms-processor-v6 ADR 0001). */
+const CLASSIFY_ACTION = 'ai-classify';
+const UNKNOWN_LABEL = 'UNKNOWN';
+
+/** Report columns appended for forms with AI classification, after the answer columns. */
+const CLASSIFICATION_COLUMNS = [
+  'classification',
+  'classification_status',
+  'classification_confidence',
+  'classification_needs_review',
+  'classification_run_id',
+];
+const CLASSIFICATION_LABELS: Record<string, string> = {
+  classification: 'Classification',
+  classification_status: 'Classification status',
+  classification_confidence: 'Confidence',
+  classification_needs_review: 'Needs review',
+  classification_run_id: 'AI run',
+};
+
+/**
+ * Maps a stored classification, or its absence, to report cells.
+ * @param row The submission's classification row, or null when never classified.
+ * @returns Cells keyed by CLASSIFICATION_COLUMNS; a missing row reports as UNKNOWN.
+ * @throws Does not throw.
+ */
+export function classificationCells(
+  row:
+    | {
+        label: string;
+        status: string;
+        confidence: Prisma.Decimal | null;
+        needsReview: boolean | null;
+        workflowRunId: string | null;
+      }
+    | null
+    | undefined,
+): Record<string, unknown> {
+  return {
+    classification: row?.label ?? UNKNOWN_LABEL,
+    classification_status: row?.status ?? null,
+    classification_confidence: row?.confidence?.toFixed() ?? null,
+    classification_needs_review: row?.needsReview ?? null,
+    classification_run_id: row?.workflowRunId ?? null,
+  };
+}
+
+/**
+ * Builds the relation filter for one label. UNKNOWN also matches submissions with no row.
+ * @param label Validated classification label. @returns Prisma relation filter.
+ * @throws Does not throw.
+ */
+export function classificationFilter(
+  label: string,
+): Prisma.SubmissionWhereInput {
+  return label === UNKNOWN_LABEL
+    ? {
+        OR: [
+          { classification: { is: null } },
+          { classification: { is: { label: UNKNOWN_LABEL } } },
+        ],
+      }
+    : { classification: { is: { label } } };
+}
+
 /** Manages immutable form revisions, publication, typed submission writes, optional Bus API events, and private reports. */
 @Injectable()
 export class FormsService {
@@ -140,9 +205,17 @@ export class FormsService {
       },
     });
     if (!form) throw new NotFoundException('Form not found.');
+    // Forms routed through the processor's AI classification report its label.
+    const classified =
+      (await this.db.processorFlow.count({
+        where: { formKey: key, action: CLASSIFY_ACTION },
+      })) > 0;
     const where: Prisma.SubmissionWhereInput = {
       versionId: { in: form.versions.map((version) => version.id) },
       createdAt: { ...dates, ...(snapshot ? { lte: snapshot } : {}) },
+      ...(classified && page.classification
+        ? classificationFilter(page.classification)
+        : {}),
     };
     if (
       page.after &&
@@ -168,6 +241,7 @@ export class FormsService {
             },
           },
           version: true,
+          classification: classified,
         },
       }),
       this.db.submission.count({ where }),
@@ -186,6 +260,7 @@ export class FormsService {
       'source_page',
       'form_version',
       ...fields.map((field) => field.key),
+      ...(classified ? CLASSIFICATION_COLUMNS : []),
     ];
     const data = submissions.slice(0, page.limit).map((submission) => {
       const row: Record<string, unknown> = Object.fromEntries(
@@ -212,14 +287,17 @@ export class FormsService {
               answer.option?.key ??
               null);
       }
+      if (classified)
+        Object.assign(row, classificationCells(submission.classification));
       return row;
     });
     return {
       form: key,
       columns,
-      labels: Object.fromEntries(
-        fields.map((field) => [field.key, field.label]),
-      ),
+      labels: {
+        ...Object.fromEntries(fields.map((field) => [field.key, field.label])),
+        ...(classified ? CLASSIFICATION_LABELS : {}),
+      },
       data,
       total,
       nextCursor: