feat(dashboards): add live activity dashboard widget

frontend/src/scenes/dashboard/tileLayouts.test.ts

@@ -134,6 +134,12 @@ describe('calculating tile layouts', () => {
             expectedMinH: 3,
             expectedMinW: 3,
         },
+        {
+            name: 'uses catalog minW for live activity widgets',
+            widgetType: 'live_activity',
+            expectedMinH: 3,
+            expectedMinW: 3,
+        },
     ])('$name', ({ widgetType, expectedMinH, expectedMinW }) => {
         const tiles: DashboardTile<QueryBasedInsightModel>[] = [
             {

posthog/settings/web.py

@@ -548,6 +548,7 @@ def static_varies_origin(headers, path, url):
         # stable component name; value is the singleton list even though length is 1.
         "ActivityEventsListWidgetTypeEnum": ["activity_events_list"],
         "ErrorTrackingListWidgetTypeEnum": ["error_tracking_list"],
+        "LiveActivityWidgetTypeEnum": ["live_activity"],
         "SessionReplayListWidgetTypeEnum": ["session_replay_list"],
         "OrderByEnum": ["latest", "earliest"],
         "PropertyGroupTypeEnum": ["cohort", "person", "group"],

products/dashboards/backend/api/test/test_dashboard_widgets.py

@@ -570,9 +570,14 @@ def test_widget_catalog_lists_registered_types(self) -> None:
 
         results = response.json()["results"]
         assert any(entry["widget_type"] == "error_tracking_list" for entry in results)
+        assert any(entry["widget_type"] == "live_activity" for entry in results)
         error_tracking_list = next(entry for entry in results if entry["widget_type"] == "error_tracking_list")
         assert error_tracking_list["config_schema"]["properties"]["limit"]["default"] == DEFAULT_WIDGET_LIST_LIMIT
         assert error_tracking_list["availability_requirements"] == ["exception_autocapture"]
+        live_activity = next(entry for entry in results if entry["widget_type"] == "live_activity")
+        assert live_activity["group_id"] == "activity"
+        assert live_activity["config_schema"]["properties"]["refreshIntervalSeconds"]["minimum"] == 15
+        assert live_activity["config_schema"]["properties"]["refreshIntervalSeconds"]["maximum"] == 60
 
     @override_settings(IN_UNIT_TESTING=True)
     def test_add_widget_via_batch_endpoint(self) -> None:
@@ -594,6 +599,56 @@ def test_add_widget_via_batch_endpoint(self) -> None:
         assert tile["widget"]["name"] == "Errors"
         assert tile["layouts"]["sm"]["w"] == 6
 
+    @override_settings(IN_UNIT_TESTING=True)
+    def test_add_live_activity_widget_via_batch_endpoint_uses_compact_layout(self) -> None:
+        dashboard_id, _ = self.dashboard_api.create_dashboard({"name": "dashboard"})
+
+        response = self.client.post(
+            f"/api/projects/{self.team.id}/dashboards/{dashboard_id}/widgets/batch/",
+            {
+                "widgets": [
+                    {
+                        "widget_type": "live_activity",
+                        "config": {"limit": 5, "refreshIntervalSeconds": 15},
+                        "name": "Live activity",
+                    },
+                ]
+            },
+        )
+        assert response.status_code == status.HTTP_201_CREATED
+
+        tile = response.json()["tiles"][0]
+        assert tile["widget"]["widget_type"] == "live_activity"
+        assert tile["widget"]["config"] == {"limit": 5, "refreshIntervalSeconds": 15}
+        assert tile["widget"]["name"] == "Live activity"
+        assert tile["layouts"]["sm"]["w"] == 4
+        assert tile["layouts"]["sm"]["h"] == 4
+
+    @override_settings(IN_UNIT_TESTING=True)
+    def test_live_activity_widget_create_and_update_validate_refresh_interval(self) -> None:
+        dashboard_id, _ = self.dashboard_api.create_dashboard({"name": "dashboard"})
+
+        response = self.client.post(
+            f"/api/projects/{self.team.id}/dashboards/{dashboard_id}/widgets/batch/",
+            {"widgets": [{"widget_type": "live_activity", "config": {"refreshIntervalSeconds": 5}}]},
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+        _, dashboard_json = self.dashboard_api.create_widget_tile(
+            dashboard_id,
+            widget_type="live_activity",
+            config={"limit": 5, "refreshIntervalSeconds": 15},
+        )
+        tile = dashboard_json["tiles"][0]
+        tile["widget"]["config"] = {"limit": 5, "refreshIntervalSeconds": 61}
+
+        response = self.client.patch(
+            f"/api/projects/{self.team.id}/dashboards/{dashboard_id}",
+            {"tiles": [tile]},
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert response.json()["attr"] == "config"
+
     @override_settings(IN_UNIT_TESTING=True)
     def test_widgets_batch_endpoint_rejects_legacy_error_tracking_widget_type(self) -> None:
         dashboard_id, _ = self.dashboard_api.create_dashboard({"name": "dashboard"})

products/dashboards/backend/api/test/test_run_widgets.py

@@ -1,6 +1,7 @@
 import typing
 from typing import Any, cast
 
+from freezegun import freeze_time
 from posthog.test.base import APIBaseTest
 from unittest.mock import MagicMock, patch
 
@@ -28,6 +29,8 @@
     ACTIVITY_EVENTS_DEFAULT_LIMIT,
     ACTIVITY_EVENTS_MAX_LIMIT,
     DEFAULT_WIDGET_LIST_LIMIT,
+    LIVE_ACTIVITY_DEFAULT_LIMIT,
+    LIVE_ACTIVITY_MAX_LIMIT,
     MAX_WIDGET_RESULT_LIMIT,
     MAX_WIDGETS_BATCH_SIZE,
 )
@@ -36,6 +39,7 @@
 from products.dashboards.backend.widget_specs.configs import (
     ACTIVITY_EVENTS_LIST_WIDGET_TYPE,
     ERROR_TRACKING_LIST_WIDGET_TYPE,
+    LIVE_ACTIVITY_WIDGET_TYPE,
     SESSION_REPLAY_LIST_WIDGET_TYPE,
     ErrorTrackingListWidgetConfig,
     SessionReplayOrderBy,
@@ -46,6 +50,11 @@
     run_activity_events_list_widget,
 )
 from products.dashboards.backend.widgets.error_tracking_list import run_error_tracking_list_widget
+from products.dashboards.backend.widgets.live_activity import (
+    LIVE_ACTIVITY_BUCKET_COUNT,
+    LIVE_ACTIVITY_BUCKET_SECONDS,
+    run_live_activity_widget,
+)
 from products.dashboards.backend.widgets.session_replay_list import run_session_replay_list_widget
 from products.error_tracking.backend.api.query_utils import ERROR_TRACKING_LISTING_VOLUME_RESOLUTION
 
@@ -103,10 +112,26 @@ def test_validate_activity_events_list_config_rejects_limit_above_max(self) -> N
         with self.assertRaises(Exception):
             validate_widget_config(ACTIVITY_EVENTS_LIST_WIDGET_TYPE, {"limit": ACTIVITY_EVENTS_MAX_LIMIT + 1})
 
+    def test_validate_live_activity_config_defaults(self) -> None:
+        validated = validate_widget_config(LIVE_ACTIVITY_WIDGET_TYPE, {})
+        assert validated["limit"] == LIVE_ACTIVITY_DEFAULT_LIMIT
+        assert validated["refreshIntervalSeconds"] == 15
+        assert "filterTestAccounts" not in validated
+
+    def test_validate_live_activity_config_accepts_limit_up_to_max(self) -> None:
+        validated = validate_widget_config(LIVE_ACTIVITY_WIDGET_TYPE, {"limit": LIVE_ACTIVITY_MAX_LIMIT})
+        assert validated["limit"] == LIVE_ACTIVITY_MAX_LIMIT
+
+    @parameterized.expand([0, 14, 61])
+    def test_validate_live_activity_config_rejects_unsupported_refresh_interval(self, refresh_interval: int) -> None:
+        with self.assertRaises(Exception):
+            validate_widget_config(LIVE_ACTIVITY_WIDGET_TYPE, {"refreshIntervalSeconds": refresh_interval})
+
     @parameterized.expand(
         [
             ("activity_events", ACTIVITY_EVENTS_LIST_WIDGET_TYPE),
             ("error_tracking", ERROR_TRACKING_LIST_WIDGET_TYPE),
+            ("live_activity", LIVE_ACTIVITY_WIDGET_TYPE),
             ("session_replay", SESSION_REPLAY_LIST_WIDGET_TYPE),
         ]
     )
@@ -121,6 +146,7 @@ def test_validate_list_config_rejects_invalid_filter_test_accounts(self, _label:
         [
             ("activity_events", ACTIVITY_EVENTS_LIST_WIDGET_TYPE),
             ("error_tracking", ERROR_TRACKING_LIST_WIDGET_TYPE),
+            ("live_activity", LIVE_ACTIVITY_WIDGET_TYPE),
             ("session_replay", SESSION_REPLAY_LIST_WIDGET_TYPE),
         ]
     )
@@ -346,6 +372,73 @@ def test_runs_activity_events_widget_for_requested_tile(self, mock_runner_cls: M
         self.assertEqual(query.after, "-24h")
         self.assertEqual(query.limit, 10)
 
+    @freeze_time("2026-06-17T12:00:00Z")
+    @patch("products.dashboards.backend.widgets.live_activity.HogQLQueryRunner")
+    def test_runs_live_activity_widget_for_requested_tile(self, mock_runner_cls: MagicMock) -> None:
+        def runner_side_effect(*_args: object, **kwargs: object) -> MagicMock:
+            query = kwargs["query"]
+            if "events_in_window" in query.query:
+                results = [[9, 3, "2026-06-17T11:59:50Z"]]
+            elif "bucket_index" in query.query:
+                results = [[19, 9]]
+            else:
+                results = [
+                    [
+                        "0196e9ab-0000-0000-0000-000000000000",
+                        "$pageview",
+                        "Alex Chen",
+                        "https://app.example.test/dashboard",
+                        "web",
+                        "2026-06-17T11:59:50Z",
+                        "user-1",
+                    ]
+                ]
+            return MagicMock(
+                calculate=MagicMock(return_value=MagicMock(model_dump=lambda mode="json": {"results": results}))
+            )
+
+        mock_runner_cls.side_effect = runner_side_effect
+        dashboard_id, _ = self.dashboard_api.create_dashboard({"name": "dash"})
+        _, dashboard_json = self.dashboard_api.create_widget_tile(
+            dashboard_id,
+            widget_type=LIVE_ACTIVITY_WIDGET_TYPE,
+            config={"limit": 5, "refreshIntervalSeconds": 15},
+        )
+        tile_id = dashboard_json["tiles"][0]["id"]
+
+        body = self._run(dashboard_id, [tile_id])
+
+        self.assertEqual(len(body["results"]), 1)
+        self.assertEqual(body["results"][0]["tile_id"], tile_id)
+        self.assertEqual(body["results"][0]["widget_type"], LIVE_ACTIVITY_WIDGET_TYPE)
+        self.assertIsNone(body["results"][0]["error"])
+        result = body["results"][0]["result"]
+        self.assertEqual(result["activeUsers"], 3)
+        self.assertEqual(result["eventsInWindow"], 9)
+        self.assertEqual(result["events"][0]["surface"], "web")
+        self.assertEqual(result["limit"], 5)
+        self.assertEqual(result["bucketSeconds"], LIVE_ACTIVITY_BUCKET_SECONDS)
+        self.assertEqual(len(result["pulse"]), LIVE_ACTIVITY_BUCKET_COUNT)
+        self.assertEqual(mock_runner_cls.call_count, 3)
+        for call in mock_runner_cls.call_args_list:
+            self.assertEqual(call.kwargs["user"], self.user)
+
+    @patch("products.dashboards.backend.widgets.live_activity.HogQLQueryRunner")
+    def test_live_activity_widget_tags_queries_in_debug_mode(self, mock_runner_cls: MagicMock) -> None:
+        def calculate_with_tag_assertion() -> MagicMock:
+            tags = get_query_tags()
+            if tags.product != Product.PRODUCT_ANALYTICS or tags.feature != Feature.QUERY:
+                raise UntaggedQueryError("live activity widget must tag ClickHouse queries")
+            return MagicMock(model_dump=lambda mode="json": {"results": []})
+
+        mock_runner_cls.return_value.calculate.side_effect = calculate_with_tag_assertion
+
+        with override_settings(DEBUG=True, TEST=False):
+            result = run_live_activity_widget(self.team, {"limit": 5}, user=self.user)
+
+        self.assertEqual(result["limit"], 5)
+        self.assertEqual(mock_runner_cls.call_count, 3)
+
     @patch("products.dashboards.backend.widgets.activity_events_list.EventsQueryRunner")
     def test_activity_events_widget_applies_date_range_and_widget_filters(self, mock_runner_cls: MagicMock) -> None:
         mock_runner_cls.return_value.calculate.return_value = MagicMock(

products/dashboards/backend/api/test/test_widget_config_schema_parity.py

@@ -4,15 +4,18 @@
     ACTIVITY_EVENTS_DEFAULT_LIMIT,
     ACTIVITY_EVENTS_MAX_LIMIT,
     DEFAULT_WIDGET_LIST_LIMIT,
+    LIVE_ACTIVITY_DEFAULT_LIMIT,
+    LIVE_ACTIVITY_MAX_LIMIT,
     MAX_WIDGET_RESULT_LIMIT,
 )
 from products.dashboards.backend.widget_catalog import WIDGET_CATALOG
-from products.dashboards.backend.widget_specs.configs import ACTIVITY_EVENTS_LIST_WIDGET_TYPE
+from products.dashboards.backend.widget_specs.configs import ACTIVITY_EVENTS_LIST_WIDGET_TYPE, LIVE_ACTIVITY_WIDGET_TYPE
 from products.dashboards.backend.widget_specs.registry import WIDGET_SPECS
 
 # Activity events allows a larger page than other list widgets; other types share the default bounds.
 WIDGET_LIMIT_BOUNDS = {
     ACTIVITY_EVENTS_LIST_WIDGET_TYPE: (ACTIVITY_EVENTS_MAX_LIMIT, ACTIVITY_EVENTS_DEFAULT_LIMIT),
+    LIVE_ACTIVITY_WIDGET_TYPE: (LIVE_ACTIVITY_MAX_LIMIT, LIVE_ACTIVITY_DEFAULT_LIMIT),
 }
 
 

products/dashboards/backend/constants.py

@@ -11,6 +11,10 @@
 ACTIVITY_EVENTS_MAX_LIMIT = 50
 ACTIVITY_EVENTS_DEFAULT_LIMIT = 25
 
+# Live activity is intentionally compact: active users + pulse + a short recent-events feed.
+LIVE_ACTIVITY_MAX_LIMIT = 10
+LIVE_ACTIVITY_DEFAULT_LIMIT = 5
+
 # Cap widgets per batch create / run-widgets request.
 MAX_WIDGETS_BATCH_SIZE = 10
 

products/dashboards/backend/test/test_widget_layouts.py

@@ -1,6 +1,7 @@
 from parameterized import parameterized
 
 from products.dashboards.backend.widget_layouts import stack_widget_layout_at_bottom
+from products.dashboards.backend.widget_specs.configs import LIVE_ACTIVITY_WIDGET_TYPE
 
 
 class TestWidgetLayouts:
@@ -75,12 +76,14 @@ def test_stack_widget_layout_at_bottom_narrow_tile_anchors_to_tallest_column(sel
         ]
 
         layouts = stack_widget_layout_at_bottom(
-            widget_type="error_tracking_list",
+            widget_type=LIVE_ACTIVITY_WIDGET_TYPE,
             existing_sm_layouts=existing,
         )
 
         placement = layouts["sm"]
         assert placement["y"] == 14
+        assert placement["w"] == 4
+        assert placement["h"] == 4
         # column span must include the tallest column (8)
         assert placement["x"] <= 8 < placement["x"] + placement["w"]
 
@@ -97,3 +100,12 @@ def test_stack_widget_layout_at_bottom_stacks_batch_on_empty_dashboard(self) ->
 
         assert first["sm"] == {"x": 0, "y": 0, "w": 6, "h": 5}
         assert second["sm"] == {"x": 0, "y": 5, "w": 6, "h": 5}
+
+    def test_live_activity_uses_compact_default_layout(self) -> None:
+        layouts = stack_widget_layout_at_bottom(
+            widget_type=LIVE_ACTIVITY_WIDGET_TYPE,
+            existing_sm_layouts=[],
+        )
+
+        assert layouts["sm"] == {"x": 0, "y": 0, "w": 4, "h": 4}
+        assert layouts["xs"] == layouts["sm"]

products/dashboards/backend/widget_catalog.py

@@ -51,8 +51,8 @@ def get_widget_product_access_denied_message(required_product_access: str) -> st
 
 
 def get_default_widget_layouts(widget_type: str) -> dict[str, dict[str, int]]:
-    width = 6
-    height = 5
+    spec = WIDGET_SPECS.get(widget_type)
+    width, height = spec.default_layout if spec else (6, 5)
     return {
         "sm": {"x": 0, "y": 0, "w": width, "h": height},
         "xs": {"x": 0, "y": 0, "w": width, "h": height},

products/dashboards/backend/widget_registry.py

@@ -9,6 +9,7 @@
 from products.dashboards.backend.widget_specs.configs import (
     ACTIVITY_EVENTS_LIST_WIDGET_TYPE,
     ERROR_TRACKING_LIST_WIDGET_TYPE,
+    LIVE_ACTIVITY_WIDGET_TYPE,
     SESSION_REPLAY_LIST_WIDGET_TYPE,
 )
 from products.dashboards.backend.widget_specs.registry import (
@@ -29,6 +30,7 @@
 __all__ = [
     "ACTIVITY_EVENTS_LIST_WIDGET_TYPE",
     "ERROR_TRACKING_LIST_WIDGET_TYPE",
+    "LIVE_ACTIVITY_WIDGET_TYPE",
     "SESSION_REPLAY_LIST_WIDGET_TYPE",
     "EXPECTED_WIDGET_TYPES",
     "DashboardWidgetType",

products/dashboards/backend/widget_specs/common.py

@@ -8,6 +8,7 @@
 
 from products.dashboards.backend.constants import (
     ACTIVITY_EVENTS_MAX_LIMIT,
+    LIVE_ACTIVITY_MAX_LIMIT,
     MAX_WIDGET_RESULT_LIMIT,
     WIDGET_DATE_FROM_VALUES_ORDERED,
 )
@@ -91,3 +92,4 @@ def validate_date_range(cls, value: object) -> WidgetDateRange | None:
 
 WidgetLimit = Annotated[int, Field(ge=1, le=MAX_WIDGET_RESULT_LIMIT)]
 ActivityWidgetLimit = Annotated[int, Field(ge=1, le=ACTIVITY_EVENTS_MAX_LIMIT)]
+LiveActivityWidgetLimit = Annotated[int, Field(ge=1, le=LIVE_ACTIVITY_MAX_LIMIT)]