`:erpc` fan-out to its peers (short timeout each): cluster membership and

  quorum (`OpenAgents.Cluster`), Raft membership (`OpenAgents.Cluster.Ra`), and per-node

  release version, hot-load revision (`OpenAgents.BuildInfo`), relup marker, and

  uptime. Counts only, never content: connected controller machines and

  active work jobs appear as integers — no names, goals, ids, or addresses.

  uptime. Public SCV activity includes only a pseudonymous label, lifecycle

  state, admitted tool category, and normalized action. Connected controller

  machines and active work jobs remain counts only. No names, goals, internal

  ids, addresses, prompts, repository paths, tool output, or reports appear.

  Honesty rules the shape:

  - A peer that does not answer in time renders `"unreachable"` — the page

      },

      "nodes" => nodes,

      "counts" => counts(),

      "scvs" => safely(fn -> OpenAgents.SCV.Activity.public_projection() end) || [],

      "forge" => forge_section(),

      "generated_at" => DateTime.utc_now() |> DateTime.to_iso8601()

    }

        {DynamicSupervisor, strategy: :one_for_one, name: OpenAgents.TurnSupervisor},

        {Registry, keys: :unique, name: OpenAgents.VoiceSessionRegistry},

        {DynamicSupervisor, strategy: :one_for_one, name: OpenAgents.VoiceSessionSupervisor},

        OpenAgents.SCV.Activity,

        OpenAgents.Leaderboard.Server,

        {Task.Supervisor, name: OpenAgents.ProviderTaskSupervisor},

        {Task.Supervisor, name: OpenAgents.ToolTaskSupervisor},

defmodule OpenAgents.SCV.Activity do

  @moduledoc """

  Publishes a bounded, content-free projection of active SCV runs.

  The projection accepts only normalized SCV lifecycle metadata. It replaces

  the internal run ID with a one-way public label and derives activity text

  from admitted event and tool names. Objectives, repository paths, arguments,

  tool output, report text, credentials, and diagnostic content never enter the

  public state.

  """

  use GenServer

  @telemetry_event [:openagents, :scv, :event]

  @public_topic "scv_activity:public"

  @replication_topic "scv_activity:replication"

  @maximum_entries 32

  @default_expire_after_ms :timer.seconds(30)

  @default_prune_interval_ms :timer.seconds(5)

  @admitted_tools ~w(apply_patch bash edit glob grep list read todowrite write)

  @type public_entry :: %{String.t() => String.t() | float()}

  @spec start_link(keyword()) :: GenServer.on_start()

  def start_link(options) do

    case Keyword.get(options, :name, __MODULE__) do

      nil -> GenServer.start_link(__MODULE__, options)

      name -> GenServer.start_link(__MODULE__, options, name: name)

    end

  end

  @doc "Observes one versioned SCV event."

  @spec observe(map(), GenServer.server()) :: :ok

  def observe(event, server \\ __MODULE__), do: GenServer.cast(server, {:observe, event})

  @doc "Returns the bounded public projection, newest activity first."

  @spec public_projection(GenServer.server()) :: [public_entry()]

  def public_projection(server \\ __MODULE__) do

    GenServer.call(server, :public_projection)

  catch

    :exit, _reason -> []

  end

  @doc "Subscribes the caller to `{:scv_activity, entries}` updates."

  @spec subscribe(module()) :: :ok | {:error, term()}

  def subscribe(pubsub \\ OpenAgents.PubSub),

    do: Phoenix.PubSub.subscribe(pubsub, @public_topic)

  @doc false

  def handle_telemetry(_event_name, _measurements, metadata, activity) do

    send(activity, {:telemetry_event, metadata})

  end

  @impl true

  def init(options) do

    pubsub = Keyword.get(options, :pubsub, OpenAgents.PubSub)

    telemetry? = Keyword.get(options, :telemetry, true)

    expire_after_ms = Keyword.get(options, :expire_after_ms, @default_expire_after_ms)

    prune_interval_ms = Keyword.get(options, :prune_interval_ms, @default_prune_interval_ms)

    if pubsub, do: Phoenix.PubSub.subscribe(pubsub, @replication_topic)

    schedule_prune(prune_interval_ms)

    handler_id = {__MODULE__, self()}

    if telemetry? do

      :ok =

        :telemetry.attach(handler_id, @telemetry_event, &__MODULE__.handle_telemetry/4, self())

    end

    {:ok,

     %{

       entries: %{},

       expire_after_ms: expire_after_ms,

       handler_id: if(telemetry?, do: handler_id),

       prune_interval_ms: prune_interval_ms,

       pubsub: pubsub

     }}

  end

  @impl true

  def handle_call(:public_projection, _from, state) do

    {:reply, project(state.entries), state}

  end

  @impl true

  def handle_cast({:observe, event}, state) do

    case public_command(event) do

      :ignore ->

        {:noreply, state}

      command ->

        state = apply_and_publish(state, command)

        replicate(state.pubsub, command)

        {:noreply, state}

    end

  end

  @impl true

  def handle_info({:telemetry_event, event}, state) do

    case public_command(event) do

      :ignore ->

        {:noreply, state}

      command ->

        state = apply_and_publish(state, command)

        replicate(state.pubsub, command)

        {:noreply, state}

    end

  end

  def handle_info({:scv_activity_replication, origin, _command}, state)

      when origin == self(),

      do: {:noreply, state}

  def handle_info({:scv_activity_replication, _origin, command}, state) do

    {:noreply, apply_and_publish(state, command)}

  end

  def handle_info(:prune, state) do

    entries = prune_expired(state.entries)

    if entries != state.entries do

      broadcast_public(state.pubsub, project(entries))

    end

    schedule_prune(state.prune_interval_ms)

    {:noreply, %{state | entries: entries}}

  end

  def handle_info(_message, state), do: {:noreply, state}

  @impl true

  def terminate(_reason, %{handler_id: nil}), do: :ok

  def terminate(_reason, %{handler_id: handler_id}) do

    :telemetry.detach(handler_id)

    :ok

  end

  defp public_command(event) when is_map(event) do

    with "openagents.scv.event.v1" <- value(event, :schema),

         run_id when is_binary(run_id) <- value(event, :run_id),

         {:ok, _uuid} <- Ecto.UUID.cast(run_id),

         type when is_binary(type) <- value(event, :type) do

      id = public_id(run_id)

      case type do

        terminal when terminal in ["process_finished", "run_finished"] ->

          {:delete, id}

        "heartbeat" ->

          {:touch, id, base_entry(id, "Working within its resource budget")}

        "run_preparing" ->

          {:upsert, id, base_entry(id, "Preparing an admitted SCV run")}

        "process_starting" ->

          {:upsert, id, base_entry(id, "Starting its coding runtime")}

        "process_started" ->

          {:upsert, id, base_entry(id, "Coding runtime started")}

        "opencode_event" ->

          {:upsert, id, open_code_entry(id, event)}

        _other ->

          :ignore

      end

    else

      _invalid -> :ignore

    end

  end

  defp public_command(_event), do: :ignore

  defp open_code_entry(id, event) do

    event_type = value(event, :event_type)

    tool = admitted_tool(value(event, :tool))

    {text, tool} =

      case {event_type, tool} do

        {"tool_use", "read"} -> {"Reading repository context", "read"}

        {"tool_use", "grep"} -> {"Searching repository context", "grep"}

        {"tool_use", "glob"} -> {"Mapping repository files", "glob"}

        {"tool_use", "list"} -> {"Listing repository context", "list"}

        {"tool_use", "edit"} -> {"Applying a bounded code edit", "edit"}

        {"tool_use", "apply_patch"} -> {"Applying a bounded code patch", "apply_patch"}

        {"tool_use", "write"} -> {"Writing an admitted workspace file", "write"}

        {"tool_use", "bash"} -> {"Running an admitted command", "bash"}

        {"tool_use", "todowrite"} -> {"Updating its work plan", "todowrite"}

        {"step_start", _tool} -> {"Starting its next model step", nil}

        {"step_finish", _tool} -> {"Finished a model step", nil}

        {"text", _tool} -> {"Preparing its bounded report", nil}

        {_event_type, _tool} -> {"Working on its admitted objective", nil}

      end

    id

    |> base_entry(text)

    |> maybe_put_tool(tool)

  end

  defp admitted_tool(tool) when tool in @admitted_tools, do: tool

  defp admitted_tool(_tool), do: nil

  defp base_entry(id, text) do

    %{

      "id" => id,

      "label" => id |> String.replace_prefix("scv-", "SCV ") |> String.upcase(),

      "status" => "running",

      "weight" => 0.4,

      "text" => text

    }

  end

  defp maybe_put_tool(entry, nil), do: Map.delete(entry, "tool")

  defp maybe_put_tool(entry, tool), do: Map.put(entry, "tool", tool)

  defp apply_and_publish(state, command) do

    entries = apply_command(state.entries, command, state.expire_after_ms)

    entries = retain_latest(entries)

    broadcast_public(state.pubsub, project(entries))

    %{state | entries: entries}

  end

  defp apply_command(entries, {:delete, id}, _expire_after_ms), do: Map.delete(entries, id)

  defp apply_command(entries, {:upsert, id, public}, expire_after_ms) do

    Map.put(entries, id, timed_entry(public, expire_after_ms))

  end

  defp apply_command(entries, {:touch, id, public}, expire_after_ms) do

    current = Map.get(entries, id, %{public: public})

    Map.put(entries, id, timed_entry(current.public, expire_after_ms))

  end

  defp timed_entry(public, expire_after_ms) do

    %{

      expires_at: monotonic_ms() + expire_after_ms,

      order: next_order(),

      public: public

    }

  end

  defp prune_expired(entries) do

    now = monotonic_ms()

    Map.reject(entries, fn {_id, entry} -> entry.expires_at <= now end)

  end

  defp retain_latest(entries) when map_size(entries) <= @maximum_entries, do: entries

  defp retain_latest(entries) do

    entries

    |> Enum.sort_by(fn {_id, entry} -> entry.order end, :desc)

    |> Enum.take(@maximum_entries)

    |> Map.new()

  end

  defp project(entries) do

    entries

    |> Map.values()

    |> Enum.sort_by(& &1.order, :desc)

    |> Enum.map(& &1.public)

  end

  defp replicate(nil, _command), do: :ok

  defp replicate(pubsub, command) do

    Phoenix.PubSub.broadcast(pubsub, @replication_topic, {

      :scv_activity_replication,

      self(),

      command

    })

  end

  defp broadcast_public(nil, _projection), do: :ok

  defp broadcast_public(pubsub, projection) do

    Phoenix.PubSub.broadcast(pubsub, @public_topic, {:scv_activity, projection})

  end

  defp public_id(run_id) do

    digest = :crypto.hash(:sha256, run_id) |> Base.encode16(case: :lower) |> String.slice(0, 8)

    "scv-" <> digest

  end

  defp value(event, key), do: Map.get(event, key) || Map.get(event, Atom.to_string(key))

  defp monotonic_ms, do: System.monotonic_time(:millisecond)

  defp next_order, do: System.unique_integer([:monotonic, :positive])

  defp schedule_prune(nil), do: :ok

  defp schedule_prune(interval_ms), do: Process.send_after(self(), :prune, interval_ms)

end

  use OpenAgentsWeb, :live_view

  alias OpenAgents.NetworkStatus

  alias OpenAgents.SCV.Activity

  alias OpenAgentsWeb.UI.Graph

  @tick_ms 5_000

  @events_kept 20

  def mount(_params, _session, socket) do

    if connected?(socket) do

      :ok = NetworkStatus.subscribe()

      :ok = Activity.subscribe()

      Enum.each(["forge:pushes", "forge:target", "forge:builds", "forge:deploys"], fn topic ->

        Phoenix.PubSub.subscribe(OpenAgents.PubSub, topic)

    {:noreply, socket |> track_transitions(projection) |> assign_projection(projection)}

  end

  def handle_info({:scv_activity, entries}, socket) do

    {:noreply, assign(socket, :scvs, public_scvs(entries))}

  end

  def handle_info(:tick, socket) do

    Process.send_after(self(), :tick, @tick_ms)

    projection = NetworkStatus.projection(refresh: true)

    socket

    |> assign(:projection, projection)

    |> assign(:scvs, public_scvs(Activity.public_projection()))

    |> assign(:overall, overall(projection))

  end

  defp public_scvs(entries) when is_list(entries) do

    Enum.map(entries, fn entry ->

      %{

        id: entry["id"],

        label: entry["label"],

        status: public_scv_status(entry["status"]),

        weight: entry["weight"],

        tool: entry["tool"],

        text: entry["text"]

      }

    end)

  end

  defp public_scvs(_entries), do: []

  defp public_scv_status("running"), do: :running

  defp public_scv_status(_status), do: :idle

  # Version divergence across reachable nodes IS the rollout visualization:

  # when a roll or hot deploy is sweeping the fleet, nodes disagree — surface

  # that as an explicit "rolling" state with progress.

            newest build. The cluster never drops below quorum during a roll.

          </.alert>

          <.card id="status-scvs">

            <h2>Active SCVs</h2>

            <p class="status-forge__intro">

              Follow the bounded public activity from SCVs that are running now.

              Prompts, repository paths, command arguments, tool output, and reports

              stay private.

            </p>

            <.empty :if={@scvs == []} id="status-no-scvs" title="No active SCVs">

              The next admitted SCV run will appear here as it works.

            </.empty>

            <Graph.scv_streams

              :if={@scvs != []}

              id="public-scv-streams"

              scvs={@scvs}

            />

          </.card>

          <div id="status-nodes" class="status-nodes">

            <.card

              :for={{node, index} <- Enum.with_index(@projection["nodes"])}

    assert %{"machines_connected" => machines, "active_jobs" => jobs} = projection["counts"]

    assert is_nil(machines) or is_integer(machines)

    assert is_nil(jobs) or is_integer(jobs)

    assert is_list(projection["scvs"])

  end

  test "an unreachable peer degrades to an honest per-node report, not a crash" do

defmodule OpenAgents.SCV.ActivityTest do

  use ExUnit.Case, async: false

  alias OpenAgents.SCV.Activity

  test "projects telemetry events into bounded public activity" do

    activity = start_supervised!({Activity, name: nil, pubsub: nil, telemetry: false})

    run_id = Ecto.UUID.generate()

    Activity.observe(

      %{

        schema: "openagents.scv.event.v1",

        run_id: run_id,

        type: "run_preparing",

        objective: "private objective",

        repository: "/private/repository",

        model: "openai/gpt-5.6-luna"

      },

      activity

    )

    assert [entry] = Activity.public_projection(activity)

    assert entry["status"] == "running"

    assert entry["text"] == "Preparing an admitted SCV run"

    assert entry["id"] =~ ~r/^scv-[0-9a-f]{8}$/

    assert entry["label"] =~ ~r/^SCV [0-9A-F]{8}$/

    rendered = inspect(entry)

    refute rendered =~ run_id

    refute rendered =~ "private objective"

    refute rendered =~ "/private/repository"

    refute rendered =~ "gpt-5.6-luna"

    Activity.observe(

      %{

        schema: "openagents.scv.event.v1",

        run_id: run_id,

        type: "opencode_event",

        event_type: "tool_use",

        tool: "grep",

        tool_status: "completed",

        output: "private tool output"

      },

      activity

    )

    assert [entry] = Activity.public_projection(activity)

    assert entry["tool"] == "grep"

    assert entry["text"] == "Searching repository context"

    refute inspect(entry) =~ "private tool output"

    Activity.observe(

      %{schema: "openagents.scv.event.v1", run_id: run_id, type: "run_finished"},

      activity

    )

    assert Activity.public_projection(activity) == []

  end

  test "observes the executor telemetry event" do

    activity = start_supervised!({Activity, name: nil, pubsub: nil})

    run_id = Ecto.UUID.generate()

    :telemetry.execute(

      [:openagents, :scv, :event],

      %{count: 1},

      %{

        schema: "openagents.scv.event.v1",

        run_id: run_id,

        type: "heartbeat"

      }

    )

    assert [%{"text" => "Working within its resource budget"}] =

             Activity.public_projection(activity)

    :telemetry.execute(

      [:openagents, :scv, :event],

      %{count: 1},

      %{schema: "openagents.scv.event.v1", run_id: run_id, type: "run_finished"}

    )

    assert Activity.public_projection(activity) == []

    assert Activity.public_projection() == []

  end

  test "ignores malformed and unrelated events" do

    activity = start_supervised!({Activity, name: nil, pubsub: nil, telemetry: false})

    Activity.observe(

      %{schema: "other", run_id: Ecto.UUID.generate(), type: "heartbeat"},

      activity

    )

    Activity.observe(%{schema: "openagents.scv.event.v1", type: "heartbeat"}, activity)

    assert Activity.public_projection(activity) == []

  end

  test "returns an empty projection while the activity process is unavailable" do

    assert Activity.public_projection(OpenAgents.SCV.UnavailableActivity) == []

  end

  test "expires an SCV that stops emitting heartbeats" do

    activity =

      start_supervised!(

        {Activity,

         name: nil, pubsub: nil, telemetry: false, expire_after_ms: 0, prune_interval_ms: nil}

      )

    Activity.observe(

      %{

        schema: "openagents.scv.event.v1",

        run_id: Ecto.UUID.generate(),

        type: "heartbeat"

      },

      activity

    )

    assert [_entry] = Activity.public_projection(activity)

    send(activity, :prune)

    assert Activity.public_projection(activity) == []

  end

end

    assert html =~ "BEAM nodes"

    assert html =~ "node 1"

    assert has_element?(view, ".status-metric__label", "computers connected")

    assert has_element?(view, "#status-scvs")

    assert has_element?(view, "#status-no-scvs")

    # Content-free: the serving node's internal name never reaches the page.

    refute html =~ to_string(node())

  end

  test "renders bounded live SCV activity without private event content", %{conn: conn} do

    run_id = Ecto.UUID.generate()

    OpenAgents.SCV.Activity.observe(%{

      schema: "openagents.scv.event.v1",

      run_id: run_id,

      type: "opencode_event",

      event_type: "tool_use",

      tool: "grep",

      objective: "private parity objective",

      repository: "/workspace/private-repository",

      output: "private tool output"

    })

    _projection = OpenAgents.SCV.Activity.public_projection()

    on_exit(fn ->

      OpenAgents.SCV.Activity.observe(%{

        schema: "openagents.scv.event.v1",

        run_id: run_id,

        type: "run_finished"

      })

      _projection = OpenAgents.SCV.Activity.public_projection()

    end)

    conn = put_req_header(conn, "accept", "text/html")

    {:ok, view, html} = live(conn, ~p"/status")

    assert has_element?(view, "#public-scv-streams")

    assert html =~ "Searching repository context"

    refute html =~ run_id

    refute html =~ "private parity objective"

    refute html =~ "/workspace/private-repository"

    refute html =~ "private tool output"

  end

  test "legacy JSON pollers of /status keep the old health payload", %{conn: conn} do

    # No Accept header (probe-style) → legacy JSON.

    response = conn |> get(~p"/status") |> json_response(200)