mosaicstack/stack
1
0
Fork 1
Code Issues 290 Pull Requests 18 Actions Packages Projects Releases 15 Wiki Activity

fix(web): close P3 chat re-review findings

Browse Source
This commit is contained in:
shaggy (mosaic-dev box)
2026-08-10 01:58:27 -05:00
parent caebf9ef70
commit 48bb19310d
10 changed files with 774 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/web/src/spa/chat/use-chat-connection.ts
+236 -31
View File
@@ -8,7 +8,13 @@ import {
MAX_STREAM_CHARS,
MAX_TOOLS,
} from './limits';
import { asConversationId, asFiniteNumber, asString, isRecord } from './runtime-guards';
import {
asConversationId,
asFiniteNumber,
asString,
asStringArray,
isRecord,
} from './runtime-guards';
import type {
AgentEndPayload,
AgentStartPayload,
@@ -35,11 +41,64 @@ export interface ToolCallState {
status: 'running' | 'success' | 'error' | 'anomaly';
}
/** Appends `addition` to `existing`, keeping at most `max` characters by
* dropping the oldest (leading) characters once the cap is exceeded. */
function capAppendString(existing: string, addition: string, max: number): string {
const next = existing + addition;
return next.length > max ? next.slice(next.length - max) : next;
/** 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. */
@@ -72,6 +131,17 @@ 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
@@ -166,7 +236,14 @@ export interface ChatConnectionState {
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;
@@ -189,6 +266,20 @@ 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;
}
export interface ChatConnectionActions {
@@ -212,7 +303,9 @@ const initialState: ChatConnectionState = {
ack: null,
streaming: false,
text: '',
textDroppedChars: 0,
thinking: '',
thinkingDroppedChars: 0,
tools: [],
usage: null,
sessionInfo: null,
@@ -226,6 +319,8 @@ const initialState: ChatConnectionState = {
messages: [],
messageSeq: 0,
toolSeq: 0,
turnToken: 0,
armedTurnToken: null,
};
type Action =
@@ -274,7 +369,14 @@ function resolveScopedConversation(
return { active: true, state };
}
if (state.conversationId === null && state.pendingSend) {
return { active: true, state: { ...state, conversationId, pendingSend: false } };
// 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.
return {
active: true,
state: { ...state, conversationId, pendingSend: false, armedTurnToken: state.turnToken },
};
}
return { active: false, state: null };
}
@@ -311,37 +413,61 @@ 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,
};
}
if (payload.conversationId !== state.conversationId) return state;
return { ...state, ack: sanitizeAck(payload, state.conversationId) };
return {
...state,
ack: sanitizeAck(payload, state.conversationId),
armedTurnToken: state.turnToken,
};
}
case 'server/agent:start': {
const resolved = resolveScopedConversation(state, action.payload.conversationId);
if (!resolved.active) return state;
return { ...resolved.state, streaming: true, text: '', thinking: '', tools: [], error: null };
return {
...resolved.state,
streaming: true,
text: '',
thinking: '',
textDroppedChars: 0,
thinkingDroppedChars: 0,
tools: [],
error: null,
armedTurnToken: resolved.state.turnToken,
};
}
case 'server/agent:text': {
const resolved = resolveScopedConversation(state, action.payload.conversationId);
if (!resolved.active) return state;
return {
...resolved.state,
text: capAppendString(resolved.state.text, asString(action.payload.text), MAX_STREAM_CHARS),
};
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: capAppendString(
resolved.state.thinking,
asString(action.payload.text),
MAX_STREAM_CHARS,
),
thinking: capped.displayed,
thinkingDroppedChars: capped.dropped,
};
}
@@ -413,12 +539,20 @@ function reduce(state: ChatConnectionState, action: Action): ChatConnectionState
sending: false,
streaming: false,
approvalRequestPending: false,
armedTurnToken: null,
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.
return next;
}
const hasContent = next.text.length > 0 || next.thinking.length > 0;
const messages = hasContent
? capPush(
@@ -435,12 +569,23 @@ 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: false,
sending: canRelease ? false : next.sending,
armedTurnToken: canRelease ? null : next.armedTurnToken,
text: '',
thinking: '',
textDroppedChars: 0,
thinkingDroppedChars: 0,
usage: payload.usage ?? next.usage,
messages,
messageSeq: hasContent ? next.messageSeq + 1 : next.messageSeq,
@@ -448,9 +593,23 @@ function reduce(state: ChatConnectionState, action: Action): ChatConnectionState
}
case 'server/session:info': {
const resolved = resolveScopedConversation(state, action.payload.conversationId);
const { payload } = action;
const resolved = resolveScopedConversation(state, payload.conversationId);
if (!resolved.active) return state;
return { ...resolved.state, sessionInfo: action.payload };
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': {
@@ -512,12 +671,24 @@ 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.
const commands = capList<CommandDef>(payload.commands, MAX_MANIFEST_ITEMS);
const skills = capList<SkillCommandDef>(payload.skills, MAX_MANIFEST_ITEMS);
return {
...state,
systemReload: { ...payload, message: asString(payload.message, 'Commands reloaded.') },
systemReload: {
...payload,
commands,
skills,
message: asString(payload.message, 'Commands reloaded.'),
},
manifest: {
commands: capList<CommandDef>(payload.commands, MAX_MANIFEST_ITEMS),
skills: capList<SkillCommandDef>(payload.skills, MAX_MANIFEST_ITEMS),
commands,
skills,
version: state.manifest?.version ?? 0,
},
};
@@ -534,16 +705,31 @@ function reduce(state: ChatConnectionState, action: Action): ChatConnectionState
sending: false,
streaming: false,
approvalRequestPending: false,
armedTurnToken: null,
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;
}
// 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;
return {
...resolved.state,
...next,
error: asString(payload.error, 'An error occurred.'),
streaming: false,
sending: false,
sending: canRelease ? false : next.sending,
armedTurnToken: canRelease ? null : next.armedTurnToken,
// 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.
@@ -566,6 +752,12 @@ 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,
pendingSend: state.conversationId === null ? true : state.pendingSend,
};
}
@@ -579,6 +771,7 @@ function reduce(state: ChatConnectionState, action: Action): ChatConnectionState
sending: false,
pendingSend: false,
approvalRequestPending: false,
armedTurnToken: null,
};
}
@@ -766,12 +959,24 @@ export function useChatConnection(): ChatConnectionValue {
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)) return;
if (executedApprovalIds.current.has(approval.approvalId)) {
dispatch({ type: 'local/consume-approval' });
return;
}
if (executedApprovalIds.current.size >= MAX_EXECUTED_APPROVAL_IDS) {
// Bounded dedup set: evict the oldest entry (Sets iterate in
// insertion order) so this can never grow without limit.
const oldest = executedApprovalIds.current.values().next().value;
if (oldest !== undefined) executedApprovalIds.current.delete(oldest);
// 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 so the
// denial is visible/recoverable by remounting.
dispatch({ type: 'local/consume-approval' });
return;
}
executedApprovalIds.current.add(approval.approvalId);
const socket = getSocket();