feat(p7): wire AI into P6 routing + AiProjector (KG-05 structural proof)
Task 7.4: route-simulator now unions RULE flags with persisted AI moderation flags before decide() — so AI flags influence routing exactly like rule flags, only ever REVIEW/DENY, never ALLOW. SUMMARY/TRANSCRIPT/DIGEST previews render a proposed AI summary (aiSourced flag) instead of the P6 stub, still preview-only. AiProjector auto-runs CLASSIFY on interaction.normalized (channel-bearing only), idempotent via claim(). 4 tests prove KG-05: AI flag forces REVIEW on an auto-ALLOW binding, DENY on CHILD, AI summary in preview with 0 sends, projector idempotency. Full suite 79 green. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,113 @@
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
|
||||
import { PrismaClient } from '@prisma/client';
|
||||
import { resetDb } from '../test-utils/reset-db';
|
||||
import { makeFakePorts, makeFakeInference } from '@insignia/iios-testkit';
|
||||
import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts';
|
||||
import { ActorResolver } from '../identity/actor.resolver';
|
||||
import { IngestService } from '../interactions/ingest.service';
|
||||
import { RuleEngine } from '../routing/rule-engine';
|
||||
import { RouteSimulator } from '../routing/route-simulator';
|
||||
import { AiJobService } from './ai.service';
|
||||
import { AiBudgetGuard } from './ai-budget.guard';
|
||||
import { AiProjector } from './ai.projector';
|
||||
import { OutboxBus } from '../outbox/outbox.bus';
|
||||
import type { PrismaService } from '../prisma/prisma.service';
|
||||
|
||||
const url = process.env.DATABASE_URL ?? 'postgresql://iios:iios@localhost:5434/iios?schema=public';
|
||||
const prisma = new PrismaClient({ datasources: { db: { url } } });
|
||||
const asService = prisma as unknown as PrismaService;
|
||||
const actors = new ActorResolver(asService);
|
||||
|
||||
const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors);
|
||||
const sim = () => new RouteSimulator(asService, new RuleEngine());
|
||||
|
||||
async function ingestInbound(text: string): Promise<{ interactionId: string; scopeId: string; event: CloudEvent }> {
|
||||
const ingest = new IngestService(asService, makeFakePorts(), actors);
|
||||
const res = await ingest.ingest(
|
||||
{
|
||||
scope: { orgId: 'org_demo', appId: 'portal-demo' },
|
||||
channel: { type: 'WEBHOOK', externalChannelId: 'webhook' },
|
||||
source: { handleKind: 'EXTERNAL_VISITOR', externalId: 'ext-1' },
|
||||
kind: 'MESSAGE',
|
||||
thread: { externalThreadId: `wh-${randomUUID().slice(0, 6)}` },
|
||||
parts: [{ kind: 'TEXT', bodyText: text }],
|
||||
occurredAt: new Date().toISOString(),
|
||||
providerEventId: `pe-${randomUUID()}`,
|
||||
},
|
||||
`ing-${randomUUID()}`,
|
||||
);
|
||||
const interaction = await prisma.iiosInteraction.findUniqueOrThrow({ where: { id: res.interactionId } });
|
||||
const ev = (await prisma.iiosOutboxEvent.findFirstOrThrow({
|
||||
where: { eventType: IIOS_EVENTS.interactionNormalized, aggregateId: res.interactionId },
|
||||
})).cloudEvent as unknown as CloudEvent;
|
||||
return { interactionId: res.interactionId, scopeId: interaction.scopeId, event: ev };
|
||||
}
|
||||
|
||||
function binding(scopeId: string, destinationRef: string, opts: { restriction?: string; requiresReview?: boolean; outputFormat?: 'FORWARD' | 'SUMMARY' } = {}) {
|
||||
return prisma.iiosRouteBinding.create({
|
||||
data: {
|
||||
scopeId,
|
||||
originChannelType: 'WEBHOOK',
|
||||
originRef: 'webhook',
|
||||
destinationChannelType: 'PORTAL',
|
||||
destinationRef,
|
||||
restrictionProfile: opts.restriction,
|
||||
requiresReview: opts.requiresReview ?? false,
|
||||
outputFormat: opts.outputFormat ?? 'FORWARD',
|
||||
mode: 'MANUAL',
|
||||
enabled: true,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
const stateFor = async (interactionId: string, destinationRef: string) => {
|
||||
const b = await prisma.iiosRouteBinding.findFirstOrThrow({ where: { destinationRef } });
|
||||
return (await prisma.iiosRouteDecision.findUniqueOrThrow({ where: { interactionId_routeBindingId: { interactionId, routeBindingId: b.id } } })).decisionState;
|
||||
};
|
||||
|
||||
beforeAll(async () => { await prisma.$connect(); });
|
||||
afterAll(async () => { await prisma.$disconnect(); });
|
||||
beforeEach(async () => { await resetDb(prisma); });
|
||||
|
||||
describe('AI × routing (P7, KG-05: AI proposes, never approves)', () => {
|
||||
it('an AI flag forces REVIEW on an otherwise-auto-ALLOW binding — never ALLOW', async () => {
|
||||
const { interactionId, scopeId } = await ingestInbound('There is a party at 9 PM tonight!');
|
||||
// requiresReview:false + no rule match would be ALLOW — but CLASSIFY proposes an AI flag.
|
||||
await binding(scopeId, 'adults', { requiresReview: false });
|
||||
await ai().runJob({ interactionId, jobType: 'CLASSIFY' });
|
||||
|
||||
await sim().simulate(interactionId, { channelType: 'WEBHOOK', ref: 'webhook' });
|
||||
expect(await stateFor(interactionId, 'adults')).toBe('REVIEW');
|
||||
});
|
||||
|
||||
it('AI flag on a restricted (CHILD) destination → DENY (deny-by-default)', async () => {
|
||||
const { interactionId, scopeId } = await ingestInbound('There is a party at 9 PM tonight!');
|
||||
await binding(scopeId, 'children', { restriction: 'CHILD' });
|
||||
await ai().runJob({ interactionId, jobType: 'CLASSIFY' });
|
||||
|
||||
await sim().simulate(interactionId, { channelType: 'WEBHOOK', ref: 'webhook' });
|
||||
expect(await stateFor(interactionId, 'children')).toBe('DENY');
|
||||
});
|
||||
|
||||
it('SUMMARY binding renders the AI summary into the preview — and still sends nothing', async () => {
|
||||
const { interactionId, scopeId } = await ingestInbound('The committee met today. Budget discussed.');
|
||||
await binding(scopeId, 'digest-lane', { outputFormat: 'SUMMARY' });
|
||||
await ai().runJob({ interactionId, jobType: 'SUMMARIZE' });
|
||||
|
||||
const { decisions } = await sim().simulate(interactionId, { channelType: 'WEBHOOK', ref: 'webhook' });
|
||||
const preview = decisions[0]!.previewPayload as { text: string; aiSourced: boolean };
|
||||
expect(preview.text).toContain('[ai]');
|
||||
expect(preview.aiSourced).toBe(true);
|
||||
expect(await prisma.iiosOutboundCommand.count()).toBe(0); // preview only, no send
|
||||
});
|
||||
|
||||
it('AiProjector auto-classifies a channel-bearing interaction, idempotently', async () => {
|
||||
const { interactionId, event } = await ingestInbound('There is a party at 9 PM tonight!');
|
||||
const p = new AiProjector(asService, new OutboxBus(), ai());
|
||||
await p.onNormalized(event);
|
||||
await p.onNormalized(event); // replay
|
||||
expect(await prisma.iiosAiJob.count({ where: { interactionId, jobType: 'CLASSIFY' } })).toBe(1);
|
||||
expect(await prisma.iiosModerationFlag.count({ where: { interactionId, proposedBy: 'AI' } })).toBe(1);
|
||||
});
|
||||
});
|
||||
@@ -3,6 +3,7 @@ import { OutboxModule } from '../outbox/outbox.module';
|
||||
import { INFERENCE_PORT, LocalDeterministicInference } from './inference.port';
|
||||
import { AiBudgetGuard } from './ai-budget.guard';
|
||||
import { AiJobService } from './ai.service';
|
||||
import { AiProjector } from './ai.projector';
|
||||
|
||||
/**
|
||||
* AI Interaction SDK (P7). Proposal layer only. `INFERENCE_PORT` binds the local
|
||||
@@ -14,6 +15,7 @@ import { AiJobService } from './ai.service';
|
||||
{ provide: INFERENCE_PORT, useClass: LocalDeterministicInference },
|
||||
AiBudgetGuard,
|
||||
AiJobService,
|
||||
AiProjector,
|
||||
],
|
||||
exports: [AiJobService, AiBudgetGuard],
|
||||
})
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
import { Injectable, OnModuleInit } from '@nestjs/common';
|
||||
import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts';
|
||||
import { PrismaService } from '../prisma/prisma.service';
|
||||
import { OutboxBus } from '../outbox/outbox.bus';
|
||||
import { AiJobService } from './ai.service';
|
||||
|
||||
/**
|
||||
* Auto-proposal (P7). Reacts to `interaction.normalized` — channel-bearing messages
|
||||
* that arrived via an adapter — and runs a CLASSIFY job so inbound content gets an AI
|
||||
* moderation proposal automatically. Those flags only ever push routing toward
|
||||
* REVIEW/DENY (KG-05); nothing is sent or approved. Idempotent via `claim()` and the
|
||||
* job's own idempotency key.
|
||||
*/
|
||||
@Injectable()
|
||||
export class AiProjector implements OnModuleInit {
|
||||
private readonly consumer = 'ai-projector';
|
||||
|
||||
constructor(
|
||||
private readonly prisma: PrismaService,
|
||||
private readonly bus: OutboxBus,
|
||||
private readonly ai: AiJobService,
|
||||
) {}
|
||||
|
||||
onModuleInit(): void {
|
||||
this.bus.on(IIOS_EVENTS.interactionNormalized, (p) => void this.onNormalized(p as CloudEvent).catch(() => {}));
|
||||
}
|
||||
|
||||
async onNormalized(event: CloudEvent): Promise<void> {
|
||||
if (!(await this.claim(event.id))) return;
|
||||
const { interactionId } = event.data as { interactionId: string };
|
||||
const interaction = await this.prisma.iiosInteraction.findUnique({
|
||||
where: { id: interactionId },
|
||||
include: { channel: true },
|
||||
});
|
||||
if (!interaction?.channel) return; // native DMs have no origin channel → not auto-classified
|
||||
await this.ai.runJob({ interactionId, jobType: 'CLASSIFY' });
|
||||
}
|
||||
|
||||
private async claim(eventId: string): Promise<boolean> {
|
||||
const res = await this.prisma.iiosProcessedEvent.createMany({
|
||||
data: [{ consumerName: this.consumer, eventId }],
|
||||
skipDuplicates: true,
|
||||
});
|
||||
return res.count > 0;
|
||||
}
|
||||
}
|
||||
@@ -18,7 +18,8 @@ import { AiBudgetGuard } from './ai-budget.guard';
|
||||
export interface RunJobInput {
|
||||
interactionId: string;
|
||||
jobType: IiosAiJobType;
|
||||
principal: MessagePrincipal;
|
||||
/** Optional — the job is scoped by the interaction, so the auto-projector needs no principal. */
|
||||
principal?: MessagePrincipal;
|
||||
idempotencyKey?: string;
|
||||
}
|
||||
|
||||
|
||||
@@ -32,9 +32,17 @@ export class RouteSimulator {
|
||||
|
||||
const text = interaction.parts.find((p) => p.kind === 'TEXT')?.bodyText ?? '';
|
||||
const hasAttachment = interaction.parts.some((p) => !!p.contentRef);
|
||||
const flags = this.rules.flag(text);
|
||||
const flagTypes = flags.map((f) => f.flagType);
|
||||
await this.persistFlags(interactionId, flags);
|
||||
const ruleFlags = this.rules.flag(text);
|
||||
await this.persistFlags(interactionId, ruleFlags);
|
||||
|
||||
// Union deterministic RULE flags with any persisted flags — including
|
||||
// proposedBy:'AI' from the P7 layer. AI flags therefore influence routing
|
||||
// exactly like rule flags: they can only push toward REVIEW/DENY (KG-05).
|
||||
const persisted = await this.prisma.iiosModerationFlag.findMany({ where: { interactionId } });
|
||||
const flagTypes = Array.from(new Set([...ruleFlags.map((f) => f.flagType), ...persisted.map((p) => p.flagType)]));
|
||||
|
||||
// A P7 AI summary (if one was proposed) renders into SUMMARY/TRANSCRIPT/DIGEST previews.
|
||||
const aiSummary = await this.latestAiSummary(interactionId);
|
||||
|
||||
const simulationId = opts?.simulationId ?? randomUUID();
|
||||
const bindings = await this.prisma.iiosRouteBinding.findMany({
|
||||
@@ -49,7 +57,7 @@ export class RouteSimulator {
|
||||
const decisions = [];
|
||||
for (const b of bindings) {
|
||||
const { state, reasonCodes } = this.rules.decide(b, flagTypes);
|
||||
const previewPayload = this.render(b, text, hasAttachment);
|
||||
const previewPayload = this.render(b, text, hasAttachment, aiSummary);
|
||||
const decision = await this.prisma.iiosRouteDecision.upsert({
|
||||
where: { interactionId_routeBindingId: { interactionId, routeBindingId: b.id } },
|
||||
create: {
|
||||
@@ -94,24 +102,36 @@ export class RouteSimulator {
|
||||
return { simulationId, decisions };
|
||||
}
|
||||
|
||||
private render(binding: IiosRouteBinding, text: string, hasAttachment: boolean) {
|
||||
private render(binding: IiosRouteBinding, text: string, hasAttachment: boolean, aiSummary?: string) {
|
||||
const attach = binding.includeAttachments && hasAttachment ? ' [+attachment]' : '';
|
||||
// SUMMARY/TRANSCRIPT/DIGEST use the P7 AI summary if one was proposed; else the
|
||||
// P6 deterministic stub. FORWARD/THREADED are verbatim. Still preview-only — no send.
|
||||
const rendered =
|
||||
binding.outputFormat === 'DIGEST'
|
||||
? `[digest] ${text}`
|
||||
? `[digest] ${aiSummary ?? text}`
|
||||
: binding.outputFormat === 'SUMMARY'
|
||||
? `[summary] ${text.slice(0, 60)}` // deterministic stub; real summary is P7 (AI)
|
||||
? (aiSummary ?? `[summary] ${text.slice(0, 60)}`)
|
||||
: binding.outputFormat === 'TRANSCRIPT'
|
||||
? `[transcript] ${text}`
|
||||
? `[transcript] ${aiSummary ?? text}`
|
||||
: text; // FORWARD / THREADED
|
||||
return {
|
||||
format: binding.outputFormat,
|
||||
destinationChannelType: binding.destinationChannelType,
|
||||
destinationRef: binding.destinationRef,
|
||||
aiSourced: !!aiSummary && binding.outputFormat !== 'FORWARD' && binding.outputFormat !== 'THREADED',
|
||||
text: rendered + attach,
|
||||
};
|
||||
}
|
||||
|
||||
private async latestAiSummary(interactionId: string): Promise<string | undefined> {
|
||||
const artifact = await this.prisma.iiosAiArtifact.findFirst({
|
||||
where: { artifactType: 'SUMMARY', job: { interactionId } },
|
||||
orderBy: { createdAt: 'desc' },
|
||||
});
|
||||
const content = artifact?.contentRef as { text?: string } | undefined;
|
||||
return content?.text;
|
||||
}
|
||||
|
||||
private async persistFlags(interactionId: string, flags: RuleFlagLike[]): Promise<void> {
|
||||
if (flags.length === 0) return;
|
||||
const existing = await this.prisma.iiosModerationFlag.count({ where: { interactionId, proposedBy: 'RULE' } });
|
||||
|
||||
Reference in New Issue
Block a user