docs: labels in the manual and the SDK guide, and one SDK release (#1637)

CHANGELOG.md

@@ -98,6 +98,15 @@ upgrade, is in
   on the team, which opens its conversation and provisions its computer.
   Re-applying moves the name and the bindings and provisions no second
   computer (#1636).
+- Conversations carry free-form `labels`, a map of at most 32 key/value
+  strings. Set them on creation or with `PATCH /api/conversations/:id/labels`,
+  which merges. A running agent stamps its own conversation with the
+  `_fountain/labels` ACP extension notification, and a sandbox callback token
+  can label only the conversation it was minted for, on every door that writes
+  labels. `GET /api/conversations`
+  and `GET /api/team/:agent_id/conversations` take a repeatable `label=key:value`
+  filter, combined with AND. `conversation.*` webhook payloads carry `labels`,
+  and the console's conversation lists render them as chips.
 
 ### Fixed
 

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
@@ -975,9 +991,9 @@ defmodule Fountain.Conversations do
   `:sprite_may_not_label_another_conversation`. Without that check a worker
   holding an account-scoped callback key could relabel every other run on the
   account, which is exactly the loop ADR 0045 describes. That is why the
-  check lives here and not in a controller: more than one request path will
-  write labels, and the rule has to hold on the door rather than on whichever
-  of them remembered.
+  check lives here and not in each controller: `PATCH .../labels`, the team
+  message and a `channel_id` resume all write labels, and the rule has to
+  hold on the door rather than on whichever of them remembered.
 
   `labels` that is not a map at all is a validation failure, not a silent
   no-op, so every door refuses `{"labels": "env=prod"}` the same way.
@@ -1011,10 +1027,12 @@ defmodule Fountain.Conversations do
   Merge `labels` into a conversation row, with no tenant scoping and no
   credential rule.
 
-  Unscoped, hence the prefix. The legitimate caller is
+  Unscoped, hence the prefix. The legitimate callers are
   `set_conversation_labels/4`, which scopes and applies the sandbox rule
-  before delegating here. A request path that calls this directly has skipped
-  the rule that stops one sandbox relabelling another, so do not add one.
+  before delegating here, and `Labels._unsafe_stamp/2`, which runs inside the
+  conversation's own server and holds the row that server was started for. A
+  request path that calls this directly has skipped the rule that stops one
+  sandbox relabelling another, so do not add one.
 
   A merge that changes nothing writes nothing and records nothing — a
   deterministic run re-stamping the same outcome on every tick is the normal

apps/fountain/lib/fountain/conversations/conversation.ex

@@ -56,10 +56,11 @@ defmodule Fountain.Conversations.Conversation do
     field :permission_policy, :map
     # The caller-defined tools of the bridge (#1202, `Fountain.CallerTools`).
     field :caller_tools, {:array, :map}, default: []
-    # Free-form `key => value` strings (#1637), set at launch.
-    # `Conversations.Labels` owns the limits and the merge; writes here go
-    # through `Labels.changeset/1` below, which is why every door enforces
-    # the same rule.
+    # Free-form `key => value` strings (#1637). Set at launch, merged by the
+    # labels route and by the agent's own `_fountain/labels` ACP notification,
+    # and filtered on with jsonb containment. `Conversations.Labels` owns the
+    # limits and the merge; writes here go through `Labels.changeset/1` below,
+    # which is why every door enforces the same rule.
     field :labels, :map, default: %{}
 
     # Populated by list_conversations_by_activity/1 — not persisted.

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

@@ -8,9 +8,9 @@ defmodule Fountain.Conversations.Labels do
   `env=prod`, `drift=true`, `gated=apply`. Those facts are not text worth
   searching, so they are not in the full-text index and never will be.
 
-  Every door that writes labels goes through `changeset/1` here, called from
-  `Fountain.Conversations.Conversation.changeset/2`, so the limits cannot
-  drift between one writer and the next:
+  Every door that writes labels goes through `changeset/1` here, so the
+  limits cannot drift between the create request, the labels route, the team
+  message and the ACP extension notification:
 
     * at most 32 entries;
     * a key is a non-empty string of at most 64 bytes;
@@ -27,10 +27,19 @@ 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]
 
+  require Logger
+
   @max_entries 32
   @max_key_bytes 64
   @max_value_bytes 256
@@ -254,4 +263,94 @@ defmodule Fountain.Conversations.Labels do
      |> Enum.filter(&Map.has_key?(current, &1))
      |> Enum.sort()}
   end
+
+  @doc """
+  Merge the labels an agent stamped on its own run over the ACP extension
+  notification (#1637).
+
+  Unscoped, hence the prefix: it takes a bare conversation id and writes to
+  that row without a `user_id` and without a credential check. The only
+  caller is `Fountain.Conversations.TurnMachine`, running inside the
+  conversation's own `ConversationServer`, holding the id that server was
+  started with — so it cannot name another tenant's conversation, or another
+  conversation of the same tenant. A request-shaped caller wants
+  `Fountain.Conversations.set_conversation_labels/4`, which scopes by
+  `user_id` and applies the sandbox rule.
+
+  Recorded as `sprite` — code in the sandbox acting on the tenant's behalf
+  (ADR 0013).
+
+  **Nothing a label contains can take the turn down.** A stamp the limits
+  refuse is logged and dropped, and so is one that raises on its way to the
+  database. The run is mid-turn and doing real work, and losing it because a
+  value was 300 bytes long, or held a byte `jsonb` will not store, would be
+  the worse outcome by a distance.
+  """
+  @spec _unsafe_stamp(String.t() | nil, map()) :: :ok
+  def _unsafe_stamp(conversation_id, labels)
+
+  def _unsafe_stamp(conversation_id, labels)
+      when is_binary(conversation_id) and is_map(labels) do
+    # ownership: `conversation_id` is the id the calling `ConversationServer`
+    # was started with, held by its own turn machine. It cannot name another
+    # conversation, so both unscoped calls below are on this server's own row.
+    with %{} = conv <- Fountain.Conversations._unsafe_get_conversation(conversation_id),
+         {:error, changeset} <-
+           Fountain.Conversations._unsafe_merge_labels(conv, labels, actor: "sprite") do
+      Logger.warning(
+        "conv #{conversation_id}: refused _fountain/labels update: #{inspect(changeset.errors)}"
+      )
+    end
+
+    :ok
+  rescue
+    error ->
+      # The belt to `check_entry/2`'s braces. Every shape we know of is
+      # refused above with a message; this is here so that a shape we do not
+      # know of costs the stamp and not the turn.
+      Logger.warning(
+        "conv #{conversation_id}: _fountain/labels update raised: #{Exception.message(error)}"
+      )
+
+      :ok
+  end
+
+  def _unsafe_stamp(_conversation_id, _labels), do: :ok
+
+  @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/conversations/turn_machine.ex

@@ -44,7 +44,7 @@ defmodule Fountain.Conversations.TurnMachine do
   require OpenTelemetry.Tracer
 
   alias Fountain.{Agents, Conversations}
-  alias Fountain.Conversations.Conversation
+  alias Fountain.Conversations.{Conversation, Labels}
 
   @typedoc "What the peer reports about a turn, with the command ref already matched."
   @type payload :: tuple()
@@ -185,13 +185,34 @@ defmodule Fountain.Conversations.TurnMachine do
   # the full quiet window: `running` status, deferred idle park, billed turn
   # time. Metadata is nothing the agent did, so it opens no turn and re-arms
   # no quiet timer; with a turn in flight it still lands on the transcript.
+  #
+  # `_fountain/labels` is the third exception (#1637). ACP reserves a leading
+  # underscore for extensions, and `session/update` is the only notification
+  # the peer forwards, so that is where a deterministic run stamps its own
+  # outcome. It is a control message and not something the agent said, so it
+  # opens no turn, re-arms no quiet timer and never reaches the transcript.
   def handle(%__MODULE__{} = turn, {:lines, stream, data}, ctx) do
+    labels = if stream == "acp", do: label_update(data)
+
     cond do
       stream == "acp" and MapSet.member?(turn.replay_dedup, data) ->
         # A replayed line we already hold (ACP reattach). Each persisted line
         # suppresses at most one arrival, so a legitimate later repeat survives.
         {%{turn | replay_dedup: MapSet.delete(turn.replay_dedup, data)}, []}
 
+      is_map(labels) ->
+        # Written here rather than handed back as an effect: nothing about it
+        # is the server's — no state, no process, no timer — and this machine
+        # already writes what a report means.
+        #
+        # Ownership: `turn.conversation_id` is the id this machine's own
+        # `ConversationServer` was started with, so the unscoped write below
+        # cannot name another conversation, of this tenant or any other.
+        # Nothing a stamp contains can fail the turn; `_unsafe_stamp/2` logs
+        # and drops instead.
+        Labels._unsafe_stamp(turn.conversation_id, labels)
+        {turn, []}
+
       stream == "acp" and Managoat.ACP.Protocol.session_metadata?(data) ->
         if is_nil(turn.row) do
           # No turn to attach it to, and not worth opening one: dropped.
@@ -507,6 +528,41 @@ defmodule Fountain.Conversations.TurnMachine do
     {turn, finish ++ [{:drop_connection, "failed"}]}
   end
 
+  # The extension update a run stamps its own outcome with (#1637):
+  #
+  #     {"jsonrpc":"2.0","method":"session/update","params":{
+  #       "sessionId":"...",
+  #       "update":{"sessionUpdate":"_fountain/labels",
+  #                 "labels":{"drift":"true","env":"prod"}}}}
+  #
+  # ACP reserves a leading `_` for extensions, and `Managoat.ACP.Peer`
+  # forwards `session/update` and drops every other notification method, so
+  # this rides there rather than on a method of its own.
+  #
+  # A cheap substring test before the decode: this runs on every protocol line
+  # of every turn, and all but a handful of them are agent output.
+  @label_kind "_fountain/labels"
+
+  defp label_update(data) do
+    if String.contains?(data, @label_kind), do: decode_label_update(data)
+  end
+
+  defp decode_label_update(data) do
+    case Managoat.ACP.Protocol.classify_line(data) do
+      {:notification, "session/update", %{"update" => %{"sessionUpdate" => @label_kind} = update}} ->
+        # An `update` carrying no `labels` object is read as an empty merge,
+        # which writes nothing. Raising on it would cost the run its turn for
+        # a stamp, which is the wrong trade.
+        case Map.get(update, "labels") do
+          labels when is_map(labels) -> labels
+          _ -> %{}
+        end
+
+      _ ->
+        nil
+    end
+  end
+
   # ── ending a turn ─────────────────────────────────────────────────────────
 
   @doc """

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
       [] ->
         []