diff --git a/apps/gateway/src/commands/command-executor-p8012.spec.ts b/apps/gateway/src/commands/command-executor-p8012.spec.ts index b22ed05a..59f0ed6a 100644 --- a/apps/gateway/src/commands/command-executor-p8012.spec.ts +++ b/apps/gateway/src/commands/command-executor-p8012.spec.ts @@ -1,3 +1,4 @@ +import { Logger } from '@nestjs/common'; import { describe, it, expect, vi, beforeEach } from 'vitest'; import { CommandExecutorService } from './command-executor.service.js'; import type { SlashCommandPayload } from '@mosaicstack/types'; @@ -258,4 +259,39 @@ describe('CommandExecutorService — P8-012 commands', () => { expect(result.command).toBe('tools'); expect(result.message).toContain('tools'); }); + + // Top-level catch sanitization (P3-4 re-review finding #1): a rejected + // Redis `set` inside /provider login is the only reachable path into the + // top-level catch in `execute()`. The raw exception must be logged + // server-side but never handed back to the socket client. + it('sanitizes the top-level command catch, logging the raw exception but never returning it to the client', async () => { + const distinctiveRawFailure = 'ECONNREFUSED distinctive-raw-redis-failure-token-9f31'; + const failingRedis = { + set: vi.fn().mockRejectedValue(new Error(distinctiveRawFailure)), + get: vi.fn(), + del: vi.fn(), + }; + const failingService = buildService(failingRedis as unknown as typeof mockRedis); + const loggerErrorSpy = vi.spyOn(Logger.prototype, 'error').mockImplementation(() => undefined); + + const payload: SlashCommandPayload = { + command: 'provider', + args: 'login anthropic', + conversationId, + }; + const result = await failingService.execute(payload, userScope); + + expect(result.success).toBe(false); + expect(result.command).toBe('provider'); + expect(result.message).toBe('Command failed due to an internal error.'); + expect(result.message).not.toContain(distinctiveRawFailure); + expect(result.message).not.toContain('ECONNREFUSED'); + + // The real exception is still logged server-side. + expect(loggerErrorSpy).toHaveBeenCalled(); + const loggedText = loggerErrorSpy.mock.calls.map((call) => String(call[0])).join(' '); + expect(loggedText).toContain(distinctiveRawFailure); + + loggerErrorSpy.mockRestore(); + }); }); diff --git a/apps/gateway/src/commands/command-executor.service.ts b/apps/gateway/src/commands/command-executor.service.ts index ca2d3cfd..4388ecaa 100644 --- a/apps/gateway/src/commands/command-executor.service.ts +++ b/apps/gateway/src/commands/command-executor.service.ts @@ -160,7 +160,12 @@ export class CommandExecutorService { } } catch (err) { this.logger.error(`Command /${command} failed: ${err}`); - return { command, conversationId, success: false, message: String(err) }; + return { + command, + conversationId, + success: false, + message: 'Command failed due to an internal error.', + }; } } diff --git a/apps/web/src/spa/chat/commands-panel.spec.tsx b/apps/web/src/spa/chat/commands-panel.spec.tsx index ade87a98..fccc3a08 100644 --- a/apps/web/src/spa/chat/commands-panel.spec.tsx +++ b/apps/web/src/spa/chat/commands-panel.spec.tsx @@ -234,6 +234,27 @@ describe('CommandsPanel', () => { expect(container?.textContent).toContain('Command failed.'); }); + it('bounds an oversized command result message at the render site as defense-in-depth', async () => { + const hostileMessage = 'y'.repeat(50_000); + await render( + , + ); + + const text = container?.textContent ?? ''; + expect(text.length).toBeLessThan(hostileMessage.length); + }); + it('does not throw when the manifest fields are malformed (non-array commands/skills)', async () => { const manifest = { commands: 'not-an-array', diff --git a/apps/web/src/spa/chat/commands-panel.tsx b/apps/web/src/spa/chat/commands-panel.tsx index b0d50eca..827f17e1 100644 --- a/apps/web/src/spa/chat/commands-panel.tsx +++ b/apps/web/src/spa/chat/commands-panel.tsx @@ -1,5 +1,6 @@ import { useState, type ReactElement } from 'react'; import type { PendingApproval } from './use-chat-connection'; +import { MAX_COMMAND_MESSAGE_CHARS } from './limits'; import { asNonEmptyString, asString } from './runtime-guards'; import type { CommandManifest, @@ -13,6 +14,16 @@ import type { * directly, never a raw thrown exception, stack trace, or object value. */ const COMMAND_FAILURE_COPY = 'Command failed.'; +/** Render-site defense-in-depth: `use-chat-connection.ts` already bounds a + * stored command:result message at ingestion, but this component must never + * assume every caller went through that path — bounding again here means a + * hostile/oversized message can never force an unbounded render. */ +function boundMessage(value: string): string { + return value.length > MAX_COMMAND_MESSAGE_CHARS + ? value.slice(0, MAX_COMMAND_MESSAGE_CHARS) + : value; +} + interface CommandsPanelProps { manifest: CommandManifest | null; results: SlashCommandResultPayload[]; @@ -141,9 +152,9 @@ export function CommandsPanel({ /{asString(result.command)}: {result.success ? 'success' : 'failed'} {result.success ? typeof result.message === 'string' && result.message - ? ` — ${result.message}` + ? ` — ${boundMessage(result.message)}` : '' - : ` — ${asNonEmptyString(result.message, COMMAND_FAILURE_COPY)}`} + : ` — ${boundMessage(asNonEmptyString(result.message, COMMAND_FAILURE_COPY))}`} ))} diff --git a/apps/web/src/spa/chat/limits.ts b/apps/web/src/spa/chat/limits.ts index 8c38bc21..1299f4da 100644 --- a/apps/web/src/spa/chat/limits.ts +++ b/apps/web/src/spa/chat/limits.ts @@ -16,3 +16,7 @@ export const MAX_COMMAND_RESULTS = 200; export const MAX_MANIFEST_ITEMS = 500; /** Max executed approval IDs remembered for single-flight dedup. */ export const MAX_EXECUTED_APPROVAL_IDS = 200; +/** Max characters retained for a single command:result message — a hostile + * or malfunctioning gateway must not be able to push an unbounded curated + * success/failure reason into state (or, defensively, onto the page). */ +export const MAX_COMMAND_MESSAGE_CHARS = 1_000; 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 95ad3b90..d340c40a 100644 --- a/apps/web/src/spa/chat/use-chat-connection.spec.tsx +++ b/apps/web/src/spa/chat/use-chat-connection.spec.tsx @@ -287,6 +287,26 @@ describe('useChatConnection', () => { ]); }); + it('bounds an oversized command:result message to a fixed cap before it ever reaches state', async () => { + await act(async () => { + fake.serverEmit('message:ack', { conversationId: 'c1', messageId: 'm1' }); + }); + + const hostileMessage = 'x'.repeat(50_000); + await act(async () => { + fake.serverEmit('command:result', { + conversationId: 'c1', + command: 'model', + success: false, + message: hostileMessage, + }); + }); + + const stored = latest?.state.commandResults.at(-1)?.message ?? ''; + expect(stored.length).toBeLessThan(hostileMessage.length); + expect(stored.length).toBeLessThanOrEqual(1000); + }); + it('pairs a successful command:approval with the pending request so it can be run with its approvalId', async () => { await act(async () => { fake.serverEmit('message:ack', { conversationId: 'c1', messageId: 'm1' }); @@ -1051,7 +1071,7 @@ describe('useChatConnection', () => { expect(latest?.state.manifest?.commands).toHaveLength(MAX_MANIFEST_ITEMS); }); - it('caps a system:reload manifest replacement at MAX_MANIFEST_ITEMS for both the raw stored reload and the manifest', async () => { + it('caps a system:reload manifest replacement at MAX_MANIFEST_ITEMS for commands, skills, and providers, and drops hostile extra fields instead of spreading the raw payload into state', async () => { const commands = Array.from({ length: MAX_MANIFEST_ITEMS + 5 }, (_, i) => ({ name: `cmd${i}`, aliases: [], @@ -1065,15 +1085,21 @@ describe('useChatConnection', () => { description: '', available: true, })); + const providers = Array.from({ length: 550 }, (_, i) => `provider-${i}`); - await act(async () => { - fake.serverEmit('system:reload', { - commands, - skills, - providers: ['anthropic'], - message: 'reloaded', - }); - }); + await expect( + act(async () => { + fake.serverEmitRaw('system:reload', { + commands, + skills, + providers, + message: 'reloaded', + // Hostile field not part of the SystemReloadPayload contract — + // must never survive into state.systemReload. + maliciousExtra: 'should-not-survive', + }); + }), + ).resolves.not.toThrow(); expect(latest?.state.manifest?.commands).toHaveLength(MAX_MANIFEST_ITEMS); expect(latest?.state.manifest?.skills).toHaveLength(MAX_MANIFEST_ITEMS); @@ -1082,6 +1108,8 @@ describe('useChatConnection', () => { // must never leave the uncapped raw payload sitting in state. expect(latest?.state.systemReload?.commands).toHaveLength(MAX_MANIFEST_ITEMS); expect(latest?.state.systemReload?.skills).toHaveLength(MAX_MANIFEST_ITEMS); + expect(latest?.state.systemReload?.providers).toHaveLength(500); + expect(latest?.state.systemReload).not.toHaveProperty('maliciousExtra'); }); it('caps availableThinkingLevels before storing a hostile session:info payload', async () => { @@ -1390,6 +1418,16 @@ describe('useChatConnection', () => { latest?.actions.runApprovedCommand(); }); + // The rejected 201st approval is not silently dropped — it must be + // consumed but also surface a stable, visible notice so the user knows + // why the command did not run. ChatPage renders state.error as + // role="alert". + expect(latest?.state.error).toBe( + 'Approval limit reached for this session. This command was not run.', + ); + expect(latest?.state.approval).toBeNull(); + expect(latest?.state.pendingApproval).toBeNull(); + // A fresh local approval request, replaying the very first approvalId — // this is the replay the old eviction policy would have let through a // second time because it had forgotten ap-0 ever ran. @@ -1473,55 +1511,7 @@ describe('useChatConnection', () => { expect(fake.emitted.filter((e) => e.event === 'message')).toHaveLength(2); }); - it('does not unlock turn B when a stale same-conversation error from turn A arrives', async () => { - await act(async () => { - fake.serverEmit('message:ack', { conversationId: 'c1', messageId: 'm1' }); - fake.serverEmit('agent:start', { conversationId: 'c1' }); - fake.serverEmit('agent:end', { conversationId: 'c1' }); - }); - expect(latest?.state.sending).toBe(false); - - await act(async () => { - latest?.actions.sendMessage({ content: 'turn B' }); - }); - expect(latest?.state.sending).toBe(true); - - // A stale error for turn A, same conversationId, arrives before turn - // B's own ack/start — must not unlock turn B. - await act(async () => { - fake.serverEmit('error', { conversationId: 'c1', error: 'stale turn A failure' }); - }); - expect(latest?.state.sending).toBe(true); - // A recognized-stale terminal must be a true no-op — its message must - // never be displayed/stored, and it must not touch streaming, which - // legitimately belongs to the still in-flight turn B. - expect(latest?.state.error).toBeNull(); - expect(latest?.state.streaming).toBe(false); - - await act(async () => { - latest?.actions.sendMessage({ content: 'turn C attempt' }); - }); - expect(fake.emitted.filter((e) => e.event === 'message')).toHaveLength(1); - expect(latest?.state.messages.some((m) => m.text === 'turn C attempt')).toBe(false); - - // Turn B's own current ack/start arms the lock; its own terminal event - // (here, its own error) can then legitimately release it. - await act(async () => { - fake.serverEmit('message:ack', { conversationId: 'c1', messageId: 'm2' }); - fake.serverEmit('agent:start', { conversationId: 'c1' }); - }); - await act(async () => { - fake.serverEmit('error', { conversationId: 'c1', error: 'turn B failed' }); - }); - expect(latest?.state.sending).toBe(false); - - await act(async () => { - latest?.actions.sendMessage({ content: 'turn D' }); - }); - expect(fake.emitted.filter((e) => e.event === 'message')).toHaveLength(2); - }); - - it('a recognized-stale same-conversation error during an unarmed turn B does not clear approvalRequestPending, does not overwrite the frozen pendingApproval, and does not surface its message', async () => { + it('settles turn B and releases its lock when a pre-ack error arrives on an already-established conversation, allowing a later send', async () => { // Turn A completes normally on c1. await act(async () => { fake.serverEmit('message:ack', { conversationId: 'c1', messageId: 'm1' }); @@ -1530,7 +1520,41 @@ describe('useChatConnection', () => { }); expect(latest?.state.sending).toBe(false); - // Turn B is sent but has not yet been armed by its own ack/start. + await act(async () => { + latest?.actions.sendMessage({ content: 'turn B' }); + }); + expect(latest?.state.sending).toBe(true); + + // A same-conversation error arrives before turn B's own ack/start. Once + // turn A has already fully settled via its own terminal event, ordered + // Socket.IO delivery means this cannot be a leftover of A — the Gateway + // has nothing left in flight to emit for a turn it already finished. It + // can only be a genuine error for the newly sent turn B (e.g. a + // session-creation failure emitted before ack), so it must settle B. + await act(async () => { + fake.serverEmit('error', { conversationId: 'c1', error: 'turn B session failure' }); + }); + expect(latest?.state.sending).toBe(false); + expect(latest?.state.streaming).toBe(false); + expect(latest?.state.error).toBe('turn B session failure'); + + await act(async () => { + latest?.actions.sendMessage({ content: 'turn C' }); + }); + expect(fake.emitted.filter((e) => e.event === 'message')).toHaveLength(2); + expect(latest?.state.messages.some((m) => m.text === 'turn C')).toBe(true); + }); + + it('invalidates an outstanding approval request when a pre-ack error settles turn B on an already-established conversation', async () => { + // Turn A completes normally on c1. + await act(async () => { + fake.serverEmit('message:ack', { conversationId: 'c1', messageId: 'm1' }); + fake.serverEmit('agent:start', { conversationId: 'c1' }); + fake.serverEmit('agent:end', { conversationId: 'c1' }); + }); + expect(latest?.state.sending).toBe(false); + + // Turn B is sent but has not yet received its own ack/start. await act(async () => { latest?.actions.sendMessage({ content: 'turn B' }); }); @@ -1538,38 +1562,86 @@ describe('useChatConnection', () => { // An approval request is outstanding on c1 — approveCommand has no // dependency on `sending`/`streaming`, so this is legitimate even while - // turn B is unarmed. + // turn B has not yet started. await act(async () => { latest?.actions.approveCommand({ command: 'deploy', args: 'prod' }); }); expect(latest?.state.approvalRequestPending).toBe(true); - expect(latest?.state.pendingApproval).toEqual({ command: 'deploy', args: 'prod' }); - expect(fake.emitted.filter((e) => e.event === 'command:approve')).toHaveLength(1); - // A stale error from turn A, same conversationId, arrives before turn - // B's own ack/start. This must be a true no-op: it must not release - // `sending`, must not display/store its message, must not touch - // `streaming`, and — critically — must not clear - // `approvalRequestPending`/`pendingApproval`, which would re-arm the - // approve UI for a request that is still outstanding. + // A genuine pre-ack error for turn B is terminal — same as an + // active-turn error, it invalidates any approval request still awaiting + // a response, since the Gateway that just errored is unlikely to still + // answer it. await act(async () => { - fake.serverEmit('error', { conversationId: 'c1', error: 'stale turn A failure' }); + fake.serverEmit('error', { conversationId: 'c1', error: 'turn B session failure' }); }); - expect(latest?.state.sending).toBe(true); - expect(latest?.state.approvalRequestPending).toBe(true); - expect(latest?.state.pendingApproval).toEqual({ command: 'deploy', args: 'prod' }); - expect(latest?.state.error).toBeNull(); - expect(latest?.state.streaming).toBe(false); + expect(latest?.state.sending).toBe(false); + expect(latest?.state.approvalRequestPending).toBe(false); + expect(latest?.state.error).toBe('turn B session failure'); - // A second approval attempt while the first is still outstanding must - // still be rejected — exactly one command:approve total, and the - // original frozen command+args must be unchanged. + // A fresh approval request can be issued again after the invalidation. await act(async () => { latest?.actions.approveCommand({ command: 'deploy', args: 'staging' }); }); - expect(fake.emitted.filter((e) => e.event === 'command:approve')).toHaveLength(1); - expect(latest?.state.pendingApproval).toEqual({ command: 'deploy', args: 'prod' }); + expect(fake.emitted.filter((e) => e.event === 'command:approve')).toHaveLength(2); + }); + + it("does not settle turn B when a duplicate agent:end from already-settled turn A is redelivered after B's own ack but before B's own start", async () => { + // Turn A completes normally on c1. + await act(async () => { + fake.serverEmit('message:ack', { conversationId: 'c1', messageId: 'm1' }); + fake.serverEmit('agent:start', { conversationId: 'c1' }); + fake.serverEmit('agent:end', { conversationId: 'c1' }); + }); + expect(latest?.state.sending).toBe(false); + + // Turn B is sent and its own ack arrives. Under the old + // ack-arms-the-lock design this alone made the lock releasable by any + // same-conversation terminal — the precise bug: it could not yet + // distinguish B's own eventual agent:end from a late duplicate delivery + // of A's already-consumed one. + await act(async () => { + latest?.actions.sendMessage({ content: 'turn B' }); + }); + await act(async () => { + fake.serverEmit('message:ack', { conversationId: 'c1', messageId: 'm2' }); + }); + expect(latest?.state.sending).toBe(true); + + // A duplicate/straggler agent:end for turn A (already fully settled + // above) is redelivered for the same conversation before B's own + // agent:start ever arrived. Only B's own accepted agent:start may move + // it into the active phase that a real agent:end may settle — this + // duplicate must be a true no-op. + await act(async () => { + fake.serverEmit('agent:end', { conversationId: 'c1' }); + }); + expect(latest?.state.sending).toBe(true); + expect(latest?.state.streaming).toBe(false); + + // Turn C must still be blocked — the lock is still genuinely held by B. + await act(async () => { + latest?.actions.sendMessage({ content: 'turn C attempt' }); + }); + expect(fake.emitted.filter((e) => e.event === 'message')).toHaveLength(1); + expect(latest?.state.messages.some((m) => m.text === 'turn C attempt')).toBe(false); + + // B's own start, then its own end, legitimately unlocks it. + await act(async () => { + fake.serverEmit('agent:start', { conversationId: 'c1' }); + }); + expect(latest?.state.streaming).toBe(true); + + await act(async () => { + fake.serverEmit('agent:end', { conversationId: 'c1' }); + }); + expect(latest?.state.sending).toBe(false); + + await act(async () => { + latest?.actions.sendMessage({ content: 'turn D' }); + }); + expect(fake.emitted.filter((e) => e.event === 'message')).toHaveLength(2); }); it('removes every listener and tears down the socket on cleanup, using no network', async () => { diff --git a/apps/web/src/spa/chat/use-chat-connection.ts b/apps/web/src/spa/chat/use-chat-connection.ts index 319d6391..9f85ad99 100644 --- a/apps/web/src/spa/chat/use-chat-connection.ts +++ b/apps/web/src/spa/chat/use-chat-connection.ts @@ -1,6 +1,7 @@ import { useEffect, useReducer, useRef } from 'react'; import { destroySocket, getSocket } from '@/lib/socket'; import { + MAX_COMMAND_MESSAGE_CHARS, MAX_COMMAND_RESULTS, MAX_EXECUTED_APPROVAL_IDS, MAX_MANIFEST_ITEMS, @@ -121,6 +122,14 @@ function isValidToolCallId(value: unknown): value is string { * startup failure instead of leaving `sending`/`pendingSend` stuck forever. */ const CONVERSATION_START_FAILURE = 'Unable to start this conversation. Please try again.'; +/** Shown when the non-evicting executed-approval dedup cache is already at + * MAX_EXECUTED_APPROVAL_IDS and a distinct, never-before-seen approval is + * denied as a result (see `runApprovedCommand`'s saturation branch below). + * The approval is still consumed (the UI lock is released) but the user + * must see why the command did not run rather than have it silently + * dropped. */ +const APPROVAL_LIMIT_MESSAGE = 'Approval limit reached for this session. This command was not run.'; + /** True only for the narrow case a malformed-conversationId `error`/`agent:end` * must be treated as a terminal startup failure: no conversation has ever been * established yet, and a send is still pending one. Once a conversation is @@ -131,17 +140,6 @@ function isUnrecoverableStartupFailure(state: ChatConnectionState): boolean { return state.conversationId === null && state.pendingSend; } -/** True when a scoped `agent:end`/`error` is a recognized-stale terminal: a - * send is genuinely in flight (`sending`) but the current turn's own - * ack/start has not armed `armedTurnToken` yet — so this event must be a - * leftover from an earlier turn on the same conversation, not this one's own - * terminal. Recognizing this must make the event a true no-op: it may not - * finalize/clear transient state (text, thinking, streaming, error, - * approvalRequestPending) that belongs to the still in-flight turn. */ -function isRecognizedStaleTerminal(state: ChatConnectionState): boolean { - return state.sending && state.armedTurnToken !== state.turnToken; -} - /** Deterministic, unique-per-event fallback id for a malformed `toolCallId`. * Sourced from a monotonically increasing counter carried in state (`toolSeq`) * rather than `tools.length`, so it stays collision-free even once `tools` is @@ -159,6 +157,15 @@ function capList(arr: unknown, max: number): T[] { return (arr.length > max ? arr.slice(0, max) : arr) as T[]; } +/** Like `capList`, but for a server-provided string collection (e.g. + * `system:reload.providers`): a non-string entry is dropped rather than + * invalidating the whole list, and the result is capped at `max`. */ +function capStringList(arr: unknown, max: number): string[] { + if (!Array.isArray(arr)) return []; + const strings = arr.filter((item): item is string => typeof item === 'string'); + return strings.length > max ? strings.slice(0, max) : strings; +} + /** Normalizes a raw message:ack payload before it is stored — `messageId` is * given a stable, visible "unknown" fallback rather than storing a raw * non-string value (which would either render as "[object Object]" or, @@ -198,7 +205,10 @@ function sanitizeApproval( /** Normalizes a raw command:result payload before it is stored — `success` * is coerced to a literal boolean so a truthy non-boolean value (e.g. an - * object) can never be displayed as a successful result. */ + * object) can never be displayed as a successful result, and a curated + * `message` is bounded to MAX_COMMAND_MESSAGE_CHARS so a hostile/malformed + * gateway cannot push an unbounded string into state (the render site in + * commands-panel.tsx applies the same bound again as defense-in-depth). */ function sanitizeCommandResult( payload: SlashCommandResultPayload, conversationId: string, @@ -207,7 +217,10 @@ function sanitizeCommandResult( conversationId, command: asString(payload.command, 'unknown'), success: payload.success === true, - message: typeof payload.message === 'string' ? payload.message : undefined, + message: + typeof payload.message === 'string' + ? payload.message.slice(0, MAX_COMMAND_MESSAGE_CHARS) + : undefined, }; } @@ -266,20 +279,32 @@ export interface ChatConnectionState { * tools remain, so fallback ids stay unique across the MAX_TOOLS cap * boundary. */ toolSeq: number; - /** Monotonically increasing id for the current in-flight turn, minted at - * local send time. Wire payloads (agent:end/error) carry only a - * conversationId, not a turn-scoped correlation id — two different turns - * on the SAME conversation are indistinguishable on the wire, so this - * local counter is what actually tells them apart. */ - turnToken: number; - /** The `turnToken` value (if any) that has been armed to accept a normal - * terminal release, set only by an accepted (conversation-scoped) - * message:ack or agent:start for the CURRENT token. A same-conversation - * agent:end/error may only set `sending: false` when this equals - * `turnToken` — otherwise it is a stale terminal event from an earlier - * turn on the same conversation and must not unlock a still-in-flight - * later turn. Fails closed: an unarmed token can never be unlocked. */ - armedTurnToken: number | null; + /** Explicit per-turn phase for the CURRENT (most recently sent) turn. Wire + * payloads (agent:end/error) carry only a conversationId, not a + * turn-scoped correlation id — two different turns on the SAME + * conversation are indistinguishable on the wire by conversationId alone, + * so this local phase is what actually tells them apart: + * + * - 'pending': minted by `local/send`; the turn has been sent but has not + * yet received its own `agent:start`. A same-conversation `agent:end` + * arriving in this phase cannot be trusted as this turn's own — Socket.IO + * delivers in order and the Gateway always emits `agent:start` before a + * legitimate `agent:end`, so this can only be a stale/duplicate end left + * over from an earlier turn (a `message:ack` alone does NOT advance this + * phase, precisely to close that hole). A same-conversation `error`, + * however, IS trusted in this phase: the Gateway can legitimately emit a + * session-creation error before ack/start, and once an earlier turn on + * this conversation has already fully settled (via its own terminal + * event) ordered delivery guarantees nothing further remains in flight + * for it — so any later scoped error can only belong to this turn. + * - 'active': set only by this turn's own accepted `agent:start`. Only + * from this phase may a scoped `agent:end` settle the turn. + * - 'settled': set once this turn's own terminal (`agent:end` or `error`) + * has been processed. A further scoped `agent:end`/`error` in this phase + * is a true no-op — it must not re-finalize or re-release anything. + * + * Fails closed: `agent:end` can only ever settle from 'active'. */ + turnPhase: 'pending' | 'active' | 'settled'; } export interface ChatConnectionActions { @@ -319,8 +344,7 @@ const initialState: ChatConnectionState = { messages: [], messageSeq: 0, toolSeq: 0, - turnToken: 0, - armedTurnToken: null, + turnPhase: 'settled', }; type Action = @@ -340,6 +364,7 @@ type Action = | { type: 'local/send'; content: string } | { type: 'local/approve-request'; command: string; args?: string } | { type: 'local/consume-approval' } + | { type: 'local/approval-saturated' } | { type: 'local/disconnect' }; /** @@ -371,11 +396,13 @@ function resolveScopedConversation( if (state.conversationId === null && state.pendingSend) { // The very first scoped event for a brand-new conversation can only belong // to the currently in-flight turn — no prior same-conversation turn exists to - // be confused with. Arm the current turnToken so a terminal event (error, - // agent:end) arriving before ack/start can still properly release the send lock. + // be confused with. This only establishes conversation identity; it does not + // advance `turnPhase` — only this turn's own `agent:start` may do that (see + // the 'server/agent:start' case), and only this turn's own `error` may settle + // it directly from 'pending' (see the 'server/error' case). return { active: true, - state: { ...state, conversationId, pendingSend: false, armedTurnToken: state.turnToken }, + state: { ...state, conversationId, pendingSend: false }, }; } return { active: false, state: null }; @@ -413,17 +440,15 @@ function reduce(state: ChatConnectionState, action: Action): ChatConnectionState conversationId, pendingSend: false, ack: sanitizeAck(payload, conversationId), - // The very first scoped event ever received for a brand-new - // conversation can only belong to the currently in-flight turn — - // no prior same-conversation turn can exist to be confused with. - armedTurnToken: state.turnToken, + // Deliberately does NOT advance `turnPhase` — an ack alone must + // never be enough to make a later `agent:end` settle this turn. + // Only this turn's own `agent:start` may do that. }; } if (payload.conversationId !== state.conversationId) return state; return { ...state, ack: sanitizeAck(payload, state.conversationId), - armedTurnToken: state.turnToken, }; } @@ -439,7 +464,9 @@ function reduce(state: ChatConnectionState, action: Action): ChatConnectionState thinkingDroppedChars: 0, tools: [], error: null, - armedTurnToken: resolved.state.turnToken, + // Only this turn's own accepted agent:start may move it into the + // 'active' phase that a real agent:end may later settle. + turnPhase: 'active', }; } @@ -539,18 +566,24 @@ function reduce(state: ChatConnectionState, action: Action): ChatConnectionState sending: false, streaming: false, approvalRequestPending: false, - armedTurnToken: null, + turnPhase: 'settled', error: CONVERSATION_START_FAILURE, }; } return state; } const next = resolved.state; - if (isRecognizedStaleTerminal(next)) { - // A same-conversation agent:end recognized as stale must be a true - // no-op — it must not finalize the stale turn's leftover - // text/thinking into a message or reset streaming, both of which - // legitimately belong to the still in-flight later turn. + if (next.turnPhase !== 'active') { + // Only an 'active' turn (one that has received its own agent:start) + // may be settled by an agent:end. A 'pending' phase means this + // turn has not started yet — Socket.IO delivers in order and the + // Gateway always emits agent:start before a legitimate agent:end, so + // this can only be a stale/duplicate end left over from an earlier + // turn. A 'settled' phase means this turn's own end already fired — + // this is a redelivered duplicate. Either way this must be a true + // no-op: it must not finalize leftover text/thinking into a message + // or reset streaming, both of which legitimately belong to the + // still in-flight (or already-finalized) current turn. return next; } const hasContent = next.text.length > 0 || next.thinking.length > 0; @@ -569,19 +602,11 @@ function reduce(state: ChatConnectionState, action: Action): ChatConnectionState MAX_MESSAGES, ) : next.messages; - // The wire carries only conversationId, not a turn id — a stale - // agent:end left over from an EARLIER turn on this SAME conversation - // is indistinguishable from this turn's own by conversationId alone. - // Only release `sending` when this turn's own accepted ack/start armed - // the current token; otherwise this is treated as a stale echo and the - // lock stays held (fail closed) so it cannot unlock a later, still - // in-flight turn. - const canRelease = next.armedTurnToken === next.turnToken; return { ...next, streaming: false, - sending: canRelease ? false : next.sending, - armedTurnToken: canRelease ? null : next.armedTurnToken, + sending: false, + turnPhase: 'settled', text: '', thinking: '', textDroppedChars: 0, @@ -671,19 +696,19 @@ function reduce(state: ChatConnectionState, action: Action): ChatConnectionState case 'server/system:reload': { const { payload } = action; - // Computed once and reused for both `systemReload` and `manifest` — - // spreading the raw `payload` into `systemReload` first and only - // capping the copy handed to `manifest` left the raw, uncapped - // commands/skills sitting in `state.systemReload`. The sanitized - // fields are placed after the spread below so they always win. + // `systemReload` is built from exactly the SystemReloadPayload + // contract fields (commands, skills, providers, message) after + // runtime normalization — the raw payload is never spread into state, + // so a hostile/malfunctioning gateway cannot smuggle extra fields in. const commands = capList(payload.commands, MAX_MANIFEST_ITEMS); const skills = capList(payload.skills, MAX_MANIFEST_ITEMS); + const providers = capStringList(payload.providers, MAX_MANIFEST_ITEMS); return { ...state, systemReload: { - ...payload, commands, skills, + providers, message: asString(payload.message, 'Commands reloaded.'), }, manifest: { @@ -705,31 +730,33 @@ function reduce(state: ChatConnectionState, action: Action): ChatConnectionState sending: false, streaming: false, approvalRequestPending: false, - armedTurnToken: null, + turnPhase: 'settled', error: CONVERSATION_START_FAILURE, }; } return state; } const next = resolved.state; - if (isRecognizedStaleTerminal(next)) { - // Same true-no-op guard as `server/agent:end` above: a - // same-conversation error recognized as stale must not display/store - // its message, touch streaming, or clear approvalRequestPending — - // doing so would re-arm the approve UI for a request that belongs to - // the still in-flight later turn while the first remains outstanding. - return next; + if (next.turnPhase === 'settled') { + // No in-flight current turn: the error is still visibly surfaced, + // but there is no turn/approval lock left to mutate. + return { ...next, error: asString(payload.error, 'An error occurred.') }; } - // Same fail-closed turn-token guard as `server/agent:end` above: a - // same-conversation error left over from an earlier, already-finished - // turn must not release a later turn's still-in-flight send lock. - const canRelease = next.armedTurnToken === next.turnToken; + // A current turn-scoped error is terminal from either 'pending' + // (pre-ack — the Gateway can emit a session-creation error before + // ack) or 'active' (a prompt error after ack/start) state. Once an + // earlier turn on this conversation has already fully settled (its + // own terminal event processed, moving turnPhase to 'settled' before + // this turn was even sent), ordered Socket.IO delivery guarantees + // nothing further remains in flight for it — so a scoped error seen + // while turnPhase is 'pending' or 'active' can only belong to THIS + // turn, and must settle and release the lock. return { ...next, error: asString(payload.error, 'An error occurred.'), streaming: false, - sending: canRelease ? false : next.sending, - armedTurnToken: canRelease ? null : next.armedTurnToken, + sending: false, + turnPhase: 'settled', // A turn-scoped error also invalidates any approval request awaiting // a response — the gateway that just errored is unlikely to still // answer it, and the synchronous approveLockRef mirrors this field. @@ -752,12 +779,11 @@ function reduce(state: ChatConnectionState, action: Action): ChatConnectionState messageSeq: state.messageSeq + 1, error: null, sending: true, - // Sending a new turn mints a fresh token and immediately clears - // terminal eligibility — only THIS turn's own accepted ack/start may - // arm it, so a same-conversation terminal left over from the turn - // that just finished can never be mistaken for this one's. - turnToken: state.turnToken + 1, - armedTurnToken: null, + // Sending a new turn begins a fresh 'pending' phase — only THIS + // turn's own accepted agent:start may move it to 'active', so a + // same-conversation agent:end left over from the turn that just + // finished can never be mistaken for this one's. + turnPhase: 'pending', pendingSend: state.conversationId === null ? true : state.pendingSend, }; } @@ -771,7 +797,7 @@ function reduce(state: ChatConnectionState, action: Action): ChatConnectionState sending: false, pendingSend: false, approvalRequestPending: false, - armedTurnToken: null, + turnPhase: 'settled', }; } @@ -792,6 +818,14 @@ function reduce(state: ChatConnectionState, action: Action): ChatConnectionState return { ...state, approval: null, pendingApproval: null }; } + case 'local/approval-saturated': { + // The non-evicting executed-approval cache is full and this is a + // distinct, never-before-seen approval — it is denied (fail closed), + // but unlike a plain replay of an already-executed ID, this must be + // visibly surfaced rather than silently dropped. + return { ...state, approval: null, pendingApproval: null, error: APPROVAL_LIMIT_MESSAGE }; + } + default: return state; } @@ -973,9 +1007,10 @@ export function useChatConnection(): ChatConnectionValue { // unmounted (lifecycle reset), not recoverable by re-approving — a // false positive/availability cost but a genuine replay of any ID // ever seen by this hook can never execute a second time. Consuming - // (rather than silently no-op'ing) releases the UI lock so the - // denial is visible/recoverable by remounting. - dispatch({ type: 'local/consume-approval' }); + // (rather than silently no-op'ing) releases the UI lock, and unlike + // the plain-replay branch above, sets a visible error notice so the + // denial is surfaced rather than silently dropped. + dispatch({ type: 'local/approval-saturated' }); return; } executedApprovalIds.current.add(approval.approvalId);