feat(s090): the Activity page, its bounded buffer, and a LiveDashboard sessions page

Trinity.Telemetry.Activity keeps the last 200 events and drops the oldest.
Bounded because an unbounded event buffer is a memory leak with a nice name, and
in memory only because what must survive a restart is a receipt or a usage row,
both written to disk by the code that caused them rather than by a listener.

Its handler does one send and nothing else. A telemetry handler runs in the
process that emitted the event, so work done there is charged to whatever turn
happened to emit it. The handler is attached in init and detached in terminate so
its lifetime is the buffer's: a handler that outlives its target sends to a dead
pid on every event for the rest of the run.

The Activity page shows those events beside the cost totals, because what
happened and what it cost are the same question asked twice and splitting them
across two pages makes a person do the joining. It reads and never writes.

The test that matters there is a negative one. Events carry no prompt text, no
completion text and no tool arguments, so a test asserting rows appear would pass
equally well on a page that had quietly started showing conversations. A secret
is passed as a tool argument and the page is asserted not to contain it.

The LiveDashboard page answers what LiveDashboard cannot: which processes are
Trinity sessions and what state each machine is in, which is the question someone
asks when a turn looks stuck. It asks the process rather than reading the row,
because a row reads active long after its process has gone: slice 080 found that
in a panel that promised which are running and delivered which are not archived.

6 more tests. mix gate exit 0; 671 passed; coverage 80.31%.

Signed-off-by: Ayla Croft <aylacroft@proton.me>

Author: HackTuah

lib/trinity.ex

@@ -30,6 +30,8 @@ defmodule Trinity do
         # redaction filter is exported because the application installs it as a primary Logger
         # filter at boot.
         Telemetry,
+        Telemetry.Activity,
+        Telemetry.Costs,
         Telemetry.Redaction,
         Repo,
         Repo.Receipts,

lib/trinity/application.ex

@@ -33,6 +33,9 @@ defmodule Trinity.Application do
       desktop_children() ++
         [
           TrinityWeb.Telemetry,
+          # Slice 090: the Activity page's bounded buffer. Early, so it is listening before
+          # anything it would want to have recorded has happened.
+          Trinity.Telemetry.Activity,
           # Slice 010: one node per data directory. Before the Repo, so a refused boot has
           # opened no database file; the reason names the holder's OS pid and mode.
           {Trinity.DataDir.Lock, dir: lock_dir(), mode: mode()},

lib/trinity/telemetry/activity.ex

@@ -0,0 +1,112 @@
+# SPDX-FileCopyrightText: Sudo Apt Holdings LLC
+# SPDX-License-Identifier: Apache-2.0
+defmodule Trinity.Telemetry.Activity do
+  @moduledoc """
+  A bounded, in-memory record of recent events, for the Activity page (slice 090).
+
+  **Bounded, because an unbounded event buffer is a memory leak with a nice name.** It keeps the
+  last `limit/0` events and drops the oldest; nothing here is durable and nothing here is meant to
+  be. What must survive a restart is a receipt (slice 024) or a usage row (slice 011), and both are
+  written to disk by the code that caused them rather than by a listener.
+
+  **It holds what the catalogue allows and nothing else.** The events themselves carry no prompt
+  text, no completion text and no tool arguments (`docs/telemetry.md`), so neither does this. That
+  is worth stating because an activity feed is exactly the feature that tempts someone to enrich a
+  record with "just the first line of the message", and the place to refuse that is here, where the
+  refusal is visible, rather than in review.
+
+  It is a `GenServer` rather than an ETS table so that the handler's work is bounded by a mailbox
+  it cannot outrun: a telemetry handler runs **in the process that emitted the event**, and doing
+  real work there would put this module's cost on the turn's latency.
+  """
+
+  use GenServer
+
+  alias Trinity.Telemetry
+
+  @handler "trinity-activity"
+  @limit 200
+
+  @type entry :: %{
+          at: DateTime.t(),
+          event: [atom()],
+          measurements: map(),
+          metadata: map()
+        }
+
+  @doc "How many events are kept."
+  @spec limit() :: pos_integer()
+  def limit, do: @limit
+
+  def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+
+  @doc """
+  The recent events, newest first.
+
+  Options: `:session_id` and `:event` (a prefix, so `[:trinity, :tool]` matches both the start and
+  the stop).
+  """
+  @spec recent(keyword()) :: [entry()]
+  def recent(opts \\ []) do
+    GenServer.call(__MODULE__, :recent)
+    |> filter_session(opts[:session_id])
+    |> filter_event(opts[:event])
+  catch
+    :exit, _ -> []
+  end
+
+  @doc "Empties the buffer. For tests, and for a person who wants the page to start again."
+  @spec clear() :: :ok
+  def clear do
+    GenServer.call(__MODULE__, :clear)
+  catch
+    :exit, _ -> :ok
+  end
+
+  @impl true
+  def init(_opts) do
+    # Attached here rather than in the application so the handler's lifetime is this process's: a
+    # handler that outlives its target sends to a dead pid on every event for the rest of the run.
+    :telemetry.attach_many(@handler, Telemetry.catalogue(), &__MODULE__.handle/4, nil)
+    {:ok, %{entries: []}}
+  end
+
+  @impl true
+  def terminate(_reason, _state) do
+    :telemetry.detach(@handler)
+    :ok
+  end
+
+  @doc false
+  # Runs in the emitting process. It does one `send` and nothing else, on purpose: a handler that
+  # did real work here would charge it to whatever turn happened to emit the event.
+  def handle(event, measurements, metadata, _config) do
+    send(__MODULE__, {:event, event, measurements, metadata, DateTime.utc_now()})
+    :ok
+  rescue
+    # A send to a name that is not registered raises; an unstarted buffer must not break a turn.
+    _ -> :ok
+  end
+
+  @impl true
+  def handle_info({:event, event, measurements, metadata, at}, state) do
+    entry = %{at: at, event: event, measurements: measurements, metadata: metadata}
+    {:noreply, %{state | entries: Enum.take([entry | state.entries], @limit)}}
+  end
+
+  def handle_info(_other, state), do: {:noreply, state}
+
+  @impl true
+  def handle_call(:recent, _from, state), do: {:reply, state.entries, state}
+  def handle_call(:clear, _from, state), do: {:reply, :ok, %{state | entries: []}}
+
+  defp filter_session(entries, nil), do: entries
+
+  defp filter_session(entries, session_id),
+    do: Enum.filter(entries, &(&1.metadata[:session_id] == session_id))
+
+  defp filter_event(entries, nil), do: entries
+
+  defp filter_event(entries, prefix) when is_list(prefix),
+    do: Enum.filter(entries, &List.starts_with?(&1.event, prefix))
+end

lib/trinity_web/dashboard/sessions_page.ex

@@ -0,0 +1,92 @@
+# SPDX-FileCopyrightText: Sudo Apt Holdings LLC
+# SPDX-License-Identifier: Apache-2.0
+defmodule TrinityWeb.Dashboard.SessionsPage do
+  @moduledoc """
+  A LiveDashboard page listing the session processes that are alive, with their `gen_statem` state
+  (slice 090, AC3).
+
+  LiveDashboard already lists every process in the VM. What it cannot say is which of them is a
+  Trinity session and what state that session's machine is in, which is the question someone asks
+  when a turn appears stuck: waiting on a model, waiting on an approval, or idle and the page did
+  not notice.
+
+  It reads the registry and asks each process, so it shows the running system rather than the
+  database's opinion of it. A session row reads `active` long after its process has gone; slice 080
+  found that the hard way, in a panel that promised "which are running" and delivered "which are
+  not archived".
+  """
+
+  use Phoenix.LiveDashboard.PageBuilder
+
+  alias Trinity.Sessions
+
+  @impl true
+  def menu_link(_, _), do: {:ok, "Trinity sessions"}
+
+  @impl true
+  def render(assigns) do
+    ~H"""
+    <.live_table
+      id="trinity-sessions"
+      dom_id="trinity-sessions"
+      page={@page}
+      title="Trinity sessions"
+      row_fetcher={&fetch_rows/2}
+      rows_name="sessions"
+    >
+      <:col field={:id} header="Session" />
+      <:col field={:state} header="State" />
+      <:col field={:title} header="Title" />
+      <:col field={:origin} header="Origin" />
+      <:col field={:pid} header="Process" />
+      <:col field={:memory} header="Memory (KB)" sortable={:desc} />
+    </.live_table>
+    """
+  end
+
+  @doc false
+  def fetch_rows(params, _node) do
+    rows = rows()
+    {sort_rows(rows, params), length(rows)}
+  end
+
+  defp rows do
+    for session <- Sessions.list_sessions(limit: 200),
+        pid = Sessions.whereis(session.id),
+        is_pid(pid) do
+      %{
+        id: short(session.id),
+        state: state_of(session.id),
+        title: session.title || "(untitled)",
+        origin: session.origin,
+        pid: inspect(pid),
+        memory: memory_kb(pid)
+      }
+    end
+  end
+
+  # Asked of the process, never read from the row: a row's status is the session's lifecycle and
+  # says nothing about whether its machine is doing anything.
+  defp state_of(session_id) do
+    case Sessions.state(session_id) do
+      %{state: state} -> to_string(state)
+      _ -> "-"
+    end
+  catch
+    :exit, _ -> "-"
+  end
+
+  defp memory_kb(pid) do
+    case Process.info(pid, :memory) do
+      {:memory, bytes} -> div(bytes, 1024)
+      _ -> 0
+    end
+  end
+
+  defp short(id), do: String.slice(id, -8, 8)
+
+  defp sort_rows(rows, %{sort_by: by, sort_dir: dir}) when is_atom(by),
+    do: Enum.sort_by(rows, &Map.get(&1, by), dir)
+
+  defp sort_rows(rows, _params), do: rows
+end