Merge origin/main into flow-chat-attachments

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

CONTEXT.md

@@ -37,6 +37,25 @@ _Avoid_: argument field, param
 Any other place a property can be picked into: the loop iterator, skip and early-stop predicates, the retry condition, a branch predicate, timeout. Its prop picker opens in a popover from the connect button rather than taking a pane.
 _Avoid_: JS field, code input
 
+### Flow chat
+
+**Conversation**:
+One thread of messages against one chat-enabled flow, with its own agent memory. A flow has
+many; the chat shows one at a time.
+_Avoid_: thread, session (that names an AI session, a different thing), chat (that names the surface)
+
+**Turn**:
+One question and the answer to it: the run the question started, the handle that stops it,
+and the rows it is writing. At most one per conversation, and the chat is held for its whole
+length — from the moment the question takes the chat, before it has a job, until it is ended.
+_Avoid_: request, exchange, message round
+
+**Transcript**:
+The rows a conversation's chat holds. Not the conversation: it is the newest page plus
+whatever older pages the reader has scrolled back through, so a question it cannot answer
+from what it holds is one to ask the server rather than to guess at.
+_Avoid_: history, messages (too easily read as "all of them")
+
 ### Permissions
 
 **Member**:

backend/.sqlx/query-07a005f0f9e80a156cd2a5a0ae39a1fabeaa167818206a25abfe31d5582f942a.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM flow_conversation c\n              WHERE c.id = ANY($1)\n                AND c.workspace_id = $2\n                AND NOT EXISTS (\n                    SELECT 1 FROM flow_conversation_message m WHERE m.conversation_id = c.id\n                )",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "UuidArray",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "07a005f0f9e80a156cd2a5a0ae39a1fabeaa167818206a25abfe31d5582f942a"
+}

backend/.sqlx/query-462d2b2822b185a6f51fafcfa957cb3b31ee6b69abae79a214dddba0dee4425c.json

@@ -1,15 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM flow_conversation_message m\n         USING flow_conversation c\n         WHERE m.conversation_id = c.id AND c.workspace_id = $1 AND m.job_id = ANY($2)",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text",
-        "UuidArray"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "462d2b2822b185a6f51fafcfa957cb3b31ee6b69abae79a214dddba0dee4425c"
-}

backend/.sqlx/query-4f52bf546579f26a1d22c239b8b0054b753cfbeb5dad1e8120fd5e8a672d50ef.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id FROM flow_conversation WHERE id = ANY($1) AND workspace_id = $2 ORDER BY id FOR UPDATE",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "UuidArray",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "4f52bf546579f26a1d22c239b8b0054b753cfbeb5dad1e8120fd5e8a672d50ef"
+}

backend/.sqlx/query-69bfbe9b39414b724488532cc3b3659915d9fcb3d58f16532aaffe06c44ec976.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM flow_conversation_message m\n         USING flow_conversation c\n         WHERE m.conversation_id = c.id AND c.workspace_id = $1 AND m.job_id = ANY($2)\n         RETURNING m.conversation_id",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "conversation_id",
+        "type_info": "Uuid"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "UuidArray"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "69bfbe9b39414b724488532cc3b3659915d9fcb3d58f16532aaffe06c44ec976"
+}

backend/.sqlx/query-6f32c1feed096ff706ae359ad6a3ca33b3f82ca38289dfa4a69aa95041027d57.json

@@ -0,0 +1,59 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id, workspace_id, flow_path, title, created_at, updated_at, created_by\n         FROM flow_conversation\n         WHERE id = $1 AND workspace_id = $2\n         FOR UPDATE",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 1,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "flow_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "title",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 4,
+        "name": "created_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 5,
+        "name": "updated_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 6,
+        "name": "created_by",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      true,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "6f32c1feed096ff706ae359ad6a3ca33b3f82ca38289dfa4a69aa95041027d57"
+}

backend/.sqlx/query-89ea81b765550cf665e30533efc9672f8c98d2579fc72c07752361cc5fd683dc.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM ai_agent_memory a\n               USING flow_conversation c\n              WHERE c.id = ANY($1)\n                AND c.workspace_id = $2\n                AND a.conversation_id = c.id\n                AND a.workspace_id = c.workspace_id\n                AND NOT EXISTS (\n                    SELECT 1 FROM flow_conversation_message m WHERE m.conversation_id = c.id\n                )",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "UuidArray",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "89ea81b765550cf665e30533efc9672f8c98d2579fc72c07752361cc5fd683dc"
+}

backend/.sqlx/query-90b910da8d00a7c7bcf29c167e38e44eb1c0062a8241dc8fe0ed3dd95b65f89a.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM flow_conversation c\n              WHERE c.id = ANY($1)\n                AND NOT EXISTS (\n                    SELECT 1 FROM flow_conversation_message m WHERE m.conversation_id = c.id\n                )",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "UuidArray"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "90b910da8d00a7c7bcf29c167e38e44eb1c0062a8241dc8fe0ed3dd95b65f89a"
+}

backend/.sqlx/query-967f52005f4a044b3a2e9f02ceadf90dad5246681bde6caaa633b85a5e8b2352.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM flow_conversation_message WHERE job_id = ANY($1) RETURNING conversation_id",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "conversation_id",
+        "type_info": "Uuid"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "UuidArray"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "967f52005f4a044b3a2e9f02ceadf90dad5246681bde6caaa633b85a5e8b2352"
+}

backend/.sqlx/query-a1b23f3e62c6433d95cdac58215741a51ca0ce66bf1c674097705cf2e1b72eff.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM ai_agent_memory a\n               USING flow_conversation c\n              WHERE c.id = ANY($1)\n                AND a.conversation_id = c.id\n                AND a.workspace_id = c.workspace_id\n                AND NOT EXISTS (\n                    SELECT 1 FROM flow_conversation_message m WHERE m.conversation_id = c.id\n                )",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "UuidArray"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "a1b23f3e62c6433d95cdac58215741a51ca0ce66bf1c674097705cf2e1b72eff"
+}

backend/.sqlx/query-bfdd60b42e32bd81e2d20b327462893147b4e5ff078531de36147d908132d636.json

@@ -1,14 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM flow_conversation_message WHERE job_id = ANY($1)",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "UuidArray"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "bfdd60b42e32bd81e2d20b327462893147b4e5ff078531de36147d908132d636"
-}

backend/.sqlx/query-c1e3ed3ecc3bcb98f60ba8196d33fee4a74f61b061e5025ecb75882208b3ba8f.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "INSERT INTO flow_conversation (id, workspace_id, flow_path, created_by, title)\n         VALUES ($1, $2, $3, $4, $5)\n         RETURNING id, workspace_id, flow_path, title, created_at, updated_at, created_by",
+  "query": "INSERT INTO flow_conversation (id, workspace_id, flow_path, created_by, title)\n         VALUES ($1, $2, $3, $4, $5)\n         ON CONFLICT (id) DO NOTHING\n         RETURNING id, workspace_id, flow_path, title, created_at, updated_at, created_by",
   "describe": {
     "columns": [
       {
@@ -58,5 +58,5 @@
       false
     ]
   },
-  "hash": "6bd23a98838e3eec309e6b696edc776bd56fc9dae1238b3272557d1562400dbe"
+  "hash": "c1e3ed3ecc3bcb98f60ba8196d33fee4a74f61b061e5025ecb75882208b3ba8f"
 }

backend/.sqlx/query-ec295b3890a0018475ec0a3774c7a30d71a5689efe72daf58bd1e8f6cf90c410.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id FROM flow_conversation WHERE id = ANY($1) ORDER BY id FOR UPDATE",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "UuidArray"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "ec295b3890a0018475ec0a3774c7a30d71a5689efe72daf58bd1e8f6cf90c410"
+}

backend/tests/v2_job_delete_orphans.rs

@@ -37,7 +37,7 @@ async fn seed_side_rows(db: &Pool<Postgres>, ws: &str, job_id: Uuid) -> anyhow::
     .bind(ws)
     .execute(db)
     .await?;
-    // created_seq is assigned by a trigger; inserting a value is rejected.
+    // created_seq is an identity column; supplying a value is rejected.
     sqlx::query(
         "INSERT INTO flow_conversation_message (conversation_id, message_type, content, job_id)
          VALUES ($1, 'assistant', 'hi', $2)",
@@ -121,6 +121,184 @@ async fn test_delete_jobs_removes_side_rows(db: Pool<Postgres>) -> anyhow::Resul
     Ok(())
 }
 
+/// (conversation rows, agent-memory rows) for one conversation.
+async fn conversation_and_memory_counts(
+    db: &Pool<Postgres>,
+    conversation_id: Uuid,
+) -> anyhow::Result<(i64, i64)> {
+    Ok((
+        count(
+            db,
+            "SELECT count(*) FROM flow_conversation WHERE id = $1",
+            conversation_id,
+        )
+        .await?,
+        count(
+            db,
+            "SELECT count(*) FROM ai_agent_memory WHERE conversation_id = $1",
+            conversation_id,
+        )
+        .await?,
+    ))
+}
+
+/// A conversation outlives the jobs behind its messages until the last one goes: only then
+/// are the row and the agent's memory for it left with nothing, and only then are they
+/// deleted. Both halves matter — the surviving half is what a single data-modifying CTE
+/// would break, since its emptiness check would read the snapshot from before the delete.
+#[sqlx::test(fixtures("base"))]
+async fn test_delete_jobs_removes_a_conversation_once_its_last_message_goes(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    let first_job = Uuid::new_v4();
+    let second_job = Uuid::new_v4();
+    insert_job(&db, WS, first_job).await?;
+    insert_job(&db, WS, second_job).await?;
+
+    let conv_id = Uuid::new_v4();
+    sqlx::query(
+        "INSERT INTO flow_conversation (id, workspace_id, flow_path, created_by)
+         VALUES ($1, $2, 'f/flow', 'test-user')",
+    )
+    .bind(conv_id)
+    .bind(WS)
+    .execute(&db)
+    .await?;
+    for job_id in [first_job, second_job] {
+        sqlx::query(
+            "INSERT INTO flow_conversation_message (conversation_id, message_type, content, job_id)
+             VALUES ($1, 'assistant', 'hi', $2)",
+        )
+        .bind(conv_id)
+        .bind(job_id)
+        .execute(&db)
+        .await?;
+    }
+    sqlx::query(
+        "INSERT INTO ai_agent_memory (workspace_id, conversation_id, step_id, messages)
+         VALUES ($1, $2, 'a', '[]'::jsonb)",
+    )
+    .bind(WS)
+    .bind(conv_id)
+    .execute(&db)
+    .await?;
+
+    let mut conn = db.acquire().await?;
+    windmill_common::jobs::delete_jobs(&mut conn, &[first_job]).await?;
+    drop(conn);
+    assert_eq!(
+        conversation_and_memory_counts(&db, conv_id).await?,
+        (1, 1),
+        "a conversation with a message left must survive, memory included"
+    );
+
+    let mut conn = db.acquire().await?;
+    windmill_common::jobs::delete_jobs(&mut conn, &[second_job]).await?;
+    drop(conn);
+    assert_eq!(
+        conversation_and_memory_counts(&db, conv_id).await?,
+        (0, 0),
+        "the last message going should take the conversation and its memory"
+    );
+    Ok(())
+}
+
+/// Turns that start while retention is collecting their conversation must land, not fail:
+/// the conversation lookup locks the row, so each turn waits for the collector's commit,
+/// finds the conversation gone, and creates it again — the first insert wins and the other
+/// reads its row. Without the lock a turn's message insert is what waits, on the parent
+/// row's key lock, and fails its FK check afterwards.
+#[sqlx::test(fixtures("base"))]
+async fn test_new_turns_wait_for_conversation_cleanup_and_recreate(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    let old_job = Uuid::new_v4();
+    let new_jobs = [Uuid::new_v4(), Uuid::new_v4()];
+    insert_job(&db, WS, old_job).await?;
+    for job in new_jobs {
+        insert_job(&db, WS, job).await?;
+    }
+
+    let conv_id = Uuid::new_v4();
+    sqlx::query(
+        "INSERT INTO flow_conversation (id, workspace_id, flow_path, created_by)
+         VALUES ($1, $2, 'f/flow', 'test-user')",
+    )
+    .bind(conv_id)
+    .bind(WS)
+    .execute(&db)
+    .await?;
+    sqlx::query(
+        "INSERT INTO flow_conversation_message (conversation_id, message_type, content, job_id)
+         VALUES ($1, 'user', 'hi', $2)",
+    )
+    .bind(conv_id)
+    .bind(old_job)
+    .execute(&db)
+    .await?;
+
+    // The collector holds the conversation row locked and deleted, uncommitted.
+    let mut cleanup = db.begin().await?;
+    windmill_common::jobs::delete_jobs(&mut *cleanup, &[old_job]).await?;
+
+    let turns: Vec<_> = new_jobs
+        .into_iter()
+        .map(|new_job| {
+            let db = db.clone();
+            tokio::spawn(async move {
+                let mut tx = db.begin().await?;
+                windmill_common::flow_conversations::get_or_create_conversation_with_id(
+                    &mut tx,
+                    WS,
+                    "f/flow",
+                    "test-user",
+                    "hi again",
+                    conv_id,
+                )
+                .await?;
+                windmill_common::flow_conversations::add_message_to_conversation_tx(
+                    &mut tx,
+                    conv_id,
+                    Some(new_job),
+                    "hi again",
+                    windmill_common::flow_conversations::MessageType::User,
+                    None,
+                    true,
+                )
+                .await?;
+                tx.commit().await?;
+                anyhow::Ok(())
+            })
+        })
+        .collect();
+    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
+    cleanup.commit().await?;
+    for turn in turns {
+        turn.await??;
+    }
+
+    assert_eq!(
+        conversation_and_memory_counts(&db, conv_id).await?.0,
+        1,
+        "the turns must have created the conversation again, once"
+    );
+    assert_eq!(
+        count(
+            &db,
+            "SELECT count(*) FROM flow_conversation_message WHERE conversation_id = $1",
+            conv_id,
+        )
+        .await?,
+        2,
+        "both turns' messages should be there"
+    );
+    Ok(())
+}
+
 #[sqlx::test(fixtures("base"))]
 async fn test_clear_schedule_removes_side_rows(db: Pool<Postgres>) -> anyhow::Result<()> {
     initialize_tracing().await;
@@ -192,6 +370,74 @@ async fn test_workspace_delete_removes_side_rows(db: Pool<Postgres>) -> anyhow::
     Ok(())
 }
 
+/// The purge endpoint carries its own copy of the emptied-conversation rule, so it gets the
+/// same guard: the conversation and its memory go with the last message, and not before.
+#[sqlx::test(fixtures("base"))]
+async fn test_jobs_export_delete_removes_a_conversation_once_its_last_message_goes(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    let first_job = Uuid::new_v4();
+    let second_job = Uuid::new_v4();
+    insert_job(&db, WS, first_job).await?;
+    insert_job(&db, WS, second_job).await?;
+
+    let conv_id = Uuid::new_v4();
+    sqlx::query(
+        "INSERT INTO flow_conversation (id, workspace_id, flow_path, created_by)
+         VALUES ($1, $2, 'f/flow', 'test-user')",
+    )
+    .bind(conv_id)
+    .bind(WS)
+    .execute(&db)
+    .await?;
+    for job_id in [first_job, second_job] {
+        sqlx::query(
+            "INSERT INTO flow_conversation_message (conversation_id, message_type, content, job_id)
+             VALUES ($1, 'assistant', 'hi', $2)",
+        )
+        .bind(conv_id)
+        .bind(job_id)
+        .execute(&db)
+        .await?;
+    }
+    sqlx::query(
+        "INSERT INTO ai_agent_memory (workspace_id, conversation_id, step_id, messages)
+         VALUES ($1, $2, 'a', '[]'::jsonb)",
+    )
+    .bind(WS)
+    .bind(conv_id)
+    .execute(&db)
+    .await?;
+
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+    let purge = |job_id: Uuid| async move {
+        reqwest::Client::new()
+            .post(format!("http://localhost:{port}/api/w/{WS}/jobs/delete"))
+            .header("Authorization", "Bearer SECRET_TOKEN")
+            .json(&[job_id])
+            .send()
+            .await
+    };
+
+    assert!(purge(first_job).await?.status().is_success());
+    assert_eq!(
+        conversation_and_memory_counts(&db, conv_id).await?,
+        (1, 1),
+        "a conversation with a message left must survive the purge endpoint too"
+    );
+
+    assert!(purge(second_job).await?.status().is_success());
+    assert_eq!(
+        conversation_and_memory_counts(&db, conv_id).await?,
+        (0, 0),
+        "the last message going should take the conversation and its memory"
+    );
+    Ok(())
+}
+
 /// The `/jobs/delete` purge endpoint must scope every side-table delete to the path
 /// workspace. A `test-workspace` admin passing a job id from another workspace must not be
 /// able to delete that workspace's job or side rows (the side tables no longer cascade, so

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

@@ -668,10 +668,16 @@ pub async fn handle_chat_conversation_messages(
     flow_path: &str,
     run_query: &RunJobQuery,
     user_message_raw: Option<&Box<serde_json::value::RawValue>>,
+    job_id: Uuid,
 ) -> error::Result<()> {
+    // Names the query parameter rather than the field: it is not a flow argument, and
+    // supplying it as one is the first thing tried on reading `memory_id is required`.
     let memory_id = run_query.memory_id.ok_or_else(|| {
         windmill_common::error::Error::BadRequest(
-            "memory_id is required for chat-enabled flows".to_string(),
+            "memory_id is required for chat-enabled flows. Pass it as the `memory_id` query \
+             parameter, not as a flow argument: it names the conversation the turn belongs to, \
+             so a fresh UUID starts one and reusing a UUID continues it."
+                .to_string(),
         )
     })?;
 
@@ -698,10 +704,13 @@ pub async fn handle_chat_conversation_messages(
     )
     .await?;
 
+    // The run this message started. Its args are the only record of what the message
+    // carried besides its text — attachments and every other flow input — and nothing
+    // written later points at them: an assistant row holds the AI agent step's job.
     add_message_to_conversation_tx(
         tx,
         memory_id,
-        None,
+        Some(job_id),
         &user_message,
         MessageType::User,
         None,
@@ -826,6 +835,7 @@ pub async fn run_flow<'c>(
             &flow_path.to_string(),
             &run_query,
             args.args.get("user_message"),
+            uuid,
         )
         .await?;
     }

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

@@ -692,16 +692,64 @@ pub async fn delete_jobs(
     .await?
     .rows_affected();
 
-    let conversation_message_deleted = sqlx::query!(
+    // One row per message deleted, so the conversation of a chat losing several appears
+    // several times: the count is taken before the dedup below.
+    let mut conversation_ids: Vec<Uuid> = sqlx::query_scalar!(
         "DELETE FROM flow_conversation_message m
          USING flow_conversation c
-         WHERE m.conversation_id = c.id AND c.workspace_id = $1 AND m.job_id = ANY($2)",
+         WHERE m.conversation_id = c.id AND c.workspace_id = $1 AND m.job_id = ANY($2)
+         RETURNING m.conversation_id",
         &w_id,
         &job_ids
     )
-    .execute(&mut *tx)
-    .await?
-    .rows_affected();
+    .fetch_all(&mut *tx)
+    .await?;
+    let conversation_message_deleted = conversation_ids.len() as u64;
+
+    // Same rule, lock and statement order as retention (windmill_common::jobs::delete_jobs,
+    // which says why): a conversation with no messages left goes, and its memory with it.
+    conversation_ids.sort_unstable();
+    conversation_ids.dedup();
+    let mut memory_deleted = 0;
+    let mut conversation_deleted = 0;
+    if !conversation_ids.is_empty() {
+        sqlx::query_scalar!(
+            "SELECT id FROM flow_conversation WHERE id = ANY($1) AND workspace_id = $2 ORDER BY id FOR UPDATE",
+            &conversation_ids,
+            &w_id
+        )
+        .fetch_all(&mut *tx)
+        .await?;
+        memory_deleted = sqlx::query!(
+            "DELETE FROM ai_agent_memory a
+               USING flow_conversation c
+              WHERE c.id = ANY($1)
+                AND c.workspace_id = $2
+                AND a.conversation_id = c.id
+                AND a.workspace_id = c.workspace_id
+                AND NOT EXISTS (
+                    SELECT 1 FROM flow_conversation_message m WHERE m.conversation_id = c.id
+                )",
+            &conversation_ids,
+            &w_id
+        )
+        .execute(&mut *tx)
+        .await?
+        .rows_affected();
+        conversation_deleted = sqlx::query!(
+            "DELETE FROM flow_conversation c
+              WHERE c.id = ANY($1)
+                AND c.workspace_id = $2
+                AND NOT EXISTS (
+                    SELECT 1 FROM flow_conversation_message m WHERE m.conversation_id = c.id
+                )",
+            &conversation_ids,
+            &w_id
+        )
+        .execute(&mut *tx)
+        .await?
+        .rows_affected();
+    }
 
     // Resolutions are not exported, so a delete-then-reimport of the same UUID would
     // otherwise resurrect the old annotation on a job that never carried one.
@@ -737,6 +785,8 @@ pub async fn delete_jobs(
         + zombie_deleted
         + dispatch_event_deleted
         + conversation_message_deleted
+        + memory_deleted
+        + conversation_deleted
         + resolution_deleted
         + jobs_deleted;
 

backend/windmill-api/src/jobs.rs

@@ -9553,6 +9553,7 @@ async fn run_preview_flow_job(
             &flow_path,
             &run_query,
             user_message.as_ref(),
+            uuid,
         )
         .await?;
     }

backend/windmill-common/src/flow_conversations.rs

@@ -36,30 +36,20 @@ pub async fn get_or_create_conversation_with_id(
     title: &str,
     conversation_id: Uuid,
 ) -> Result<FlowConversation> {
-    // Check if conversation already exists
-    let existing_conversation = sqlx::query_as!(
-        FlowConversation,
-        "SELECT id, workspace_id, flow_path, title, created_at, updated_at, created_by
-         FROM flow_conversation
-         WHERE id = $1 AND workspace_id = $2",
-        conversation_id,
-        w_id
-    )
-    .fetch_optional(&mut **tx)
-    .await?;
-
-    if let Some(existing) = existing_conversation {
+    if let Some(existing) = lock_conversation(tx, w_id, conversation_id).await? {
         return Ok(existing);
     }
 
     // Truncate title to 25 characters max
     let title = truncate_with_ellipsis(title, 25);
 
-    // Create new conversation with provided ID
-    let conversation = sqlx::query_as!(
+    // Every turn released by the same collector's commit finds no row: the first insert
+    // wins, the others wait on it, do nothing, and read the row it created.
+    let created = sqlx::query_as!(
         FlowConversation,
         "INSERT INTO flow_conversation (id, workspace_id, flow_path, created_by, title)
          VALUES ($1, $2, $3, $4, $5)
+         ON CONFLICT (id) DO NOTHING
          RETURNING id, workspace_id, flow_path, title, created_at, updated_at, created_by",
         conversation_id,
         w_id,
@@ -67,10 +57,41 @@ pub async fn get_or_create_conversation_with_id(
         username,
         title
     )
-    .fetch_one(&mut **tx)
+    .fetch_optional(&mut **tx)
     .await?;
+    if let Some(conversation) = created {
+        return Ok(conversation);
+    }
 
-    Ok(conversation)
+    lock_conversation(tx, w_id, conversation_id)
+        .await?
+        .ok_or_else(|| {
+            crate::error::Error::BadRequest(format!(
+                "conversation {conversation_id} belongs to another workspace"
+            ))
+        })
+}
+
+/// Locked, so a turn orders against retention collecting the conversation
+/// (windmill_common::jobs::delete_jobs): either the turn goes first and the collector then
+/// sees its message, or it waits and finds the row gone and creates it again. Unlocked, the
+/// message insert would wait on the parent row's lock instead and then fail its FK check.
+async fn lock_conversation(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    w_id: &str,
+    conversation_id: Uuid,
+) -> Result<Option<FlowConversation>> {
+    Ok(sqlx::query_as!(
+        FlowConversation,
+        "SELECT id, workspace_id, flow_path, title, created_at, updated_at, created_by
+         FROM flow_conversation
+         WHERE id = $1 AND workspace_id = $2
+         FOR UPDATE",
+        conversation_id,
+        w_id
+    )
+    .fetch_optional(&mut **tx)
+    .await?)
 }
 
 /// Add a message to a conversation using an existing transaction

backend/windmill-common/src/jobs.rs

@@ -478,6 +478,12 @@ 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.
+/// A conversation is collected only once every message row of it has gone with a job; a
+/// row written with no job id (an MCP tool call, persisted under no job of its own) keeps
+/// its conversation and the agent's memory for it alive for as long as it exists.
 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)",
@@ -485,12 +491,55 @@ pub async fn delete_jobs(conn: &mut sqlx::PgConnection, ids: &[uuid::Uuid]) -> e
     )
     .execute(&mut *conn)
     .await?;
-    sqlx::query!(
-        "DELETE FROM flow_conversation_message WHERE job_id = ANY($1)",
+    let mut conversation_ids: Vec<uuid::Uuid> = sqlx::query_scalar!(
+        "DELETE FROM flow_conversation_message WHERE job_id = ANY($1) RETURNING conversation_id",
         ids
     )
-    .execute(&mut *conn)
+    .fetch_all(&mut *conn)
     .await?;
+    conversation_ids.sort_unstable();
+    conversation_ids.dedup();
+    if !conversation_ids.is_empty() {
+        // A conversation is a view over its messages: once the last one goes with its job,
+        // the row and the agent's memory for it are all that is left, and nothing else
+        // collects them — `ai_agent_memory` carries no job id for retention to match on.
+        // Two statements rather than one CTE: a data-modifying CTE reads the snapshot from
+        // before the delete above, so every conversation would still look non-empty.
+        // Two calls each deleting one of a conversation's last messages would each still see
+        // the other's row — uncommitted deletes are invisible across transactions — so
+        // 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) ORDER BY id FOR UPDATE",
+            &conversation_ids
+        )
+        .fetch_all(&mut *conn)
+        .await?;
+        // Memory first, since it reads the conversation row for its workspace.
+        sqlx::query!(
+            "DELETE FROM ai_agent_memory a
+               USING flow_conversation c
+              WHERE c.id = ANY($1)
+                AND a.conversation_id = c.id
+                AND a.workspace_id = c.workspace_id
+                AND NOT EXISTS (
+                    SELECT 1 FROM flow_conversation_message m WHERE m.conversation_id = c.id
+                )",
+            &conversation_ids
+        )
+        .execute(&mut *conn)
+        .await?;
+        sqlx::query!(
+            "DELETE FROM flow_conversation c
+              WHERE c.id = ANY($1)
+                AND NOT EXISTS (
+                    SELECT 1 FROM flow_conversation_message m WHERE m.conversation_id = c.id
+                )",
+            &conversation_ids
+        )
+        .execute(&mut *conn)
+        .await?;
+    }
     sqlx::query!("DELETE FROM zombie_job_counter WHERE job_id = ANY($1)", ids)
         .execute(&mut *conn)
         .await?;

cli/src/guidance/skills.gen.ts

@@ -5280,6 +5280,42 @@ tool, \`websearch\` for web search.
   "resource": "$res:<path>", "model": <model id> }\`. Required unless the module links to a saved
   agent through \`value.agent\`
 
+### Chat-Mode Flows
+
+A flow with \`value.chat_input_enabled: true\` is run from a chat instead of a form: the composer
+sends one message per turn and renders the conversation. It needs a required \`user_message\` string
+input, read by the agent. Any other flow input stays and is asked for under Configure inputs.
+
+\`\`\`json
+{
+  "id": "chat_agent",
+  "value": {
+    "type": "aiagent",
+    "input_transforms": {
+      "provider": {
+        "type": "static",
+        "value": { "kind": "anthropic", "resource": "$res:f/ai/claude", "model": "claude-sonnet-5" }
+      },
+      "user_message": { "type": "javascript", "expr": "flow_input.user_message" },
+      "user_attachments": { "type": "javascript", "expr": "flow_input.files" },
+      "memory": { "type": "static", "value": { "kind": "auto", "context_length": 10 } },
+      "streaming": { "type": "static", "value": true },
+      "output_type": { "type": "static", "value": "text" }
+    },
+    "tools": []
+  }
+}
+\`\`\`
+
+- \`memory\` is what lets the agent see earlier turns; without it every message starts from nothing
+- \`streaming\` on makes the answer and its thinking appear token by token instead of all at once
+- \`user_attachments\` points at a flow input typed as an array of s3 objects
+  (\`{ "type": "array", "items": { "type": "object", "resourceType": "s3object" } }\`), so files
+  sent with a message reach the agent
+- Running one needs a \`memory_id\` **query parameter** — not a flow argument — naming the
+  conversation the turn belongs to: a fresh UUID starts one, reusing a UUID continues it. The chat
+  supplies it itself; a run driven any other way has to pass it or the server refuses the job
+
 ### Tool Naming Rules
 
 These rules cover \`flowmodule\` tools, the ones the agent calls by name. A \`websearch\` tool's

frontend/src/lib/common.ts

@@ -45,6 +45,8 @@ export interface SchemaProperty {
 	required?: string[]
 	showExpr?: string
 	hideWhenChatEnabled?: boolean
+	/** Why the oneOf variant is chat mode's to pick. Set = selector disabled, reason shown. */
+	lockOneOfWhenChatEnabled?: string
 	password?: boolean
 	order?: string[]
 	nullable?: boolean

frontend/src/lib/components/ArgInput.svelte

@@ -123,6 +123,8 @@
 		workspace?: string | undefined
 		s3StorageConfigured?: boolean
 		chatInputEnabled?: boolean
+		/** Why the oneOf variant is fixed. Set = the selector is disabled and says so. */
+		oneOfLockedReason?: string
 		actions?: import('svelte').Snippet
 		innerBottomSnippet?: import('svelte').Snippet
 		fieldHeaderActions?: import('svelte').Snippet
@@ -184,6 +186,7 @@
 		workspace = undefined,
 		s3StorageConfigured = true,
 		chatInputEnabled = false,
+		oneOfLockedReason = undefined,
 		actions,
 		innerBottomSnippet,
 		fieldHeaderActions,
@@ -1104,11 +1107,15 @@
 		{:else if inputCat == 'object' || inputCat == 'resource-object' || isListJson}
 			{#if oneOf && oneOf.length >= 2}
 				<div class="flex flex-col gap-2 w-full border rounded-md p-4">
+					{#if oneOfLockedReason !== undefined}
+						<div class="text-2xs text-tertiary">{oneOfLockedReason}</div>
+					{/if}
 					{#if oneOf && oneOf.length >= 2}
 						<ToggleButtonGroup
 							selected={oneOfSelected}
 							wrap
 							class="mb-4"
+							disabled={disabled || oneOfLockedReason !== undefined}
 							on:selected={({ detail }) => {
 								oneOfSelected = detail
 								const selectedObjProperties =

frontend/src/lib/components/EditableSchemaForm.svelte

@@ -43,6 +43,8 @@
 	interface Props {
 		schema: Schema | any
 		hiddenArgs?: string[]
+		/** Fields another part of the app owns: shown, but not renameable, deletable or retypeable. */
+		lockedArgs?: string[]
 		args?: Record<string, any>
 		shouldHideNoInputs?: boolean
 		noVariablePicker?: boolean
@@ -89,6 +91,7 @@
 	let {
 		schema = $bindable(),
 		hiddenArgs = [],
+		lockedArgs = [],
 		args = $bindable(undefined),
 		shouldHideNoInputs = false,
 		noVariablePicker = false,
@@ -587,6 +590,7 @@
 						>
 							{#if keys.length > 0}
 								{#each keys as argName, i (argName)}
+									{@const locked = lockedArgs.includes(argName)}
 									<div>
 										<!-- svelte-ignore a11y_click_events_have_key_events -->
 										<!-- svelte-ignore a11y_no_static_element_interactions -->
@@ -605,7 +609,7 @@
 										>
 											<div class="flex flex-row gap-2 text-sm">
 												{argName}
-												{#if !uiOnly}
+												{#if !uiOnly && !locked}
 													<div onclick={stopPropagation(preventDefault(bubble('click')))}>
 														<Popover placement="bottom-end" closeButton>
 															{#snippet trigger()}
@@ -654,7 +658,7 @@
 												<span class="text-red-500 text-xs"> Required </span>
 											{/if}
 
-											{#if !uiOnly}
+											{#if !uiOnly && !locked}
 												<button
 													class="delete-schema-field-button
 													rounded-full p-1 text-gray-500 bg-white
@@ -701,6 +705,7 @@
 																		<ToggleButtonGroup
 																			tabListClass="flex-wrap"
 																			class="h-auto"
+																			disabled={lockedArgs.includes(opened ?? '')}
 																			bind:selected={
 																				() => computeSelected(schema.properties[opened ?? '']),
 																				(v) => {

frontend/src/lib/components/InputTransformForm.svelte

@@ -999,6 +999,11 @@
 									{helperScript}
 									{s3StorageConfigured}
 									{chatInputEnabled}
+									oneOfLockedReason={chatInputEnabled &&
+									arg?.type === 'static' &&
+									(arg.value as any)?.kind !== 'off'
+										? schema.properties[argName]?.lockOneOfWhenChatEnabled
+										: undefined}
 									otherArgs={Object.fromEntries(
 										Object.entries(otherArgs).map(([key, transform]) => [
 											key,

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

@@ -26,7 +26,8 @@
 		Save,
 		X,
 		Check,
-		Settings2
+		Settings2,
+		MessageSquare
 	} from 'lucide-svelte'
 	import CaptureIcon from '$lib/components/triggers/CaptureIcon.svelte'
 	import FlowInputEditor from './FlowInputEditor.svelte'
@@ -47,9 +48,12 @@
 	import type { AiAgent, InputTransform, ScriptLang } from '$lib/gen'
 	import { deepEqual } from 'fast-equals'
 	import Toggle from '$lib/components/Toggle.svelte'
+	import ToggleButtonGroup from '$lib/components/common/toggleButton-v2/ToggleButtonGroup.svelte'
+	import ToggleButton from '$lib/components/common/toggleButton-v2/ToggleButton.svelte'
 	import { AI_AGENT_SCHEMA } from '../flowInfers'
 	import { nextId } from '../flowModuleNextId'
-	import ConfirmationModal from '$lib/components/common/confirmationModal/ConfirmationModal.svelte'
+	import { fetchAgentWithDraft, normalizeAgentRef } from '../linkedAgentDrafts'
+	import type { AIAgentConfig } from '../agentResourceUtils'
 	import FlowChat from '../conversations/FlowChat.svelte'
 	import { SPECIAL_MODULE_IDS } from '$lib/components/copilot/chat/shared'
 
@@ -96,9 +100,10 @@
 	)
 
 	let chatInputEnabled = $state(Boolean(flowStore.val.value?.chat_input_enabled))
-	let showChatModeWarning = $state(false)
-	let showAdditionalInputs = $state(false)
+	// Chat mode shows one of the two at a time: the conversation, or the inputs it sends.
+	let chatPanelTab = $state<'chat' | 'inputs'>('chat')
 	let chatInputsEditTab = $state(false)
+	let chatEditableSchemaForm: EditableSchemaForm | undefined = $state(undefined)
 	let chatInputsAddPropertyV2: AddPropertyV2 | undefined = $state(undefined)
 
 	let addPropertyV2: AddPropertyV2 | undefined = $state(undefined)
@@ -511,23 +516,9 @@
 		return jobId
 	}
 
-	function hasOtherInputs(): boolean {
-		const properties = flowStore.val.schema?.properties
-		return Boolean(
-			properties &&
-				Object.keys(properties).length > 0 &&
-				!(Object.keys(properties).length === 1 && Object.keys(properties).includes('user_message'))
-		)
-	}
-
 	function handleToggleChatMode() {
 		if (!flowStore.val.value?.chat_input_enabled) {
-			// Check if there are existing inputs
-			if (hasOtherInputs()) {
-				showChatModeWarning = true
-			} else {
-				enableChatMode()
-			}
+			enableChatMode()
 		} else {
 			// Disable chat input - remove from flow.value
 			if (flowStore.val.value) {
@@ -536,21 +527,77 @@
 		}
 	}
 
+	/**
+	 * Add the flow input the agent's `user_attachments` reads, and return its name. Chat
+	 * mode means files dropped in the composer, and that only works through a flow input —
+	 * so it is set up with the message and the memory rather than left to be discovered.
+	 */
+	function addAttachmentsInput(): string {
+		const schema = (flowStore.val.schema ?? {}) as Record<string, any>
+		const properties: Record<string, any> = (schema.properties ??= {})
+		let name = 'files'
+		for (let i = 2; name in properties; i++) name = `files_${i}`
+		properties[name] = {
+			type: 'array',
+			items: { type: 'object', resourceType: 's3object' },
+			description: 'Images or PDFs for the agent to read'
+		}
+		flowStore.val.schema = schema
+		return name
+	}
+
+	// The step panel seeds an array-typed field with [] and undefined persists as null
+	// through JSON round-trips, so both count as nothing configured.
+	function isEmptyAgentChatInputValue(value: unknown): boolean {
+		return value == null || value === '' || (Array.isArray(value) && value.length === 0)
+	}
+
+	/**
+	 * Memory and streaming of a linked agent are the resource's, so enabling chat mode cannot
+	 * set them. Read the agent, its draft first, and say what the chat will lack.
+	 */
+	async function warnIfLinkedAgentCannotChat(path: string) {
+		if (!opWs) return
+		let args: AIAgentConfig
+		try {
+			const { response, draft } = await fetchAgentWithDraft(path, opWs)
+			args = draft?.args ?? ((response.value ?? {}) as AIAgentConfig)
+		} catch {
+			// Unreadable here means unreadable at run time too; that run reports it.
+			return
+		}
+		const missing: string[] = []
+		if (!args.memory || (args.memory as { kind?: string }).kind === 'off') missing.push('memory')
+		if (args.streaming !== true) missing.push('streaming')
+		if (missing.length > 0) {
+			sendUserToast(
+				`The linked agent ${path} has ${missing.join(' and ')} off. The chat needs both; turn them on in the agent.`,
+				true
+			)
+		}
+	}
+
 	function enableChatMode() {
 		// Enable chat input - set in flow.value
 		flowStore.val.value.chat_input_enabled = true
 
-		// Set up the schema for chat input
+		// The chat fills the flow's form rather than standing in for it: all it needs is a
+		// `user_message` string, which the server requires under that exact argument name.
+		// Every other input stays — the composer drives the ones an agent field reads, and
+		// the rest are asked for in the Configure-inputs modal.
+		const schema = flowStore.val.schema ?? {}
+		const properties = { ...(schema.properties ?? {}) }
+		// Only a string can carry the message; anything else here cannot be what chat sends.
+		if (properties['user_message']?.type !== 'string') {
+			properties['user_message'] = { type: 'string', description: 'Message from user' }
+		}
+		const required: string[] = Array.isArray(schema.required) ? schema.required : []
 		flowStore.val.schema = {
 			$schema: 'https://json-schema.org/draft/2020-12/schema',
+			...schema,
 			type: 'object',
-			properties: {
-				user_message: {
-					type: 'string',
-					description: 'Message from user'
-				}
-			},
-			required: ['user_message']
+			properties,
+			required: required.includes('user_message') ? required : [...required, 'user_message']
 		}
 
 		// Find all AI agent modules
@@ -570,8 +617,12 @@
 							(accu, key) => {
 								if (key === 'user_message') {
 									accu[key] = { type: 'javascript', expr: 'flow_input.user_message' }
+								} else if (key === 'user_attachments') {
+									accu[key] = { type: 'javascript', expr: `flow_input.${addAttachmentsInput()}` }
 								} else if (key === 'memory') {
 									accu[key] = { type: 'static', value: { kind: 'auto', context_length: 10 } }
+								} else if (key === 'streaming') {
+									accu[key] = { type: 'static', value: true }
 								} else {
 									accu[key] = {
 										type: 'static',
@@ -586,25 +637,23 @@
 				}
 			]
 			sendUserToast(
-				'Chat mode enabled. AI agent created with user message input and context memory set to 10.',
+				'Chat mode enabled. AI agent created with user message and attachments inputs, context memory set to 10 and streaming turned on.',
 				false
 			)
 		} else if (aiAgentModules.length === 1) {
 			// Exactly one AI agent exists: fill in defaults only for inputs the
 			// user hasn't configured, so re-enabling chat mode on an already
-			// configured agent doesn't clobber a custom user_message expression
-			// or a deliberate memory choice (e.g. off).
+			// configured agent doesn't clobber a custom user_message expression.
 			const aiAgent = aiAgentModules[0]
 			const value = aiAgent.value as AiAgent
 
 			// Degenerate shapes the input form can produce without deliberate
-			// configuration count as unconfigured: empty static value (undefined
-			// persists as null through JSON round-trips), blank JS expression
-			// (the JS toggle seeds a bare backtick pair), or an AI transform
-			// (meaningless for the chat input).
+			// configuration count as unconfigured: empty static value, blank JS
+			// expression (the JS toggle seeds a bare backtick pair), or an AI
+			// transform (meaningless for the chat input).
 			const isUnconfigured = (transform: InputTransform | undefined) =>
 				transform === undefined ||
-				(transform.type === 'static' && (transform.value == null || transform.value === '')) ||
+				(transform.type === 'static' && isEmptyAgentChatInputValue(transform.value)) ||
 				(transform.type === 'javascript' && transform.expr.replaceAll('`', '').trim() === '') ||
 				transform.type === 'ai'
 
@@ -617,12 +666,47 @@
 				applied.push('user message input')
 			}
 
-			if (isUnconfigured(value.input_transforms['memory'])) {
-				value.input_transforms['memory'] = {
-					type: 'static',
-					value: { kind: 'auto', context_length: 10 }
+			if (isUnconfigured(value.input_transforms['user_attachments'])) {
+				value.input_transforms['user_attachments'] = {
+					type: 'javascript',
+					expr: `flow_input.${addAttachmentsInput()}`
+				}
+				applied.push('attachments input')
+			}
+
+			// A linked step's brain lives in the agent resource: the worker overlays only the
+			// flow-local keys from the step, so a memory or streaming transform written here
+			// would be ignored. Those are checked on the agent itself below instead.
+			const linkedAgent = value.agent ? normalizeAgentRef(value.agent) : undefined
+			if (linkedAgent === undefined) {
+				// `off` is the first oneOf variant of the memory field, so a step added by hand
+				// carries it without anyone choosing it — and an agent that forgets every turn
+				// makes the chat a series of unrelated questions. Overwritten rather than left
+				// alone; the toast below says it happened.
+				const memoryIsOff = (transform: InputTransform | undefined) =>
+					transform?.type === 'static' && (transform.value as any)?.kind === 'off'
+
+				if (
+					isUnconfigured(value.input_transforms['memory']) ||
+					memoryIsOff(value.input_transforms['memory'])
+				) {
+					value.input_transforms['memory'] = {
+						type: 'static',
+						value: { kind: 'auto', context_length: 10 }
+					}
+					applied.push('context memory set to 10')
+				}
+
+				// Without streaming the chat has no SSE to read, so a turn shows nothing —
+				// no thinking, no answer — until the run ends and its rows are written.
+				if (
+					isUnconfigured(value.input_transforms['streaming']) ||
+					(value.input_transforms['streaming']?.type === 'static' &&
+						value.input_transforms['streaming'].value === false)
+				) {
+					value.input_transforms['streaming'] = { type: 'static', value: true }
+					applied.push('streaming turned on')
 				}
-				applied.push('context memory set to 10')
 			}
 
 			sendUserToast(
@@ -631,42 +715,53 @@
 					: 'Chat mode enabled. Existing AI agent configuration kept unchanged.',
 				false
 			)
+			if (linkedAgent !== undefined) {
+				warnIfLinkedAgentCannotChat(linkedAgent)
+			}
 		}
 		// If there are multiple AI agents, don't auto-configure (ambiguous which one to configure)
-
-		showChatModeWarning = false
 	}
 </script>
 
 <!-- Add svelte:window to listen for keyboard events -->
 <svelte:window onkeydown={handleKeydown} />
 
-<ConfirmationModal
-	open={showChatModeWarning}
-	title="Enable Chat Mode?"
-	confirmationText="Continue"
-	onConfirmed={enableChatMode}
-	onCanceled={() => {
-		showChatModeWarning = false
-		chatInputEnabled = false
-	}}
->
-	<p class="text-sm text-secondary">
-		Enabling Chat Mode will replace all existing flow inputs with a single
-		<span class="font-mono text-xs bg-surface-secondary px-1 rounded">user_message</span>
-		parameter.
-	</p>
-	<p class="text-sm text-secondary mt-2">
-		Your current input configuration will be lost. Are you sure you want to continue?
-	</p>
-</ConfirmationModal>
+<!-- The edit toggle and the add-input target, shared by both panels below: chat mode
+     carries a smaller set of side tabs, but the controls themselves must not differ. -->
+{#snippet inputsEditButton(open: boolean, toggle: () => void)}
+	<Button
+		onClick={toggle}
+		{...open
+			? {
+					title: 'Close input editor',
+					startIcon: { icon: ChevronRight },
+					btnClasses: 'rounded-none rounded-tl-md'
+				}
+			: {
+					title: 'Open input editor',
+					startIcon: { icon: Pen }
+				}}
+		variant="accent"
+		iconOnly
+		wrapperClasses="h-full"
+	/>
+{/snippet}
+
+{#snippet inputsAddTrigger()}
+	<div
+		class="w-full py-2 flex justify-center items-center border border-dashed rounded-md hover:bg-surface-hover"
+		id="add-flow-input-btn"
+	>
+		<Plus size={14} />
+	</div>
+{/snippet}
 
 <FlowCard {noEditor} title="Flow Input">
 	{#snippet action()}
 		{#if !disabled}
 			<div class="flex items-center gap-2">
 				<Toggle
-					size="sm"
+					size="xs"
 					bind:checked={chatInputEnabled}
 					on:change={() => {
 						handleToggleChatMode()
@@ -674,20 +769,25 @@
 					options={{
 						right: 'Chat Mode',
 						rightTooltip:
-							'When enabled, the flow execution page will show a chat interface where each message sent runs the flow with the message as "user_message" input parameter. The flow schema will be automatically set to accept only a user_message string input.'
+							'Turns this flow\'s page into a chat. Each message runs the flow with the message as its "user_message" input, and is kept as a chat — one conversation per chat, each with its own AI agent memory.',
+						rightDocumentationLink:
+							'https://www.windmill.dev/docs/core_concepts/ai_agents#chat-mode'
 					}}
 				/>
 				{#if flowStore.val.value?.chat_input_enabled}
-					<Button
-						size="xs"
-						variant="border"
-						color={showAdditionalInputs ? 'blue' : 'light'}
-						startIcon={{ icon: Settings2 }}
-						title="Manage inputs"
-						on:click={() => (showAdditionalInputs = !showAdditionalInputs)}
-					>
-						Manage inputs
-					</Button>
+					<ToggleButtonGroup bind:selected={chatPanelTab} noWFull>
+						{#snippet children({ item })}
+							<ToggleButton size="sm" value="chat" label="Chat" icon={MessageSquare} {item} />
+							<ToggleButton
+								size="sm"
+								value="inputs"
+								label="Inputs"
+								icon={Settings2}
+								tooltip="Edit the flow inputs the chat sends alongside each message"
+								{item}
+							/>
+						{/snippet}
+					</ToggleButtonGroup>
 				{/if}
 			</div>
 		{/if}
@@ -696,11 +796,15 @@
 		<div class="flex flex-col h-full">
 			{#if flowStore.val.value?.chat_input_enabled}
 				<div class="flex flex-col h-full">
-					{#if showAdditionalInputs}
-						<div class="border-b p-2">
+					{#if chatPanelTab === 'inputs'}
+						<!-- EditableSchemaForm scrolls internally against `h-full`, so the wrapper has
+						     to be bounded (flex-1 min-h-0) or the form grows to content height and
+						     spills out of the panel. -->
+						<div class="py-2 px-4 flex-1 min-h-0">
 							<EditableSchemaForm
+								bind:this={chatEditableSchemaForm}
 								bind:schema={flowStore.val.schema}
-								hiddenArgs={['user_message']}
+								lockedArgs={['user_message']}
 								isFlowInput
 								showSensitiveToggle
 								workspace={opWs}
@@ -713,45 +817,43 @@
 								}}
 							>
 								{#snippet openEditTab()}
-									<Button
-										size="xs"
-										variant={chatInputsEditTab ? 'contained' : 'border'}
-										color={chatInputsEditTab ? 'blue' : 'light'}
-										startIcon={{ icon: chatInputsEditTab ? ChevronRight : Pen }}
-										title={chatInputsEditTab ? 'Close editor' : 'Edit inputs'}
-										onClick={() => {
-											chatInputsEditTab = !chatInputsEditTab
-										}}
-									/>
+									{@render inputsEditButton(
+										chatInputsEditTab,
+										() => (chatInputsEditTab = !chatInputsEditTab)
+									)}
 								{/snippet}
 								{#snippet addProperty()}
 									<AddPropertyV2
 										bind:this={chatInputsAddPropertyV2}
 										bind:schema={flowStore.val.schema}
-										onAddNew={() => {}}
+										onAddNew={(argName) => {
+											chatInputsEditTab = true
+											chatEditableSchemaForm?.openField(argName)
+											refreshStateStore(flowStore)
+										}}
 									>
 										{#snippet trigger()}
-											<Button
-												size="xs"
-												color="light"
-												startIcon={{ icon: Plus }}
-												title="Add additional input"
-											>
-												Add input
-											</Button>
+											{@render inputsAddTrigger()}
 										{/snippet}
 									</AddPropertyV2>
 								{/snippet}
 							</EditableSchemaForm>
 						</div>
 					{/if}
-					<FlowChat
-						onRunFlow={runFlowWithMessage}
-						path={$pathStore}
-						hideSidebar={true}
-						inputSchema={flowStore.val.schema}
-						flowModules={flowStore.val.value?.modules}
-					/>
+					<!-- Hidden rather than unmounted: tearing the chat down destroys its SDK chat,
+					     which ends the stream and poller of the turn in flight, so a turn started
+					     here would finish server-side with nothing following it and the reader would
+					     come back to their own message and no answer. One display class at a time,
+					     so the two cannot race in the cascade. -->
+					<div class={chatPanelTab === 'inputs' ? 'hidden' : 'flex flex-col flex-1 min-h-0'}>
+						<FlowChat
+							onRunFlow={runFlowWithMessage}
+							path={$pathStore}
+							hideSidebar={true}
+							inputSchema={flowStore.val.schema}
+							flowModules={flowStore.val.value?.modules}
+						/>
+					</div>
 				</div>
 			{:else}
 				<div class="py-2 px-4 flex-1 min-h-0">
@@ -816,22 +918,9 @@
 							<div class={twMerge('flex flex-row divide-x', ButtonType.ColorVariants.blue.divider)}>
 								<SideBarTab {dropdownItems} fullMenu={!!$flowInputEditorState?.selectedTab}>
 									{#snippet close_button()}
-										<Button
-											onClick={() => handleEditSchema()}
-											{...!!$flowInputEditorState?.selectedTab
-												? {
-														title: 'Close input editor',
-														startIcon: { icon: ChevronRight },
-														btnClasses: 'rounded-none rounded-tl-md'
-													}
-												: {
-														title: 'Open input editor',
-														startIcon: { icon: Pen }
-													}}
-											variant="accent"
-											iconOnly
-											wrapperClasses="h-full"
-										/>
+										{@render inputsEditButton(!!$flowInputEditorState?.selectedTab, () =>
+											handleEditSchema()
+										)}
 									{/snippet}
 								</SideBarTab>
 							</div>
@@ -885,12 +974,7 @@
 									}}
 								>
 									{#snippet trigger()}
-										<div
-											class="w-full py-2 flex justify-center items-center border border-dashed rounded-md hover:bg-surface-hover"
-											id="add-flow-input-btn"
-										>
-											<Plus size={14} />
-										</div>
+										{@render inputsAddTrigger()}
 									{/snippet}
 								</AddPropertyV2>
 							{/if}

frontend/src/lib/components/flows/flowInfers.ts

@@ -39,6 +39,12 @@ export const AI_AGENT_SCHEMA: Schema = {
 		},
 		memory: {
 			type: 'object',
+			// Chat mode keys memory on the conversation, so a chat whose agent has memory off
+			// forgets every turn. Enabling chat mode sets `auto`; this keeps it there. A step
+			// sitting at `off` stays switchable, or a flow that reached that state before —
+			// an agent added to an already-chat-enabled flow — would have no way out of it.
+			lockOneOfWhenChatEnabled:
+				"Chat mode keys this agent's history on the conversation, so memory stays on while it is enabled.",
 			description: 'History sent between the system message and the user message.',
 			oneOf: [
 				{

system_prompts/auto-generated/flow.md

@@ -106,6 +106,42 @@ tool, `websearch` for web search.
   "resource": "$res:<path>", "model": <model id> }`. Required unless the module links to a saved
   agent through `value.agent`
 
+### Chat-Mode Flows
+
+A flow with `value.chat_input_enabled: true` is run from a chat instead of a form: the composer
+sends one message per turn and renders the conversation. It needs a required `user_message` string
+input, read by the agent. Any other flow input stays and is asked for under Configure inputs.
+
+```json
+{
+  "id": "chat_agent",
+  "value": {
+    "type": "aiagent",
+    "input_transforms": {
+      "provider": {
+        "type": "static",
+        "value": { "kind": "anthropic", "resource": "$res:f/ai/claude", "model": "claude-sonnet-5" }
+      },
+      "user_message": { "type": "javascript", "expr": "flow_input.user_message" },
+      "user_attachments": { "type": "javascript", "expr": "flow_input.files" },
+      "memory": { "type": "static", "value": { "kind": "auto", "context_length": 10 } },
+      "streaming": { "type": "static", "value": true },
+      "output_type": { "type": "static", "value": "text" }
+    },
+    "tools": []
+  }
+}
+```
+
+- `memory` is what lets the agent see earlier turns; without it every message starts from nothing
+- `streaming` on makes the answer and its thinking appear token by token instead of all at once
+- `user_attachments` points at a flow input typed as an array of s3 objects
+  (`{ "type": "array", "items": { "type": "object", "resourceType": "s3object" } }`), so files
+  sent with a message reach the agent
+- Running one needs a `memory_id` **query parameter** — not a flow argument — naming the
+  conversation the turn belongs to: a fresh UUID starts one, reusing a UUID continues it. The chat
+  supplies it itself; a run driven any other way has to pass it or the server refuses the job
+
 ### Tool Naming Rules
 
 These rules cover `flowmodule` tools, the ones the agent calls by name. A `websearch` tool's

system_prompts/auto-generated/prompts.ts

@@ -137,6 +137,42 @@ tool, \`websearch\` for web search.
   "resource": "$res:<path>", "model": <model id> }\`. Required unless the module links to a saved
   agent through \`value.agent\`
 
+### Chat-Mode Flows
+
+A flow with \`value.chat_input_enabled: true\` is run from a chat instead of a form: the composer
+sends one message per turn and renders the conversation. It needs a required \`user_message\` string
+input, read by the agent. Any other flow input stays and is asked for under Configure inputs.
+
+\`\`\`json
+{
+  "id": "chat_agent",
+  "value": {
+    "type": "aiagent",
+    "input_transforms": {
+      "provider": {
+        "type": "static",
+        "value": { "kind": "anthropic", "resource": "$res:f/ai/claude", "model": "claude-sonnet-5" }
+      },
+      "user_message": { "type": "javascript", "expr": "flow_input.user_message" },
+      "user_attachments": { "type": "javascript", "expr": "flow_input.files" },
+      "memory": { "type": "static", "value": { "kind": "auto", "context_length": 10 } },
+      "streaming": { "type": "static", "value": true },
+      "output_type": { "type": "static", "value": "text" }
+    },
+    "tools": []
+  }
+}
+\`\`\`
+
+- \`memory\` is what lets the agent see earlier turns; without it every message starts from nothing
+- \`streaming\` on makes the answer and its thinking appear token by token instead of all at once
+- \`user_attachments\` points at a flow input typed as an array of s3 objects
+  (\`{ "type": "array", "items": { "type": "object", "resourceType": "s3object" } }\`), so files
+  sent with a message reach the agent
+- Running one needs a \`memory_id\` **query parameter** — not a flow argument — naming the
+  conversation the turn belongs to: a fresh UUID starts one, reusing a UUID continues it. The chat
+  supplies it itself; a run driven any other way has to pass it or the server refuses the job
+
 ### Tool Naming Rules
 
 These rules cover \`flowmodule\` tools, the ones the agent calls by name. A \`websearch\` tool's

system_prompts/auto-generated/skills/write-flow/SKILL.md

@@ -194,6 +194,42 @@ tool, `websearch` for web search.
   "resource": "$res:<path>", "model": <model id> }`. Required unless the module links to a saved
   agent through `value.agent`
 
+### Chat-Mode Flows
+
+A flow with `value.chat_input_enabled: true` is run from a chat instead of a form: the composer
+sends one message per turn and renders the conversation. It needs a required `user_message` string
+input, read by the agent. Any other flow input stays and is asked for under Configure inputs.
+
+```json
+{
+  "id": "chat_agent",
+  "value": {
+    "type": "aiagent",
+    "input_transforms": {
+      "provider": {
+        "type": "static",
+        "value": { "kind": "anthropic", "resource": "$res:f/ai/claude", "model": "claude-sonnet-5" }
+      },
+      "user_message": { "type": "javascript", "expr": "flow_input.user_message" },
+      "user_attachments": { "type": "javascript", "expr": "flow_input.files" },
+      "memory": { "type": "static", "value": { "kind": "auto", "context_length": 10 } },
+      "streaming": { "type": "static", "value": true },
+      "output_type": { "type": "static", "value": "text" }
+    },
+    "tools": []
+  }
+}
+```
+
+- `memory` is what lets the agent see earlier turns; without it every message starts from nothing
+- `streaming` on makes the answer and its thinking appear token by token instead of all at once
+- `user_attachments` points at a flow input typed as an array of s3 objects
+  (`{ "type": "array", "items": { "type": "object", "resourceType": "s3object" } }`), so files
+  sent with a message reach the agent
+- Running one needs a `memory_id` **query parameter** — not a flow argument — naming the
+  conversation the turn belongs to: a fresh UUID starts one, reusing a UUID continues it. The chat
+  supplies it itself; a run driven any other way has to pass it or the server refuses the job
+
 ### Tool Naming Rules
 
 These rules cover `flowmodule` tools, the ones the agent calls by name. A `websearch` tool's

system_prompts/base/flow-base.md

@@ -106,6 +106,42 @@ tool, `websearch` for web search.
   "resource": "$res:<path>", "model": <model id> }`. Required unless the module links to a saved
   agent through `value.agent`
 
+### Chat-Mode Flows
+
+A flow with `value.chat_input_enabled: true` is run from a chat instead of a form: the composer
+sends one message per turn and renders the conversation. It needs a required `user_message` string
+input, read by the agent. Any other flow input stays and is asked for under Configure inputs.
+
+```json
+{
+  "id": "chat_agent",
+  "value": {
+    "type": "aiagent",
+    "input_transforms": {
+      "provider": {
+        "type": "static",
+        "value": { "kind": "anthropic", "resource": "$res:f/ai/claude", "model": "claude-sonnet-5" }
+      },
+      "user_message": { "type": "javascript", "expr": "flow_input.user_message" },
+      "user_attachments": { "type": "javascript", "expr": "flow_input.files" },
+      "memory": { "type": "static", "value": { "kind": "auto", "context_length": 10 } },
+      "streaming": { "type": "static", "value": true },
+      "output_type": { "type": "static", "value": "text" }
+    },
+    "tools": []
+  }
+}
+```
+
+- `memory` is what lets the agent see earlier turns; without it every message starts from nothing
+- `streaming` on makes the answer and its thinking appear token by token instead of all at once
+- `user_attachments` points at a flow input typed as an array of s3 objects
+  (`{ "type": "array", "items": { "type": "object", "resourceType": "s3object" } }`), so files
+  sent with a message reach the agent
+- Running one needs a `memory_id` **query parameter** — not a flow argument — naming the
+  conversation the turn belongs to: a fresh UUID starts one, reusing a UUID continues it. The chat
+  supplies it itself; a run driven any other way has to pass it or the server refuses the job
+
 ### Tool Naming Rules
 
 These rules cover `flowmodule` tools, the ones the agent calls by name. A `websearch` tool's