feat(api): filter a conversation's messages by kind

A chat reopened while its flow is still running has to find the message
that started the turn: it carries the flow job, which is the only thing
that knows whether the run is over. An agent writes a row per round and
per tool call, so that message sits an unbounded number of pages back in
the transcript, and a client reading towards it is searching with no end
it can name.

`message_type` answers it in one request. The filter is applied inside
the paginated subquery, so `page=1&per_page=1&message_type=user` really
is the newest user row rather than the newest row that happens to be one.

`MessageType` carries a fourth kind, `System`, that nothing writes and
the Postgres enum has no label for. A request to filter on it is refused
before the database, so what reaches the caller is a word about their
request rather than an error about an enum.

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

backend/.sqlx/query-2802cb8dcabbd601c9de44c85cf512b1d4961b4bb3b8d598ee69f1c1a154bac9.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT id, conversation_id, message_type as \"message_type: MessageType\", content, job_id, created_at, created_seq, step_name, success, tool_arguments, tool_result, reasoning\n             FROM flow_conversation_message\n             WHERE conversation_id = $1\n               AND created_seq > $2\n             ORDER BY created_seq ASC\n             LIMIT $3\n             ",
+  "query": "SELECT id, conversation_id, message_type as \"message_type: MessageType\", content, job_id, created_at, created_seq, step_name, success, tool_arguments, tool_result, reasoning\n             FROM flow_conversation_message\n             WHERE conversation_id = $1\n               AND created_seq > $2\n               AND ($4::message_type IS NULL OR message_type = $4)\n             ORDER BY created_seq ASC\n             LIMIT $3\n             ",
   "describe": {
     "columns": [
       {
@@ -79,7 +79,19 @@
       "Left": [
         "Uuid",
         "Int8",
-        "Int8"
+        "Int8",
+        {
+          "Custom": {
+            "name": "message_type",
+            "kind": {
+              "Enum": [
+                "user",
+                "assistant",
+                "tool"
+              ]
+            }
+          }
+        }
       ]
     },
     "nullable": [
@@ -97,5 +109,5 @@
       true
     ]
   },
-  "hash": "318ed7a45d8326e3ecbc9727ee3812adaeeead8b39b007a32badadf67ec5ca17"
+  "hash": "2802cb8dcabbd601c9de44c85cf512b1d4961b4bb3b8d598ee69f1c1a154bac9"
 }

backend/.sqlx/query-456678fe3907343f7e991d4469ec1152788a32124ca6e9a101eca8ccea6fc840.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT id, conversation_id, message_type as \"message_type: MessageType\", content, job_id, created_at, created_seq, step_name, success, tool_arguments, tool_result, reasoning\n             FROM (\n                SELECT id, conversation_id, message_type, content, job_id, created_at, created_seq, step_name, success, tool_arguments, tool_result, reasoning\n                FROM flow_conversation_message\n                WHERE conversation_id = $1\n                ORDER BY created_seq DESC\n                LIMIT $2 OFFSET $3\n             ) AS messages\n             ORDER BY created_seq ASC\n             ",
+  "query": "SELECT id, conversation_id, message_type as \"message_type: MessageType\", content, job_id, created_at, created_seq, step_name, success, tool_arguments, tool_result, reasoning\n             FROM (\n                SELECT id, conversation_id, message_type, content, job_id, created_at, created_seq, step_name, success, tool_arguments, tool_result, reasoning\n                FROM flow_conversation_message\n                WHERE conversation_id = $1\n                  AND ($4::message_type IS NULL OR message_type = $4)\n                ORDER BY created_seq DESC\n                LIMIT $2 OFFSET $3\n             ) AS messages\n             ORDER BY created_seq ASC\n             ",
   "describe": {
     "columns": [
       {
@@ -79,7 +79,19 @@
       "Left": [
         "Uuid",
         "Int8",
-        "Int8"
+        "Int8",
+        {
+          "Custom": {
+            "name": "message_type",
+            "kind": {
+              "Enum": [
+                "user",
+                "assistant",
+                "tool"
+              ]
+            }
+          }
+        }
       ]
     },
     "nullable": [
@@ -97,5 +109,5 @@
       true
     ]
   },
-  "hash": "d6c65ce1d443d1e1783818d36c7eabd9633bbe5219bfd9683f336de19763cb58"
+  "hash": "456678fe3907343f7e991d4469ec1152788a32124ca6e9a101eca8ccea6fc840"
 }

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

@@ -67,6 +67,10 @@ pub struct ListConversationsQuery {
 #[derive(Deserialize)]
 pub struct ListMessagesQuery {
     pub after_seq: Option<i64>,
+    /// Keep only the rows of one kind. A chat reopened while a run is going asks for the
+    /// newest `user` row this way: it carries the flow job of the turn that is running, and
+    /// an agent writes enough rows per round to push it many pages back in the transcript.
+    pub message_type: Option<MessageType>,
 }
 
 async fn list_conversations(
@@ -226,6 +230,16 @@ async fn list_messages(
     Query(query): Query<ListMessagesQuery>,
 ) -> JsonResult<Vec<FlowConversationMessage>> {
     let (per_page, offset) = paginate(pagination);
+
+    // `MessageType` carries a fourth kind the column has no label for and nothing writes.
+    // Refused here rather than bound: Postgres rejects the label, and what reaches the
+    // caller then is a database error about an enum rather than a word about their request.
+    if matches!(query.message_type, Some(MessageType::System)) {
+        return Err(windmill_common::error::Error::BadRequest(
+            "message_type must be one of user, assistant, tool".to_string(),
+        ));
+    }
+
     let mut tx = user_db.clone().begin(&authed).await?;
 
     // Verify the conversation exists and belongs to the user
@@ -252,12 +266,14 @@ async fn list_messages(
              FROM flow_conversation_message
              WHERE conversation_id = $1
                AND created_seq > $2
+               AND ($4::message_type IS NULL OR message_type = $4)
              ORDER BY created_seq ASC
              LIMIT $3
              "#,
             conversation_id,
             after_seq,
-            per_page as i64
+            per_page as i64,
+            query.message_type as Option<MessageType>
         )
         .fetch_all(&mut *tx)
         .await?
@@ -270,6 +286,7 @@ async fn list_messages(
                 SELECT id, conversation_id, message_type, content, job_id, created_at, created_seq, step_name, success, tool_arguments, tool_result, reasoning
                 FROM flow_conversation_message
                 WHERE conversation_id = $1
+                  AND ($4::message_type IS NULL OR message_type = $4)
                 ORDER BY created_seq DESC
                 LIMIT $2 OFFSET $3
              ) AS messages
@@ -277,7 +294,8 @@ async fn list_messages(
              "#,
             conversation_id,
             per_page as i64,
-            offset as i64
+            offset as i64,
+            query.message_type as Option<MessageType>
         )
         .fetch_all(&mut *tx)
         .await?

backend/windmill-api/openapi.yaml

@@ -12507,6 +12507,16 @@ paths:
           schema:
             type: integer
             format: int64
+        - name: message_type
+          description: Keep only the messages of that kind
+          in: query
+          required: false
+          schema:
+            type: string
+            # The labels the MESSAGE_TYPE column actually has. The response schema below
+            # carries a fourth, `system`, which nothing writes and the column cannot hold;
+            # the handler refuses a request to filter on it.
+            enum: [user, assistant, tool]
       responses:
         "200":
           description: conversation messages