feat(tess): wire durable interaction surfaces (#732)

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<void> {
     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<void> => {
+    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<void> => {
     let persisted: unknown;
     const ingest = async (entry: unknown): Promise<unknown> => {

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<string, string> }) => void;
+    const snapshot = new Promise<{ identity: Record<string, string> }>((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) };

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<RuntimeStreamEvent> | undefined;
+      let cancelled = false;
+      void (async (): Promise<void> => {
+        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,

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' });
+  }
+}

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);

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<DurableSessionSnapshot | null> {
@@ -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
   );
 }
 

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<void> {
+    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<void> {
     const snapshot = await this.coordinator.snapshot(input.sessionId);
     this.assertScope(snapshot.identity.ownerId, snapshot.identity.tenantId, input);