mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
Peter Steinberger
096cb91175
* fix(ui): make indexed access explicit across the Control UI Burns down all 322 ui-lane noUncheckedIndexedAccess errors: untrusted markdown/diff/patch parsing gets miss-tolerant guards (renderer rules degrade to empty output instead of throwing mid-render), accumulators use canonical sparse initialization, DOM lookups stay null-guarded, and length-checked constructions carry named invariants. * feat(tooling): enforce noUncheckedIndexedAccess in the core and ui lanes Flips the flag on for tsconfig.core.json and tsconfig.ui.json (covering src, all packages including the two deferred ones, and ui) and retires the strict-ratchet lane wholesale: config, script, projects reference, check/CI wiring, changed-lane routing, and sync test. The assertion-ban oxlint override stays as an independent surface.
2531 lines
84 KiB
TypeScript
2531 lines
84 KiB
TypeScript
// Control UI module implements app chat behavior.
|
|
import { shouldForwardModelCommandToServer } from "../../../../src/auto-reply/commands-registry.shared.js";
|
|
import { isNonTerminalAgentRunStatus } from "../../../../src/shared/agent-run-status.js";
|
|
import {
|
|
GatewayRequestError,
|
|
type GatewayBrowserClient,
|
|
type GatewayHelloOk,
|
|
} from "../../api/gateway.ts";
|
|
import type { AgentsListResult } from "../../api/types.ts";
|
|
import { setLastActiveSessionKey } from "../../app/settings.ts";
|
|
import type {
|
|
ChatAttachment,
|
|
ChatQueueItem,
|
|
ChatQueueSkillWorkshopRevision,
|
|
} from "../../lib/chat/chat-types.ts";
|
|
import { parseSlashCommand } from "../../lib/chat/commands.ts";
|
|
import { extractSideQuestionDisplayText } from "../../lib/chat/side-question.ts";
|
|
import {
|
|
retirePendingChatSideQuestion,
|
|
type ChatSideResultPending,
|
|
} from "../../lib/chat/side-result.ts";
|
|
import { isSessionRunActive } from "../../lib/session-run-state.ts";
|
|
import {
|
|
scopedAgentIdForSession,
|
|
visibleSessionMatches,
|
|
type SessionCapability,
|
|
type SessionRefreshTarget,
|
|
} from "../../lib/sessions/index.ts";
|
|
import {
|
|
isUiGlobalSessionKey,
|
|
normalizeAgentId,
|
|
resolveUiSelectedSessionAgentId,
|
|
} from "../../lib/sessions/session-key.ts";
|
|
import { normalizeLowercaseStringOrEmpty } from "../../lib/string-coerce.ts";
|
|
import { generateUUID } from "../../lib/uuid.ts";
|
|
import {
|
|
discardChatAttachmentDataUrls,
|
|
getChatAttachmentDataUrl,
|
|
releaseChatAttachmentPayloads,
|
|
} from "./attachment-payload-store.ts";
|
|
import {
|
|
dispatchChatSlashCommand,
|
|
type ChatCommandHost,
|
|
type ChatCommandResetOptions,
|
|
shouldQueueLocalSlashCommand,
|
|
} from "./chat-commands.ts";
|
|
import { loadChatHistory, type ChatHistoryResult, type ChatState } from "./chat-history.ts";
|
|
import {
|
|
admitQueuedMessageForSession,
|
|
clearTransientQueuedMessageProjection,
|
|
enqueueChatMessage,
|
|
excludeComposerAttachments,
|
|
isVolatileQueuedMessage,
|
|
readQueuedMessageById,
|
|
removeQueuedMessageWithoutReleasing,
|
|
removeVisibleOrScopedQueuedMessageWithoutReleasing,
|
|
replacePendingQueuedMessageProjection,
|
|
syncChatQueueFromStoredOutbox,
|
|
setTransientQueuedMessageProjection,
|
|
updateQueuedMessage,
|
|
updateQueuedMessageForSession,
|
|
updateVolatileQueuedMessage,
|
|
} from "./chat-queue.ts";
|
|
import type {
|
|
ChatSendAck,
|
|
ChatSendAckServerTiming,
|
|
ChatSendTimingEntry,
|
|
} from "./chat-send-contract.ts";
|
|
import {
|
|
chatSendAckServerTimingEventFields,
|
|
recordChatSendTiming,
|
|
registerChatSendTiming,
|
|
schedulePendingSendPaintTiming,
|
|
updateChatSendAckTiming,
|
|
} from "./chat-send-timing.ts";
|
|
import { getPendingChatPickerPatch, refreshChatSessionListForTarget } from "./chat-session.ts";
|
|
import {
|
|
INTERRUPTED_SETTINGS_WAIT_ERROR,
|
|
listStoredChatOutboxes,
|
|
storedChatOutboxScopeKey,
|
|
type StoredChatOutbox,
|
|
type StoredChatOutboxScope,
|
|
} from "./composer-persistence.ts";
|
|
import { formatConnectError } from "./connect-error.ts";
|
|
import {
|
|
handleChatDraftChange,
|
|
handleChatInputHistoryKey,
|
|
navigateChatInputHistory,
|
|
recordNonTranscriptInputHistory,
|
|
resetChatInputHistoryNavigation,
|
|
type ChatInputHistoryKeyInput,
|
|
type ChatInputHistoryKeyResult,
|
|
type ChatInputHistoryState,
|
|
} from "./input-history.ts";
|
|
import { controlUiNowMs, roundedControlUiDurationMs } from "./performance.ts";
|
|
import type { RenderLifecycle } from "./render-lifecycle.ts";
|
|
import {
|
|
handleAbortChat,
|
|
isChatBusy,
|
|
isChatStopCommand,
|
|
reconcileChatRunLifecycle,
|
|
} from "./run-lifecycle.ts";
|
|
import { scheduleChatScroll, resetChatScroll } from "./scroll.ts";
|
|
import { resetToolStream } from "./tool-stream.ts";
|
|
import { buildUserChatMessageContentBlocks } from "./user-message-content.ts";
|
|
|
|
export type ChatHost = ChatInputHistoryState &
|
|
ChatCommandHost & {
|
|
sessions: SessionCapability;
|
|
client: GatewayBrowserClient | null;
|
|
chatStream: string | null;
|
|
connected: boolean;
|
|
connectionEpoch?: number;
|
|
chatAttachments: ChatAttachment[];
|
|
chatQueue: ChatQueueItem[];
|
|
chatQueueByScope?: Record<string, ChatQueueItem[]>;
|
|
chatRunId: string | null;
|
|
chatSending: boolean;
|
|
chatSendingScopeKey?: string | null;
|
|
lastError?: string | null;
|
|
chatError?: string | null;
|
|
hello: GatewayHelloOk | null;
|
|
renderLifecycle?: RenderLifecycle;
|
|
requestUpdate?: () => void;
|
|
refreshSessionsAfterChat: Map<string, SessionRefreshTarget>;
|
|
chatSubmitGuards?: Map<string, Promise<void>>;
|
|
chatSendTimingsByRun?: Map<string, ChatSendTimingEntry>;
|
|
eventLogBuffer?: unknown[];
|
|
assistantAgentId?: string | null;
|
|
agentsList?: ChatAgentsListSnapshot | null;
|
|
/** Selected message to reply to (right-click / keyboard shortcut). */
|
|
chatReplyTarget?: { messageId: string; text: string; senderLabel?: string | null } | null;
|
|
/** Placeholder for an in-flight /btw side question awaiting chat.side_result. */
|
|
chatSideResultPending?: ChatSideResultPending | null;
|
|
/** Retired/handled BTW run ids whose late events must not reach the transcript. */
|
|
chatSideResultTerminalRuns?: Set<string>;
|
|
/** Side-chat panel closed via X/Escape; a new question reopens it. */
|
|
chatSideChatHidden?: boolean;
|
|
};
|
|
|
|
type ChatAgentsListSnapshot = Partial<Omit<AgentsListResult, "agents">> & {
|
|
agents?: AgentsListResult["agents"];
|
|
};
|
|
|
|
function setChatError(
|
|
host: { lastError?: string | null; chatError?: string | null },
|
|
error: string | null,
|
|
) {
|
|
host.lastError = error;
|
|
host.chatError = error;
|
|
}
|
|
|
|
function sendResetSlashCommand(
|
|
host: ChatHost,
|
|
message: string,
|
|
opts: ChatCommandResetOptions,
|
|
): Promise<void> {
|
|
return sendChatMessageNow(host, message, {
|
|
refreshSessions: true,
|
|
previousDraft: opts.previousDraft,
|
|
restoreDraft: opts.restoreDraft,
|
|
routingSessionKey: host.sessionKey,
|
|
}).then(() => undefined);
|
|
}
|
|
|
|
type AcceptedChatSendAck = ChatSendAck & { status: "started" | "in_flight" | "ok" };
|
|
type TerminalFailureChatSendAck = ChatSendAck & { status: "timeout" | "error" };
|
|
|
|
function isAcceptedChatSendAck(ack: ChatSendAck | null): ack is AcceptedChatSendAck {
|
|
return ack != null && (ack.status === "ok" || isNonTerminalAgentRunStatus(ack.status));
|
|
}
|
|
|
|
function isTerminalFailureChatSendAck(ack: ChatSendAck | null): ack is TerminalFailureChatSendAck {
|
|
return ack?.status === "timeout" || ack?.status === "error";
|
|
}
|
|
|
|
function formatTerminalChatSendAckError(
|
|
ack: TerminalFailureChatSendAck,
|
|
context: "chat" | "detached" | "steer",
|
|
): string {
|
|
if (ack.status === "error") {
|
|
if (context === "steer") {
|
|
return "Steer failed before it reached the run; try again.";
|
|
}
|
|
return "Chat failed before the run started; try again.";
|
|
}
|
|
if (context === "detached") {
|
|
return "The active run ended before the detached message was accepted.";
|
|
}
|
|
if (context === "steer") {
|
|
return "The active run ended before the steer message was accepted.";
|
|
}
|
|
return "The run ended before the message was accepted.";
|
|
}
|
|
|
|
type ChatSendOptions = {
|
|
confirmReset?: boolean;
|
|
restoreDraft?: boolean;
|
|
skillWorkshopRevision?: ChatQueueSkillWorkshopRevision;
|
|
/** Side-chat follow-ups embed prior-turn context in the /btw command; the
|
|
* pending turn must display the user's typed question instead. */
|
|
sideQuestionDisplayText?: string;
|
|
/** Lets the side-chat panel restore its typed follow-up when the detached
|
|
* send is not accepted (the panel input is not a managed draft). */
|
|
onSideQuestionSendRejected?: () => void;
|
|
};
|
|
|
|
function dataUrlToBase64(dataUrl: string): { content: string; mimeType: string } | null {
|
|
const match = /^data:([^;]+);base64,(.+)$/.exec(dataUrl);
|
|
if (!match) {
|
|
return null;
|
|
}
|
|
const mimeType = match[1];
|
|
const content = match[2];
|
|
return mimeType && content ? { mimeType, content } : null;
|
|
}
|
|
|
|
function buildApiAttachments(attachments?: ChatAttachment[]) {
|
|
const hasAttachments = attachments && attachments.length > 0;
|
|
return hasAttachments
|
|
? attachments
|
|
.map((att) => {
|
|
const dataUrl = getChatAttachmentDataUrl(att);
|
|
const parsed = dataUrl ? dataUrlToBase64(dataUrl) : null;
|
|
if (!parsed) {
|
|
return null;
|
|
}
|
|
return {
|
|
type: parsed.mimeType.startsWith("image/") ? "image" : "file",
|
|
mimeType: parsed.mimeType,
|
|
fileName: att.fileName,
|
|
content: parsed.content,
|
|
};
|
|
})
|
|
.filter((a): a is NonNullable<typeof a> => a !== null)
|
|
: undefined;
|
|
}
|
|
|
|
function normalizeAckTimingValue(value: unknown): number | undefined {
|
|
return typeof value === "number" && Number.isFinite(value) && value >= 0 ? value : undefined;
|
|
}
|
|
|
|
export type {
|
|
ChatSendAck,
|
|
ChatSendAckServerTiming,
|
|
ChatSendAckStatus,
|
|
} from "./chat-send-contract.ts";
|
|
|
|
function normalizeChatSendAckServerTiming(value: unknown): ChatSendAckServerTiming | undefined {
|
|
if (!value || typeof value !== "object") {
|
|
return undefined;
|
|
}
|
|
const record = value as Record<string, unknown>;
|
|
const receivedToAckMs = normalizeAckTimingValue(record.receivedToAckMs);
|
|
const loadSessionMs = normalizeAckTimingValue(record.loadSessionMs);
|
|
const prepareAttachmentsMs = normalizeAckTimingValue(record.prepareAttachmentsMs);
|
|
const timing: ChatSendAckServerTiming = {
|
|
...(receivedToAckMs !== undefined ? { receivedToAckMs } : {}),
|
|
...(loadSessionMs !== undefined ? { loadSessionMs } : {}),
|
|
...(prepareAttachmentsMs !== undefined ? { prepareAttachmentsMs } : {}),
|
|
};
|
|
return Object.keys(timing).length > 0 ? timing : undefined;
|
|
}
|
|
|
|
function normalizeChatSendAck(payload: unknown, fallbackRunId: string): ChatSendAck {
|
|
if (!payload || typeof payload !== "object") {
|
|
return { runId: fallbackRunId, status: "started" };
|
|
}
|
|
const record = payload as Record<string, unknown>;
|
|
const runId =
|
|
typeof record.runId === "string" && record.runId.trim() ? record.runId.trim() : fallbackRunId;
|
|
const status = record.status;
|
|
const serverTiming = normalizeChatSendAckServerTiming(record.serverTiming);
|
|
return {
|
|
runId,
|
|
status:
|
|
status === "in_flight" || status === "ok" || status === "timeout" || status === "error"
|
|
? status
|
|
: "started",
|
|
...(serverTiming ? { serverTiming } : {}),
|
|
};
|
|
}
|
|
|
|
export async function requestChatSend(
|
|
state: ChatState,
|
|
params: {
|
|
message: string;
|
|
attachments?: ChatAttachment[];
|
|
runId: string;
|
|
sessionKey?: string;
|
|
agentId?: string;
|
|
},
|
|
): Promise<ChatSendAck> {
|
|
const routing = resolveChatSendRouting(state, params);
|
|
const controlUiReconnectResume = Boolean(
|
|
routing.sessionId && state.reconnectResumeSessionId === routing.sessionId,
|
|
);
|
|
const payload = await state.client!.request("chat.send", {
|
|
sessionKey: routing.sessionKey,
|
|
...(isUiGlobalSessionKey(routing.sessionKey) && routing.selectedAgentId
|
|
? { agentId: routing.selectedAgentId }
|
|
: {}),
|
|
...(routing.sessionId ? { sessionId: routing.sessionId } : {}),
|
|
...(controlUiReconnectResume ? { __controlUiReconnectResume: true } : {}),
|
|
message: params.message,
|
|
deliver: false,
|
|
idempotencyKey: params.runId,
|
|
attachments: buildApiAttachments(params.attachments),
|
|
});
|
|
if (controlUiReconnectResume) {
|
|
state.reconnectResumeSessionId = null;
|
|
}
|
|
return normalizeChatSendAck(payload, params.runId);
|
|
}
|
|
|
|
function resolveChatSendRouting(
|
|
state: ChatState,
|
|
params: {
|
|
sessionKey?: string;
|
|
agentId?: string;
|
|
},
|
|
): { selectedAgentId?: string; sessionId?: string; sessionKey: string } {
|
|
const sessionKey = params.sessionKey ?? state.sessionKey;
|
|
const selectedAgentId = params.agentId
|
|
? normalizeAgentId(params.agentId)
|
|
: resolveUiSelectedSessionAgentId(state);
|
|
const currentSessionId = state.currentSessionId;
|
|
const canReuseCurrentSessionId =
|
|
sessionKey === state.sessionKey &&
|
|
(!isUiGlobalSessionKey(sessionKey) ||
|
|
(selectedAgentId !== undefined &&
|
|
selectedAgentId === resolveUiSelectedSessionAgentId(state)));
|
|
const sessionId =
|
|
canReuseCurrentSessionId && typeof currentSessionId === "string" && currentSessionId.trim()
|
|
? currentSessionId.trim()
|
|
: undefined;
|
|
return {
|
|
sessionKey,
|
|
...(selectedAgentId ? { selectedAgentId } : {}),
|
|
...(sessionId ? { sessionId } : {}),
|
|
};
|
|
}
|
|
|
|
export async function requestSkillWorkshopRevisionChatSend(
|
|
state: ChatState,
|
|
params: {
|
|
proposalId: string;
|
|
instructions: string;
|
|
runId: string;
|
|
sessionKey?: string;
|
|
agentId?: string;
|
|
targetAgentId?: string;
|
|
},
|
|
): Promise<ChatSendAck> {
|
|
const routing = resolveChatSendRouting(state, {
|
|
sessionKey: params.sessionKey,
|
|
agentId: params.targetAgentId,
|
|
});
|
|
const payload = await state.client!.request("skills.proposals.requestRevision", {
|
|
...(params.agentId ? { agentId: normalizeAgentId(params.agentId) } : {}),
|
|
...(routing.selectedAgentId ? { targetAgentId: routing.selectedAgentId } : {}),
|
|
proposalId: params.proposalId,
|
|
instructions: params.instructions,
|
|
sessionKey: routing.sessionKey,
|
|
...(routing.sessionId ? { sessionId: routing.sessionId } : {}),
|
|
idempotencyKey: params.runId,
|
|
});
|
|
return normalizeChatSendAck(payload, params.runId);
|
|
}
|
|
|
|
function appendUserChatMessage(
|
|
state: ChatState,
|
|
message: string,
|
|
attachments?: ChatAttachment[],
|
|
timestamp = Date.now(),
|
|
) {
|
|
const entry = {
|
|
role: "user" as const,
|
|
content: buildUserChatMessageContentBlocks(message, attachments),
|
|
timestamp,
|
|
};
|
|
state.chatMessages = [...state.chatMessages, entry];
|
|
return entry;
|
|
}
|
|
|
|
async function sendChatMessageWithGeneratedRunId(
|
|
state: ChatState,
|
|
message: string,
|
|
attachments?: ChatAttachment[],
|
|
canApplyError: () => boolean = () => true,
|
|
runIdOverride?: string,
|
|
): Promise<ChatSendAck | null> {
|
|
if (!state.client || !state.connected) {
|
|
return null;
|
|
}
|
|
const msg = message.trim();
|
|
const hasAttachments = attachments && attachments.length > 0;
|
|
if (!msg && !hasAttachments) {
|
|
return null;
|
|
}
|
|
if (canApplyError()) {
|
|
setChatError(state, null);
|
|
}
|
|
const runId = runIdOverride ?? generateUUID();
|
|
try {
|
|
return await requestChatSend(state, { message: msg, attachments, runId });
|
|
} catch (err) {
|
|
if (canApplyError()) {
|
|
setChatError(state, formatConnectError(err));
|
|
}
|
|
return null;
|
|
}
|
|
}
|
|
|
|
export async function sendDetachedChatMessage(
|
|
state: ChatState,
|
|
message: string,
|
|
attachments?: ChatAttachment[],
|
|
runId?: string,
|
|
): Promise<ChatSendAck | null> {
|
|
return sendChatMessageWithGeneratedRunId(state, message, attachments, () => true, runId);
|
|
}
|
|
|
|
export async function sendSteerChatMessage(
|
|
state: ChatState,
|
|
message: string,
|
|
attachments?: ChatAttachment[],
|
|
): Promise<ChatSendAck | null> {
|
|
return sendChatMessageWithGeneratedRunId(state, message, attachments);
|
|
}
|
|
|
|
export {
|
|
handleChatDraftChange,
|
|
handleChatInputHistoryKey,
|
|
navigateChatInputHistory,
|
|
resetChatInputHistoryNavigation,
|
|
};
|
|
export type { ChatInputHistoryKeyInput, ChatInputHistoryKeyResult };
|
|
|
|
function isChatResetCommand(text: string) {
|
|
const parsed = parseSlashCommand(text);
|
|
if (!parsed || (parsed.command.key !== "new" && parsed.command.key !== "reset")) {
|
|
return false;
|
|
}
|
|
if (parsed.command.key === "new") {
|
|
return true;
|
|
}
|
|
if (/^soft(?:\s|$)/.test(normalizeLowercaseStringOrEmpty(parsed.args))) {
|
|
return false;
|
|
}
|
|
return true;
|
|
}
|
|
|
|
function confirmChatResetCommand(text: string) {
|
|
if (!isChatResetCommand(text)) {
|
|
return true;
|
|
}
|
|
if (typeof globalThis.confirm !== "function") {
|
|
return false;
|
|
}
|
|
return globalThis.confirm("Start a new session? This will reset the current chat.");
|
|
}
|
|
|
|
function isBtwCommand(text: string) {
|
|
return /^\/(?:btw|side)(?::|\s|$)/i.test(text.trim());
|
|
}
|
|
|
|
function enqueuePendingSendMessage(
|
|
host: ChatHost,
|
|
text: string,
|
|
attachments?: ChatAttachment[],
|
|
refreshSessions?: boolean,
|
|
submittedAtMs = controlUiNowMs(),
|
|
sendState?: ChatQueueItem["sendState"],
|
|
skillWorkshopRevision?: ChatQueueSkillWorkshopRevision,
|
|
): ChatQueueItem | null {
|
|
const trimmed = text.trim();
|
|
const hasAttachments = Boolean(attachments && attachments.length > 0);
|
|
if (!trimmed && !hasAttachments) {
|
|
return null;
|
|
}
|
|
const pending: ChatQueueItem = {
|
|
id: generateUUID(),
|
|
text: trimmed,
|
|
createdAt: Date.now(),
|
|
attachments: hasAttachments ? attachments : undefined,
|
|
refreshSessions,
|
|
sendAttempts: 0,
|
|
sendRunId: generateUUID(),
|
|
sendState,
|
|
sendSubmittedAtMs: submittedAtMs,
|
|
sessionKey: host.sessionKey,
|
|
agentId: scopedAgentIdForSession(host, host.sessionKey),
|
|
...(skillWorkshopRevision ? { skillWorkshopRevision } : {}),
|
|
};
|
|
host.chatQueue = [...host.chatQueue, pending];
|
|
recordChatSendTiming(host, pending, "pending-visible", submittedAtMs);
|
|
if (sendState === "waiting-model" || sendState === "waiting-reconnect") {
|
|
recordChatSendTiming(host, pending, sendState, submittedAtMs);
|
|
}
|
|
schedulePendingSendPaintTiming(host, pending, submittedAtMs);
|
|
scheduleChatScroll(host as unknown as Parameters<typeof scheduleChatScroll>[0], true, false, {
|
|
source: "manual",
|
|
});
|
|
return pending;
|
|
}
|
|
|
|
function isRecoverableChatSendError(err: unknown, formattedError: string): boolean {
|
|
if (err instanceof GatewayRequestError) {
|
|
return err.retryable;
|
|
}
|
|
return /gateway (?:not connected|closed)|websocket|disconnected/i.test(formattedError);
|
|
}
|
|
|
|
function isProvablyPreTransportChatSendError(err: unknown): boolean {
|
|
return (
|
|
err instanceof Error &&
|
|
!(err instanceof GatewayRequestError) &&
|
|
err.message === "gateway not connected"
|
|
);
|
|
}
|
|
|
|
function restoreComposerAfterFailedSend(
|
|
host: ChatHost,
|
|
opts: {
|
|
previousAttachments?: ChatAttachment[];
|
|
previousDraft?: string;
|
|
},
|
|
) {
|
|
if (opts.previousDraft != null && !host.chatMessage.trim()) {
|
|
host.chatMessage = opts.previousDraft;
|
|
}
|
|
if (opts.previousAttachments?.length && host.chatAttachments.length === 0) {
|
|
host.chatAttachments = opts.previousAttachments;
|
|
}
|
|
}
|
|
|
|
type PendingComposerSnapshot = {
|
|
previousAttachments?: ChatAttachment[];
|
|
previousDraft?: string;
|
|
};
|
|
|
|
function pendingComposerRestorePlan(host: ChatHost, snapshot: PendingComposerSnapshot) {
|
|
const willRestoreDraft = snapshot.previousDraft != null && !host.chatMessage.trim();
|
|
const willRestoreAttachments = Boolean(
|
|
snapshot.previousAttachments?.length &&
|
|
host.chatAttachments.length === 0 &&
|
|
(willRestoreDraft || !host.chatMessage.trim()),
|
|
);
|
|
return {
|
|
complete:
|
|
(!snapshot.previousDraft?.trim() || willRestoreDraft) &&
|
|
(!snapshot.previousAttachments?.length || willRestoreAttachments),
|
|
willRestoreAttachments,
|
|
willRestoreDraft,
|
|
};
|
|
}
|
|
|
|
function cancelPendingSendBeforeRequest(
|
|
host: ChatHost,
|
|
queued: ChatQueueItem,
|
|
opts: PendingComposerSnapshot & {
|
|
restoreComposer?: boolean;
|
|
},
|
|
) {
|
|
const removed = removeVisibleOrScopedQueuedMessageWithoutReleasing(
|
|
host,
|
|
queued.id,
|
|
queued.sessionKey,
|
|
);
|
|
const restoreComposer = opts.restoreComposer !== false && removed != null;
|
|
const restorePlan = pendingComposerRestorePlan(host, opts);
|
|
const willRestoreDraft = restoreComposer && restorePlan.willRestoreDraft;
|
|
const willRestoreAttachments = restoreComposer && restorePlan.willRestoreAttachments;
|
|
if (restoreComposer) {
|
|
if (willRestoreDraft) {
|
|
host.chatMessage = opts.previousDraft ?? "";
|
|
}
|
|
if (willRestoreAttachments) {
|
|
host.chatAttachments = opts.previousAttachments ?? [];
|
|
}
|
|
}
|
|
if (removed && !willRestoreAttachments) {
|
|
releaseChatAttachmentPayloads(excludeComposerAttachments(host, removed.attachments));
|
|
}
|
|
}
|
|
|
|
type QueuedChatSendResult = "sent" | "pending" | "failed";
|
|
type QueuedChatStorageMode = "durable" | "memory";
|
|
type QueuedChatSendOptions = {
|
|
previousAttachments?: ChatAttachment[];
|
|
previousDraft?: string;
|
|
routingSessionKey?: string;
|
|
storageMode?: QueuedChatStorageMode;
|
|
};
|
|
|
|
function reconnectSafeQueuedSendState(
|
|
host: Pick<ChatHost, "client" | "connected">,
|
|
): "waiting-idle" | "waiting-reconnect" {
|
|
return host.connected && host.client ? "waiting-idle" : "waiting-reconnect";
|
|
}
|
|
|
|
function updateQueuedSendItem(
|
|
host: ChatHost,
|
|
storageMode: QueuedChatStorageMode,
|
|
sessionKey: string,
|
|
id: string,
|
|
update: (item: ChatQueueItem) => ChatQueueItem,
|
|
): ChatQueueItem | null {
|
|
return storageMode === "memory"
|
|
? updateVolatileQueuedMessage(host, id, update)
|
|
: updateQueuedMessageForSession(host, sessionKey, id, update);
|
|
}
|
|
|
|
function canSendVolatileQueueItem(
|
|
host: ChatHost,
|
|
item: ChatQueueItem,
|
|
routingSessionKey = item.sessionKey ?? host.sessionKey,
|
|
): boolean {
|
|
return (
|
|
host.connected &&
|
|
Boolean(host.client) &&
|
|
!isChatBusy(host) &&
|
|
!getPendingChatPickerPatch(host, routingSessionKey, item.agentId) &&
|
|
host.sessionKey === routingSessionKey &&
|
|
visibleSessionMatches(host, routingSessionKey, item.agentId) &&
|
|
host.chatQueue[0]?.id === item.id
|
|
);
|
|
}
|
|
|
|
const OFFLINE_QUEUE_STORAGE_ERROR =
|
|
"Could not store this message for reconnect. Free browser storage or reconnect before sending.";
|
|
const UNCONFIRMED_CHAT_SEND_ERROR =
|
|
"Delivery could not be confirmed after reconnect. Check the conversation before retrying.";
|
|
const UNCONFIRMED_STEER_ERROR =
|
|
"Steer delivery could not be confirmed. Check the active run before retrying.";
|
|
const UNCERTAIN_CLEAR_SUCCESSOR_ERROR =
|
|
"A preceding /clear may have completed. Review the current conversation before retrying.";
|
|
const STORED_OUTBOX_RETRY_DEFAULT_MS = 500;
|
|
const STORED_OUTBOX_RETRY_MIN_MS = 100;
|
|
const STORED_OUTBOX_RETRY_MAX_MS = 30_000;
|
|
|
|
function beginScopedChatSending(host: ChatHost, scope: StoredChatOutboxScope): void {
|
|
if (!visibleSessionMatches(host, scope.sessionKey, scope.agentId)) {
|
|
return;
|
|
}
|
|
host.chatSendingScopeKey = storedChatOutboxScopeKey(scope);
|
|
host.chatSending = true;
|
|
}
|
|
|
|
function finishScopedChatSending(host: ChatHost, scope: StoredChatOutboxScope): void {
|
|
if (host.chatSendingScopeKey !== storedChatOutboxScopeKey(scope)) {
|
|
return;
|
|
}
|
|
host.chatSendingScopeKey = null;
|
|
host.chatSending = false;
|
|
}
|
|
|
|
function retryableGatewayDelayMs(err: unknown): number | null {
|
|
if (!(err instanceof GatewayRequestError) || !err.retryable) {
|
|
return null;
|
|
}
|
|
const requested = err.retryAfterMs ?? STORED_OUTBOX_RETRY_DEFAULT_MS;
|
|
return Math.min(Math.max(requested, STORED_OUTBOX_RETRY_MIN_MS), STORED_OUTBOX_RETRY_MAX_MS);
|
|
}
|
|
|
|
function ensureQueuedSendState(
|
|
host: ChatHost,
|
|
item: ChatQueueItem,
|
|
fallbackSessionKey = host.sessionKey,
|
|
storageMode: QueuedChatStorageMode = "durable",
|
|
): ChatQueueItem | null {
|
|
if (item.sendRunId && item.sendState) {
|
|
return item;
|
|
}
|
|
const sessionKey = item.sessionKey ?? fallbackSessionKey;
|
|
const agentId = item.agentId ?? scopedAgentIdForSession(host, sessionKey);
|
|
const prepared: ChatQueueItem = {
|
|
...item,
|
|
sendAttempts: item.sendAttempts ?? 0,
|
|
sendRunId: item.sendRunId ?? generateUUID(),
|
|
sendState: host.connected && host.client ? "sending" : "waiting-reconnect",
|
|
sessionKey,
|
|
agentId,
|
|
};
|
|
return updateQueuedSendItem(host, storageMode, sessionKey, item.id, () => prepared);
|
|
}
|
|
|
|
async function sendQueuedChatMessage(
|
|
host: ChatHost,
|
|
id: string,
|
|
opts?: QueuedChatSendOptions,
|
|
queuedSessionKey = host.sessionKey,
|
|
): Promise<QueuedChatSendResult> {
|
|
const storageMode = opts?.storageMode ?? "durable";
|
|
let queued = readQueuedMessageById(host, id);
|
|
if (!queued || queued.pendingRunId || queued.localCommandName) {
|
|
return "failed";
|
|
}
|
|
// Foreground sends keep the submitted route for picker admission. Durable
|
|
// storage may canonicalize an agent-main alias to its global outbox scope.
|
|
const queueSessionKey = queued.sessionKey ?? queuedSessionKey;
|
|
const pickerSessionKey = opts?.routingSessionKey ?? queueSessionKey;
|
|
const pendingSettings = getPendingChatPickerPatch(host, pickerSessionKey, queued.agentId);
|
|
if (pendingSettings) {
|
|
// Final admission gate for retries/reconnect replays and picker patches
|
|
// that start after the composer-level snapshot.
|
|
updateQueuedSendItem(host, storageMode, queueSessionKey, id, (item) => ({
|
|
...item,
|
|
sendError: undefined,
|
|
sendState: "waiting-model",
|
|
}));
|
|
host.requestUpdate?.();
|
|
if (
|
|
!(await waitForPendingChatSettings(host, pickerSessionKey, pendingSettings, queued.agentId))
|
|
) {
|
|
const canRestoreComposer =
|
|
opts?.previousDraft !== undefined &&
|
|
!host.chatMessage.trim() &&
|
|
host.chatAttachments.length === 0;
|
|
if (
|
|
canRestoreComposer &&
|
|
host.sessionKey === pickerSessionKey &&
|
|
visibleSessionMatches(host, pickerSessionKey, queued.agentId)
|
|
) {
|
|
cancelPendingSendBeforeRequest(host, queued, {
|
|
previousDraft: opts.previousDraft,
|
|
previousAttachments: opts.previousAttachments,
|
|
});
|
|
} else {
|
|
updateQueuedSendItem(host, storageMode, queueSessionKey, id, (item) => ({
|
|
...item,
|
|
sendError: INTERRUPTED_SETTINGS_WAIT_ERROR,
|
|
sendState: "failed",
|
|
}));
|
|
}
|
|
host.requestUpdate?.();
|
|
return "failed";
|
|
}
|
|
queued = readQueuedMessageById(host, id);
|
|
if (!queued) {
|
|
return "failed";
|
|
}
|
|
}
|
|
if (
|
|
opts?.routingSessionKey &&
|
|
(host.sessionKey !== opts.routingSessionKey ||
|
|
!visibleSessionMatches(host, opts.routingSessionKey, queued.agentId))
|
|
) {
|
|
const parked = updateQueuedSendItem(host, storageMode, queueSessionKey, id, (item) => ({
|
|
...item,
|
|
sendError: undefined,
|
|
sendState: host.connected && host.client ? "waiting-idle" : "waiting-reconnect",
|
|
}));
|
|
if (!parked) {
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
}
|
|
return "pending";
|
|
}
|
|
const queuedForRoute = opts?.routingSessionKey
|
|
? { ...queued, sessionKey: opts.routingSessionKey }
|
|
: queued;
|
|
const prepared = ensureQueuedSendState(host, queuedForRoute, queuedSessionKey, storageMode);
|
|
if (!prepared) {
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
return "pending";
|
|
}
|
|
const message = prepared.text.trim();
|
|
const attachments = prepared.attachments ?? [];
|
|
const hasAttachments = attachments.length > 0;
|
|
if (!message && !hasAttachments) {
|
|
removeQueuedMessageWithoutReleasing(host, id, prepared.sessionKey ?? host.sessionKey);
|
|
return "sent";
|
|
}
|
|
if (prepared.skillWorkshopRevision && hasAttachments) {
|
|
updateQueuedSendItem(host, storageMode, prepared.sessionKey ?? host.sessionKey, id, (item) => ({
|
|
...item,
|
|
sendError: "Skill Workshop revision requests do not support attachments.",
|
|
sendState: "failed",
|
|
}));
|
|
return "failed";
|
|
}
|
|
const sessionKey = prepared.sessionKey ?? host.sessionKey;
|
|
if (!host.connected || !host.client) {
|
|
const waiting = updateQueuedSendItem(host, storageMode, sessionKey, id, (item) => ({
|
|
...item,
|
|
sendState: "waiting-reconnect",
|
|
sendError: undefined,
|
|
}));
|
|
if (!waiting) {
|
|
const hasComposerSnapshot =
|
|
opts?.previousDraft !== undefined || opts?.previousAttachments !== undefined;
|
|
const canRestoreComposer =
|
|
hasComposerSnapshot && pendingComposerRestorePlan(host, opts ?? {}).complete;
|
|
if (canRestoreComposer) {
|
|
cancelPendingSendBeforeRequest(host, waiting ?? prepared, {
|
|
previousDraft: opts?.previousDraft,
|
|
previousAttachments: opts?.previousAttachments,
|
|
});
|
|
} else {
|
|
updateQueuedSendItem(host, storageMode, sessionKey, id, (item) => ({
|
|
...item,
|
|
sendError: OFFLINE_QUEUE_STORAGE_ERROR,
|
|
sendState: "failed",
|
|
}));
|
|
}
|
|
if (visibleSessionMatches(host, sessionKey, prepared.agentId)) {
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
}
|
|
return canRestoreComposer ? "failed" : "pending";
|
|
}
|
|
return "pending";
|
|
}
|
|
|
|
const runId = prepared.sendRunId ?? generateUUID();
|
|
const startedAt = Date.now();
|
|
const requestStartedAtMs = controlUiNowMs();
|
|
const sendingItem = updateQueuedSendItem(host, storageMode, sessionKey, id, (item) => ({
|
|
...item,
|
|
sendAttempts: (item.sendAttempts ?? 0) + 1,
|
|
sendError: undefined,
|
|
sendRunId: runId,
|
|
sendState: "sending",
|
|
sendRequestStartedAtMs: requestStartedAtMs,
|
|
sessionKey,
|
|
agentId: prepared.agentId,
|
|
}));
|
|
if (!sendingItem) {
|
|
if (visibleSessionMatches(host, sessionKey, prepared.agentId)) {
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
}
|
|
return "pending";
|
|
}
|
|
registerChatSendTiming(host, sendingItem, runId, requestStartedAtMs);
|
|
recordChatSendTiming(host, sendingItem, "request-start", sendingItem.sendSubmittedAtMs);
|
|
const sendingScope: StoredChatOutboxScope = {
|
|
sessionKey,
|
|
...(prepared.agentId ? { agentId: prepared.agentId } : {}),
|
|
};
|
|
beginScopedChatSending(host, sendingScope);
|
|
const isVisibleSession = () => visibleSessionMatches(host, sessionKey, prepared.agentId);
|
|
if (isVisibleSession()) {
|
|
resetToolStream(host as unknown as Parameters<typeof resetToolStream>[0]);
|
|
resetChatScroll(host as unknown as Parameters<typeof resetChatScroll>[0]);
|
|
setChatError(host, null);
|
|
reconcileChatRunLifecycle(host as unknown as Parameters<typeof reconcileChatRunLifecycle>[0], {
|
|
clearRunStatus: true,
|
|
});
|
|
}
|
|
|
|
try {
|
|
const ack = prepared.skillWorkshopRevision
|
|
? await requestSkillWorkshopRevisionChatSend(host as unknown as ChatState, {
|
|
proposalId: prepared.skillWorkshopRevision.proposalId,
|
|
...(prepared.skillWorkshopRevision.agentId
|
|
? { agentId: prepared.skillWorkshopRevision.agentId }
|
|
: {}),
|
|
...(prepared.agentId ? { targetAgentId: prepared.agentId } : {}),
|
|
instructions: message,
|
|
runId,
|
|
sessionKey,
|
|
})
|
|
: await requestChatSend(host as unknown as ChatState, {
|
|
message,
|
|
attachments: hasAttachments ? attachments : undefined,
|
|
runId,
|
|
sessionKey,
|
|
agentId: prepared.agentId,
|
|
});
|
|
updateChatSendAckTiming(host, runId, ack, sendingItem, requestStartedAtMs);
|
|
recordChatSendTiming(host, sendingItem, "ack", sendingItem.sendSubmittedAtMs, {
|
|
ackStatus: ack.status,
|
|
requestDurationMs: roundedControlUiDurationMs(controlUiNowMs() - requestStartedAtMs),
|
|
...chatSendAckServerTimingEventFields(ack),
|
|
});
|
|
if (isTerminalFailureChatSendAck(ack)) {
|
|
const error = formatTerminalChatSendAckError(ack, "chat");
|
|
updateQueuedSendItem(host, storageMode, sessionKey, id, (item) => ({
|
|
...item,
|
|
sendError: error,
|
|
sendState: "failed",
|
|
}));
|
|
if (isVisibleSession()) {
|
|
reconcileChatRunLifecycle(
|
|
host as unknown as Parameters<typeof reconcileChatRunLifecycle>[0],
|
|
{
|
|
outcome: "interrupted",
|
|
sessionStatus: ack.status === "error" ? "failed" : "killed",
|
|
runId: ack.runId,
|
|
sessionKey,
|
|
clearLocalRun: true,
|
|
clearChatStream: true,
|
|
clearToolStream: true,
|
|
clearSideResultTerminalRuns: true,
|
|
publishRunStatus: false,
|
|
armLocalTerminalReconcile: ack.runId === runId,
|
|
},
|
|
);
|
|
setChatError(host, error);
|
|
restoreComposerAfterFailedSend(host, opts ?? {});
|
|
}
|
|
recordChatSendTiming(host, sendingItem, "failed", sendingItem.sendSubmittedAtMs, {
|
|
error,
|
|
ackStatus: ack.status,
|
|
});
|
|
return "failed";
|
|
}
|
|
const retireOnAck = ack.status === "ok" || storageMode === "memory";
|
|
if (retireOnAck) {
|
|
removeQueuedMessageWithoutReleasing(host, id, sessionKey);
|
|
}
|
|
if (isVisibleSession()) {
|
|
if (retireOnAck) {
|
|
appendUserChatMessage(
|
|
host as unknown as ChatState,
|
|
message,
|
|
hasAttachments ? attachments : undefined,
|
|
startedAt,
|
|
);
|
|
}
|
|
if (ack.status === "ok") {
|
|
reconcileChatRunLifecycle(
|
|
host as unknown as Parameters<typeof reconcileChatRunLifecycle>[0],
|
|
{
|
|
outcome: "done",
|
|
sessionStatus: "done",
|
|
runId: ack.runId,
|
|
sessionKey,
|
|
clearLocalRun: true,
|
|
clearChatStream: true,
|
|
clearToolStream: true,
|
|
clearSideResultTerminalRuns: true,
|
|
publishRunStatus: false,
|
|
armLocalTerminalReconcile: true,
|
|
},
|
|
);
|
|
void loadChatHistory(host as unknown as ChatState);
|
|
} else if (isNonTerminalAgentRunStatus(ack.status)) {
|
|
const hasAlreadyAdoptedRunStream =
|
|
host.chatRunId === ack.runId && typeof host.chatStream === "string";
|
|
host.chatRunId = ack.runId;
|
|
// Gateway can deliver the first delta before the chat.send ACK resolves.
|
|
// Preserve that adopted stream; resetting here makes first replies vanish
|
|
// until a later delta or final event arrives.
|
|
if (!hasAlreadyAdoptedRunStream) {
|
|
host.chatStream = "";
|
|
(host as ChatHost & { chatStreamStartedAt?: number | null }).chatStreamStartedAt =
|
|
startedAt;
|
|
}
|
|
} else {
|
|
reconcileChatRunLifecycle(
|
|
host as unknown as Parameters<typeof reconcileChatRunLifecycle>[0],
|
|
{
|
|
outcome: "interrupted",
|
|
sessionStatus: ack.status === "error" ? "failed" : "killed",
|
|
runId: ack.runId,
|
|
sessionKey,
|
|
clearLocalRun: true,
|
|
clearChatStream: true,
|
|
clearToolStream: true,
|
|
clearSideResultTerminalRuns: true,
|
|
publishRunStatus: false,
|
|
armLocalTerminalReconcile: ack.runId === runId,
|
|
},
|
|
);
|
|
}
|
|
}
|
|
if (prepared.refreshSessions) {
|
|
const refreshTarget = {
|
|
sessionKey,
|
|
agentId: prepared.agentId,
|
|
};
|
|
if (ack.status === "ok") {
|
|
void refreshChatSessionListForTarget(host, refreshTarget);
|
|
} else if (isNonTerminalAgentRunStatus(ack.status)) {
|
|
host.refreshSessionsAfterChat.set(ack.runId, refreshTarget);
|
|
}
|
|
}
|
|
discardChatAttachmentDataUrls(excludeComposerAttachments(host, attachments));
|
|
return retireOnAck ? "sent" : "pending";
|
|
} catch (err) {
|
|
finishScopedChatSending(host, sendingScope);
|
|
const error = formatConnectError(err);
|
|
if (isRecoverableChatSendError(err, error)) {
|
|
const failedBeforeTransport = isProvablyPreTransportChatSendError(err);
|
|
const retryDelayMs = retryableGatewayDelayMs(err);
|
|
const safelyRejected = failedBeforeTransport || retryDelayMs !== null;
|
|
if (storageMode === "memory") {
|
|
const hasComposerSnapshot =
|
|
opts?.previousDraft !== undefined || opts?.previousAttachments !== undefined;
|
|
const canRestoreSafely =
|
|
hasComposerSnapshot &&
|
|
safelyRejected &&
|
|
pendingComposerRestorePlan(host, opts ?? {}).complete;
|
|
if (canRestoreSafely) {
|
|
cancelPendingSendBeforeRequest(host, prepared, {
|
|
previousDraft: opts?.previousDraft,
|
|
previousAttachments: opts?.previousAttachments,
|
|
});
|
|
} else {
|
|
updateQueuedSendItem(host, storageMode, sessionKey, id, (item) => ({
|
|
...item,
|
|
...(safelyRejected
|
|
? {
|
|
sendAttempts: queued.sendAttempts,
|
|
sendRequestStartedAtMs: queued.sendRequestStartedAtMs,
|
|
}
|
|
: {}),
|
|
sendError: safelyRejected ? error : UNCONFIRMED_CHAT_SEND_ERROR,
|
|
sendState: safelyRejected ? "failed" : "unconfirmed",
|
|
}));
|
|
}
|
|
if (isVisibleSession()) {
|
|
setChatError(host, canRestoreSafely ? error : OFFLINE_QUEUE_STORAGE_ERROR);
|
|
}
|
|
recordChatSendTiming(host, prepared, "failed", prepared.sendSubmittedAtMs, {
|
|
error: canRestoreSafely ? error : OFFLINE_QUEUE_STORAGE_ERROR,
|
|
});
|
|
return canRestoreSafely ? "failed" : "pending";
|
|
}
|
|
const waiting = updateQueuedMessageForSession(host, sessionKey, id, (item) => ({
|
|
...item,
|
|
...(safelyRejected
|
|
? {
|
|
sendAttempts: queued.sendAttempts,
|
|
sendRequestStartedAtMs: queued.sendRequestStartedAtMs,
|
|
}
|
|
: {}),
|
|
sendError: error,
|
|
sendState: "waiting-reconnect",
|
|
}));
|
|
if (!waiting) {
|
|
const hasComposerSnapshot =
|
|
opts?.previousDraft !== undefined || opts?.previousAttachments !== undefined;
|
|
const canRestorePreTransport =
|
|
hasComposerSnapshot &&
|
|
failedBeforeTransport &&
|
|
pendingComposerRestorePlan(host, opts ?? {}).complete;
|
|
if (canRestorePreTransport) {
|
|
cancelPendingSendBeforeRequest(host, waiting ?? prepared, {
|
|
previousDraft: opts?.previousDraft,
|
|
previousAttachments: opts?.previousAttachments,
|
|
});
|
|
} else {
|
|
// The request may have reached the Gateway. Retain its run id so a
|
|
// manual retry remains idempotent even when this tab cannot persist it.
|
|
updateQueuedMessageForSession(host, sessionKey, id, (item) => ({
|
|
...item,
|
|
sendError: OFFLINE_QUEUE_STORAGE_ERROR,
|
|
sendState: "failed",
|
|
}));
|
|
}
|
|
if (isVisibleSession()) {
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
}
|
|
recordChatSendTiming(host, prepared, "failed", prepared.sendSubmittedAtMs, {
|
|
error: OFFLINE_QUEUE_STORAGE_ERROR,
|
|
});
|
|
return canRestorePreTransport ? "failed" : "pending";
|
|
}
|
|
if (isVisibleSession()) {
|
|
setChatError(
|
|
host,
|
|
retryDelayMs === null
|
|
? "Message will send when the Gateway reconnects."
|
|
: "The Gateway asked us to retry this message shortly.",
|
|
);
|
|
}
|
|
if (retryDelayMs !== null) {
|
|
scheduleStoredChatOutboxRetry(
|
|
host,
|
|
{ sessionKey, agentId: prepared.agentId },
|
|
retryDelayMs,
|
|
);
|
|
}
|
|
recordChatSendTiming(host, prepared, "waiting-reconnect", prepared.sendSubmittedAtMs, {
|
|
error,
|
|
});
|
|
return "pending";
|
|
}
|
|
updateQueuedSendItem(host, storageMode, sessionKey, id, (item) => ({
|
|
...item,
|
|
sendError: error,
|
|
sendState: "failed",
|
|
}));
|
|
if (isVisibleSession()) {
|
|
setChatError(host, error);
|
|
restoreComposerAfterFailedSend(host, opts ?? {});
|
|
}
|
|
recordChatSendTiming(host, prepared, "failed", prepared.sendSubmittedAtMs, { error });
|
|
return "failed";
|
|
} finally {
|
|
finishScopedChatSending(host, sendingScope);
|
|
}
|
|
}
|
|
|
|
async function sendChatMessageNow(
|
|
host: ChatHost,
|
|
message: string,
|
|
opts?: {
|
|
queueItemId?: string;
|
|
previousDraft?: string;
|
|
restoreDraft?: boolean;
|
|
attachments?: ChatAttachment[];
|
|
previousAttachments?: ChatAttachment[];
|
|
restoreAttachments?: boolean;
|
|
refreshSessions?: boolean;
|
|
routingSessionKey?: string;
|
|
storageMode?: QueuedChatStorageMode;
|
|
submittedAtMs?: number;
|
|
},
|
|
): Promise<QueuedChatSendResult> {
|
|
const queued =
|
|
opts?.queueItemId != null
|
|
? (host.chatQueue.find((item) => item.id === opts.queueItemId) ?? null)
|
|
: enqueuePendingSendMessage(
|
|
host,
|
|
message,
|
|
opts?.attachments,
|
|
opts?.refreshSessions,
|
|
opts?.submittedAtMs,
|
|
reconnectSafeQueuedSendState(host),
|
|
);
|
|
if (!queued) {
|
|
return "failed";
|
|
}
|
|
const queuedSessionKey = queued.sessionKey ?? host.sessionKey;
|
|
if (opts?.queueItemId == null && !admitQueuedMessageForSession(host, queuedSessionKey, queued)) {
|
|
cancelPendingSendBeforeRequest(host, queued, {
|
|
previousDraft: opts?.previousDraft,
|
|
previousAttachments: opts?.previousAttachments,
|
|
});
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
return "failed";
|
|
}
|
|
const storageMode = opts?.storageMode ?? "durable";
|
|
let result: QueuedChatSendResult;
|
|
if (storageMode === "memory") {
|
|
result = await sendQueuedChatMessage(
|
|
host,
|
|
queued.id,
|
|
{
|
|
previousDraft: opts?.previousDraft,
|
|
previousAttachments: opts?.previousAttachments,
|
|
routingSessionKey: opts?.routingSessionKey ?? queuedSessionKey,
|
|
storageMode,
|
|
},
|
|
queuedSessionKey,
|
|
);
|
|
} else {
|
|
const queuedOutbox = listStoredChatOutboxes(host).find((outbox) =>
|
|
outbox.queue.some((item) => item.id === queued.id),
|
|
);
|
|
if (!queuedOutbox) {
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
return "pending";
|
|
}
|
|
await scheduleStoredChatOutboxDrain(host, queuedOutbox, queued.id, {
|
|
previousDraft: opts?.previousDraft,
|
|
previousAttachments: opts?.previousAttachments,
|
|
routingSessionKey: opts?.routingSessionKey ?? queuedSessionKey,
|
|
});
|
|
const storedItem = listStoredChatOutboxes(host)
|
|
.flatMap((outbox) => outbox.queue)
|
|
.find((item) => item.id === queued.id);
|
|
result = !storedItem ? "sent" : storedItem.sendState === "failed" ? "failed" : "pending";
|
|
}
|
|
const sent = result === "sent";
|
|
if (sent && host.sessionKey === queuedSessionKey) {
|
|
setLastActiveSessionKey(
|
|
host as unknown as Parameters<typeof setLastActiveSessionKey>[0],
|
|
queuedSessionKey,
|
|
);
|
|
resetChatInputHistoryNavigation(host);
|
|
}
|
|
if (
|
|
sent &&
|
|
host.sessionKey === queuedSessionKey &&
|
|
opts?.restoreDraft &&
|
|
opts.previousDraft?.trim()
|
|
) {
|
|
host.chatMessage = opts.previousDraft;
|
|
}
|
|
if (
|
|
sent &&
|
|
host.sessionKey === queuedSessionKey &&
|
|
opts?.restoreAttachments &&
|
|
opts.previousAttachments?.length
|
|
) {
|
|
host.chatAttachments = opts.previousAttachments;
|
|
}
|
|
// Force scroll after sending to ensure viewport is at bottom for incoming stream
|
|
if (host.sessionKey === queuedSessionKey) {
|
|
scheduleChatScroll(host as unknown as Parameters<typeof scheduleChatScroll>[0], true);
|
|
}
|
|
if (sent && host.sessionKey === queuedSessionKey && !host.chatRunId) {
|
|
void flushChatQueue(host);
|
|
}
|
|
return result;
|
|
}
|
|
|
|
function attachmentSubmitSignature(attachment: ChatAttachment): string {
|
|
const dataUrl = getChatAttachmentDataUrl(attachment);
|
|
return JSON.stringify([
|
|
attachment.id,
|
|
attachment.mimeType,
|
|
attachment.fileName ?? "",
|
|
attachment.sizeBytes ?? 0,
|
|
dataUrl?.length ?? 0,
|
|
dataUrl?.slice(0, 64) ?? "",
|
|
]);
|
|
}
|
|
|
|
function chatSubmitKey(
|
|
host: ChatHost,
|
|
kind: "detached" | "local" | "message",
|
|
message: string,
|
|
attachments: ChatAttachment[],
|
|
skillWorkshopRevision?: ChatQueueSkillWorkshopRevision,
|
|
): string {
|
|
return JSON.stringify([
|
|
kind,
|
|
host.sessionKey,
|
|
message.trim(),
|
|
skillWorkshopRevision?.proposalId ?? "",
|
|
skillWorkshopRevision?.agentId ?? "",
|
|
attachments.map(attachmentSubmitSignature),
|
|
]);
|
|
}
|
|
|
|
async function withChatSubmitGuard<T>(
|
|
host: ChatHost,
|
|
key: string,
|
|
run: () => Promise<T>,
|
|
): Promise<T | undefined> {
|
|
const guards = (host.chatSubmitGuards ??= new Map<string, Promise<void>>());
|
|
if (guards.has(key)) {
|
|
return undefined;
|
|
}
|
|
let releaseGuard!: () => void;
|
|
const guard = new Promise<void>((resolve) => {
|
|
releaseGuard = resolve;
|
|
});
|
|
guards.set(key, guard);
|
|
try {
|
|
return await run();
|
|
} finally {
|
|
releaseGuard();
|
|
if (guards.get(key) === guard) {
|
|
guards.delete(key);
|
|
}
|
|
}
|
|
}
|
|
|
|
async function waitForPendingChatSettings(
|
|
host: ChatHost,
|
|
sessionKey: string,
|
|
initialPending: Promise<boolean>,
|
|
agentId?: string,
|
|
): Promise<boolean> {
|
|
let pending = initialPending;
|
|
while (await pending) {
|
|
const nextPending = getPendingChatPickerPatch(host, sessionKey, agentId);
|
|
if (!nextPending || nextPending === pending) {
|
|
return true;
|
|
}
|
|
pending = nextPending;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
function clearSubmittedComposerState(
|
|
host: ChatHost,
|
|
submittedDraft: string,
|
|
submittedAttachments: ChatAttachment[],
|
|
): {
|
|
previousAttachments?: ChatAttachment[];
|
|
previousDraft?: string;
|
|
} {
|
|
const attachmentsUnchanged =
|
|
host.chatAttachments.length === submittedAttachments.length &&
|
|
host.chatAttachments.every((attachment, index) => {
|
|
const submitted = submittedAttachments[index];
|
|
return (
|
|
submitted !== undefined &&
|
|
attachmentSubmitSignature(attachment) === attachmentSubmitSignature(submitted)
|
|
);
|
|
});
|
|
const clearedDraft = host.chatMessage === submittedDraft && attachmentsUnchanged;
|
|
const clearedAttachments = clearedDraft;
|
|
if (clearedDraft) {
|
|
host.chatMessage = "";
|
|
}
|
|
if (clearedAttachments) {
|
|
host.chatAttachments = [];
|
|
}
|
|
if (clearedDraft || clearedAttachments) {
|
|
resetChatInputHistoryNavigation(host);
|
|
}
|
|
return {
|
|
previousAttachments: clearedAttachments ? submittedAttachments : undefined,
|
|
previousDraft: clearedDraft ? submittedDraft : undefined,
|
|
};
|
|
}
|
|
|
|
function snapshotChatAttachments(attachments: readonly ChatAttachment[]): ChatAttachment[] {
|
|
return attachments.map((attachment) => {
|
|
const dataUrl = getChatAttachmentDataUrl(attachment);
|
|
return {
|
|
...attachment,
|
|
...(dataUrl ? { dataUrl } : {}),
|
|
};
|
|
});
|
|
}
|
|
|
|
async function sendDetachedCommandMessage(
|
|
host: ChatHost,
|
|
message: string,
|
|
opts?: {
|
|
previousDraft?: string;
|
|
attachments?: ChatAttachment[];
|
|
previousAttachments?: ChatAttachment[];
|
|
runId?: string;
|
|
},
|
|
) {
|
|
const ack = await sendDetachedChatMessage(
|
|
host as unknown as ChatState,
|
|
message,
|
|
opts?.attachments,
|
|
opts?.runId,
|
|
);
|
|
const ok = isAcceptedChatSendAck(ack);
|
|
if (!ok && opts?.previousDraft != null) {
|
|
host.chatMessage = opts.previousDraft;
|
|
}
|
|
if (!ok && opts?.previousAttachments) {
|
|
host.chatAttachments = opts.previousAttachments;
|
|
}
|
|
if (isTerminalFailureChatSendAck(ack)) {
|
|
setChatError(host, formatTerminalChatSendAckError(ack, "detached"));
|
|
}
|
|
if (ok) {
|
|
setLastActiveSessionKey(
|
|
host as unknown as Parameters<typeof setLastActiveSessionKey>[0],
|
|
host.sessionKey,
|
|
);
|
|
releaseChatAttachmentPayloads(excludeComposerAttachments(host, opts?.attachments));
|
|
}
|
|
return ack;
|
|
}
|
|
|
|
export async function steerQueuedChatMessage(host: ChatHost, id: string) {
|
|
if (!host.connected || !host.chatRunId) {
|
|
return;
|
|
}
|
|
const activeRunId = host.chatRunId;
|
|
const item = host.chatQueue.find(
|
|
(entry) =>
|
|
entry.id === id &&
|
|
!entry.pendingRunId &&
|
|
!entry.localCommandName &&
|
|
(entry.sendState === undefined || entry.sendState === "waiting-idle"),
|
|
);
|
|
if (!item) {
|
|
return;
|
|
}
|
|
const itemSessionKey = item.sessionKey ?? host.sessionKey;
|
|
const message = item.text.trim();
|
|
const attachments = item.attachments ?? [];
|
|
const hasAttachments = attachments.length > 0;
|
|
if (!message && !hasAttachments) {
|
|
return;
|
|
}
|
|
|
|
// Claim the durable row before transport so a crash or ambiguous ACK cannot
|
|
// replay the original queued turn after the steer may already be accepted.
|
|
const claimed = updateQueuedMessage(host, id, (entry) => ({
|
|
...entry,
|
|
sendError: UNCONFIRMED_STEER_ERROR,
|
|
sendState: "unconfirmed",
|
|
}));
|
|
if (!claimed) {
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
return;
|
|
}
|
|
const pendingIndicator: ChatQueueItem = {
|
|
id: item.id,
|
|
text: item.text,
|
|
createdAt: item.createdAt,
|
|
kind: "steered",
|
|
...(item.attachments?.length ? { attachments: item.attachments } : {}),
|
|
pendingRunId: activeRunId,
|
|
};
|
|
const hasTransientProjection = setTransientQueuedMessageProjection(
|
|
host,
|
|
itemSessionKey,
|
|
{
|
|
...claimed,
|
|
kind: "steered",
|
|
sendError: undefined,
|
|
sendState: "steering",
|
|
},
|
|
item.agentId,
|
|
);
|
|
if (!hasTransientProjection) {
|
|
const restored = updateQueuedMessage(host, id, () => item);
|
|
if (!restored) {
|
|
host.chatQueue = host.chatQueue.map((entry) => (entry.id === id ? item : entry));
|
|
}
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
return;
|
|
}
|
|
host.chatQueue = host.chatQueue.map((entry) => (entry.id === id ? pendingIndicator : entry));
|
|
const ack = await sendChatMessageWithGeneratedRunId(
|
|
host as unknown as ChatState,
|
|
message,
|
|
hasAttachments ? attachments : undefined,
|
|
() => visibleSessionMatches(host, itemSessionKey, item.agentId),
|
|
);
|
|
const pendingStillVisible = host.chatQueue.some(
|
|
(entry) => entry.id === id && entry.pendingRunId === activeRunId,
|
|
);
|
|
replacePendingQueuedMessageProjection(
|
|
host,
|
|
itemSessionKey,
|
|
id,
|
|
activeRunId,
|
|
claimed,
|
|
item.agentId,
|
|
);
|
|
clearTransientQueuedMessageProjection(host, itemSessionKey, id, item.agentId);
|
|
const itemStillVisible = visibleSessionMatches(host, itemSessionKey, item.agentId);
|
|
if (!ack) {
|
|
// A transport failure does not prove the steer was rejected. Keep the
|
|
// durable row parked so reconnect cannot replay it as a separate turn.
|
|
if (itemStillVisible) {
|
|
setChatError(host, UNCONFIRMED_STEER_ERROR);
|
|
}
|
|
return;
|
|
}
|
|
if (isTerminalFailureChatSendAck(ack)) {
|
|
const restored = updateQueuedMessage(host, id, (entry) => ({
|
|
...item,
|
|
...(entry.attachments?.length ? { attachments: entry.attachments } : {}),
|
|
}));
|
|
if (!restored) {
|
|
if (itemStillVisible) {
|
|
setChatError(host, UNCONFIRMED_STEER_ERROR);
|
|
}
|
|
} else {
|
|
if (itemStillVisible) {
|
|
setChatError(host, formatTerminalChatSendAckError(ack, "steer"));
|
|
}
|
|
const restoredOutbox = listStoredChatOutboxes(host).find((outbox) =>
|
|
outbox.queue.some((entry) => entry.id === id),
|
|
);
|
|
if (!host.chatRunId && restoredOutbox) {
|
|
void scheduleStoredChatOutboxDrain(host, restoredOutbox);
|
|
}
|
|
}
|
|
return;
|
|
}
|
|
const removed = removeQueuedMessageWithoutReleasing(host, id, itemSessionKey, item.agentId);
|
|
if (!removed) {
|
|
if (itemStillVisible) {
|
|
setChatError(host, UNCONFIRMED_STEER_ERROR);
|
|
}
|
|
return;
|
|
}
|
|
if (
|
|
ack.status !== "ok" &&
|
|
pendingStillVisible &&
|
|
host.chatRunId === activeRunId &&
|
|
itemStillVisible
|
|
) {
|
|
host.chatQueue = [...host.chatQueue, pendingIndicator].toSorted(
|
|
(left, right) => left.createdAt - right.createdAt,
|
|
);
|
|
} else {
|
|
releaseChatAttachmentPayloads(attachments);
|
|
}
|
|
if (itemStillVisible) {
|
|
setLastActiveSessionKey(
|
|
host as unknown as Parameters<typeof setLastActiveSessionKey>[0],
|
|
itemSessionKey,
|
|
);
|
|
scheduleChatScroll(host as unknown as Parameters<typeof scheduleChatScroll>[0]);
|
|
}
|
|
}
|
|
|
|
type StoredChatOutboxDrainResult = "blocked" | "empty";
|
|
type StoredChatOutboxDrainLane = {
|
|
freshAdmissions: Set<string>;
|
|
host: ChatHost;
|
|
pendingOptions: Map<string, QueuedChatSendOptions>;
|
|
promise: Promise<void>;
|
|
rerun: boolean;
|
|
};
|
|
|
|
const storedChatOutboxDrainLanesByClient = new WeakMap<
|
|
GatewayBrowserClient,
|
|
Map<string, StoredChatOutboxDrainLane>
|
|
>();
|
|
const storedChatOutboxRetryTimersByClient = new WeakMap<
|
|
GatewayBrowserClient,
|
|
Map<string, ReturnType<typeof setTimeout>>
|
|
>();
|
|
|
|
function storedChatOutboxDrainLanesForClient(
|
|
client: GatewayBrowserClient,
|
|
): Map<string, StoredChatOutboxDrainLane> {
|
|
const existing = storedChatOutboxDrainLanesByClient.get(client);
|
|
if (existing) {
|
|
return existing;
|
|
}
|
|
const lanes = new Map<string, StoredChatOutboxDrainLane>();
|
|
storedChatOutboxDrainLanesByClient.set(client, lanes);
|
|
return lanes;
|
|
}
|
|
|
|
function storedChatOutboxRetryTimersForClient(
|
|
client: GatewayBrowserClient,
|
|
): Map<string, ReturnType<typeof setTimeout>> {
|
|
const existing = storedChatOutboxRetryTimersByClient.get(client);
|
|
if (existing) {
|
|
return existing;
|
|
}
|
|
const timers = new Map<string, ReturnType<typeof setTimeout>>();
|
|
storedChatOutboxRetryTimersByClient.set(client, timers);
|
|
return timers;
|
|
}
|
|
|
|
function cancelStoredChatOutboxRetry(client: GatewayBrowserClient, scope: StoredChatOutboxScope) {
|
|
const timers = storedChatOutboxRetryTimersByClient.get(client);
|
|
const key = storedChatOutboxScopeKey(scope);
|
|
const timer = timers?.get(key);
|
|
if (timer !== undefined) {
|
|
clearTimeout(timer);
|
|
timers?.delete(key);
|
|
}
|
|
}
|
|
|
|
function scheduleStoredChatOutboxRetry(
|
|
host: ChatHost,
|
|
scope: StoredChatOutboxScope,
|
|
delayMs: number,
|
|
) {
|
|
const client = host.client;
|
|
if (!host.connected || !client) {
|
|
return;
|
|
}
|
|
const connectionEpoch = host.connectionEpoch;
|
|
const timers = storedChatOutboxRetryTimersForClient(client);
|
|
const key = storedChatOutboxScopeKey(scope);
|
|
if (timers.has(key)) {
|
|
return;
|
|
}
|
|
const timer = setTimeout(() => {
|
|
timers.delete(key);
|
|
if (host.connected && host.client === client && host.connectionEpoch === connectionEpoch) {
|
|
void scheduleStoredChatOutboxDrain(host, scope);
|
|
}
|
|
}, delayMs);
|
|
timers.set(key, timer);
|
|
}
|
|
|
|
function sameStoredChatOutboxScope(
|
|
outbox: StoredChatOutbox,
|
|
scope: StoredChatOutboxScope,
|
|
): boolean {
|
|
return outbox.sessionKey === scope.sessionKey && outbox.agentId === scope.agentId;
|
|
}
|
|
|
|
function readStoredChatOutbox(
|
|
host: ChatHost,
|
|
scope: StoredChatOutboxScope,
|
|
): StoredChatOutbox | undefined {
|
|
return listStoredChatOutboxes(host).find((outbox) => sameStoredChatOutboxScope(outbox, scope));
|
|
}
|
|
|
|
function nextAutomaticStoredChatQueueItem(outbox: StoredChatOutbox): ChatQueueItem | undefined {
|
|
for (const item of outbox.queue) {
|
|
if (item.sendState !== "failed") {
|
|
return item;
|
|
}
|
|
// A failed command may have changed session state before reporting an
|
|
// error. Preserve FIFO until the user explicitly retries or removes it.
|
|
if (item.localCommandName) {
|
|
return undefined;
|
|
}
|
|
}
|
|
return undefined;
|
|
}
|
|
|
|
function sameQueuedDeliveryVersion(left: ChatQueueItem, right: ChatQueueItem): boolean {
|
|
return (
|
|
left.id === right.id &&
|
|
left.sendRunId === right.sendRunId &&
|
|
left.sendAttempts === right.sendAttempts &&
|
|
left.sendState === right.sendState &&
|
|
left.agentId === right.agentId &&
|
|
left.sessionKey === right.sessionKey
|
|
);
|
|
}
|
|
|
|
function historyContainsQueuedSend(history: ChatHistoryResult, item: ChatQueueItem): boolean {
|
|
if (!item.sendRunId) {
|
|
return false;
|
|
}
|
|
const messages = Array.isArray(history.messages) ? history.messages : [];
|
|
return messages.some((message) => {
|
|
if (!message || typeof message !== "object" || Array.isArray(message)) {
|
|
return false;
|
|
}
|
|
const record = message as Record<string, unknown>;
|
|
const marker = record["__openclaw"];
|
|
const metadata =
|
|
marker && typeof marker === "object" && !Array.isArray(marker)
|
|
? (marker as Record<string, unknown>)
|
|
: undefined;
|
|
const idempotencyKey = metadata?.idempotencyKey ?? record.idempotencyKey;
|
|
return idempotencyKey === item.sendRunId || idempotencyKey === `${item.sendRunId}:user`;
|
|
});
|
|
}
|
|
|
|
function historySessionIsIdle(history: ChatHistoryResult): boolean {
|
|
return Boolean(
|
|
history.sessionInfo &&
|
|
history.sessionInfo.hasActiveRun !== true &&
|
|
!isSessionRunActive(history.sessionInfo),
|
|
);
|
|
}
|
|
|
|
function removeHistoryProvenQueuedSend(
|
|
host: ChatHost,
|
|
outbox: StoredChatOutbox,
|
|
item: ChatQueueItem,
|
|
): boolean {
|
|
const removed = removeQueuedMessageWithoutReleasing(host, item.id, outbox.sessionKey);
|
|
if (!removed) {
|
|
return false;
|
|
}
|
|
releaseChatAttachmentPayloads(excludeComposerAttachments(host, removed.attachments));
|
|
if (visibleSessionMatches(host, outbox.sessionKey, outbox.agentId)) {
|
|
void loadChatHistory(host as unknown as ChatState);
|
|
}
|
|
return true;
|
|
}
|
|
|
|
async function reconcileStoredChatOutboxHead(
|
|
host: ChatHost,
|
|
outbox: StoredChatOutbox,
|
|
item: ChatQueueItem,
|
|
): Promise<"blocked" | "continue" | "send"> {
|
|
const client = host.client;
|
|
const connectionEpoch = host.connectionEpoch;
|
|
if (!client || !host.connected) {
|
|
return "blocked";
|
|
}
|
|
let history: ChatHistoryResult;
|
|
try {
|
|
history = await client.request<ChatHistoryResult>("chat.history", {
|
|
sessionKey: outbox.sessionKey,
|
|
...(isUiGlobalSessionKey(outbox.sessionKey) && outbox.agentId
|
|
? { agentId: outbox.agentId }
|
|
: {}),
|
|
limit: 1000,
|
|
});
|
|
} catch (err) {
|
|
const retryDelayMs = retryableGatewayDelayMs(err);
|
|
if (
|
|
retryDelayMs !== null &&
|
|
host.client === client &&
|
|
host.connectionEpoch === connectionEpoch &&
|
|
host.connected
|
|
) {
|
|
scheduleStoredChatOutboxRetry(host, outbox, retryDelayMs);
|
|
}
|
|
return "blocked";
|
|
}
|
|
const currentOutbox = readStoredChatOutbox(host, outbox);
|
|
const currentItem = currentOutbox?.queue.find((entry) => entry.id === item.id);
|
|
if (host.client !== client || host.connectionEpoch !== connectionEpoch || !host.connected) {
|
|
return "blocked";
|
|
}
|
|
if (!currentOutbox || !currentItem || !sameQueuedDeliveryVersion(currentItem, item)) {
|
|
return "continue";
|
|
}
|
|
syncChatQueueFromStoredOutbox(host, currentOutbox);
|
|
if (historyContainsQueuedSend(history, item)) {
|
|
return removeHistoryProvenQueuedSend(host, outbox, item) ? "continue" : "blocked";
|
|
}
|
|
if (visibleSessionMatches(host, outbox.sessionKey, outbox.agentId) && isChatBusy(host)) {
|
|
return "blocked";
|
|
}
|
|
if (!historySessionIsIdle(history)) {
|
|
return "blocked";
|
|
}
|
|
if ((item.sendAttempts ?? 0) > 0) {
|
|
// History messages and active-run metadata are not captured atomically.
|
|
// Re-read after the first idle snapshot before classifying delivery as unknown.
|
|
let verifiedHistory: ChatHistoryResult;
|
|
try {
|
|
verifiedHistory = await client.request<ChatHistoryResult>("chat.history", {
|
|
sessionKey: outbox.sessionKey,
|
|
...(isUiGlobalSessionKey(outbox.sessionKey) && outbox.agentId
|
|
? { agentId: outbox.agentId }
|
|
: {}),
|
|
limit: 1000,
|
|
});
|
|
} catch (err) {
|
|
const retryDelayMs = retryableGatewayDelayMs(err);
|
|
if (
|
|
retryDelayMs !== null &&
|
|
host.client === client &&
|
|
host.connectionEpoch === connectionEpoch &&
|
|
host.connected
|
|
) {
|
|
scheduleStoredChatOutboxRetry(host, outbox, retryDelayMs);
|
|
}
|
|
return "blocked";
|
|
}
|
|
const verifiedOutbox = readStoredChatOutbox(host, outbox);
|
|
const verifiedItem = verifiedOutbox?.queue.find((entry) => entry.id === item.id);
|
|
if (host.client !== client || host.connectionEpoch !== connectionEpoch || !host.connected) {
|
|
return "blocked";
|
|
}
|
|
if (!verifiedOutbox || !verifiedItem || !sameQueuedDeliveryVersion(verifiedItem, item)) {
|
|
return "continue";
|
|
}
|
|
syncChatQueueFromStoredOutbox(host, verifiedOutbox);
|
|
if (historyContainsQueuedSend(verifiedHistory, item)) {
|
|
return removeHistoryProvenQueuedSend(host, outbox, item) ? "continue" : "blocked";
|
|
}
|
|
if (!historySessionIsIdle(verifiedHistory)) {
|
|
return "blocked";
|
|
}
|
|
const parked = updateQueuedMessageForSession(host, outbox.sessionKey, item.id, (entry) => ({
|
|
...entry,
|
|
sendError: UNCONFIRMED_CHAT_SEND_ERROR,
|
|
sendState: "unconfirmed",
|
|
}));
|
|
if (parked && visibleSessionMatches(host, outbox.sessionKey, outbox.agentId)) {
|
|
setChatError(host, UNCONFIRMED_CHAT_SEND_ERROR);
|
|
}
|
|
return "blocked";
|
|
}
|
|
return "send";
|
|
}
|
|
|
|
async function drainStoredChatOutbox(
|
|
lane: StoredChatOutboxDrainLane,
|
|
scope: StoredChatOutboxScope,
|
|
): Promise<StoredChatOutboxDrainResult> {
|
|
while (true) {
|
|
const host = lane.host;
|
|
if (!host.connected || !host.client) {
|
|
return "blocked";
|
|
}
|
|
const outbox = readStoredChatOutbox(host, scope);
|
|
if (!outbox) {
|
|
return "empty";
|
|
}
|
|
const item = nextAutomaticStoredChatQueueItem(outbox);
|
|
if (!item) {
|
|
return "empty";
|
|
}
|
|
if (
|
|
item.sendState === "failed" ||
|
|
item.sendState === "unconfirmed" ||
|
|
item.sendState === "waiting-model"
|
|
) {
|
|
syncChatQueueFromStoredOutbox(host, outbox);
|
|
return "blocked";
|
|
}
|
|
const visible = visibleSessionMatches(host, outbox.sessionKey, outbox.agentId);
|
|
if (item.localCommandName) {
|
|
if (!visible || isChatBusy(host)) {
|
|
lane.freshAdmissions.delete(item.id);
|
|
lane.pendingOptions.delete(item.id);
|
|
return "blocked";
|
|
}
|
|
syncChatQueueFromStoredOutbox(host, outbox);
|
|
if (item.localCommandName === "reset") {
|
|
const resetText = item.localCommandArgs ? `/reset ${item.localCommandArgs}` : "/reset";
|
|
const converted = updateQueuedMessageForSession(
|
|
host,
|
|
outbox.sessionKey,
|
|
item.id,
|
|
(entry) => ({
|
|
...entry,
|
|
localCommandArgs: undefined,
|
|
localCommandName: undefined,
|
|
refreshSessions: true,
|
|
text: resetText,
|
|
}),
|
|
);
|
|
if (!converted) {
|
|
return "blocked";
|
|
}
|
|
continue;
|
|
}
|
|
// This token exists only in the live drain that admitted the row. Consume
|
|
// it before command execution so a manual retry cannot inherit it.
|
|
const freshAdmission = lane.freshAdmissions.delete(item.id);
|
|
lane.pendingOptions.delete(item.id);
|
|
if (!freshAdmission) {
|
|
const reconciled = await reconcileStoredChatOutboxHead(host, outbox, item);
|
|
if (reconciled === "blocked") {
|
|
return "blocked";
|
|
}
|
|
if (reconciled === "continue") {
|
|
continue;
|
|
}
|
|
}
|
|
// Claim in place before executing. This preserves FIFO on command failure
|
|
// and leaves a manual-review marker if the page disappears mid-command.
|
|
const claimed = updateQueuedMessageForSession(host, outbox.sessionKey, item.id, (entry) => ({
|
|
...entry,
|
|
sendError: undefined,
|
|
sendState: "executing-command",
|
|
}));
|
|
if (!claimed) {
|
|
return "blocked";
|
|
}
|
|
const commandClient = host.client;
|
|
const commandConnectionEpoch = host.connectionEpoch;
|
|
const commandScopeIsCurrent = () =>
|
|
host.connected &&
|
|
host.client === commandClient &&
|
|
host.connectionEpoch === commandConnectionEpoch &&
|
|
visibleSessionMatches(host, outbox.sessionKey, outbox.agentId);
|
|
try {
|
|
const dispatchResult = await dispatchChatSlashCommand(
|
|
host,
|
|
claimed.localCommandName ?? item.localCommandName,
|
|
claimed.localCommandArgs ?? "",
|
|
{
|
|
sendResetMessage: (message, resetOpts) =>
|
|
sendResetSlashCommand(host, message, resetOpts),
|
|
},
|
|
);
|
|
if (dispatchResult === "failed") {
|
|
const commandStillCurrent = commandScopeIsCurrent();
|
|
const error =
|
|
(commandStillCurrent ? host.lastError : null) ??
|
|
`Command /${item.localCommandName} failed.`;
|
|
if (
|
|
!updateQueuedMessageForSession(host, outbox.sessionKey, item.id, (entry) => ({
|
|
...entry,
|
|
sendError: error,
|
|
sendState: "failed",
|
|
}))
|
|
) {
|
|
if (commandStillCurrent) {
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
}
|
|
}
|
|
return "blocked";
|
|
}
|
|
if (dispatchResult === "uncertain") {
|
|
const currentOutbox = readStoredChatOutbox(host, outbox);
|
|
const currentIndex =
|
|
currentOutbox?.queue.findIndex((entry) => entry.id === item.id) ?? -1;
|
|
const successor = currentIndex >= 0 ? currentOutbox?.queue[currentIndex + 1] : undefined;
|
|
if (
|
|
successor &&
|
|
!updateQueuedMessageForSession(
|
|
host,
|
|
outbox.sessionKey,
|
|
successor.id,
|
|
(entry) => ({
|
|
...entry,
|
|
sendError: UNCERTAIN_CLEAR_SUCCESSOR_ERROR,
|
|
sendState: "unconfirmed",
|
|
}),
|
|
outbox.agentId,
|
|
)
|
|
) {
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
// If the successor barrier cannot be made durable, keep the
|
|
// claimed clear row. Its persisted executing-command projection
|
|
// is unconfirmed, which safely blocks this lane after reload.
|
|
return "blocked";
|
|
}
|
|
}
|
|
if (!removeQueuedMessageWithoutReleasing(host, item.id, outbox.sessionKey)) {
|
|
if (commandScopeIsCurrent()) {
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
}
|
|
return "blocked";
|
|
}
|
|
if (dispatchResult === "uncertain") {
|
|
// The destructive command itself is consumed. An unconfirmed
|
|
// successor is the durable manual-review barrier for this FIFO lane.
|
|
return "blocked";
|
|
}
|
|
if (commandScopeIsCurrent()) {
|
|
setChatError(host, null);
|
|
}
|
|
} catch (err) {
|
|
const commandStillCurrent = commandScopeIsCurrent();
|
|
if (
|
|
!updateQueuedMessageForSession(host, outbox.sessionKey, item.id, (entry) => ({
|
|
...entry,
|
|
sendError: String(err),
|
|
sendState: "failed",
|
|
}))
|
|
) {
|
|
if (commandStillCurrent) {
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
}
|
|
return "blocked";
|
|
}
|
|
if (commandStillCurrent) {
|
|
setChatError(host, String(err));
|
|
}
|
|
return "blocked";
|
|
}
|
|
continue;
|
|
}
|
|
if (isUiGlobalSessionKey(outbox.sessionKey) && !outbox.agentId) {
|
|
lane.freshAdmissions.delete(item.id);
|
|
lane.pendingOptions.delete(item.id);
|
|
return "blocked";
|
|
}
|
|
// Consume fresh provenance before any await. A restored or deferred row
|
|
// has no token and must reconcile Gateway history before transport.
|
|
const freshAdmission = lane.freshAdmissions.delete(item.id);
|
|
const pendingOptions = lane.pendingOptions.get(item.id);
|
|
lane.pendingOptions.delete(item.id);
|
|
const needsHistory = !freshAdmission;
|
|
if (needsHistory) {
|
|
const reconciled = await reconcileStoredChatOutboxHead(host, outbox, item);
|
|
if (reconciled === "blocked") {
|
|
return "blocked";
|
|
}
|
|
if (reconciled === "continue") {
|
|
continue;
|
|
}
|
|
}
|
|
if (visible && isChatBusy(host)) {
|
|
syncChatQueueFromStoredOutbox(host, outbox);
|
|
updateQueuedMessageForSession(host, outbox.sessionKey, item.id, (entry) => ({
|
|
...entry,
|
|
sendState: host.connected && host.client ? "waiting-idle" : "waiting-reconnect",
|
|
}));
|
|
return "blocked";
|
|
}
|
|
const currentOutbox = readStoredChatOutbox(host, scope);
|
|
const currentItem = currentOutbox?.queue.find((entry) => entry.id === item.id);
|
|
if (!currentOutbox || !currentItem || !sameQueuedDeliveryVersion(currentItem, item)) {
|
|
continue;
|
|
}
|
|
syncChatQueueFromStoredOutbox(host, currentOutbox);
|
|
const result = await sendQueuedChatMessage(host, item.id, pendingOptions, outbox.sessionKey);
|
|
if (result === "pending") {
|
|
// A pending ACK/reconnect state owns the next wakeup. Any rerun requested
|
|
// while this RPC was in flight is already reflected in the durable queue.
|
|
lane.rerun = false;
|
|
return "blocked";
|
|
}
|
|
if (result === "failed") {
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
|
|
async function scheduleStoredChatOutboxDrain(
|
|
host: ChatHost,
|
|
scope: StoredChatOutboxScope,
|
|
itemId?: string,
|
|
options?: QueuedChatSendOptions,
|
|
): Promise<void> {
|
|
const client = host.client;
|
|
if (!host.connected || !client) {
|
|
return;
|
|
}
|
|
cancelStoredChatOutboxRetry(client, scope);
|
|
// Drain ownership follows the live gateway client. A disconnected client can
|
|
// leave an RPC pending, but its lane must never capture a replacement client.
|
|
const lanes = storedChatOutboxDrainLanesForClient(client);
|
|
const key = storedChatOutboxScopeKey(scope);
|
|
const existing = lanes.get(key);
|
|
if (existing) {
|
|
const existingHostOwnsScope =
|
|
existing.host.connected &&
|
|
existing.host.client === client &&
|
|
visibleSessionMatches(existing.host, scope.sessionKey, scope.agentId);
|
|
const candidateOwnsScope = visibleSessionMatches(host, scope.sessionKey, scope.agentId);
|
|
// Local commands need the visible pane's session-bound UI state. Keep that
|
|
// owner while connected; an inactive split pane may still request a rerun.
|
|
if (!existingHostOwnsScope && candidateOwnsScope) {
|
|
existing.host = host;
|
|
} else if (!existing.host.connected || existing.host.client !== client) {
|
|
existing.host = host;
|
|
}
|
|
existing.rerun = true;
|
|
if (itemId && options) {
|
|
existing.pendingOptions.set(itemId, options);
|
|
}
|
|
if (itemId) {
|
|
existing.freshAdmissions.add(itemId);
|
|
}
|
|
await existing.promise;
|
|
return;
|
|
}
|
|
let resolveLane!: () => void;
|
|
let rejectLane!: (reason?: unknown) => void;
|
|
const promise = new Promise<void>((resolve, reject) => {
|
|
resolveLane = resolve;
|
|
rejectLane = reject;
|
|
});
|
|
const lane: StoredChatOutboxDrainLane = {
|
|
freshAdmissions: new Set(itemId ? [itemId] : []),
|
|
host,
|
|
pendingOptions: new Map(itemId && options ? [[itemId, options]] : []),
|
|
promise,
|
|
rerun: false,
|
|
};
|
|
lanes.set(key, lane);
|
|
void (async () => {
|
|
do {
|
|
lane.rerun = false;
|
|
await drainStoredChatOutbox(lane, scope);
|
|
} while (lane.rerun);
|
|
})().then(resolveLane, rejectLane);
|
|
try {
|
|
await lane.promise;
|
|
} finally {
|
|
if (lanes.get(key) === lane) {
|
|
lanes.delete(key);
|
|
}
|
|
}
|
|
}
|
|
|
|
export async function resumeStoredChatOutboxes(host: ChatHost) {
|
|
if (!host.connected || !host.client) {
|
|
return;
|
|
}
|
|
await Promise.allSettled(
|
|
listStoredChatOutboxes(host).map((outbox) => scheduleStoredChatOutboxDrain(host, outbox)),
|
|
);
|
|
}
|
|
|
|
async function flushChatQueue(host: ChatHost) {
|
|
const outbox = listStoredChatOutboxes(host).find((candidate) =>
|
|
visibleSessionMatches(host, candidate.sessionKey, candidate.agentId),
|
|
);
|
|
if (outbox) {
|
|
await scheduleStoredChatOutboxDrain(host, outbox);
|
|
}
|
|
}
|
|
|
|
export async function retryReconnectableQueuedChatSends(host: ChatHost) {
|
|
await resumeStoredChatOutboxes(host);
|
|
}
|
|
|
|
export async function retryQueuedChatMessage(host: ChatHost, id: string) {
|
|
const item = host.chatQueue.find((entry) => entry.id === id);
|
|
if (
|
|
!item ||
|
|
item.pendingRunId ||
|
|
item.sendState === "executing-command" ||
|
|
item.sendState === "steering" ||
|
|
item.sendState === "sending" ||
|
|
item.sendState === "waiting-model"
|
|
) {
|
|
return;
|
|
}
|
|
let outbox = listStoredChatOutboxes(host).find((candidate) =>
|
|
candidate.queue.some((entry) => entry.id === item.id),
|
|
);
|
|
if (!outbox) {
|
|
const wasVolatile = isVolatileQueuedMessage(host, item.id);
|
|
if (!admitQueuedMessageForSession(host, item.sessionKey ?? host.sessionKey, item)) {
|
|
if (
|
|
wasVolatile &&
|
|
!item.localCommandName &&
|
|
item.sendRunId &&
|
|
(item.sendState === "failed" || item.sendState === "unconfirmed") &&
|
|
canSendVolatileQueueItem(host, item)
|
|
) {
|
|
const retry = updateVolatileQueuedMessage(host, id, (entry) => ({
|
|
...entry,
|
|
sendAttempts: 0,
|
|
sendError: undefined,
|
|
sendRunId: entry.sendState === "failed" ? generateUUID() : entry.sendRunId,
|
|
sendState: undefined,
|
|
}));
|
|
if (!retry) {
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
return;
|
|
}
|
|
await sendQueuedChatMessage(
|
|
host,
|
|
retry.id,
|
|
{
|
|
routingSessionKey: retry.sessionKey ?? host.sessionKey,
|
|
storageMode: "memory",
|
|
},
|
|
retry.sessionKey ?? host.sessionKey,
|
|
);
|
|
return;
|
|
}
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
return;
|
|
}
|
|
outbox = listStoredChatOutboxes(host).find((candidate) =>
|
|
candidate.queue.some((entry) => entry.id === item.id),
|
|
);
|
|
}
|
|
if (!outbox) {
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
return;
|
|
}
|
|
const retry = updateQueuedMessage(host, id, (entry) => ({
|
|
...entry,
|
|
sendAttempts: 0,
|
|
sendError: undefined,
|
|
sendRunId: entry.sendState === "failed" ? generateUUID() : entry.sendRunId,
|
|
sendState: reconnectSafeQueuedSendState(host),
|
|
}));
|
|
if (!retry) {
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
return;
|
|
}
|
|
outbox = listStoredChatOutboxes(host).find((candidate) =>
|
|
candidate.queue.some((entry) => entry.id === retry.id),
|
|
);
|
|
if (!outbox) {
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
return;
|
|
}
|
|
const drain = scheduleStoredChatOutboxDrain(host, outbox);
|
|
if (host.chatSending && host.chatSendingScopeKey === storedChatOutboxScopeKey(outbox)) {
|
|
void drain;
|
|
return;
|
|
}
|
|
await drain;
|
|
if (!host.chatRunId) {
|
|
void flushChatQueue(host);
|
|
}
|
|
}
|
|
|
|
export async function handleSendChat(
|
|
host: ChatHost,
|
|
messageOverride?: string,
|
|
opts?: ChatSendOptions,
|
|
) {
|
|
const previousDraft = host.chatMessage;
|
|
const message = (messageOverride ?? host.chatMessage).trim();
|
|
const submittedAtMs = controlUiNowMs();
|
|
const submittedSessionKey = host.sessionKey;
|
|
const attachments = host.chatAttachments ?? [];
|
|
const attachmentsToSend = messageOverride == null ? snapshotChatAttachments(attachments) : [];
|
|
const hasAttachments = attachmentsToSend.length > 0;
|
|
const skillWorkshopRevision = opts?.skillWorkshopRevision;
|
|
const shouldInterpretChatCommands = !skillWorkshopRevision;
|
|
|
|
if (!message && !hasAttachments) {
|
|
return;
|
|
}
|
|
|
|
if (messageOverride != null && opts?.confirmReset && !confirmChatResetCommand(message)) {
|
|
return;
|
|
}
|
|
|
|
if (shouldInterpretChatCommands) {
|
|
if (isChatStopCommand(message)) {
|
|
if (messageOverride == null) {
|
|
recordNonTranscriptInputHistory(host, message);
|
|
}
|
|
await handleAbortChat(host);
|
|
return;
|
|
}
|
|
|
|
const parsed = parseSlashCommand(message);
|
|
// The backend resolves /approve before active-run admission. Send it now so
|
|
// the approval command cannot queue behind the run that is waiting for it.
|
|
const shouldSendDetachedCommand =
|
|
isBtwCommand(message) || (parsed?.command.key === "approve" && isChatBusy(host));
|
|
if (shouldSendDetachedCommand) {
|
|
const submitKey = chatSubmitKey(host, "detached", message, attachmentsToSend);
|
|
// Covers every non-accepted path — early exits, guard dedupe, and
|
|
// rejected acks — so the side-chat panel can restore its typed
|
|
// follow-up even when no request was sent.
|
|
let detachedSendAccepted = false;
|
|
await withChatSubmitGuard(host, submitKey, async () => {
|
|
const pendingSettings = getPendingChatPickerPatch(host, submittedSessionKey);
|
|
if (
|
|
pendingSettings &&
|
|
!(await waitForPendingChatSettings(host, submittedSessionKey, pendingSettings))
|
|
) {
|
|
return;
|
|
}
|
|
if (host.sessionKey !== submittedSessionKey) {
|
|
return;
|
|
}
|
|
const cleared =
|
|
messageOverride == null
|
|
? clearSubmittedComposerState(host, previousDraft, attachmentsToSend)
|
|
: {};
|
|
if (messageOverride == null) {
|
|
recordNonTranscriptInputHistory(host, message);
|
|
}
|
|
// BTW runs detached and delivers via chat.side_result only; show a
|
|
// pending turn immediately so the send has visible feedback. The run
|
|
// id is generated upfront so the turn is correlatable before the ack
|
|
// returns.
|
|
const btwPending = isBtwCommand(message)
|
|
? {
|
|
question: opts?.sideQuestionDisplayText ?? extractSideQuestionDisplayText(message),
|
|
ts: Date.now(),
|
|
runId: generateUUID(),
|
|
}
|
|
: null;
|
|
if (btwPending) {
|
|
// The superseded run loses its pending record; retire it so its
|
|
// late side_result/terminal events cannot reach the panel or the
|
|
// transcript. Completed turns stay: the panel is a conversation.
|
|
retirePendingChatSideQuestion(host);
|
|
host.chatSideResultPending = btwPending;
|
|
host.chatSideChatHidden = false;
|
|
host.requestUpdate?.();
|
|
}
|
|
const ack = await sendDetachedCommandMessage(host, message, {
|
|
previousDraft: cleared.previousDraft,
|
|
attachments: hasAttachments ? attachmentsToSend : undefined,
|
|
previousAttachments: cleared.previousAttachments,
|
|
runId: btwPending?.runId,
|
|
});
|
|
detachedSendAccepted = isAcceptedChatSendAck(ack);
|
|
// Touch only this send's card: a side_result (or a newer question)
|
|
// may already have replaced it while the ack was in flight.
|
|
if (btwPending && host.chatSideResultPending === btwPending && !detachedSendAccepted) {
|
|
host.chatSideResultPending = null;
|
|
host.requestUpdate?.();
|
|
}
|
|
});
|
|
if (!detachedSendAccepted) {
|
|
opts?.onSideQuestionSendRejected?.();
|
|
}
|
|
return;
|
|
}
|
|
|
|
// Intercept local slash commands (/status, /model, /compact, etc.)
|
|
const forwardModelCommand =
|
|
parsed?.command.key === "model" && shouldForwardModelCommandToServer(parsed.args);
|
|
if (parsed?.command.executeLocal && !forwardModelCommand) {
|
|
const shouldQueueCommand = shouldQueueLocalSlashCommand(parsed.command.key);
|
|
if (shouldQueueCommand) {
|
|
if (messageOverride == null) {
|
|
recordNonTranscriptInputHistory(host, message);
|
|
host.chatMessage = "";
|
|
resetChatInputHistoryNavigation(host);
|
|
}
|
|
const queued = enqueueChatMessage(host, message, undefined, isChatResetCommand(message), {
|
|
args: parsed.args,
|
|
name: parsed.command.key,
|
|
});
|
|
if (queued) {
|
|
queued.sendState = reconnectSafeQueuedSendState(host);
|
|
}
|
|
if (!queued) {
|
|
return;
|
|
}
|
|
if (!admitQueuedMessageForSession(host, host.sessionKey, queued)) {
|
|
removeQueuedMessageWithoutReleasing(host, queued.id);
|
|
if (messageOverride == null) {
|
|
host.chatMessage = previousDraft;
|
|
host.chatAttachments = attachmentsToSend;
|
|
}
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
return;
|
|
}
|
|
if (host.connected && host.client && !isChatBusy(host)) {
|
|
const outbox = listStoredChatOutboxes(host).find((candidate) =>
|
|
candidate.queue.some((entry) => entry.id === queued.id),
|
|
);
|
|
if (outbox) {
|
|
await scheduleStoredChatOutboxDrain(host, outbox, queued.id, {
|
|
routingSessionKey: host.sessionKey,
|
|
});
|
|
}
|
|
}
|
|
return;
|
|
}
|
|
const waitsForPicker = parsed.command.key === "redirect";
|
|
const dispatchLocalCommand = async () => {
|
|
if (waitsForPicker) {
|
|
const pendingSettings = getPendingChatPickerPatch(host, submittedSessionKey);
|
|
if (
|
|
pendingSettings &&
|
|
!(await waitForPendingChatSettings(host, submittedSessionKey, pendingSettings))
|
|
) {
|
|
return;
|
|
}
|
|
if (host.sessionKey !== submittedSessionKey) {
|
|
return;
|
|
}
|
|
}
|
|
let prevDraft = messageOverride == null ? previousDraft : undefined;
|
|
if (messageOverride == null) {
|
|
recordNonTranscriptInputHistory(host, message);
|
|
if (waitsForPicker) {
|
|
prevDraft = clearSubmittedComposerState(
|
|
host,
|
|
previousDraft,
|
|
attachmentsToSend,
|
|
).previousDraft;
|
|
} else {
|
|
host.chatMessage = "";
|
|
host.chatAttachments = [];
|
|
resetChatInputHistoryNavigation(host);
|
|
}
|
|
}
|
|
const dispatchResult = await dispatchChatSlashCommand(
|
|
host,
|
|
parsed.command.key,
|
|
parsed.args,
|
|
{
|
|
previousDraft: prevDraft,
|
|
restoreDraft: Boolean(messageOverride && opts?.restoreDraft),
|
|
sendResetMessage: (resetMessage, resetOpts) =>
|
|
sendResetSlashCommand(host, resetMessage, resetOpts),
|
|
},
|
|
);
|
|
if (dispatchResult === "failed" && messageOverride == null) {
|
|
const restorePlan = pendingComposerRestorePlan(host, {
|
|
previousAttachments: attachmentsToSend,
|
|
previousDraft,
|
|
});
|
|
if (restorePlan.willRestoreDraft) {
|
|
host.chatMessage = previousDraft;
|
|
}
|
|
if (restorePlan.willRestoreAttachments) {
|
|
host.chatAttachments = attachmentsToSend;
|
|
}
|
|
}
|
|
};
|
|
if (waitsForPicker) {
|
|
const submitKey = chatSubmitKey(host, "local", message, attachmentsToSend);
|
|
await withChatSubmitGuard(host, submitKey, dispatchLocalCommand);
|
|
} else {
|
|
await dispatchLocalCommand();
|
|
}
|
|
return;
|
|
}
|
|
}
|
|
|
|
const replyTarget = host.chatReplyTarget;
|
|
const effectiveMessage = replyTarget ? prependReplyQuote(message, replyTarget) : message;
|
|
|
|
const refreshSessions = shouldInterpretChatCommands && isChatResetCommand(message);
|
|
const submitKey = chatSubmitKey(
|
|
host,
|
|
"message",
|
|
effectiveMessage,
|
|
attachmentsToSend,
|
|
skillWorkshopRevision,
|
|
);
|
|
await withChatSubmitGuard(host, submitKey, async () => {
|
|
if (host.sessionKey !== submittedSessionKey) {
|
|
return;
|
|
}
|
|
const cleared =
|
|
messageOverride == null
|
|
? clearSubmittedComposerState(host, previousDraft, attachmentsToSend)
|
|
: {};
|
|
if (messageOverride == null) {
|
|
recordNonTranscriptInputHistory(host, message);
|
|
}
|
|
|
|
const pendingSettings = getPendingChatPickerPatch(host, submittedSessionKey);
|
|
const waitingForSettings = pendingSettings !== undefined;
|
|
const initialSendState: ChatQueueItem["sendState"] = waitingForSettings
|
|
? "waiting-model"
|
|
: reconnectSafeQueuedSendState(host);
|
|
const queued = enqueuePendingSendMessage(
|
|
host,
|
|
effectiveMessage,
|
|
hasAttachments ? attachmentsToSend : undefined,
|
|
refreshSessions,
|
|
submittedAtMs,
|
|
initialSendState,
|
|
skillWorkshopRevision,
|
|
);
|
|
if (!queued) {
|
|
return;
|
|
}
|
|
const admittedDurably = admitQueuedMessageForSession(host, submittedSessionKey, queued);
|
|
const canSendFromMemory =
|
|
!admittedDurably &&
|
|
!waitingForSettings &&
|
|
canSendVolatileQueueItem(host, queued, submittedSessionKey);
|
|
if (!admittedDurably && !canSendFromMemory) {
|
|
cancelPendingSendBeforeRequest(host, queued, {
|
|
previousDraft: cleared.previousDraft,
|
|
previousAttachments: cleared.previousAttachments,
|
|
});
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
return;
|
|
}
|
|
|
|
if (
|
|
pendingSettings &&
|
|
!(await waitForPendingChatSettings(host, submittedSessionKey, pendingSettings))
|
|
) {
|
|
const canRestoreComposer =
|
|
cleared.previousDraft !== undefined &&
|
|
!host.chatMessage.trim() &&
|
|
host.chatAttachments.length === 0;
|
|
const submittedScopeVisible =
|
|
host.sessionKey === submittedSessionKey &&
|
|
visibleSessionMatches(host, submittedSessionKey, queued.agentId);
|
|
if (canRestoreComposer && submittedScopeVisible) {
|
|
cancelPendingSendBeforeRequest(host, queued, {
|
|
previousDraft: cleared.previousDraft,
|
|
previousAttachments: cleared.previousAttachments,
|
|
});
|
|
} else {
|
|
updateQueuedMessageForSession(host, submittedSessionKey, queued.id, (item) => ({
|
|
...item,
|
|
sendError: INTERRUPTED_SETTINGS_WAIT_ERROR,
|
|
sendState: "failed",
|
|
}));
|
|
}
|
|
return;
|
|
}
|
|
if (waitingForSettings) {
|
|
const ready = updateQueuedMessageForSession(host, submittedSessionKey, queued.id, (item) => ({
|
|
...item,
|
|
sendError: undefined,
|
|
sendState: reconnectSafeQueuedSendState(host),
|
|
}));
|
|
if (!ready) {
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
return;
|
|
}
|
|
}
|
|
if (
|
|
host.sessionKey !== submittedSessionKey ||
|
|
!visibleSessionMatches(host, submittedSessionKey, queued.agentId)
|
|
) {
|
|
const parked = updateQueuedMessageForSession(
|
|
host,
|
|
submittedSessionKey,
|
|
queued.id,
|
|
(item) => ({
|
|
...item,
|
|
sendError: undefined,
|
|
sendState: host.connected && host.client ? "waiting-idle" : "waiting-reconnect",
|
|
}),
|
|
);
|
|
if (!parked) {
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
return;
|
|
}
|
|
const outbox = listStoredChatOutboxes(host).find((candidate) =>
|
|
candidate.queue.some((item) => item.id === queued.id),
|
|
);
|
|
if (outbox) {
|
|
await scheduleStoredChatOutboxDrain(host, outbox);
|
|
}
|
|
return;
|
|
}
|
|
|
|
let sendResult: QueuedChatSendResult;
|
|
if (isChatBusy(host)) {
|
|
const pending = updateQueuedMessage(host, queued.id, (item) => ({
|
|
...item,
|
|
sendError: undefined,
|
|
sendState: host.connected && host.client ? "waiting-idle" : "waiting-reconnect",
|
|
}));
|
|
if (!pending) {
|
|
setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR);
|
|
sendResult = "failed";
|
|
} else {
|
|
recordChatSendTiming(host, pending, "queued-busy", submittedAtMs);
|
|
sendResult = "pending";
|
|
}
|
|
} else {
|
|
sendResult = await sendChatMessageNow(host, effectiveMessage, {
|
|
queueItemId: queued.id,
|
|
previousDraft: cleared.previousDraft,
|
|
restoreDraft: Boolean(messageOverride && opts?.restoreDraft),
|
|
attachments: hasAttachments ? attachmentsToSend : undefined,
|
|
previousAttachments: cleared.previousAttachments,
|
|
restoreAttachments: Boolean(messageOverride && opts?.restoreDraft),
|
|
refreshSessions,
|
|
routingSessionKey: submittedSessionKey,
|
|
storageMode: canSendFromMemory ? "memory" : "durable",
|
|
submittedAtMs,
|
|
});
|
|
}
|
|
if (
|
|
sendResult !== "failed" &&
|
|
replyTarget &&
|
|
host.chatReplyTarget?.messageId === replyTarget.messageId &&
|
|
host.sessionKey === submittedSessionKey
|
|
) {
|
|
// A reconnect queue owns the quoted turn before the Gateway ACK. Consume
|
|
// its reply target so later offline turns cannot reuse stale context.
|
|
host.chatReplyTarget = null;
|
|
}
|
|
});
|
|
}
|
|
|
|
function prependReplyQuote(
|
|
message: string,
|
|
replyTarget: NonNullable<ChatHost["chatReplyTarget"]>,
|
|
): string {
|
|
const label = escapeMarkdownInline(replyTarget.senderLabel ?? "User");
|
|
const text = replyTarget.text.trim();
|
|
if (!text.includes("\n")) {
|
|
return `> **${label}:** ${text}\n\n${message}`;
|
|
}
|
|
const quoted = text
|
|
.split("\n")
|
|
.map((line) => `> ${line}`)
|
|
.join("\n");
|
|
return `> **${label}:**\n${quoted}\n\n${message}`;
|
|
}
|
|
|
|
function escapeMarkdownInline(value: string): string {
|
|
return value.replace(/([\\`*_{}[\]()#+\-.!|>])/g, "\\$1");
|
|
}
|
|
|
|
export const flushChatQueueForEvent = flushChatQueue;
|