feat(conversations): labels with an AND list filter

CHANGELOG.md

@@ -36,6 +36,15 @@ upgrade, is in
   record already matched the document. Apply stays additive and prunes nothing.
   Rebinding a teammate retires the computer its old environment and vault named,
   and is refused while a turn is running on it.
+- 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

@@ -11,7 +11,7 @@ defmodule Fountain.Conversations do
   require Logger
 
   alias Fountain.Audit
-  alias Fountain.Conversations.{Blocks, Conversation, LogEvent, Sandbox, Turn, TurnImage}
+  alias Fountain.Conversations.{Blocks, Conversation, Labels, LogEvent, Sandbox, Turn, TurnImage}
   alias Fountain.Repo
 
   # ── on the _unsafe_ prefix ────────────────────────────────────────────────
@@ -696,6 +696,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.
@@ -729,6 +733,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)
 
@@ -751,16 +758,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
@@ -889,6 +905,134 @@ defmodule Fountain.Conversations do
     end
   end
 
+  @doc """
+  Merge `labels` into `conversation_id`'s. **The door every request-shaped
+  caller uses** (#1637).
+
+  Merge, not replace: a key that is not named is left alone and a key whose
+  value is `nil` is removed, so a run can stamp one outcome without reading
+  the rest first. `Conversations.Labels` owns the limits, and a write that
+  breaks one comes back as a changeset naming the offending key.
+
+  **A sandbox may label its own conversation only.** Pass
+  `sandbox_key_id: key.id` whenever the caller authenticated with a
+  `sprite`-scoped token: the conversation must be the one that token was
+  minted for (`callback_api_key_id`), or the write is refused with
+  `: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 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.
+
+  Tenant-scoped: an id belonging to another account reads as `:not_found`.
+  """
+  @spec set_conversation_labels(binary(), binary(), term(), keyword()) ::
+          {:ok, Conversation.t()} | {:error, term()}
+  def set_conversation_labels(conversation_id, user_id, labels, opts \\ [])
+      when is_binary(conversation_id) and is_binary(user_id) do
+    case get_conversation(conversation_id, user_id) do
+      nil ->
+        {:error, :not_found}
+
+      %Conversation{} = conv ->
+        if sandbox_owns?(conv, Keyword.get(opts, :sandbox_key_id)) do
+          # Ownership: `conv` came from the tenant-scoped fetch above.
+          _unsafe_merge_labels(conv, labels, opts)
+        else
+          {:error, :sprite_may_not_label_another_conversation}
+        end
+    end
+  end
+
+  # No sandbox key on the request is the owner's own credential, which may
+  # label any conversation it can already fetch.
+  defp sandbox_owns?(_conv, nil), do: true
+  defp sandbox_owns?(%Conversation{callback_api_key_id: id}, key_id), do: id == key_id
+
+  @doc """
+  Merge `labels` into a conversation row, with no tenant scoping and no
+  credential rule.
+
+  Unscoped, hence the prefix. The legitimate callers are
+  `set_conversation_labels/4`, which scopes and applies the sandbox rule
+  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
+  case. Audited as `conversation.labels_set` with the keys written and the
+  keys removed, never the values (ADR 0013).
+  """
+  @spec _unsafe_merge_labels(Conversation.t(), term(), keyword()) ::
+          {:ok, Conversation.t()} | {:error, Ecto.Changeset.t()}
+  def _unsafe_merge_labels(conv, labels, opts \\ [])
+
+  def _unsafe_merge_labels(%Conversation{} = conv, labels, opts) when is_map(labels) do
+    current = conv.labels || %{}
+    merged = Labels.merge(current, labels)
+
+    cond do
+      merged == current -> {:ok, conv}
+      true -> write_labels(conv, current, labels, merged, opts)
+    end
+  end
+
+  # Anything that is not a map is a validation failure with the same shape a
+  # broken limit produces, so a caller reads one answer whichever door it
+  # came through.
+  def _unsafe_merge_labels(%Conversation{} = conv, labels, _opts) do
+    {:error, label_refusal(conv, Labels.check(labels))}
+  end
+
+  defp write_labels(conv, current, labels, merged, opts) do
+    case Labels.check_merge(current, labels) do
+      :ok ->
+        {written, removed} = Labels.changed_keys(current, labels)
+
+        conv
+        |> Conversation.changeset(%{labels: merged})
+        |> Repo.update()
+        |> tap(fn
+          {:ok, updated} -> record_labels_set(updated, written, removed, merged, opts)
+          _ -> :ok
+        end)
+
+      refusal ->
+        {:error, label_refusal(conv, refusal)}
+    end
+  end
+
+  defp record_labels_set(conv, written, removed, merged, opts) do
+    Audit.record(%{
+      user_id: conv.user_id,
+      action: "conversation.labels_set",
+      resource_type: "conversation",
+      resource_id: conv.id,
+      actor: Keyword.get(opts, :actor, "self"),
+      request_ip: Keyword.get(opts, :request_ip),
+      metadata: %{
+        "keys" => written,
+        "removed_keys" => removed,
+        "label_count" => map_size(merged)
+      }
+    })
+  end
+
+  # `Labels.check_merge/2` words the refusal from the write the caller made;
+  # this is what turns that sentence into the `errors.labels` a 422 renders,
+  # the same key the changeset validator would have used.
+  defp label_refusal(%Conversation{} = conv, {:error, message}) do
+    conv
+    |> Ecto.Changeset.change()
+    |> Ecto.Changeset.add_error(:labels, message)
+  end
+
   def update_conversation(%Conversation{} = conv, attrs) do
     conv
     |> Conversation.changeset(attrs)
@@ -1780,8 +1924,10 @@ defmodule Fountain.Conversations do
   independent of `inserted_at`'s one-second precision.
 
   Two concurrent first calls for one channel can both create; the next call
-  resumes whichever is newer. Nothing is audited on the resume path — nothing
-  changed.
+  resumes whichever is newer. Nothing is audited on the resume path unless
+  `attrs["labels"]` actually changes something: it is the same conversation,
+  so labels merge into the row it hands back (#1637) and that write records
+  `conversation.labels_set` like any other.
   """
   def start_or_resume_conversation(attrs, opts \\ [])
 
@@ -1801,6 +1947,7 @@ defmodule Fountain.Conversations do
                  do: {:ok, fresh, :created}
           else
             with :ok <- check_sandbox_api_resume(conv, attrs["sandbox_api_access"]),
+                 {:ok, conv} <- resume_labels(conv, attrs["labels"], opts),
                  do: {:ok, conv, :resumed}
           end
 
@@ -1814,6 +1961,20 @@ defmodule Fountain.Conversations do
     with {:ok, conv} <- start_conversation(attrs, opts), do: {:ok, conv, :created}
   end
 
+  # A resume lands on the conversation the binding already has, so labels on
+  # the request are merged into it rather than dropped (#1637). A caller that
+  # sends none changes nothing, and the resume stays the silent path it was.
+  #
+  # Through `set_conversation_labels/4` rather than the writer beneath it:
+  # this runs on `POST /api/conversations`, which a sandbox's own token may
+  # call, and a resume names an *existing* conversation. Writing here
+  # directly would let a sprite minted for one conversation relabel any other
+  # of the tenant's by resuming its channel.
+  defp resume_labels(%Conversation{} = conv, nil, _opts), do: {:ok, conv}
+
+  defp resume_labels(%Conversation{} = conv, labels, opts),
+    do: set_conversation_labels(conv.id, conv.user_id, labels, opts)
+
   # `true` or `"true"` — the ACP adapter sends a JSON boolean, a hand-built
   # request may send a string. Anything else is not a request.
   defp fresh_requested?(%{"fresh" => fresh}), do: fresh in [true, "true"]
@@ -1905,6 +2066,7 @@ defmodule Fountain.Conversations do
     - `source`                — optional; one of "ui", "api", "agent" (default "api")
     - `parent_conversation_id` — optional; UUID of the conversation that spawned this one
     - `title`                 — optional display title (the team page names a teammate with it)
+    - `labels`                — optional `key => value` strings (#1637); see `Conversations.Labels`
   """
   def start_conversation(attrs, opts \\ [])
 
@@ -1974,7 +2136,8 @@ defmodule Fountain.Conversations do
              title: attrs["title"],
              sandbox_api_access: api_access,
              permission_policy: perm_policy,
-             caller_tools: attrs["caller_tools"] || []
+             caller_tools: attrs["caller_tools"] || [],
+             labels: attrs["labels"] || %{}
            }) do
       # Recorded here rather than in either branch below: both of them return
       # {:ok, conv}. The row exists and the sandbox reservation is spent even
@@ -2496,7 +2659,8 @@ defmodule Fountain.Conversations do
              permission_policy: perm_policy,
              # The bridge's tools (#1202) ride on both create paths: this
              # one is what a home sandbox's second conversation takes.
-             caller_tools: attrs["caller_tools"] || []
+             caller_tools: attrs["caller_tools"] || [],
+             labels: attrs["labels"] || %{}
            }) do
       Audit.record(%{
         user_id: user_id,

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

@@ -56,6 +56,12 @@ 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, 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.
     field :turn_count, :integer, virtual: true, default: 0
@@ -120,7 +126,8 @@ defmodule Fountain.Conversations.Conversation do
       :environment_id,
       :channel_id,
       :permission_policy,
-      :caller_tools
+      :caller_tools,
+      :labels
     ])
     |> validate_required([:runtime, :status, :sandbox_id, :user_id])
     |> validate_length(:channel_id, max: 255)
@@ -129,6 +136,7 @@ defmodule Fountain.Conversations.Conversation do
     |> validate_inclusion(:source, @sources)
     |> validate_inclusion(:sandbox_api_access, @sandbox_api_access_modes)
     |> validate_sandbox_api_access_immutable()
+    |> Fountain.Conversations.Labels.changeset()
     |> foreign_key_constraint(:sandbox_id)
     |> foreign_key_constraint(:agent_id)
     |> foreign_key_constraint(:agent_version_id)