refactor(chat): route browser chat through one runtime

Co-Authored-By: Claude Opus 4.8 <[email protected]>
Claude-Session: https://claude.ai/code/session_01ESFAnh2t9HmLwng8oW95St

apps/gateway/src/agent/__tests__/session-ownership.test.ts

@@ -1,7 +1,7 @@
 import 'reflect-metadata';
 import { readFileSync } from 'node:fs';
 import { resolve } from 'node:path';
-import { NotFoundException } from '@nestjs/common';
+import { ForbiddenException, NotFoundException } from '@nestjs/common';
 import { Test, type TestingModule } from '@nestjs/testing';
 import { describe, expect, it, vi } from 'vitest';
 
@@ -28,6 +28,8 @@ import { CommandExecutorService } from '../../commands/command-executor.service.
 import { RoutingEngineService } from '../routing/routing-engine.service.js';
 import { ChatRuntimeRouter } from '../../chat/chat-runtime-router.js';
 import { EmbeddedChatRuntime } from '../../chat/embedded-chat.runtime.js';
+import { ownConversation } from '../../chat/chat-runtime.js';
+import type { LegacyRuntimeStream } from '../../chat/chat-runtime.js';
 import { HarnessChatRuntime } from '../../chat/harness-chat.runtime.js';
 import { HarnessRegistry } from '../../harness/harness.registry.js';
 import { HARNESS_CONVERSATION_SERVICE_UNAVAILABLE } from '../../harness/harness.tokens.js';
@@ -650,3 +652,152 @@ describe('TESS-M1-SEC-002 WebSocket session ownership and tenant binding (router
     expect(harnessConversation.send).not.toHaveBeenCalled();
   });
 });
+
+// ---------------------------------------------------------------------------
+// Task-5 AMEND — embedded runtime lease lifecycle (G1) + ownership collapse (G5).
+// These drive the real EmbeddedChatRuntime directly over a shape-complete AgentService
+// fake (every touched method exists, so a RED can only come from behavior, never a
+// `getSession is not a function` TypeError). Ownership context is minted through the
+// real `ownConversation` factory — the only sanctioned way to reach a port op.
+// ---------------------------------------------------------------------------
+
+const EMBEDDED_SCOPE = { userId: USER_A.id, tenantId: USER_A.tenantId };
+const CONVERSATION_UNAVAILABLE_RESULT = {
+  ok: false,
+  code: 'conversation_unavailable',
+  retryable: false,
+} as const;
+
+/** A stream sink; `channelId` is server-derived, `onEvent` records nothing here. */
+function makeStream(): LegacyRuntimeStream {
+  return { channelId: 'websocket:test-1', onEvent: vi.fn() };
+}
+
+/**
+ * getSession → undefined (session missing), createSession → rejects with `err`. Exercises the
+ * `resolveOrCreate` collapse branch. `prompt` exists so its ABSENCE from the call record proves
+ * the turn short-circuited before any dispatch.
+ */
+function makeCollapsingAgentService(err: Error) {
+  return {
+    getSession: vi.fn(() => undefined),
+    createSession: vi.fn().mockRejectedValue(err),
+    onEvent: vi.fn(() => vi.fn()),
+    addChannel: vi.fn(),
+    removeChannel: vi.fn(),
+    prompt: vi.fn().mockResolvedValue(undefined),
+    recordTokenUsage: vi.fn(),
+  };
+}
+
+/** getSession → a live owned session, so `resolveOrCreate` succeeds and a lease is built. */
+function makeLeaseAgentService() {
+  const session = makeAgentSession(USER_A);
+  const unsubscribe = vi.fn();
+  const svc = {
+    getSession: vi.fn(() => session),
+    createSession: vi.fn(),
+    onEvent: vi.fn(() => unsubscribe),
+    addChannel: vi.fn(),
+    removeChannel: vi.fn(),
+    prompt: vi.fn().mockResolvedValue(undefined),
+    recordTokenUsage: vi.fn(),
+  };
+  return { svc, unsubscribe, session };
+}
+
+describe('TESS Task-5 embedded ownership collapse (missing and foreign are indistinguishable, never throw)', () => {
+  const ctx = ownConversation(CONVERSATION_ID, EMBEDDED_SCOPE);
+
+  it('collapses a foreign (Forbidden) create to conversation_unavailable and never throws', async () => {
+    const svc = makeCollapsingAgentService(new ForbiddenException('foreign owner'));
+    const runtime = new EmbeddedChatRuntime(svc as never);
+
+    const result = await runtime.completeLegacyRestTurn(ctx, { content: 'take over' });
+
+    expect(result).toEqual(CONVERSATION_UNAVAILABLE_RESULT);
+    expect(svc.prompt).not.toHaveBeenCalled();
+  });
+
+  it('collapses a missing (NotFound) create to conversation_unavailable and never throws', async () => {
+    const svc = makeCollapsingAgentService(new NotFoundException('no such conversation'));
+    const runtime = new EmbeddedChatRuntime(svc as never);
+
+    const result = await runtime.completeLegacyRestTurn(ctx, { content: 'hello' });
+
+    expect(result).toEqual(CONVERSATION_UNAVAILABLE_RESULT);
+    expect(svc.prompt).not.toHaveBeenCalled();
+  });
+
+  it('returns the IDENTICAL collapse for foreign and missing so neither can be distinguished', async () => {
+    const foreign = new EmbeddedChatRuntime(
+      makeCollapsingAgentService(new ForbiddenException('foreign owner')) as never,
+    );
+    const missing = new EmbeddedChatRuntime(
+      makeCollapsingAgentService(new NotFoundException('no such conversation')) as never,
+    );
+
+    const foreignResult = await foreign.completeLegacyRestTurn(ctx, { content: 'x' });
+    const missingResult = await missing.completeLegacyRestTurn(ctx, { content: 'x' });
+
+    expect(foreignResult).toEqual(missingResult);
+    expect(foreignResult).toEqual(CONVERSATION_UNAVAILABLE_RESULT);
+  });
+});
+
+describe('TESS Task-5 embedded socket lease lifecycle (one-shot dispatch, idempotent dispose, partial-setup rollback)', () => {
+  const ctx = ownConversation(CONVERSATION_ID, EMBEDDED_SCOPE);
+
+  it('dispatches the turn exactly once; a second dispatch is a no-op turn_already_dispatched', async () => {
+    const { svc } = makeLeaseAgentService();
+    const runtime = new EmbeddedChatRuntime(svc as never);
+
+    const prepared = await runtime.prepareLegacySocketTurn(ctx, { content: 'first' }, makeStream());
+    expect(prepared.ok).toBe(true);
+    if (!prepared.ok) throw new Error('prepareLegacySocketTurn should succeed');
+    const lease = prepared.value;
+
+    const first = await lease.dispatch();
+    expect(first).toEqual({ ok: true, value: undefined });
+    expect(svc.prompt).toHaveBeenCalledTimes(1);
+
+    const second = await lease.dispatch();
+    expect(second).toEqual({ ok: false, code: 'turn_already_dispatched', retryable: false });
+    // Zero additional effect — the second dispatch must not prompt again.
+    expect(svc.prompt).toHaveBeenCalledTimes(1);
+  });
+
+  it('disposes once; a second dispose is a silent no-op that never re-detaches or destroys the session', async () => {
+    const { svc, unsubscribe, session } = makeLeaseAgentService();
+    const runtime = new EmbeddedChatRuntime(svc as never);
+
+    const prepared = await runtime.prepareLegacySocketTurn(ctx, { content: 'x' }, makeStream());
+    expect(prepared.ok).toBe(true);
+    if (!prepared.ok) throw new Error('prepareLegacySocketTurn should succeed');
+    const lease = prepared.value;
+
+    await lease.dispose();
+    await lease.dispose();
+
+    // Listener + channel torn down exactly once across two dispose calls.
+    expect(unsubscribe).toHaveBeenCalledTimes(1);
+    expect(svc.removeChannel).toHaveBeenCalledTimes(1);
+    // Disposal never terminates the underlying session or process.
+    expect(session.piSession.abort).not.toHaveBeenCalled();
+    expect(session.piSession.dispose).not.toHaveBeenCalled();
+  });
+
+  it('rolls back the acquired listener and returns a total safe failure when channel attach fails mid-setup', async () => {
+    const { svc, unsubscribe } = makeLeaseAgentService();
+    svc.addChannel = vi.fn(() => {
+      throw new Error('channel attach failed');
+    });
+    const runtime = new EmbeddedChatRuntime(svc as never);
+
+    // Must NOT throw out of the port — a partial setup collapses to a total safe failure.
+    const prepared = await runtime.prepareLegacySocketTurn(ctx, { content: 'x' }, makeStream());
+    expect(prepared.ok).toBe(false);
+    // Exactly what was acquired (the event listener) is rolled back.
+    expect(unsubscribe).toHaveBeenCalledTimes(1);
+  });
+});

apps/gateway/src/chat/__tests__/chat-security.test.ts

@@ -485,11 +485,13 @@ describe('Non-Discord ("Telegram-equivalent") socket ingress rejection (Task Fiv
 
       await gateway.handleConnection(client as never);
 
-      // Rejected at the door: disconnected, no trust flag, no user scope, no manifest.
+      // Rejected at the door: disconnected, no trust flag, no user scope, no manifest, and — the
+      // Task 5 send-capability rule — no send-protocol advertisement to an unauthenticated socket.
       expect(client.disconnect).toHaveBeenCalled();
       expect(client.data.discordService).not.toBe(true);
       expect(client.data.user).toBeUndefined();
       expect(client.emit).not.toHaveBeenCalledWith('commands:manifest', expect.anything());
+      expect(client.emit).not.toHaveBeenCalledWith('chat:send-capability', expect.anything());
 
       // Even if the ignored socket then attempts a message, it carries no scope, so the send path
       // never begins and no runtime is dispatched.
@@ -507,6 +509,291 @@ describe('Non-Discord ("Telegram-equivalent") socket ingress rejection (Task Fiv
       expect(harnessConversations.append).not.toHaveBeenCalled();
     },
   );
+
+  // A minimal authenticated-browser connection harness for the send-capability advertisement.
+  const connectAuthedBrowser = async (
+    mode: 'legacy' | 'pi-rpc',
+    clientId: string,
+  ): Promise<{ emit: ReturnType<typeof vi.fn>; disconnect: ReturnType<typeof vi.fn> }> => {
+    process.env['CHAT_HARNESS_RUNTIME'] = mode;
+    const agentService = {
+      getSession: vi.fn().mockReturnValue(undefined),
+      createSession: vi.fn(),
+      recordMessage: vi.fn(),
+      onEvent: vi.fn().mockReturnValue((): void => undefined),
+      addChannel: vi.fn(),
+      prompt: vi.fn().mockResolvedValue(undefined),
+    };
+    const harnessConversations = { append: vi.fn() };
+    const auth = {
+      api: {
+        getSession: vi
+          .fn()
+          .mockResolvedValue({ user: { id: 'user-a' }, session: { id: 'session-a' } }),
+      },
+    };
+    const gateway = new ChatGateway(
+      readyRouter(mode, agentService, harnessConversations) as never,
+      auth as never,
+      { conversations: { addMessage: vi.fn().mockResolvedValue(undefined) } } as never,
+      { getManifest: vi.fn().mockReturnValue({ commands: [] }) } as never,
+      {} as never,
+      {} as never,
+    );
+    const client = {
+      id: clientId,
+      handshake: { auth: {}, headers: { cookie: 'session=abc' } },
+      data: {} as Record<string, unknown>,
+      emit: vi.fn(),
+      disconnect: vi.fn(),
+    };
+    await gateway.handleConnection(client as never);
+    return client as never;
+  };
+
+  it('advertises legacy-message exactly once to an authenticated browser in legacy mode (Task 5 MAJOR-1)', async () => {
+    const client = await connectAuthedBrowser('legacy', 'browser-cap-legacy');
+
+    // Legacy mode: the connected Gateway handles the `message` event, so it advertises
+    // `legacy-message` — targeted, connection-bound, exactly once.
+    expect(client.emit).toHaveBeenCalledWith('chat:send-capability', {
+      protocol: 'legacy-message',
+      connectionId: 'browser-cap-legacy',
+    });
+    const capabilityCalls = client.emit.mock.calls.filter(
+      (call: unknown[]) => call[0] === 'chat:send-capability',
+    );
+    expect(capabilityCalls).toHaveLength(1);
+    expect(client.disconnect).not.toHaveBeenCalled();
+  });
+
+  it('advertises unavailable (never turn-send) to an authenticated browser in pi-rpc mode (Task 5 MAJOR-1)', async () => {
+    const client = await connectAuthedBrowser('pi-rpc', 'browser-cap-pirpc');
+
+    // pi-rpc mode: the legacy `message` handler fails closed and the authenticated `turn:send`
+    // handler lands in Task 15, so Task 5 advertises `unavailable` — never `turn-send`.
+    expect(client.emit).toHaveBeenCalledWith('chat:send-capability', {
+      protocol: 'unavailable',
+      connectionId: 'browser-cap-pirpc',
+    });
+    const capabilityCalls = client.emit.mock.calls.filter(
+      (call: unknown[]) => call[0] === 'chat:send-capability',
+    );
+    expect(capabilityCalls).toHaveLength(1);
+    expect(capabilityCalls[0]?.[1]).not.toMatchObject({ protocol: 'turn-send' });
+  });
+
+  it.each(['legacy', 'pi-rpc'] as const)(
+    'never advertises send-capability to a Discord service socket (%s mode, Task 5 MAJOR-1)',
+    async (mode) => {
+      process.env['CHAT_HARNESS_RUNTIME'] = mode;
+      process.env['DISCORD_SERVICE_TOKEN'] = 'super-secret-discord-token';
+      const agentService = {
+        getSession: vi.fn().mockReturnValue(undefined),
+        createSession: vi.fn(),
+        recordMessage: vi.fn(),
+        onEvent: vi.fn().mockReturnValue((): void => undefined),
+        addChannel: vi.fn(),
+        prompt: vi.fn().mockResolvedValue(undefined),
+      };
+      const gateway = new ChatGateway(
+        readyRouter(mode, agentService, { append: vi.fn() }) as never,
+        { api: { getSession: vi.fn() } } as never,
+        { conversations: { addMessage: vi.fn().mockResolvedValue(undefined) } } as never,
+        {} as never,
+        {} as never,
+        {} as never,
+      );
+      const client = {
+        id: `discord-service-${mode}`,
+        handshake: { auth: { discordServiceToken: 'super-secret-discord-token' }, headers: {} },
+        data: {} as Record<string, unknown>,
+        emit: vi.fn(),
+        disconnect: vi.fn(),
+      };
+
+      await gateway.handleConnection(client as never);
+
+      // The trusted Discord service socket is not a browser; it never receives a browser
+      // send-protocol advertisement.
+      expect(client.data.discordService).toBe(true);
+      expect(client.emit).not.toHaveBeenCalledWith('chat:send-capability', expect.anything());
+      delete process.env['DISCORD_SERVICE_TOKEN'];
+    },
+  );
+
+  // Record-and-forward instrumentation at the REAL returned-lease boundary: wrap the lease's own
+  // dispatch/dispose so the test observes the Gateway's invocation counts through the real
+  // ChatRuntimeRouter -> EmbeddedChatRuntime path — no canned lease, synthesized method, or shim.
+  const instrumentLease = (
+    router: ChatRuntimeRouter,
+    order: string[],
+    counters: { dispatch: number; dispose: number },
+  ): void => {
+    const realPrepare = router.prepareLegacySocketTurn.bind(router) as (
+      ...args: unknown[]
+    ) => Promise<{ ok: boolean; value?: { dispatch: () => unknown; dispose: () => unknown } }>;
+    vi.spyOn(router, 'prepareLegacySocketTurn').mockImplementation((async (...args: unknown[]) => {
+      const result = await realPrepare(...args);
+      if (result.ok && result.value) {
+        const lease = result.value;
+        const realDispatch = lease.dispatch.bind(lease);
+        const realDispose = lease.dispose.bind(lease);
+        lease.dispatch = (): unknown => {
+          counters.dispatch += 1;
+          order.push('dispatch');
+          return realDispatch();
+        };
+        lease.dispose = (): unknown => {
+          counters.dispose += 1;
+          return realDispose();
+        };
+      }
+      return result;
+    }) as never);
+  };
+
+  it('orders a legacy browser turn persist -> ack -> lease.dispatch -> prompt, each exactly once (Task 5 G2)', async () => {
+    process.env['CHAT_HARNESS_RUNTIME'] = 'legacy';
+    const order: string[] = [];
+    const counters = { dispatch: 0, dispose: 0 };
+    const session = {
+      provider: 'configured-provider',
+      modelId: 'configured-model',
+      piSession: {
+        thinkingLevel: 'medium',
+        getAvailableThinkingLevels: (): string[] => ['medium'],
+      },
+    };
+    const agentService = {
+      getSession: vi.fn().mockReturnValue(undefined),
+      createSession: vi.fn().mockResolvedValue(session),
+      recordMessage: vi.fn(),
+      onEvent: vi.fn().mockReturnValue((): void => undefined),
+      addChannel: vi.fn(),
+      removeChannel: vi.fn(),
+      prompt: vi.fn().mockImplementation(async (): Promise<void> => {
+        order.push('prompt');
+      }),
+    };
+    const harnessConversations = { append: vi.fn() };
+    const brain = {
+      conversations: {
+        findById: vi.fn().mockResolvedValue({ id: 'conversation-order-1' }),
+        findMessages: vi.fn().mockResolvedValue([]),
+        create: vi.fn().mockResolvedValue(undefined),
+        update: vi.fn().mockResolvedValue(undefined),
+        addMessage: vi.fn().mockImplementation(async (): Promise<void> => {
+          order.push('persist');
+        }),
+      },
+    };
+    const router = readyRouter('legacy', agentService, harnessConversations);
+    instrumentLease(router, order, counters);
+    const gateway = new ChatGateway(
+      router as never,
+      { api: { getSession: vi.fn() } } as never,
+      brain as never,
+      {} as never,
+      {} as never,
+      {} as never,
+    );
+    const client = {
+      id: 'browser-order-1',
+      data: { user: { id: 'user-a' } },
+      emit: vi.fn().mockImplementation((event: string): void => {
+        if (event === 'message:ack') order.push('ack');
+      }),
+    };
+
+    await gateway.handleMessage(
+      client as never,
+      {
+        conversationId: 'conversation-order-1',
+        content: 'ordered hello',
+      } as never,
+    );
+
+    // The Gateway persists the user turn, THEN acks, THEN invokes the one-shot lease dispatch which
+    // finally prompts. Observed at the real lease boundary: dispatch is invoked exactly once and the
+    // runtime prompts exactly once.
+    expect(order).toEqual(['persist', 'ack', 'dispatch', 'prompt']);
+    expect(counters.dispatch).toBe(1);
+    expect(agentService.prompt).toHaveBeenCalledTimes(1);
+    expect(harnessConversations.append).not.toHaveBeenCalled();
+  });
+
+  it('drops a legacy browser turn on persistence failure: zero info/ack/dispatch/prompt, one clean disposal (Task 5 G2)', async () => {
+    process.env['CHAT_HARNESS_RUNTIME'] = 'legacy';
+    const order: string[] = [];
+    const counters = { dispatch: 0, dispose: 0 };
+    const unsub = vi.fn();
+    const session = {
+      provider: 'configured-provider',
+      modelId: 'configured-model',
+      piSession: {
+        thinkingLevel: 'medium',
+        getAvailableThinkingLevels: (): string[] => ['medium'],
+      },
+    };
+    const agentService = {
+      getSession: vi.fn().mockReturnValue(undefined),
+      createSession: vi.fn().mockResolvedValue(session),
+      recordMessage: vi.fn(),
+      onEvent: vi.fn().mockReturnValue(unsub),
+      addChannel: vi.fn(),
+      removeChannel: vi.fn(),
+      prompt: vi.fn().mockResolvedValue(undefined),
+    };
+    const harnessConversations = { append: vi.fn() };
+    const brain = {
+      conversations: {
+        findById: vi.fn().mockResolvedValue({ id: 'conversation-fail-1' }),
+        findMessages: vi.fn().mockResolvedValue([]),
+        create: vi.fn().mockResolvedValue(undefined),
+        update: vi.fn().mockResolvedValue(undefined),
+        addMessage: vi.fn().mockRejectedValue(new Error('persistence unavailable')),
+      },
+    };
+    const router = readyRouter('legacy', agentService, harnessConversations);
+    instrumentLease(router, order, counters);
+    const gateway = new ChatGateway(
+      router as never,
+      { api: { getSession: vi.fn() } } as never,
+      brain as never,
+      {} as never,
+      {} as never,
+      {} as never,
+    );
+    const client = {
+      id: 'browser-fail-1',
+      data: { user: { id: 'user-a' } },
+      emit: vi.fn(),
+    };
+
+    await gateway.handleMessage(
+      client as never,
+      {
+        conversationId: 'conversation-fail-1',
+        content: 'will fail persistence',
+      } as never,
+    );
+
+    // A failed user-message persistence aborts the turn with no accept-then-lose: no session:info,
+    // no ack, the lease never dispatches or prompts, and the just-prepared lease is disposed exactly
+    // once (listener/channel cleanup) without throwing. The fixed safe error is surfaced.
+    expect(client.emit).not.toHaveBeenCalledWith('session:info', expect.anything());
+    expect(client.emit).not.toHaveBeenCalledWith('message:ack', expect.anything());
+    expect(counters.dispatch).toBe(0);
+    expect(agentService.prompt).not.toHaveBeenCalled();
+    expect(counters.dispose).toBe(1);
+    expect(unsub).toHaveBeenCalledTimes(1);
+    expect(client.emit).toHaveBeenCalledWith(
+      'error',
+      expect.objectContaining({ code: 'persist_failed' }),
+    );
+    expect(harnessConversations.append).not.toHaveBeenCalled();
+  });
 });
 
 describe('Chat DTO validation', () => {

apps/gateway/src/chat/chat.gateway-command-approval.spec.ts

@@ -91,3 +91,114 @@ describe('ChatGateway command approval ingress', () => {
     });
   });
 });
+
+/**
+ * Task 5 (G3) command runtime fence. Under pi-rpc there is no embedded chat session, so
+ * embedded slash-commands (/model, /agent, and every other non-audited command) are fixed
+ * "unsupported" and MUST fail closed BEFORE reaching the command executor — never a silent
+ * fall-through to embedded execution. Only runtime-independent audited system commands
+ * (/reload) pass through as a positive control, and the approval path stays runtime-independent.
+ * The router stub here carries `runtimeMode: 'pi-rpc'` and throws if any runtime is resolved, so
+ * a fence bypass surfaces as a thrown error rather than a silent embedded dispatch.
+ */
+function buildPiRpcGateway(commandExecutor: {
+  execute: ReturnType<typeof vi.fn>;
+  createApproval: ReturnType<typeof vi.fn>;
+}): ChatGateway {
+  const piRpcRouter = {
+    runtimeMode: 'pi-rpc' as const,
+    onModuleInit: () => {
+      throw new Error('chat runtime router must not initialise on the pi-rpc command path');
+    },
+    get active(): never {
+      throw new Error('chat runtime must not be resolved on the pi-rpc command path');
+    },
+  };
+  return new ChatGateway(
+    piRpcRouter as never,
+    {} as never,
+    {} as never,
+    {} as never,
+    commandExecutor as never,
+    {} as never,
+  );
+}
+
+describe('ChatGateway command runtime fence (Task 5 G3, pi-rpc)', () => {
+  const UNSUPPORTED = 'Slash commands are not available on this deployment.';
+
+  it.each(['model', 'agent', 'gc'])(
+    'fails /%s closed before the executor under pi-rpc (execute never called)',
+    async (command): Promise<void> => {
+      const commandExecutor = {
+        execute: vi
+          .fn()
+          .mockResolvedValue({ command, conversationId: 'conversation-1', success: true }),
+        createApproval: vi.fn(),
+      };
+      const gateway = buildPiRpcGateway(commandExecutor);
+      const client = { data: { user: { id: 'admin-1' } }, emit: vi.fn() };
+
+      await gateway.handleCommandExecute(client as never, {
+        command,
+        conversationId: 'conversation-1',
+      });
+
+      expect(commandExecutor.execute).toHaveBeenCalledTimes(0);
+      expect(client.emit).toHaveBeenCalledWith('command:result', {
+        command,
+        conversationId: 'conversation-1',
+        success: false,
+        message: UNSUPPORTED,
+      });
+    },
+  );
+
+  it('passes the audited /reload system command through as a positive control under pi-rpc', async (): Promise<void> => {
+    const reloadResult = { command: 'reload', conversationId: 'conversation-1', success: true };
+    const commandExecutor = {
+      execute: vi.fn().mockResolvedValue(reloadResult),
+      createApproval: vi.fn(),
+    };
+    const gateway = buildPiRpcGateway(commandExecutor);
+    const client = { data: { user: { id: 'admin-1' } }, emit: vi.fn() };
+
+    await gateway.handleCommandExecute(client as never, {
+      command: 'reload',
+      conversationId: 'conversation-1',
+    });
+
+    expect(commandExecutor.execute).toHaveBeenCalledTimes(1);
+    expect(commandExecutor.execute).toHaveBeenCalledWith(
+      { command: 'reload', conversationId: 'conversation-1' },
+      { userId: 'admin-1', tenantId: 'admin-1' },
+    );
+    expect(client.emit).toHaveBeenCalledWith('command:result', reloadResult);
+  });
+
+  it('keeps command approval runtime-independent under pi-rpc (createApproval still runs)', async (): Promise<void> => {
+    const commandExecutor = {
+      execute: vi.fn(),
+      createApproval: vi.fn().mockResolvedValue({
+        approvalId: 'approval-1',
+        expiresAt: '2026-07-12T00:05:00.000Z',
+      }),
+    };
+    const gateway = buildPiRpcGateway(commandExecutor);
+    const client = { data: { user: { id: 'admin-1' } }, emit: vi.fn() };
+
+    await gateway.handleCommandApproval(client as never, {
+      command: 'gc',
+      conversationId: 'conversation-1',
+    });
+
+    expect(commandExecutor.createApproval).toHaveBeenCalledWith(
+      { command: 'gc', conversationId: 'conversation-1' },
+      { userId: 'admin-1', tenantId: 'admin-1' },
+    );
+    expect(client.emit).toHaveBeenCalledWith(
+      'command:approval',
+      expect.objectContaining({ success: true, approvalId: 'approval-1' }),
+    );
+  });
+});

apps/gateway/src/chat/chat.gateway.ts

@@ -87,6 +87,14 @@ interface ClientSession {
  * Keyed by conversationId, value is the model name to use.
  */
 const modelOverrides = new Map<string, string>();
+/**
+ * Task 5 (G3): commands whose effect is runtime-independent — they operate on gateway/system
+ * state rather than an embedded chat session — and therefore stay available under pi-rpc. Every
+ * other command is an embedded-session command and is fixed-"unsupported" under pi-rpc, failing
+ * closed before the executor. Kept as an explicit allowlist so adding a runtime-independent
+ * command is a deliberate edit, never an accidental fall-through.
+ */
+const RUNTIME_INDEPENDENT_COMMANDS: ReadonlySet<string> = new Set(['reload']);
 const MAX_REDACTION_BUFFER_LENGTH = 8_192;
 const MAX_CHANNEL_ATTACHMENTS = 10;
 const MAX_ATTACHMENT_METADATA_BYTES = 16_384;
@@ -273,6 +281,17 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
     client.data.session = session.session;
     this.logger.log(`Client connected: ${client.id}`);
     client.emit('commands:manifest', { manifest: this.commandRegistry.getManifest() });
+
+    // Send-protocol advertisement (Task 5): a conversation id or harness selection never proves the
+    // connected Gateway handles a given wire event, so after BetterAuth authentication and
+    // user/session assignment advertise — exactly once, bound to this connection — which send event
+    // the browser may use. Legacy mode handles `message` (`legacy-message`); `pi-rpc` fails that
+    // handler closed and its authenticated `turn:send` handler lands in Task 15, so it advertises
+    // `unavailable` and never `turn-send`. Capability is routing information, never authorization.
+    client.emit('chat:send-capability', {
+      protocol: this.runtime.runtimeMode === 'pi-rpc' ? 'unavailable' : 'legacy-message',
+      connectionId: client.id,
+    });
   }
 
   handleDisconnect(client: Socket): void {
@@ -386,7 +405,24 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
     }
 
     this.registerClientSession(client, conversationId, stream.channelId, prepared.value, scope);
-    await this.persistUserMessage(conversationId, scope.userId, data.content, data.attachments);
+    // Persist the user turn BEFORE acknowledging or dispatching. If persistence fails, abort the
+    // turn: tear down the just-prepared lease and surface the failure — never ack-then-lose.
+    const persisted = await this.persistUserMessage(
+      conversationId,
+      scope.userId,
+      data.content,
+      data.attachments,
+    );
+    if (!persisted) {
+      await this.disposeExistingSession(key);
+      client.emit('error', {
+        conversationId,
+        code: 'persist_failed',
+        retryable: true,
+        error: 'Your message could not be saved. Please try again.',
+      });
+      return;
+    }
 
     client.emit('session:info', { conversationId, ...prepared.value.presentation });
     client.emit('message:ack', { conversationId, messageId: uuid() });
@@ -473,11 +509,32 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
     }
 
     this.registerClientSession(client, conversationId, stream.channelId, prepared.value, scope);
-    await this.persistUserMessage(conversationId, scope.userId, ingress.content, attachments, {
-      correlationId: ingress.correlationId,
-      discordMessageId: ingress.messageId,
-      discordUserId: ingress.userId,
-    });
+    // Persist BEFORE acknowledging or dispatching; on persistence failure abort the verified turn
+    // (tear down the lease, surface the error) rather than ack-then-lose the Discord message.
+    const persisted = await this.persistUserMessage(
+      conversationId,
+      scope.userId,
+      ingress.content,
+      attachments,
+      {
+        correlationId: ingress.correlationId,
+        discordMessageId: ingress.messageId,
+        discordUserId: ingress.userId,
+      },
+    );
+    if (!persisted) {
+      await this.disposeExistingSession(key);
+      client.emit('error', {
+        conversationId,
+        code: 'persist_failed',
+        retryable: true,
+        correlationId: ingress.correlationId,
+        discordMessageId: ingress.messageId,
+        discordUserId: ingress.userId,
+        error: 'Your message could not be saved. Please try again.',
+      });
+      return;
+    }
 
     client.emit('session:info', { conversationId, ...prepared.value.presentation });
     client.emit('message:ack', {
@@ -529,8 +586,8 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
     content: string,
     attachments: readonly ChannelAttachmentDto[] | undefined,
     discord?: { correlationId: string; discordMessageId: string; discordUserId: string },
-  ): Promise<void> {
-    if (!userId) return;
+  ): Promise<boolean> {
+    if (!userId) return true;
     await this.ensureConversation(conversationId, userId);
     try {
       await this.brain.conversations.addMessage(
@@ -563,11 +620,13 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
         },
         userId,
       );
+      return true;
     } catch (err) {
       this.logger.error(
         `Failed to persist user message for conversation=${conversationId}`,
         err instanceof Error ? err.stack : String(err),
       );
+      return false;
     }
   }
 
@@ -660,6 +719,22 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
       return;
     }
 
+    // Task 5 (G3): under pi-rpc there is no embedded chat session, so embedded slash-commands are
+    // unsupported. Fail closed BEFORE the executor — never fall back to embedded execution — while
+    // runtime-independent audited system commands (e.g. /reload) still pass through.
+    if (
+      this.runtime.runtimeMode === 'pi-rpc' &&
+      !RUNTIME_INDEPENDENT_COMMANDS.has(payload.command)
+    ) {
+      client.emit('command:result', {
+        command: payload.command,
+        conversationId: payload.conversationId,
+        success: false,
+        message: 'Slash commands are not available on this deployment.',
+      });
+      return;
+    }
+
     const result = await this.commandExecutor.execute(payload, scope);
     client.emit('command:result', result);
   }

apps/gateway/src/chat/embedded-chat.runtime.ts

@@ -116,7 +116,18 @@ export class EmbeddedChatRuntime implements ChatRuntime, LegacyEmbeddedChatPort
     });
     if (!resolved.ok) return resolved;
 
-    const detach = this.subscribe(conversationId, scope, stream);
+    let detach: () => void;
+    try {
+      detach = this.subscribe(conversationId, scope, stream);
+    } catch (err) {
+      // A partial listener/channel setup rolled itself back inside subscribe(); surface a total
+      // safe failure instead of throwing out of the port. Retryable — the attach is transient.
+      this.logger.error(
+        `Embedded socket subscription failed for conversation=${conversationId}`,
+        err instanceof Error ? err.message : String(err),
+      );
+      return { ok: false, code: 'runtime_unavailable', retryable: true };
+    }
 
     return {
       ok: true,
@@ -224,7 +235,18 @@ export class EmbeddedChatRuntime implements ChatRuntime, LegacyEmbeddedChatPort
     });
     if (!resolved.ok) return resolved;
 
-    const detach = this.subscribe(conversationId, scope, stream);
+    let detach: () => void;
+    try {
+      detach = this.subscribe(conversationId, scope, stream);
+    } catch (err) {
+      // A partial listener/channel setup rolled itself back inside subscribe(); surface a total
+      // safe failure instead of throwing out of the port. Retryable — the attach is transient.
+      this.logger.error(
+        `Embedded Discord subscription failed for conversation=${conversationId}`,
+        err instanceof Error ? err.message : String(err),
+      );
+      return { ok: false, code: 'runtime_unavailable', retryable: true };
+    }
 
     return {
       ok: true,
@@ -293,7 +315,19 @@ export class EmbeddedChatRuntime implements ChatRuntime, LegacyEmbeddedChatPort
       },
       scope,
     );
-    this.agentService.addChannel(conversationId, stream.channelId, scope);
+    try {
+      this.agentService.addChannel(conversationId, stream.channelId, scope);
+    } catch (err) {
+      // Partial setup: the listener was acquired but the channel attach failed. Roll back
+      // exactly what was acquired (the listener) before the failure escapes, so no leaked
+      // subscription survives; the caller converts the rethrow into a total safe failure.
+      try {
+        unsubscribe();
+      } catch {
+        /* idempotent teardown */
+      }
+      throw err;
+    }
     return () => {
       try {
         unsubscribe();

apps/gateway/src/plugin/discord-ingress.security.spec.ts

@@ -594,6 +594,91 @@ describe('Discord ingress security', () => {
     expect(harnessConversations.append).not.toHaveBeenCalled();
   });
 
+  it('dispatches a verified Discord SEND once and drops a byte-identical replay with zero additional dispatch/persist/ack (Task 5 G4)', async () => {
+    configureDiscordEnv();
+    process.env['CHAT_HARNESS_RUNTIME'] = 'pi-rpc';
+    process.env['DISCORD_ALLOWED_CHANNEL_IDS'] = 'channel-001';
+    process.env['DISCORD_INTERACTION_BINDINGS'] = JSON.stringify([
+      {
+        instanceId: 'Nova',
+        agentConfigId: 'agent-config-nova',
+        guildId: 'guild-001',
+        channelId: 'channel-001',
+        pairedUsers: {
+          'user-001': { role: 'operator', mosaicUserId: 'mosaic-operator-001' },
+        },
+      },
+    ]);
+    const session = {
+      provider: 'configured-provider',
+      modelId: 'configured-model',
+      piSession: {
+        thinkingLevel: 'medium',
+        getAvailableThinkingLevels: (): string[] => ['medium'],
+      },
+    };
+    const createSession = vi.fn().mockResolvedValue(session);
+    const prompt = vi.fn().mockResolvedValue(undefined);
+    const agentService = {
+      getSession: vi.fn().mockReturnValue(undefined),
+      createSession,
+      recordMessage: vi.fn(),
+      onEvent: vi.fn().mockReturnValue((): void => undefined),
+      addChannel: vi.fn(),
+      removeChannel: vi.fn(),
+      prompt,
+    };
+    const addMessage = vi.fn().mockResolvedValue(undefined);
+    const brain = {
+      agents: { findById: vi.fn((id: string) => Promise.resolve({ id, name: 'Nova' })) },
+      conversations: {
+        findById: vi.fn().mockResolvedValue({ id: 'Nova:discord:channel-001' }),
+        findMessages: vi.fn().mockResolvedValue([]),
+        create: vi.fn().mockResolvedValue(undefined),
+        update: vi.fn().mockResolvedValue(undefined),
+        addMessage,
+      },
+    };
+    const harnessConversations = { append: vi.fn() };
+    const gateway = new ChatGateway(
+      piRpcRouterFronting(agentService, harnessConversations) as never,
+      {} as never,
+      brain as never,
+      {} as never,
+      {} as never,
+      { resolve: vi.fn() } as never,
+    );
+    const client = {
+      id: 'discord-client-replay',
+      data: { discordService: true },
+      emit: vi.fn(),
+    };
+    const ackCount = (): number =>
+      client.emit.mock.calls.filter((call) => call[0] === 'message:ack').length;
+
+    // One fully-valid signed envelope; the replay reuses the SAME object (same messageId).
+    const envelope = ingressEnvelope('verified once', 'discord-replay-001', {
+      conversationId: 'Nova:discord:channel-001',
+    });
+
+    // First delivery: the verified-Discord SEND runs the full embedded dispatch exactly once.
+    await gateway.handleMessage(client as never, envelope);
+    expect(createSession).toHaveBeenCalledTimes(1);
+    expect(prompt).toHaveBeenCalledTimes(1);
+    expect(addMessage).toHaveBeenCalledTimes(1);
+    expect(ackCount()).toBe(1);
+
+    // Byte-identical replay: the messageId is already claimed, so resolveDiscordIngress returns
+    // null and the SEND handler bails before dispatch/persist/ack. Every effect stays at exactly one.
+    await gateway.handleMessage(client as never, envelope);
+    expect(createSession).toHaveBeenCalledTimes(1);
+    expect(prompt).toHaveBeenCalledTimes(1);
+    expect(addMessage).toHaveBeenCalledTimes(1);
+    expect(ackCount()).toBe(1);
+    // The harness runtime is never touched on either delivery.
+    expect(harnessConversations.append).not.toHaveBeenCalled();
+  });
+
   it('retains validated persisted attachments in resumed conversation history', async () => {
     const attachment = {
       id: 'attachment-history',