diff --git a/docs/api.md b/docs/api.md index fbf1568..6b3ec1e 100644 --- a/docs/api.md +++ b/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 diff --git a/docs/data-model.md b/docs/data-model.md index 2db65d6..c4ae0c5 100644 --- a/docs/data-model.md +++ b/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 diff --git a/docs/lets-talk.md b/docs/lets-talk.md index afb77aa..de5d202 100644 --- a/docs/lets-talk.md +++ b/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`. diff --git a/prisma/migrations/20261022010000_lets_talk_classification/migration.sql b/prisma/migrations/20261022010000_lets_talk_classification/migration.sql new file mode 100644 index 0000000..ab8ab50 --- /dev/null +++ b/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; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 7acbad7..0e3a16d 100644 --- a/prisma/schema.prisma +++ b/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) diff --git a/src/forms/dto.ts b/src/forms/dto.ts index 9eafdde..d3dcc97 100644 --- a/src/forms/dto.ts +++ b/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. */ diff --git a/src/forms/forms.service.ts b/src/forms/forms.service.ts index e6eb7a0..de847d4 100644 --- a/src/forms/forms.service.ts +++ b/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 = { + 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 { + 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 = 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: diff --git a/test/forms.integration.spec.ts b/test/forms.integration.spec.ts index 8409c29..e414bb1 100644 --- a/test/forms.integration.spec.ts +++ b/test/forms.integration.spec.ts @@ -331,7 +331,9 @@ describe('forms API with real PostgreSQL', () => { expect(csv.text).toContain('"\'=SUM(1,2)"'); expect(csv.text).toContain(first.id); expect(csv.text).toContain(second.id); - expect(csv.text.indexOf(second.id)).toBeLessThan(csv.text.indexOf(first.id)); + expect(csv.text.indexOf(second.id)).toBeLessThan( + csv.text.indexOf(first.id), + ); await request(app.getHttpServer()) .get(`${path}/export`) .query({ startDate: '2020-02-30' }) @@ -344,6 +346,197 @@ describe('forms API with real PostgreSQL', () => { .expect(400); }); + it("seeds the Let's talk classification flows disabled and leaves the sales email unchanged", async () => { + const flows = await db.processorFlow.findMany({ + where: { formKey: 'lets_talk' }, + orderBy: { id: 'asc' }, + }); + const rules = Object.fromEntries( + flows.map((flow) => [flow.id, flow.rules]), + ); + expect(rules).toEqual({ + 'lets-talk-classify': { all: [] }, + 'lets-talk-community-general-reply': { + all: [], + classification: ['COMMUNITY_GENERAL'], + }, + 'lets-talk-community-opportunity-reply': { + all: [], + classification: ['COMMUNITY_OPPORTUNITY'], + }, + 'lets-talk-community-skills-reply': { + all: [], + classification: ['COMMUNITY_SKILLS'], + }, + 'lets-talk-sales-email': { all: [] }, + 'lets-talk-sales-reply': { all: [], classification: ['SALES_LEAD'] }, + 'lets-talk-support-forward': { + all: [], + classification: ['COMMUNITY_SUPPORT'], + }, + 'lets-talk-wipro-queries-reply': { + all: [], + classification: ['WIPRO_QUERIES'], + }, + }); + expect( + flows + .filter((flow) => flow.id !== 'lets-talk-sales-email') + .every((flow) => !flow.enabled), + ).toBe(true); + expect(flows.find((flow) => flow.id === 'lets-talk-classify')?.action).toBe( + 'ai-classify', + ); + }); + + it('reports AI classification columns and filters by label only for forms with a classify flow', async () => { + const key = await create({ + ...definition(), + fields: definition().fields.map((field) => ({ + ...field, + required: false, + })), + }); + const version = await db.formVersion.findFirstOrThrow({ + where: { form: { key } }, + }); + const reportPath = `/v6/forms/${key}/submissions`; + const columnsOf = async () => + ( + await request(app.getHttpServer()) + .get(reportPath) + .auth(reporter, { type: 'bearer' }) + .expect(200) + ).body.columns as string[]; + // No classify flow yet: the report is unchanged. + expect(await columnsOf()).not.toContain('classification'); + + await db.processorFlow.create({ + data: { id: `${key}-classify`, formKey: key, action: 'ai-classify' }, + }); + const [lead, junk, pending, never] = Array.from({ length: 4 }, () => + randomUUID(), + ); + await db.submission.createMany({ + data: [lead, junk, pending, never].map((id, index) => ({ + id, + versionId: version.id, + idempotencyKey: randomUUID(), + requestHash: 'a'.repeat(64), + createdAt: new Date(Date.UTC(2020, 9, 1 + index)), + })), + }); + const runId = randomUUID(); + await db.submissionClassification.createMany({ + data: [ + { + submissionId: lead, + label: 'SALES_LEAD', + status: 'CLASSIFIED', + confidence: '0.9100', + needsReview: false, + workflowId: 'lets-talk-classification', + workflowRunId: runId, + classifiedAt: new Date(), + }, + { + submissionId: junk, + label: 'JUNK', + status: 'CLASSIFIED', + confidence: '0.6', + needsReview: true, + workflowId: 'lets-talk-classification', + classifiedAt: new Date(), + }, + { submissionId: pending, workflowId: 'lets-talk-classification' }, + ], + }); + + const report = await request(app.getHttpServer()) + .get(reportPath) + .auth(reporter, { type: 'bearer' }) + .expect(200); + expect(report.body.columns.slice(-5)).toEqual([ + 'classification', + 'classification_status', + 'classification_confidence', + 'classification_needs_review', + 'classification_run_id', + ]); + expect(report.body.labels.classification).toBe('Classification'); + const byId = Object.fromEntries( + report.body.data.map((row: Record) => [ + row.submission_id, + row, + ]), + ); + expect(byId[lead]).toMatchObject({ + classification: 'SALES_LEAD', + classification_status: 'CLASSIFIED', + classification_confidence: '0.91', + classification_needs_review: false, + classification_run_id: runId, + }); + expect(byId[junk]).toMatchObject({ + classification: 'JUNK', + classification_needs_review: true, + }); + expect(byId[pending]).toMatchObject({ + classification: 'UNKNOWN', + classification_status: 'PENDING', + }); + // Submitted before classification existed: no row at all. + expect(byId[never]).toMatchObject({ + classification: 'UNKNOWN', + classification_status: null, + classification_run_id: null, + }); + + const filtered = async (classification: string) => + ( + await request(app.getHttpServer()) + .get(reportPath) + .query({ classification }) + .auth(reporter, { type: 'bearer' }) + .expect(200) + ).body; + expect( + (await filtered('SALES_LEAD')).data.map( + (r: { submission_id: string }) => r.submission_id, + ), + ).toEqual([lead]); + const unknown = await filtered('UNKNOWN'); + expect(unknown.total).toBe(2); + expect( + unknown.data.map((r: { submission_id: string }) => r.submission_id), + ).toEqual([never, pending]); + expect((await filtered('COMMUNITY_SKILLS')).total).toBe(0); + await request(app.getHttpServer()) + .get(reportPath) + .query({ classification: 'sales_lead' }) + .auth(reporter, { type: 'bearer' }) + .expect(400); + + const csv = await request(app.getHttpServer()) + .get(`${reportPath}/export`) + .query({ classification: 'SALES_LEAD' }) + .auth(reporter, { type: 'bearer' }) + .expect(200); + const lines = csv.text.trim().split('\r\n'); + expect(lines).toHaveLength(2); + expect(lines[0]).toContain('"classification","classification_status"'); + expect(lines[1]).toContain('"SALES_LEAD"'); + expect(lines[1]).toContain(`"${runId}"`); + + // Deleting a submission erases its classification. + await db.submission.delete({ where: { id: lead } }); + expect( + await db.submissionClassification.count({ + where: { submissionId: lead }, + }), + ).toBe(0); + }); + it('exports more than one thousand rows with one header and no table-page truncation', async () => { const key = await create({ ...definition(), @@ -376,7 +569,10 @@ describe('forms API with real PostgreSQL', () => { 1002, ); expect(lines.slice(1).map((line) => line.split(',')[0])).toEqual( - [...ids].sort().reverse().map((id) => `"${id}"`), + [...ids] + .sort() + .reverse() + .map((id) => `"${id}"`), ); expect(csv.headers['content-disposition']).toContain(`${key}.csv`); expect(csv.headers['cache-control']).toBe('no-store');