feat(tasks): add task-centric activity feed endpoint

.semgrep/rules/security/idor-team-scoped-models.yaml

@@ -283,6 +283,7 @@ rules:
                               |Tag
                               |Tagger
                               |Task
+                              |TaskActivity
                               |TaskArtifact
                               |TaskAutomation
                               |TaskPresence
@@ -598,6 +599,7 @@ rules:
                         |Tag
                         |Tagger
                         |Task
+                        |TaskActivity
                         |TaskArtifact
                         |TaskAutomation
                         |TaskPresence

posthog/test/setup_receivers_baseline.txt

@@ -187,6 +187,7 @@ post_save:products.signals.backend.receivers.close_pr_when_report_dismissed
 post_save:products.slack_app.backend.signals.invalidate_repo_list_on_user_github_change
 post_save:products.slack_app.backend.signals.onboard_slack_inbox_on_install
 post_save:products.surveys.backend.models.survey_changed
+post_save:products.tasks.backend.models.project_task_created_activity
 post_save:products.tasks.backend.models.track_task_run_completion
 post_save:products.workflows.backend.models.hog_flow.hog_flow.action_saved_for_hog_flows
 post_save:products.workflows.backend.models.hog_flow.hog_flow.hog_flow_saved

products/tasks/backend/agent_proxy_callback.py

@@ -14,7 +14,7 @@
     AgentProxyCallbackResponseSerializer,
     TaskRunErrorResponseSerializer,
 )
-from products.tasks.backend.push_dispatcher import notify_task_run_awaiting_input
+from products.tasks.backend.push_dispatcher import notify_task_run_turn_completed
 
 logger = logging.getLogger(__name__)
 
@@ -131,7 +131,7 @@ def agent_proxy_callback(request, run_id: str) -> JsonResponse:
                 id=run_id, task_id=task_id, team_id=team_id
             )
             if task_run.mode == "interactive":
-                notify_task_run_awaiting_input(task_run)
+                notify_task_run_turn_completed(task_run)
                 dispatched = True
         except TaskRun.DoesNotExist:
             logger.warning("agent_proxy_callback.run_not_found", extra={"run_id": run_id})

products/tasks/backend/facade/api.py

@@ -59,6 +59,7 @@
     SandboxEnvironment,
     SandboxSnapshot,
     Task,
+    TaskActivity,
     TaskAutomation,
     TaskRun,
     TaskThreadMessage,
@@ -5151,6 +5152,10 @@ def create_thread_message(
     if _visible_task(task_id, team_id, user_id) is None:
         return None
     message = TaskThreadMessage.objects.create(team_id=team_id, task_id=task_id, author_id=user_id, content=content)
+    try:
+        project_thread_message_activity(message)
+    except Exception:
+        logger.exception("Failed to project thread message activity", extra={"message_id": str(message.id)})
     try:
         _index_thread_message_mentions(message)
     except Exception:
@@ -5169,19 +5174,29 @@ def _index_thread_message_mentions(message: TaskThreadMessage) -> None:
     mentioned_user_ids = resolve_mentioned_user_ids(
         User, message.content, team_id=message.team_id, author_id=message.author_id
     )
+    mentions = [
+        TaskThreadMessageMention(
+            team_id=message.team_id,
+            message_id=message.id,
+            task_id=message.task_id,
+            mentioned_user_id=mentioned_user_id,
+            created_at=message.created_at,
+        )
+        for mentioned_user_id in mentioned_user_ids
+    ]
     TaskThreadMessageMention.objects.for_team(message.team_id).bulk_create(
-        [
-            TaskThreadMessageMention(
-                team_id=message.team_id,
-                message_id=message.id,
-                task_id=message.task_id,
-                mentioned_user_id=mentioned_user_id,
-                created_at=message.created_at,
-            )
-            for mentioned_user_id in mentioned_user_ids
-        ],
+        mentions,
         ignore_conflicts=True,
     )
+    for mention in mentions:
+        TaskActivity.record(
+            team_id=message.team_id,
+            user_id=mention.mentioned_user_id,
+            task_id=message.task_id,
+            kind=TaskActivity.Kind.MENTION,
+            activity_at=message.created_at,
+            message_id=message.id,
+        )
 
 
 def list_mentions(
@@ -5217,6 +5232,129 @@ def list_mentions(
     ]
 
 
+def project_thread_message_activity(message: TaskThreadMessage) -> None:
+    """Project a new thread message onto the feed of everyone it concerns."""
+    recipient_ids = {recipient_id for recipient_id in (message.author_id, message.task.created_by_id) if recipient_id}
+    for recipient_id in recipient_ids:
+        TaskActivity.record(
+            team_id=message.team_id,
+            user_id=recipient_id,
+            task_id=message.task_id,
+            kind=TaskActivity.Kind.MESSAGE,
+            activity_at=message.created_at,
+            message_id=message.id,
+            actor_id=message.author_id,
+        )
+
+
+def project_awaiting_input_activity(task_run: "TaskRun") -> None:
+    """Flag the task creator's feed row when a run stops and needs them.
+
+    Called from ``push_dispatcher.notify_task_run_awaiting_input`` so every path that
+    decides a run is waiting (stream ingest, agent proxy callback, sandbox relay) projects
+    the same row. Deliberately outside the push feature flag and its Redis cooldown — the
+    in-app feed should update even where the mobile push is off.
+    """
+    creator_id = task_run.task.created_by_id
+    if creator_id is None:
+        return
+    TaskActivity.record(
+        team_id=task_run.task.team_id,
+        user_id=creator_id,
+        task_id=task_run.task_id,
+        kind=TaskActivity.Kind.AWAITING_INPUT,
+        activity_at=django_timezone.now(),
+    )
+
+
+def project_completed_activity(task_run: "TaskRun") -> None:
+    creator_id = task_run.task.created_by_id
+    if creator_id is None:
+        return
+    TaskActivity.record(
+        team_id=task_run.task.team_id,
+        user_id=creator_id,
+        task_id=task_run.task_id,
+        kind=TaskActivity.Kind.COMPLETED,
+        activity_at=task_run.completed_at or django_timezone.now(),
+    )
+
+
+def _task_activity_qs(team_id: int, user_id: int) -> QuerySet[TaskActivity]:
+    """The requester's feed rows, gated to tasks they can still see.
+
+    Rows outlive visibility changes (a task moving to a private channel, say), so the
+    visibility gate belongs on read rather than being enforced when projecting.
+    """
+    return TaskActivity.objects.filter(team_id=team_id, user_id=user_id, task__in=_visible_task_qs(team_id, user_id))
+
+
+def count_unread_task_activity(team_id: int, user_id: int | None) -> int:
+    """Unread tasks across the requester's whole feed. Backs the sidebar badge."""
+    if user_id is None:
+        return 0
+    return _task_activity_qs(team_id, user_id).filter(read_at__isnull=True).count()
+
+
+def list_task_activity(
+    team_id: int,
+    user_id: int | None,
+    *,
+    limit: int = 100,
+    before: datetime | None = None,
+    before_id: UUID | None = None,
+) -> contracts.TaskActivityPageDTO:
+    """The requester's feed: one row per task they are involved in, newest activity first.
+
+    ``unread_count`` counts every unread row the requester can see, not just the ones in
+    this page, so the sidebar badge stays honest past ``limit``.
+    """
+    if user_id is None:
+        return contracts.TaskActivityPageDTO(results=[], unread_count=0)
+    qs = _task_activity_qs(team_id, user_id)
+    if before is not None and before_id is not None:
+        qs = qs.filter(Q(activity_at__lt=before) | Q(activity_at=before, id__lt=before_id))
+    rows = list(qs.select_related("task__channel", "message__author").order_by("-activity_at", "-id")[: limit + 1])
+    has_more = len(rows) > limit
+    rows = rows[:limit]
+    next_row = rows[-1] if has_more else None
+    return contracts.TaskActivityPageDTO(
+        results=[
+            contracts.TaskActivityDTO(
+                id=row.id,
+                task_id=row.task_id,
+                task_title=row.task.title,
+                channel_id=row.task.channel_id,
+                channel_name=row.task.channel.name if row.task.channel else None,
+                activity_at=row.activity_at,
+                activity_kind=row.kind,
+                snippet=row.message.content if row.message else "",
+                latest_author=_user_basic_info(row.message.author if row.message and row.message.author_id else None),
+                latest_message_id=row.message_id,
+                is_unread=row.read_at is None,
+            )
+            for row in rows
+        ],
+        unread_count=_task_activity_qs(team_id, user_id).filter(read_at__isnull=True).count(),
+        next_before=next_row.activity_at if next_row else None,
+        next_before_id=next_row.id if next_row else None,
+    )
+
+
+def mark_task_activity_read(team_id: int, user_id: int | None, activities: Sequence[tuple[UUID, datetime]]) -> int:
+    """Mark feed rows read only when their latest activity was visible to the requester."""
+    if user_id is None or not activities:
+        return 0
+    activity_versions = Q()
+    for task_id, seen_before in activities:
+        activity_versions |= Q(task_id=task_id, activity_at__lte=seen_before)
+    return (
+        TaskActivity.objects.filter(team_id=team_id, user_id=user_id, read_at__isnull=True)
+        .filter(activity_versions)
+        .update(read_at=django_timezone.now())
+    )
+
+
 def delete_thread_message(message_id: str | UUID, task_id: str | UUID, team_id: int, user_id: int | None) -> str:
     """Delete own thread message. Returns ``ok`` / ``not_found`` / ``forbidden``."""
     message = TaskThreadMessage.objects.filter(id=message_id, task_id=task_id, team_id=team_id).first()
@@ -5298,6 +5436,7 @@ def _create_agent_thread_message(task: Task, content: str, *, event: str, payloa
         payload=payload or {},
         content=content,
     )
+    project_thread_message_activity(message)
     try:
         _index_thread_message_mentions(message)
     except Exception:

products/tasks/backend/facade/contracts.py

@@ -208,6 +208,38 @@ class TaskMentionDTO:
     author: "TaskUserBasicInfo | None" = None
 
 
+@dataclass(frozen=True)
+class TaskActivityDTO:
+    """One task the requesting user is involved in, for the task-centric activity feed.
+
+    Unlike ``TaskMentionDTO`` (one row per mention message), this is one row per task,
+    surfacing the most recent relevant activity. ``activity_kind`` classifies the winning
+    signal so the client can pick row copy; ``snippet``/``latest_author``/``latest_message_id``
+    describe the thread message tied to ``activity_at`` (empty/None when the winning signal is
+    task creation, which has no message).
+    """
+
+    id: UUID
+    task_id: UUID
+    task_title: str
+    channel_id: UUID | None
+    channel_name: str | None
+    activity_at: datetime
+    activity_kind: str
+    snippet: str
+    latest_author: "TaskUserBasicInfo | None" = None
+    latest_message_id: UUID | None = None
+    is_unread: bool = True
+
+
+@dataclass(frozen=True)
+class TaskActivityPageDTO:
+    results: list[TaskActivityDTO]
+    unread_count: int
+    next_before: datetime | None = None
+    next_before_id: UUID | None = None
+
+
 @dataclass(frozen=True)
 class TaskLatestRunSummaryDTO:
     """The latest-run status/environment pair nested in a task summary response."""

products/tasks/backend/logic/stream/event_ingest.py

@@ -12,13 +12,11 @@
 from django.db import InterfaceError, OperationalError, close_old_connections
 
 import structlog
-import posthoganalytics
 from asgiref.sync import sync_to_async
 from jwt import PyJWTError
 
 from posthog.ph_client import ph_scoped_capture
 
-from products.tasks.backend.constants import STREAM_VIA_PROXY_FEATURE_FLAG
 from products.tasks.backend.logic.services.connection_token import (
     SandboxEventIngestTokenPayload,
     validate_sandbox_event_ingest_token,
@@ -31,7 +29,7 @@
     get_task_run_stream_key,
 )
 from products.tasks.backend.models import TaskRun
-from products.tasks.backend.push_dispatcher import notify_task_run_awaiting_input
+from products.tasks.backend.push_dispatcher import notify_task_run_turn_completed
 
 from ee.hogai.sandbox import is_turn_complete
 
@@ -374,7 +372,7 @@ def _parse_ingest_line(line: str) -> EventIngestEventLine | EventIngestCompleteL
 async def _heartbeat_workflow_if_needed(redis_stream: TaskRunRedisStream, run_id: str, event: dict) -> None:
     if is_turn_complete(event):
         await redis_stream.set_agent_active(False)
-        await _dispatch_awaiting_input_if_interactive(run_id)
+        await _dispatch_turn_completed_if_interactive(run_id)
         return
 
     if _is_session_update(event):
@@ -410,54 +408,24 @@ def _heartbeat_workflow(run_id: str, agent_active: bool) -> None:
     task_run.heartbeat_workflow(agent_active=agent_active)
 
 
-async def _dispatch_awaiting_input_if_interactive(run_id: str) -> None:
-    """Notify when an interactive run finishes a turn and idles for input."""
-    await sync_to_async(_dispatch_awaiting_input_if_interactive_sync, thread_sensitive=True)(run_id)
+async def _dispatch_turn_completed_if_interactive(run_id: str) -> None:
+    await sync_to_async(_dispatch_turn_completed_if_interactive_sync, thread_sensitive=True)(run_id)
 
 
-def _dispatch_awaiting_input_if_interactive_sync(run_id: str) -> None:
+def _dispatch_turn_completed_if_interactive_sync(run_id: str) -> None:
     if not settings.TEST:
         close_old_connections()
 
     try:
         task_run = TaskRun.objects.select_related("task__created_by", "team").get(id=run_id)
     except TaskRun.DoesNotExist:
-        logger.warning("task_run_event_ingest_awaiting_input_run_missing", run_id=run_id)
+        logger.warning("task_run_event_ingest_turn_completed_run_missing", run_id=run_id)
         return
 
     if task_run.mode != "interactive":
         return
 
-    if not _awaiting_input_push_enabled(task_run):
-        return
-
-    notify_task_run_awaiting_input(task_run)
-
-
-def _awaiting_input_push_enabled(task_run: TaskRun) -> bool:
-    """Awaiting-input pushes ship with the proxy-streaming rollout: gate them on the same flag
-    so deploying this code changes nothing until the rollout starts. Local dev disables the
-    analytics SDK, so the flag never evaluates there; DEBUG is the opt-in, mirroring the
-    stream_token endpoint. Fails closed on flag-evaluation errors."""
-    if settings.DEBUG:
-        return True
-    user = task_run.task.created_by
-    if user is None:
-        return False
-    organization_id = str(task_run.team.organization_id)
-    try:
-        return bool(
-            posthoganalytics.feature_enabled(
-                STREAM_VIA_PROXY_FEATURE_FLAG,
-                user.distinct_id or f"user_{user.id}",
-                groups={"organization": organization_id},
-                group_properties={"organization": {"id": organization_id}},
-                only_evaluate_locally=False,
-                send_feature_flag_events=False,
-            )
-        )
-    except Exception:
-        return False
+    notify_task_run_turn_completed(task_run)
 
 
 def _is_session_update(event: dict) -> bool: