Control-Center/backend/ControlCenter/Hubs/AgentStatusHub.cs

using Microsoft.AspNetCore.SignalR;

namespace ControlCenter.Hubs;

/// <summary>
/// SignalR hub for real-time agent status updates in the Command Hub.
///
/// <para>Usage flow:</para>
/// <list type="number">
///   <item>Client connects to <c>/hubs/agent-status</c></item>
///   <item>Client calls <see cref="JoinFleet"/> to subscribe to all agent updates</item>
///   <item>Client calls <see cref="JoinAgentGroup"/> to subscribe to a specific agent</item>
///   <item>Server pushes <see cref="IAgentStatusClient.AgentStatusChanged"/>
///         and <see cref="IAgentStatusClient.AgentTaskProgress"/> events</item>
///   <item>Client calls <see cref="GetFleetSnapshot"/> for initial state on connect</item>
/// </list>
///
/// <para>Group naming:</para>
/// <list type="bullet">
///   <item>Fleet group: <c>fleet</c> — receives all agent updates</item>
///   <item>Agent group: <c>agent:{agentId}</c> — receives updates for one agent</item>
/// </list>
///
/// <para>Typed client: <see cref="IAgentStatusClient"/> — all server-to-client
/// calls go through this interface for compile-time safety.</para>
///
/// <para>Architecture note: This hub bridges OpenClaw Gateway WebSocket events
/// to SignalR clients. A background service (<see cref="Services.GatewayEventBridgeService"/>)
/// subscribes to Gateway events and pushes them through this hub's extension methods.</para>
/// </summary>
public class AgentStatusHub : Hub<IAgentStatusClient>
{
    private readonly ILogger<AgentStatusHub> _logger;

    public AgentStatusHub(ILogger<AgentStatusHub> logger)
    {
        _logger = logger;
    }

    /// <summary>
    /// Broadcasts an agent status update to all connected clients.
    ///
    /// <para>
    /// Any connected client (or server-side caller) can invoke this method
    /// to push a status update to every subscriber. The DTO is converted to
    /// an <see cref="AgentStatusUpdate"/> record and relayed through the
    /// <see cref="IAgentStatusClient.AgentStatusChanged"/> callback.
    /// </para>
    /// </summary>
    /// <param name="update">The agent status update DTO to broadcast.</param>
    public async Task SendStatusUpdate(AgentStatusUpdateDto update)
    {
        _logger.LogInformation(
            "Broadcasting status update for agent {AgentId}: {Status}",
            update.AgentId, update.Status);

        var agentUpdate = update.ToUpdate();

        // Broadcast to all connected clients
        await Clients.All.AgentStatusChanged(agentUpdate);

        // Also push to the specific agent's group
        var agentGroup = AgentGroupName(update.AgentId);
        await Clients.Group(agentGroup).AgentStatusChanged(agentUpdate);
    }

    /// <summary>
    /// Adds the calling connection to the fleet group.
    /// Once joined, the client will receive all agent status changes
    /// and task progress updates across the entire fleet.
    /// </summary>
    public async Task JoinFleet()
    {
        await Groups.AddToGroupAsync(Context.ConnectionId, FleetGroupName);
        _logger.LogDebug("Connection {ConnectionId} joined fleet group", Context.ConnectionId);
    }

    /// <summary>
    /// Removes the calling connection from the fleet group.
    /// </summary>
    public async Task LeaveFleet()
    {
        await Groups.RemoveFromGroupAsync(Context.ConnectionId, FleetGroupName);
        _logger.LogDebug("Connection {ConnectionId} left fleet group", Context.ConnectionId);
    }

    /// <summary>
    /// Adds the calling connection to a specific agent's group.
    /// Once joined, the client will receive updates only for that agent.
    /// </summary>
    /// <param name="agentId">The agent identifier, e.g. "otto", "dex".</param>
    /// <exception cref="HubException">Thrown if agentId is null or empty.</exception>
    public async Task JoinAgentGroup(string agentId)
    {
        if (string.IsNullOrWhiteSpace(agentId))
            throw new HubException("agentId is required");

        var groupName = AgentGroupName(agentId);
        await Groups.AddToGroupAsync(Context.ConnectionId, groupName);
        _logger.LogDebug("Connection {ConnectionId} joined agent group {GroupName}",
            Context.ConnectionId, groupName);
    }

    /// <summary>
    /// Removes the calling connection from a specific agent's group.
    /// </summary>
    /// <param name="agentId">The agent identifier.</param>
    public async Task LeaveAgentGroup(string agentId)
    {
        if (string.IsNullOrWhiteSpace(agentId)) return;

        var groupName = AgentGroupName(agentId);
        await Groups.RemoveFromGroupAsync(Context.ConnectionId, groupName);
    }

    /// <summary>
    /// Returns a snapshot of the current fleet state.
    /// Called by clients on initial connection to get the full picture
    /// before incremental updates begin arriving.
    /// </summary>
    /// <returns>An array of <see cref="AgentCardData"/> representing all known agents.</returns>
    public Task<AgentCardData[]> GetFleetSnapshot()
    {
        // The fleet state is managed by the GatewayEventBridgeService.
        // For now, return an empty array — the bridge service will push
        // updates as they arrive from the Gateway.
        _logger.LogDebug("Fleet snapshot requested by {ConnectionId}", Context.ConnectionId);
        return Task.FromResult(Array.Empty<AgentCardData>());
    }

    /// <summary>
    /// Overrides <see cref="Hub.OnDisconnectedAsync"/> to perform cleanup.
    /// SignalR automatically removes disconnected connections from all groups.
    /// </summary>
    /// <param name="exception">Exception that caused the disconnection, if any.</param>
    public override Task OnDisconnectedAsync(Exception? exception)
    {
        _logger.LogDebug("Connection {ConnectionId} disconnected", Context.ConnectionId);
        return base.OnDisconnectedAsync(exception);
    }

    /// <summary>
    /// The SignalR group name for the entire fleet (all agents).
    /// </summary>
    internal const string FleetGroupName = "fleet";

    /// <summary>
    /// Returns the SignalR group name for a specific agent.
    /// Format: <c>agent:{agentId}</c> (lowercase for consistency).
    /// </summary>
    /// <param name="agentId">The agent identifier.</param>
    internal static string AgentGroupName(string agentId) =>
        $"agent:{agentId.ToLowerInvariant()}";
}

/// <summary>
/// Extension methods for pushing real-time agent updates through
/// the <see cref="IHubContext{T}"/> of <see cref="AgentStatusHub"/>.
///
/// <para>These methods are intended to be called from background services
/// (e.g., <see cref="Services.GatewayEventBridgeService"/>) or other
/// server-side code that detects an agent state change.</para>
/// </summary>
public static class AgentStatusHubExtensions
{
    /// <summary>
    /// Pushes an agent status change to all clients subscribed to
    /// the fleet group and the specific agent's group.
    ///
    /// <para>Call this from any background service when an agent's
    /// operational status changes (e.g., the Gateway reports a
    /// session transition from "running" to "done").</para>
    /// </summary>
    /// <param name="hubContext">The hub context injected via DI.</param>
    /// <param name="update">The agent status update payload.</param>
    /// <returns>A Task that completes when the message has been sent to all group members.</returns>
    public static async Task PushAgentStatusAsync(
        this IHubContext<AgentStatusHub, IAgentStatusClient> hubContext,
        AgentStatusUpdate update)
    {
        // Broadcast to the fleet group (all subscribers)
        await hubContext.Clients.Group(AgentStatusHub.FleetGroupName)
            .AgentStatusChanged(update);

        // Also push to the specific agent's group
        var agentGroup = AgentStatusHub.AgentGroupName(update.AgentId);
        await hubContext.Clients.Group(agentGroup)
            .AgentStatusChanged(update);
    }

    /// <summary>
    /// Pushes a task progress update to all clients subscribed to
    /// the fleet group and the specific agent's group.
    /// </summary>
    /// <param name="hubContext">The hub context injected via DI.</param>
    /// <param name="progress">The task progress update payload.</param>
    /// <returns>A Task that completes when the message has been sent to all group members.</returns>
    public static async Task PushTaskProgressAsync(
        this IHubContext<AgentStatusHub, IAgentStatusClient> hubContext,
        TaskProgressUpdate progress)
    {
        // Broadcast to the fleet group
        await hubContext.Clients.Group(AgentStatusHub.FleetGroupName)
            .AgentTaskProgress(progress);

        // Also push to the specific agent's group
        var agentGroup = AgentStatusHub.AgentGroupName(progress.AgentId);
        await hubContext.Clients.Group(agentGroup)
            .AgentTaskProgress(progress);
    }
}