feat(ai-chat): let a purged conversation go, and bound a renamed title

A chat's messages are deleted with the jobs behind them, and the user row now
carries its run, so retention can empty a conversation completely. When it does,
the `flow_conversation` row and the agent memory keyed on it are all that is
left, and nothing else collects them — `ai_agent_memory` has no job id for
retention to match on. Both delete paths (the retention sweep in
`windmill_common::jobs` and the `/jobs/delete` endpoint) now drop a conversation
that has no messages left, and its memory with it. The trigger is the emptiness,
so a partly-purged chat keeps what it still has.

Two statements rather than one CTE: a data-modifying CTE reads the snapshot from
before the message delete, so every conversation would still look non-empty.
Both halves are pinned by a test on each path.

Also from review: the rename endpoint bounds the title three short of the
column, since the helper appends an ellipsis; and `OpenAIResponsesSSEParser`
loses the reasoning field it never wrote — that stream has no reasoning-summary
event, so the value could only ever be empty.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QN7VboDEm9HAB1t4sMxMdE

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-5b9c9eb64051f291fed4be9bc0b0cc0aef2e7bde732899976eddac36a2da7658.json

@@ -0,0 +1,24 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE flow_conversation SET title = $1, updated_at = updated_at\n         WHERE id = $2 AND workspace_id = $3\n         RETURNING id",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "5b9c9eb64051f291fed4be9bc0b0cc0aef2e7bde732899976eddac36a2da7658"
+}

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-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/summarized_schema.txt

@@ -97,9 +97,9 @@ email_trigger: path(char), local_part(char), workspaced_local_part(bool), script
 favorite: usr(char), workspace_id(char), path(char), favorite_kind(favorite_kind)
 flow: workspace_id(char), path(char), summary(text), description(text), value(jsonb), edited_by(char), edited_at(ts), archived(bool), schema(json), extra_perms(jsonb), dependency_job(uuid), draft_only(bool), tag(char), ws_error_handler_muted(bool), dedicated_worker(bool), timeout(int), visible_to_runner_only(bool), concurrency_key(char), versions(bigint[]), on_behalf_of(varchar), on_behalf_of_email(text), lock_error_logs(text), labels(text[])
   FK: (workspace_id) -> workspace(id)
-flow_conversation: id(uuid), workspace_id(char), flow_path(char), title(char), created_at(ts), updated_at(ts), created_by(char)
+flow_conversation: id(uuid), workspace_id(char), flow_path(char), title(char), created_at(ts), updated_at(ts), created_by(char), is_test(bool)
   FK: (workspace_id) -> workspace(id)
-flow_conversation_message: id(uuid), conversation_id(uuid), message_type(message_type), content(text), job_id(uuid), created_at(ts), created_seq(int8), step_name(char), success(bool)
+flow_conversation_message: id(uuid), conversation_id(uuid), message_type(message_type), content(text), job_id(uuid), created_at(ts), created_seq(int8), step_name(char), success(bool), tool_arguments(text), tool_result(text), reasoning(text)
   FK: (conversation_id) -> flow_conversation(id) | (job_id) -> v2_job(id)
 flow_iterator_data: job_id(uuid), itered(jsonb)
 flow_node: id(bigint), workspace_id(char), hash(bigint), path(char), lock(text), code(text), flow(jsonb), hash_v2(char(64))

backend/tests/v2_job_delete_orphans.rs

@@ -121,6 +121,90 @@ 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(())
+}
+
 #[sqlx::test(fixtures("base"))]
 async fn test_clear_schedule_removes_side_rows(db: Pool<Postgres>) -> anyhow::Result<()> {
     initialize_tracing().await;
@@ -196,6 +280,75 @@ async fn test_workspace_delete_removes_side_rows(db: Pool<Postgres>) -> anyhow::
 /// 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
 /// the scoping has to live in each explicit delete).
+/// The purge endpoint carries its own copy of the emptied-conversation rule, so it gets the
+/// same guard: the conversation and its memory go only with the last message, and only in
+/// the caller's workspace.
+#[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(())
+}
+
 #[sqlx::test(fixtures("base"))]
 async fn test_jobs_export_delete_is_workspace_scoped(db: Pool<Postgres>) -> anyhow::Result<()> {
     initialize_tracing().await;

backend/windmill-ai/src/providers/openai.rs

@@ -538,8 +538,9 @@ impl QueryBuilder for OpenAIQueryBuilder {
             } else {
                 Some(parser.accumulated_content)
             },
-            reasoning: (!parser.accumulated_reasoning.is_empty())
-                .then_some(parser.accumulated_reasoning),
+            // The Responses stream has no reasoning-summary event in
+            // `OpenAIResponsesSSEEvent`, so nothing thinks out loud on this path yet.
+            reasoning: None,
             tool_calls: parser.accumulated_tool_calls.into_values().collect(),
             events_str: Some(parser.events_str),
             annotations: parser.annotations,

backend/windmill-ai/src/sse.rs

@@ -823,8 +823,6 @@ pub enum OpenAIResponsesSSEEvent {
 /// OpenAI Responses API SSE Parser for streaming responses
 pub struct OpenAIResponsesSSEParser {
     pub accumulated_content: String,
-    /// The thinking streamed before the answer, kept so it can be stored with it.
-    pub accumulated_reasoning: String,
     pub accumulated_tool_calls: HashMap<String, OpenAIToolCall>,
     /// Maps item_id -> (name, call_id) for function calls
     tool_call_metadata: HashMap<String, (String, String)>,
@@ -844,7 +842,6 @@ impl OpenAIResponsesSSEParser {
     pub fn new(stream_event_processor: Box<dyn StreamEventSink>) -> Self {
         Self {
             accumulated_content: String::new(),
-            accumulated_reasoning: String::new(),
             accumulated_tool_calls: HashMap::new(),
             tool_call_metadata: HashMap::new(),
             tool_call_arguments: HashMap::new(),

backend/windmill-api-flow-conversations/src/lib.rs

@@ -15,7 +15,7 @@ use windmill_common::{
     db::{UserDB, DB},
     error::{JsonResult, Result},
     flow_conversations::MessageType,
-    utils::{not_found_if_none, paginate, Pagination},
+    utils::{not_found_if_none, paginate, truncate_with_ellipsis, Pagination},
 };
 
 pub fn workspaced_service() -> Router {
@@ -193,13 +193,18 @@ async fn update_conversation(
     Path((w_id, conversation_id)): Path<(String, Uuid)>,
     Json(update): Json<UpdateConversation>,
 ) -> Result<String> {
+    // The column is VARCHAR(255) and the helper appends an ellipsis to what it cuts, so the
+    // bound it takes is three short of the column's. A longer title would otherwise reach
+    // Postgres as a 22001 and come back a 500.
+    let title = truncate_with_ellipsis(update.title.trim(), 252);
+
     let mut tx = user_db.clone().begin(&authed).await?;
 
     let updated = sqlx::query_scalar!(
         "UPDATE flow_conversation SET title = $1, updated_at = updated_at
          WHERE id = $2 AND workspace_id = $3
          RETURNING id",
-        update.title.trim(),
+        title,
         conversation_id,
         &w_id
     )

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

@@ -692,16 +692,52 @@ pub async fn delete_jobs(
     .await?
     .rows_affected();
 
-    let conversation_message_deleted = sqlx::query!(
+    let emptied_conversations: 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 = emptied_conversations.len() as u64;
+
+    // Same rule as retention (windmill_common::jobs::delete_jobs): a conversation with no
+    // messages left goes, and the agent's memory for it with it.
+    let mut conversation_ids = emptied_conversations;
+    conversation_ids.sort_unstable();
+    conversation_ids.dedup();
+    if !conversation_ids.is_empty() {
+        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?;
+        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?;
+    }
 
     // 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.

backend/windmill-common/src/jobs.rs

@@ -485,12 +485,45 @@ 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.
+        // 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?;