fix(chat): guard every write-back behind the re-point, and order the conversation locks

The streaming path had the same unguarded gap the settle path was fixed for: its
final poll is awaited and then `endTurn(settled)` runs regardless. Settling
releases the conversation's queue, so a message typed before a re-point was
started against the flow now loaded, writing it into the previous flow's
conversation and agent memory — the thing the forgetting exists to prevent.

The poller and the sidebar's loader were unguarded too. Both are guarded at their
own await now, which covers every caller rather than each call site.

The editor chats keyed off `$initialPathStore || $pathStore`, which still falls
back to the typed path on a flow that was never deployed — and there the summary
field rewrites the path on every keystroke, so naming a new flow emptied the
composer as you typed. They take the editor's stable identity instead, which is
also what a preview run records.

`FOR UPDATE` over several rows takes them in whatever order the plan plans, so
two purges could acquire two conversations in opposite orders. Both sites order
by id. The transaction the lock depends on is now stated on `delete_jobs`, which
takes a bare connection and would silently lose the serialisation on one that
autocommits.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

backend/.sqlx/query-4f52bf546579f26a1d22c239b8b0054b753cfbeb5dad1e8120fd5e8a672d50ef.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT id FROM flow_conversation WHERE id = ANY($1) AND workspace_id = $2 FOR UPDATE",
+  "query": "SELECT id FROM flow_conversation WHERE id = ANY($1) AND workspace_id = $2 ORDER BY id FOR UPDATE",
   "describe": {
     "columns": [
       {
@@ -19,5 +19,5 @@
       false
     ]
   },
-  "hash": "72a67160f31a0e66e4487a4508a7955ed919ef4e455ea47efef171395b7fec96"
+  "hash": "4f52bf546579f26a1d22c239b8b0054b753cfbeb5dad1e8120fd5e8a672d50ef"
 }

backend/.sqlx/query-ec295b3890a0018475ec0a3774c7a30d71a5689efe72daf58bd1e8f6cf90c410.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT id FROM flow_conversation WHERE id = ANY($1) FOR UPDATE",
+  "query": "SELECT id FROM flow_conversation WHERE id = ANY($1) ORDER BY id FOR UPDATE",
   "describe": {
     "columns": [
       {
@@ -18,5 +18,5 @@
       false
     ]
   },
-  "hash": "0a5502ee13a1af720ca7cc0596a32f0fe91257bb31b595c05f64c7903ad80ead"
+  "hash": "ec295b3890a0018475ec0a3774c7a30d71a5689efe72daf58bd1e8f6cf90c410"
 }

backend/windmill-api-jobs/src/jobs_export.rs

@@ -715,7 +715,7 @@ pub async fn delete_jobs(
         // delete one of a conversation's last messages, neither sees the other's uncommitted
         // delete, and the conversation and its memory are left with nothing to collect them.
         sqlx::query_scalar!(
-            "SELECT id FROM flow_conversation WHERE id = ANY($1) AND workspace_id = $2 FOR UPDATE",
+            "SELECT id FROM flow_conversation WHERE id = ANY($1) AND workspace_id = $2 ORDER BY id FOR UPDATE",
             &conversation_ids,
             &w_id
         )

backend/windmill-common/src/jobs.rs

@@ -478,6 +478,9 @@ pub static WORKER_INTERNAL_SERVER_INLINE_UTILS: OnceCell<WorkerInternalServerInl
 /// set-based deletes below cost one scan per table per call instead. Because the cascade no
 /// longer fires, every code path that deletes from `v2_job` by id must go through this helper
 /// (or delete these tables itself) or it will leave orphan rows behind.
+/// **Transaction contract:** call this inside a transaction. The conversation cleanup below
+/// locks rows to serialise itself against a concurrent delete, and on an autocommit
+/// connection that lock is released at statement end, silently restoring the race.
 pub async fn delete_jobs(conn: &mut sqlx::PgConnection, ids: &[uuid::Uuid]) -> error::Result<()> {
     sqlx::query!(
         "DELETE FROM dispatch_event WHERE producer_job_id = ANY($1)",
@@ -504,7 +507,7 @@ pub async fn delete_jobs(conn: &mut sqlx::PgConnection, ids: &[uuid::Uuid]) -> e
         // neither would collect it and nothing would try again. Taking the conversation row
         // first serialises them: the second reads the first's delete and finds it empty.
         sqlx::query_scalar!(
-            "SELECT id FROM flow_conversation WHERE id = ANY($1) FOR UPDATE",
+            "SELECT id FROM flow_conversation WHERE id = ANY($1) ORDER BY id FOR UPDATE",
             &conversation_ids
         )
         .fetch_all(&mut *conn)

frontend/src/lib/components/FlowPreviewContent.svelte

@@ -481,7 +481,7 @@
 						}}
 						conversationKind="test"
 						frame="boxed"
-						path={$initialPathStore || $pathStore}
+						path={$initialPathStore || fakeInitialPath}
 						inputSchema={flowStore.val.schema}
 						flowModules={flowStore.val.value?.modules}
 					/>

frontend/src/lib/components/flows/content/FlowInput.svelte

@@ -814,7 +814,7 @@
 						<FlowChat
 							onRunFlow={runFlowWithMessage}
 							conversationKind="test"
-							path={$initialPathStore || $pathStore}
+							path={$initialPathStore || fakeInitialPath}
 							useStreaming={shouldUseStreaming}
 							inputSchema={flowStore.val.schema}
 							flowModules={flowStore.val.value?.modules}

frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts

@@ -781,6 +781,7 @@ export class FlowChatManager {
 	private async loadConversations(page: number, perPage: number) {
 		if (!this.#workspace() || !this.#path) return []
 
+		const startedIn = this.#generation
 		try {
 			const response = await FlowConversationsService.listFlowConversations({
 				workspace: this.#workspace()!,
@@ -789,6 +790,9 @@ export class FlowChatManager {
 				page: page,
 				perPage: perPage
 			})
+			// The list is bound straight into the sidebar, so one fetched for the flow just
+			// left would otherwise become the list of the flow that replaced it.
+			if (startedIn !== this.#generation) return []
 
 			return response
 		} catch (error) {
@@ -919,6 +923,7 @@ export class FlowChatManager {
 
 		try {
 			const lastSeq = this.getLastPersistedMessageSeq(conversationId)
+			const startedIn = this.#generation
 			const response = await FlowConversationsService.listConversationMessages({
 				workspace: this.#workspace()!,
 				conversationId: conversationId,
@@ -926,6 +931,9 @@ export class FlowChatManager {
 				perPage: 50,
 				afterSeq: lastSeq
 			})
+			// An interval tick already dispatched outlives `clearInterval`, and a turn's final
+			// poll outlives its abort — either would put a forgotten flow's rows back.
+			if (startedIn !== this.#generation) return
 
 			if (options?.isNewConversation) {
 				await this.refreshConversations()
@@ -1223,9 +1231,13 @@ export class FlowChatManager {
 					// Anything still buffered would be dropped by the temp-row sweep below.
 					this.#flushReveals(currentConversationId)
 					// Do a final poll to get all messages from database
+					const startedIn = this.#generation
 					await this.pollConversationMessages(currentConversationId, {
 						removeTempMessages: true
 					})
+					// Settling releases this conversation's queue, and after a re-point that
+					// would run the message against the flow now loaded.
+					if (controller.signal.aborted || startedIn !== this.#generation) return
 					this.endTurn(currentConversationId, { settled: true })
 				}
 				return