1047 lines
44 KiB
TypeScript
1047 lines
44 KiB
TypeScript
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,
|
|
MAX_MESSAGES,
|
|
MAX_STREAM_CHARS,
|
|
MAX_TOOLS,
|
|
} from './limits';
|
|
import {
|
|
asConversationId,
|
|
asFiniteNumber,
|
|
asString,
|
|
asStringArray,
|
|
isRecord,
|
|
} from './runtime-guards';
|
|
import type {
|
|
AgentEndPayload,
|
|
AgentStartPayload,
|
|
AgentTextPayload,
|
|
AgentThinkingPayload,
|
|
CommandDef,
|
|
CommandManifest,
|
|
CommandManifestPayload,
|
|
ErrorPayload,
|
|
MessageAckPayload,
|
|
SessionInfoPayload,
|
|
SessionUsagePayload,
|
|
SkillCommandDef,
|
|
SlashCommandApprovalResultPayload,
|
|
SlashCommandResultPayload,
|
|
SystemReloadPayload,
|
|
ToolEndPayload,
|
|
ToolStartPayload,
|
|
} from '@/lib/chat-contract';
|
|
|
|
export interface ToolCallState {
|
|
toolCallId: string;
|
|
toolName: string;
|
|
status: 'running' | 'success' | 'error' | 'anomaly';
|
|
}
|
|
|
|
/** Formats the visible marker prepended to a capped stream buffer once any
|
|
* original characters have been dropped from it. Reconstructing the prior
|
|
* tail (see `capAppendStream` below) always slices this exact computed
|
|
* length off the front of the previous displayed value — it never scans
|
|
* buffer content for marker-shaped text, so real streamed content that
|
|
* happens to look like a marker can never be mistaken for one. */
|
|
function formatTruncationMarker(dropped: number): string {
|
|
return `…[truncated ${dropped} characters]…`;
|
|
}
|
|
|
|
interface CappedStream {
|
|
/** The full value to store/display — a plain tail when nothing has been
|
|
* dropped yet, otherwise the marker followed by the retained tail. Always
|
|
* bounded to at most `max` characters in total. */
|
|
displayed: string;
|
|
/** Total original stream characters dropped so far, cumulative across
|
|
* every append — never reset while the buffer is still accumulating. */
|
|
dropped: number;
|
|
}
|
|
|
|
/** Appends `addition` to a stream buffer capped at `max` displayed
|
|
* characters (marker included), honestly disclosing how many original
|
|
* characters have been dropped so far rather than silently retaining only
|
|
* the tail. `priorDisplayed`/`priorDropped` come from state; the marker (if
|
|
* any) already present in `priorDisplayed` is stripped by the exact length
|
|
* `formatTruncationMarker(priorDropped)` computes, not by pattern-matching. */
|
|
function capAppendStream(
|
|
priorDisplayed: string,
|
|
priorDropped: number,
|
|
addition: string,
|
|
max: number,
|
|
): CappedStream {
|
|
const priorTail =
|
|
priorDropped > 0
|
|
? priorDisplayed.slice(formatTruncationMarker(priorDropped).length)
|
|
: priorDisplayed;
|
|
const combinedTail = priorTail + addition;
|
|
|
|
if (priorDropped === 0 && combinedTail.length <= max) {
|
|
return { displayed: combinedTail, dropped: 0 };
|
|
}
|
|
|
|
// The marker's own text grows (rarely) as `dropped`'s digit count grows,
|
|
// which shrinks the budget left for the tail, which can in turn increase
|
|
// `dropped` further — a handful of iterations is always enough to reach a
|
|
// fixed point for any realistic character count.
|
|
let dropped = priorDropped;
|
|
for (let i = 0; i < 8; i += 1) {
|
|
const budget = Math.max(0, max - formatTruncationMarker(dropped).length);
|
|
const nextDropped = priorDropped + Math.max(0, combinedTail.length - budget);
|
|
if (nextDropped === dropped) break;
|
|
dropped = nextDropped;
|
|
}
|
|
|
|
const marker = formatTruncationMarker(dropped);
|
|
const budget = Math.max(0, max - marker.length);
|
|
const tail = combinedTail.slice(Math.max(0, combinedTail.length - budget));
|
|
return { displayed: marker + tail, dropped };
|
|
}
|
|
|
|
/** Pushes `item` onto `arr`, dropping the oldest entries once `max` is exceeded. */
|
|
function capPush<T>(arr: T[], item: T, max: number): T[] {
|
|
const next = [...arr, item];
|
|
return next.length > max ? next.slice(next.length - max) : next;
|
|
}
|
|
|
|
/** A valid `toolCallId` from the wire is a non-empty string — a malformed or
|
|
* absent one (non-string, or empty string) must never be treated as if it
|
|
* named a real tool call. */
|
|
function isValidToolCallId(value: unknown): value is string {
|
|
return typeof value === 'string' && value.length > 0;
|
|
}
|
|
|
|
/** Shown when a scoped `error`/`agent:end` cannot be attributed to any
|
|
* conversation (a malformed/missing conversationId) while a send is still
|
|
* pending and no conversation has ever been established. There is no valid
|
|
* identity left to recover the turn under, so it is surfaced as a terminal
|
|
* 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
|
|
* active, or when no send is pending, an unattributable event stays a no-op —
|
|
* dropping it (rather than guessing which turn it belongs to) is what lets a
|
|
* later genuinely valid event still recover the turn. */
|
|
function isUnrecoverableStartupFailure(state: ChatConnectionState): boolean {
|
|
return state.conversationId === null && state.pendingSend;
|
|
}
|
|
|
|
/** 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
|
|
* capped at MAX_TOOLS and its length stops changing. */
|
|
function nextToolFallbackId(toolSeq: number): string {
|
|
return `unknown-tool-call-${toolSeq}`;
|
|
}
|
|
|
|
/** Truncates a server-provided collection to at most `max` entries. `arr` is
|
|
* `unknown` because it comes straight from a raw socket payload — a
|
|
* malformed/compromised gateway can send `null` or any non-array value here,
|
|
* which must normalize to an empty list rather than throw. */
|
|
function capList<T>(arr: unknown, max: number): T[] {
|
|
if (!Array.isArray(arr)) return [];
|
|
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,
|
|
* with an unguarded render site, throw). */
|
|
function sanitizeAck(payload: MessageAckPayload, conversationId: string): MessageAckPayload {
|
|
return { conversationId, messageId: asString(payload.messageId, 'unknown') };
|
|
}
|
|
|
|
/** Normalizes a raw command:approval payload before it is stored. A command
|
|
* string matching `pendingApproval.command` alone is not sufficient proof of
|
|
* a genuine, executable approval: `success` must be the literal boolean
|
|
* `true`, and a successful approval requires a non-empty string
|
|
* `approvalId`. Anything else is normalized to a denial (`success: false`,
|
|
* no `approvalId`) so the pending request resolves and the approve lock is
|
|
* released instead of leaving the UI hung on a response it can never trust
|
|
* enough to enable "Run approved command" for. `expiresAt`/`message` are
|
|
* retained only when they are strings (message is never rendered as-is). */
|
|
function sanitizeApproval(
|
|
payload: SlashCommandApprovalResultPayload,
|
|
conversationId: string,
|
|
command: string,
|
|
): SlashCommandApprovalResultPayload {
|
|
const approvalId =
|
|
typeof payload.approvalId === 'string' && payload.approvalId.length > 0
|
|
? payload.approvalId
|
|
: undefined;
|
|
const success = payload.success === true && approvalId !== undefined;
|
|
return {
|
|
conversationId,
|
|
command,
|
|
success,
|
|
approvalId: success ? approvalId : undefined,
|
|
expiresAt: typeof payload.expiresAt === 'string' ? payload.expiresAt : undefined,
|
|
message: typeof payload.message === 'string' ? payload.message : undefined,
|
|
};
|
|
}
|
|
|
|
/** 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, 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,
|
|
): SlashCommandResultPayload {
|
|
return {
|
|
conversationId,
|
|
command: asString(payload.command, 'unknown'),
|
|
success: payload.success === true,
|
|
message:
|
|
typeof payload.message === 'string'
|
|
? payload.message.slice(0, MAX_COMMAND_MESSAGE_CHARS)
|
|
: undefined,
|
|
};
|
|
}
|
|
|
|
export interface ChatTranscriptMessage {
|
|
id: string;
|
|
role: 'user' | 'assistant';
|
|
text: string;
|
|
thinking?: string;
|
|
}
|
|
|
|
export interface PendingApproval {
|
|
command: string;
|
|
args?: string;
|
|
}
|
|
|
|
export interface ChatConnectionState {
|
|
conversationId: string | null;
|
|
/** True once a message has been sent while no conversation is active yet, so the
|
|
* first scoped server event naming a conversation may establish it (the gateway
|
|
* does not guarantee message:ack is the first event for a new conversation). */
|
|
pendingSend: boolean;
|
|
/** True from the moment a message is sent until its turn ends (agent:end),
|
|
* errors, or the connection drops — guards against a second turn starting
|
|
* while one is already in flight. */
|
|
sending: boolean;
|
|
ack: MessageAckPayload | null;
|
|
streaming: boolean;
|
|
text: string;
|
|
/** Total original `agent:text` characters dropped so far by the
|
|
* MAX_STREAM_CHARS cap on `text` — tracked separately from `text` itself
|
|
* so the honest cumulative count survives across multiple appends without
|
|
* re-parsing any marker embedded in the displayed string. */
|
|
textDroppedChars: number;
|
|
thinking: string;
|
|
/** Same accounting as `textDroppedChars`, for `thinking`. */
|
|
thinkingDroppedChars: number;
|
|
tools: ToolCallState[];
|
|
usage: SessionUsagePayload | null;
|
|
sessionInfo: SessionInfoPayload | null;
|
|
manifest: CommandManifest | null;
|
|
commandResults: SlashCommandResultPayload[];
|
|
approval: SlashCommandApprovalResultPayload | null;
|
|
pendingApproval: PendingApproval | null;
|
|
/** True while an approval request has been sent and no response has arrived yet. */
|
|
approvalRequestPending: boolean;
|
|
systemReload: SystemReloadPayload | null;
|
|
error: string | null;
|
|
messages: ChatTranscriptMessage[];
|
|
/** Monotonically increasing counter used to mint transcript message ids —
|
|
* never reset while retained messages remain, so ids stay unique across the
|
|
* MAX_MESSAGES cap boundary (unlike a `messages.length`-derived id, which
|
|
* plateaus once the array is capped). */
|
|
messageSeq: number;
|
|
/** Monotonically increasing counter used to mint fallback tool-call ids for
|
|
* malformed (non-string/empty) `toolCallId`s — never reset while retained
|
|
* tools remain, so fallback ids stay unique across the MAX_TOOLS cap
|
|
* boundary. */
|
|
toolSeq: number;
|
|
/** 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 {
|
|
sendMessage: (input: { content: string; provider?: string; modelId?: string }) => void;
|
|
abort: () => void;
|
|
setThinking: (level: string) => void;
|
|
executeCommand: (input: { command: string; args?: string }) => void;
|
|
approveCommand: (input: { command: string; args?: string }) => void;
|
|
runApprovedCommand: () => void;
|
|
}
|
|
|
|
export interface ChatConnectionValue {
|
|
state: ChatConnectionState;
|
|
actions: ChatConnectionActions;
|
|
}
|
|
|
|
const initialState: ChatConnectionState = {
|
|
conversationId: null,
|
|
pendingSend: false,
|
|
sending: false,
|
|
ack: null,
|
|
streaming: false,
|
|
text: '',
|
|
textDroppedChars: 0,
|
|
thinking: '',
|
|
thinkingDroppedChars: 0,
|
|
tools: [],
|
|
usage: null,
|
|
sessionInfo: null,
|
|
manifest: null,
|
|
commandResults: [],
|
|
approval: null,
|
|
pendingApproval: null,
|
|
approvalRequestPending: false,
|
|
systemReload: null,
|
|
error: null,
|
|
messages: [],
|
|
messageSeq: 0,
|
|
toolSeq: 0,
|
|
turnPhase: 'settled',
|
|
};
|
|
|
|
type Action =
|
|
| { type: 'server/message:ack'; payload: MessageAckPayload }
|
|
| { type: 'server/agent:start'; payload: AgentStartPayload }
|
|
| { type: 'server/agent:text'; payload: AgentTextPayload }
|
|
| { type: 'server/agent:thinking'; payload: AgentThinkingPayload }
|
|
| { type: 'server/agent:tool:start'; payload: ToolStartPayload }
|
|
| { type: 'server/agent:tool:end'; payload: ToolEndPayload }
|
|
| { type: 'server/agent:end'; payload: AgentEndPayload }
|
|
| { type: 'server/session:info'; payload: SessionInfoPayload }
|
|
| { type: 'server/commands:manifest'; payload: CommandManifestPayload }
|
|
| { type: 'server/command:result'; payload: SlashCommandResultPayload }
|
|
| { type: 'server/command:approval'; payload: SlashCommandApprovalResultPayload }
|
|
| { type: 'server/system:reload'; payload: SystemReloadPayload }
|
|
| { type: 'server/error'; payload: ErrorPayload }
|
|
| { type: 'local/send'; content: string }
|
|
| { type: 'local/approve-request'; command: string; args?: string }
|
|
| { type: 'local/consume-approval' }
|
|
| { type: 'local/approval-saturated' }
|
|
| { type: 'local/disconnect' };
|
|
|
|
/**
|
|
* Resolves whether a scoped server event (one carrying a conversationId) belongs to
|
|
* the active conversation. The gateway does not guarantee message:ack is the first
|
|
* event for a new conversation (session:info, and error on auth/session-creation
|
|
* failure, can both arrive first) — so while a message is pending and no conversation
|
|
* is active yet, the first scoped event names the conversation instead of being
|
|
* dropped. Once a conversation is active, only its own events pass.
|
|
*
|
|
* `rawConversationId` is `unknown`, not `string` — every payload is runtime-untrusted
|
|
* regardless of its compile-time contract, and this is the single place that may
|
|
* establish `state.conversationId`. A malformed value (non-string/empty) is rejected
|
|
* via the central `asConversationId` guard and never adopted: the event is treated as
|
|
* inactive/dropped, leaving state (including `pendingSend`) untouched so a later
|
|
* genuinely valid event can still establish or recover the turn.
|
|
*/
|
|
function resolveScopedConversation(
|
|
state: ChatConnectionState,
|
|
rawConversationId: unknown,
|
|
): { active: true; state: ChatConnectionState } | { active: false; state: null } {
|
|
const conversationId = asConversationId(rawConversationId);
|
|
if (conversationId === null) {
|
|
return { active: false, state: null };
|
|
}
|
|
if (state.conversationId === conversationId) {
|
|
return { active: true, state };
|
|
}
|
|
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. 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 },
|
|
};
|
|
}
|
|
return { active: false, state: null };
|
|
}
|
|
|
|
type ServerAction = Extract<Action, { type: `server/${string}` }>;
|
|
|
|
function isServerAction(action: Action): action is ServerAction {
|
|
return action.type.startsWith('server/');
|
|
}
|
|
|
|
function reduce(state: ChatConnectionState, action: Action): ChatConnectionState {
|
|
// A malformed packet (the whole payload is null/undefined/a primitive, not
|
|
// an object) is ignored outright rather than crashing the reducer on the
|
|
// first field dereference in the case below — every server/* action shape
|
|
// declares a `payload` field, so this guard covers all of them uniformly.
|
|
if (isServerAction(action) && !isRecord(action.payload)) {
|
|
return state;
|
|
}
|
|
|
|
switch (action.type) {
|
|
case 'server/message:ack': {
|
|
const { payload } = action;
|
|
if (state.conversationId === null) {
|
|
const conversationId = asConversationId(payload.conversationId);
|
|
// A malformed first frame (non-string/empty conversationId) is ignored
|
|
// outright rather than adopted — adopting it would permanently
|
|
// desynchronize every later scoped event's strict-equality check
|
|
// against a value that can never again match. Leaving state untouched
|
|
// (conversationId stays null) lets a later genuinely valid ack/start
|
|
// still establish the turn.
|
|
if (conversationId === null) return state;
|
|
return {
|
|
...state,
|
|
conversationId,
|
|
pendingSend: false,
|
|
ack: sanitizeAck(payload, conversationId),
|
|
// 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),
|
|
};
|
|
}
|
|
|
|
case 'server/agent:start': {
|
|
const resolved = resolveScopedConversation(state, action.payload.conversationId);
|
|
if (!resolved.active) return state;
|
|
return {
|
|
...resolved.state,
|
|
streaming: true,
|
|
text: '',
|
|
thinking: '',
|
|
textDroppedChars: 0,
|
|
thinkingDroppedChars: 0,
|
|
tools: [],
|
|
error: null,
|
|
// 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',
|
|
};
|
|
}
|
|
|
|
case 'server/agent:text': {
|
|
const resolved = resolveScopedConversation(state, action.payload.conversationId);
|
|
if (!resolved.active) return state;
|
|
const capped = capAppendStream(
|
|
resolved.state.text,
|
|
resolved.state.textDroppedChars,
|
|
asString(action.payload.text),
|
|
MAX_STREAM_CHARS,
|
|
);
|
|
return { ...resolved.state, text: capped.displayed, textDroppedChars: capped.dropped };
|
|
}
|
|
|
|
case 'server/agent:thinking': {
|
|
const resolved = resolveScopedConversation(state, action.payload.conversationId);
|
|
if (!resolved.active) return state;
|
|
const capped = capAppendStream(
|
|
resolved.state.thinking,
|
|
resolved.state.thinkingDroppedChars,
|
|
asString(action.payload.text),
|
|
MAX_STREAM_CHARS,
|
|
);
|
|
return {
|
|
...resolved.state,
|
|
thinking: capped.displayed,
|
|
thinkingDroppedChars: capped.dropped,
|
|
};
|
|
}
|
|
|
|
case 'server/agent:tool:start': {
|
|
const { payload } = action;
|
|
const resolved = resolveScopedConversation(state, payload.conversationId);
|
|
if (!resolved.active) return state;
|
|
const valid = isValidToolCallId(payload.toolCallId);
|
|
const toolSeq = resolved.state.toolSeq;
|
|
const tool: ToolCallState = {
|
|
// Each malformed toolCallId gets its own fresh fallback id (never the
|
|
// fixed 'unknown-tool-call' constant) so distinct malformed starts
|
|
// never collide with each other.
|
|
toolCallId: valid ? payload.toolCallId : nextToolFallbackId(toolSeq),
|
|
toolName: asString(payload.toolName, 'Unknown tool'),
|
|
status: 'running',
|
|
};
|
|
return {
|
|
...resolved.state,
|
|
tools: capPush(resolved.state.tools, tool, MAX_TOOLS),
|
|
toolSeq: valid ? toolSeq : toolSeq + 1,
|
|
};
|
|
}
|
|
|
|
case 'server/agent:tool:end': {
|
|
const { payload } = action;
|
|
const resolved = resolveScopedConversation(state, payload.conversationId);
|
|
if (!resolved.active) return state;
|
|
const valid = isValidToolCallId(payload.toolCallId);
|
|
const toolSeq = resolved.state.toolSeq;
|
|
const toolCallId = valid ? payload.toolCallId : nextToolFallbackId(toolSeq);
|
|
const toolName = asString(payload.toolName, 'Unknown tool');
|
|
const nextToolSeq = valid ? toolSeq : toolSeq + 1;
|
|
// A fresh fallback id (malformed toolCallId) can never match an
|
|
// existing entry, so this only ever finds a genuine prior tool:start.
|
|
const matchIndex = resolved.state.tools.findIndex((tool) => tool.toolCallId === toolCallId);
|
|
if (matchIndex === -1) {
|
|
// A tool:end for an ID we never saw a tool:start for is a protocol
|
|
// anomaly, not a no-op — surface it visibly instead of silently
|
|
// dropping it.
|
|
const anomaly: ToolCallState = { toolCallId, toolName, status: 'anomaly' };
|
|
return {
|
|
...resolved.state,
|
|
tools: capPush(resolved.state.tools, anomaly, MAX_TOOLS),
|
|
toolSeq: nextToolSeq,
|
|
};
|
|
}
|
|
// Update at most the first matching entry — if `toolCallId` is
|
|
// (unexpectedly) shared by more than one retained tool, updating every
|
|
// match would cross-contaminate unrelated statuses (and could silently
|
|
// overwrite a distinct entry's 'anomaly' status).
|
|
return {
|
|
...resolved.state,
|
|
tools: resolved.state.tools.map((tool, index) =>
|
|
index === matchIndex ? { ...tool, status: payload.isError ? 'error' : 'success' } : tool,
|
|
),
|
|
toolSeq: nextToolSeq,
|
|
};
|
|
}
|
|
|
|
case 'server/agent:end': {
|
|
const { payload } = action;
|
|
const resolved = resolveScopedConversation(state, payload.conversationId);
|
|
if (!resolved.active) {
|
|
if (isUnrecoverableStartupFailure(state)) {
|
|
return {
|
|
...state,
|
|
pendingSend: false,
|
|
sending: false,
|
|
streaming: false,
|
|
approvalRequestPending: false,
|
|
turnPhase: 'settled',
|
|
error: CONVERSATION_START_FAILURE,
|
|
};
|
|
}
|
|
return state;
|
|
}
|
|
const next = resolved.state;
|
|
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;
|
|
}
|
|
// Reaching here means turnPhase is 'active' — but AgentEndPayload
|
|
// carries only a conversationId, no turn/message identity. In-order
|
|
// delivery proves the boundary enforced above (a same-conversation end
|
|
// arriving before this turn's own agent:start is provably stale); once
|
|
// 'active', a genuine end for this turn and a same-conversation
|
|
// stale/duplicate end are indistinguishable here. This residual is
|
|
// accepted — full correlation requires a wire turnId, deferred to P5.
|
|
const hasContent = next.text.length > 0 || next.thinking.length > 0;
|
|
const messages = hasContent
|
|
? capPush(
|
|
next.messages,
|
|
{
|
|
// Sourced from the reducer-owned `messageSeq` counter, not
|
|
// `messages.length` — the latter plateaus once MAX_MESSAGES is
|
|
// reached under a flood, producing duplicate ids/React keys.
|
|
id: `assistant-${payload.conversationId}-${next.messageSeq}`,
|
|
role: 'assistant' as const,
|
|
text: next.text,
|
|
thinking: next.thinking || undefined,
|
|
},
|
|
MAX_MESSAGES,
|
|
)
|
|
: next.messages;
|
|
return {
|
|
...next,
|
|
streaming: false,
|
|
sending: false,
|
|
turnPhase: 'settled',
|
|
text: '',
|
|
thinking: '',
|
|
textDroppedChars: 0,
|
|
thinkingDroppedChars: 0,
|
|
usage: payload.usage ?? next.usage,
|
|
messages,
|
|
messageSeq: hasContent ? next.messageSeq + 1 : next.messageSeq,
|
|
};
|
|
}
|
|
|
|
case 'server/session:info': {
|
|
const { payload } = action;
|
|
const resolved = resolveScopedConversation(state, payload.conversationId);
|
|
if (!resolved.active) return state;
|
|
return {
|
|
...resolved.state,
|
|
sessionInfo: {
|
|
...payload,
|
|
// A hostile/malfunctioning gateway can flood this list; cap it
|
|
// before it ever reaches state so a render site (e.g. the
|
|
// thinking-level <select>) can never be forced to lay out an
|
|
// unbounded number of options.
|
|
availableThinkingLevels: asStringArray(payload.availableThinkingLevels).slice(
|
|
0,
|
|
MAX_MANIFEST_ITEMS,
|
|
),
|
|
},
|
|
};
|
|
}
|
|
|
|
case 'server/commands:manifest': {
|
|
// `manifest` itself, and its `commands`/`skills` fields, are raw
|
|
// socket payload values — any of them can be null/non-object at
|
|
// runtime regardless of the compile-time contract.
|
|
const manifest: Record<string, unknown> = isRecord(action.payload.manifest)
|
|
? action.payload.manifest
|
|
: {};
|
|
return {
|
|
...state,
|
|
manifest: {
|
|
commands: capList<CommandDef>(manifest.commands, MAX_MANIFEST_ITEMS),
|
|
skills: capList<SkillCommandDef>(manifest.skills, MAX_MANIFEST_ITEMS),
|
|
version: asFiniteNumber(manifest.version, state.manifest?.version ?? 0),
|
|
},
|
|
};
|
|
}
|
|
|
|
case 'server/command:result': {
|
|
const { payload } = action;
|
|
const resolved = resolveScopedConversation(state, payload.conversationId);
|
|
if (!resolved.active) return state;
|
|
const next = resolved.state;
|
|
const conversationId = next.conversationId;
|
|
if (conversationId === null) return next;
|
|
return {
|
|
...next,
|
|
commandResults: capPush(
|
|
next.commandResults,
|
|
sanitizeCommandResult(payload, conversationId),
|
|
MAX_COMMAND_RESULTS,
|
|
),
|
|
};
|
|
}
|
|
|
|
case 'server/command:approval': {
|
|
const { payload } = action;
|
|
const resolved = resolveScopedConversation(state, payload.conversationId);
|
|
if (!resolved.active) return state;
|
|
const next = resolved.state;
|
|
const conversationId = next.conversationId;
|
|
if (
|
|
conversationId === null ||
|
|
!next.pendingApproval ||
|
|
typeof payload.command !== 'string' ||
|
|
payload.command !== next.pendingApproval.command
|
|
) {
|
|
// Stale or mismatched response for a request that is no longer (or never was)
|
|
// the one outstanding approval — do not let it replace active approval state.
|
|
return next;
|
|
}
|
|
return {
|
|
...next,
|
|
approval: sanitizeApproval(payload, conversationId, payload.command),
|
|
approvalRequestPending: false,
|
|
};
|
|
}
|
|
|
|
case 'server/system:reload': {
|
|
const { payload } = action;
|
|
// `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: {
|
|
commands,
|
|
skills,
|
|
providers,
|
|
message: asString(payload.message, 'Commands reloaded.'),
|
|
},
|
|
manifest: {
|
|
commands,
|
|
skills,
|
|
version: state.manifest?.version ?? 0,
|
|
},
|
|
};
|
|
}
|
|
|
|
case 'server/error': {
|
|
const { payload } = action;
|
|
const resolved = resolveScopedConversation(state, payload.conversationId);
|
|
if (!resolved.active) {
|
|
if (isUnrecoverableStartupFailure(state)) {
|
|
return {
|
|
...state,
|
|
pendingSend: false,
|
|
sending: false,
|
|
streaming: false,
|
|
approvalRequestPending: false,
|
|
turnPhase: 'settled',
|
|
error: CONVERSATION_START_FAILURE,
|
|
};
|
|
}
|
|
return state;
|
|
}
|
|
const next = resolved.state;
|
|
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.') };
|
|
}
|
|
// 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.
|
|
//
|
|
// Same wire limitation as agent:end above: ErrorPayload also carries
|
|
// only a conversationId, no turn identity. A pre-ack ('pending') error
|
|
// is accepted as current by design, per the reasoning above. Once
|
|
// 'active', a genuine error for this turn and a same-conversation
|
|
// stale/duplicate error are equally indistinguishable — accepted
|
|
// residual; full correlation needs a wire turnId, deferred to P5.
|
|
return {
|
|
...next,
|
|
error: asString(payload.error, 'An error occurred.'),
|
|
streaming: false,
|
|
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.
|
|
approvalRequestPending: false,
|
|
};
|
|
}
|
|
|
|
case 'local/send': {
|
|
const message: ChatTranscriptMessage = {
|
|
// Sourced from the reducer-owned `messageSeq` counter — see the
|
|
// `server/agent:end` case for why `messages.length`/`Date.now()` are
|
|
// not collision-safe once MAX_MESSAGES is reached under a flood.
|
|
id: `user-${state.messageSeq}`,
|
|
role: 'user',
|
|
text: action.content,
|
|
};
|
|
return {
|
|
...state,
|
|
messages: capPush(state.messages, message, MAX_MESSAGES),
|
|
messageSeq: state.messageSeq + 1,
|
|
error: null,
|
|
sending: true,
|
|
// 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,
|
|
};
|
|
}
|
|
|
|
case 'local/disconnect': {
|
|
// A transient socket disconnect must not leave the UI stuck waiting on
|
|
// a turn/approval/send that will never resolve on this connection.
|
|
return {
|
|
...state,
|
|
streaming: false,
|
|
sending: false,
|
|
pendingSend: false,
|
|
approvalRequestPending: false,
|
|
turnPhase: 'settled',
|
|
};
|
|
}
|
|
|
|
case 'local/approve-request': {
|
|
// Only one approval request may be outstanding at a time; a second request
|
|
// before the first resolves is ignored so it cannot overwrite the original
|
|
// command+args pair with a mismatched one.
|
|
if (state.approvalRequestPending) return state;
|
|
return {
|
|
...state,
|
|
pendingApproval: { command: action.command, args: action.args },
|
|
approvalRequestPending: true,
|
|
approval: null,
|
|
};
|
|
}
|
|
|
|
case 'local/consume-approval': {
|
|
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;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Owns the `/chat` socket for the lifetime of the consuming page: subscribes
|
|
* once, narrows the socket to the typed contract, and tears the socket down
|
|
* on cleanup so the next visit gets a fresh authenticated connection.
|
|
*/
|
|
export function useChatConnection(): ChatConnectionValue {
|
|
const [state, dispatch] = useReducer(reduce, initialState);
|
|
// Reducer state updates are batched/async; a ref lets a double-click on "Run
|
|
// approved command" be rejected synchronously, before React ever re-renders.
|
|
const executedApprovalIds = useRef<Set<string>>(new Set());
|
|
// Synchronous send lock: `state.sending` (reducer) drives the reactive UI
|
|
// disabled state, but reducer updates are batched/async, so two sendMessage
|
|
// calls issued in the same tick would both read the same stale `state`. This
|
|
// ref is the source of truth the guard checks against.
|
|
//
|
|
// It is intentionally never cleared directly by an event handler — a stale
|
|
// agent:end/error for a conversation other than the active one must not be
|
|
// able to re-arm the lock for a turn that is still genuinely in flight.
|
|
// Instead it is synchronized from `state.sending` below, which the reducer
|
|
// only flips on an *accepted* (conversation-scoped, or unconditional
|
|
// disconnect) transition — resolveScopedConversation already rejects a
|
|
// foreign-conversation agent:end/error by returning the unchanged state, so
|
|
// `state.sending` (and therefore this ref) cannot be released by one.
|
|
const sendLockRef = useRef(false);
|
|
useEffect(() => {
|
|
sendLockRef.current = state.sending;
|
|
}, [state.sending]);
|
|
|
|
// Same design as sendLockRef, mirroring `state.approvalRequestPending`:
|
|
// closes the same-tick double-dispatch race for approveCommand, and is
|
|
// synchronized from (not cleared by) the reducer's own accepted
|
|
// command:approval/error/disconnect transitions.
|
|
const approveLockRef = useRef(false);
|
|
useEffect(() => {
|
|
approveLockRef.current = state.approvalRequestPending;
|
|
}, [state.approvalRequestPending]);
|
|
|
|
useEffect(() => {
|
|
const socket = getSocket();
|
|
|
|
const onMessageAck = (payload: MessageAckPayload): void =>
|
|
dispatch({ type: 'server/message:ack', payload });
|
|
const onAgentStart = (payload: AgentStartPayload): void =>
|
|
dispatch({ type: 'server/agent:start', payload });
|
|
const onAgentText = (payload: AgentTextPayload): void =>
|
|
dispatch({ type: 'server/agent:text', payload });
|
|
const onAgentThinking = (payload: AgentThinkingPayload): void =>
|
|
dispatch({ type: 'server/agent:thinking', payload });
|
|
const onToolStart = (payload: ToolStartPayload): void =>
|
|
dispatch({ type: 'server/agent:tool:start', payload });
|
|
const onToolEnd = (payload: ToolEndPayload): void =>
|
|
dispatch({ type: 'server/agent:tool:end', payload });
|
|
const onAgentEnd = (payload: AgentEndPayload): void => {
|
|
dispatch({ type: 'server/agent:end', payload });
|
|
};
|
|
const onSessionInfo = (payload: SessionInfoPayload): void =>
|
|
dispatch({ type: 'server/session:info', payload });
|
|
const onCommandsManifest = (payload: CommandManifestPayload): void =>
|
|
dispatch({ type: 'server/commands:manifest', payload });
|
|
const onCommandResult = (payload: SlashCommandResultPayload): void =>
|
|
dispatch({ type: 'server/command:result', payload });
|
|
const onCommandApproval = (payload: SlashCommandApprovalResultPayload): void =>
|
|
dispatch({ type: 'server/command:approval', payload });
|
|
const onSystemReload = (payload: SystemReloadPayload): void =>
|
|
dispatch({ type: 'server/system:reload', payload });
|
|
const onError = (payload: ErrorPayload): void => {
|
|
dispatch({ type: 'server/error', payload });
|
|
};
|
|
const onDisconnect = (): void => {
|
|
dispatch({ type: 'local/disconnect' });
|
|
};
|
|
|
|
socket.on('message:ack', onMessageAck);
|
|
socket.on('agent:start', onAgentStart);
|
|
socket.on('agent:text', onAgentText);
|
|
socket.on('agent:thinking', onAgentThinking);
|
|
socket.on('agent:tool:start', onToolStart);
|
|
socket.on('agent:tool:end', onToolEnd);
|
|
socket.on('agent:end', onAgentEnd);
|
|
socket.on('session:info', onSessionInfo);
|
|
socket.on('commands:manifest', onCommandsManifest);
|
|
socket.on('command:result', onCommandResult);
|
|
socket.on('command:approval', onCommandApproval);
|
|
socket.on('system:reload', onSystemReload);
|
|
socket.on('error', onError);
|
|
socket.on('disconnect', onDisconnect);
|
|
|
|
if (!socket.connected) {
|
|
socket.connect();
|
|
}
|
|
|
|
return () => {
|
|
socket.off('message:ack', onMessageAck);
|
|
socket.off('agent:start', onAgentStart);
|
|
socket.off('agent:text', onAgentText);
|
|
socket.off('agent:thinking', onAgentThinking);
|
|
socket.off('agent:tool:start', onToolStart);
|
|
socket.off('agent:tool:end', onToolEnd);
|
|
socket.off('agent:end', onAgentEnd);
|
|
socket.off('session:info', onSessionInfo);
|
|
socket.off('commands:manifest', onCommandsManifest);
|
|
socket.off('command:result', onCommandResult);
|
|
socket.off('command:approval', onCommandApproval);
|
|
socket.off('system:reload', onSystemReload);
|
|
socket.off('error', onError);
|
|
socket.off('disconnect', onDisconnect);
|
|
destroySocket();
|
|
};
|
|
}, []);
|
|
|
|
const actions: ChatConnectionActions = {
|
|
sendMessage: ({ content, provider, modelId }) => {
|
|
if (sendLockRef.current || state.streaming || state.sending) return;
|
|
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,
|
|
});
|
|
},
|
|
|
|
abort: () => {
|
|
if (state.conversationId === null) return;
|
|
const socket = getSocket();
|
|
socket.emit('abort', { conversationId: state.conversationId });
|
|
},
|
|
|
|
setThinking: (level) => {
|
|
if (state.conversationId === null) return;
|
|
const socket = getSocket();
|
|
socket.emit('set:thinking', { conversationId: state.conversationId, level });
|
|
},
|
|
|
|
executeCommand: ({ command, args }) => {
|
|
if (state.conversationId === null) return;
|
|
const socket = getSocket();
|
|
socket.emit('command:execute', { conversationId: state.conversationId, command, args });
|
|
},
|
|
|
|
approveCommand: ({ command, args }) => {
|
|
if (state.conversationId === null) return;
|
|
if (approveLockRef.current || state.approvalRequestPending) return;
|
|
approveLockRef.current = true;
|
|
dispatch({ type: 'local/approve-request', command, args });
|
|
const socket = getSocket();
|
|
socket.emit('command:approve', { conversationId: state.conversationId, command, args });
|
|
},
|
|
|
|
runApprovedCommand: () => {
|
|
const { conversationId, approval, pendingApproval } = state;
|
|
if (conversationId === null) return;
|
|
// Defense-in-depth: the reducer already normalizes `success`/`approvalId`
|
|
// before storing `approval` (a matching command string alone is not
|
|
// proof of a genuine approval), but this action must never rely on that
|
|
// alone — a literal-`true` and non-empty-string check here too.
|
|
if (approval?.success !== true) return;
|
|
if (typeof approval.approvalId !== 'string' || approval.approvalId.length === 0) return;
|
|
if (!pendingApproval || pendingApproval.command !== approval.command) return;
|
|
if (executedApprovalIds.current.has(approval.approvalId)) {
|
|
dispatch({ type: 'local/consume-approval' });
|
|
return;
|
|
}
|
|
if (executedApprovalIds.current.size >= MAX_EXECUTED_APPROVAL_IDS) {
|
|
// Security tradeoff, chosen deliberately: this dedup set never
|
|
// evicts. Evicting the oldest entry to make room (the old behavior)
|
|
// would let a replay of that forgotten ID execute again once it
|
|
// scrolled out of the set — a false negative that lets a privileged
|
|
// command run twice. Once the set is full, every *unseen* approval
|
|
// is denied instead. The mounted hook remains fail-closed until
|
|
// 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, 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);
|
|
const socket = getSocket();
|
|
socket.emit('command:execute', {
|
|
conversationId,
|
|
// Sourced entirely from the frozen local pendingApproval, not the
|
|
// server-echoed approval payload — the equality guard above is
|
|
// defense-in-depth, not the source of truth for what gets executed.
|
|
command: pendingApproval.command,
|
|
args: pendingApproval.args,
|
|
approvalId: approval.approvalId,
|
|
});
|
|
dispatch({ type: 'local/consume-approval' });
|
|
},
|
|
};
|
|
|
|
return { state, actions };
|
|
}
|