feat(team): labels on a teammate message, and on the teammate's list (#1637)

apps/fountain/lib/fountain/conversations.ex

@@ -732,6 +732,10 @@ defmodule Fountain.Conversations do
   for `limit: n` — which the console's dashboard uses to ask for the five it
   shows instead of every row a busy account has.
 
+  `labels: %{"env" => "prod"}` (#1637) keeps the conversations carrying every
+  one of those pairs — jsonb containment, so a row with more labels than the
+  filter names still matches, and the GIN index on the column serves it.
+
   Populates the `last_active_at` virtual field using `kind: "output"` log
   events only — stage events (reconnects, lifecycle) are excluded so
   reconnects don't produce false unread indicators.
@@ -765,6 +769,9 @@ defmodule Fountain.Conversations do
         {:channel_id, id}, q when is_binary(id) and id != "" ->
           where(q, [conv: c], c.channel_id == ^id)
 
+        {:labels, labels}, q when is_map(labels) and map_size(labels) > 0 ->
+          where(q, [conv: c], fragment("? @> ?", c.labels, type(^labels, :map)))
+
         {:status, [_ | _] = statuses}, q ->
           where(q, [conv: c], c.status in ^statuses)
 
@@ -787,16 +794,25 @@ defmodule Fountain.Conversations do
   the surface reading a channel — the team page — wants the last transcript
   even when nothing is running.
   """
-  def list_channel_conversations(user_id, channel_id)
+  def list_channel_conversations(user_id, channel_id, opts \\ [])
       when is_binary(user_id) and is_binary(channel_id) do
     from(c in annotated_query(user_id),
       where: c.channel_id == ^channel_id,
       order_by: [desc: c.inserted_at, desc: c.id]
     )
+    |> filter_by_labels(Keyword.get(opts, :labels))
     |> Repo.all()
     |> Repo.preload([:agent, :sandbox])
   end
 
+  # The same containment filter `list_conversations/2` applies (#1637), for
+  # the channel-bound list behind the team route.
+  defp filter_by_labels(query, labels) when is_map(labels) and map_size(labels) > 0 do
+    where(query, [conv: c], fragment("? @> ?", c.labels, type(^labels, :map)))
+  end
+
+  defp filter_by_labels(query, _labels), do: query
+
   @doc """
   Scoped fetch that also populates the read-model annotations —
   `turn_count` and `last_active_at` — which `get_conversation/2` leaves at

apps/fountain/lib/fountain/conversations/labels.ex

@@ -27,6 +27,13 @@ defmodule Fountain.Conversations.Labels do
   whose value is `null` is removed. That is what lets a run add one label
   without reading the others first, and what makes a channel resume keep the
   labels the binding already carried.
+
+  ## The list filter
+
+  `GET /api/conversations?label=env:prod&label=drift:true` is repeatable and
+  AND-combined. Each value splits on its **first** colon only, so
+  `label=path:a:b` filters `path` for `a:b`. The query is jsonb containment,
+  which the GIN index on the column serves.
   """
 
   import Ecto.Changeset, only: [get_change: 2, add_error: 3]
@@ -254,4 +261,41 @@ defmodule Fountain.Conversations.Labels do
      |> Enum.filter(&Map.has_key?(current, &1))
      |> Enum.sort()}
   end
+
+  @doc """
+  Every `label` value in a raw query string, in the order they were sent.
+
+  Read from the query string rather than from the parsed params because Plug
+  collapses a repeated key to its last value, and the filter is repeatable by
+  design. `label[]=` is accepted as well, for a client whose HTTP layer only
+  builds arrays that way.
+  """
+  @spec from_query_string(String.t() | nil) :: [String.t()]
+  def from_query_string(nil), do: []
+
+  def from_query_string(query) when is_binary(query) do
+    for {key, value} <- URI.query_decoder(query), key in ["label", "label[]"], do: value
+  end
+
+  @doc """
+  Turn repeated `key:value` filter values into the map the list query
+  contains against. Splits on the first colon only, so a value may contain
+  colons of its own.
+
+  `{:error, :invalid_label_filter}` for a value with no colon or an empty
+  key: a caller who typed `?label=prod` meant something, and matching
+  everything would be the wrong guess.
+  """
+  @spec parse_filter([String.t()] | nil) :: {:ok, map()} | {:error, :invalid_label_filter}
+  def parse_filter(nil), do: {:ok, %{}}
+
+  def parse_filter(values) when is_list(values) do
+    Enum.reduce_while(values, {:ok, %{}}, fn value, {:ok, acc} ->
+      case value |> to_string() |> String.trim() |> String.split(":", parts: 2) do
+        ["" | _] -> {:halt, {:error, :invalid_label_filter}}
+        [key, filter_value] -> {:cont, {:ok, Map.put(acc, key, filter_value)}}
+        [_no_colon] -> {:halt, {:error, :invalid_label_filter}}
+      end
+    end)
+  end
 end

apps/fountain/lib/fountain/team.ex

@@ -373,17 +373,38 @@ defmodule Fountain.Team do
 
       %{conversation: conv} ->
         if live?(conv) do
-          case ConversationServer.send_prompt(conv.id, text, images, opts) do
-            :ok -> {:ok, Conversations.get_conversation(conv.id, user_id) || conv}
-            {:error, :gone} -> start_fresh(user_id, agent_id, conv, text, images, opts)
-            {:error, _} = err -> err
+          # Labels before the prompt (#1637): merged first, so a label the
+          # limits refuse means nothing happened at all rather than "the
+          # message went and the labels did not".
+          with {:ok, conv} <- label(conv, opts) do
+            case ConversationServer.send_prompt(conv.id, text, images, opts) do
+              :ok -> {:ok, Conversations.get_conversation(conv.id, user_id) || conv}
+              {:error, :gone} -> start_fresh(user_id, agent_id, conv, text, images, opts)
+              {:error, _} = err -> err
+            end
           end
         else
           start_fresh(user_id, agent_id, conv, text, images, opts)
         end
     end
   end
 
+  # `opts[:labels]` goes onto the conversation the message lands on (#1637).
+  # On the fresh path they ride in the create attrs instead, so a conversation
+  # that never existed is not labelled twice.
+  #
+  # Through `Conversations.set_conversation_labels/4`, not the writer beneath
+  # it: this route accepts a sandbox's own `sprite` token, and the teammate's
+  # conversation is somebody else's conversation as far as that token is
+  # concerned. The door is where the rule lives, so the refusal is the same
+  # one `PATCH .../labels` gives.
+  defp label(%Conversation{} = conv, opts) do
+    case Keyword.get(opts, :labels) do
+      nil -> {:ok, conv}
+      labels -> Conversations.set_conversation_labels(conv.id, conv.user_id, labels, opts)
+    end
+  end
+
   # A new conversation under the team binding, seeded with the message. Not
   # `start_or_resume`: we are here precisely because the bound conversation
   # cannot be resumed, and `find_channel_conversation` would agree — but
@@ -401,7 +422,8 @@ defmodule Fountain.Team do
           "images" => images,
           "title" => prev.title,
           "environment_id" => prev.environment_id,
-          "vault_id" => prev.vault_id
+          "vault_id" => prev.vault_id,
+          "labels" => Keyword.get(opts, :labels) || %{}
         },
         opts
       )
@@ -784,10 +806,10 @@ defmodule Fountain.Team do
   first (`live?/1`, else the newest), then the retired ones newest first — a previous computer's thread, still bound to the channel until the
   teammate is removed. `[]` when the agent is not on the team.
   """
-  def list_teammate_conversations(user_id, agent_id)
+  def list_teammate_conversations(user_id, agent_id, opts \\ [])
       when is_binary(user_id) and is_binary(agent_id) do
     case user_id
-         |> Conversations.list_channel_conversations(@channel)
+         |> Conversations.list_channel_conversations(@channel, opts)
          |> Enum.filter(&(&1.agent_id == agent_id)) do
       [] ->
         []

apps/fountain/lib/fountain_web/controllers/conversation_controller.ex

@@ -9,11 +9,17 @@ defmodule FountainWeb.ConversationController do
   alias Fountain.Conversations
   alias Fountain.Conversations.{ConversationServer, LogEvent}
   alias FountainWeb.Audited
+  alias FountainWeb.LabelFilter
   alias FountainWeb.SandboxKey
   alias FountainWeb.Schemas
 
   action_fallback FountainWeb.FallbackController
 
+  # Before the cast, and only where the parameter exists: `label` is a
+  # repeated key, and the cast reads query parameters out of Plug, which
+  # keeps only the last of them. See the plug's moduledoc.
+  plug FountainWeb.Plugs.RepeatedQueryParam, "label" when action in [:index]
+
   plug OpenApiSpex.Plug.CastAndValidate,
     replace_params: false,
     render_error: FountainWeb.Plugs.CastRenderError
@@ -68,6 +74,20 @@ defmodule FountainWeb.ConversationController do
             "400 outside that range). Without it the whole list is returned, which on a " <>
             "busy account is hundreds of rows per call — a client that needs one " <>
             "conversation should filter (`agent_id`, `sandbox_id`, `channel_id`) and cap."
+      ],
+      label: [
+        in: :query,
+        schema: %OpenApiSpex.Schema{type: :array, items: %OpenApiSpex.Schema{type: :string}},
+        style: :form,
+        explode: true,
+        required: false,
+        description:
+          "Only conversations carrying these `key:value` labels (#1637). Repeat the " <>
+            "parameter to combine them with AND: `?label=env:prod&label=drift:true` keeps " <>
+            "the conversations that carry both. `label[]=` is accepted as well. Each value " <>
+            "splits on its first colon only, so `label=path:a:b` matches the label `path` " <>
+            "with the value `a:b`. 400 `invalid_label_filter` on a value with no colon or " <>
+            "an empty key."
       ]
     ],
     responses: [
@@ -82,7 +102,8 @@ defmodule FountainWeb.ConversationController do
     roots_only = parse_bool_param(params["roots_only"], false)
 
     with {:ok, statuses} <- parse_statuses(params["status"]),
-         {:ok, limit} <- parse_list_limit(params["limit"]) do
+         {:ok, limit} <- parse_list_limit(params["limit"]),
+         {:ok, labels} <- LabelFilter.from(conn) do
       render(conn, :index,
         conversations:
           Conversations.list_conversations(user.id,
@@ -91,7 +112,8 @@ defmodule FountainWeb.ConversationController do
             channel_id: params["channel_id"],
             sandbox_id: params["sandbox_id"],
             status: statuses,
-            limit: limit
+            limit: limit,
+            labels: labels
           )
       )
     end

apps/fountain/lib/fountain_web/controllers/team_controller.ex

@@ -19,6 +19,9 @@ defmodule FountainWeb.TeamController do
 
   action_fallback FountainWeb.FallbackController
 
+  # Before the cast: `label` is a repeated key. See the plug's moduledoc.
+  plug FountainWeb.Plugs.RepeatedQueryParam, "label" when action in [:conversations]
+
   plug OpenApiSpex.Plug.CastAndValidate,
     replace_params: false,
     render_error: FountainWeb.Plugs.CastRenderError
@@ -120,25 +123,36 @@ defmodule FountainWeb.TeamController do
         "thread, read-only — behind it. Each is a full conversation object; read a " <>
         "retired thread with `GET /api/conversations/:id/events`. 404 when the agent is " <>
         "not on the team.",
-    parameters: [agent_id: [in: :path, type: :string, required: true]],
+    parameters: [
+      agent_id: [in: :path, type: :string, required: true],
+      label: [
+        in: :query,
+        schema: %OpenApiSpex.Schema{type: :array, items: %OpenApiSpex.Schema{type: :string}},
+        style: :form,
+        explode: true,
+        required: false,
+        description:
+          "Only conversations carrying these `key:value` labels (#1637). Repeatable and " <>
+            "AND-combined, exactly as on `GET /api/conversations`. 400 " <>
+            "`invalid_label_filter` on a value with no colon or an empty key."
+      ]
+    ],
     responses: [
       ok: {"Conversations", "application/json", Schemas.TeammateConversationListResponse},
+      bad_request: {"Invalid label filter", "application/json", Schemas.Error},
       not_found: {"Not on the team", "application/json", Schemas.Error}
     ]
   )
 
   def conversations(conn, %{"agent_id" => agent_id}) do
     user = conn.assigns.current_user
 
-    case Team.get_teammate(user.id, agent_id) do
-      nil ->
-        {:error, :not_found}
-
-      %{conversation: current} ->
-        render(conn, :conversations,
-          conversations: Team.list_teammate_conversations(user.id, agent_id),
-          current_id: current.id
-        )
+    with {:ok, labels} <- FountainWeb.LabelFilter.from(conn),
+         %{conversation: current} <- Team.get_teammate(user.id, agent_id) || {:error, :not_found} do
+      render(conn, :conversations,
+        conversations: Team.list_teammate_conversations(user.id, agent_id, labels: labels),
+        current_id: current.id
+      )
     end
   end
 
@@ -370,21 +384,30 @@ defmodule FountainWeb.TeamController do
         "seeded with this message, so the response names the conversation the message " <>
         "went to. 400 `conversation_busy` while the previous turn is still running " <>
         "(the same shape as `POST /api/conversations/:id/prompts`), 503 while the " <>
-        "computer is still starting.",
+        "computer is still starting.\n\n" <>
+        "`labels` merges onto the conversation the message lands on (#1637), before the " <>
+        "turn is queued, so a label the limits refuse leaves the message unsent.",
     parameters: [agent_id: [in: :path, type: :string, required: true]],
     request_body: {"Message", "application/json", Schemas.TeamMessageRequest},
     responses: [
       accepted: {"Queued", "application/json", Schemas.TeamMessageResponse},
       not_found: {"Not on the team", "application/json", Schemas.Error},
+      unprocessable_entity: {"Invalid labels", "application/json", Schemas.ChangesetError},
+      forbidden:
+        {"A sandbox token labelling another conversation", "application/json", Schemas.Error},
       bad_request: {"A turn is still running", "application/json", Schemas.Error}
     ]
   )
 
   def message(conn, %{"agent_id" => agent_id, "prompt" => prompt} = params) do
     user = conn.assigns.current_user
 
+    # `SandboxKey.opts/1`: this route writes labels onto the teammate's
+    # conversation (#1637), which a sandbox's own token does not own.
     with {:ok, images} <- FountainWeb.PromptImages.decode(params["images"]),
-         opts = [source: "api"] ++ Audited.attribution(conn),
+         opts =
+           [source: "api", labels: params["labels"]] ++
+             FountainWeb.SandboxKey.opts(conn) ++ Audited.attribution(conn),
          {:ok, conv} <- Team.send_message(user.id, agent_id, prompt, images, opts) do
       conn
       |> put_status(:accepted)

apps/fountain/lib/fountain_web/label_filter.ex

@@ -0,0 +1,41 @@
+defmodule FountainWeb.LabelFilter do
+  @moduledoc """
+  The `?label=key:value` filter on a conversation list (#1637), for both
+  routes that take it.
+
+  `GET /api/conversations` and `GET /api/team/:agent_id/conversations` accept
+  the same repeatable, AND-combined parameter, and a second copy of the
+  parsing in the team controller is exactly how the two would drift.
+
+  Read from `conn.query_string` and not from `conn.params`, because Plug
+  collapses a repeated key to its last value and this filter is repeatable by
+  design. `Fountain.Conversations.Labels` owns the vocabulary; this is the
+  Plug half plus the error the fallback controller renders as a 400.
+  """
+
+  alias Fountain.Conversations.Labels
+
+  @doc """
+  The label filter on a request: `{:ok, %{"env" => "prod"}}`, or
+  `{:error, "invalid_label_filter"}` for a value with no colon or an empty
+  key, which `FountainWeb.FallbackController` renders as a 400.
+
+  Reads the list `FountainWeb.Plugs.RepeatedQueryParam` left on the request.
+  `List.wrap/1` rather than a match, so a route that has not been given that
+  plug degrades to the single value Plug kept instead of raising.
+  """
+  @spec from(Plug.Conn.t()) :: {:ok, map()} | {:error, String.t()}
+  def from(%Plug.Conn{} = conn) do
+    conn.params
+    |> Map.get("label")
+    |> List.wrap()
+    |> parse()
+  end
+
+  defp parse(values) do
+    case Labels.parse_filter(values) do
+      {:ok, labels} -> {:ok, labels}
+      {:error, :invalid_label_filter} -> {:error, "invalid_label_filter"}
+    end
+  end
+end