From 451f7e04ec6c45adb12007dd7131da6a2b199962 Mon Sep 17 00:00:00 2001 From: Jarvis Date: Mon, 13 Jul 2026 04:28:27 -0500 Subject: [PATCH] feat(tess): wire durable interaction surfaces --- .../tess-cross-surface.integration.test.ts | 159 +++++++++++++++++ .../runtime-provider-registry.service.test.ts | 21 +++ .../src/agent/interaction.controller.test.ts | 163 ++++++++++++++++++ .../src/agent/interaction.controller.ts | 89 +++++++++- .../agent/runtime-approval-denied.filter.ts | 13 ++ .../runtime-provider-registry.service.ts | 15 +- .../agent/tess-durable-session.repository.ts | 19 +- .../src/agent/tess-durable-session.service.ts | 16 +- apps/gateway/src/chat/chat.gateway.ts | 116 ++++++++++--- .../plugin/discord-ingress.security.spec.ts | 76 ++++++-- docs/scratchpads/tess-20260712.md | 8 + .../agent/src/tess-durable-session.test.ts | 17 ++ packages/agent/src/tess-durable-session.ts | 10 +- .../mosaic/src/commands/interaction.test.ts | 1 + packages/mosaic/src/commands/interaction.ts | 48 +++++- packages/mosaic/src/tui/gateway-api.ts | 69 ++++++++ 16 files changed, 782 insertions(+), 58 deletions(-) create mode 100644 apps/gateway/src/__tests__/integration/tess-cross-surface.integration.test.ts create mode 100644 apps/gateway/src/agent/runtime-approval-denied.filter.ts diff --git a/apps/gateway/src/__tests__/integration/tess-cross-surface.integration.test.ts b/apps/gateway/src/__tests__/integration/tess-cross-surface.integration.test.ts new file mode 100644 index 0000000..ae43c9a --- /dev/null +++ b/apps/gateway/src/__tests__/integration/tess-cross-surface.integration.test.ts @@ -0,0 +1,159 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { InMemoryDurableSessionStore } from '@mosaicstack/agent'; +import { + createDiscordIngressEnvelope, + type DiscordIngressPayload, +} from '@mosaicstack/discord-plugin'; +import { InteractionController } from '../../agent/interaction.controller.js'; +import { RuntimeProviderService } from '../../agent/runtime-provider-registry.service.js'; +import { TessDurableSessionService } from '../../agent/tess-durable-session.service.js'; +import { ChatGateway } from '../../chat/chat.gateway.js'; +import { CommandAuthorizationService } from '../../commands/command-authorization.service.js'; + +const SERVICE_TOKEN = 'test-discord-service-token'; +const envKeys = [ + 'DISCORD_SERVICE_TOKEN', + 'DISCORD_SERVICE_TENANT_ID', + 'DISCORD_INTERACTION_BINDINGS', + 'DISCORD_ALLOWED_GUILD_IDS', + 'DISCORD_ALLOWED_CHANNEL_IDS', + 'DISCORD_ALLOWED_USER_IDS', + 'MOSAIC_AGENT_NAME', +] as const; +const priorEnv = new Map(); + +function payload(content: string, messageId: string, correlationId: string): DiscordIngressPayload { + return { + content, + messageId, + correlationId, + guildId: 'guild-1', + channelId: 'channel-1', + userId: 'discord-admin-1', + conversationId: 'conversation-1', + }; +} + +function authorization(): CommandAuthorizationService { + const entries = new Map(); + return new CommandAuthorizationService( + { + select: () => ({ + from: () => ({ where: () => ({ limit: async () => [{ role: 'admin' }] }) }), + }), + } as never, + { + get: async (key: string) => entries.get(key) ?? null, + set: async (key: string, value: string) => entries.set(key, value), + del: async (key: string) => Number(entries.delete(key)), + }, + ); +} + +describe('Tess Discord/CLI durable-session integration', () => { + afterEach(() => { + for (const key of envKeys) { + const value = priorEnv.get(key); + if (value === undefined) delete process.env[key]; + else process.env[key] = value; + } + priorEnv.clear(); + }); + + it('enrolls through the CLI surface then resolves the same durable session from Discord', async () => { + for (const key of envKeys) priorEnv.set(key, process.env[key]); + process.env['MOSAIC_AGENT_NAME'] = 'Nova'; + process.env['DISCORD_SERVICE_TOKEN'] = SERVICE_TOKEN; + process.env['DISCORD_SERVICE_TENANT_ID'] = 'tenant-1'; + process.env['DISCORD_ALLOWED_GUILD_IDS'] = 'guild-1'; + process.env['DISCORD_ALLOWED_CHANNEL_IDS'] = 'channel-1'; + process.env['DISCORD_ALLOWED_USER_IDS'] = 'discord-admin-1'; + process.env['DISCORD_INTERACTION_BINDINGS'] = JSON.stringify([ + { + instanceId: 'Nova', + guildId: 'guild-1', + channelId: 'channel-1', + pairedUsers: { + 'discord-admin-1': { role: 'admin', mosaicUserId: 'mosaic-admin-1' }, + }, + }, + ]); + + const durable = new TessDurableSessionService( + new InMemoryDurableSessionStore() as never, + {} as never, + ); + const enrollmentRuntime = { + listSessions: vi.fn().mockResolvedValue([{ id: 'runtime-1' }]), + }; + const controller = new InteractionController(enrollmentRuntime as never, durable); + await controller.enroll( + 'Nova', + 'conversation-1', + { providerId: 'fleet', runtimeSessionId: 'runtime-1' }, + { id: 'mosaic-admin-1', tenantId: 'tenant-1' }, + 'cli-enrollment-correlation', + ); + + const authz = authorization(); + const terminated = vi.fn().mockResolvedValue(undefined); + const runtime = new RuntimeProviderService( + { + require: () => ({ + capabilities: async () => ({ supported: ['session.terminate'] }), + terminate: terminated, + }), + } as never, + { record: async () => undefined } as never, + { + consume: (approvalId, action) => + authz.consumeRuntimeTerminationApproval(approvalId, action), + }, + ); + const gateway = new ChatGateway( + {} as never, + {} as never, + {} as never, + {} as never, + {} as never, + {} as never, + authz, + runtime, + durable, + ); + const client = { data: { discordService: true }, emit: vi.fn() }; + + await gateway.handleDiscordApproval( + client as never, + createDiscordIngressEnvelope( + payload('/approve', 'approve-1', 'discord-approve-correlation'), + SERVICE_TOKEN, + ), + ); + const approval = client.emit.mock.calls.find( + ([event]) => event === 'discord:approval', + )?.[1] as { + approvalId: string; + success: boolean; + }; + expect(approval.success).toBe(true); + + await gateway.handleDiscordStop( + client as never, + createDiscordIngressEnvelope( + payload(`/stop ${approval.approvalId}`, 'stop-1', 'discord-stop-correlation'), + SERVICE_TOKEN, + ), + ); + + expect(terminated).toHaveBeenCalledWith( + 'runtime-1', + approval.approvalId, + expect.objectContaining({ actorId: 'mosaic-admin-1' }), + ); + expect(client.emit).toHaveBeenCalledWith('discord:stop', { + correlationId: 'discord-stop-correlation', + success: true, + }); + }); +}); diff --git a/apps/gateway/src/agent/__tests__/runtime-provider-registry.service.test.ts b/apps/gateway/src/agent/__tests__/runtime-provider-registry.service.test.ts index a70eafe..11c57f6 100644 --- a/apps/gateway/src/agent/__tests__/runtime-provider-registry.service.test.ts +++ b/apps/gateway/src/agent/__tests__/runtime-provider-registry.service.test.ts @@ -37,6 +37,7 @@ class RecordingRuntimeProvider implements AgentRuntimeProvider { readonly sentMessages: RuntimeMessage[] = []; terminateCalls = 0; throwAfterSend = false; + throwAuthorization = false; constructor(private readonly supported: RuntimeCapability[]) {} @@ -76,6 +77,9 @@ class RecordingRuntimeProvider implements AgentRuntimeProvider { ): Promise { this.receivedScopes.push(scope); this.sentMessages.push(message); + if (this.throwAuthorization) { + throw Object.assign(new Error('provider authorization denied'), { code: 'forbidden' }); + } if (this.throwAfterSend) { throw new Error('provider acknowledgement failed'); } @@ -295,6 +299,23 @@ describe('RuntimeProviderService security boundary', (): void => { }); }); + it('records a provider authorization rejection as denied rather than provider failure', async (): Promise => { + const provider = new RecordingRuntimeProvider(['session.send']); + provider.throwAuthorization = true; + const audit = new RecordingAuditSink(); + const service = makeService(provider, audit); + + await expect( + service.sendMessage( + 'fleet', + 'session-1', + { content: 'hello', idempotencyKey: 'key-1' }, + CONTEXT, + ), + ).rejects.toThrow(/provider authorization denied/); + expect(audit.events.at(-1)).toMatchObject({ outcome: 'denied', errorCode: 'policy_denied' }); + }); + it('persists only metadata-only runtime audit fields', async (): Promise => { let persisted: unknown; const ingest = async (entry: unknown): Promise => { diff --git a/apps/gateway/src/agent/interaction.controller.test.ts b/apps/gateway/src/agent/interaction.controller.test.ts index 8e7c382..db0dad4 100644 --- a/apps/gateway/src/agent/interaction.controller.test.ts +++ b/apps/gateway/src/agent/interaction.controller.test.ts @@ -1,9 +1,27 @@ +import { firstValueFrom } from 'rxjs'; import { afterEach, describe, expect, it, vi } from 'vitest'; +import { RuntimeApprovalDeniedError } from './runtime-provider-registry.service.js'; +import { RuntimeApprovalDeniedFilter } from './runtime-approval-denied.filter.js'; import { InteractionController } from './interaction.controller.js'; describe('InteractionController', (): void => { afterEach(() => vi.restoreAllMocks()); + it('maps a denied runtime approval to Fastify HTTP 403', () => { + const send = vi.fn(); + const status = vi.fn().mockReturnValue({ send }); + const response = { status }; + const host = { switchToHttp: () => ({ getResponse: () => response }) }; + + new RuntimeApprovalDeniedFilter().catch(new RuntimeApprovalDeniedError(), host as never); + + expect(status).toHaveBeenCalledWith(403); + expect(send).toHaveBeenCalledWith({ + statusCode: 403, + message: 'Runtime termination approval denied', + }); + }); + it('honors a differently named configured instance without a code change', async () => { const prior = process.env['MOSAIC_AGENT_NAME']; process.env['MOSAIC_AGENT_NAME'] = 'Nova'; @@ -30,6 +48,36 @@ describe('InteractionController', (): void => { ); }); + it('enrolls a visible runtime session under the cross-surface conversation handle', async () => { + process.env['MOSAIC_AGENT_NAME'] = 'Nova'; + const runtime = { + listSessions: vi.fn().mockResolvedValue([{ id: 'runtime-1' }]), + }; + const durable = { enroll: vi.fn().mockResolvedValue(undefined) }; + const controller = new InteractionController(runtime as never, durable as never); + + await expect( + controller.enroll( + 'Nova', + 'conversation-1', + { providerId: 'fleet', runtimeSessionId: 'runtime-1' }, + { id: 'owner', tenantId: 'team' }, + 'corr-1', + ), + ).resolves.toEqual({ status: 'enrolled', sessionId: 'conversation-1' }); + expect(durable.enroll).toHaveBeenCalledWith( + { + agentName: 'Nova', + sessionId: 'conversation-1', + tenantId: 'team', + ownerId: 'owner', + providerId: 'fleet', + runtimeSessionId: 'runtime-1', + }, + expect.objectContaining({ correlationId: 'corr-1' }), + ); + }); + it('rejects an invalid attach mode before invoking a provider', async () => { process.env['MOSAIC_AGENT_NAME'] = 'Nova'; const controller = new InteractionController({ attach: vi.fn() } as never, {} as never); @@ -39,6 +87,121 @@ describe('InteractionController', (): void => { ).rejects.toThrow('Interaction attach mode is invalid'); }); + it('resumes a durable session by attaching and streaming its runtime events', async () => { + process.env['MOSAIC_AGENT_NAME'] = 'Nova'; + const runtimeEvent = { + type: 'message.delta' as const, + sessionId: 'runtime-1', + cursor: 'cursor-1', + occurredAt: '2026-07-13T00:00:00.000Z', + content: 'resumed', + }; + const runtime = { + attach: vi.fn().mockResolvedValue({ attachmentId: 'attach-1', sessionId: 'runtime-1' }), + streamSession: vi.fn(async function* () { + yield runtimeEvent; + }), + }; + const durable = { + getSnapshot: vi.fn().mockResolvedValue({ + identity: { agentName: 'Nova', providerId: 'fleet', runtimeSessionId: 'runtime-1' }, + }), + }; + const controller = new InteractionController(runtime as never, durable as never); + + await controller.attach('Nova', 'conversation-1', { mode: 'read' }, { id: 'owner' }, 'corr-1'); + await expect( + firstValueFrom( + controller.stream('Nova', 'conversation-1', undefined, { id: 'owner' }, 'corr-1'), + ), + ).resolves.toEqual({ data: runtimeEvent }); + + expect(runtime.attach).toHaveBeenCalledWith( + 'fleet', + 'runtime-1', + 'read', + expect.objectContaining({ correlationId: 'corr-1' }), + ); + expect(runtime.streamSession).toHaveBeenCalledWith( + 'fleet', + 'runtime-1', + undefined, + expect.objectContaining({ correlationId: 'corr-1' }), + ); + }); + + it('does not create a runtime stream after the SSE subscriber disconnects during snapshot lookup', async () => { + process.env['MOSAIC_AGENT_NAME'] = 'Nova'; + let resolveSnapshot!: (value: { identity: Record }) => void; + const snapshot = new Promise<{ identity: Record }>((resolve) => { + resolveSnapshot = resolve; + }); + const runtime = { streamSession: vi.fn() }; + const durable = { getSnapshot: vi.fn().mockReturnValue(snapshot) }; + const controller = new InteractionController(runtime as never, durable as never); + + const subscription = controller + .stream('Nova', 'conversation-1', undefined, { id: 'owner' }, 'corr-1') + .subscribe(); + subscription.unsubscribe(); + resolveSnapshot({ + identity: { agentName: 'Nova', providerId: 'fleet', runtimeSessionId: 'runtime-1' }, + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(runtime.streamSession).not.toHaveBeenCalled(); + }); + + it.each([ + ['wrong actor', { getSnapshot: vi.fn().mockRejectedValue(new Error('scope mismatch')) }], + [ + 'session-agent mismatch', + { + getSnapshot: vi.fn().mockResolvedValue({ + identity: { agentName: 'Other', providerId: 'fleet', runtimeSessionId: 'runtime-1' }, + }), + }, + ], + ])('denies a CLI stop for %s', async (_reason, durable) => { + process.env['MOSAIC_AGENT_NAME'] = 'Nova'; + const runtime = { terminate: vi.fn().mockResolvedValue(undefined) }; + const controller = new InteractionController(runtime as never, durable as never); + + await expect( + controller.stop( + 'Nova', + 'durable-1', + { approvalRef: 'approval-1' }, + { id: 'owner' }, + 'corr-1', + ), + ).rejects.toBeDefined(); + expect(runtime.terminate).not.toHaveBeenCalled(); + }); + + it('surfaces a denied runtime approval to the CLI interaction surface', async () => { + process.env['MOSAIC_AGENT_NAME'] = 'Nova'; + const runtime = { + terminate: vi.fn().mockRejectedValue(new Error('Runtime termination approval denied')), + }; + const durable = { + getSnapshot: vi.fn().mockResolvedValue({ + identity: { agentName: 'Nova', providerId: 'fleet', runtimeSessionId: 'runtime-1' }, + }), + }; + const controller = new InteractionController(runtime as never, durable as never); + + await expect( + controller.stop( + 'Nova', + 'durable-1', + { approvalRef: 'approval-1' }, + { id: 'owner' }, + 'corr-1', + ), + ).rejects.toThrow('Runtime termination approval denied'); + }); + it('uses the durable session identity and runtime registry for an approved stop', async () => { process.env['MOSAIC_AGENT_NAME'] = 'Nova'; const runtime = { terminate: vi.fn().mockResolvedValue(undefined) }; diff --git a/apps/gateway/src/agent/interaction.controller.ts b/apps/gateway/src/agent/interaction.controller.ts index 5103660..cc247fd 100644 --- a/apps/gateway/src/agent/interaction.controller.ts +++ b/apps/gateway/src/agent/interaction.controller.ts @@ -4,17 +4,21 @@ import { ForbiddenException, Get, Headers, + Sse, Inject, Param, Post, Query, UseGuards, + UseFilters, } from '@nestjs/common'; -import type { RuntimeAttachMode } from '@mosaicstack/types'; +import type { RuntimeAttachMode, RuntimeStreamEvent } from '@mosaicstack/types'; +import { Observable } from 'rxjs'; import { AuthGuard } from '../auth/auth.guard.js'; import { CurrentUser } from '../auth/current-user.decorator.js'; import { scopeFromUser, type AuthenticatedUserLike } from '../auth/session-scope.js'; import { TessDurableSessionService } from './tess-durable-session.service.js'; +import { RuntimeApprovalDeniedFilter } from './runtime-approval-denied.filter.js'; import { RuntimeProviderService, type RuntimeProviderRequestContext, @@ -26,6 +30,7 @@ import { */ @Controller('api/interaction/:agentName') @UseGuards(AuthGuard) +@UseFilters(RuntimeApprovalDeniedFilter) export class InteractionController { constructor( @Inject(RuntimeProviderService) private readonly runtime: RuntimeProviderService, @@ -60,6 +65,41 @@ export class InteractionController { ); } + /** + * Bind an existing, authorized runtime session to the stable conversation ID. + * This is the lifecycle boundary where both runtime identifiers are known. + */ + @Post('sessions/:sessionId/enroll') + async enroll( + @Param('agentName') agentName: string, + @Param('sessionId') sessionId: string, + @Body() body: { providerId?: string; runtimeSessionId?: string } = {}, + @CurrentUser() user: AuthenticatedUserLike, + @Headers('x-correlation-id') correlationId?: string, + ) { + this.assertConfiguredAgent(agentName); + const providerId = this.requiredProvider(body.providerId ?? ''); + const runtimeSessionId = body.runtimeSessionId?.trim(); + if (!runtimeSessionId) throw new ForbiddenException('Runtime session identity is required'); + const context = this.context(user, correlationId); + const sessions = await this.runtime.listSessions(providerId, context); + if (!sessions.some((session): boolean => session.id === runtimeSessionId)) { + throw new ForbiddenException('Runtime session is not visible to this actor'); + } + await this.durable.enroll( + { + agentName, + sessionId, + tenantId: context.actorScope.tenantId, + ownerId: context.actorScope.userId, + providerId, + runtimeSessionId, + }, + context, + ); + return { status: 'enrolled', sessionId }; + } + @Post('sessions/:sessionId/attach') async attach( @Param('agentName') agentName: string, @@ -84,6 +124,53 @@ export class InteractionController { ); } + @Sse('sessions/:sessionId/stream') + stream( + @Param('agentName') agentName: string, + @Param('sessionId') sessionId: string, + @Query('cursor') cursor: string | undefined, + @CurrentUser() user: AuthenticatedUserLike, + @Headers('x-correlation-id') correlationId?: string, + ): Observable<{ data: RuntimeStreamEvent }> { + this.assertConfiguredAgent(agentName); + const context = this.context(user, correlationId); + return new Observable((subscriber) => { + let iterator: AsyncIterator | undefined; + let cancelled = false; + void (async (): Promise => { + try { + const snapshot = await this.durable.getSnapshot(sessionId, context); + if (cancelled || subscriber.closed) return; + this.assertSessionAgent(snapshot.identity.agentName, agentName); + iterator = this.runtime + .streamSession( + snapshot.identity.providerId, + snapshot.identity.runtimeSessionId, + cursor?.trim() || undefined, + context, + ) + [Symbol.asyncIterator](); + if (cancelled || subscriber.closed) { + await iterator.return?.(); + return; + } + while (!cancelled && !subscriber.closed) { + const next = await iterator.next(); + if (next.done || cancelled || subscriber.closed) break; + subscriber.next({ data: next.value }); + } + if (!subscriber.closed) subscriber.complete(); + } catch (error: unknown) { + if (!subscriber.closed) subscriber.error(error); + } + })(); + return (): void => { + cancelled = true; + void iterator?.return?.(); + }; + }); + } + @Post('sessions/:sessionId/send') async send( @Param('agentName') agentName: string, diff --git a/apps/gateway/src/agent/runtime-approval-denied.filter.ts b/apps/gateway/src/agent/runtime-approval-denied.filter.ts new file mode 100644 index 0000000..22aae38 --- /dev/null +++ b/apps/gateway/src/agent/runtime-approval-denied.filter.ts @@ -0,0 +1,13 @@ +import { Catch, type ArgumentsHost, type ExceptionFilter } from '@nestjs/common'; +import { RuntimeApprovalDeniedError } from './runtime-provider-registry.service.js'; + +/** Maps a consumed/missing runtime approval to a stable HTTP authorization response. */ +@Catch(RuntimeApprovalDeniedError) +export class RuntimeApprovalDeniedFilter implements ExceptionFilter { + catch(_exception: RuntimeApprovalDeniedError, host: ArgumentsHost): void { + const response = host.switchToHttp().getResponse<{ + status(code: number): { send(body: { statusCode: number; message: string }): void }; + }>(); + response.status(403).send({ statusCode: 403, message: 'Runtime termination approval denied' }); + } +} diff --git a/apps/gateway/src/agent/runtime-provider-registry.service.ts b/apps/gateway/src/agent/runtime-provider-registry.service.ts index cb16a29..b824bb0 100644 --- a/apps/gateway/src/agent/runtime-provider-registry.service.ts +++ b/apps/gateway/src/agent/runtime-provider-registry.service.ts @@ -77,7 +77,7 @@ function configuredAgentName(): string { return agentName; } -class RuntimeApprovalDeniedError extends Error { +export class RuntimeApprovalDeniedError extends Error { constructor() { super('Runtime termination approval denied'); } @@ -301,7 +301,7 @@ export class RuntimeProviderService { return result; } catch (error: unknown) { const durationMs = Date.now() - startedAt; - if (invocationStarted && !(error instanceof RuntimeApprovalDeniedError)) { + if (invocationStarted && !this.isAuthorizationDenied(error)) { await this.recordFailure(providerId, operation, scope, resourceId, durationMs); } else { await this.record( @@ -360,6 +360,17 @@ export class RuntimeProviderService { } } + private isAuthorizationDenied(error: unknown): boolean { + return ( + error instanceof RuntimeApprovalDeniedError || + error instanceof ForbiddenException || + (typeof error === 'object' && + error !== null && + 'code' in error && + (error as { code?: unknown }).code === 'forbidden') + ); + } + private provider(providerId: string): AgentRuntimeProvider { try { return this.registry.require(providerId); diff --git a/apps/gateway/src/agent/tess-durable-session.repository.ts b/apps/gateway/src/agent/tess-durable-session.repository.ts index 8d2f52c..52d853b 100644 --- a/apps/gateway/src/agent/tess-durable-session.repository.ts +++ b/apps/gateway/src/agent/tess-durable-session.repository.ts @@ -50,9 +50,20 @@ export class TessDurableSessionRepository implements DurableSessionStore { .onConflictDoNothing(); const existing = await this.session(identity.sessionId); - if (!existing || !sameIdentity(existing, identity)) { + if (!existing || !sameEnrollmentScope(existing, identity)) { throw new Error(`Durable Tess session identity conflict: ${identity.sessionId}`); } + // A recovered/re-enrolled runtime can receive a new provider session ID; + // the conversation handle and owner scope remain immutable. + if ( + existing.providerId !== identity.providerId || + existing.runtimeSessionId !== identity.runtimeSessionId + ) { + await this.db + .update(interactionSessions) + .set({ providerId: identity.providerId, runtimeSessionId: identity.runtimeSessionId }) + .where(eq(interactionSessions.id, identity.sessionId)); + } } async snapshot(sessionId: string): Promise { @@ -427,14 +438,12 @@ function matchesCheckpointDigest(stored: string, digest: string): boolean { return stored === digest; } -function sameIdentity(left: DurableSessionIdentity, right: DurableSessionIdentity): boolean { +function sameEnrollmentScope(left: DurableSessionIdentity, right: DurableSessionIdentity): boolean { return ( left.agentName === right.agentName && left.sessionId === right.sessionId && left.tenantId === right.tenantId && - left.ownerId === right.ownerId && - left.providerId === right.providerId && - left.runtimeSessionId === right.runtimeSessionId + left.ownerId === right.ownerId ); } diff --git a/apps/gateway/src/agent/tess-durable-session.service.ts b/apps/gateway/src/agent/tess-durable-session.service.ts index 2dc3619..340694f 100644 --- a/apps/gateway/src/agent/tess-durable-session.service.ts +++ b/apps/gateway/src/agent/tess-durable-session.service.ts @@ -1,5 +1,5 @@ import { ForbiddenException, Inject, Injectable } from '@nestjs/common'; -import { DurableSessionCoordinator } from '@mosaicstack/agent'; +import { DurableSessionCoordinator, type DurableSessionIdentity } from '@mosaicstack/agent'; import type { TessProviderOutboxDto } from './tess-durable-session.dto.js'; import { TessDurableSessionRepository } from './tess-durable-session.repository.js'; import { @@ -23,6 +23,20 @@ export class TessDurableSessionService { this.coordinator = new DurableSessionCoordinator(repository); } + /** Enroll a verified runtime session under the stable cross-surface conversation handle. */ + async enroll( + identity: DurableSessionIdentity, + context: RuntimeProviderRequestContext, + ): Promise { + if ( + identity.ownerId !== context.actorScope.userId || + identity.tenantId !== context.actorScope.tenantId + ) { + throw new ForbiddenException('Durable Tess enrollment scope mismatch'); + } + await this.coordinator.create(identity); + } + async queueProviderSend(input: TessProviderOutboxDto): Promise { const snapshot = await this.coordinator.snapshot(input.sessionId); this.assertScope(snapshot.identity.ownerId, snapshot.identity.tenantId, input); diff --git a/apps/gateway/src/chat/chat.gateway.ts b/apps/gateway/src/chat/chat.gateway.ts index f3fd1b5..a7549e4 100644 --- a/apps/gateway/src/chat/chat.gateway.ts +++ b/apps/gateway/src/chat/chat.gateway.ts @@ -32,7 +32,12 @@ import type { AbortPayload, } from '@mosaicstack/types'; import { AgentService, type ConversationHistoryMessage } from '../agent/agent.service.js'; -import { RuntimeProviderService } from '../agent/runtime-provider-registry.service.js'; +import { + RUNTIME_PROVIDER_AUDIT_SINK, + RuntimeProviderService, + type RuntimeAuditSink, +} from '../agent/runtime-provider-registry.service.js'; +import { TessDurableSessionService } from '../agent/tess-durable-session.service.js'; import { AUTH } from '../auth/auth.tokens.js'; import { scopeFromUser, @@ -134,6 +139,12 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa @Optional() @Inject(RuntimeProviderService) private readonly runtimeRegistry: RuntimeProviderService | null = null, + @Optional() + @Inject(TessDurableSessionService) + private readonly durableSessions: TessDurableSessionService | null = null, + @Optional() + @Inject(RUNTIME_PROVIDER_AUDIT_SINK) + private readonly runtimeAudit: RuntimeAuditSink | null = null, ) {} afterInit(): void { @@ -656,9 +667,16 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa ): Promise { if (!client.data.discordService) return; const ingress = this.resolveDiscordIngress(client, envelope, 'approve'); - const actionParts = ingress?.content.match(/^\/approve\s+([^\s]+)\s+([^\s]+)$/i); + const isApprovalCommand = /^\/approve\s*$/i.test(ingress?.content ?? ''); const tenantId = process.env['DISCORD_SERVICE_TENANT_ID']?.trim(); - if (!ingress || !actionParts || !tenantId || !this.commandAuthorization) return; + if ( + !ingress || + !isApprovalCommand || + !tenantId || + !this.commandAuthorization || + !this.durableSessions + ) + return; const binding = resolveDiscordInteractionBinding( parseDiscordInteractionBindings(process.env['DISCORD_INTERACTION_BINDINGS']), ingress.guildId, @@ -680,22 +698,63 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa }); return; } - const providerId = actionParts[1]!; - const sessionId = actionParts[2]!; + let snapshot; + try { + snapshot = await this.durableSessions.getSnapshot(ingress.conversationId, { + actorScope: { userId: actorId, tenantId }, + channelId: ingress.channelId, + correlationId: ingress.correlationId, + }); + } catch { + client.emit('discord:approval', { + correlationId: ingress.correlationId, + success: false, + approvalId: undefined, + expiresAt: undefined, + }); + return; + } + if (snapshot.identity.agentName !== agentName) { + client.emit('discord:approval', { + correlationId: ingress.correlationId, + success: false, + approvalId: undefined, + expiresAt: undefined, + }); + return; + } const approval = await this.commandAuthorization.createRuntimeTerminationApproval({ - providerId, - sessionId, + providerId: snapshot.identity.providerId, + sessionId: snapshot.identity.runtimeSessionId, actorId, tenantId, channelId: ingress.channelId, correlationId: this.discordRuntimeActionCorrelation( binding.instanceId, ingress, - providerId, - sessionId, + snapshot.identity.providerId, + snapshot.identity.runtimeSessionId, ), agentName, }); + if (!approval) { + await this.runtimeAudit?.record({ + providerId: snapshot.identity.providerId, + operation: 'session.terminate', + outcome: 'denied', + actorId, + tenantId, + channelId: ingress.channelId, + correlationId: this.discordRuntimeActionCorrelation( + binding.instanceId, + ingress, + snapshot.identity.providerId, + snapshot.identity.runtimeSessionId, + ), + resourceId: snapshot.identity.runtimeSessionId, + errorCode: 'policy_denied', + }); + } client.emit('discord:approval', { correlationId: ingress.correlationId, success: approval !== null, @@ -711,9 +770,10 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa ): Promise { if (!client.data.discordService) return; const ingress = this.resolveDiscordIngress(client, envelope, 'stop'); - const actionParts = ingress?.content.match(/^\/stop\s+([^\s]+)\s+([^\s]+)\s+([^\s]+)$/i); + const approvalRef = /^\/stop\s+([^\s]+)$/i.exec(ingress?.content ?? '')?.[1]; const tenantId = process.env['DISCORD_SERVICE_TENANT_ID']?.trim(); - if (!ingress || !actionParts || !tenantId || !this.runtimeRegistry) return; + if (!ingress || !approvalRef || !tenantId || !this.runtimeRegistry || !this.durableSessions) + return; const binding = resolveDiscordInteractionBinding( parseDiscordInteractionBindings(process.env['DISCORD_INTERACTION_BINDINGS']), ingress.guildId, @@ -723,25 +783,31 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa ); const actorId = binding && resolveDiscordInteractionActorId(binding, ingress.userId); if (!actorId) return; - const providerId = actionParts[1]!; - const sessionId = actionParts[2]!; try { + const context = { + actorScope: { userId: actorId, tenantId }, + channelId: ingress.channelId, + correlationId: ingress.correlationId, + }; + const snapshot = await this.durableSessions.getSnapshot(ingress.conversationId, context); + if (snapshot.identity.agentName !== binding.instanceId) throw new Error('agent mismatch'); // RuntimeProviderService consumes the durable approval exactly once using the // provisioned approving-admin identity, never the Discord service account. - await this.runtimeRegistry.terminate(providerId, sessionId, actionParts[3]!, { - actorScope: { - userId: actorId, - tenantId, + await this.runtimeRegistry.terminate( + snapshot.identity.providerId, + snapshot.identity.runtimeSessionId, + approvalRef, + { + ...context, + correlationId: this.discordRuntimeActionCorrelation( + binding.instanceId, + ingress, + snapshot.identity.providerId, + snapshot.identity.runtimeSessionId, + ), }, - channelId: ingress.channelId, - correlationId: this.discordRuntimeActionCorrelation( - binding.instanceId, - ingress, - providerId, - sessionId, - ), - }); + ); client.emit('discord:stop', { correlationId: ingress.correlationId, success: true }); } catch { client.emit('discord:stop', { correlationId: ingress.correlationId, success: false }); diff --git a/apps/gateway/src/plugin/discord-ingress.security.spec.ts b/apps/gateway/src/plugin/discord-ingress.security.spec.ts index 3ceb840..0c8bb46 100644 --- a/apps/gateway/src/plugin/discord-ingress.security.spec.ts +++ b/apps/gateway/src/plugin/discord-ingress.security.spec.ts @@ -77,9 +77,17 @@ function discordGateway(role: 'admin' | 'member'): { gateway: ChatGateway; client: { data: { discordService: boolean }; emit: ReturnType }; consumedActions: Array<{ actorId: string; correlationId: string }>; + durable: { getSnapshot: ReturnType }; + audit: { record: ReturnType }; } { const authorization = commandAuthorization(role); const consumedActions: Array<{ actorId: string; correlationId: string }> = []; + const durable = { + getSnapshot: vi.fn().mockResolvedValue({ + identity: { agentName: 'Nova', providerId: 'fleet', runtimeSessionId: 'runtime-1' }, + }), + }; + const audit = { record: vi.fn().mockResolvedValue(undefined) }; const runtimeRegistry = new RuntimeProviderService( { require: () => ({ @@ -105,9 +113,13 @@ function discordGateway(role: 'admin' | 'member'): { {} as never, authorization, runtimeRegistry, + durable as never, + audit as never, ), client: { data: { discordService: true }, emit: vi.fn() }, consumedActions, + durable, + audit, }; } @@ -241,7 +253,7 @@ describe('Discord ingress security', () => { const { gateway, client, consumedActions } = discordGateway('admin'); await gateway.handleDiscordApproval( client as never, - ingressEnvelope('/approve fleet runtime-1', 'approve-message', { + ingressEnvelope('/approve', 'approve-message', { correlationId: 'approval-ingress-correlation', }), ); @@ -255,7 +267,7 @@ describe('Discord ingress security', () => { await gateway.handleDiscordStop( client as never, - ingressEnvelope(`/stop fleet runtime-1 ${approval.approvalId}`, 'stop-message', { + ingressEnvelope(`/stop ${approval.approvalId}`, 'stop-message', { correlationId: 'stop-ingress-correlation', }), ); @@ -271,14 +283,13 @@ describe('Discord ingress security', () => { ]); }); - it('rejects approval when the binding targets a different runtime agent', async () => { + it('audits a Discord mint-side authorization denial', async () => { configureDiscordEnv(); - process.env['MOSAIC_AGENT_NAME'] = 'Other'; - const { gateway, client } = discordGateway('admin'); + const { gateway, client, audit } = discordGateway('member'); await gateway.handleDiscordApproval( client as never, - ingressEnvelope('/approve fleet runtime-1', 'mismatched-agent-approve'), + ingressEnvelope('/approve', 'denied-approve'), ); expect(client.emit).toHaveBeenCalledWith('discord:approval', { @@ -287,14 +298,57 @@ describe('Discord ingress security', () => { approvalId: undefined, expiresAt: undefined, }); + expect(audit.record).toHaveBeenCalledWith( + expect.objectContaining({ + outcome: 'denied', + operation: 'session.terminate', + errorCode: 'policy_denied', + }), + ); }); + it.each([ + [ + 'binding', + () => { + process.env['MOSAIC_AGENT_NAME'] = 'Other'; + }, + ], + [ + 'durable session', + (durable: { getSnapshot: ReturnType }) => { + durable.getSnapshot.mockResolvedValueOnce({ + identity: { agentName: 'Other', providerId: 'fleet', runtimeSessionId: 'runtime-1' }, + }); + }, + ], + ])( + 'rejects approval when the %s targets a different runtime agent', + async (_source, configure) => { + configureDiscordEnv(); + const { gateway, client, durable } = discordGateway('admin'); + configure(durable); + + await gateway.handleDiscordApproval( + client as never, + ingressEnvelope('/approve', 'mismatched-agent-approve'), + ); + + expect(client.emit).toHaveBeenCalledWith('discord:approval', { + correlationId: 'correlation-001', + success: false, + approvalId: undefined, + expiresAt: undefined, + }); + }, + ); + it('rejects unpaired and non-admin Discord users for approval and stop', async () => { configureDiscordEnv(); const { gateway, client } = discordGateway('member'); await gateway.handleDiscordApproval( client as never, - ingressEnvelope('/approve fleet runtime-1', 'member-approve'), + ingressEnvelope('/approve', 'member-approve'), ); expect(client.emit).toHaveBeenCalledWith('discord:approval', { correlationId: 'correlation-001', @@ -306,7 +360,7 @@ describe('Discord ingress security', () => { process.env['DISCORD_INTERACTION_BINDINGS'] = JSON.stringify([]); await gateway.handleDiscordStop( client as never, - ingressEnvelope('/stop fleet runtime-1 forged', 'unpaired-stop'), + ingressEnvelope('/stop forged', 'unpaired-stop'), ); expect(client.emit).not.toHaveBeenCalledWith('discord:stop', expect.anything()); }); @@ -316,7 +370,7 @@ describe('Discord ingress security', () => { const { gateway, client } = discordGateway('admin'); await gateway.handleDiscordApproval( client as never, - ingressEnvelope('/approve fleet runtime-1', 'replay-approve', { + ingressEnvelope('/approve', 'replay-approve', { correlationId: 'replay-approval-correlation', }), ); @@ -327,13 +381,13 @@ describe('Discord ingress security', () => { }; await gateway.handleDiscordStop( client as never, - ingressEnvelope(`/stop fleet runtime-1 ${approval.approvalId}`, 'replay-stop-one', { + ingressEnvelope(`/stop ${approval.approvalId}`, 'replay-stop-one', { correlationId: 'replay-stop-correlation-one', }), ); await gateway.handleDiscordStop( client as never, - ingressEnvelope(`/stop fleet runtime-1 ${approval.approvalId}`, 'replay-stop-two', { + ingressEnvelope(`/stop ${approval.approvalId}`, 'replay-stop-two', { correlationId: 'replay-stop-correlation-two', }), ); diff --git a/docs/scratchpads/tess-20260712.md b/docs/scratchpads/tess-20260712.md index 819d5c5..5f21ca9 100644 --- a/docs/scratchpads/tess-20260712.md +++ b/docs/scratchpads/tess-20260712.md @@ -56,3 +56,11 @@ **Final focused review:** PASS. Deterministic audit validated all task repository roots with zero missing paths; no planning placeholders remained; security prerequisites still gate Tess exposure; observability traceability is explicit. **Current gate:** planning PR must merge to `main` with terminal-green CI before any source-code worker starts. + +## 2026-07-13 — M3 cross-surface delivery + +**Branch:** `feat/tess-m3-integration` from `main` at `84d884b9`. + +**Delivered:** Stable Discord `conversationId` enrollment after a visible provider/runtime session is known; idempotent provider-session rebinding that preserves agent/tenant/owner scope; Discord approval/stop target resolution through the durable snapshot; SSE runtime streaming after CLI attach; denial/audit parity including provider authorization denials and HTTP 403 approval-denial mapping. + +**Evidence:** Gateway targeted suite: 37 tests passed; Mosaic CLI interaction test passed; agent durable-session test passed; gateway and CLI typechecks passed; changed-file format and whitespace checks passed. Codex security review found no confident vulnerability. Code review identified a Fastify exception-response mismatch and two UX/acknowledgement issues; all were corrected before the final validation run. diff --git a/packages/agent/src/tess-durable-session.test.ts b/packages/agent/src/tess-durable-session.test.ts index 64e5ca8..3cadaf4 100644 --- a/packages/agent/src/tess-durable-session.test.ts +++ b/packages/agent/src/tess-durable-session.test.ts @@ -64,6 +64,23 @@ describe('DurableSessionCoordinator', () => { expect(recovered.handoffs).toMatchObject([{ handoffId: 'handoff-1', status: 'pending' }]); }); + it('rebinds a recovered runtime while preserving the immutable conversation owner scope', async () => { + const coordinator = new DurableSessionCoordinator(new InMemoryDurableSessionStore()); + await coordinator.create(IDENTITY); + await coordinator.create({ + ...IDENTITY, + providerId: 'fleet-next', + runtimeSessionId: 'nova-next', + }); + + await expect(coordinator.snapshot(IDENTITY.sessionId)).resolves.toMatchObject({ + identity: { ...IDENTITY, providerId: 'fleet-next', runtimeSessionId: 'nova-next' }, + }); + await expect(coordinator.create({ ...IDENTITY, ownerId: 'other-owner' })).rejects.toThrow( + /identity conflict/, + ); + }); + it('deduplicates duplicate ingress and never reprocesses an inbox record after restart or compaction', async () => { const store = new InMemoryDurableSessionStore(); const firstProcess = new DurableSessionCoordinator(store); diff --git a/packages/agent/src/tess-durable-session.ts b/packages/agent/src/tess-durable-session.ts index 039fe06..adc84c6 100644 --- a/packages/agent/src/tess-durable-session.ts +++ b/packages/agent/src/tess-durable-session.ts @@ -256,9 +256,11 @@ export class InMemoryDurableSessionStore implements DurableSessionStore { async create(identity: DurableSessionIdentity): Promise { const existing = this.sessions.get(identity.sessionId); if (existing) { - if (!identitiesEqual(existing.identity, identity)) { + if (!sameEnrollmentScope(existing.identity, identity)) { throw new Error(`Durable Tess session identity conflict: ${identity.sessionId}`); } + existing.identity.providerId = identity.providerId; + existing.identity.runtimeSessionId = identity.runtimeSessionId; return; } this.sessions.set(identity.sessionId, { @@ -416,14 +418,12 @@ export class InMemoryDurableSessionStore implements DurableSessionStore { } } -function identitiesEqual(left: DurableSessionIdentity, right: DurableSessionIdentity): boolean { +function sameEnrollmentScope(left: DurableSessionIdentity, right: DurableSessionIdentity): boolean { return ( left.agentName === right.agentName && left.sessionId === right.sessionId && left.tenantId === right.tenantId && - left.ownerId === right.ownerId && - left.providerId === right.providerId && - left.runtimeSessionId === right.runtimeSessionId + left.ownerId === right.ownerId ); } diff --git a/packages/mosaic/src/commands/interaction.test.ts b/packages/mosaic/src/commands/interaction.test.ts index 79cd774..856a375 100644 --- a/packages/mosaic/src/commands/interaction.test.ts +++ b/packages/mosaic/src/commands/interaction.test.ts @@ -10,6 +10,7 @@ describe('generic interaction CLI', (): void => { expect(command.commands.map((item) => item.name()).sort()).toEqual([ 'attach', 'chat', + 'enroll', 'health', 'recover', 'send', diff --git a/packages/mosaic/src/commands/interaction.ts b/packages/mosaic/src/commands/interaction.ts index 8b53a70..2e543d0 100644 --- a/packages/mosaic/src/commands/interaction.ts +++ b/packages/mosaic/src/commands/interaction.ts @@ -3,6 +3,7 @@ import type { Command } from 'commander'; import { withAuth } from './with-auth.js'; import { attachInteractionSession, + enrollInteractionSession, fetchInteractionHealth, fetchInteractionSessions, fetchInteractionStatus, @@ -10,6 +11,7 @@ import { recoverInteractionSession, sendInteractionMessage, stopInteractionSession, + streamInteractionSession, } from '../tui/gateway-api.js'; interface InteractionOptions { @@ -103,18 +105,48 @@ export function registerInteractionCommand(program: Command): Command { }); options( - command.command('attach ').description('Create a scoped read or write attachment'), + command + .command('enroll ') + .description('Bind an authorized runtime session to a durable conversation handle'), + ).action( + async ( + sessionId: string, + providerId: string, + runtimeSessionId: string, + opts: InteractionOptions, + ) => { + const request = await authRequest(opts); + print( + await enrollInteractionSession(request.gateway, request.cookie, { + ...request, + sessionId, + providerId, + runtimeSessionId, + }), + ); + }, + ); + + options( + command + .command('attach ') + .description('Create a scoped read or write attachment and stream the session'), ) .option('--control', 'Request control mode (provider policy may deny it)') .action(async (sessionId: string, opts: InteractionOptions & { control?: boolean }) => { const request = await authRequest(opts); - print( - await attachInteractionSession(request.gateway, request.cookie, { - ...request, - sessionId, - mode: opts.control ? 'control' : 'read', - }), - ); + const attachment = await attachInteractionSession(request.gateway, request.cookie, { + ...request, + sessionId, + mode: opts.control ? 'control' : 'read', + }); + print(attachment); + for await (const event of streamInteractionSession(request.gateway, request.cookie, { + ...request, + sessionId, + })) { + print(event); + } }); const send = options( diff --git a/packages/mosaic/src/tui/gateway-api.ts b/packages/mosaic/src/tui/gateway-api.ts index d5ad4dd..7e169bd 100644 --- a/packages/mosaic/src/tui/gateway-api.ts +++ b/packages/mosaic/src/tui/gateway-api.ts @@ -406,6 +406,32 @@ export async function fetchInteractionTree( return handleResponse(res, 'Failed to get interaction session tree'); } +export async function enrollInteractionSession( + gatewayUrl: string, + sessionCookie: string, + request: InteractionRequest & { + sessionId: string; + providerId: string; + runtimeSessionId: string; + }, +): Promise<{ status: string; sessionId: string }> { + const res = await fetch( + `${gatewayUrl}${interactionPath(request.agentName, `/sessions/${encodeURIComponent(request.sessionId)}/enroll`)}`, + { + method: 'POST', + headers: interactionHeaders(sessionCookie, gatewayUrl, request.correlationId), + body: JSON.stringify({ + providerId: request.providerId, + runtimeSessionId: request.runtimeSessionId, + }), + }, + ); + return handleResponse<{ status: string; sessionId: string }>( + res, + 'Failed to enroll interaction session', + ); +} + export async function attachInteractionSession( gatewayUrl: string, sessionCookie: string, @@ -422,6 +448,49 @@ export async function attachInteractionSession( return handleResponse(res, 'Failed to attach interaction session'); } +export async function* streamInteractionSession( + gatewayUrl: string, + sessionCookie: string, + request: InteractionRequest & { sessionId: string; cursor?: string }, +): AsyncIterable { + const params = request.cursor?.trim() + ? `?${new URLSearchParams({ cursor: request.cursor.trim() }).toString()}` + : ''; + const res = await fetch( + `${gatewayUrl}${interactionPath(request.agentName, `/sessions/${encodeURIComponent(request.sessionId)}/stream${params}`)}`, + { headers: interactionHeaders(sessionCookie, gatewayUrl, request.correlationId) }, + ); + if (!res.ok || !res.body) { + const body = await res.text().catch(() => ''); + throw new Error(`Failed to stream interaction session (${res.status}): ${body}`); + } + + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + let pending = ''; + try { + for (;;) { + const chunk = await reader.read(); + if (chunk.done) break; + pending += decoder.decode(chunk.value, { stream: true }); + for (;;) { + const separator = pending.indexOf('\n\n'); + if (separator < 0) break; + const frame = pending.slice(0, separator); + pending = pending.slice(separator + 2); + const data = frame + .split('\n') + .find((line: string): boolean => line.startsWith('data:')) + ?.slice('data:'.length) + .trim(); + if (data) yield JSON.parse(data) as unknown; + } + } + } finally { + reader.releaseLock(); + } +} + export async function sendInteractionMessage( gatewayUrl: string, sessionCookie: string,