feat(p9): route OutboundService through the Capability Broker

Task 9.3: OutboundService now delegates the execute step to CapabilityBroker
(policy-gated + obligation-enforced) instead of calling a provider directly, while
keeping idempotency + rate-limit + the command/attempt ledger. BLOCKED obligations
→ command BLOCKED; a fail-closed throw marks the command FAILED then propagates.
AdaptersModule imports CapabilityModule; P4/P6/P8 egress now flows through the
broker unchanged. Updated 6 spec constructions; added a DENY_OUTBOUND test.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-01 20:04:11 +05:30
parent c50a501f32
commit 28e517fc1b
7 changed files with 50 additions and 23 deletions
@@ -1,6 +1,7 @@
import { Module } from '@nestjs/common'; import { Module } from '@nestjs/common';
import { OutboxModule } from '../outbox/outbox.module'; import { OutboxModule } from '../outbox/outbox.module';
import { InteractionsModule } from '../interactions/interactions.module'; import { InteractionsModule } from '../interactions/interactions.module';
import { CapabilityModule } from '../capability/capability.module';
import { AdapterRegistry } from './adapter.registry'; import { AdapterRegistry } from './adapter.registry';
import { InboundService } from './inbound.service'; import { InboundService } from './inbound.service';
import { OutboundService } from './outbound.service'; import { OutboundService } from './outbound.service';
@@ -8,7 +9,7 @@ import { RawEventProjector } from './raw-event.projector';
import { AdaptersController } from './adapters.controller'; import { AdaptersController } from './adapters.controller';
@Module({ @Module({
imports: [OutboxModule, InteractionsModule], imports: [OutboxModule, InteractionsModule, CapabilityModule],
controllers: [AdaptersController], controllers: [AdaptersController],
providers: [AdapterRegistry, InboundService, OutboundService, RawEventProjector], providers: [AdapterRegistry, InboundService, OutboundService, RawEventProjector],
exports: [AdapterRegistry, InboundService, OutboundService], exports: [AdapterRegistry, InboundService, OutboundService],
@@ -7,6 +7,8 @@ import { signedFixture, emailFixture } from '@insignia/iios-adapter-sdk';
import { AdapterRegistry } from './adapter.registry'; import { AdapterRegistry } from './adapter.registry';
import { InboundService } from './inbound.service'; import { InboundService } from './inbound.service';
import { OutboundService } from './outbound.service'; import { OutboundService } from './outbound.service';
import { CapabilityBroker } from '../capability/capability.broker';
import { CapabilityProviderRegistry } from '../capability/capability.registry';
import { RawEventProjector } from './raw-event.projector'; import { RawEventProjector } from './raw-event.projector';
import { IngestService } from '../interactions/ingest.service'; import { IngestService } from '../interactions/ingest.service';
import { ActorResolver } from '../identity/actor.resolver'; import { ActorResolver } from '../identity/actor.resolver';
@@ -77,7 +79,7 @@ describe('Adapter inbound (P5)', () => {
describe('Adapter outbound (P5)', () => { describe('Adapter outbound (P5)', () => {
const outbound = (limit?: number): OutboundService => { const outbound = (limit?: number): OutboundService => {
const s = new OutboundService(asService, registry()); const s = new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()));
if (limit) s.limit = limit; if (limit) s.limit = limit;
return s; return s;
}; };
@@ -104,4 +106,13 @@ describe('Adapter outbound (P5)', () => {
expect(b.id).toBe(a.id); expect(b.id).toBe(a.id);
expect(await prisma.iiosOutboundCommand.count()).toBe(1); expect(await prisma.iiosOutboundCommand.count()).toBe(1);
}); });
it('broker obligation DENY_OUTBOUND → command BLOCKED, no send (P9)', async () => {
const ports = makeFakePorts();
ports.opa.setDecision({ allow: true, obligations: [{ kind: 'DENY_OUTBOUND', reason: 'recipient opted out' }] });
const svc = new OutboundService(asService, new CapabilityBroker(ports, new CapabilityProviderRegistry()));
const cmd = await svc.send('WEBHOOK', 'user@x', { text: 'hi' }, 'blk-1');
expect(cmd.status).toBe('BLOCKED');
expect(cmd.providerRef).toBe('blocked');
});
}); });
@@ -1,13 +1,14 @@
import { randomUUID } from 'node:crypto'; import { randomUUID } from 'node:crypto';
import { Injectable, NotFoundException } from '@nestjs/common'; import { Injectable } from '@nestjs/common';
import { Prisma } from '@prisma/client'; import { Prisma } from '@prisma/client';
import { PrismaService } from '../prisma/prisma.service'; import { PrismaService } from '../prisma/prisma.service';
import { AdapterRegistry } from './adapter.registry'; import { CapabilityBroker } from '../capability/capability.broker';
/** /**
* Outbound to a provider — in P5 the adapter's `send` is a SANDBOX sink (no real * Outbound command layer (P5). Owns idempotency + per-(channelType,target) rate
* network). Per-(channelType,target) rate limiting; idempotent per command key; * limiting + the delivery ledger. As of P9 the actual execute step is delegated to
* every attempt recorded. Real delivery is gated to P6+. * the governed CapabilityBroker (policy gate + obligations + provider selection);
* this layer no longer talks to a provider directly.
*/ */
@Injectable() @Injectable()
export class OutboundService { export class OutboundService {
@@ -17,7 +18,7 @@ export class OutboundService {
constructor( constructor(
private readonly prisma: PrismaService, private readonly prisma: PrismaService,
private readonly registry: AdapterRegistry, private readonly broker: CapabilityBroker,
) {} ) {}
async send( async send(
@@ -25,10 +26,8 @@ export class OutboundService {
target: string, target: string,
payload: Record<string, unknown>, payload: Record<string, unknown>,
idempotencyKey: string = randomUUID(), idempotencyKey: string = randomUUID(),
scopeId?: string,
) { ) {
const reg = this.registry.get(channelType);
if (!reg) throw new NotFoundException(`unknown channel type: ${channelType}`);
const existing = await this.prisma.iiosOutboundCommand.findUnique({ where: { idempotencyKey } }); const existing = await this.prisma.iiosOutboundCommand.findUnique({ where: { idempotencyKey } });
if (existing) return existing; if (existing) return existing;
@@ -43,11 +42,23 @@ export class OutboundService {
const cmd = await this.prisma.iiosOutboundCommand.create({ const cmd = await this.prisma.iiosOutboundCommand.create({
data: { channelType, target, payload: payload as Prisma.InputJsonValue, status: 'PENDING', idempotencyKey }, data: { channelType, target, payload: payload as Prisma.InputJsonValue, status: 'PENDING', idempotencyKey },
}); });
const result = await reg.adapter.send({ channelType, target, payload }); // sandbox — no network
let result;
try {
result = await this.broker.invoke({ capability: 'channel.send', channelType, target, payload, idempotencyKey, scopeId });
} catch (err) {
// Fail-closed (e.g. PolicyDeniedError): mark the command FAILED, record the attempt, propagate.
await this.prisma.$transaction([
this.prisma.iiosOutboundCommand.update({ where: { id: cmd.id }, data: { status: 'FAILED' } }),
this.prisma.iiosDeliveryAttempt.create({ data: { commandId: cmd.id, attemptNo: 1, status: 'FAILED', errorCode: (err as Error).name } }),
]);
throw err;
}
const [updated] = await this.prisma.$transaction([ const [updated] = await this.prisma.$transaction([
this.prisma.iiosOutboundCommand.update({ where: { id: cmd.id }, data: { status: 'SENT', providerRef: result.providerRef } }), this.prisma.iiosOutboundCommand.update({ where: { id: cmd.id }, data: { status: result.outcome, providerRef: result.providerRef } }),
this.prisma.iiosDeliveryAttempt.create({ this.prisma.iiosDeliveryAttempt.create({
data: { commandId: cmd.id, attemptNo: 1, status: 'SENT', providerRef: result.providerRef, latencyMs: result.latencyMs }, data: { commandId: cmd.id, attemptNo: 1, status: result.outcome, providerRef: result.providerRef, latencyMs: result.latencyMs, errorCode: result.errorCode },
}), }),
]); ]);
return updated; return updated;
@@ -7,7 +7,8 @@ import { ActorResolver } from '../identity/actor.resolver';
import { AiJobService } from '../ai/ai.service'; import { AiJobService } from '../ai/ai.service';
import { AiBudgetGuard } from '../ai/ai-budget.guard'; import { AiBudgetGuard } from '../ai/ai-budget.guard';
import { OutboundService } from '../adapters/outbound.service'; import { OutboundService } from '../adapters/outbound.service';
import { AdapterRegistry } from '../adapters/adapter.registry'; import { CapabilityBroker } from '../capability/capability.broker';
import { CapabilityProviderRegistry } from '../capability/capability.registry';
import { CalendarService } from './calendar.service'; import { CalendarService } from './calendar.service';
import type { PrismaService } from '../prisma/prisma.service'; import type { PrismaService } from '../prisma/prisma.service';
import type { FakePorts } from '@insignia/iios-testkit'; import type { FakePorts } from '@insignia/iios-testkit';
@@ -21,7 +22,7 @@ const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'po
const START = '2026-07-02T10:00:00.000Z'; const START = '2026-07-02T10:00:00.000Z';
const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors); const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors);
const cal = (ports?: FakePorts) => new CalendarService(asService, ports ?? makeFakePorts(), actors, ai(), new OutboundService(asService, new AdapterRegistry())); const cal = (ports?: FakePorts) => new CalendarService(asService, ports ?? makeFakePorts(), actors, ai(), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry())));
async function scheduledMeeting(attendees: { userId: string; visibility?: 'FULL' | 'NONE' }[]) { async function scheduledMeeting(attendees: { userId: string; visibility?: 'FULL' | 'NONE' }[]) {
const m = await cal().scheduleDirect(alice, { meetingType: 'INTERNAL', title: 'Board sync', startAt: START, attendees }); const m = await cal().scheduleDirect(alice, { meetingType: 'INTERNAL', title: 'Board sync', startAt: START, attendees });
@@ -9,7 +9,8 @@ import { ActorResolver } from '../identity/actor.resolver';
import { AiJobService } from '../ai/ai.service'; import { AiJobService } from '../ai/ai.service';
import { AiBudgetGuard } from '../ai/ai-budget.guard'; import { AiBudgetGuard } from '../ai/ai-budget.guard';
import { OutboundService } from '../adapters/outbound.service'; import { OutboundService } from '../adapters/outbound.service';
import { AdapterRegistry } from '../adapters/adapter.registry'; import { CapabilityBroker } from '../capability/capability.broker';
import { CapabilityProviderRegistry } from '../capability/capability.registry';
import { CalendarService } from './calendar.service'; import { CalendarService } from './calendar.service';
import { CalendarProjector } from './calendar.projector'; import { CalendarProjector } from './calendar.projector';
import { OutboxBus } from '../outbox/outbox.bus'; import { OutboxBus } from '../outbox/outbox.bus';
@@ -24,7 +25,7 @@ const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'po
const START = '2026-07-02T10:00:00.000Z'; const START = '2026-07-02T10:00:00.000Z';
const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors); const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors);
const cal = () => new CalendarService(asService, makeFakePorts(), actors, ai(), new OutboundService(asService, new AdapterRegistry())); const cal = () => new CalendarService(asService, makeFakePorts(), actors, ai(), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry())));
async function makeInteraction(text: string): Promise<string> { async function makeInteraction(text: string): Promise<string> {
const m = new MessageService(asService, makeFakePorts(), actors); const m = new MessageService(asService, makeFakePorts(), actors);
@@ -9,7 +9,8 @@ import { ActorResolver } from '../identity/actor.resolver';
import { AiJobService } from '../ai/ai.service'; import { AiJobService } from '../ai/ai.service';
import { AiBudgetGuard } from '../ai/ai-budget.guard'; import { AiBudgetGuard } from '../ai/ai-budget.guard';
import { OutboundService } from '../adapters/outbound.service'; import { OutboundService } from '../adapters/outbound.service';
import { AdapterRegistry } from '../adapters/adapter.registry'; import { CapabilityBroker } from '../capability/capability.broker';
import { CapabilityProviderRegistry } from '../capability/capability.registry';
import { CalendarService } from './calendar.service'; import { CalendarService } from './calendar.service';
import type { PrismaService } from '../prisma/prisma.service'; import type { PrismaService } from '../prisma/prisma.service';
import type { FakePorts } from '@insignia/iios-testkit'; import type { FakePorts } from '@insignia/iios-testkit';
@@ -23,7 +24,7 @@ const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'po
const START = '2026-07-02T10:00:00.000Z'; const START = '2026-07-02T10:00:00.000Z';
const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors); const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors);
const outbound = () => new OutboundService(asService, new AdapterRegistry()); const outbound = () => new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()));
const cal = (ports?: FakePorts) => new CalendarService(asService, ports ?? makeFakePorts(), actors, ai(), outbound()); const cal = (ports?: FakePorts) => new CalendarService(asService, ports ?? makeFakePorts(), actors, ai(), outbound());
async function makeInteraction(text: string): Promise<string> { async function makeInteraction(text: string): Promise<string> {
@@ -11,7 +11,8 @@ import { RouteService } from './route.service';
import { RouteProjector } from './route.projector'; import { RouteProjector } from './route.projector';
import { IngestService } from '../interactions/ingest.service'; import { IngestService } from '../interactions/ingest.service';
import { OutboundService } from '../adapters/outbound.service'; import { OutboundService } from '../adapters/outbound.service';
import { AdapterRegistry } from '../adapters/adapter.registry'; import { CapabilityBroker } from '../capability/capability.broker';
import { CapabilityProviderRegistry } from '../capability/capability.registry';
import { OutboxBus } from '../outbox/outbox.bus'; import { OutboxBus } from '../outbox/outbox.bus';
import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts'; import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts';
import type { PrismaService } from '../prisma/prisma.service'; import type { PrismaService } from '../prisma/prisma.service';
@@ -103,7 +104,7 @@ describe('Route simulator (P6, preview-first)', () => {
describe('RouteService approve/deny → execute (P6)', () => { describe('RouteService approve/deny → execute (P6)', () => {
const admin: MessagePrincipal = { userId: 'admin', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Admin' }; const admin: MessagePrincipal = { userId: 'admin', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Admin' };
const svc = () => new RouteService(asService, new RouteSimulator(asService, new RuleEngine()), new OutboundService(asService, new AdapterRegistry()), actors); const svc = () => new RouteService(asService, new RouteSimulator(asService, new RuleEngine()), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry())), actors);
it('approving a REVIEW decision executes the forward to the sandbox', async () => { it('approving a REVIEW decision executes the forward to the sandbox', async () => {
const { interactionId, scopeId } = await makeInteraction('party at 9 PM'); const { interactionId, scopeId } = await makeInteraction('party at 9 PM');
@@ -138,7 +139,7 @@ describe('RouteService approve/deny → execute (P6)', () => {
}); });
describe('RouteProjector — AUTOMATIC forwarding (P6)', () => { describe('RouteProjector — AUTOMATIC forwarding (P6)', () => {
const routeSvc = () => new RouteService(asService, new RouteSimulator(asService, new RuleEngine()), new OutboundService(asService, new AdapterRegistry()), actors); const routeSvc = () => new RouteService(asService, new RouteSimulator(asService, new RuleEngine()), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry())), actors);
const projector = () => new RouteProjector(asService, new OutboxBus(), routeSvc()); const projector = () => new RouteProjector(asService, new OutboxBus(), routeSvc());
// Ingest an inbound (channel-bearing) interaction — routable, unlike native DMs. // Ingest an inbound (channel-bearing) interaction — routable, unlike native DMs.