From dedf85a34d249cf5b2f6d752186cfe37341e3875 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 23 Jul 2026 08:51:29 -0700 Subject: [PATCH] refactor(ui): split chat-send.ts along ack/request/drain/lifecycle seams (#113052) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refactor(ui): split chat-send.ts along ack/request/drain/lifecycle seams chat-send.ts was ~2,330 lines under a grandfathered max-lines suppression, mixing ack normalization, wire requests/routing, the stored-outbox drain scheduler, retry/steer actions, and the send lifecycle. Mechanical extraction into owned modules: - chat-send-contract.ts absorbs ack normalization (it owns the ack shape) - chat-send-request.ts: wire requests + session routing - chat-outbox-drain.ts: drain lanes, retry timers, head reconciliation — colocated with the outbox ownership boundary from the composer split - chat-send-actions.ts / chat-send-queue-state.ts / chat-send-submit.ts: retry/steer actions, queue-state helpers, submission routing - chat-send.ts keeps the send lifecycle (719 lines); its max-lines suppression and baseline entry are removed No compatibility re-exports; cycles broken by hoisting ChatHost into the contract module and injecting the drain's two lifecycle callbacks. Closes #112742. * fix(ui): extract chat-send ack shapes into a leaf module to break the madge type cycle --- config/max-lines-baseline.txt | 1 - ui/src/pages/chat/chat-outbox-drain.ts | 628 +++++++ ui/src/pages/chat/chat-pane.ts | 2 +- ui/src/pages/chat/chat-send-ack.ts | 80 + ui/src/pages/chat/chat-send-actions.ts | 191 +++ ui/src/pages/chat/chat-send-contract.ts | 93 +- ui/src/pages/chat/chat-send-queue-state.ts | 124 ++ ui/src/pages/chat/chat-send-request.ts | 101 ++ ui/src/pages/chat/chat-send-submit.ts | 648 ++++++++ ui/src/pages/chat/chat-send-timing.ts | 2 +- ui/src/pages/chat/chat-send.test.ts | 16 +- ui/src/pages/chat/chat-send.ts | 1733 +------------------- ui/src/pages/chat/chat-state.ts | 10 +- ui/src/pages/chat/steer-lifecycle.ts | 2 +- 14 files changed, 1903 insertions(+), 1728 deletions(-) create mode 100644 ui/src/pages/chat/chat-outbox-drain.ts create mode 100644 ui/src/pages/chat/chat-send-ack.ts create mode 100644 ui/src/pages/chat/chat-send-actions.ts create mode 100644 ui/src/pages/chat/chat-send-queue-state.ts create mode 100644 ui/src/pages/chat/chat-send-request.ts create mode 100644 ui/src/pages/chat/chat-send-submit.ts diff --git a/config/max-lines-baseline.txt b/config/max-lines-baseline.txt index a13f9035efbc..0e4ff20d8403 100644 --- a/config/max-lines-baseline.txt +++ b/config/max-lines-baseline.txt @@ -1093,7 +1093,6 @@ ui/src/pages/chat/chat-history.ts ui/src/pages/chat/chat-pane.ts ui/src/pages/chat/chat-responsive.browser.test.ts ui/src/pages/chat/chat-send.test.ts -ui/src/pages/chat/chat-send.ts ui/src/pages/chat/chat-state.test.ts ui/src/pages/chat/chat-state.ts ui/src/pages/chat/chat-thread.test.ts diff --git a/ui/src/pages/chat/chat-outbox-drain.ts b/ui/src/pages/chat/chat-outbox-drain.ts new file mode 100644 index 000000000000..5cec28410b0d --- /dev/null +++ b/ui/src/pages/chat/chat-outbox-drain.ts @@ -0,0 +1,628 @@ +import { GatewayRequestError, type GatewayBrowserClient } from "../../api/gateway.ts"; +import type { ChatAttachment, ChatQueueItem } from "../../lib/chat/chat-types.ts"; +import { isSessionRunActive } from "../../lib/session-run-state.ts"; +import { visibleSessionMatches } from "../../lib/sessions/index.ts"; +import { isUiGlobalSessionKey } from "../../lib/sessions/session-key.ts"; +import { releaseChatAttachmentPayloads } from "./attachment-payload-store.ts"; +import { + confirmConversationResetForCurrentSession, + dispatchChatSlashCommand, + type ChatCommandResetOptions, +} from "./chat-commands.ts"; +import { loadChatHistory, type ChatHistoryResult, type ChatState } from "./chat-history.ts"; +import { + excludeComposerAttachments, + removeQueuedMessageWithoutReleasing, + syncChatQueueFromStoredOutbox, + updateQueuedMessageForSession, +} from "./chat-queue.ts"; +import type { ChatHost } from "./chat-send-contract.ts"; +import { + listStoredChatOutboxes, + storedChatOutboxScopeKey, + type StoredChatOutbox, + type StoredChatOutboxScope, +} from "./composer-persistence.ts"; +import { isChatBusy } from "./run-lifecycle.ts"; +import { + chatMessagesContainQueuedSend, + OFFLINE_QUEUE_STORAGE_ERROR, + preserveQueuedUserTurn, +} from "./steer-lifecycle.ts"; + +export type QueuedChatSendResult = "sent" | "pending" | "failed"; +export type QueuedChatStorageMode = "durable" | "memory"; +export type QueuedChatSendOptions = { + previousAttachments?: ChatAttachment[]; + previousDraft?: string; + routingSessionKey?: string; + storageMode?: QueuedChatStorageMode; +}; + +export type ChatOutboxDrainDependencies = { + sendQueuedChatMessage: ( + host: ChatHost, + id: string, + opts?: QueuedChatSendOptions, + queuedSessionKey?: string, + ) => Promise; + sendResetSlashCommand: ( + host: ChatHost, + message: string, + opts: ChatCommandResetOptions, + ) => Promise; + setChatError: ( + host: { lastError?: string | null; chatError?: string | null }, + error: string | null, + ) => void; +}; + +type StoredChatOutboxDrainResult = "blocked" | "empty"; +type StoredChatOutboxDrainLane = { + freshAdmissions: Set; + host: ChatHost; + pendingOptions: Map; + promise: Promise; + rerun: boolean; +}; + +const STORED_OUTBOX_RETRY_DEFAULT_MS = 500; +const STORED_OUTBOX_RETRY_MIN_MS = 100; +const STORED_OUTBOX_RETRY_MAX_MS = 30_000; +export const UNCONFIRMED_CHAT_SEND_ERROR = + "Delivery could not be confirmed after reconnect. Check the conversation before retrying."; +const UNCERTAIN_CLEAR_SUCCESSOR_ERROR = + "A preceding /clear may have completed. Review the current conversation before retrying."; + +const storedChatOutboxDrainLanesByClient = new WeakMap< + GatewayBrowserClient, + Map +>(); +const storedChatOutboxRetryTimersByClient = new WeakMap< + GatewayBrowserClient, + Map> +>(); + +function storedChatOutboxClientMap( + store: WeakMap>, + client: GatewayBrowserClient, +): Map { + const existing = store.get(client); + if (existing) { + return existing; + } + const created = new Map(); + store.set(client, created); + return created; +} + +export 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); +} + +export function scheduleStoredChatOutboxRetry( + host: ChatHost, + scope: StoredChatOutboxScope, + delayMs: number, + dependencies: ChatOutboxDrainDependencies, +) { + const client = host.client; + if (!host.connected || !client) { + return; + } + const connectionEpoch = host.connectionEpoch; + const timers = storedChatOutboxClientMap(storedChatOutboxRetryTimersByClient, 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, dependencies); + } + }, delayMs); + timers.set(key, timer); +} + +function readStoredChatOutbox( + host: ChatHost, + scope: StoredChatOutboxScope, +): StoredChatOutbox | undefined { + return listStoredChatOutboxes(host).find( + (outbox) => outbox.sessionKey === scope.sessionKey && outbox.agentId === scope.agentId, + ); +} + +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 + ); +} + +async function readCurrentStoredChatHistory( + host: ChatHost, + outbox: StoredChatOutbox, + item: ChatQueueItem, + client: NonNullable, + connectionEpoch: number | undefined, + dependencies: ChatOutboxDrainDependencies, +): Promise { + let history: ChatHistoryResult; + try { + history = await client.request("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, dependencies); + } + 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 (chatMessagesContainQueuedSend(history.messages, item)) { + // Server history owns the turn, but the visible transcript may not have + // reloaded yet; materialize the turn locally before dropping the queue row + // or the bubble vanishes until loadChatHistory below resolves. + preserveQueuedUserTurn(host, item); + const removed = removeQueuedMessageWithoutReleasing(host, item.id, outbox.sessionKey); + if (!removed) { + return "blocked"; + } + releaseChatAttachmentPayloads(excludeComposerAttachments(host, removed.attachments)); + if (visibleSessionMatches(host, outbox.sessionKey, outbox.agentId)) { + void loadChatHistory(host as unknown as ChatState); + } + return "continue"; + } + if ( + !history.sessionInfo || + history.sessionInfo.hasActiveRun === true || + isSessionRunActive(history.sessionInfo) + ) { + return "blocked"; + } + return history; +} + +async function reconcileStoredChatOutboxHead( + host: ChatHost, + outbox: StoredChatOutbox, + item: ChatQueueItem, + dependencies: ChatOutboxDrainDependencies, +): Promise<"blocked" | "continue" | "send"> { + const client = host.client; + const connectionEpoch = host.connectionEpoch; + if (!client || !host.connected) { + return "blocked"; + } + const history = await readCurrentStoredChatHistory( + host, + outbox, + item, + client, + connectionEpoch, + dependencies, + ); + if (history === "blocked" || history === "continue") { + return history; + } + if (visibleSessionMatches(host, outbox.sessionKey, outbox.agentId) && isChatBusy(host)) { + 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. + const verifiedHistory = await readCurrentStoredChatHistory( + host, + outbox, + item, + client, + connectionEpoch, + dependencies, + ); + if (verifiedHistory === "blocked" || verifiedHistory === "continue") { + return verifiedHistory; + } + 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)) { + dependencies.setChatError(host, UNCONFIRMED_CHAT_SEND_ERROR); + } + return "blocked"; + } + return "send"; +} + +async function drainStoredChatOutbox( + lane: StoredChatOutboxDrainLane, + scope: StoredChatOutboxScope, + dependencies: ChatOutboxDrainDependencies, +): Promise { + while (true) { + const host = lane.host; + if (!host.connected || !host.client) { + return "blocked"; + } + const outbox = readStoredChatOutbox(host, scope); + if (!outbox) { + return "empty"; + } + // Failed non-command rows are skipped; a failed command may have changed + // session state before reporting an error, so it blocks the drain and + // preserves FIFO until the user explicitly retries or removes it. + let item: ChatQueueItem | undefined; + for (const entry of outbox.queue) { + if (entry.sendState !== "failed") { + item = entry; + break; + } + if (entry.localCommandName) { + break; + } + } + if (!item) { + return "empty"; + } + if (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 convertResetToMessage = (sendState?: ChatQueueItem["sendState"]) => + updateQueuedMessageForSession(host, outbox.sessionKey, item.id, (entry) => ({ + ...entry, + localCommandArgs: undefined, + localCommandName: undefined, + refreshSessions: true, + text: resetText, + ...(sendState ? { sendState } : {}), + })); + const confirmation = await confirmConversationResetForCurrentSession(host, { + sessionKey: outbox.sessionKey, + ...(outbox.agentId ? { agentId: outbox.agentId } : {}), + }); + if (confirmation === "deferred") { + const approvedDuringRun = + visibleSessionMatches(host, outbox.sessionKey, outbox.agentId) && host.chatRunId; + const deferred = approvedDuringRun + ? convertResetToMessage("waiting-idle") + : updateQueuedMessageForSession(host, outbox.sessionKey, item.id, (entry) => ({ + ...entry, + sendError: undefined, + sendState: "waiting-idle", + })); + if (!deferred) { + return "blocked"; + } + return "blocked"; + } + if (confirmation === "cancelled") { + if (!removeQueuedMessageWithoutReleasing(host, item.id, outbox.sessionKey)) { + return "blocked"; + } + continue; + } + const converted = convertResetToMessage(); + 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, dependencies); + 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) => + dependencies.sendResetSlashCommand(host, message, resetOpts), + }, + ); + if (dispatchResult === "deferred") { + updateQueuedMessageForSession(host, outbox.sessionKey, item.id, (entry) => ({ + ...entry, + sendError: undefined, + sendState: "waiting-idle", + })); + return "blocked"; + } + 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) { + dependencies.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, + ) + ) { + dependencies.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()) { + dependencies.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()) { + dependencies.setChatError(host, null); + } + } catch (err) { + const commandStillCurrent = commandScopeIsCurrent(); + if ( + !updateQueuedMessageForSession(host, outbox.sessionKey, item.id, (entry) => ({ + ...entry, + sendError: String(err), + sendState: "failed", + })) + ) { + if (commandStillCurrent) { + dependencies.setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR); + } + return "blocked"; + } + if (commandStillCurrent) { + dependencies.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, dependencies); + 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 dependencies.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; + } + } +} + +export async function scheduleStoredChatOutboxDrain( + host: ChatHost, + scope: StoredChatOutboxScope, + dependencies: ChatOutboxDrainDependencies, + itemId?: string, + options?: QueuedChatSendOptions, +): Promise { + const client = host.client; + if (!host.connected || !client) { + return; + } + const key = storedChatOutboxScopeKey(scope); + const retryTimers = storedChatOutboxRetryTimersByClient.get(client); + const retryTimer = retryTimers?.get(key); + if (retryTimer !== undefined) { + clearTimeout(retryTimer); + retryTimers?.delete(key); + } + // 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 = storedChatOutboxClientMap(storedChatOutboxDrainLanesByClient, client); + 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((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, dependencies); + } 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, + dependencies: ChatOutboxDrainDependencies, +) { + if (!host.connected || !host.client) { + return; + } + await Promise.allSettled( + listStoredChatOutboxes(host).map((outbox) => + scheduleStoredChatOutboxDrain(host, outbox, dependencies), + ), + ); +} + +export async function flushStoredChatOutbox( + host: ChatHost, + dependencies: ChatOutboxDrainDependencies, +) { + const outbox = listStoredChatOutboxes(host).find((candidate) => + visibleSessionMatches(host, candidate.sessionKey, candidate.agentId), + ); + if (outbox) { + await scheduleStoredChatOutboxDrain(host, outbox, dependencies); + } +} diff --git a/ui/src/pages/chat/chat-pane.ts b/ui/src/pages/chat/chat-pane.ts index 0a843b7f901c..0b941baabbe7 100644 --- a/ui/src/pages/chat/chat-pane.ts +++ b/ui/src/pages/chat/chat-pane.ts @@ -169,7 +169,7 @@ import { import { markQueuedChatSendsWaitingForReconnect } from "./chat-queue.ts"; import { dismissRealtimeTalkError } from "./chat-realtime.ts"; import { activeChatRunStartupStatus } from "./chat-run-startup.ts"; -import { flushChatQueueForEvent, retryReconnectableQueuedChatSends } from "./chat-send.ts"; +import { flushChatQueueForEvent, retryReconnectableQueuedChatSends } from "./chat-send-actions.ts"; import { flushChatQueueAfterIdleSessionReconciliation, switchChatFastMode, diff --git a/ui/src/pages/chat/chat-send-ack.ts b/ui/src/pages/chat/chat-send-ack.ts new file mode 100644 index 000000000000..41648181e74a --- /dev/null +++ b/ui/src/pages/chat/chat-send-ack.ts @@ -0,0 +1,80 @@ +// Leaf contract for chat.send acknowledgment shapes and timing records. +// Kept import-free of chat-page modules so lifecycle/steer/history layers +// can consume ack types without forming import cycles. +import type { ChatQueueItem } from "../../lib/chat/chat-types.ts"; + +type ChatSendAckStatus = "started" | "in_flight" | "ok" | "timeout" | "error"; + +type ChatSendAckServerTiming = { + receivedToAckMs?: number; + loadSessionMs?: number; + prepareAttachmentsMs?: number; +}; + +export type ChatSendAck = { + runId: string; + status: ChatSendAckStatus; + serverTiming?: ChatSendAckServerTiming; +}; + +function normalizeAckTimingValue(value: unknown): number | undefined { + return typeof value === "number" && Number.isFinite(value) && value >= 0 ? value : undefined; +} + +function normalizeChatSendAckServerTiming(value: unknown): ChatSendAckServerTiming | undefined { + if (!value || typeof value !== "object") { + return undefined; + } + const record = value as Record; + 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; +} + +export function normalizeChatSendAck(payload: unknown, fallbackRunId: string): ChatSendAck { + if (!payload || typeof payload !== "object") { + return { runId: fallbackRunId, status: "started" }; + } + const record = payload as Record; + 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 type TerminalFailureChatSendAck = ChatSendAck & { status: "timeout" | "error" }; + +// ChatSendAck's status is a union field, not a discriminant across object +// types; callers need this predicate to narrow the whole ack object. +export function isTerminalFailureChatSendAck( + ack: ChatSendAck | null, +): ack is TerminalFailureChatSendAck { + return ack?.status === "timeout" || ack?.status === "error"; +} + +export type ChatSendTimingEntry = { + runId: string; + sessionKey?: string; + agentId?: string; + sendAttempts: number; + sendState?: ChatQueueItem["sendState"]; + submittedAtMs: number; + requestStartedAtMs?: number; + ackAtMs?: number; + ackStatus?: ChatSendAckStatus; + firstAssistantVisibleRecorded?: boolean; +}; diff --git a/ui/src/pages/chat/chat-send-actions.ts b/ui/src/pages/chat/chat-send-actions.ts new file mode 100644 index 000000000000..d009f49943d6 --- /dev/null +++ b/ui/src/pages/chat/chat-send-actions.ts @@ -0,0 +1,191 @@ +import type { QueueMode } from "../../../../src/auto-reply/reply/queue/types.js"; +import type { ChatAttachment } from "../../lib/chat/chat-types.ts"; +import { generateUUID } from "../../lib/uuid.ts"; +import { loadChatHistory, type ChatState } from "./chat-history.ts"; +import { + flushStoredChatOutbox, + resumeStoredChatOutboxes as resumeStoredChatOutboxesDrain, + scheduleStoredChatOutboxDrain, +} from "./chat-outbox-drain.ts"; +import { + admitQueuedMessageForSession, + isVolatileQueuedMessage, + updateQueuedMessage, + updateVolatileQueuedMessage, +} from "./chat-queue.ts"; +import type { ChatSendAck } from "./chat-send-ack.ts"; +import type { ChatHost } from "./chat-send-contract.ts"; +import { + canSendVolatileQueueItem, + reconnectSafeQueuedSendState, + setChatError, +} from "./chat-send-queue-state.ts"; +import { requestChatSend } from "./chat-send-request.ts"; +import { chatOutboxDrainDependencies, sendQueuedChatMessage } from "./chat-send.ts"; +import { listStoredChatOutboxes, storedChatOutboxScopeKey } from "./composer-persistence.ts"; +import { formatConnectError } from "./connect-error.ts"; +import { + OFFLINE_QUEUE_STORAGE_ERROR, + steerQueuedChatMessage as steerQueuedChatMessageLifecycle, + type SteerSendDependencies, +} from "./steer-lifecycle.ts"; +import { isInflightSteer } from "./steered-chip.ts"; + +export async function sendChatMessageWithGeneratedRunId( + state: ChatState, + message: string, + attachments?: ChatAttachment[], + options: { + canApplyError?: () => boolean; + queueMode?: QueueMode; + runId?: string; + } = {}, +): Promise { + if (!state.client || !state.connected) { + return null; + } + const msg = message.trim(); + const hasAttachments = attachments && attachments.length > 0; + if (!msg && !hasAttachments) { + return null; + } + const canApplyError = options.canApplyError ?? (() => true); + if (canApplyError()) { + setChatError(state, null); + } + const runId = options.runId ?? generateUUID(); + try { + return await requestChatSend(state, { + message: msg, + attachments, + runId, + ...(options.queueMode ? { queueMode: options.queueMode } : {}), + }); + } catch (err) { + if (canApplyError()) { + setChatError(state, formatConnectError(err)); + } + return null; + } +} + +export const steerSendDependencies: SteerSendDependencies = { + loadChatHistory: (host) => void loadChatHistory(host as unknown as ChatState), + resumeRestoredOutbox: (host, itemId) => { + const restoredOutbox = listStoredChatOutboxes(host).find((outbox) => + outbox.queue.some((item) => item.id === itemId), + ); + if (!host.chatRunId && restoredOutbox) { + void scheduleStoredChatOutboxDrain( + host as ChatHost, + restoredOutbox, + chatOutboxDrainDependencies, + ); + } + }, + sendChatMessage: (host, message, attachments, options) => + sendChatMessageWithGeneratedRunId(host as unknown as ChatState, message, attachments, options), +}; + +export function steerQueuedChatMessage(host: ChatHost, id: string) { + return steerQueuedChatMessageLifecycle(host, id, steerSendDependencies); +} + +export async function resumeStoredChatOutboxes(host: ChatHost) { + await resumeStoredChatOutboxesDrain(host, chatOutboxDrainDependencies); +} + +export async function flushChatQueueForEvent(host: ChatHost) { + await flushStoredChatOutbox(host, chatOutboxDrainDependencies); +} + +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" || + isInflightSteer(item) || + 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, chatOutboxDrainDependencies); + if (host.chatSending && host.chatSendingScopeKey === storedChatOutboxScopeKey(outbox)) { + void drain; + return; + } + await drain; + if (!host.chatRunId) { + void flushStoredChatOutbox(host, chatOutboxDrainDependencies); + } +} diff --git a/ui/src/pages/chat/chat-send-contract.ts b/ui/src/pages/chat/chat-send-contract.ts index 0ce50c3aa70f..d4414509bf13 100644 --- a/ui/src/pages/chat/chat-send-contract.ts +++ b/ui/src/pages/chat/chat-send-contract.ts @@ -1,38 +1,61 @@ -import type { ChatQueueItem } from "../../lib/chat/chat-types.ts"; +import type { GatewayBrowserClient, GatewayHelloOk } from "../../api/gateway.ts"; +import type { AgentsListResult } from "../../api/types.ts"; +import type { ChatFollowUpMode } from "../../app/settings.ts"; +import type { ChatAttachment, ChatQueueItem } from "../../lib/chat/chat-types.ts"; +import type { ControlUiFollowUpMode } from "../../lib/chat/follow-up-mode.ts"; +import type { ChatSideResultPending } from "../../lib/chat/side-result.ts"; +import type { SessionCapability, SessionRefreshTarget } from "../../lib/sessions/index.ts"; +import type { ChatCommandHost } from "./chat-commands.ts"; +import type { ChatRunStartupState } from "./chat-run-startup.ts"; +import type { ChatSendTimingEntry } from "./chat-send-ack.ts"; +import type { ChatInputHistoryState } from "./input-history.ts"; +import type { RenderLifecycle } from "./render-lifecycle.ts"; -type ChatSendAckStatus = "started" | "in_flight" | "ok" | "timeout" | "error"; - -export type ChatSendAckServerTiming = { - receivedToAckMs?: number; - loadSessionMs?: number; - prepareAttachmentsMs?: number; +type ChatAgentsListSnapshot = Partial> & { + agents?: AgentsListResult["agents"]; }; -export type ChatSendAck = { - runId: string; - status: ChatSendAckStatus; - serverTiming?: ChatSendAckServerTiming; -}; - -export type TerminalFailureChatSendAck = ChatSendAck & { status: "timeout" | "error" }; - -// ChatSendAck's status is a union field, not a discriminant across object -// types; callers need this predicate to narrow the whole ack object. -export function isTerminalFailureChatSendAck( - ack: ChatSendAck | null, -): ack is TerminalFailureChatSendAck { - return ack?.status === "timeout" || ack?.status === "error"; -} - -export type ChatSendTimingEntry = { - runId: string; - sessionKey?: string; - agentId?: string; - sendAttempts: number; - sendState?: ChatQueueItem["sendState"]; - submittedAtMs: number; - requestStartedAtMs?: number; - ackAtMs?: number; - ackStatus?: ChatSendAckStatus; - firstAssistantVisibleRecorded?: boolean; -}; +export type ChatHost = ChatInputHistoryState & + ChatCommandHost & { + sessions: SessionCapability; + client: GatewayBrowserClient | null; + chatStream: string | null; + connected: boolean; + connectionEpoch?: number; + chatAttachments: ChatAttachment[]; + chatQueue: ChatQueueItem[]; + chatQueueByScope?: Record; + chatRunId: string | null; + chatRunStartup?: ChatRunStartupState | null; + chatRunUsageById?: Map; + chatSending: boolean; + chatSendingScopeKey?: string | null; + chatRunError?: { summary: string } | null; + lastError?: string | null; + chatError?: string | null; + hello: GatewayHelloOk | null; + renderLifecycle?: RenderLifecycle; + requestUpdate?: () => void; + refreshSessionsAfterChat: Map; + chatSubmitGuards?: Map>; + chatSendTimingsByRun?: Map; + eventLogBuffer?: unknown[]; + assistantAgentId?: string | null; + agentsList?: ChatAgentsListSnapshot | null; + settings?: { chatFollowUpMode?: ChatFollowUpMode }; + /** Prepared from the browser override and current Gateway effective queue mode. */ + chatFollowUpMode?: ControlUiFollowUpMode; + /** Selected message to reply to (right-click / keyboard shortcut). */ + chatReplyTarget?: { + messageId: string; + text: string; + senderLabel?: string | null; + sourceMessageId?: 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; + /** Side-chat panel closed via X/Escape; a new question reopens it. */ + chatSideChatHidden?: boolean; + }; diff --git a/ui/src/pages/chat/chat-send-queue-state.ts b/ui/src/pages/chat/chat-send-queue-state.ts new file mode 100644 index 000000000000..0e963846dbf6 --- /dev/null +++ b/ui/src/pages/chat/chat-send-queue-state.ts @@ -0,0 +1,124 @@ +import type { ChatAttachment, ChatQueueItem } from "../../lib/chat/chat-types.ts"; +import { resolveCurrentUserIdentity } from "../../lib/chat/current-user-identity.ts"; +import { scopedAgentIdForSession, visibleSessionMatches } from "../../lib/sessions/index.ts"; +import { generateUUID } from "../../lib/uuid.ts"; +import type { QueuedChatStorageMode } from "./chat-outbox-drain.ts"; +import { updateQueuedMessageForSession, updateVolatileQueuedMessage } from "./chat-queue.ts"; +import type { ChatHost } from "./chat-send-contract.ts"; +import { recordChatSendTiming, schedulePendingSendPaintTiming } from "./chat-send-timing.ts"; +import { getPendingChatPickerPatch } from "./chat-session.ts"; +import { storedChatOutboxScopeKey, type StoredChatOutboxScope } from "./composer-persistence.ts"; +import { controlUiNowMs } from "./performance.ts"; +import { isChatBusy } from "./run-lifecycle.ts"; +import { scheduleChatScroll } from "./scroll.ts"; + +export function setChatError( + host: { lastError?: string | null; chatError?: string | null }, + error: string | null, +) { + host.lastError = error; + host.chatError = error; +} + +export function enqueuePendingSendMessage( + host: ChatHost, + text: string, + attachments?: ChatAttachment[], + refreshSessions?: boolean, + submittedAtMs = controlUiNowMs(), + sendState?: ChatQueueItem["sendState"], + skillWorkshopRevision?: ChatQueueItem["skillWorkshopRevision"], + replyToId?: string, +): ChatQueueItem | null { + const trimmed = text.trim(); + const hasAttachments = Boolean(attachments && attachments.length > 0); + if (!trimmed && !hasAttachments) { + return null; + } + const sender = resolveCurrentUserIdentity(host.hello, host.client?.instanceId); + 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), + ...(sender ? { sender } : {}), + ...(skillWorkshopRevision ? { skillWorkshopRevision } : {}), + ...(replyToId ? { replyToId } : {}), + }; + 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[0], true, false, { + source: "manual", + }); + return pending; +} + +export function reconnectSafeQueuedSendState( + host: Pick, +): "waiting-idle" | "waiting-reconnect" { + return host.connected && host.client ? "waiting-idle" : "waiting-reconnect"; +} + +export 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); +} + +export 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 + ); +} + +export function finishScopedChatSending(host: ChatHost, scope: StoredChatOutboxScope): void { + if (host.chatSendingScopeKey !== storedChatOutboxScopeKey(scope)) { + return; + } + host.chatSendingScopeKey = null; + host.chatSending = false; +} + +export async function waitForPendingChatSettings( + host: ChatHost, + sessionKey: string, + initialPending: Promise, + agentId?: string, +): Promise { + let pending = initialPending; + while (await pending) { + const nextPending = getPendingChatPickerPatch(host, sessionKey, agentId); + if (!nextPending || nextPending === pending) { + return true; + } + pending = nextPending; + } + return false; +} diff --git a/ui/src/pages/chat/chat-send-request.ts b/ui/src/pages/chat/chat-send-request.ts new file mode 100644 index 000000000000..d4099b6a7426 --- /dev/null +++ b/ui/src/pages/chat/chat-send-request.ts @@ -0,0 +1,101 @@ +import type { QueueMode } from "../../../../src/auto-reply/reply/queue/types.js"; +import type { ChatAttachment } from "../../lib/chat/chat-types.ts"; +import { + isUiGlobalSessionKey, + normalizeAgentId, + resolveUiSelectedSessionAgentId, +} from "../../lib/sessions/session-key.ts"; +import { buildChatApiAttachments } from "./attachment-api.ts"; +import type { ChatState } from "./chat-history.ts"; +import { normalizeChatSendAck, type ChatSendAck } from "./chat-send-ack.ts"; + +export async function requestChatSend( + state: ChatState, + params: { + message: string; + attachments?: ChatAttachment[]; + runId: string; + sessionKey?: string; + agentId?: string; + queueMode?: QueueMode; + replyToId?: string; + }, +): Promise { + 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, + ...(params.replyToId ? { replyToId: params.replyToId } : {}), + ...(params.queueMode ? { queueMode: params.queueMode } : {}), + idempotencyKey: params.runId, + attachments: buildChatApiAttachments(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 { + 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); +} diff --git a/ui/src/pages/chat/chat-send-submit.ts b/ui/src/pages/chat/chat-send-submit.ts new file mode 100644 index 000000000000..ecc50ccbc2dc --- /dev/null +++ b/ui/src/pages/chat/chat-send-submit.ts @@ -0,0 +1,648 @@ +import { shouldForwardModelCommandToServer } from "../../../../src/auto-reply/commands-registry.shared.js"; +import { normalizeChatFollowUpModeOverride, setLastActiveSessionKey } from "../../app/settings.ts"; +import type { + ChatAttachment, + ChatQueueItem, + ChatQueueSkillWorkshopRevision, +} from "../../lib/chat/chat-types.ts"; +import { parseSlashCommand } from "../../lib/chat/commands.ts"; +import { resolveCurrentUserIdentity } from "../../lib/chat/current-user-identity.ts"; +import { extractSideQuestionDisplayText } from "../../lib/chat/side-question.ts"; +import { retirePendingChatSideQuestion } from "../../lib/chat/side-result.ts"; +import { visibleSessionMatches } from "../../lib/sessions/index.ts"; +import { normalizeLowercaseStringOrEmpty } from "../../lib/string-coerce.ts"; +import { generateUUID } from "../../lib/uuid.ts"; +import { + getChatAttachmentDataUrl, + releaseChatAttachmentPayloads, +} from "./attachment-payload-store.ts"; +import { dispatchChatSlashCommand, shouldQueueLocalSlashCommand } from "./chat-commands.ts"; +import type { ChatState } from "./chat-history.ts"; +import { scheduleStoredChatOutboxDrain } from "./chat-outbox-drain.ts"; +import { + admitQueuedMessageForSession, + enqueueChatMessage, + excludeComposerAttachments, + removeQueuedMessageWithoutReleasing, + updateQueuedMessage, + updateQueuedMessageForSession, +} from "./chat-queue.ts"; +import { isTerminalFailureChatSendAck } from "./chat-send-ack.ts"; +import { sendChatMessageWithGeneratedRunId, steerSendDependencies } from "./chat-send-actions.ts"; +import type { ChatHost } from "./chat-send-contract.ts"; +import { + canSendVolatileQueueItem, + enqueuePendingSendMessage, + reconnectSafeQueuedSendState, + setChatError, + waitForPendingChatSettings, +} from "./chat-send-queue-state.ts"; +import { recordChatSendTiming } from "./chat-send-timing.ts"; +import { + cancelPendingSendBeforeRequest, + chatOutboxDrainDependencies, + pendingComposerRestorePlan, + sendChatMessageNow, + withChatSubmitGuard, +} from "./chat-send.ts"; +import { getPendingChatPickerPatch } from "./chat-session.ts"; +import { INTERRUPTED_SETTINGS_WAIT_ERROR, listStoredChatOutboxes } from "./composer-persistence.ts"; +import { + recordNonTranscriptInputHistory, + resetChatInputHistoryNavigation, +} from "./input-history.ts"; +import { controlUiNowMs } from "./performance.ts"; +import { + handleAbortChat, + hasAbortableSessionRun, + isChatBusy, + isChatStopCommand, +} from "./run-lifecycle.ts"; +import { + formatTerminalChatSendAckError, + OFFLINE_QUEUE_STORAGE_ERROR, + sendQueuedChatMessageWithQueueMode as sendQueuedChatMessageWithQueueModeLifecycle, +} from "./steer-lifecycle.ts"; + +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; + /** Lets request-scoped UI actions recover when their local slash command + * fails before the Gateway accepts it. */ + onLocalCommandSendRejected?: () => void; +}; + +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 isBtwCommand(text: string) { + return /^\/(?:btw|side)(?::|\s|$)/i.test(text.trim()); +} + +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), + ]); +} + +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 sendChatMessageWithGeneratedRunId( + host as unknown as ChatState, + message, + opts?.attachments, + { runId: opts?.runId }, + ); + const ok = ack?.status === "ok" || ack?.status === "started" || ack?.status === "in_flight"; + 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[0], + host.sessionKey, + ); + releaseChatAttachmentPayloads(excludeComposerAttachments(host, opts?.attachments)); + } + return ack; +} + +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 && + isChatResetCommand(message) && + (typeof globalThis.confirm !== "function" || + !globalThis.confirm("Start a new thread? This will reset the current chat.")) + ) { + return; + } + + host.chatRunError = null; + + if (shouldInterpretChatCommands) { + // Natural words such as "wait" and "exit" are stop aliases only while a + // run exists. Keep the explicit /stop command available at any time. + const shouldAbort = + isChatStopCommand(message) && + (message.trim().startsWith("/") || hasAbortableSessionRun(host)); + if (shouldAbort) { + 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 = + ack?.status === "ok" || ack?.status === "started" || ack?.status === "in_flight"; + // 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, + }, + resolveCurrentUserIdentity(host.hello, host.client?.instanceId) ?? undefined, + ); + 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, + chatOutboxDrainDependencies, + 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) => + chatOutboxDrainDependencies.sendResetSlashCommand(host, resetMessage, resetOpts), + }, + ); + if (dispatchResult === "failed") { + opts?.onLocalCommandSendRejected?.(); + } + if ( + (dispatchResult === "failed" || dispatchResult === "cancelled") && + 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; + // Persisted transcript ids ride chat.send as replyToId so the Gateway can + // hydrate reply context like Discord; synthetic ids fall back to a quote. + const replyToId = replyTarget?.sourceMessageId?.trim() || undefined; + const effectiveMessage = + replyTarget && !replyToId ? 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, + replyToId, + ); + 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, chatOutboxDrainDependencies); + } + return; + } + + let sendResult: "sent" | "pending" | "failed"; + if (isChatBusy(host) || hasAbortableSessionRun(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"; + // Inherited policy belongs to the Gateway: preserve steer, followup, + // collect, and interrupt semantics. Browser-local queueing only applies + // to an explicit browser override. + const followUpMode = + host.chatFollowUpMode ?? + normalizeChatFollowUpModeOverride(host.settings?.chatFollowUpMode); + if ( + !skillWorkshopRevision && + followUpMode !== "queue" && + host.connected && + hasAbortableSessionRun(host) + ) { + void sendQueuedChatMessageWithQueueModeLifecycle( + host, + pending.id, + followUpMode, + steerSendDependencies, + ); + } + } + } 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, +): 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"); +} diff --git a/ui/src/pages/chat/chat-send-timing.ts b/ui/src/pages/chat/chat-send-timing.ts index 6ef9980d1413..0087e241f5e5 100644 --- a/ui/src/pages/chat/chat-send-timing.ts +++ b/ui/src/pages/chat/chat-send-timing.ts @@ -1,7 +1,7 @@ import type { ChatQueueItem } from "../../lib/chat/chat-types.ts"; import { visibleSessionMatches, type SessionScopeHost } from "../../lib/sessions/index.ts"; import { readChatQueueForScope } from "./chat-queue.ts"; -import type { ChatSendAck, ChatSendTimingEntry } from "./chat-send-contract.ts"; +import type { ChatSendAck, ChatSendTimingEntry } from "./chat-send-ack.ts"; import { controlUiNowMs, recordControlUiPerformanceEvent, diff --git a/ui/src/pages/chat/chat-send.test.ts b/ui/src/pages/chat/chat-send.test.ts index f3f3815a42e6..97608b88853b 100644 --- a/ui/src/pages/chat/chat-send.test.ts +++ b/ui/src/pages/chat/chat-send.test.ts @@ -17,7 +17,7 @@ import { import { refreshChatAvatar } from "./chat-avatar.ts"; import * as chatCommandExecutor from "./chat-command-executor.ts"; import type { executeSlashCommand } from "./chat-command-executor.ts"; -import type { ChatHost } from "./chat-send.ts"; +import type { ChatHost } from "./chat-send-contract.ts"; import { getPendingChatPickerPatch, switchChatFastMode, @@ -138,8 +138,8 @@ afterEach(() => { vi.restoreAllMocks(); }); -let handleSendChat: typeof import("./chat-send.ts").handleSendChat; -let steerQueuedChatMessage: typeof import("./chat-send.ts").steerQueuedChatMessage; +let handleSendChat: typeof import("./chat-send-submit.ts").handleSendChat; +let steerQueuedChatMessage: typeof import("./chat-send-actions.ts").steerQueuedChatMessage; let handleAbortChat: typeof import("./run-lifecycle.ts").handleAbortChat; let hasAbortableSessionRun: typeof import("./run-lifecycle.ts").hasAbortableSessionRun; let handlePageGatewayEvent: typeof import("./chat-state.ts").handlePageGatewayEvent; @@ -152,20 +152,20 @@ let removeVisibleOrScopedQueuedMessageWithoutReleasing: typeof import("./chat-qu let markQueuedChatSendsWaitingForReconnect: typeof import("./chat-queue.ts").markQueuedChatSendsWaitingForReconnect; let subscribeChatOutboxProjection: typeof import("./chat-queue.ts").subscribeChatOutboxProjection; let syncChatQueueFromStoredOutbox: typeof import("./chat-queue.ts").syncChatQueueFromStoredOutbox; -let flushChatQueueForEvent: typeof import("./chat-send.ts").flushChatQueueForEvent; -let retryReconnectableQueuedChatSends: typeof import("./chat-send.ts").retryReconnectableQueuedChatSends; -let retryQueuedChatMessage: typeof import("./chat-send.ts").retryQueuedChatMessage; +let flushChatQueueForEvent: typeof import("./chat-send-actions.ts").flushChatQueueForEvent; +let retryReconnectableQueuedChatSends: typeof import("./chat-send-actions.ts").retryReconnectableQueuedChatSends; +let retryQueuedChatMessage: typeof import("./chat-send-actions.ts").retryQueuedChatMessage; let recordChatSendServerTiming: typeof import("./chat-send-timing.ts").recordChatSendServerTiming; let refreshPageChat: typeof import("./chat-state.ts").refreshPageChat; async function loadChatHelpers(): Promise { ({ - handleSendChat, steerQueuedChatMessage, flushChatQueueForEvent, retryReconnectableQueuedChatSends, retryQueuedChatMessage, - } = await import("./chat-send.ts")); + } = await import("./chat-send-actions.ts")); + ({ handleSendChat } = await import("./chat-send-submit.ts")); ({ recordChatSendServerTiming } = await import("./chat-send-timing.ts")); const chatState = await import("./chat-state.ts"); handlePageGatewayEvent = chatState.handlePageGatewayEvent; diff --git a/ui/src/pages/chat/chat-send.ts b/ui/src/pages/chat/chat-send.ts index ed38c0504953..4f8384eab3ab 100644 --- a/ui/src/pages/chat/chat-send.ts +++ b/ui/src/pages/chat/chat-send.ts @@ -1,84 +1,50 @@ // Control UI module implements app chat behavior. -import { shouldForwardModelCommandToServer } from "../../../../src/auto-reply/commands-registry.shared.js"; -import type { QueueMode } from "../../../../src/auto-reply/reply/queue/types.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 { - normalizeChatFollowUpModeOverride, - setLastActiveSessionKey, - type ChatFollowUpMode, -} from "../../app/settings.ts"; -import type { - ChatAttachment, - ChatQueueItem, - ChatQueueSkillWorkshopRevision, -} from "../../lib/chat/chat-types.ts"; -import { parseSlashCommand } from "../../lib/chat/commands.ts"; -import { resolveCurrentUserIdentity } from "../../lib/chat/current-user-identity.ts"; -import type { ControlUiFollowUpMode } from "../../lib/chat/follow-up-mode.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 { GatewayRequestError } from "../../api/gateway.ts"; +import { setLastActiveSessionKey } from "../../app/settings.ts"; +import type { ChatAttachment, ChatQueueItem } from "../../lib/chat/chat-types.ts"; +import { scopedAgentIdForSession, visibleSessionMatches } from "../../lib/sessions/index.ts"; import { generateUUID } from "../../lib/uuid.ts"; -import { buildChatApiAttachments } from "./attachment-api.ts"; import { discardChatAttachmentDataUrls, - getChatAttachmentDataUrl, releaseChatAttachmentPayloads, } from "./attachment-payload-store.ts"; +import type { ChatCommandResetOptions } from "./chat-commands.ts"; +import { loadChatHistory, type ChatState } from "./chat-history.ts"; import { - confirmConversationResetForCurrentSession, - dispatchChatSlashCommand, - type ChatCommandHost, - type ChatCommandResetOptions, - shouldQueueLocalSlashCommand, -} from "./chat-commands.ts"; -import { loadChatHistory, type ChatHistoryResult, type ChatState } from "./chat-history.ts"; + flushStoredChatOutbox, + retryableGatewayDelayMs, + scheduleStoredChatOutboxDrain, + scheduleStoredChatOutboxRetry, + UNCONFIRMED_CHAT_SEND_ERROR, + type ChatOutboxDrainDependencies, + type QueuedChatSendOptions, + type QueuedChatSendResult, + type QueuedChatStorageMode, +} from "./chat-outbox-drain.ts"; import { admitQueuedMessageForSession, - enqueueChatMessage, excludeComposerAttachments, - isVolatileQueuedMessage, readQueuedMessageById, removeQueuedMessageWithoutReleasing, removeVisibleOrScopedQueuedMessageWithoutReleasing, - syncChatQueueFromStoredOutbox, - updateQueuedMessage, updateQueuedMessageForSession, - updateVolatileQueuedMessage, } from "./chat-queue.ts"; -import type { ChatRunStartupState } from "./chat-run-startup.ts"; +import { isTerminalFailureChatSendAck } from "./chat-send-ack.ts"; +import type { ChatHost } from "./chat-send-contract.ts"; import { - isTerminalFailureChatSendAck, - type ChatSendAck, - type ChatSendAckServerTiming, - type ChatSendTimingEntry, -} from "./chat-send-contract.ts"; + enqueuePendingSendMessage, + finishScopedChatSending, + reconnectSafeQueuedSendState, + setChatError, + updateQueuedSendItem, + waitForPendingChatSettings, +} from "./chat-send-queue-state.ts"; +import { requestChatSend, requestSkillWorkshopRevisionChatSend } from "./chat-send-request.ts"; import { chatSendAckServerTimingEventFields, recordChatSendTiming, registerChatSendTiming, - schedulePendingSendPaintTiming, updateChatSendAckTiming, } from "./chat-send-timing.ts"; import { getPendingChatPickerPatch, refreshChatSessionListForTarget } from "./chat-session.ts"; @@ -86,353 +52,17 @@ import { INTERRUPTED_SETTINGS_WAIT_ERROR, listStoredChatOutboxes, storedChatOutboxScopeKey, - type StoredChatOutbox, type StoredChatOutboxScope, } from "./composer-persistence.ts"; import { formatConnectError } from "./connect-error.ts"; -import { - recordNonTranscriptInputHistory, - resetChatInputHistoryNavigation, - type ChatInputHistoryState, -} from "./input-history.ts"; +import { resetChatInputHistoryNavigation } from "./input-history.ts"; import { controlUiNowMs, roundedControlUiDurationMs } from "./performance.ts"; -import type { RenderLifecycle } from "./render-lifecycle.ts"; -import { - handleAbortChat, - hasAbortableSessionRun, - isChatBusy, - isChatStopCommand, - reconcileChatRunLifecycle, -} from "./run-lifecycle.ts"; +import { reconcileChatRunLifecycle } from "./run-lifecycle.ts"; import { scheduleChatScroll, resetChatScroll } from "./scroll.ts"; -import { - formatTerminalChatSendAckError, - chatMessagesContainQueuedSend, - OFFLINE_QUEUE_STORAGE_ERROR, - preserveQueuedUserTurn, - sendQueuedChatMessageWithQueueMode as sendQueuedChatMessageWithQueueModeLifecycle, - steerQueuedChatMessage as steerQueuedChatMessageLifecycle, - type SteerSendDependencies, -} from "./steer-lifecycle.ts"; -import { isInflightSteer } from "./steered-chip.ts"; +import { formatTerminalChatSendAckError, OFFLINE_QUEUE_STORAGE_ERROR } from "./steer-lifecycle.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; - chatRunId: string | null; - chatRunStartup?: ChatRunStartupState | null; - chatRunUsageById?: Map; - chatSending: boolean; - chatSendingScopeKey?: string | null; - chatRunError?: { summary: string } | null; - lastError?: string | null; - chatError?: string | null; - hello: GatewayHelloOk | null; - renderLifecycle?: RenderLifecycle; - requestUpdate?: () => void; - refreshSessionsAfterChat: Map; - chatSubmitGuards?: Map>; - chatSendTimingsByRun?: Map; - eventLogBuffer?: unknown[]; - assistantAgentId?: string | null; - agentsList?: ChatAgentsListSnapshot | null; - settings?: { chatFollowUpMode?: ChatFollowUpMode }; - /** Prepared from the browser override and current Gateway effective queue mode. */ - chatFollowUpMode?: ControlUiFollowUpMode; - /** Selected message to reply to (right-click / keyboard shortcut). */ - chatReplyTarget?: { - messageId: string; - text: string; - senderLabel?: string | null; - sourceMessageId?: 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; - /** Side-chat panel closed via X/Escape; a new question reopens it. */ - chatSideChatHidden?: boolean; - }; - -type ChatAgentsListSnapshot = Partial> & { - 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 { - return sendChatMessageNow(host, message, { - refreshSessions: true, - previousDraft: opts.previousDraft, - restoreDraft: opts.restoreDraft, - routingSessionKey: host.sessionKey, - }).then(() => undefined); -} - -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; - /** Lets request-scoped UI actions recover when their local slash command - * fails before the Gateway accepts it. */ - onLocalCommandSendRejected?: () => void; -}; - -function normalizeAckTimingValue(value: unknown): number | undefined { - return typeof value === "number" && Number.isFinite(value) && value >= 0 ? value : undefined; -} - -function normalizeChatSendAckServerTiming(value: unknown): ChatSendAckServerTiming | undefined { - if (!value || typeof value !== "object") { - return undefined; - } - const record = value as Record; - 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; - 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 } : {}), - }; -} - -async function requestChatSend( - state: ChatState, - params: { - message: string; - attachments?: ChatAttachment[]; - runId: string; - sessionKey?: string; - agentId?: string; - queueMode?: QueueMode; - replyToId?: string; - }, -): Promise { - 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, - ...(params.replyToId ? { replyToId: params.replyToId } : {}), - ...(params.queueMode ? { queueMode: params.queueMode } : {}), - idempotencyKey: params.runId, - attachments: buildChatApiAttachments(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 } : {}), - }; -} - -async function requestSkillWorkshopRevisionChatSend( - state: ChatState, - params: { - proposalId: string; - instructions: string; - runId: string; - sessionKey?: string; - agentId?: string; - targetAgentId?: string; - }, -): Promise { - 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); -} - -async function sendChatMessageWithGeneratedRunId( - state: ChatState, - message: string, - attachments?: ChatAttachment[], - options: { - canApplyError?: () => boolean; - queueMode?: QueueMode; - runId?: string; - } = {}, -): Promise { - if (!state.client || !state.connected) { - return null; - } - const msg = message.trim(); - const hasAttachments = attachments && attachments.length > 0; - if (!msg && !hasAttachments) { - return null; - } - const canApplyError = options.canApplyError ?? (() => true); - if (canApplyError()) { - setChatError(state, null); - } - const runId = options.runId ?? generateUUID(); - try { - return await requestChatSend(state, { - message: msg, - attachments, - runId, - ...(options.queueMode ? { queueMode: options.queueMode } : {}), - }); - } catch (err) { - if (canApplyError()) { - setChatError(state, formatConnectError(err)); - } - return null; - } -} - -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 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, - replyToId?: string, -): ChatQueueItem | null { - const trimmed = text.trim(); - const hasAttachments = Boolean(attachments && attachments.length > 0); - if (!trimmed && !hasAttachments) { - return null; - } - const sender = resolveCurrentUserIdentity(host.hello, host.client?.instanceId); - 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), - ...(sender ? { sender } : {}), - ...(skillWorkshopRevision ? { skillWorkshopRevision } : {}), - ...(replyToId ? { replyToId } : {}), - }; - 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[0], true, false, { - source: "manual", - }); - return pending; -} - function restoreComposerAfterFailedSend( host: ChatHost, opts: { @@ -453,7 +83,7 @@ type PendingComposerSnapshot = { previousDraft?: string; }; -function pendingComposerRestorePlan(host: ChatHost, snapshot: PendingComposerSnapshot) { +export function pendingComposerRestorePlan(host: ChatHost, snapshot: PendingComposerSnapshot) { const willRestoreDraft = snapshot.previousDraft != null && !host.chatMessage.trim(); const willRestoreAttachments = Boolean( snapshot.previousAttachments?.length && @@ -469,7 +99,7 @@ function pendingComposerRestorePlan(host: ChatHost, snapshot: PendingComposerSna }; } -function cancelPendingSendBeforeRequest( +export function cancelPendingSendBeforeRequest( host: ChatHost, queued: ChatQueueItem, opts: PendingComposerSnapshot & { @@ -498,74 +128,19 @@ function cancelPendingSendBeforeRequest( } } -type QueuedChatSendResult = "sent" | "pending" | "failed"; -type QueuedChatStorageMode = "durable" | "memory"; -type QueuedChatSendOptions = { - previousAttachments?: ChatAttachment[]; - previousDraft?: string; - routingSessionKey?: string; - storageMode?: QueuedChatStorageMode; +export const chatOutboxDrainDependencies: ChatOutboxDrainDependencies = { + sendQueuedChatMessage, + sendResetSlashCommand: (host, message, opts: ChatCommandResetOptions) => + sendChatMessageNow(host, message, { + refreshSessions: true, + previousDraft: opts.previousDraft, + restoreDraft: opts.restoreDraft, + routingSessionKey: host.sessionKey, + }).then(() => undefined), + setChatError, }; -function reconnectSafeQueuedSendState( - host: Pick, -): "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 UNCONFIRMED_CHAT_SEND_ERROR = - "Delivery could not be confirmed after reconnect. Check the conversation 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 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); -} - -async function sendQueuedChatMessage( +export async function sendQueuedChatMessage( host: ChatHost, id: string, opts?: QueuedChatSendOptions, @@ -985,6 +560,7 @@ async function sendQueuedChatMessage( host, { sessionKey, agentId: prepared.agentId }, retryDelayMs, + chatOutboxDrainDependencies, ); } recordChatSendTiming(host, prepared, "waiting-reconnect", prepared.sendSubmittedAtMs, { @@ -1008,7 +584,7 @@ async function sendQueuedChatMessage( } } -async function sendChatMessageNow( +export async function sendChatMessageNow( host: ChatHost, message: string, opts?: { @@ -1069,11 +645,17 @@ async function sendChatMessageNow( setChatError(host, OFFLINE_QUEUE_STORAGE_ERROR); return "pending"; } - await scheduleStoredChatOutboxDrain(host, queuedOutbox, queued.id, { - previousDraft: opts?.previousDraft, - previousAttachments: opts?.previousAttachments, - routingSessionKey: opts?.routingSessionKey ?? queuedSessionKey, - }); + await scheduleStoredChatOutboxDrain( + host, + queuedOutbox, + chatOutboxDrainDependencies, + 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); @@ -1108,41 +690,12 @@ async function sendChatMessageNow( scheduleChatScroll(host as unknown as Parameters[0], true); } if (sent && host.sessionKey === queuedSessionKey && !host.chatRunId) { - void flushChatQueue(host); + void flushStoredChatOutbox(host, chatOutboxDrainDependencies); } 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( +export async function withChatSubmitGuard( host: ChatHost, key: string, run: () => Promise, @@ -1165,1175 +718,3 @@ async function withChatSubmitGuard( } } } - -async function waitForPendingChatSettings( - host: ChatHost, - sessionKey: string, - initialPending: Promise, - agentId?: string, -): Promise { - 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 sendChatMessageWithGeneratedRunId( - host as unknown as ChatState, - message, - opts?.attachments, - { runId: opts?.runId }, - ); - const ok = ack?.status === "ok" || ack?.status === "started" || ack?.status === "in_flight"; - 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[0], - host.sessionKey, - ); - releaseChatAttachmentPayloads(excludeComposerAttachments(host, opts?.attachments)); - } - return ack; -} - -const steerSendDependencies: SteerSendDependencies = { - loadChatHistory: (host) => void loadChatHistory(host as unknown as ChatState), - resumeRestoredOutbox: (host, itemId) => { - const restoredOutbox = listStoredChatOutboxes(host).find((outbox) => - outbox.queue.some((item) => item.id === itemId), - ); - if (!host.chatRunId && restoredOutbox) { - void scheduleStoredChatOutboxDrain(host as ChatHost, restoredOutbox); - } - }, - sendChatMessage: (host, message, attachments, options) => - sendChatMessageWithGeneratedRunId(host as unknown as ChatState, message, attachments, options), -}; - -export function steerQueuedChatMessage(host: ChatHost, id: string) { - return steerQueuedChatMessageLifecycle(host, id, steerSendDependencies); -} - -type StoredChatOutboxDrainResult = "blocked" | "empty"; -type StoredChatOutboxDrainLane = { - freshAdmissions: Set; - host: ChatHost; - pendingOptions: Map; - promise: Promise; - rerun: boolean; -}; - -const storedChatOutboxDrainLanesByClient = new WeakMap< - GatewayBrowserClient, - Map ->(); -const storedChatOutboxRetryTimersByClient = new WeakMap< - GatewayBrowserClient, - Map> ->(); - -function storedChatOutboxClientMap( - store: WeakMap>, - client: GatewayBrowserClient, -): Map { - const existing = store.get(client); - if (existing) { - return existing; - } - const created = new Map(); - store.set(client, created); - return created; -} - -function scheduleStoredChatOutboxRetry( - host: ChatHost, - scope: StoredChatOutboxScope, - delayMs: number, -) { - const client = host.client; - if (!host.connected || !client) { - return; - } - const connectionEpoch = host.connectionEpoch; - const timers = storedChatOutboxClientMap(storedChatOutboxRetryTimersByClient, 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 readStoredChatOutbox( - host: ChatHost, - scope: StoredChatOutboxScope, -): StoredChatOutbox | undefined { - return listStoredChatOutboxes(host).find( - (outbox) => outbox.sessionKey === scope.sessionKey && outbox.agentId === scope.agentId, - ); -} - -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 - ); -} - -async function readCurrentStoredChatHistory( - host: ChatHost, - outbox: StoredChatOutbox, - item: ChatQueueItem, - client: NonNullable, - connectionEpoch: number | undefined, -): Promise { - let history: ChatHistoryResult; - try { - history = await client.request("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 (chatMessagesContainQueuedSend(history.messages, item)) { - // Server history owns the turn, but the visible transcript may not have - // reloaded yet; materialize the turn locally before dropping the queue row - // or the bubble vanishes until loadChatHistory below resolves. - preserveQueuedUserTurn(host, item); - const removed = removeQueuedMessageWithoutReleasing(host, item.id, outbox.sessionKey); - if (!removed) { - return "blocked"; - } - releaseChatAttachmentPayloads(excludeComposerAttachments(host, removed.attachments)); - if (visibleSessionMatches(host, outbox.sessionKey, outbox.agentId)) { - void loadChatHistory(host as unknown as ChatState); - } - return "continue"; - } - if ( - !history.sessionInfo || - history.sessionInfo.hasActiveRun === true || - isSessionRunActive(history.sessionInfo) - ) { - return "blocked"; - } - return history; -} - -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"; - } - const history = await readCurrentStoredChatHistory(host, outbox, item, client, connectionEpoch); - if (history === "blocked" || history === "continue") { - return history; - } - if (visibleSessionMatches(host, outbox.sessionKey, outbox.agentId) && isChatBusy(host)) { - 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. - const verifiedHistory = await readCurrentStoredChatHistory( - host, - outbox, - item, - client, - connectionEpoch, - ); - if (verifiedHistory === "blocked" || verifiedHistory === "continue") { - return verifiedHistory; - } - 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 { - while (true) { - const host = lane.host; - if (!host.connected || !host.client) { - return "blocked"; - } - const outbox = readStoredChatOutbox(host, scope); - if (!outbox) { - return "empty"; - } - // Failed non-command rows are skipped; a failed command may have changed - // session state before reporting an error, so it blocks the drain and - // preserves FIFO until the user explicitly retries or removes it. - let item: ChatQueueItem | undefined; - for (const entry of outbox.queue) { - if (entry.sendState !== "failed") { - item = entry; - break; - } - if (entry.localCommandName) { - break; - } - } - if (!item) { - return "empty"; - } - if (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 convertResetToMessage = (sendState?: ChatQueueItem["sendState"]) => - updateQueuedMessageForSession(host, outbox.sessionKey, item.id, (entry) => ({ - ...entry, - localCommandArgs: undefined, - localCommandName: undefined, - refreshSessions: true, - text: resetText, - ...(sendState ? { sendState } : {}), - })); - const confirmation = await confirmConversationResetForCurrentSession(host, { - sessionKey: outbox.sessionKey, - ...(outbox.agentId ? { agentId: outbox.agentId } : {}), - }); - if (confirmation === "deferred") { - const approvedDuringRun = - visibleSessionMatches(host, outbox.sessionKey, outbox.agentId) && host.chatRunId; - const deferred = approvedDuringRun - ? convertResetToMessage("waiting-idle") - : updateQueuedMessageForSession(host, outbox.sessionKey, item.id, (entry) => ({ - ...entry, - sendError: undefined, - sendState: "waiting-idle", - })); - if (!deferred) { - return "blocked"; - } - return "blocked"; - } - if (confirmation === "cancelled") { - if (!removeQueuedMessageWithoutReleasing(host, item.id, outbox.sessionKey)) { - return "blocked"; - } - continue; - } - const converted = convertResetToMessage(); - 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 === "deferred") { - updateQueuedMessageForSession(host, outbox.sessionKey, item.id, (entry) => ({ - ...entry, - sendError: undefined, - sendState: "waiting-idle", - })); - return "blocked"; - } - 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 { - const client = host.client; - if (!host.connected || !client) { - return; - } - const key = storedChatOutboxScopeKey(scope); - const retryTimers = storedChatOutboxRetryTimersByClient.get(client); - const retryTimer = retryTimers?.get(key); - if (retryTimer !== undefined) { - clearTimeout(retryTimer); - retryTimers?.delete(key); - } - // 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 = storedChatOutboxClientMap(storedChatOutboxDrainLanesByClient, client); - 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((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" || - isInflightSteer(item) || - 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 && - isChatResetCommand(message) && - (typeof globalThis.confirm !== "function" || - !globalThis.confirm("Start a new thread? This will reset the current chat.")) - ) { - return; - } - - host.chatRunError = null; - - if (shouldInterpretChatCommands) { - // Natural words such as "wait" and "exit" are stop aliases only while a - // run exists. Keep the explicit /stop command available at any time. - const shouldAbort = - isChatStopCommand(message) && - (message.trim().startsWith("/") || hasAbortableSessionRun(host)); - if (shouldAbort) { - 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 = - ack?.status === "ok" || ack?.status === "started" || ack?.status === "in_flight"; - // 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, - }, - resolveCurrentUserIdentity(host.hello, host.client?.instanceId) ?? undefined, - ); - 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") { - opts?.onLocalCommandSendRejected?.(); - } - if ( - (dispatchResult === "failed" || dispatchResult === "cancelled") && - 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; - // Persisted transcript ids ride chat.send as replyToId so the Gateway can - // hydrate reply context like Discord; synthetic ids fall back to a quote. - const replyToId = replyTarget?.sourceMessageId?.trim() || undefined; - const effectiveMessage = - replyTarget && !replyToId ? 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, - replyToId, - ); - 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) || hasAbortableSessionRun(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"; - // Inherited policy belongs to the Gateway: preserve steer, followup, - // collect, and interrupt semantics. Browser-local queueing only applies - // to an explicit browser override. - const followUpMode = - host.chatFollowUpMode ?? - normalizeChatFollowUpModeOverride(host.settings?.chatFollowUpMode); - if ( - !skillWorkshopRevision && - followUpMode !== "queue" && - host.connected && - hasAbortableSessionRun(host) - ) { - void sendQueuedChatMessageWithQueueModeLifecycle( - host, - pending.id, - followUpMode, - steerSendDependencies, - ); - } - } - } 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, -): 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; -/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ diff --git a/ui/src/pages/chat/chat-state.ts b/ui/src/pages/chat/chat-state.ts index f96e7ada4c53..6a454f2559e0 100644 --- a/ui/src/pages/chat/chat-state.ts +++ b/ui/src/pages/chat/chat-state.ts @@ -77,16 +77,16 @@ import { resetChatRealtimeConversation, type ChatRealtimeState, } from "./chat-realtime.ts"; -import type { ChatSendTimingEntry } from "./chat-send-contract.ts"; -import { recordChatSendServerTiming } from "./chat-send-timing.ts"; +import type { ChatSendTimingEntry } from "./chat-send-ack.ts"; import { flushChatQueueForEvent, - handleSendChat, resumeStoredChatOutboxes, retryQueuedChatMessage, steerQueuedChatMessage, - type ChatHost, -} from "./chat-send.ts"; +} from "./chat-send-actions.ts"; +import type { ChatHost } from "./chat-send-contract.ts"; +import { handleSendChat } from "./chat-send-submit.ts"; +import { recordChatSendServerTiming } from "./chat-send-timing.ts"; import { flushChatQueueAfterIdleSessionReconciliation, refreshCurrentChatSessionList, diff --git a/ui/src/pages/chat/steer-lifecycle.ts b/ui/src/pages/chat/steer-lifecycle.ts index db92354ae055..6982acee1a56 100644 --- a/ui/src/pages/chat/steer-lifecycle.ts +++ b/ui/src/pages/chat/steer-lifecycle.ts @@ -22,7 +22,7 @@ import { isTerminalFailureChatSendAck, type ChatSendAck, type TerminalFailureChatSendAck, -} from "./chat-send-contract.ts"; +} from "./chat-send-ack.ts"; import { hasAbortableSessionRun } from "./run-lifecycle.ts"; import { scheduleChatScroll } from "./scroll.ts"; import {