feat(p7): AiJobService + AiBudgetGuard + local inference (gate→budget→propose)

Task 7.3: AiJobService runs scoped/purposed AI jobs that only propose — fails
closed via OPA (KG-03), degrades on budget (AiBudgetGuard sums costUnits, KG-12),
then persists job+modelRun+artifact(PROPOSED)+claims+evidence with full
provenance. CLASSIFY also proposes IiosModerationFlag(proposedBy:'AI'). accept/
reject record human feedback + emit events. Idempotent per job key. INFERENCE_PORT
binds LocalDeterministicInference for nest start; specs inject testkit FakeInference.
8 tests: classify/summarize/extract, fail-closed, budget-degrade, idempotent, accept.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-01 16:34:32 +05:30
parent e03f2f72e4
commit c6dc1cf112
6 changed files with 491 additions and 0 deletions
@@ -0,0 +1,26 @@
import { Injectable } from '@nestjs/common';
import { PrismaService } from '../prisma/prisma.service';
/**
* Budget governor (Critics KG-12). Sums `IiosAiModelRun.costUnits` for a scope and
* compares against a configured quota. When exceeded the caller degrades — the job
* ABSTAINS and no artifact is produced — rather than letting AI cost run unbounded.
*/
@Injectable()
export class AiBudgetGuard {
private quota = Number(process.env.IIOS_AI_BUDGET_UNITS ?? '1000000');
constructor(private readonly prisma: PrismaService) {}
/** Test/config hook to force a low quota. */
setQuota(units: number): this {
this.quota = units;
return this;
}
async check(scopeId: string): Promise<{ ok: boolean; spent: number; quota: number }> {
const agg = await this.prisma.iiosAiModelRun.aggregate({ _sum: { costUnits: true }, where: { scopeId } });
const spent = agg._sum.costUnits ?? 0;
return { ok: spent < this.quota, spent, quota: this.quota };
}
}
+20
View File
@@ -0,0 +1,20 @@
import { Module } from '@nestjs/common';
import { OutboxModule } from '../outbox/outbox.module';
import { INFERENCE_PORT, LocalDeterministicInference } from './inference.port';
import { AiBudgetGuard } from './ai-budget.guard';
import { AiJobService } from './ai.service';
/**
* AI Interaction SDK (P7). Proposal layer only. `INFERENCE_PORT` binds the local
* deterministic golden provider for `nest start`; specs inject the testkit fake.
*/
@Module({
imports: [OutboxModule],
providers: [
{ provide: INFERENCE_PORT, useClass: LocalDeterministicInference },
AiBudgetGuard,
AiJobService,
],
exports: [AiJobService, AiBudgetGuard],
})
export class AiModule {}
+240
View File
@@ -0,0 +1,240 @@
import { createHash } from 'node:crypto';
import { Inject, Injectable, NotFoundException } from '@nestjs/common';
import { Prisma, IiosAiJobType } from '@prisma/client';
import {
CloudEvent,
IIOS_EVENTS,
IiosPlatformPorts,
InferencePort,
InferenceResult,
} from '@insignia/iios-contracts';
import { PrismaService } from '../prisma/prisma.service';
import { PLATFORM_PORTS } from '../platform/platform-ports';
import { decideOrThrow } from '../platform/fail-closed';
import { ActorResolver, type MessagePrincipal } from '../identity/actor.resolver';
import { INFERENCE_PORT } from './inference.port';
import { AiBudgetGuard } from './ai-budget.guard';
export interface RunJobInput {
interactionId: string;
jobType: IiosAiJobType;
principal: MessagePrincipal;
idempotencyKey?: string;
}
/**
* The AI proposal layer (P7). Runs a scoped, purposed AI job that **only proposes**:
* it fails closed through OPA, degrades on budget (KG-12), and persists a durable
* artifact + claims + evidence + model-run with full provenance. It never writes a
* route decision, sends a message, merges identity, or approves anything — a human
* (or a deterministic P6 rule) is always the gate. CLASSIFY additionally proposes
* `IiosModerationFlag(proposedBy:'AI')` which can only push routing toward
* REVIEW/DENY, never ALLOW (KG-05).
*/
@Injectable()
export class AiJobService {
constructor(
private readonly prisma: PrismaService,
@Inject(PLATFORM_PORTS) private readonly ports: IiosPlatformPorts,
@Inject(INFERENCE_PORT) private readonly inference: InferencePort,
private readonly budget: AiBudgetGuard,
private readonly actors: ActorResolver,
) {}
async runJob(input: RunJobInput) {
const { interactionId, jobType } = input;
const idempotencyKey = input.idempotencyKey ?? `ai:${jobType}:${interactionId}`;
// Idempotent: a repeated key returns the existing job + its artifact — no duplicate work.
const prior = await this.prisma.iiosAiJob.findUnique({
where: { idempotencyKey },
include: { artifacts: { include: { claims: true, evidence: true } } },
});
if (prior) return { job: prior, artifact: prior.artifacts[0] ?? null };
const interaction = await this.prisma.iiosInteraction.findUnique({
where: { id: interactionId },
include: { parts: { orderBy: { partIndex: 'asc' } } },
});
if (!interaction) throw new NotFoundException('interaction not found');
const scopeId = interaction.scopeId;
const text = interaction.parts.find((p) => p.kind === 'TEXT')?.bodyText ?? '';
const purpose = `ai_${jobType.toLowerCase()}`;
const inputScopeHash = createHash('sha256').update(`${scopeId}:${interactionId}`).digest('hex').slice(0, 16);
const base = {
scopeId,
jobType,
interactionId,
inputWindowRef: interactionId,
purpose,
inputScopeHash,
idempotencyKey,
traceId: interaction.traceId ?? undefined,
};
// 1. Fail-closed authorization (Critics KG-03). Deny/timeout ⇒ FAILED job, no artifact.
let policyDecisionRef: string;
try {
const decision = await decideOrThrow(this.ports, { action: 'iios.ai.infer', scopeId, jobType, purpose });
policyDecisionRef = decision.decisionRef;
} catch (err) {
await this.prisma.iiosAiJob.create({ data: { ...base, status: 'FAILED', abstentionReason: 'POLICY_DENIED' } });
throw err;
}
// 2. Budget governor (Critics KG-12). Exceeded ⇒ ABSTAINED job, no artifact (degrade mode).
const budget = await this.budget.check(scopeId);
if (!budget.ok) {
const job = await this.prisma.iiosAiJob.create({
data: { ...base, status: 'ABSTAINED', policyDecisionRef, abstentionReason: 'BUDGET_EXCEEDED' },
});
return { job, artifact: null };
}
// 3. Inference. A provider error ⇒ FAILED job, no artifact (fail closed on the model too).
let result: InferenceResult;
try {
result = await this.inference.infer({ jobType, text, purpose, scopeId, interactionId });
} catch {
const job = await this.prisma.iiosAiJob.create({
data: { ...base, status: 'FAILED', policyDecisionRef, abstentionReason: 'PROVIDER_ERROR' },
});
return { job, artifact: null };
}
// 4. Persist the proposal (job + model run + artifact + claims + evidence).
const job = await this.prisma.iiosAiJob.create({ data: { ...base, status: 'DONE', policyDecisionRef } });
const modelRun = await this.prisma.iiosAiModelRun.create({ data: { scopeId, ...result.modelRun } });
const artifact = await this.prisma.iiosAiArtifact.create({
data: {
jobId: job.id,
modelRunId: modelRun.id,
artifactType: result.artifactType,
confidence: result.confidence,
contentRef: result.content as Prisma.InputJsonValue,
explanationRef: result.explanation,
abstentionReason: result.abstentionReason,
claims: {
create: result.claims.map((c) => ({
claimType: c.claimType,
claimJson: c.claimJson as Prisma.InputJsonValue,
confidence: c.confidence,
})),
},
evidence: {
create: result.evidence.map((e) => ({
sourceType: e.sourceType,
sourceId: e.sourceId,
spanRef: e.spanRef,
relevanceScore: e.relevanceScore,
})),
},
},
include: { claims: true, evidence: true },
});
// CLASSIFY proposes AI moderation flags (the P6 seam). Once per (interaction, flagType)
// for replay-safety. These feed rule-engine.decide() — REVIEW/DENY only, never ALLOW.
if (jobType === 'CLASSIFY') {
for (const claim of result.claims.filter((c) => c.claimType === 'MODERATION_FLAG')) {
const flagType = String(claim.claimJson.flagType ?? '');
if (!flagType) continue;
const exists = await this.prisma.iiosModerationFlag.count({ where: { interactionId, flagType, proposedBy: 'AI' } });
if (exists === 0) {
await this.prisma.iiosModerationFlag.create({
data: {
interactionId,
flagType,
proposedBy: 'AI',
confidence: claim.confidence,
explanation: String(claim.claimJson.explanation ?? result.explanation ?? ''),
},
});
}
}
}
await this.emit(IIOS_EVENTS.aiArtifactProposed, scopeId, artifact.id, {
jobId: job.id,
artifactId: artifact.id,
artifactType: artifact.artifactType,
interactionId,
});
return { job, artifact };
}
async listArtifacts(opts: { interactionId?: string; status?: string } = {}) {
return this.prisma.iiosAiArtifact.findMany({
where: {
...(opts.status ? { status: opts.status as Prisma.EnumIiosAiArtifactStatusFilter } : {}),
...(opts.interactionId ? { job: { interactionId: opts.interactionId } } : {}),
},
orderBy: { createdAt: 'desc' },
take: 100,
include: { claims: true, evidence: true, modelRun: true, job: true },
});
}
async getArtifact(id: string) {
const artifact = await this.prisma.iiosAiArtifact.findUnique({
where: { id },
include: { claims: true, evidence: true, modelRun: true, job: true },
});
if (!artifact) throw new NotFoundException('artifact not found');
return artifact;
}
/** Human acceptance feedback. Records who accepted; does NOT send or approve any route. */
async accept(artifactId: string, principal: MessagePrincipal) {
const artifact = await this.prisma.iiosAiArtifact.findUnique({ where: { id: artifactId }, include: { job: true } });
if (!artifact) throw new NotFoundException('artifact not found');
const actor = await this.actors.resolveActor(artifact.job.scopeId, principal);
const updated = await this.prisma.iiosAiArtifact.update({
where: { id: artifactId },
data: { status: 'ACCEPTED', acceptedByActorId: actor.id },
});
await this.prisma.iiosAiClaim.updateMany({
where: { artifactId },
data: { validationStatus: 'ACCEPTED', acceptedByRef: actor.id },
});
await this.emit(IIOS_EVENTS.aiArtifactAccepted, artifact.job.scopeId, artifactId, { artifactId, acceptedByActorId: actor.id });
return updated;
}
async reject(artifactId: string, principal: MessagePrincipal) {
const artifact = await this.prisma.iiosAiArtifact.findUnique({ where: { id: artifactId }, include: { job: true } });
if (!artifact) throw new NotFoundException('artifact not found');
const actor = await this.actors.resolveActor(artifact.job.scopeId, principal);
const updated = await this.prisma.iiosAiArtifact.update({
where: { id: artifactId },
data: { status: 'REJECTED', acceptedByActorId: actor.id },
});
await this.prisma.iiosAiClaim.updateMany({ where: { artifactId }, data: { validationStatus: 'REJECTED', acceptedByRef: actor.id } });
return updated;
}
private async emit(type: string, scopeId: string, aggregateId: string, data: Record<string, unknown>): Promise<void> {
const event: CloudEvent = {
specversion: '1.0',
id: `evt_ai_${aggregateId}_${type.split('.').slice(-2, -1)[0]}`,
type,
source: `iios/ai/${scopeId}`,
subject: `ai_artifact/${aggregateId}`,
time: new Date().toISOString(),
datacontenttype: 'application/json',
insignia: { scopeSnapshotId: scopeId, idempotencyKey: `${type}:${aggregateId}`, dataClass: 'internal' },
data,
};
await this.prisma.iiosOutboxEvent.create({
data: {
aggregateType: 'ai_artifact',
aggregateId,
eventType: type,
cloudEvent: event as unknown as Prisma.InputJsonValue,
partitionKey: `${scopeId}:${aggregateId}`,
},
});
}
}
+109
View File
@@ -0,0 +1,109 @@
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 { PolicyDeniedError } from '@insignia/iios-contracts';
import { MessageService, type MessagePrincipal } from '../messaging/message.service';
import { ActorResolver } from '../identity/actor.resolver';
import { AiJobService } from './ai.service';
import { AiBudgetGuard } from './ai-budget.guard';
import type { PrismaService } from '../prisma/prisma.service';
import type { FakeInference } from '@insignia/iios-testkit';
import type { FakePorts } from '@insignia/iios-testkit';
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 user: MessagePrincipal = { userId: 'admin-1', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Admin' };
function svc(opts?: { ports?: FakePorts; inference?: FakeInference; quota?: number }) {
const ports = opts?.ports ?? makeFakePorts();
const inference = opts?.inference ?? makeFakeInference();
const budget = new AiBudgetGuard(asService);
if (opts?.quota !== undefined) budget.setQuota(opts.quota);
return new AiJobService(asService, ports, inference, budget, actors);
}
async function makeInteraction(text: string): Promise<string> {
const m = new MessageService(asService, makeFakePorts(), actors);
const { threadId } = await m.openThread(null, user);
const msg = await m.send(threadId, user, { content: text }, `k-${randomUUID()}`);
return msg.id;
}
beforeAll(async () => { await prisma.$connect(); });
afterAll(async () => { await prisma.$disconnect(); });
beforeEach(async () => { await resetDb(prisma); });
describe('AiJobService (P7 proposal layer)', () => {
it('CLASSIFY "party at 9 PM" → CLASSIFICATION artifact + MODERATION_FLAG claim + AI moderation flag', async () => {
const interactionId = await makeInteraction('There is a party at 9 PM tonight!');
const { job, artifact } = await svc().runJob({ interactionId, jobType: 'CLASSIFY', principal: user });
expect(job.status).toBe('DONE');
expect(artifact?.artifactType).toBe('CLASSIFICATION');
expect(artifact?.claims.some((c) => c.claimType === 'MODERATION_FLAG')).toBe(true);
const flag = await prisma.iiosModerationFlag.findFirst({ where: { interactionId, proposedBy: 'AI' } });
expect(flag?.flagType).toBe('AFTER_HOURS_EVENT');
expect(flag?.confidence).toBeGreaterThan(0);
expect(flag?.explanation).toBeTruthy();
});
it('SUMMARIZE → SUMMARY artifact with a cited evidence link', async () => {
const interactionId = await makeInteraction('The committee met today. Budget discussed.');
const { artifact } = await svc().runJob({ interactionId, jobType: 'SUMMARIZE', principal: user });
expect(artifact?.artifactType).toBe('SUMMARY');
expect((artifact?.contentRef as { text?: string }).text).toContain('[ai]');
expect(artifact?.evidence.some((e) => e.sourceId === interactionId)).toBe(true);
});
it('EXTRACT → EVENT claim stays certainty=DISCUSSION (Scenario 7)', async () => {
const interactionId = await makeInteraction('Maybe a party at 9 PM?');
const { artifact } = await svc().runJob({ interactionId, jobType: 'EXTRACT', principal: user });
const event = artifact?.claims.find((c) => c.claimType === 'EVENT');
expect((event?.claimJson as { certainty?: string }).certainty).toBe('DISCUSSION');
});
it('fails closed: OPA deny → FAILED job, PolicyDeniedError, 0 artifacts (KG-03)', async () => {
const interactionId = await makeInteraction('hello');
const ports = makeFakePorts();
ports.opa.deny('no ai purpose');
await expect(svc({ ports }).runJob({ interactionId, jobType: 'CLASSIFY', principal: user })).rejects.toBeInstanceOf(PolicyDeniedError);
expect(await prisma.iiosAiArtifact.count()).toBe(0);
expect((await prisma.iiosAiJob.findFirstOrThrow({ where: { interactionId } })).status).toBe('FAILED');
});
it('budget governor: quota 0 → ABSTAINED job, 0 artifacts (KG-12 degrade)', async () => {
const interactionId = await makeInteraction('The committee met today.');
const { job, artifact } = await svc({ quota: 0 }).runJob({ interactionId, jobType: 'SUMMARIZE', principal: user });
expect(job.status).toBe('ABSTAINED');
expect(job.abstentionReason).toBe('BUDGET_EXCEEDED');
expect(artifact).toBeNull();
expect(await prisma.iiosAiArtifact.count()).toBe(0);
});
it('is idempotent: replaying the same job key twice → exactly 1 artifact', async () => {
const interactionId = await makeInteraction('There is a party at 9 PM tonight!');
const s = svc();
const key = `ai:CLASSIFY:${interactionId}`;
await s.runJob({ interactionId, jobType: 'CLASSIFY', principal: user, idempotencyKey: key });
await s.runJob({ interactionId, jobType: 'CLASSIFY', principal: user, idempotencyKey: key });
expect(await prisma.iiosAiJob.count({ where: { interactionId } })).toBe(1);
expect(await prisma.iiosAiArtifact.count()).toBe(1);
});
it('accept records human feedback (ACCEPTED + claims validated) and emits an event', async () => {
const interactionId = await makeInteraction('There is a party at 9 PM tonight!');
const { artifact } = await svc().runJob({ interactionId, jobType: 'CLASSIFY', principal: user });
const updated = await svc().accept(artifact!.id, user);
expect(updated.status).toBe('ACCEPTED');
expect(updated.acceptedByActorId).toBeTruthy();
const claim = await prisma.iiosAiClaim.findFirst({ where: { artifactId: artifact!.id } });
expect(claim?.validationStatus).toBe('ACCEPTED');
const evt = await prisma.iiosOutboxEvent.findFirst({ where: { eventType: 'com.insignia.iios.ai.artifact.accepted.v1', aggregateId: artifact!.id } });
expect(evt).toBeTruthy();
});
});
@@ -0,0 +1,94 @@
import type { InferencePort, InferenceRequest, InferenceResult, InferenceEvidence } from '@insignia/iios-contracts';
/** DI token for the AI inference port (P7). Tests inject the testkit FakeInference. */
export const INFERENCE_PORT = Symbol('INFERENCE_PORT');
/**
* The `nest start` binding: a deterministic, golden, no-network inference provider
* — the same split as `LocalDevPorts` vs the testkit fakes. Behaviour mirrors
* `@insignia/iios-testkit`'s `FakeInference` (kept runtime-independent of testkit,
* which is a dev dependency). A real provider via the Capability Broker replaces
* this in P9.
*/
export class LocalDeterministicInference implements InferencePort {
async infer(req: InferenceRequest): Promise<InferenceResult> {
const text = req.text ?? '';
const evidence: InferenceEvidence[] = [
{ sourceType: 'INTERACTION', sourceId: req.interactionId ?? 'unknown', relevanceScore: 1 },
];
const tokensIn = text.length === 0 ? 0 : text.trim().split(/\s+/).length;
const model =
req.jobType === 'CLASSIFY' ? 'golden-classifier' : req.jobType === 'SUMMARIZE' ? 'golden-summarizer' : 'golden-extractor';
const modelRun = {
model,
modelVersion: '1',
promptVersion: 'p1',
tokensIn,
tokensOut: Math.max(1, Math.ceil(tokensIn / 4)),
costUnits: Math.max(1, Math.ceil(text.length / 10)),
latencyMs: 1,
};
if (req.jobType === 'CLASSIFY') {
const flagTypes = this.flags(text);
return {
artifactType: 'CLASSIFICATION',
content: { flags: flagTypes },
confidence: flagTypes.length > 0 ? 0.9 : 0.95,
explanation: flagTypes.length > 0 ? `flagged: ${flagTypes.join(', ')}` : 'no moderation concerns',
claims: flagTypes.map((flagType) => ({
claimType: 'MODERATION_FLAG' as const,
claimJson: { flagType, explanation: `keyword rule matched for ${flagType}` },
confidence: 0.9,
})),
evidence,
modelRun,
};
}
if (req.jobType === 'SUMMARIZE') {
const first = (text.split(/[.!?]/)[0] ?? '').trim();
return {
artifactType: 'SUMMARY',
content: { text: `[ai] ${first}`, sourceInteractionId: req.interactionId ?? null },
confidence: 0.8,
explanation: 'extractive first-sentence summary (golden)',
claims: [],
evidence,
modelRun,
};
}
const event = this.extractEvent(text);
if (!event) {
return { artifactType: 'EXTRACTION', content: { claims: [] }, confidence: 0, abstentionReason: 'NO_CLAIM', claims: [], evidence, modelRun };
}
return {
artifactType: 'EXTRACTION',
content: { claims: [event] },
confidence: 0.7,
explanation: 'candidate event extracted (unconfirmed)',
claims: [{ claimType: 'EVENT', claimJson: event, confidence: 0.7 }],
evidence,
modelRun,
};
}
private flags(text: string): string[] {
const t = text.toLowerCase();
const out: string[] = [];
if (t.includes('party') || /\b(9|10|11)\s*pm\b/.test(t)) out.push('AFTER_HOURS_EVENT');
if (t.includes('sale') || t.includes('promo') || t.includes('discount')) out.push('PROMOTION');
return out;
}
private extractEvent(text: string): { title: string; when: string; certainty: string } | null {
const t = text.toLowerCase();
const timeMatch = t.match(/\b(9|10|11|12|[1-8])\s*(am|pm)\b/);
if (t.includes('party') || t.includes('event') || t.includes('meeting') || timeMatch) {
const title = t.includes('party') ? 'party' : t.includes('meeting') ? 'meeting' : 'event';
return { title, when: timeMatch ? timeMatch[0] : 'unspecified', certainty: 'DISCUSSION' };
}
return null;
}
}
+2
View File
@@ -10,6 +10,7 @@ import { InboxModule } from './inbox/inbox.module';
import { SupportModule } from './support/support.module';
import { AdaptersModule } from './adapters/adapters.module';
import { RoutingModule } from './routing/routing.module';
import { AiModule } from './ai/ai.module';
import { HealthController } from './health.controller';
import { DevController } from './dev/dev.controller';
@@ -26,6 +27,7 @@ import { DevController } from './dev/dev.controller';
SupportModule,
AdaptersModule,
RoutingModule,
AiModule,
],
controllers: [HealthController, DevController],
})