From f4e7a4ddb7e83ad9a0b25f9f485ab9d217ccb098 Mon Sep 17 00:00:00 2001 From: shaggy Date: Wed, 12 Aug 2026 14:43:59 -0500 Subject: [PATCH] refactor(chat): route browser chat through one runtime Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01ESFAnh2t9HmLwng8oW95St --- .../agent/__tests__/session-ownership.test.ts | 153 ++++++- .../src/chat/__tests__/chat-security.test.ts | 289 ++++++++++++- .../chat.gateway-command-approval.spec.ts | 111 +++++ apps/gateway/src/chat/chat.gateway.ts | 91 +++- .../gateway/src/chat/embedded-chat.runtime.ts | 40 +- .../plugin/discord-ingress.security.spec.ts | 85 ++++ apps/web/src/lib/chat-contract.ts | 4 + apps/web/src/spa/chat/composer.tsx | 21 +- .../spa/chat/test-support/fake-chat-socket.ts | 14 +- .../src/spa/chat/use-chat-connection.spec.tsx | 399 +++++++++++++++++- apps/web/src/spa/chat/use-chat-connection.ts | 210 +++++++-- apps/web/src/spa/pages/chat.spec.tsx | 40 ++ packages/types/src/chat/events.ts | 20 + packages/types/src/chat/index.ts | 2 + 14 files changed, 1399 insertions(+), 80 deletions(-) diff --git a/apps/gateway/src/agent/__tests__/session-ownership.test.ts b/apps/gateway/src/agent/__tests__/session-ownership.test.ts index ec297c34..2ebb1235 100644 --- a/apps/gateway/src/agent/__tests__/session-ownership.test.ts +++ b/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); + }); +}); diff --git a/apps/gateway/src/chat/__tests__/chat-security.test.ts b/apps/gateway/src/chat/__tests__/chat-security.test.ts index e455e767..7343576f 100644 --- a/apps/gateway/src/chat/__tests__/chat-security.test.ts +++ b/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; disconnect: ReturnType }> => { + 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, + 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, + 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 => { + 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 => { + 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', () => { diff --git a/apps/gateway/src/chat/chat.gateway-command-approval.spec.ts b/apps/gateway/src/chat/chat.gateway-command-approval.spec.ts index 2d1364ef..08f18f95 100644 --- a/apps/gateway/src/chat/chat.gateway-command-approval.spec.ts +++ b/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; + createApproval: ReturnType; +}): 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 => { + 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 => { + 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 => { + 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' }), + ); + }); +}); diff --git a/apps/gateway/src/chat/chat.gateway.ts b/apps/gateway/src/chat/chat.gateway.ts index 60665ca0..4f63c5a9 100644 --- a/apps/gateway/src/chat/chat.gateway.ts +++ b/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(); +/** + * 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 = 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 { - if (!userId) return; + ): Promise { + 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); } diff --git a/apps/gateway/src/chat/embedded-chat.runtime.ts b/apps/gateway/src/chat/embedded-chat.runtime.ts index bbcfe9b0..d8f69063 100644 --- a/apps/gateway/src/chat/embedded-chat.runtime.ts +++ b/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(); diff --git a/apps/gateway/src/plugin/discord-ingress.security.spec.ts b/apps/gateway/src/plugin/discord-ingress.security.spec.ts index b11eead6..6698730b 100644 --- a/apps/gateway/src/plugin/discord-ingress.security.spec.ts +++ b/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', diff --git a/apps/web/src/lib/chat-contract.ts b/apps/web/src/lib/chat-contract.ts index 57e700a1..e6763123 100644 --- a/apps/web/src/lib/chat-contract.ts +++ b/apps/web/src/lib/chat-contract.ts @@ -10,6 +10,8 @@ import type { AgentTextPayload, AgentThinkingPayload, ChatMessagePayload, + ChatSendCapabilityPayload, + ChatSendProtocol, ClientToServerEvents, CommandDef, CommandManifest, @@ -40,6 +42,8 @@ export type { AgentTextPayload, AgentThinkingPayload, ChatMessagePayload, + ChatSendCapabilityPayload, + ChatSendProtocol, ClientToServerEvents, CommandDef, CommandManifest, diff --git a/apps/web/src/spa/chat/composer.tsx b/apps/web/src/spa/chat/composer.tsx index f43bfbf8..07ae631d 100644 --- a/apps/web/src/spa/chat/composer.tsx +++ b/apps/web/src/spa/chat/composer.tsx @@ -3,12 +3,7 @@ import type { HarnessSelection } from '@/lib/types'; import type { HarnessSelectionValue } from './use-harness-selection'; interface ComposerProps { - onSend: (input: { - content: string; - provider?: string; - modelId?: string; - selection: HarnessSelection; - }) => boolean; + onSend: (input: { content: string; selection: HarnessSelection }) => boolean; onStop: () => void; streaming: boolean; /** True from local send time through server turn startup/ack and @@ -53,17 +48,11 @@ export function Composer({ if (!harness.canSend || harness.persistedSelection === null) return; const trimmed = content.trim(); if (!trimmed) return; - // Pass BOTH the nested selection tuple (the turn-runtime contract) and the - // flat provider/modelId (the legacy first-send that creates the conversation) - // — both derived from the same validated persisted tuple. The hook decides - // which path applies from whether a conversation is already established. + // Pass the validated, persisted selection tuple only. The hook derives the + // wire projection (legacy `message` provider/model, or `turn:send`) from the + // negotiated `chat:send-capability` protocol — never from flat caller input. const selection = harness.persistedSelection; - const ok = onSend({ - content: trimmed, - selection, - provider: selection.providerId, - modelId: selection.modelId, - }); + const ok = onSend({ content: trimmed, selection }); // Clear the input only when the send was accepted — a refused turn (e.g. a // failed idempotency mint) must retain the user's text so it is not lost. if (ok) setContent(''); diff --git a/apps/web/src/spa/chat/test-support/fake-chat-socket.ts b/apps/web/src/spa/chat/test-support/fake-chat-socket.ts index ce80117b..b2f45d08 100644 --- a/apps/web/src/spa/chat/test-support/fake-chat-socket.ts +++ b/apps/web/src/spa/chat/test-support/fake-chat-socket.ts @@ -14,6 +14,10 @@ export interface EmittedEvent { /** The subset of a Socket.IO `ChatSocket` that `useChatConnection` drives. */ export interface FakeChatSocket { connected: boolean; + /** Mirrors socket.io-client's `Socket.id`: the connection identity the server + * echoes in a `chat:send-capability` payload. The generation-bound send + * protocol accepts an advertisement only when `payload.connectionId === id`. */ + id: string; connect(): FakeChatSocket; on(event: K, handler: ServerHandler): FakeChatSocket; off(event: K, handler: ServerHandler): FakeChatSocket; @@ -51,8 +55,10 @@ export function createFakeChatSocket(): { /** Simulates socket.io-client's automatic reconnect of the *same* * instance after a transient disconnect: marks the socket connected again * and fires any handler(s) registered via `socket.on('connect', ...)`, - * without clearing or replacing any listeners. */ - simulateReconnect(): void; + * without clearing or replacing any listeners. A real reconnect is assigned + * a fresh `Socket.id`; pass `nextId` to model that new connection identity + * (defaults to the current id so existing callers are unaffected). */ + simulateReconnect(nextId?: string): void; } { const listeners = new Map void>>(); const emitted: EmittedEvent[] = []; @@ -63,6 +69,7 @@ export function createFakeChatSocket(): { // type-checked against ServerToClientEvents/ClientToServerEvents. const socket = { connected: false, + id: 'socket-a', connect: vi.fn(function connect(this: void) { socket.connected = true; return socket; @@ -105,8 +112,9 @@ export function createFakeChatSocket(): { } } - function simulateReconnect(): void { + function simulateReconnect(nextId: string = socket.id): void { socket.connected = true; + socket.id = nextId; const lifecycleKey = 'connect' satisfies LifecycleEvent as unknown as ServerEvent; for (const handler of listeners.get(lifecycleKey) ?? []) { (handler as () => void)(); diff --git a/apps/web/src/spa/chat/use-chat-connection.spec.tsx b/apps/web/src/spa/chat/use-chat-connection.spec.tsx index 3892a307..66aaf826 100644 --- a/apps/web/src/spa/chat/use-chat-connection.spec.tsx +++ b/apps/web/src/spa/chat/use-chat-connection.spec.tsx @@ -21,7 +21,7 @@ vi.mock('@/lib/socket', () => ({ destroySocket: destroySocketMock, })); -import type { HarnessSelection } from '@mosaicstack/types'; +import type { ChatSendProtocol, HarnessSelection } from '@mosaicstack/types'; import { useChatConnection, type ChatConnectionValue } from './use-chat-connection'; let fake: ReturnType; @@ -52,6 +52,21 @@ function harnessSend(): HarnessSendMessage { return latest?.actions.sendMessage as unknown as HarnessSendMessage; } +/** + * Task Five MAJOR-1 (browser send-protocol negotiation) support. The Gateway + * advertises how this connection may send via a server-to-client-only + * `chat:send-capability` (already part of the typed `ServerToClientEvents` + * contract, so this uses the fake's typed `serverEmit` — no cast); the hook + * holds the advertised protocol and routes `sendMessage` through an exhaustive + * switch on it, never inferring it from conversation/selection. When no listener + * is registered yet (CURRENT impl), the emit is an inert no-op, so the reds + * below fail on BEHAVIOUR — the current send path still infers a protocol and + * emits regardless of any advertisement — not on a missing module or type. + */ +function advertiseCapability(protocol: ChatSendProtocol, connectionId: string): void { + fake.serverEmit('chat:send-capability', { protocol, connectionId }); +} + /** * Install a controllable `crypto.randomUUID` on the global crypto object and * return a restore fn. Uses defineProperty on the instance so it works whether @@ -173,6 +188,20 @@ afterEach(async () => { }); describe('useChatConnection', () => { + // Task Five MAJOR-1: the send path is PROTOCOL-driven — `sendMessage` routes + // only on the negotiated `chat:send-capability`, never on inferred + // conversation/selection state. These pre-existing cases exercise the legacy + // `message` branch, so the connection is advertised `legacy-message` once here + // (server-to-client, for this exact socket id) after the mount registers its + // listener. Sub-describes that need the pi turn-runtime reset the generation + // and re-advertise `turn-send`; the capability describe resets to the + // unadvertised `unavailable` baseline and drives the protocol itself. + beforeEach(async () => { + await act(async () => { + advertiseCapability('legacy-message', fake.socket.id); + }); + }); + it('establishes the active conversation from the first message:ack when message omitted conversationId', async () => { await act(async () => { fake.serverEmit('message:ack', { conversationId: 'c1', messageId: 'm1' }); @@ -450,7 +479,10 @@ describe('useChatConnection', () => { it('sendMessage emits optional conversationId/provider/modelId and appends an optimistic user turn', async () => { await act(async () => { - latest?.actions.sendMessage({ content: 'hello', provider: 'anthropic', modelId: 'claude' }); + latest?.actions.sendMessage({ + content: 'hello', + selection: { harnessId: 'pi', providerId: 'anthropic', modelId: 'claude' }, + }); }); expect(fake.emitted).toContainEqual({ @@ -487,6 +519,19 @@ describe('useChatConnection', () => { }; const UUID = 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa'; + // The pi turn-runtime routes sends through `turn:send`. Reset the generation + // (clearing the outer `legacy-message` advertisement + first-wins lock) and + // advertise `turn-send` for this exact connection, so every send below takes + // the turn-runtime branch. + beforeEach(async () => { + await act(async () => { + fake.simulateReconnect(); + }); + await act(async () => { + advertiseCapability('turn-send', fake.socket.id); + }); + }); + async function establishConversation(): Promise { await act(async () => { fake.serverEmit('message:ack', { conversationId: 'c1', messageId: 'm1' }); @@ -619,6 +664,18 @@ describe('useChatConnection', () => { }; const UUID = 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa'; + // turn:ack is the receipt for a `turn:send`, so these establish under the pi + // turn-runtime: reset the generation (clearing the outer `legacy-message` + // advertisement + lock) and advertise `turn-send` for this connection. + beforeEach(async () => { + await act(async () => { + fake.simulateReconnect(); + }); + await act(async () => { + advertiseCapability('turn-send', fake.socket.id); + }); + }); + // Establish the conversation and send one accepted turn under a controlled // idempotency key. Returns the crypto restore fn so callers unwind it. async function establishAndSend(): Promise<() => void> { @@ -741,6 +798,18 @@ describe('useChatConnection', () => { const UUID_B = '22222222-2222-4222-9222-222222222222'; const UUID_V4 = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; + // The idempotency key is minted only on the pi turn-runtime `turn:send` + // branch: reset the generation (clearing the outer `legacy-message` + // advertisement + lock) and advertise `turn-send` for this connection. + beforeEach(async () => { + await act(async () => { + fake.simulateReconnect(); + }); + await act(async () => { + advertiseCapability('turn-send', fake.socket.id); + }); + }); + async function establish(): Promise { await act(async () => { fake.serverEmit('message:ack', { conversationId: 'c1', messageId: 'm1' }); @@ -1155,7 +1224,16 @@ describe('useChatConnection', () => { expect(latest?.state.approvalRequestPending).toBe(false); // The send lock must also be released — a subsequent sendMessage after - // reconnect must not be permanently blocked by the interrupted turn. + // reconnect must not be permanently blocked by the interrupted turn. The + // disconnect also voids the negotiated send protocol (MAJOR-1), so model the + // reconnect handshake — the socket reconnects and the server re-advertises + // how this connection may send — before probing the released lock. + await act(async () => { + fake.simulateReconnect(); + }); + await act(async () => { + advertiseCapability('legacy-message', fake.socket.id); + }); await act(async () => { latest?.actions.sendMessage({ content: 'after reconnect' }); }); @@ -2136,4 +2214,319 @@ describe('useChatConnection', () => { } expect(destroySocketMock).toHaveBeenCalledOnce(); }); + + describe('chat:send-capability protocol negotiation (Task Five MAJOR-1, red-first)', () => { + const capSelection: HarnessSelection = { + harnessId: 'pi', + providerId: 'anthropic', + modelId: 'claude', + }; + const UUID = 'bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb'; + // The one fixed, safe user-facing notice the hook must surface (code + // `send_protocol_unavailable`) when a send is attempted on a connection whose + // advertised protocol is `unavailable`/unknown/absent. Contract-frozen string. + const UNAVAILABLE_NOTICE = 'Chat sending is unavailable on this connection.'; + + // These tests each drive the protocol negotiation themselves, so they must + // start from a clean, unadvertised generation. Reconnect resets protocolRef + // to `unavailable` and clears the outer `legacy-message` first-wins lock + // WITHOUT advertising — no client emit, so `fake.emitted` stays empty and the + // "starts unavailable" premise holds. + beforeEach(async () => { + await act(async () => { + fake.simulateReconnect(); + }); + }); + + async function establishConversation(): Promise { + await act(async () => { + fake.serverEmit('message:ack', { conversationId: 'c1', messageId: 'm1' }); + }); + } + + function connectCalls(): number { + return (fake.socket.connect as unknown as { mock: { calls: unknown[] } }).mock.calls.length; + } + + it('starts with no advertised protocol: a send is refused, emits nothing, mints no key, and surfaces the fixed unavailable notice', async () => { + // No `chat:send-capability` has arrived, so the connection has not been told + // it may send at all. The current impl infers "selection + no conversation + + // no flat provider/model → return false" but SURFACES NOTHING — the red is + // that the fixed `send_protocol_unavailable` notice is never set. + let uuidCalls = 0; + const restore = installRandomUUID(() => { + uuidCalls += 1; + return UUID; + }); + let returned: boolean | undefined; + try { + await act(async () => { + returned = harnessSend()({ content: 'hi', selection: capSelection }); + }); + } finally { + restore(); + } + + expect(returned).toBe(false); + expect(fake.emitted).toHaveLength(0); + expect(latest?.state.error).toBe(UNAVAILABLE_NOTICE); + // The test's name promises "mints no key": the unavailable branch must not + // reach the idempotency mint at all. Without this assertion a defect that + // mints a key before refusing survives. + expect(uuidCalls).toBe(0); + // ...and no user content may be optimistically appended on refusal. + expect(latest?.state.messages.some((m) => m.text === 'hi')).toBe(false); + }); + + it('legacy-message advertised overrides conversation-inference: an established conversation still routes the legacy message event, never turn:send', async () => { + // Same inputs the inference impl routes to `turn:send` (selection + active + // conversation). The advertised protocol is authoritative: it must emit the + // legacy `message` event instead. Red: current impl emits turn:send. + const restore = installRandomUUID(() => UUID); + try { + await establishConversation(); + await act(async () => { + advertiseCapability('legacy-message', fake.socket.id); + }); + await act(async () => { + harnessSend()({ content: 'hi', selection: capSelection }); + }); + } finally { + restore(); + } + + expect(fake.emitted.filter((e) => e.event === 'turn:send')).toHaveLength(0); + expect(fake.emitted).toContainEqual({ + event: 'message', + payload: { conversationId: 'c1', content: 'hi', provider: 'anthropic', modelId: 'claude' }, + }); + }); + + it('legacy-message advertised with no conversation: derives provider/model from the selection tuple and emits one message', async () => { + // The flat provider/modelId caller inputs are gone; the legacy branch must + // source them from the confirmed persisted selection. Red: current impl + // refuses a bare harness send (selection + no flat fields → return false). + let returned: boolean | undefined; + const restore = installRandomUUID(() => UUID); + try { + await act(async () => { + advertiseCapability('legacy-message', fake.socket.id); + }); + await act(async () => { + returned = harnessSend()({ content: 'first', selection: capSelection }); + }); + } finally { + restore(); + } + + expect(returned).toBe(true); + expect(fake.emitted).toContainEqual({ + event: 'message', + payload: { + conversationId: undefined, + content: 'first', + provider: 'anthropic', + modelId: 'claude', + }, + }); + expect(fake.emitted.some((e) => e.event === 'turn:send')).toBe(false); + }); + + it('unavailable advertised: refuses even with an active conversation and selection, emits nothing, surfaces the fixed notice', async () => { + // Red: current impl ignores the advertisement and emits turn:send. + let uuidCalls = 0; + const restore = installRandomUUID(() => { + uuidCalls += 1; + return UUID; + }); + let returned: boolean | undefined; + try { + await establishConversation(); + await act(async () => { + advertiseCapability('unavailable', fake.socket.id); + }); + await act(async () => { + returned = harnessSend()({ content: 'nope', selection: capSelection }); + }); + } finally { + restore(); + } + + expect(returned).toBe(false); + expect(fake.emitted).toHaveLength(0); + expect(latest?.state.error).toBe(UNAVAILABLE_NOTICE); + // Refusal must not optimistically append the user's turn to the transcript + // (a distinct leak from the emit): the unavailable branch appends nothing. + expect(latest?.state.messages.some((m) => m.text === 'nope')).toBe(false); + // ...and must not mint an idempotency key on the refused path. + expect(uuidCalls).toBe(0); + }); + + it('ignores an advertisement whose connectionId does not match the socket id: protocol stays unavailable and the send is refused', async () => { + // A capability minted for a different (stale/foreign) connection must never + // arm this one. Red: current impl has no connection-id gate and emits + // turn:send off the inferred path. + const restore = installRandomUUID(() => UUID); + let returned: boolean | undefined; + try { + await establishConversation(); + await act(async () => { + advertiseCapability('legacy-message', 'a-different-connection'); + }); + await act(async () => { + returned = harnessSend()({ content: 'spoof', selection: capSelection }); + }); + } finally { + restore(); + } + + expect(returned).toBe(false); + expect(fake.emitted).toHaveLength(0); + expect(latest?.state.error).toBe(UNAVAILABLE_NOTICE); + }); + + it('accepts only the first advertisement for the generation: a later conflicting protocol is ignored', async () => { + // legacy-message wins; the subsequent turn-send is a replay/conflict and is + // dropped. Red: current impl ignores both and infers turn:send. + const restore = installRandomUUID(() => UUID); + try { + await establishConversation(); + await act(async () => { + advertiseCapability('legacy-message', fake.socket.id); + }); + await act(async () => { + advertiseCapability('turn-send', fake.socket.id); + }); + await act(async () => { + harnessSend()({ content: 'hi', selection: capSelection }); + }); + } finally { + restore(); + } + + expect(fake.emitted.filter((e) => e.event === 'turn:send')).toHaveLength(0); + expect(fake.emitted).toContainEqual({ + event: 'message', + payload: { conversationId: 'c1', content: 'hi', provider: 'anthropic', modelId: 'claude' }, + }); + }); + + it('resets to unavailable on disconnect: a later send is refused and never reconnects the socket', async () => { + // Disconnect voids the advertised protocol for the generation. The send must + // refuse and MUST NOT call socket.connect() to force a reconnection. Red: + // current impl keeps the conversation, infers turn:send, and its turn:send + // branch calls socket.connect() when the socket is disconnected. + const restore = installRandomUUID(() => UUID); + let returned: boolean | undefined; + let connectsDuringSend = 0; + try { + await establishConversation(); + await act(async () => { + advertiseCapability('legacy-message', fake.socket.id); + }); + await act(async () => { + fake.simulateDisconnect(); + }); + const before = connectCalls(); + await act(async () => { + returned = harnessSend()({ content: 'after-drop', selection: capSelection }); + }); + connectsDuringSend = connectCalls() - before; + } finally { + restore(); + } + + expect(returned).toBe(false); + expect(fake.emitted).toHaveLength(0); + expect(latest?.state.error).toBe(UNAVAILABLE_NOTICE); + expect(connectsDuringSend).toBe(0); + }); + + it('resets on reconnect to a fresh generation: refuses until re-advertised, then honors the new advertisement', async () => { + // A reconnect mints a new Socket.id and a new generation; the prior + // advertisement (bound to the old id) is stale and must not carry over. The + // hook only trusts a fresh advertisement for the new connection. Red: + // current impl has no connect listener and keeps inferring turn:send. + const restore = installRandomUUID(() => UUID); + let refusedAfterReconnect: boolean | undefined; + try { + await establishConversation(); + await act(async () => { + advertiseCapability('legacy-message', fake.socket.id); + }); + await act(async () => { + fake.simulateReconnect('socket-b'); + }); + await act(async () => { + refusedAfterReconnect = harnessSend()({ content: 'stale', selection: capSelection }); + }); + } finally { + restore(); + } + + expect(refusedAfterReconnect).toBe(false); + expect(fake.emitted).toHaveLength(0); + expect(latest?.state.error).toBe(UNAVAILABLE_NOTICE); + + // A fresh advertisement for the reconnected id (socket-b) re-arms sending. + const restore2 = installRandomUUID(() => UUID); + try { + await act(async () => { + advertiseCapability('legacy-message', 'socket-b'); + }); + await act(async () => { + harnessSend()({ content: 'welcome-back', selection: capSelection }); + }); + } finally { + restore2(); + } + + expect(fake.emitted).toContainEqual({ + event: 'message', + payload: { + conversationId: 'c1', + content: 'welcome-back', + provider: 'anthropic', + modelId: 'claude', + }, + }); + expect(fake.emitted.some((e) => e.event === 'turn:send')).toBe(false); + }); + + it('routes on the synchronous protocol ref, not the batched reducer mirror: an advertisement and a send in the SAME tick still route by the just-advertised protocol', async () => { + // An advertisement lands and a send is issued within one synchronous tick, + // before React commits the reducer's `sendProtocol` mirror. The send is + // captured from the pre-advertisement render, so its closed-over reducer + // state still reads `sendProtocol === 'unavailable'`; the capability + // handler, however, has already set the synchronous `protocolRef` to + // `legacy-message`. The hook must route on that ref. Red (against a + // stale-mirror routing that reads `state.sendProtocol`): the send reads the + // pre-advertisement `unavailable` and refuses instead of emitting `message`. + const restore = installRandomUUID(() => UUID); + try { + await act(async () => { + // Bound to the CURRENT (pre-advertisement) render — its closure still + // sees the reset `unavailable` mirror even after the advert dispatches. + const sendBeforeCommit = harnessSend(); + advertiseCapability('legacy-message', fake.socket.id); + // Same tick, no await: React has not committed the new mirror yet, so + // only `protocolRef` reflects `legacy-message`. + sendBeforeCommit({ content: 'same-tick', selection: capSelection }); + }); + } finally { + restore(); + } + + expect(fake.emitted).toContainEqual({ + event: 'message', + payload: { + conversationId: undefined, + content: 'same-tick', + provider: 'anthropic', + modelId: 'claude', + }, + }); + expect(fake.emitted.some((e) => e.event === 'turn:send')).toBe(false); + }); + }); }); diff --git a/apps/web/src/spa/chat/use-chat-connection.ts b/apps/web/src/spa/chat/use-chat-connection.ts index 8ce304a7..b4850e1c 100644 --- a/apps/web/src/spa/chat/use-chat-connection.ts +++ b/apps/web/src/spa/chat/use-chat-connection.ts @@ -22,6 +22,8 @@ import type { AgentStartPayload, AgentTextPayload, AgentThinkingPayload, + ChatSendCapabilityPayload, + ChatSendProtocol, CommandDef, CommandManifest, CommandManifestPayload, @@ -145,6 +147,14 @@ const TURN_REJECTED_NOTICE = 'This turn could not be sent. Please try again.'; * Like {@link TURN_REJECTED_NOTICE}, it never carries the thrown message. */ const IDEMPOTENCY_UNAVAILABLE_NOTICE = 'This turn could not be sent. Please try again.'; +/** The single fixed, browser-safe notice surfaced (with safe code + * `send_protocol_unavailable`) when a send is attempted on a connection whose + * negotiated send protocol is `unavailable` — the server never advertised a + * usable `chat:send-capability`, advertised `unavailable` (e.g. a pi-rpc runtime + * in this slice), or the advertisement was rejected (wrong connection id, replay, + * or an unknown protocol). It carries no dynamic detail. */ +const SEND_PROTOCOL_UNAVAILABLE_NOTICE = 'Chat sending is unavailable on this connection.'; + /** Mints a single idempotency key for one accepted `turn:send`, fail-closed. * Returns a fresh RFC-4122 UUID from `crypto.randomUUID`, or `null` when that * source is absent (not a function) or throws — the caller then refuses the turn @@ -307,6 +317,14 @@ export interface ChatConnectionState { approvalRequestPending: boolean; systemReload: SystemReloadPayload | null; error: string | null; + /** How this connection is currently permitted to send, negotiated via the + * server-to-client-only `chat:send-capability` advertisement. Starts and resets + * to `'unavailable'` on every (re)connect and disconnect — a fresh or dropped + * connection has no usable protocol until the server (re-)advertises. This is + * the reactive/UI mirror of the synchronous `protocolRef` that `sendMessage` + * actually reads; the ref is authoritative because an advertisement and a send + * can occur in the same tick before React re-renders. */ + sendProtocol: ChatSendProtocol; /** Receipt from the most recently accepted harness `turn:ack`, or null before any * turn has been accepted. A rejected turn:ack surfaces via `error` and leaves this * untouched (a prior accepted receipt is not erased by a later rejection). */ @@ -351,12 +369,7 @@ export interface ChatConnectionState { } export interface ChatConnectionActions { - sendMessage: (input: { - content: string; - provider?: string; - modelId?: string; - selection?: HarnessSelection; - }) => boolean; + sendMessage: (input: { content: string; selection?: HarnessSelection }) => boolean; abort: () => void; setThinking: (level: string) => void; executeCommand: (input: { command: string; args?: string }) => void; @@ -389,6 +402,7 @@ const initialState: ChatConnectionState = { approvalRequestPending: false, systemReload: null, error: null, + sendProtocol: 'unavailable', turnReceipt: null, messages: [], messageSeq: 0, @@ -412,6 +426,9 @@ type Action = | { type: 'server/error'; payload: ErrorPayload } | { type: 'server/turn:ack'; payload: HarnessTurnAckPayload } | { type: 'local/send'; content: string } + | { type: 'local/capability'; protocol: ChatSendProtocol } + | { type: 'local/reset-protocol' } + | { type: 'local/send-unavailable' } | { type: 'local/turn-idempotency-unavailable' } | { type: 'local/approve-request'; command: string; args?: string } | { type: 'local/consume-approval' } @@ -877,6 +894,31 @@ function reduce(state: ChatConnectionState, action: Action): ChatConnectionState }; } + case 'local/capability': { + // The FIRST valid `chat:send-capability` for this connection generation has + // been accepted (connection-id gating + first-wins enforced in the handler); + // record how this connection may now send. This is the reactive mirror of + // the synchronous `protocolRef` the send path reads. + return { ...state, sendProtocol: action.protocol }; + } + + case 'local/reset-protocol': { + // A (re)connect or disconnect voids any negotiated protocol: a fresh or + // dropped connection has no usable send capability until the server + // (re-)advertises. Reset to `unavailable` so no stale advertisement can + // authorize a send across a connection boundary. + if (state.sendProtocol === 'unavailable') return state; + return { ...state, sendProtocol: 'unavailable' }; + } + + case 'local/send-unavailable': { + // A send was attempted while the negotiated protocol is `unavailable` + // (never advertised / advertised unavailable / rejected advertisement). + // Surface the single FIXED safe notice — nothing was emitted, minted, + // appended, or locked. + return { ...state, error: SEND_PROTOCOL_UNAVAILABLE_NOTICE }; + } + case 'local/turn-idempotency-unavailable': { // The idempotency-key mint failed closed (crypto.randomUUID absent or // throwing), so the turn was refused before emit. Surface a FIXED notice — @@ -965,6 +1007,20 @@ export function useChatConnection(): ChatConnectionValue { approveLockRef.current = state.approvalRequestPending; }, [state.approvalRequestPending]); + // Synchronous, generation-bound send protocol. `state.sendProtocol` drives the + // reactive UI, but reducer updates are batched/async — a `chat:send-capability` + // advertisement and a `sendMessage` can land in the same tick before React + // re-renders — so this ref is the source of truth the send path reads. Unlike + // sendLockRef/approveLockRef (synchronized FROM the reducer), this ref is + // written directly by the socket lifecycle/capability handlers below, which + // also dispatch the reducer mirror. It is NOT synchronized from state, because + // its whole purpose is to be correct BEFORE the reducer has re-rendered. + const protocolRef = useRef('unavailable'); + // True once the first valid advertisement for the CURRENT connection generation + // has been accepted; every later advertisement (a conflicting or replayed one) + // is ignored until the next (re)connect/disconnect resets the generation. + const protocolLockedRef = useRef(false); + useEffect(() => { const socket = getSocket(); @@ -998,7 +1054,43 @@ export function useChatConnection(): ChatConnectionValue { }; const onTurnAck = (payload: HarnessTurnAckPayload): void => dispatch({ type: 'server/turn:ack', payload }); + + // Void the negotiated send protocol at every connection-lifecycle boundary. + // A fresh or dropped connection has no usable capability until the server + // (re-)advertises, so no advertisement bound to a prior connection may carry + // across the boundary and authorize a send. Both write the synchronous ref + // AND unlock first-wins, then dispatch the reducer mirror. + const resetSendProtocol = (): void => { + protocolRef.current = 'unavailable'; + protocolLockedRef.current = false; + dispatch({ type: 'local/reset-protocol' }); + }; + const onConnect = (): void => { + resetSendProtocol(); + }; + const onCapability = (payload: ChatSendCapabilityPayload): void => { + // Server-to-client-only advertisement of how THIS connection may send. + // Accept only the FIRST valid one per generation, and only when it names + // this exact connection (`connectionId === socket.id`): a capability minted + // for another or stale connection must never arm this one. The payload is + // runtime-untrusted despite its compile-time type, so every field is + // guard-checked and an unknown protocol is dropped (leaving `unavailable`). + if (protocolLockedRef.current) return; + if (!isRecord(payload)) return; + const { protocol, connectionId } = payload as { + protocol?: unknown; + connectionId?: unknown; + }; + if (typeof connectionId !== 'string' || connectionId !== socket.id) return; + if (protocol !== 'legacy-message' && protocol !== 'turn-send' && protocol !== 'unavailable') { + return; + } + protocolLockedRef.current = true; + protocolRef.current = protocol; + dispatch({ type: 'local/capability', protocol }); + }; const onDisconnect = (): void => { + resetSendProtocol(); dispatch({ type: 'local/disconnect' }); }; @@ -1016,6 +1108,10 @@ export function useChatConnection(): ChatConnectionValue { socket.on('system:reload', onSystemReload); socket.on('error', onError); socket.on('turn:ack', onTurnAck); + // Registered BEFORE connect so the initial post-auth advertisement (and any + // reconnect) can never race ahead of its listener. + socket.on('connect', onConnect); + socket.on('chat:send-capability', onCapability); socket.on('disconnect', onDisconnect); if (!socket.connected) { @@ -1037,54 +1133,78 @@ export function useChatConnection(): ChatConnectionValue { socket.off('system:reload', onSystemReload); socket.off('error', onError); socket.off('turn:ack', onTurnAck); + socket.off('connect', onConnect); + socket.off('chat:send-capability', onCapability); socket.off('disconnect', onDisconnect); destroySocket(); }; }, []); const actions: ChatConnectionActions = { - sendMessage: ({ content, provider, modelId, selection }) => { - // Turn-runtime path: a selection tuple against an already-established - // conversation routes through the exclusive `turn:send` contract. It is - // lock-independent by design — it does not engage the send lock, does not - // dispatch `local/send` (no optimistic append), and mints exactly one - // idempotency key per accepted turn. A failed mint fails the turn closed. - if (selection != null && state.conversationId !== null) { - const idempotencyKey = mintIdempotencyKey(); - if (idempotencyKey === null) { - dispatch({ type: 'local/turn-idempotency-unavailable' }); + sendMessage: ({ content, selection }) => { + // Routing is PROTOCOL-driven, never inferred from conversation/selection/ + // provider/local mode: the server advertised, once per connection, exactly + // how this connection may send, and that advertisement is authoritative. + // The exhaustive switch maps each protocol to its ONE event; the send path + // never reconnects the socket (a dropped connection has already reset the + // protocol to `unavailable`, so no emit branch is reachable while offline). + switch (protocolRef.current) { + case 'legacy-message': { + // Embedded/legacy runtime: EVERY browser turn — the first (which + // creates the conversation) and every later one — is the `message` + // event. provider/model are sourced ONLY from the confirmed persisted + // selection tuple, never from separate flat caller inputs. + if (sendLockRef.current || state.streaming || state.sending) return false; + sendLockRef.current = true; + const socket = getSocket(); + dispatch({ type: 'local/send', content }); + socket.emit('message', { + conversationId: state.conversationId ?? undefined, + content, + provider: selection?.providerId, + modelId: selection?.modelId, + }); + return true; + } + case 'turn-send': { + // Pi turn-runtime: the exclusive `turn:send` contract. Requires an + // already-established conversation AND a confirmed persisted selection + // tuple; it is lock-independent (no send lock, no optimistic append), + // and mints exactly one idempotency key per accepted turn, failing the + // turn closed if the mint fails. A premature send (no conversation yet, + // or no selection) is refused with no emit and no notice. + if (selection == null || state.conversationId === null) return false; + const idempotencyKey = mintIdempotencyKey(); + if (idempotencyKey === null) { + dispatch({ type: 'local/turn-idempotency-unavailable' }); + return false; + } + const socket = getSocket(); + socket.emit('turn:send', { + conversationId: state.conversationId, + content, + selection, + idempotencyKey, + }); + return true; + } + case 'unavailable': { + // No usable protocol negotiated for this connection: refuse without + // emitting, minting, appending, or acquiring the lock, and surface the + // one fixed safe notice (code `send_protocol_unavailable`). + dispatch({ type: 'local/send-unavailable' }); + return false; + } + default: { + // Exhaustiveness guard: every ChatSendProtocol member is handled above. + // An unknown value can never arm a send — refuse exactly as + // `unavailable` rather than falling through to any emit. + const _exhaustive: never = protocolRef.current; + void _exhaustive; + dispatch({ type: 'local/send-unavailable' }); return false; } - const socket = getSocket(); - if (!socket.connected) socket.connect(); - socket.emit('turn:send', { - conversationId: state.conversationId, - content, - selection, - idempotencyKey, - }); - return true; } - // A selection tuple with no active conversation and no flat provider/model - // is a bare harness call that cannot create a conversation — refuse it, - // emitting nothing and appending nothing. - if (selection != null && provider === undefined && modelId === undefined) { - return false; - } - // Legacy `message` path (first send that creates the conversation, and every - // pre-turn-runtime send) — unchanged behavior, now reporting acceptance. - if (sendLockRef.current || state.streaming || state.sending) return false; - sendLockRef.current = true; - const socket = getSocket(); - if (!socket.connected) socket.connect(); - dispatch({ type: 'local/send', content }); - socket.emit('message', { - conversationId: state.conversationId ?? undefined, - content, - provider, - modelId, - }); - return true; }, abort: () => { diff --git a/apps/web/src/spa/pages/chat.spec.tsx b/apps/web/src/spa/pages/chat.spec.tsx index 0641865c..1835290b 100644 --- a/apps/web/src/spa/pages/chat.spec.tsx +++ b/apps/web/src/spa/pages/chat.spec.tsx @@ -135,6 +135,31 @@ function installRandomUUID(fn: () => string): () => void { }; } +/** + * Task Five MAJOR-1: the send path is PROTOCOL-driven — the browser may send only + * as the server advertised, once per connection, over the server-to-client-only + * `chat:send-capability`. Model that advertisement for THIS connection id so the + * page send tests take the intended branch. `legacy-message` is the default + * (advertised in `beforeEach`/`remountWithFetch`); the pi turn-runtime tests + * reset the generation and re-advertise `turn-send` via the helper below. + */ +function advertiseSendCapability(protocol: 'legacy-message' | 'turn-send' | 'unavailable'): void { + fake.serverEmit('chat:send-capability', { protocol, connectionId: fake.socket.id }); +} + +/** Reset the negotiated protocol to a fresh, unlocked generation (clearing the + * default `legacy-message` advertisement + first-wins lock), then advertise the + * pi turn-runtime `turn:send` protocol for this connection. The per-test override + * for the page send tests that route through `turn:send`. */ +async function advertiseTurnSendGeneration(): Promise { + await act(async () => { + fake.simulateReconnect(); + }); + await act(async () => { + advertiseSendCapability('turn-send'); + }); +} + let fake: ReturnType; let root: Root | null; let container: HTMLElement; @@ -164,6 +189,12 @@ beforeEach(async () => { // Settle the selection hook's mount fetches so the default in-catalog tuple // persists and `canSend` is true for the existing send-path tests. await flushAsync(); + // Model the server's post-auth send-capability advertisement (MAJOR-1). Most + // page send tests exercise the legacy `message` branch; the pi turn-runtime + // tests override to `turn-send` via advertiseTurnSendGeneration(). + await act(async () => { + advertiseSendCapability('legacy-message'); + }); }); afterEach(async () => { @@ -186,6 +217,11 @@ async function remountWithFetch(fetchImpl: typeof fetch): Promise { root?.render(); }); await flushAsync(); + // Re-advertise on the remounted connection — the prior generation's capability + // does not carry across a remount (fresh hook instance, unadvertised protocol). + await act(async () => { + advertiseSendCapability('legacy-message'); + }); } describe('ChatPage', () => { @@ -599,6 +635,7 @@ describe('ChatPage', () => { }); it('emits turn:send with the nested persisted selection tuple and a UUID idempotency key (never the legacy message event)', async () => { + await advertiseTurnSendGeneration(); const restore = installRandomUUID(() => PAGE_UUID); try { // Send is disabled without an active conversation — establish one first. @@ -635,6 +672,7 @@ describe('ChatPage', () => { }); it('keeps the composer content and emits nothing when the send cannot mint an idempotency key, so the user can retry (composer clears only on success) — Task Five group 5', async () => { + await advertiseTurnSendGeneration(); const failing = installRandomUUID(() => { throw new Error('secure random unavailable'); }); @@ -668,6 +706,7 @@ describe('ChatPage', () => { }); it('clears the composer after a successful turn:send and never falls back to the legacy message event — Task Five group 5', async () => { + await advertiseTurnSendGeneration(); const restore = installRandomUUID(() => PAGE_UUID); try { await act(async () => { @@ -696,6 +735,7 @@ describe('ChatPage', () => { }); it('sends the freshly persisted selection as a nested turn:send tuple after the user changes provider/model — never a stale default or flat fields — Task Five group 5', async () => { + await advertiseTurnSendGeneration(); const restore = installRandomUUID(() => PAGE_UUID); try { // Change the selection away from the mount default and let it persist. diff --git a/packages/types/src/chat/events.ts b/packages/types/src/chat/events.ts index 6b0d447a..51852053 100644 --- a/packages/types/src/chat/events.ts +++ b/packages/types/src/chat/events.ts @@ -150,8 +150,28 @@ export type HarnessTurnAckPayload = readonly selection?: HarnessSelection; }; +/** + * The frozen browser send-protocol advertisement (Task Five; server → client only). + * + * A conversation id or a harness selection never proves that the connected Gateway actually + * handles a given wire event, so after BetterAuth authenticates a browser Socket connection the + * Gateway advertises — exactly once, targeted to that socket — which send event the client may + * use. `legacy-message` in legacy mode, `unavailable` in `pi-rpc` (including test-ready Pi + * graphs); Task Five never advertises `turn-send` (its authenticated handler lands in Task 15). + * Capability is routing information, never authorization: every server handler still enforces + * authentication, ownership, DTO, mode, and runtime checks. + */ +export type ChatSendProtocol = 'legacy-message' | 'turn-send' | 'unavailable'; + +export interface ChatSendCapabilityPayload { + readonly protocol: ChatSendProtocol; + /** Exact Socket.IO id for the authenticated browser connection this advertisement is bound to. */ + readonly connectionId: string; +} + /** Socket.IO typed event map: server → client */ export interface ServerToClientEvents { + 'chat:send-capability': (payload: ChatSendCapabilityPayload) => void; 'message:ack': (payload: MessageAckPayload) => void; 'agent:start': (payload: AgentStartPayload) => void; 'agent:end': (payload: AgentEndPayload) => void; diff --git a/packages/types/src/chat/index.ts b/packages/types/src/chat/index.ts index d5cbaae6..036eae84 100644 --- a/packages/types/src/chat/index.ts +++ b/packages/types/src/chat/index.ts @@ -16,6 +16,8 @@ export type { ChatMessagePayload, HarnessTurnSendPayload, HarnessTurnAckPayload, + ChatSendProtocol, + ChatSendCapabilityPayload, ServerToClientEvents, ClientToServerEvents, } from './events.js';