fix(web,gateway): close P3 re-review#3 findings — sanitize command errors, harden turn-lock & caps

This commit is contained in:
shaggy (mosaic-dev box)
2026-08-10 14:21:52 -05:00
parent 48bb19310d
commit d46a2d675a
7 changed files with 350 additions and 166 deletions
@@ -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();
});
});
@@ -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.',
};
}
}
@@ -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(
<CommandsPanel
manifest={null}
results={[
{ conversationId: 'c1', command: 'model', success: false, message: hostileMessage },
]}
approval={null}
pendingApproval={null}
hasConversation={false}
onExecute={vi.fn()}
onApprove={vi.fn()}
onRunApproved={vi.fn()}
/>,
);
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',
+13 -2
View File
@@ -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))}`}
</li>
))}
</ul>
+4
View File
@@ -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;
@@ -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', {
await expect(
act(async () => {
fake.serverEmitRaw('system:reload', {
commands,
skills,
providers: ['anthropic'],
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 () => {
+119 -84
View File
@@ -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<T>(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<CommandDef>(payload.commands, MAX_MANIFEST_ITEMS);
const skills = capList<SkillCommandDef>(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);