fix(ai-sessions): mirror runs on non-secure origins, fix resync frame start

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

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)
+	})
+})

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
+}

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.

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<T>(
 	sessionId: string,
 	body: () => Promise<T>
 ): Promise<T | 'busy'> {
-	// 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<T>(
  *  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<boolean> {
+	// 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<MirrorMsg, 'kind'>
 
-/** 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 })