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

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<string, string | undefined>();
+
+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<string, string>();
+  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,
+    });
+  });
+});

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.ownerId === right.ownerId
-    left.providerId === right.providerId &&
-    left.runtimeSessionId === right.runtimeSessionId
   );
 }
 

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

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<void> {
     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]!;
+    let snapshot;
-    const sessionId = actionParts[2]!;
+    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,
+      providerId: snapshot.identity.providerId,
-      sessionId,
+      sessionId: snapshot.identity.runtimeSessionId,
       actorId,
       tenantId,
       channelId: ingress.channelId,
       correlationId: this.discordRuntimeActionCorrelation(
         binding.instanceId,
         ingress,
-        providerId,
+        snapshot.identity.providerId,
-        sessionId,
+        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<void> {
     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]!, {
+      await this.runtimeRegistry.terminate(
-        actorScope: {
+        snapshot.identity.providerId,
-          userId: actorId,
+        snapshot.identity.runtimeSessionId,
-          tenantId,
+        approvalRef,
-        },
+        {
-        channelId: ingress.channelId,
+          ...context,
           correlationId: this.discordRuntimeActionCorrelation(
             binding.instanceId,
             ingress,
-          providerId,
+            snapshot.identity.providerId,
-          sessionId,
+            snapshot.identity.runtimeSessionId,
           ),
-      });
+        },
+      );
       client.emit('discord:stop', { correlationId: ingress.correlationId, success: true });
     } catch {
       client.emit('discord:stop', { correlationId: ingress.correlationId, success: false });

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<typeof vi.fn> };
   consumedActions: Array<{ actorId: string; correlationId: string }>;
+  durable: { getSnapshot: ReturnType<typeof vi.fn> };
+  audit: { record: ReturnType<typeof vi.fn> };
 } {
   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, audit } = discordGateway('member');
-    const { gateway, client } = discordGateway('admin');
 
     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<typeof vi.fn> }) => {
+        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',
       }),
     );

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.

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

packages/agent/src/tess-durable-session.ts

@@ -256,9 +256,11 @@ export class InMemoryDurableSessionStore implements DurableSessionStore {
   async create(identity: DurableSessionIdentity): Promise<void> {
     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.ownerId === right.ownerId
-    left.providerId === right.providerId &&
-    left.runtimeSessionId === right.runtimeSessionId
   );
 }
 

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',

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 <sessionId>').description('Create a scoped read or write attachment'),
+    command
+      .command('enroll <sessionId> <provider> <runtimeSessionId>')
+      .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 <sessionId>')
+      .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(
+      const attachment = await attachInteractionSession(request.gateway, request.cookie, {
-        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(

packages/mosaic/src/tui/gateway-api.ts

@@ -406,6 +406,32 @@ export async function fetchInteractionTree(
   return handleResponse<unknown[]>(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<unknown>(res, 'Failed to attach interaction session');
 }
 
+export async function* streamInteractionSession(
+  gatewayUrl: string,
+  sessionCookie: string,
+  request: InteractionRequest & { sessionId: string; cursor?: string },
+): AsyncIterable<unknown> {
+  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,