From 28e517fc1b619dcebee9d702a1789af60de3e7c5 Mon Sep 17 00:00:00 2001 From: maaz519 Date: Wed, 1 Jul 2026 20:04:11 +0530 Subject: [PATCH] feat(p9): route OutboundService through the Capability Broker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- .../src/adapters/adapters.module.ts | 3 +- .../src/adapters/adapters.spec.ts | 13 ++++++- .../src/adapters/outbound.service.ts | 35 ++++++++++++------- .../src/calendar/calendar-consent.spec.ts | 5 +-- .../src/calendar/calendar-projector.spec.ts | 5 +-- .../src/calendar/calendar.spec.ts | 5 +-- .../iios-service/src/routing/route.spec.ts | 7 ++-- 7 files changed, 50 insertions(+), 23 deletions(-) diff --git a/packages/iios-service/src/adapters/adapters.module.ts b/packages/iios-service/src/adapters/adapters.module.ts index 71b3900..ff27d41 100644 --- a/packages/iios-service/src/adapters/adapters.module.ts +++ b/packages/iios-service/src/adapters/adapters.module.ts @@ -1,6 +1,7 @@ import { Module } from '@nestjs/common'; import { OutboxModule } from '../outbox/outbox.module'; import { InteractionsModule } from '../interactions/interactions.module'; +import { CapabilityModule } from '../capability/capability.module'; import { AdapterRegistry } from './adapter.registry'; import { InboundService } from './inbound.service'; import { OutboundService } from './outbound.service'; @@ -8,7 +9,7 @@ import { RawEventProjector } from './raw-event.projector'; import { AdaptersController } from './adapters.controller'; @Module({ - imports: [OutboxModule, InteractionsModule], + imports: [OutboxModule, InteractionsModule, CapabilityModule], controllers: [AdaptersController], providers: [AdapterRegistry, InboundService, OutboundService, RawEventProjector], exports: [AdapterRegistry, InboundService, OutboundService], diff --git a/packages/iios-service/src/adapters/adapters.spec.ts b/packages/iios-service/src/adapters/adapters.spec.ts index 3abe49b..0ccf657 100644 --- a/packages/iios-service/src/adapters/adapters.spec.ts +++ b/packages/iios-service/src/adapters/adapters.spec.ts @@ -7,6 +7,8 @@ import { signedFixture, emailFixture } from '@insignia/iios-adapter-sdk'; import { AdapterRegistry } from './adapter.registry'; import { InboundService } from './inbound.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 { IngestService } from '../interactions/ingest.service'; import { ActorResolver } from '../identity/actor.resolver'; @@ -77,7 +79,7 @@ describe('Adapter inbound (P5)', () => { describe('Adapter outbound (P5)', () => { 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; return s; }; @@ -104,4 +106,13 @@ describe('Adapter outbound (P5)', () => { expect(b.id).toBe(a.id); 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'); + }); }); diff --git a/packages/iios-service/src/adapters/outbound.service.ts b/packages/iios-service/src/adapters/outbound.service.ts index c5f319e..eefd9f4 100644 --- a/packages/iios-service/src/adapters/outbound.service.ts +++ b/packages/iios-service/src/adapters/outbound.service.ts @@ -1,13 +1,14 @@ import { randomUUID } from 'node:crypto'; -import { Injectable, NotFoundException } from '@nestjs/common'; +import { Injectable } from '@nestjs/common'; import { Prisma } from '@prisma/client'; 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 - * network). Per-(channelType,target) rate limiting; idempotent per command key; - * every attempt recorded. Real delivery is gated to P6+. + * Outbound command layer (P5). Owns idempotency + per-(channelType,target) rate + * limiting + the delivery ledger. As of P9 the actual execute step is delegated to + * the governed CapabilityBroker (policy gate + obligations + provider selection); + * this layer no longer talks to a provider directly. */ @Injectable() export class OutboundService { @@ -17,7 +18,7 @@ export class OutboundService { constructor( private readonly prisma: PrismaService, - private readonly registry: AdapterRegistry, + private readonly broker: CapabilityBroker, ) {} async send( @@ -25,10 +26,8 @@ export class OutboundService { target: string, payload: Record, 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 } }); if (existing) return existing; @@ -43,11 +42,23 @@ export class OutboundService { const cmd = await this.prisma.iiosOutboundCommand.create({ 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([ - 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({ - 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; diff --git a/packages/iios-service/src/calendar/calendar-consent.spec.ts b/packages/iios-service/src/calendar/calendar-consent.spec.ts index c635af4..9620496 100644 --- a/packages/iios-service/src/calendar/calendar-consent.spec.ts +++ b/packages/iios-service/src/calendar/calendar-consent.spec.ts @@ -7,7 +7,8 @@ import { ActorResolver } from '../identity/actor.resolver'; import { AiJobService } from '../ai/ai.service'; import { AiBudgetGuard } from '../ai/ai-budget.guard'; 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 type { PrismaService } from '../prisma/prisma.service'; 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 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' }[]) { const m = await cal().scheduleDirect(alice, { meetingType: 'INTERNAL', title: 'Board sync', startAt: START, attendees }); diff --git a/packages/iios-service/src/calendar/calendar-projector.spec.ts b/packages/iios-service/src/calendar/calendar-projector.spec.ts index 7bc53c3..160fb33 100644 --- a/packages/iios-service/src/calendar/calendar-projector.spec.ts +++ b/packages/iios-service/src/calendar/calendar-projector.spec.ts @@ -9,7 +9,8 @@ import { ActorResolver } from '../identity/actor.resolver'; import { AiJobService } from '../ai/ai.service'; import { AiBudgetGuard } from '../ai/ai-budget.guard'; 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 { CalendarProjector } from './calendar.projector'; 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 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 { const m = new MessageService(asService, makeFakePorts(), actors); diff --git a/packages/iios-service/src/calendar/calendar.spec.ts b/packages/iios-service/src/calendar/calendar.spec.ts index 8f90a70..1b43692 100644 --- a/packages/iios-service/src/calendar/calendar.spec.ts +++ b/packages/iios-service/src/calendar/calendar.spec.ts @@ -9,7 +9,8 @@ import { ActorResolver } from '../identity/actor.resolver'; import { AiJobService } from '../ai/ai.service'; import { AiBudgetGuard } from '../ai/ai-budget.guard'; 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 type { PrismaService } from '../prisma/prisma.service'; 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 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()); async function makeInteraction(text: string): Promise { diff --git a/packages/iios-service/src/routing/route.spec.ts b/packages/iios-service/src/routing/route.spec.ts index 8748aea..e28fb9e 100644 --- a/packages/iios-service/src/routing/route.spec.ts +++ b/packages/iios-service/src/routing/route.spec.ts @@ -11,7 +11,8 @@ import { RouteService } from './route.service'; import { RouteProjector } from './route.projector'; import { IngestService } from '../interactions/ingest.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 { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts'; import type { PrismaService } from '../prisma/prisma.service'; @@ -103,7 +104,7 @@ describe('Route simulator (P6, preview-first)', () => { describe('RouteService approve/deny → execute (P6)', () => { 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 () => { const { interactionId, scopeId } = await makeInteraction('party at 9 PM'); @@ -138,7 +139,7 @@ describe('RouteService approve/deny → execute (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()); // Ingest an inbound (channel-bearing) interaction — routable, unlike native DMs.