feat(tasks): channel feed messages — durable, multiplayer system announcements

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

@@ -265,6 +265,7 @@ rules:
                               |TaskThreadMessage
                               |TaskThreadMessageMention
                               |Channel
+                              |ChannelFeedMessage
                               |EmailChannel
                               |EvaluationReport
                               |Text
@@ -554,6 +555,7 @@ rules:
                         |TaskThreadMessage
                         |TaskThreadMessageMention
                         |Channel
+                        |ChannelFeedMessage
                         |EmailChannel
                         |EvaluationReport
                         |Text

posthog/api/file_system/file_system.py

@@ -3,8 +3,9 @@
 import shlex
 import builtins
 from typing import Any, cast
-from uuid import uuid4
+from uuid import UUID, uuid4
 
+from django.conf import settings
 from django.db import transaction
 from django.db.models import Case, F, IntegerField, Q, QuerySet, Value, When
 from django.db.models.functions import Concat, Lower
@@ -47,6 +48,7 @@
 from posthog.api.routing import TeamAndOrgViewSetMixin
 from posthog.api.shared import UserBasicSerializer
 from posthog.api.utils import action
+from posthog.auth import OAuthAccessTokenAuthentication
 from posthog.decorators import disallow_if_impersonated
 from posthog.models.file_system.file_system import (
     DEFAULT_SURFACE,
@@ -61,8 +63,11 @@
 from posthog.models.file_system.unfiled_file_saver import save_unfiled_files
 from posthog.models.team import Team
 from posthog.models.user import User
+from posthog.temporal.oauth import SANDBOX_OAUTH_APP_CLIENT_IDS
 from posthog.utils import str_to_bool
 
+from products.tasks.backend.facade import api as tasks_facade
+
 DELETE_PREVIEW_ENTRY_LIMIT = 200
 
 # Search-within-Recents scans this many of the user's most-recent views, then the text filter trims
@@ -1132,6 +1137,7 @@ def publish_canvas(self, request: Request, *args: Any, **kwargs: Any) -> Respons
             if isinstance(existing_context, str):
                 version["context"] = existing_context
             versions = list(meta.get("versions") or [])
+            first_publish = not versions and not meta.get("code")
             versions.append(version)
 
             meta.update(
@@ -1156,8 +1162,73 @@ def publish_canvas(self, request: Request, *args: Any, **kwargs: Any) -> Respons
 
             dashboard.save(update_fields=update_fields)
 
+        if first_publish:
+            self._announce_canvas_created(request, dashboard)
+
         return Response(self.get_serializer(dashboard).data)
 
+    def _announce_canvas_created(self, request: Request, dashboard: FileSystem) -> None:
+        """Announce a canvas's first publish in the generating task's thread.
+
+        The task sandbox stamps every MCP call with an X-PostHog-Task-Id header, so
+        a publish is attributable to the task that made it. The header alone is
+        forgeable, so two checks bind the announcement to a real sandbox run: the
+        request must carry an OAuth token minted under a sandbox app (those tokens
+        are only created server-side), and the facade only accepts a task created
+        by the requesting user (the sandbox authenticates with the task creator's
+        credentials). No header (a human or app save) means no announcement.
+        """
+        raw_task_id = (request.headers.get("X-PostHog-Task-Id") or "").strip()
+        try:
+            task_id = UUID(raw_task_id)
+        except ValueError:
+            return
+        if not self._is_sandbox_authenticated(request):
+            return
+        user = request.user if isinstance(request.user, User) else None
+        segments = split_path(dashboard.path)
+        tasks_facade.post_canvas_created_thread_update(
+            task_id,
+            self.team_id,
+            acting_user_id=user.id if user else None,
+            canvas_name=segments[-1] if segments else "Canvas",
+            canvas_url=self._canvas_share_url(dashboard),
+        )
+
+    @staticmethod
+    def _is_sandbox_authenticated(request: Request) -> bool:
+        """True when the request bears an OAuth token minted under a sandbox app —
+        the credential a task sandbox (via the MCP server) calls this API with."""
+        authenticator = request.successful_authenticator
+        if not isinstance(authenticator, OAuthAccessTokenAuthentication):
+            return False
+        application = authenticator.access_token.application
+        return application is not None and application.client_id in SANDBOX_OAUTH_APP_CLIENT_IDS
+
+    def _canvas_share_url(self, dashboard: FileSystem) -> str | None:
+        """The web interstitial link that deep-links into the desktop app's canvas view:
+        `/code/canvas/<channel folder id>/<dashboard id>`. The channel id is stamped on
+        the row's meta by the desktop app at create time; fall back to the parent folder
+        row for rows that predate the stamp.
+        """
+        channel_id = (dashboard.meta or {}).get("channelId")
+        if not channel_id:
+            parent_path = join_path(split_path(dashboard.path)[:-1])
+            folder = (
+                FileSystem.objects.filter(
+                    surface_q(self.file_system_surface),
+                    team_id=dashboard.team_id,
+                    type="folder",
+                    path=parent_path,
+                ).first()
+                if parent_path
+                else None
+            )
+            channel_id = str(folder.id) if folder else None
+        if not channel_id:
+            return None
+        return f"{settings.SITE_URL}/code/canvas/{channel_id}/{dashboard.id}"
+
     @extend_schema(responses={200: FolderInstructionsSerializer})
     @action(methods=["GET"], detail=True)
     def instructions(self, request: Request, *args: Any, **kwargs: Any) -> Response:

posthog/api/file_system/test/test_canvas_publish.py

@@ -1,10 +1,25 @@
-from typing import cast
+from typing import TYPE_CHECKING, cast
 
 from posthog.test.base import APIBaseTest
+from unittest.mock import patch
+
+from django.apps import apps
+from django.conf import settings
 
 from rest_framework import status
 
 from posthog.models.file_system.file_system import FileSystem
+from posthog.models.oauth import OAuthApplication
+from posthog.models.user import User
+from posthog.temporal.oauth import (
+    ARRAY_APP_CLIENT_ID_DEV,
+    ARRAY_APP_CLIENT_ID_EU,
+    ARRAY_APP_CLIENT_ID_US,
+    create_oauth_access_token_for_user,
+)
+
+if TYPE_CHECKING:
+    from products.tasks.backend.models import Task
 
 
 class TestDesktopCanvasPublishAPI(APIBaseTest):
@@ -96,6 +111,116 @@ def test_publish_canvas_requires_code(self):
         self.assertEqual(bad.status_code, status.HTTP_400_BAD_REQUEST, bad.json())
         self.assertIn("code", bad.json())
 
+    # Task models load via the app registry: this test lives outside the isolated
+    # tasks product, so it can't import its internals (tach-enforced).
+    def _create_task(self) -> "Task":
+        Task = apps.get_model("tasks", "Task")
+        return Task.objects.create(
+            team=self.team,
+            title="Generate canvas",
+            description="",
+            origin_product=Task.OriginProduct.USER_CREATED,
+            created_by=self.user,
+        )
+
+    def _thread_messages(self, task: "Task"):
+        TaskThreadMessage = apps.get_model("tasks", "TaskThreadMessage")
+        return TaskThreadMessage.objects.for_team(self.team.id).filter(task=task)
+
+    def _authenticate_as_sandbox(self) -> None:
+        """Swap session auth for a sandbox-app OAuth token — announcements only fire for
+        requests bearing one. The app is created for every region client id because
+        `create_oauth_access_token_for_user` resolves it by `get_instance_region()`."""
+        for client_id in (ARRAY_APP_CLIENT_ID_DEV, ARRAY_APP_CLIENT_ID_US, ARRAY_APP_CLIENT_ID_EU):
+            OAuthApplication.objects.get_or_create(
+                client_id=client_id,
+                defaults={
+                    "name": "Array Test App",
+                    "client_type": OAuthApplication.CLIENT_PUBLIC,
+                    "authorization_grant_type": OAuthApplication.GRANT_AUTHORIZATION_CODE,
+                    "redirect_uris": "https://app.posthog.com/callback",
+                    # RS256 is enforced by the `enforce_rs256_algorithm` DB constraint.
+                    "algorithm": "RS256",
+                },
+            )
+        token = create_oauth_access_token_for_user(self.user, self.team.id, scopes="full")
+        self.client.logout()
+        self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
+
+    @patch("products.tasks.backend.facade.api.posthoganalytics.feature_enabled", return_value=True)
+    def test_first_publish_from_task_announces_in_thread_once(self, _flag):
+        task = self._create_task()
+        item_id = self._create_dashboard(meta={"channelId": "chan-1"})
+        self._authenticate_as_sandbox()
+
+        self.client.patch(self._canvas_url(item_id), {"code": "v1"}, HTTP_X_POSTHOG_TASK_ID=str(task.id))
+
+        messages = self._thread_messages(task)
+        self.assertEqual(messages.count(), 1)
+        message = messages.get()
+        self.assertIsNone(message.author_id)
+        self.assertEqual(
+            message.content,
+            f"[MyCanvas]({settings.SITE_URL}/code/canvas/chan-1/{item_id}) has been created",
+        )
+
+        # A second publish updates the canvas, it doesn't create it again.
+        self.client.patch(self._canvas_url(item_id), {"code": "v2"}, HTTP_X_POSTHOG_TASK_ID=str(task.id))
+        self.assertEqual(messages.count(), 1)
+
+    @patch("products.tasks.backend.facade.api.posthoganalytics.feature_enabled", return_value=True)
+    def test_announcement_links_via_parent_folder_when_meta_has_no_channel(self, _flag):
+        task = self._create_task()
+        item_id = self._create_dashboard()  # no channelId stamp — rows created before the app stamped it
+        self._authenticate_as_sandbox()
+
+        self.client.patch(self._canvas_url(item_id), {"code": "v1"}, HTTP_X_POSTHOG_TASK_ID=str(task.id))
+
+        folder = FileSystem.objects.get(team=self.team, path="MyChannel", type="folder")
+        message = self._thread_messages(task).get()
+        self.assertTrue(message.content.startswith(f"[MyCanvas]({settings.SITE_URL}/code/canvas/{folder.id}/"))
+
+    @patch("products.tasks.backend.facade.api.posthoganalytics.feature_enabled", return_value=True)
+    def test_header_naming_someone_elses_task_stays_silent(self, _flag):
+        # The header selects the announcement's thread; it must not let a publisher
+        # plant agent messages in a task they didn't create.
+        other = User.objects.create_and_join(self.organization, "other@posthog.com", None)
+        Task = apps.get_model("tasks", "Task")
+        task = Task.objects.create(
+            team=self.team,
+            title="Someone else's task",
+            description="",
+            origin_product=Task.OriginProduct.USER_CREATED,
+            created_by=other,
+        )
+        item_id = self._create_dashboard()
+        self._authenticate_as_sandbox()
+
+        self.client.patch(self._canvas_url(item_id), {"code": "v1"}, HTTP_X_POSTHOG_TASK_ID=str(task.id))
+
+        self.assertFalse(self._thread_messages(task).exists())
+
+    @patch("products.tasks.backend.facade.api.posthoganalytics.feature_enabled", return_value=True)
+    def test_session_authenticated_publish_with_header_stays_silent(self, _flag):
+        # The header alone must not produce an agent announcement: a member setting it on
+        # an ordinary (session-authenticated) publish of their own task would otherwise
+        # forge a trusted-looking agent message. Only sandbox OAuth tokens qualify.
+        task = self._create_task()
+        item_id = self._create_dashboard()
+
+        response = self.client.patch(self._canvas_url(item_id), {"code": "v1"}, HTTP_X_POSTHOG_TASK_ID=str(task.id))
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertFalse(self._thread_messages(task).exists())
+
+    def test_publish_without_task_attribution_stays_silent(self):
+        item_id = self._create_dashboard()
+
+        self.client.patch(self._canvas_url(item_id), {"code": "v1"})
+
+        TaskThreadMessage = apps.get_model("tasks", "TaskThreadMessage")
+        self.assertFalse(TaskThreadMessage.objects.for_team(self.team.id).exists())
+
     def test_delete_canvas_removes_ref_less_dashboard_row(self):
         # Desktop canvases are `dashboard`-typed rows with no ref; deleting one must not
         # trip the "without a reference" guard meant for real object-backed rows.

posthog/temporal/oauth.py

@@ -16,6 +16,20 @@
 POSTHOG_AI_APP_CLIENT_ID_EU = "0Lizwa3mFSlBuEEQ8V8FMJlskUXpDuSmoEdhzxyi"
 POSTHOG_AI_APP_CLIENT_ID_DEV = "DD2ZLG6a2YEUtpPANSzSiIBPuUryYmbndLnKKUy1"
 
+# Every OAuth application sandbox agent tokens are minted under. Tokens for these apps
+# are only ever created server-side (never via the consent flow or personal API keys),
+# so a request bearing one provably originates from a sandbox run.
+SANDBOX_OAUTH_APP_CLIENT_IDS = frozenset(
+    {
+        ARRAY_APP_CLIENT_ID_US,
+        ARRAY_APP_CLIENT_ID_EU,
+        ARRAY_APP_CLIENT_ID_DEV,
+        POSTHOG_AI_APP_CLIENT_ID_US,
+        POSTHOG_AI_APP_CLIENT_ID_EU,
+        POSTHOG_AI_APP_CLIENT_ID_DEV,
+    }
+)
+
 McpScopePreset = Literal["read_only", "full", "signals_scout", "signals_scout_reports"]
 SandboxOAuthApplication = Literal["array", "posthog_ai"]