diff --git a/frontend/src/lib/components/sessions/sessionMirrorFrame.test.ts b/frontend/src/lib/components/sessions/sessionMirrorFrame.test.ts new file mode 100644 index 0000000000..927f0069ab --- /dev/null +++ b/frontend/src/lib/components/sessions/sessionMirrorFrame.test.ts @@ -0,0 +1,47 @@ +import { describe, expect, it } from 'vitest' +import { canSpliceFrame, mirrorFrameStart } from './sessionMirrorPayload' + +describe('mirrorFrameStart', () => { + // A resync exists to rescue a receiver whose prefix does not fit. Answering it + // with a frame that still starts mid-transcript fails the same check again and + // asks for another one, so the two tabs trade messages and nothing renders. + it('sends the whole transcript when full, even mid-turn', () => { + expect(mirrorFrameStart({ total: 40, turnStart: 36, full: true })).toBe(0) + }) + + it('reaches no further back than the running turn', () => { + expect(mirrorFrameStart({ total: 40, turnStart: 36, full: false })).toBe(36) + }) + + it('caps the tail when the turn itself is long', () => { + expect(mirrorFrameStart({ total: 40, turnStart: 2, full: false })).toBe(30) + }) + + it('handles a turn that has produced nothing yet', () => { + expect(mirrorFrameStart({ total: 0, turnStart: 0, full: false })).toBe(0) + }) +}) + +describe('canSpliceFrame', () => { + const base = { baseIndex: 30, total: 40, localLength: 35, onSameChat: true } + + it('accepts a tail that lands on a matching prefix', () => { + expect(canSpliceFrame(base)).toBe(true) + }) + + it('always accepts a full frame', () => { + expect(canSpliceFrame({ ...base, baseIndex: 0, onSameChat: false })).toBe(true) + }) + + it('rejects a tail when the receiver joined mid-run and has a gap', () => { + expect(canSpliceFrame({ ...base, localLength: 12 })).toBe(false) + }) + + it('rejects a tail when the receiver holds more than the sender has', () => { + expect(canSpliceFrame({ ...base, localLength: 44 })).toBe(false) + }) + + it('rejects a tail from a different conversation', () => { + expect(canSpliceFrame({ ...base, onSameChat: false })).toBe(false) + }) +}) diff --git a/frontend/src/lib/components/sessions/sessionMirrorPayload.ts b/frontend/src/lib/components/sessions/sessionMirrorPayload.ts index 8276223aed..12c2d52950 100644 --- a/frontend/src/lib/components/sessions/sessionMirrorPayload.ts +++ b/frontend/src/lib/components/sessions/sessionMirrorPayload.ts @@ -49,3 +49,62 @@ export function withoutHeavyPayloads(messages: DisplayMessage[]): DisplayMessage return stripped as DisplayMessage }) } + +/** How many of the newest messages a frame carries when it is not sending the + * whole transcript. In-place edits to already-rendered cards (a tool card + * settling) land within a few messages of the end, so a short tail carries them + * while keeping the frame bounded on a long conversation. */ +const MIRROR_TAIL = 10 + +/** + * Where the tail a frame carries starts. + * + * `full` sends the whole transcript and overrides everything else — it is what + * answers a resync, and a resync that came back partial would fail the receiver's + * prefix check again and ask for another one, forever. + * + * Otherwise the frame reaches no further back than `turnStart`, the index of the + * running turn's first message: everything from there on is either new this turn + * or a stripped copy the receiver already got from an earlier frame of the same + * turn, so overwriting it can never destroy a message held complete from the + * store. + */ +export function mirrorFrameStart({ + total, + turnStart, + full +}: { + total: number + turnStart: number + full: boolean +}): number { + if (full) return 0 + return Math.max(turnStart, Math.max(0, total - MIRROR_TAIL)) +} + +/** + * Whether a receiver can splice this frame onto what it already holds, or has to + * ask for the whole transcript instead. + * + * A frame is positional, so it is only meaningful against the same conversation + * and a prefix of the same shape. A receiver holding fewer messages than the + * frame starts at has a gap; one holding more than the sender has messages the + * sender no longer does (it compacted, or switched chats), and splicing would + * render a conversation that never existed. + */ +export function canSpliceFrame({ + baseIndex, + total, + localLength, + onSameChat +}: { + baseIndex: number + total: number + localLength: number + onSameChat: boolean +}): boolean { + // A frame that starts at 0 replaces everything, so it needs no prefix to + // agree with and can be adopted even when it names a different conversation. + if (baseIndex === 0) return true + return onSameChat && localLength >= baseIndex && localLength <= total +} diff --git a/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts b/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts index 2264ac9368..6a59643a2c 100644 --- a/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionRuntime.svelte.ts @@ -93,15 +93,13 @@ import { getNonStreamingMetadataCompletion } from '$lib/components/copilot/lib' import { sendUserToast } from '$lib/toast' import { pendingUserAction, type DisplayMessage } from '$lib/components/copilot/chat/shared' import type { ChatCompletionMessageParam } from 'openai/resources/index.mjs' -import { withoutHeavyPayloads } from './sessionMirrorPayload' +import { canSpliceFrame, mirrorFrameStart, withoutHeavyPayloads } from './sessionMirrorPayload' import { broadcastMirror, broadcastTurnEnd, isLocallyDriven, isRemotelyDriven, - RUN_OWNERSHIP_AVAILABLE, MIRROR_THROTTLE_MS, - mirrorBaseIndex, registerSyncHandlers, requestCancel, requestResync, @@ -938,10 +936,6 @@ async function initRuntime(runtime: SessionRuntime, session: Session) { // turns interleave into one chat id. manager.runGuard = async (body) => { const outcome = await withSessionRunLock(session.id, async () => { - // The whole turn protocol rides on ownership, not just the frames: a - // turn-end announced without it would reach tabs that were never - // mirroring and end a run of their own that is still going. - if (!RUN_OWNERSHIP_AVAILABLE) return body() // The first frame doubles as the "a run started here" signal: it is // posted immediately and carries the chat id the watchers need. startMirroring(session.id) @@ -1042,10 +1036,11 @@ function mirrorSnapshotOf(sessionId: string, full: boolean): MirrorSnapshot | un // Slice before cloning: `$state.snapshot` walks whatever it is handed, so // snapshotting the whole transcript to send ten messages would traverse the // entire conversation on every tick. - const baseIndex = Math.max( - turnStarts.get(sessionId) ?? 0, - mirrorBaseIndex(total, full || rewritten) - ) + const baseIndex = mirrorFrameStart({ + total, + turnStart: turnStarts.get(sessionId) ?? 0, + full: full || rewritten + }) return { sessionId, chatId: m.historyManager.getCurrentChatId(), @@ -1069,7 +1064,6 @@ function postMirror(sessionId: string, { full = false } = {}): void { } function startMirroring(sessionId: string): void { - if (!RUN_OWNERSHIP_AVAILABLE) return stopMirroring(sessionId) // Captured before the turn pushes its first message, so it is the index of // the oldest message this turn owns. @@ -1102,9 +1096,14 @@ function applyMirror(msg: MirrorMsg): void { if (!runtime) return const m = runtime.manager const onSameChat = m.historyManager.getCurrentChatId() === msg.chatId - const prefixFits = - m.displayMessages.length >= msg.baseIndex && m.displayMessages.length <= msg.total - if (msg.baseIndex > 0 && !(onSameChat && prefixFits)) { + if ( + !canSpliceFrame({ + baseIndex: msg.baseIndex, + total: msg.total, + localLength: m.displayMessages.length, + onSameChat + }) + ) { // Nothing here can host this tail: this tab joined mid-run, or the driver // rotated to a chat it isn't on. Ask for the whole transcript instead of // splicing onto a prefix that isn't the same conversation. diff --git a/frontend/src/lib/components/sessions/sessionSync.svelte.ts b/frontend/src/lib/components/sessions/sessionSync.svelte.ts index 75c0838d30..af9e4a910e 100644 --- a/frontend/src/lib/components/sessions/sessionSync.svelte.ts +++ b/frontend/src/lib/components/sessions/sessionSync.svelte.ts @@ -18,19 +18,11 @@ import type { DisplayMessage } from '$lib/components/copilot/chat/shared' const CHANNEL_BASE = 'windmill-sessions-sync' -/** Web Locks is secure-context only; BroadcastChannel is not. Without it there - * is no way to retire a driver that closed mid-turn, and a watcher would sit on - * "generating" forever with a Stop that reaches nobody — worse than not - * mirroring at all. So run mirroring rides on this and record sync, which needs - * no ownership, does not. */ -export const RUN_OWNERSHIP_AVAILABLE = BROWSER && !!globalThis.navigator?.locks?.query - -/** Tail length of the transcript sent on each mirror tick. In-place edits to - * already-rendered messages (tool cards settling) land within a few messages - * of the end, so a short tail carries them while keeping the payload bounded - * on long conversations. Anything older is corrected by the IndexedDB re-read - * at turn end. */ -const MIRROR_TAIL = 10 +/** Whether this context can hold a real lock on a run. Web Locks is + * secure-context only, so a self-hosted instance served over plain HTTP has + * none — mirroring still runs there, on the weaker footing described at + * {@link withSessionRunLock} and {@link runLockHeld}. */ +const EXCLUSIVE_OWNERSHIP = BROWSER && !!globalThis.navigator?.locks?.query /** Mirror ticks are throttled to this while a turn streams. Also the heartbeat * interval: an unchanged run still ticks, so silence means the driver is gone. */ @@ -274,10 +266,20 @@ export async function withSessionRunLock( sessionId: string, body: () => Promise ): Promise { - // No ownership to take: the run proceeds unguarded, and the caller mirrors - // nothing (see RUN_OWNERSHIP_AVAILABLE), leaving tabs as independent as they - // were before any of this. - if (!RUN_OWNERSHIP_AVAILABLE) return body() + if (!EXCLUSIVE_OWNERSHIP) { + // No lock to take, so exclusion is best-effort: refuse only when another + // tab is visibly mid-run. That covers the case that actually happens (one + // tab already going) and leaves a genuine simultaneous start racing, which + // is what a session already did before any of this. Marking the run ours + // is what keeps the two tabs from adopting each other's frames. + if (isRemotelyDriven(sessionId)) return 'busy' + locallyDriven.add(sessionId) + try { + return await body() + } finally { + locallyDriven.delete(sessionId) + } + } return (await navigator.locks.request( lockName(sessionId), { mode: 'exclusive', ifAvailable: true }, @@ -300,6 +302,12 @@ export async function withSessionRunLock( * settle a driver that stopped mirroring: a released lock proves the tab is * gone, where silence alone only suggests it. */ async function runLockHeld(sessionId: string): Promise { + // Nothing to consult without the lock API, so silence is the only evidence — + // and the caller only asks after a driver has gone quiet for the whole + // silence window. Reaping a driver that was merely starved is self-correcting: + // the watcher stops showing a run that is not visibly progressing and re-reads + // the record, which the next frame or the turn-end would have done anyway. + if (!EXCLUSIVE_OWNERSHIP) return false try { const state = await navigator.locks.query() const name = lockName(sessionId) @@ -351,15 +359,6 @@ export function sendQuestionAnswer(sessionId: string, toolId: string, choices: s export type MirrorSnapshot = Omit -/** Where the tail a frame carries should start. Callers slice — and only then - * clone — so a heartbeat on a long conversation never copies the messages it - * is not going to send; a transcript holding pasted files or image data URLs - * makes that difference megabytes per tick. `full` sends everything, which is - * what answers a resync request. */ -export function mirrorBaseIndex(total: number, full = false): number { - return full ? 0 : Math.max(0, total - MIRROR_TAIL) -} - /** Send the driver's current view of a run. */ export function broadcastMirror(snap: MirrorSnapshot): void { post({ kind: 'mirror', ...snap })