diff --git a/.semgrep/rules/security/idor-team-scoped-models.yaml b/.semgrep/rules/security/idor-team-scoped-models.yaml index 77a87128dd31..b6b48e5d3389 100644 --- a/.semgrep/rules/security/idor-team-scoped-models.yaml +++ b/.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 diff --git a/posthog/test/setup_receivers_baseline.txt b/posthog/test/setup_receivers_baseline.txt index f7af1a1a6406..313b5d1aa263 100644 --- a/posthog/test/setup_receivers_baseline.txt +++ b/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 diff --git a/products/tasks/backend/agent_proxy_callback.py b/products/tasks/backend/agent_proxy_callback.py index f3dd41be37b3..e816b0cf7751 100644 --- a/products/tasks/backend/agent_proxy_callback.py +++ b/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}) diff --git a/products/tasks/backend/facade/api.py b/products/tasks/backend/facade/api.py index cb5a8cd7b40f..b17fcdb525d2 100644 --- a/products/tasks/backend/facade/api.py +++ b/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: diff --git a/products/tasks/backend/facade/contracts.py b/products/tasks/backend/facade/contracts.py index 4abb12017136..67c7f0675a32 100644 --- a/products/tasks/backend/facade/contracts.py +++ b/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.""" diff --git a/products/tasks/backend/logic/stream/event_ingest.py b/products/tasks/backend/logic/stream/event_ingest.py index 151c9ac6d56f..f996f1388a1e 100644 --- a/products/tasks/backend/logic/stream/event_ingest.py +++ b/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: diff --git a/products/tasks/backend/migrations/0073_task_activity.py b/products/tasks/backend/migrations/0073_task_activity.py new file mode 100644 index 000000000000..9671bb36faa0 --- /dev/null +++ b/products/tasks/backend/migrations/0073_task_activity.py @@ -0,0 +1,83 @@ +import django.db.models.deletion +from django.db import migrations, models + +import posthog.uuidt + + +class Migration(migrations.Migration): + dependencies = [("tasks", "0072_loop_skill_bundles")] + operations = [ + migrations.CreateModel( + name="TaskActivity", + fields=[ + ( + "id", + models.UUIDField(default=posthog.uuidt.uuid7, editable=False, primary_key=True, serialize=False), + ), + ( + "kind", + models.CharField( + choices=[ + ("created", "Created"), + ("mention", "Mention"), + ("message", "Message"), + ("awaiting_input", "Awaiting input"), + ("completed", "Completed"), + ], + max_length=32, + ), + ), + ("activity_at", models.DateTimeField()), + ("read_at", models.DateTimeField(blank=True, null=True)), + ( + "message", + models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="activity_rows", + to="tasks.taskthreadmessage", + ), + ), + ( + "task", + models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="+", to="tasks.task"), + ), + ( + "team", + models.ForeignKey( + db_constraint=False, + on_delete=django.db.models.deletion.CASCADE, + related_name="+", + to="posthog.team", + ), + ), + ( + "user", + models.ForeignKey( + db_constraint=False, + on_delete=django.db.models.deletion.CASCADE, + related_name="+", + to="posthog.user", + ), + ), + ], + options={"db_table": "posthog_task_activity"}, + ), + migrations.AddConstraint( + model_name="taskactivity", + constraint=models.UniqueConstraint(fields=("team", "user", "task"), name="task_activity_user_task_unique"), + ), + migrations.AddIndex( + model_name="taskactivity", + index=models.Index(fields=["team", "user", "activity_at", "id"], name="task_activity_feed_idx"), + ), + migrations.AddIndex( + model_name="taskactivity", + index=models.Index( + fields=["team", "user"], + condition=models.Q(read_at__isnull=True), + name="task_activity_unread_idx", + ), + ), + ] diff --git a/products/tasks/backend/migrations/max_migration.txt b/products/tasks/backend/migrations/max_migration.txt index e12dffe2bd4e..2f9da34a1494 100644 --- a/products/tasks/backend/migrations/max_migration.txt +++ b/products/tasks/backend/migrations/max_migration.txt @@ -1 +1 @@ -0072_loop_skill_bundles +0073_task_activity diff --git a/products/tasks/backend/models.py b/products/tasks/backend/models.py index 6a64428598b4..c7b1f1e07a63 100644 --- a/products/tasks/backend/models.py +++ b/products/tasks/backend/models.py @@ -20,7 +20,7 @@ from django.conf import settings from django.contrib.postgres.fields import ArrayField from django.core.exceptions import ValidationError -from django.db import IntegrityError, models, transaction +from django.db import IntegrityError, connection, models, transaction from django.db.models.fields.json import KeyTransform from django.utils import timezone as django_timezone @@ -41,6 +41,7 @@ from posthog.models.utils import DeletedMetaFields, UUIDModel from posthog.storage import object_storage from posthog.temporal.oauth import PosthogMcpScopes +from posthog.uuidt import uuid7 from products.tasks.backend.constants import DEFAULT_TRUSTED_DOMAINS from products.tasks.backend.error_telemetry import truncate_error_message @@ -906,7 +907,9 @@ class AuthorKind(models.TextChoices): class Meta: db_table = "posthog_task_thread_message" - indexes = [models.Index(fields=["task", "created_at"], name="task_thread_msg_task_created")] + indexes = [ + models.Index(fields=["task", "created_at"], name="task_thread_msg_task_created"), + ] def __str__(self): return f"Thread message {self.id} on task {self.task_id}" @@ -940,6 +943,106 @@ def __str__(self): return f"Mention of user {self.mentioned_user_id} in message {self.message_id}" +class TaskActivity(TeamScopedRootMixin): + """One row per (user, task): the latest thing that happened on a task the user is + involved in, plus whether they have seen it. + + Collapsing to one row per task is what makes "read" a property of the task rather + than of a feed cursor, so opening the task from anywhere clears it. Rows are + projected on write by ``products.tasks.backend.facade.api``. + """ + + class Kind(models.TextChoices): + CREATED = "created", "Created" + MENTION = "mention", "Mention" + MESSAGE = "message", "Message" + AWAITING_INPUT = "awaiting_input", "Awaiting input" + COMPLETED = "completed", "Completed" + + # uuid7 rather than the uuid4 the sibling task models use: rows are insert-heavy and + # read newest-first, so a time-ordered key keeps the index appends local and makes the + # id a meaningful tiebreak when two rows share an activity_at. + id = models.UUIDField(primary_key=True, default=uuid7, editable=False) + team = models.ForeignKey("posthog.Team", on_delete=models.CASCADE, related_name="+", db_constraint=False) + user = models.ForeignKey("posthog.User", on_delete=models.CASCADE, related_name="+", db_constraint=False) + task = models.ForeignKey(Task, on_delete=models.CASCADE, related_name="+") + message = models.ForeignKey( + TaskThreadMessage, on_delete=models.SET_NULL, null=True, blank=True, related_name="activity_rows" + ) + kind = models.CharField(max_length=32, choices=Kind) + activity_at = models.DateTimeField() + read_at = models.DateTimeField(null=True, blank=True) + + class Meta: + db_table = "posthog_task_activity" + constraints = [models.UniqueConstraint(fields=["team", "user", "task"], name="task_activity_user_task_unique")] + indexes = [ + models.Index(fields=["team", "user", "activity_at", "id"], name="task_activity_feed_idx"), + models.Index( + fields=["team", "user"], condition=models.Q(read_at__isnull=True), name="task_activity_unread_idx" + ), + ] + + @classmethod + def record( + cls, + *, + team_id: int, + user_id: int, + task_id: uuid.UUID | str, + kind: str, + activity_at: datetime, + message_id: uuid.UUID | None = None, + actor_id: int | None = None, + ) -> None: + """Record the latest activity on ``task_id`` for ``user_id``, newest-wins. + + A single upsert rather than read-modify-write: two messages landing on the same + task concurrently would otherwise race and lose one. The ``WHERE`` on the conflict + clause is what makes it newest-wins, so an out-of-order write (a retried Temporal + activity, say) can't drag ``activity_at`` backwards. + + Activity the user caused themselves lands already-read — their own reply should + never light up their own unread badge. + """ + read_at = activity_at if actor_id is not None and actor_id == user_id else None + with connection.cursor() as cursor: + cursor.execute( + f""" + INSERT INTO {cls._meta.db_table} + (id, team_id, user_id, task_id, message_id, kind, activity_at, read_at) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s) + ON CONFLICT (team_id, user_id, task_id) DO UPDATE + SET message_id = EXCLUDED.message_id, + kind = EXCLUDED.kind, + activity_at = EXCLUDED.activity_at, + read_at = CASE + WHEN {cls._meta.db_table}.activity_at = EXCLUDED.activity_at + THEN {cls._meta.db_table}.read_at + ELSE EXCLUDED.read_at + END + WHERE {cls._meta.db_table}.activity_at <= EXCLUDED.activity_at + """, + [uuid7(), team_id, user_id, task_id, message_id, kind, activity_at, read_at], + ) + + +@receiver(post_save, sender=Task) +def project_task_created_activity(sender, instance: Task, created: bool, **kwargs) -> None: + """Seed the creator's activity row. A signal rather than a facade call because tasks are + created from several paths (API, automations, the sandbox warm path) and every one of them + should show up in its creator's feed.""" + if created and instance.created_by_id is not None: + TaskActivity.record( + team_id=instance.team_id, + user_id=instance.created_by_id, + task_id=instance.id, + kind=TaskActivity.Kind.CREATED, + activity_at=instance.created_at, + actor_id=instance.created_by_id, + ) + + class ChannelFeedMessage(TeamScopedRootMixin): """A durable, team-visible announcement in a channel's feed — rendered alongside task cards as a "PostHog agent" system row (e.g. "Adam created this context"). diff --git a/products/tasks/backend/presentation/serializers.py b/products/tasks/backend/presentation/serializers.py index 8b267f3edcfa..6a6aaca41a3f 100644 --- a/products/tasks/backend/presentation/serializers.py +++ b/products/tasks/backend/presentation/serializers.py @@ -27,6 +27,8 @@ ChannelFeedMessageDTO, SandboxCustomImageDTO, SandboxEnvironmentDTO, + TaskActivityDTO, + TaskActivityPageDTO, TaskAutomationDTO, TaskDetailDTO, TaskMentionDTO, @@ -780,7 +782,7 @@ class TaskRunArtifactUploadSerializer(serializers.Serializer): help_text="Optional structured metadata for special artifact types, such as skill bundles.", ) - def validate(self, attrs): + def validate(self, attrs: dict[str, Any]) -> dict[str, Any]: attrs = validate_task_run_artifact_metadata(attrs) content = attrs["content"] content_encoding = attrs.get("content_encoding", "utf-8") @@ -1474,6 +1476,117 @@ class Meta: ] +class TaskActivityQuerySerializer(serializers.Serializer): + """Query parameters for the task-centric activity feed.""" + + limit = serializers.IntegerField( + required=False, + default=100, + min_value=1, + max_value=500, + help_text="Maximum number of tasks to return (most recent activity first).", + ) + before = serializers.DateTimeField( + required=False, + help_text="Activity timestamp from the final row of the previous page.", + ) + before_id = serializers.UUIDField( + required=False, + help_text="Activity ID from the final row of the previous page.", + ) + + def validate(self, attrs): + if ("before" in attrs) != ("before_id" in attrs): + raise serializers.ValidationError("before and before_id must be provided together") + return attrs + + +class TaskActivitySerializer(DataclassSerializer): + """Response shape for one task in the requester's activity feed (one row per task).""" + + latest_author = TaskUserBasicInfoSerializer( + allow_null=True, + required=False, + help_text="Author of the thread message tied to the latest activity, when one applies.", + ) + activity_kind = serializers.ChoiceField( + choices=["awaiting_input", "completed", "mention", "message", "created"], + help_text=( + "What the latest activity on this task was: an agent run waiting on the requester " + "(awaiting_input), a completed run (completed), someone @-mentioning them (mention), " + "a thread reply (message), or their creating the task (created)." + ), + ) + snippet = serializers.CharField( + help_text="Content of the thread message tied to the latest activity; empty for task-creation rows." + ) + is_unread = serializers.BooleanField( + help_text="Whether the requester has yet to see this activity. Activity they caused themselves is never unread." + ) + + class Meta: + dataclass = TaskActivityDTO + fields = [ + "id", + "task_id", + "task_title", + "channel_id", + "channel_name", + "activity_at", + "activity_kind", + "snippet", + "latest_author", + "latest_message_id", + "is_unread", + ] + + +class TaskActivityPageSerializer(DataclassSerializer): + """A page of the requester's activity feed, plus the unread total across the whole feed.""" + + results = TaskActivitySerializer(many=True, help_text="Tasks with activity, most recent first.") + unread_count = serializers.IntegerField( + help_text="Unread tasks across the requester's whole feed, not just this page. Backs the sidebar badge." + ) + next_before = serializers.DateTimeField( + allow_null=True, + required=False, + help_text="Activity timestamp to pass as before for the next page, or null on the final page.", + ) + next_before_id = serializers.UUIDField( + allow_null=True, + required=False, + help_text="Activity ID to pass as before_id for the next page, or null on the final page.", + ) + + class Meta: + dataclass = TaskActivityPageDTO + fields = ["results", "unread_count", "next_before", "next_before_id"] + + +class TaskActivityReadMarkerSerializer(serializers.Serializer): + task_id = serializers.UUIDField(help_text="Task whose displayed activity should be marked read.") + seen_before = serializers.DateTimeField( + help_text="Mark activity at or before this timestamp read without clearing newer activity." + ) + + +class TaskActivityMarkReadSerializer(serializers.Serializer): + """Request body for clearing the unread flag on specific tasks.""" + + activities = serializers.ListField( + child=TaskActivityReadMarkerSerializer(), + allow_empty=False, + max_length=500, + help_text="Displayed task activities to mark read if they have not changed.", + ) + + +class TaskActivityMarkReadResponseSerializer(serializers.Serializer): + marked_read = serializers.IntegerField(help_text="How many feed rows changed from unread to read.") + unread_count = serializers.IntegerField(help_text="The requester's remaining unread total after the update.") + + class TaskRepositoriesResponseSerializer(serializers.Serializer): repositories = serializers.ListField( child=serializers.CharField(), diff --git a/products/tasks/backend/presentation/views/channels_api.py b/products/tasks/backend/presentation/views/channels_api.py index 1891ecc2238a..deea7f7f75e9 100644 --- a/products/tasks/backend/presentation/views/channels_api.py +++ b/products/tasks/backend/presentation/views/channels_api.py @@ -1,5 +1,7 @@ +from typing import Any from uuid import UUID +from drf_spectacular.openapi import AutoSchema from drf_spectacular.utils import OpenApiResponse, extend_schema from rest_framework import status, viewsets from rest_framework.authentication import SessionAuthentication @@ -19,6 +21,11 @@ ChannelFeedMessageWriteSerializer, ChannelSerializer, ChannelWriteSerializer, + TaskActivityMarkReadResponseSerializer, + TaskActivityMarkReadSerializer, + TaskActivityPageSerializer, + TaskActivityQuerySerializer, + TaskActivitySerializer, TaskMentionQuerySerializer, TaskMentionSerializer, TaskThreadMessageSerializer, @@ -199,6 +206,96 @@ def list(self, request, *args, **kwargs): return Response(TaskMentionSerializer(mentions, many=True).data) +class _ActivityPageEnvelopeSchema(AutoSchema): + """Stops drf-spectacular's list-view heuristic from wrapping the `list` response in an array. + + `list` returns a single page envelope (`results` + `unread_count`), not a bare collection. + Forcing the heuristic off renames the operation to `*_retrieve`, so pin the operationId back + to keep the generated client's `*List` name. + """ + + def _is_list_view(self, serializer: Any = None) -> bool: + return False + + def get_operation_id(self) -> str: + operation_id = super().get_operation_id() + if getattr(self.view, "action", None) == "list" and operation_id.endswith("_retrieve"): + return operation_id.removesuffix("_retrieve") + "_list" + return operation_id + + +class TaskActivityViewSet(TeamAndOrgViewSetMixin, viewsets.GenericViewSet): + """ + API for the requester's activity feed — one row per task they are involved in (created, + @-mentioned in, or authored a thread message on), most-recent activity first. + """ + + authentication_classes = [ + SessionAuthentication, + PersonalAPIKeyAuthentication, + OAuthAccessTokenAuthentication, + ] + permission_classes = [IsAuthenticated, APIScopePermission] + scope_object = "task" + http_method_names = ["get", "post", "head", "options"] + serializer_class = TaskActivitySerializer + # `list` hands back one envelope carrying its own unread total, so neither DRF's + # pagination wrapper nor spectacular's array wrapper describes what it sends. + pagination_class = None + schema = _ActivityPageEnvelopeSchema() + + def _user_id(self) -> int | None: + return getattr(self.request.user, "id", None) + + @validated_request( + query_serializer=TaskActivityQuerySerializer, + responses={ + 200: OpenApiResponse(response=TaskActivityPageSerializer, description="Tasks, most-recent activity first"), + }, + summary="List the requester's task activity", + description=( + "Tasks the requester is involved in (created, mentioned, or messaged), one row per task, " + "most-recent activity first, restricted to tasks they can see." + ), + ) + def list(self, request, *args, **kwargs): + activity = tasks_facade.list_task_activity( + self.team_id, + self._user_id(), + limit=request.validated_query_data["limit"], + before=request.validated_query_data.get("before"), + before_id=request.validated_query_data.get("before_id"), + ) + return Response(TaskActivityPageSerializer(activity).data) + + # @extend_schema must sit OUTSIDE @action: DRF's @action resets func.kwargs, wiping any schema + # annotation applied earlier — including @validated_request's — from the generated OpenAPI. + @extend_schema( + request=TaskActivityMarkReadSerializer, + responses={ + 200: OpenApiResponse(response=TaskActivityMarkReadResponseSerializer, description="Remaining unread total"), + }, + summary="Mark task activity read", + description=( + "Clear the unread flag on the requester's feed rows for the given tasks. Read state is per " + "task, so opening a task through any surface clears the same row." + ), + ) + @action(detail=False, methods=["post"], url_path="mark_read", required_scopes=["task:write"]) + @validated_request(request_serializer=TaskActivityMarkReadSerializer) + def mark_read(self, request, *args, **kwargs): + activities = [ + (activity["task_id"], activity["seen_before"]) for activity in request.validated_data["activities"] + ] + marked_read = tasks_facade.mark_task_activity_read(self.team_id, self._user_id(), activities) + return Response( + { + "marked_read": marked_read, + "unread_count": tasks_facade.count_unread_task_activity(self.team_id, self._user_id()), + } + ) + + class TaskThreadMessageViewSet(TeamAndOrgViewSetMixin, viewsets.GenericViewSet): """ API for a task's thread — the human-only side conversation around a task. Messages diff --git a/products/tasks/backend/push_dispatcher.py b/products/tasks/backend/push_dispatcher.py index 8bc88c33cc14..f8f4caa24e37 100644 --- a/products/tasks/backend/push_dispatcher.py +++ b/products/tasks/backend/push_dispatcher.py @@ -47,17 +47,19 @@ # they should only fire once per run lifetime — anything more is a retry. # Interactive turn-end can legitimately fire again after the user replies, # so a short cooldown is enough to absorb rapid duplicate triggers. -PushKind = Literal["completed", "failed", "cancelled", "awaiting"] +PushKind = Literal["completed", "failed", "cancelled", "awaiting", "turn_completed"] _COOLDOWN_SECONDS: dict[PushKind, int] = { "completed": 600, "failed": 600, "cancelled": 600, "awaiting": 30, + "turn_completed": 30, } def notify_task_run_completed(task_run: TaskRun) -> None: """Fire a push notification when ``task_run`` finishes successfully.""" + _project_completed_activity(task_run) _enqueue(task_run, kind="completed", body=f'"{_task_title(task_run)}" finished') @@ -73,9 +75,44 @@ def notify_task_run_cancelled(task_run: TaskRun) -> None: def notify_task_run_awaiting_input(task_run: TaskRun) -> None: """Fire a push notification when an interactive run is waiting for user input.""" + _project_awaiting_input_activity(task_run) _enqueue(task_run, kind="awaiting", body=f'"{_task_title(task_run)}" needs your input') +def notify_task_run_turn_completed(task_run: TaskRun) -> None: + _project_completed_activity(task_run) + _enqueue(task_run, kind="turn_completed", body=f'"{_task_title(task_run)}" finished') + + +def _project_awaiting_input_activity(task_run: TaskRun) -> None: + """Surface the wait in the in-app Activity feed. + + Runs ahead of, and independently of, the push guards above: the feed should update even + for users without the mobile push flag, and it has no cooldown to observe. Best-effort + for the same reason ``_enqueue`` is — this sits on the agent's turn-end path and must + never fail it. + """ + try: + from products.tasks.backend.facade.api import ( # noqa: PLC0415 - keeps the facade off the push import path + project_awaiting_input_activity, + ) + + project_awaiting_input_activity(task_run) + except Exception: + logger.warning("push_dispatcher.activity_projection_failed", run_id=str(task_run.id), exc_info=True) + + +def _project_completed_activity(task_run: TaskRun) -> None: + try: + from products.tasks.backend.facade.api import ( # noqa: PLC0415 - keeps the facade off the push import path + project_completed_activity, + ) + + project_completed_activity(task_run) + except Exception: + logger.warning("push_dispatcher.activity_projection_failed", run_id=str(task_run.id), exc_info=True) + + def _task_title(task_run: TaskRun) -> str: title = (task_run.task.title or "").strip() return title or "Untitled task" diff --git a/products/tasks/backend/routes.py b/products/tasks/backend/routes.py index 6bfbbccf691c..a9c10373ee8f 100644 --- a/products/tasks/backend/routes.py +++ b/products/tasks/backend/routes.py @@ -30,6 +30,7 @@ def register_routes(routers: RouterRegistry) -> None: ["team_id", "channel_id"], ) routers.projects.register(r"task_mentions", channels.TaskMentionViewSet, "project_task_mentions", ["team_id"]) + routers.projects.register(r"task_activity", channels.TaskActivityViewSet, "project_task_activity", ["team_id"]) routers.projects.register(r"task_automations", tasks.TaskAutomationViewSet, "project_task_automations", ["team_id"]) routers.projects.register(r"loops", loops.LoopViewSet, "project_loops", ["team_id"]) routers.projects.register( diff --git a/products/tasks/backend/temporal/process_task/activities/relay_sandbox_events.py b/products/tasks/backend/temporal/process_task/activities/relay_sandbox_events.py index 7f83fb51a35d..537f001c1f12 100644 --- a/products/tasks/backend/temporal/process_task/activities/relay_sandbox_events.py +++ b/products/tasks/backend/temporal/process_task/activities/relay_sandbox_events.py @@ -399,11 +399,10 @@ async def _relay_loop( if sandbox_id and background_logs_enabled: asyncio.create_task(_emit_agentsh_events(sandbox_id, run_id, last_audit_ts_ns)) if task_run is not None and task_run.mode == "interactive": - # Interactive run finished a turn — the agent is now idle waiting - # for the user. Hop off the event loop because the dispatcher - # does sync Redis (cache.add) and a potential network call to + # Hop off the event loop because the turn-completion dispatcher + # performs sync Redis I/O and a potential network call to # the feature-flag service. - asyncio.create_task(asyncio.to_thread(_safe_dispatch_awaiting_input, task_run)) + asyncio.create_task(asyncio.to_thread(_safe_dispatch_turn_completed, task_run)) if is_agent_design_enabled and slack_turn_active[0] and workflow_handle is not None: slack_turn_active[0] = False # Awaited in order: the final prose must be recorded before @@ -749,8 +748,8 @@ def _is_terminal_event(event_data: dict) -> bool: return method in TERMINAL_NOTIFICATION_METHODS -def _safe_dispatch_awaiting_input(task_run: TaskRunModel) -> None: - """Schedule a push when an interactive run idles waiting on the user. +def _safe_dispatch_turn_completed(task_run: TaskRunModel) -> None: + """Schedule a notification when an interactive run finishes a turn. Must be called via ``asyncio.to_thread`` (as the caller does) because the dispatcher performs sync I/O: a Redis write (``cache.add``) and a potential @@ -758,9 +757,9 @@ def _safe_dispatch_awaiting_input(task_run: TaskRunModel) -> None: dispatch never bubbles into the relay loop. """ try: - from products.tasks.backend.push_dispatcher import notify_task_run_awaiting_input + from products.tasks.backend.push_dispatcher import notify_task_run_turn_completed - notify_task_run_awaiting_input(task_run) + notify_task_run_turn_completed(task_run) except Exception: logger.warning( "relay_sandbox_events_push_dispatch_failed", diff --git a/products/tasks/backend/tests/test_agent_proxy_callback.py b/products/tasks/backend/tests/test_agent_proxy_callback.py index 66be24830178..fbf2be386173 100644 --- a/products/tasks/backend/tests/test_agent_proxy_callback.py +++ b/products/tasks/backend/tests/test_agent_proxy_callback.py @@ -140,7 +140,7 @@ def test_heartbeat_not_dispatched_when_inactive(self) -> None: def test_awaiting_input_dispatches_for_interactive_run(self) -> None: run = self.task.create_run(mode="interactive") - with patch("products.tasks.backend.agent_proxy_callback.notify_task_run_awaiting_input") as notify: + with patch("products.tasks.backend.agent_proxy_callback.notify_task_run_turn_completed") as notify: response = self._post( self._body(kind="awaiting_input", agent_active=False), token=self._token(run), @@ -151,7 +151,7 @@ def test_awaiting_input_dispatches_for_interactive_run(self) -> None: notify.assert_called_once() def test_awaiting_input_skipped_for_background_run(self) -> None: - with patch("products.tasks.backend.agent_proxy_callback.notify_task_run_awaiting_input") as notify: + with patch("products.tasks.backend.agent_proxy_callback.notify_task_run_turn_completed") as notify: response = self._post(self._body(kind="awaiting_input", agent_active=False), token=self._token()) self.assertEqual(response.status_code, 200) self.assertFalse(response.json()["dispatched"]) diff --git a/products/tasks/backend/tests/test_channels_api.py b/products/tasks/backend/tests/test_channels_api.py index 7df6b6c01316..28f5036f9c54 100644 --- a/products/tasks/backend/tests/test_channels_api.py +++ b/products/tasks/backend/tests/test_channels_api.py @@ -1,16 +1,23 @@ -from datetime import timedelta +from datetime import datetime, timedelta from unittest.mock import patch from django.test import TestCase from django.utils import timezone as django_timezone +from parameterized import parameterized from rest_framework import status from rest_framework.test import APIClient from posthog.models import Organization, OrganizationMembership, Team, User -from products.tasks.backend.models import Channel, ChannelFeedMessage, Task, TaskRun, TaskThreadMessage +from products.tasks.backend.facade import api as tasks_facade +from products.tasks.backend.models import Channel, ChannelFeedMessage, Task, TaskActivity, TaskRun, TaskThreadMessage +from products.tasks.backend.push_dispatcher import ( + notify_task_run_awaiting_input, + notify_task_run_completed, + notify_task_run_turn_completed, +) class ChannelsAPITestCase(TestCase): @@ -334,6 +341,273 @@ def test_mentions_are_team_scoped(self): self.assertEqual(len(other_team_mentions), 1) +class TaskActivityAPITestCase(ChannelTaskAPITestCase): + def _activity_url(self) -> str: + return f"/api/projects/{self.team.id}/task_activity/" + + def _thread_url(self, task=None) -> str: + return f"/api/projects/{self.team.id}/tasks/{(task or self.task).id}/thread_messages/" + + def _post_message(self, client, content: str, task=None) -> dict: + response = client.post(self._thread_url(task), {"content": content}) + self.assertEqual(response.status_code, status.HTTP_201_CREATED, response.content) + return response.json() + + def _mark_read(self, client, activities) -> dict: + response = client.post(self._activity_url() + "mark_read/", {"activities": activities}, format="json") + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content) + return response.json() + + def _rows(self, client) -> list[dict]: + return client.get(self._activity_url()).json()["results"] + + def _row_for(self, client, task) -> dict: + rows = [row for row in self._rows(client) if row["task_id"] == str(task.id)] + self.assertEqual(len(rows), 1, rows) + return rows[0] + + def _awaiting_input(self, task=None) -> None: + run = TaskRun.objects.create(team=self.team, task=task or self.task, status=TaskRun.Status.IN_PROGRESS) + # Go through the real notifier so the feed stays wired to whatever the product + # treats as "the agent is waiting", but leave the push side (flag, cooldown, + # Expo call) out of it. + with patch("products.tasks.backend.push_dispatcher._enqueue"): + notify_task_run_awaiting_input(run) + + def test_creator_sees_the_task_they_created(self): + row = self._row_for(self.author_client, self.task) + self.assertEqual(row["activity_kind"], "created") + self.assertEqual(row["snippet"], "") + self.assertEqual(row["channel_name"], "growth") + # A teammate with no relationship to the task sees nothing. + self.assertEqual(self._rows(self.peer_client), []) + + def test_authored_message_shows_as_message_with_snippet(self): + self._post_message(self.peer_client, "looking into this") + row = self._row_for(self.peer_client, self.task) + self.assertEqual(row["activity_kind"], "message") + self.assertEqual(row["snippet"], "looking into this") + self.assertEqual(row["latest_author"]["id"], self.peer.id) + + def test_agent_message_is_unread_for_the_task_creator(self): + tasks_facade._create_agent_thread_message(self.task, "Hello!", event="agent_message") + + row = self._row_for(self.author_client, self.task) + self.assertEqual(row["activity_kind"], "message") + self.assertEqual(row["snippet"], "Hello!") + self.assertIsNone(row["latest_author"]) + self.assertTrue(row["is_unread"]) + + def test_mention_shows_as_mention_with_snippet(self): + self._post_message(self.author_client, "cc @[Bob](peer@example.com) please look") + row = self._row_for(self.peer_client, self.task) + self.assertEqual(row["activity_kind"], "mention") + self.assertEqual(row["snippet"], "cc @[Bob](peer@example.com) please look") + self.assertEqual(row["latest_author"]["id"], self.author.id) + + def test_awaiting_input_projects_from_the_run_awaiting_notification(self): + self._awaiting_input() + row = self._row_for(self.author_client, self.task) + self.assertEqual(row["activity_kind"], "awaiting_input") + self.assertTrue(row["is_unread"]) + # Only the task's creator is being waited on. + self.assertEqual(self._rows(self.peer_client), []) + + def test_completed_run_replaces_awaiting_input_activity(self): + run = TaskRun.objects.create(team=self.team, task=self.task, status=TaskRun.Status.IN_PROGRESS) + with patch("products.tasks.backend.push_dispatcher._enqueue"): + notify_task_run_awaiting_input(run) + notify_task_run_completed(run) + + row = self._row_for(self.author_client, self.task) + self.assertEqual(row["activity_kind"], "completed") + self.assertTrue(row["is_unread"]) + + def test_completed_turn_is_unread_for_the_task_creator(self): + run = TaskRun.objects.create( + team=self.team, + task=self.task, + state={"mode": "interactive"}, + status=TaskRun.Status.IN_PROGRESS, + ) + with patch("products.tasks.backend.push_dispatcher._enqueue"): + notify_task_run_turn_completed(run) + + row = self._row_for(self.author_client, self.task) + self.assertEqual(row["activity_kind"], "completed") + self.assertTrue(row["is_unread"]) + + def test_multiple_signals_collapse_to_one_row_with_the_newest_winning(self): + self._post_message(self.author_client, "cc @[Bob](peer@example.com)") + self._post_message(self.peer_client, "on it") + row = self._row_for(self.peer_client, self.task) + self.assertEqual(row["activity_kind"], "message") + self.assertEqual(row["snippet"], "on it") + + def test_out_of_order_projection_does_not_move_the_row_backwards(self): + self._awaiting_input() + latest = self._row_for(self.author_client, self.task)["activity_at"] + # A retried write replaying an older event must not overwrite newer activity. + TaskActivity.record( + team_id=self.team.id, + user_id=self.author.id, + task_id=self.task.id, + kind=TaskActivity.Kind.CREATED, + activity_at=django_timezone.now() - timedelta(hours=1), + ) + row = self._row_for(self.author_client, self.task) + self.assertEqual(row["activity_kind"], "awaiting_input") + self.assertEqual(row["activity_at"], latest) + + @parameterized.expand( + [ + ("own_task_creation", None), + ("own_reply", "just thinking out loud"), + ] + ) + def test_activity_the_user_caused_themselves_is_never_unread(self, _name, own_message): + if own_message is not None: + self._post_message(self.author_client, own_message) + page = self.author_client.get(self._activity_url()).json() + self.assertEqual(page["unread_count"], 0) + self.assertFalse(page["results"][0]["is_unread"]) + + @parameterized.expand( + [ + ("mention", lambda self: self._post_message(self.author_client, "@[Bob](peer@example.com) ping")), + ("awaiting_input", lambda self: self._awaiting_input()), + ] + ) + def test_activity_someone_else_caused_is_unread(self, name, trigger): + trigger(self) + client = self.peer_client if name == "mention" else self.author_client + page = client.get(self._activity_url()).json() + self.assertEqual(page["unread_count"], 1) + self.assertTrue(self._row_for(client, self.task)["is_unread"]) + + def test_mark_read_clears_only_the_named_tasks(self): + second = Task.objects.create( + team=self.team, + created_by=self.author, + channel=self.channel, + title="Second", + description="d", + origin_product=Task.OriginProduct.USER_CREATED, + ) + self._awaiting_input() + self._awaiting_input(second) + self.assertEqual(self.author_client.get(self._activity_url()).json()["unread_count"], 2) + + row = self._row_for(self.author_client, self.task) + body = self._mark_read( + self.author_client, + [{"task_id": str(self.task.id), "seen_before": row["activity_at"]}], + ) + self.assertEqual(body, {"marked_read": 1, "unread_count": 1}) + self.assertFalse(self._row_for(self.author_client, self.task)["is_unread"]) + self.assertTrue(self._row_for(self.author_client, second)["is_unread"]) + + def test_reading_the_thread_does_not_mutate_activity(self): + self._awaiting_input() + self.assertTrue(self._row_for(self.author_client, self.task)["is_unread"]) + self.assertEqual(self.author_client.get(self._thread_url()).status_code, status.HTTP_200_OK) + self.assertTrue(self._row_for(self.author_client, self.task)["is_unread"]) + + def test_mark_read_does_not_clear_newer_activity(self): + self._awaiting_input() + listed = self._row_for(self.author_client, self.task) + TaskActivity.record( + team_id=self.team.id, + user_id=self.author.id, + task_id=self.task.id, + kind=TaskActivity.Kind.MENTION, + activity_at=django_timezone.now() + timedelta(seconds=1), + ) + + body = self._mark_read( + self.author_client, + [{"task_id": str(self.task.id), "seen_before": listed["activity_at"]}], + ) + + self.assertEqual(body["marked_read"], 0) + self.assertTrue(self._row_for(self.author_client, self.task)["is_unread"]) + + def test_replaying_the_same_activity_preserves_read_state(self): + self._awaiting_input() + listed = self._row_for(self.author_client, self.task) + self._mark_read( + self.author_client, + [{"task_id": str(self.task.id), "seen_before": listed["activity_at"]}], + ) + + TaskActivity.record( + team_id=self.team.id, + user_id=self.author.id, + task_id=self.task.id, + kind=TaskActivity.Kind.AWAITING_INPUT, + activity_at=datetime.fromisoformat(listed["activity_at"]), + ) + + self.assertFalse(self._row_for(self.author_client, self.task)["is_unread"]) + + def test_unread_count_covers_the_whole_feed_not_just_the_page(self): + for index in range(2): + task = Task.objects.create( + team=self.team, + created_by=self.author, + channel=self.channel, + title=f"Extra {index}", + description="d", + origin_product=Task.OriginProduct.USER_CREATED, + ) + self._awaiting_input(task) + self._awaiting_input() + page = self.author_client.get(self._activity_url(), {"limit": 1}).json() + self.assertEqual(len(page["results"]), 1) + self.assertEqual(page["unread_count"], 3) + + def test_newest_activity_first_and_limit_applies(self): + second = Task.objects.create( + team=self.team, + created_by=self.author, + channel=self.channel, + title="Second", + description="d", + origin_product=Task.OriginProduct.USER_CREATED, + ) + # A fresh message makes `second` the most recently active task. + self._post_message(self.author_client, "kickoff", task=second) + rows = self._rows(self.author_client) + self.assertEqual([row["task_id"] for row in rows], [str(second.id), str(self.task.id)]) + first_page = self.author_client.get(self._activity_url(), {"limit": 1}).json() + self.assertEqual([row["task_id"] for row in first_page["results"]], [str(second.id)]) + second_page = self.author_client.get( + self._activity_url(), + { + "limit": 1, + "before": first_page["next_before"], + "before_id": first_page["next_before_id"], + }, + ).json() + self.assertEqual([row["task_id"] for row in second_page["results"]], [str(self.task.id)]) + self.assertIsNone(second_page["next_before"]) + self.assertIsNone(second_page["next_before_id"]) + + def test_mark_read_rejects_an_empty_task_list(self): + response = self.author_client.post(self._activity_url() + "mark_read/", {"activities": []}, format="json") + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST, response.content) + + def test_activity_projection_failure_does_not_fail_message_creation(self): + with patch.object(TaskActivity, "record", side_effect=RuntimeError("projection unavailable")): + response = self.author_client.post(self._thread_url(), {"content": "still persisted"}) + + self.assertEqual(response.status_code, status.HTTP_201_CREATED, response.content) + self.assertEqual( + TaskThreadMessage.objects.for_team(self.team.id).filter(task=self.task, content="still persisted").count(), + 1, + ) + + class ChannelFeedMessageAPITestCase(TestCase): def setUp(self) -> None: self.organization = Organization.objects.create(name="Feed Org") diff --git a/products/tasks/backend/tests/test_event_ingest.py b/products/tasks/backend/tests/test_event_ingest.py index 15c92daa2263..5e18cf77916a 100644 --- a/products/tasks/backend/tests/test_event_ingest.py +++ b/products/tasks/backend/tests/test_event_ingest.py @@ -310,26 +310,17 @@ def test_current_project_path_ingests_with_token_scoped_task_run(self) -> None: self.assertEqual(body["accepted"], 1) self.assertEqual(self._read_notification_methods(), ["session/update"]) - @parameterized.expand([(True,), (False,)]) @override_settings(SANDBOX_JWT_PRIVATE_KEY=TEST_RSA_PRIVATE_KEY) - def test_turn_complete_ingest_notifies_interactive_run_awaiting_input_only_with_flag( - self, flag_enabled: bool - ) -> None: + def test_turn_complete_ingest_notifies_interactive_run(self) -> None: self.task.created_by = User.objects.create_user("ingest-push@posthog.com", None, "Ingest") self.task.save(update_fields=["created_by"]) self.task_run.state = {"mode": "interactive"} self.task_run.save(update_fields=["state"]) token = self._create_token() - with ( - patch( - "products.tasks.backend.logic.stream.event_ingest.notify_task_run_awaiting_input" - ) as notify_awaiting_input, - patch( - "products.tasks.backend.logic.stream.event_ingest.posthoganalytics.feature_enabled", - return_value=flag_enabled, - ), - ): + with patch( + "products.tasks.backend.logic.stream.event_ingest.notify_task_run_turn_completed" + ) as notify_turn_completed: status, body = self._call_ingest( token, [ @@ -345,11 +336,8 @@ def test_turn_complete_ingest_notifies_interactive_run_awaiting_input_only_with_ self.assertEqual(status, 200) self.assertEqual(body["accepted"], 1) - if flag_enabled: - notify_awaiting_input.assert_called_once() - self.assertEqual(notify_awaiting_input.call_args.args[0].id, self.task_run.id) - else: - notify_awaiting_input.assert_not_called() + notify_turn_completed.assert_called_once() + self.assertEqual(notify_turn_completed.call_args.args[0].id, self.task_run.id) @override_settings(SANDBOX_JWT_PRIVATE_KEY=TEST_RSA_PRIVATE_KEY) def test_workflow_heartbeat_does_not_block_event_loop(self) -> None: diff --git a/products/tasks/backend/tests/test_push_dispatcher.py b/products/tasks/backend/tests/test_push_dispatcher.py index eb1ea938570d..ee7c7cb889ea 100644 --- a/products/tasks/backend/tests/test_push_dispatcher.py +++ b/products/tasks/backend/tests/test_push_dispatcher.py @@ -15,6 +15,7 @@ notify_task_run_cancelled, notify_task_run_completed, notify_task_run_failed, + notify_task_run_turn_completed, ) @@ -46,6 +47,7 @@ def setUp(self) -> None: ("failed", notify_task_run_failed, "failed"), ("cancelled", notify_task_run_cancelled, "cancelled"), ("awaiting", notify_task_run_awaiting_input, "needs your input"), + ("turn_completed", notify_task_run_turn_completed, "finished"), ] ) @patch("products.tasks.backend.push_dispatcher.posthoganalytics.feature_enabled", return_value=True) @@ -96,7 +98,8 @@ def test_cooldown_is_per_kind(self, mock_delay, _flag): with self.captureOnCommitCallbacks(execute=True): notify_task_run_completed(self.task_run) notify_task_run_awaiting_input(self.task_run) - self.assertEqual(mock_delay.call_count, 2) + notify_task_run_turn_completed(self.task_run) + self.assertEqual(mock_delay.call_count, 3) @patch("products.tasks.backend.push_dispatcher.posthoganalytics.feature_enabled", return_value=True) @patch("products.tasks.backend.push_dispatcher.send_user_push.delay") diff --git a/products/tasks/frontend/generated/api.schemas.ts b/products/tasks/frontend/generated/api.schemas.ts index 4e529073ac9f..ff01a2d6950b 100644 --- a/products/tasks/frontend/generated/api.schemas.ts +++ b/products/tasks/frontend/generated/api.schemas.ts @@ -905,6 +905,98 @@ export interface PatchedSandboxEnvironmentWriteApi { custom_image_id?: string | null } +/** + * * `awaiting_input` - awaiting_input + * * `completed` - completed + * * `mention` - mention + * * `message` - message + * * `created` - created + */ +export type ActivityKindEnumApi = (typeof ActivityKindEnumApi)[keyof typeof ActivityKindEnumApi] + +export const ActivityKindEnumApi = { + AwaitingInput: 'awaiting_input', + Completed: 'completed', + Mention: 'mention', + Message: 'message', + Created: 'created', +} as const + +/** + * Response shape for one task in the requester's activity feed (one row per task). + */ +export interface TaskActivityDTOApi { + id: string + task_id: string + task_title: string + /** @nullable */ + channel_id: string | null + /** @nullable */ + channel_name: string | null + activity_at: string + /** What the latest activity on this task was: an agent run waiting on the requester (awaiting_input), a completed run (completed), someone @-mentioning them (mention), a thread reply (message), or their creating the task (created). + * + * * `awaiting_input` - awaiting_input + * * `completed` - completed + * * `mention` - mention + * * `message` - message + * * `created` - created */ + activity_kind: ActivityKindEnumApi + /** Content of the thread message tied to the latest activity; empty for task-creation rows. */ + snippet: string + /** Author of the thread message tied to the latest activity, when one applies. */ + latest_author?: TaskUserBasicInfoApi | null + /** @nullable */ + latest_message_id?: string | null + /** Whether the requester has yet to see this activity. Activity they caused themselves is never unread. */ + is_unread: boolean +} + +/** + * A page of the requester's activity feed, plus the unread total across the whole feed. + */ +export interface TaskActivityPageDTOApi { + /** Tasks with activity, most recent first. */ + results: TaskActivityDTOApi[] + /** Unread tasks across the requester's whole feed, not just this page. Backs the sidebar badge. */ + unread_count: number + /** + * Activity timestamp to pass as before for the next page, or null on the final page. + * @nullable + */ + next_before?: string | null + /** + * Activity ID to pass as before_id for the next page, or null on the final page. + * @nullable + */ + next_before_id?: string | null +} + +export interface TaskActivityReadMarkerApi { + /** Task whose displayed activity should be marked read. */ + task_id: string + /** Mark activity at or before this timestamp read without clearing newer activity. */ + seen_before: string +} + +/** + * Request body for clearing the unread flag on specific tasks. + */ +export interface TaskActivityMarkReadApi { + /** + * Displayed task activities to mark read if they have not changed. + * @maxItems 500 + */ + activities: TaskActivityReadMarkerApi[] +} + +export interface TaskActivityMarkReadResponseApi { + /** How many feed rows changed from unread to read. */ + marked_read: number + /** The requester's remaining unread total after the update. */ + unread_count: number +} + /** * Detail/create/update/run response for a task automation. */ @@ -3543,6 +3635,23 @@ export type SandboxListParams = { offset?: number } +export type TaskActivityListParams = { + /** + * Activity timestamp from the final row of the previous page. + */ + before?: string + /** + * Activity ID from the final row of the previous page. + */ + before_id?: string + /** + * Maximum number of tasks to return (most recent activity first). + * @minimum 1 + * @maximum 500 + */ + limit?: number +} + export type TaskAutomationsListParams = { /** * Number of results to return per page. diff --git a/products/tasks/frontend/generated/api.ts b/products/tasks/frontend/generated/api.ts index 700f657f4f23..e54734405cf8 100644 --- a/products/tasks/frontend/generated/api.ts +++ b/products/tasks/frontend/generated/api.ts @@ -56,6 +56,10 @@ import type { SandboxListParams, SlackThreadContextResponseApi, StreamReadTokenResponseApi, + TaskActivityListParams, + TaskActivityMarkReadApi, + TaskActivityMarkReadResponseApi, + TaskActivityPageDTOApi, TaskAutomationDTOApi, TaskAutomationWriteApi, TaskAutomationsListParams, @@ -625,6 +629,58 @@ export const sandboxDestroy = async (projectId: string, id: string, options?: Re }) } +export const getTaskActivityListUrl = (projectId: string, params?: TaskActivityListParams) => { + const normalizedParams = new URLSearchParams() + + Object.entries(params || {}).forEach(([key, value]) => { + if (value !== undefined) { + normalizedParams.append(key, value === null ? 'null' : String(value)) + } + }) + + const stringifiedParams = normalizedParams.toString() + + return stringifiedParams.length > 0 + ? `/api/projects/${projectId}/task_activity/?${stringifiedParams}` + : `/api/projects/${projectId}/task_activity/` +} + +/** + * Tasks the requester is involved in (created, mentioned, or messaged), one row per task, most-recent activity first, restricted to tasks they can see. + * @summary List the requester's task activity + */ +export const taskActivityList = async ( + projectId: string, + params?: TaskActivityListParams, + options?: RequestInit +): Promise => { + return apiMutator(getTaskActivityListUrl(projectId, params), { + ...options, + method: 'GET', + }) +} + +export const getTaskActivityMarkReadCreateUrl = (projectId: string) => { + return `/api/projects/${projectId}/task_activity/mark_read/` +} + +/** + * Clear the unread flag on the requester's feed rows for the given tasks. Read state is per task, so opening a task through any surface clears the same row. + * @summary Mark task activity read + */ +export const taskActivityMarkReadCreate = async ( + projectId: string, + taskActivityMarkReadApi: TaskActivityMarkReadApi, + options?: RequestInit +): Promise => { + return apiMutator(getTaskActivityMarkReadCreateUrl(projectId), { + ...options, + method: 'POST', + headers: { 'Content-Type': 'application/json', ...options?.headers }, + body: JSON.stringify(taskActivityMarkReadApi), + }) +} + export const getTaskAutomationsListUrl = (projectId: string, params?: TaskAutomationsListParams) => { const normalizedParams = new URLSearchParams() diff --git a/products/tasks/frontend/generated/api.zod.ts b/products/tasks/frontend/generated/api.zod.ts index 74fc911c81df..b53da5877760 100644 --- a/products/tasks/frontend/generated/api.zod.ts +++ b/products/tasks/frontend/generated/api.zod.ts @@ -868,6 +868,28 @@ export const SandboxPartialUpdateBody = /* @__PURE__ */ zod }) .describe('Request body for creating or updating a sandbox environment.') +/** + * Clear the unread flag on the requester's feed rows for the given tasks. Read state is per task, so opening a task through any surface clears the same row. + * @summary Mark task activity read + */ +export const taskActivityMarkReadCreateBodyActivitiesMax = 500 + +export const TaskActivityMarkReadCreateBody = /* @__PURE__ */ zod + .object({ + activities: zod + .array( + zod.object({ + task_id: zod.uuid().describe('Task whose displayed activity should be marked read.'), + seen_before: zod.iso + .datetime({ offset: true }) + .describe('Mark activity at or before this timestamp read without clearing newer activity.'), + }) + ) + .max(taskActivityMarkReadCreateBodyActivitiesMax) + .describe('Displayed task activities to mark read if they have not changed.'), + }) + .describe('Request body for clearing the unread flag on specific tasks.') + /** * API for managing scheduled task automations. */ diff --git a/products/tasks/mcp/tools.yaml b/products/tasks/mcp/tools.yaml index 31ac82fe78a8..b548c9ee6351 100644 --- a/products/tasks/mcp/tools.yaml +++ b/products/tasks/mcp/tools.yaml @@ -190,6 +190,12 @@ tools: sandbox-retrieve: operation: sandbox_retrieve enabled: false + task-activity-list: + operation: task_activity_list + enabled: false + task-activity-mark-read-create: + operation: task_activity_mark_read_create + enabled: false task-automations-create: operation: task_automations_create enabled: false diff --git a/services/mcp/src/api/generated.ts b/services/mcp/src/api/generated.ts index fb8a0758a8bd..d1e53224e3f8 100644 --- a/services/mcp/src/api/generated.ts +++ b/services/mcp/src/api/generated.ts @@ -2196,6 +2196,24 @@ export namespace Schemas { config?: ActivityEventsListWidgetConfig; } + /** + * * `awaiting_input` - awaiting_input + * * `completed` - completed + * * `mention` - mention + * * `message` - message + * * `created` - created + */ + export type ActivityKindEnum = typeof ActivityKindEnum[keyof typeof ActivityKindEnum]; + + + export const ActivityKindEnum = { + AwaitingInput: 'awaiting_input', + Completed: 'completed', + Mention: 'mention', + Message: 'message', + Created: 'created', + } as const; + export interface ActivityLog { readonly id: string; user: UserBasic; @@ -67041,6 +67059,81 @@ export namespace Schemas { deleted?: boolean; } + /** + * Response shape for one task in the requester's activity feed (one row per task). + */ + export interface TaskActivityDTO { + id: string; + task_id: string; + task_title: string; + /** @nullable */ + channel_id: string | null; + /** @nullable */ + channel_name: string | null; + activity_at: string; + /** What the latest activity on this task was: an agent run waiting on the requester (awaiting_input), a completed run (completed), someone @-mentioning them (mention), a thread reply (message), or their creating the task (created). + * + * * `awaiting_input` - awaiting_input + * * `completed` - completed + * * `mention` - mention + * * `message` - message + * * `created` - created */ + activity_kind: ActivityKindEnum; + /** Content of the thread message tied to the latest activity; empty for task-creation rows. */ + snippet: string; + /** Author of the thread message tied to the latest activity, when one applies. */ + latest_author?: TaskUserBasicInfo | null; + /** @nullable */ + latest_message_id?: string | null; + /** Whether the requester has yet to see this activity. Activity they caused themselves is never unread. */ + is_unread: boolean; + } + + export interface TaskActivityReadMarker { + /** Task whose displayed activity should be marked read. */ + task_id: string; + /** Mark activity at or before this timestamp read without clearing newer activity. */ + seen_before: string; + } + + /** + * Request body for clearing the unread flag on specific tasks. + */ + export interface TaskActivityMarkRead { + /** + * Displayed task activities to mark read if they have not changed. + * @maxItems 500 + */ + activities: TaskActivityReadMarker[]; + } + + export interface TaskActivityMarkReadResponse { + /** How many feed rows changed from unread to read. */ + marked_read: number; + /** The requester's remaining unread total after the update. */ + unread_count: number; + } + + /** + * A page of the requester's activity feed, plus the unread total across the whole feed. + */ + export interface TaskActivityPageDTO { + /** Tasks with activity, most recent first. */ + results: TaskActivityDTO[]; + /** Unread tasks across the requester's whole feed, not just this page. Backs the sidebar badge. */ + unread_count: number; + /** + * Activity timestamp to pass as before for the next page, or null on the final page. + * @nullable + */ + next_before?: string | null; + /** + * Activity ID to pass as before_id for the next page, or null on the final page. + * @nullable + */ + next_before_id?: string | null; + } + /** * * `active` - active * * `failed` - failed @@ -79865,6 +79958,23 @@ export namespace Schemas { offset?: number; }; + export type TaskActivityListParams = { + /** + * Activity timestamp from the final row of the previous page. + */ + before?: string; + /** + * Activity ID from the final row of the previous page. + */ + before_id?: string; + /** + * Maximum number of tasks to return (most recent activity first). + * @minimum 1 + * @maximum 500 + */ + limit?: number; + }; + export type TaskAutomationsListParams = { /** * Number of results to return per page.