From 23c24a9688d4c8c462f53221334d538280f16bca Mon Sep 17 00:00:00 2001 From: Guilhem Date: Thu, 17 Sep 2026 09:27:12 +0200 Subject: [PATCH 1/2] feat: collect flow conversations and agent memory once their last message goes (#11178) * feat: collect flow conversations and agent memory when their last message goes * fix: lock the conversation lookup so a new turn orders against its cleanup * fix: let concurrent turns recreate a collected conversation without conflicting --- ...9a1fabeaa167818206a25abfe31d5582f942a.json | 15 ++ ...7cb3b31ee6b69abae79a214dddba0dee4425c.json | 15 -- ...0054b753cfbeb5dad1e8120fd5e8a672d50ef.json | 23 ++ ...3659915d9fcb3d58f16532aaffe06c44ec976.json | 23 ++ ...3ca33b3f82ca38289dfa4a69aa95041027d57.json | 59 +++++ ...9672f8c98d2579fc72c07752361cc5fd683dc.json | 15 ++ ...8e44eb1c0062a8241dc8fe0ed3dd95b65f89a.json | 14 + ...df90dad5246681bde6caaa633b85a5e8b2352.json | 22 ++ ...741a51ca0ce66bf1c674097705cf2e1b72eff.json | 14 + ...2893147b4e5ff078531de36147d908132d636.json | 14 - ...fee4a74f61b061e5025ecb75882208b3ba8f.json} | 4 +- ...7a30d71a5689efe72daf58bd1e8f6cf90c410.json | 22 ++ backend/tests/v2_job_delete_orphans.rs | 248 +++++++++++++++++- backend/windmill-api-jobs/src/execution.rs | 14 +- backend/windmill-api-jobs/src/jobs_export.rs | 60 ++++- backend/windmill-api/src/jobs.rs | 1 + .../windmill-common/src/flow_conversations.rs | 55 ++-- backend/windmill-common/src/jobs.rs | 55 +++- 18 files changed, 614 insertions(+), 59 deletions(-) create mode 100644 backend/.sqlx/query-07a005f0f9e80a156cd2a5a0ae39a1fabeaa167818206a25abfe31d5582f942a.json delete mode 100644 backend/.sqlx/query-462d2b2822b185a6f51fafcfa957cb3b31ee6b69abae79a214dddba0dee4425c.json create mode 100644 backend/.sqlx/query-4f52bf546579f26a1d22c239b8b0054b753cfbeb5dad1e8120fd5e8a672d50ef.json create mode 100644 backend/.sqlx/query-69bfbe9b39414b724488532cc3b3659915d9fcb3d58f16532aaffe06c44ec976.json create mode 100644 backend/.sqlx/query-6f32c1feed096ff706ae359ad6a3ca33b3f82ca38289dfa4a69aa95041027d57.json create mode 100644 backend/.sqlx/query-89ea81b765550cf665e30533efc9672f8c98d2579fc72c07752361cc5fd683dc.json create mode 100644 backend/.sqlx/query-90b910da8d00a7c7bcf29c167e38e44eb1c0062a8241dc8fe0ed3dd95b65f89a.json create mode 100644 backend/.sqlx/query-967f52005f4a044b3a2e9f02ceadf90dad5246681bde6caaa633b85a5e8b2352.json create mode 100644 backend/.sqlx/query-a1b23f3e62c6433d95cdac58215741a51ca0ce66bf1c674097705cf2e1b72eff.json delete mode 100644 backend/.sqlx/query-bfdd60b42e32bd81e2d20b327462893147b4e5ff078531de36147d908132d636.json rename backend/.sqlx/{query-6bd23a98838e3eec309e6b696edc776bd56fc9dae1238b3272557d1562400dbe.json => query-c1e3ed3ecc3bcb98f60ba8196d33fee4a74f61b061e5025ecb75882208b3ba8f.json} (84%) create mode 100644 backend/.sqlx/query-ec295b3890a0018475ec0a3774c7a30d71a5689efe72daf58bd1e8f6cf90c410.json diff --git a/backend/.sqlx/query-07a005f0f9e80a156cd2a5a0ae39a1fabeaa167818206a25abfe31d5582f942a.json b/backend/.sqlx/query-07a005f0f9e80a156cd2a5a0ae39a1fabeaa167818206a25abfe31d5582f942a.json new file mode 100644 index 0000000000..635ec2b94d --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-462d2b2822b185a6f51fafcfa957cb3b31ee6b69abae79a214dddba0dee4425c.json b/backend/.sqlx/query-462d2b2822b185a6f51fafcfa957cb3b31ee6b69abae79a214dddba0dee4425c.json deleted file mode 100644 index 0c4d90b073..0000000000 --- a/backend/.sqlx/query-462d2b2822b185a6f51fafcfa957cb3b31ee6b69abae79a214dddba0dee4425c.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-4f52bf546579f26a1d22c239b8b0054b753cfbeb5dad1e8120fd5e8a672d50ef.json b/backend/.sqlx/query-4f52bf546579f26a1d22c239b8b0054b753cfbeb5dad1e8120fd5e8a672d50ef.json new file mode 100644 index 0000000000..a5695f75fd --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-69bfbe9b39414b724488532cc3b3659915d9fcb3d58f16532aaffe06c44ec976.json b/backend/.sqlx/query-69bfbe9b39414b724488532cc3b3659915d9fcb3d58f16532aaffe06c44ec976.json new file mode 100644 index 0000000000..52105ca605 --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-6f32c1feed096ff706ae359ad6a3ca33b3f82ca38289dfa4a69aa95041027d57.json b/backend/.sqlx/query-6f32c1feed096ff706ae359ad6a3ca33b3f82ca38289dfa4a69aa95041027d57.json new file mode 100644 index 0000000000..0e92e3aa99 --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-89ea81b765550cf665e30533efc9672f8c98d2579fc72c07752361cc5fd683dc.json b/backend/.sqlx/query-89ea81b765550cf665e30533efc9672f8c98d2579fc72c07752361cc5fd683dc.json new file mode 100644 index 0000000000..7a7110030c --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-90b910da8d00a7c7bcf29c167e38e44eb1c0062a8241dc8fe0ed3dd95b65f89a.json b/backend/.sqlx/query-90b910da8d00a7c7bcf29c167e38e44eb1c0062a8241dc8fe0ed3dd95b65f89a.json new file mode 100644 index 0000000000..1555f3683d --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-967f52005f4a044b3a2e9f02ceadf90dad5246681bde6caaa633b85a5e8b2352.json b/backend/.sqlx/query-967f52005f4a044b3a2e9f02ceadf90dad5246681bde6caaa633b85a5e8b2352.json new file mode 100644 index 0000000000..5161f716ff --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-a1b23f3e62c6433d95cdac58215741a51ca0ce66bf1c674097705cf2e1b72eff.json b/backend/.sqlx/query-a1b23f3e62c6433d95cdac58215741a51ca0ce66bf1c674097705cf2e1b72eff.json new file mode 100644 index 0000000000..d8fdd5140b --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-bfdd60b42e32bd81e2d20b327462893147b4e5ff078531de36147d908132d636.json b/backend/.sqlx/query-bfdd60b42e32bd81e2d20b327462893147b4e5ff078531de36147d908132d636.json deleted file mode 100644 index f9fbc7a58b..0000000000 --- a/backend/.sqlx/query-bfdd60b42e32bd81e2d20b327462893147b4e5ff078531de36147d908132d636.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-6bd23a98838e3eec309e6b696edc776bd56fc9dae1238b3272557d1562400dbe.json b/backend/.sqlx/query-c1e3ed3ecc3bcb98f60ba8196d33fee4a74f61b061e5025ecb75882208b3ba8f.json similarity index 84% rename from backend/.sqlx/query-6bd23a98838e3eec309e6b696edc776bd56fc9dae1238b3272557d1562400dbe.json rename to backend/.sqlx/query-c1e3ed3ecc3bcb98f60ba8196d33fee4a74f61b061e5025ecb75882208b3ba8f.json index d7f5fc45d4..50ba9d2897 100644 --- a/backend/.sqlx/query-6bd23a98838e3eec309e6b696edc776bd56fc9dae1238b3272557d1562400dbe.json +++ b/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" } diff --git a/backend/.sqlx/query-ec295b3890a0018475ec0a3774c7a30d71a5689efe72daf58bd1e8f6cf90c410.json b/backend/.sqlx/query-ec295b3890a0018475ec0a3774c7a30d71a5689efe72daf58bd1e8f6cf90c410.json new file mode 100644 index 0000000000..1e35bd15ac --- /dev/null +++ b/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" +} diff --git a/backend/tests/v2_job_delete_orphans.rs b/backend/tests/v2_job_delete_orphans.rs index 95cd1673b6..ded760b8ff 100644 --- a/backend/tests/v2_job_delete_orphans.rs +++ b/backend/tests/v2_job_delete_orphans.rs @@ -37,7 +37,7 @@ async fn seed_side_rows(db: &Pool, 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) -> anyhow::Resul Ok(()) } +/// (conversation rows, agent-memory rows) for one conversation. +async fn conversation_and_memory_counts( + db: &Pool, + 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, +) -> 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, +) -> 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) -> anyhow::Result<()> { initialize_tracing().await; @@ -192,6 +370,74 @@ async fn test_workspace_delete_removes_side_rows(db: Pool) -> 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, +) -> 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 diff --git a/backend/windmill-api-jobs/src/execution.rs b/backend/windmill-api-jobs/src/execution.rs index 310e44ae1f..c15068583d 100644 --- a/backend/windmill-api-jobs/src/execution.rs +++ b/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>, + 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?; } diff --git a/backend/windmill-api-jobs/src/jobs_export.rs b/backend/windmill-api-jobs/src/jobs_export.rs index 05f5a51bbc..6389134031 100644 --- a/backend/windmill-api-jobs/src/jobs_export.rs +++ b/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 = 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; diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 84968ebbcc..bcc2fed94f 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/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?; } diff --git a/backend/windmill-common/src/flow_conversations.rs b/backend/windmill-common/src/flow_conversations.rs index 21b1f56389..b62f768bbc 100644 --- a/backend/windmill-common/src/flow_conversations.rs +++ b/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 { - // 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> { + 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 diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 5541615838..6f3248f3fb 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -478,6 +478,12 @@ pub static WORKER_INTERNAL_SERVER_INLINE_UTILS: OnceCell 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 = 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?; From 68f2248018fc218a090bf939e1eb22ff97d5bc22 Mon Sep 17 00:00:00 2001 From: Guilhem Date: Thu, 17 Sep 2026 09:29:05 +0200 Subject: [PATCH 2/2] feat: keep flow inputs and seed the agent when chat mode is enabled (#11177) * feat: keep flow inputs and seed the agent when chat mode is enabled Co-Authored-By: Claude Fable 5.1 * fix: leave a linked agent's memory and streaming to the agent when enabling chat mode Co-Authored-By: Claude Fable 5.1 --------- Co-authored-by: Claude Fable 5.1 --- CONTEXT.md | 19 + cli/src/guidance/skills.gen.ts | 36 ++ frontend/src/lib/common.ts | 2 + frontend/src/lib/components/ArgInput.svelte | 7 + .../lib/components/EditableSchemaForm.svelte | 9 +- .../lib/components/InputTransformForm.svelte | 5 + .../components/flows/content/FlowInput.svelte | 330 +++++++++++------- .../src/lib/components/flows/flowInfers.ts | 6 + system_prompts/auto-generated/flow.md | 36 ++ system_prompts/auto-generated/prompts.ts | 36 ++ .../auto-generated/skills/write-flow/SKILL.md | 36 ++ system_prompts/base/flow-base.md | 36 ++ 12 files changed, 433 insertions(+), 125 deletions(-) diff --git a/CONTEXT.md b/CONTEXT.md index 64aa1b93d8..fbc2592cc9 100644 --- a/CONTEXT.md +++ b/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**: diff --git a/cli/src/guidance/skills.gen.ts b/cli/src/guidance/skills.gen.ts index f58092fdac..e4490276ce 100644 --- a/cli/src/guidance/skills.gen.ts +++ b/cli/src/guidance/skills.gen.ts @@ -5280,6 +5280,42 @@ tool, \`websearch\` for web search. "resource": "$res:", "model": }\`. 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 diff --git a/frontend/src/lib/common.ts b/frontend/src/lib/common.ts index 0d76b1a0b8..4f82fd252e 100644 --- a/frontend/src/lib/common.ts +++ b/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 diff --git a/frontend/src/lib/components/ArgInput.svelte b/frontend/src/lib/components/ArgInput.svelte index b7001059c2..eaca31c761 100644 --- a/frontend/src/lib/components/ArgInput.svelte +++ b/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}
+ {#if oneOfLockedReason !== undefined} +
{oneOfLockedReason}
+ {/if} {#if oneOf && oneOf.length >= 2} { oneOfSelected = detail const selectedObjProperties = diff --git a/frontend/src/lib/components/EditableSchemaForm.svelte b/frontend/src/lib/components/EditableSchemaForm.svelte index 67dfeed0b8..bd3399d9d2 100644 --- a/frontend/src/lib/components/EditableSchemaForm.svelte +++ b/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 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)}
@@ -605,7 +609,7 @@ >
{argName} - {#if !uiOnly} + {#if !uiOnly && !locked}
{#snippet trigger()} @@ -654,7 +658,7 @@ Required {/if} - {#if !uiOnly} + {#if !uiOnly && !locked} + + {#snippet children({ item })} + + + {/snippet} + {/if}
{/if} @@ -696,11 +796,15 @@
{#if flowStore.val.value?.chat_input_enabled}
- {#if showAdditionalInputs} -
+ {#if chatPanelTab === 'inputs'} + +
{#snippet openEditTab()} - + {@render inputsAddTrigger()} {/snippet} {/snippet}
{/if} - + +
+ +
{:else}
@@ -815,22 +917,9 @@
{#snippet close_button()} -
@@ -884,12 +973,7 @@ }} > {#snippet trigger()} -
- -
+ {@render inputsAddTrigger()} {/snippet} {/if} diff --git a/frontend/src/lib/components/flows/flowInfers.ts b/frontend/src/lib/components/flows/flowInfers.ts index bce923c77a..3c38682350 100644 --- a/frontend/src/lib/components/flows/flowInfers.ts +++ b/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: [ { diff --git a/system_prompts/auto-generated/flow.md b/system_prompts/auto-generated/flow.md index dd0048ec9c..8518068992 100644 --- a/system_prompts/auto-generated/flow.md +++ b/system_prompts/auto-generated/flow.md @@ -106,6 +106,42 @@ tool, `websearch` for web search. "resource": "$res:", "model": }`. 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 diff --git a/system_prompts/auto-generated/prompts.ts b/system_prompts/auto-generated/prompts.ts index 1e92c8e69b..9ae13dafbe 100644 --- a/system_prompts/auto-generated/prompts.ts +++ b/system_prompts/auto-generated/prompts.ts @@ -137,6 +137,42 @@ tool, \`websearch\` for web search. "resource": "$res:", "model": }\`. 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 diff --git a/system_prompts/auto-generated/skills/write-flow/SKILL.md b/system_prompts/auto-generated/skills/write-flow/SKILL.md index 3556cbfff0..5988dd2f01 100644 --- a/system_prompts/auto-generated/skills/write-flow/SKILL.md +++ b/system_prompts/auto-generated/skills/write-flow/SKILL.md @@ -194,6 +194,42 @@ tool, `websearch` for web search. "resource": "$res:", "model": }`. 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 diff --git a/system_prompts/base/flow-base.md b/system_prompts/base/flow-base.md index e70ac3d42a..a102710911 100644 --- a/system_prompts/base/flow-base.md +++ b/system_prompts/base/flow-base.md @@ -106,6 +106,42 @@ tool, `websearch` for web search. "resource": "$res:", "model": }`. 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