feat(tasks): post agent thread updates for canvas creation and turn completion (#70371)

Co-authored-by: tests-posthog[bot] <250237707+tests-posthog[bot]@users.noreply.github.com>

Author: k11kirky

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

@@ -259,6 +259,7 @@ rules:
                               |TaskThreadMessage
                               |TaskThreadMessageMention
                               |Channel
+                              |ChannelFeedMessage
                               |EmailChannel
                               |EvaluationReport
                               |Text
@@ -542,6 +543,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
@@ -63,6 +64,8 @@
 from posthog.models.user import User
 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 +1135,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 +1160,59 @@ 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 sandbox authenticates
+        with the task creator's credentials, so the facade only accepts a task created
+        by the requesting user — the header can't point the announcement at someone
+        else's task thread. 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
+        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),
+        )
+
+    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,18 @@
-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.user import User
+
+if TYPE_CHECKING:
+    from products.tasks.backend.models import Task
 
 
 class TestDesktopCanvasPublishAPI(APIBaseTest):
@@ -96,6 +104,80 @@ 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)
+
+    @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.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.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.client.patch(self._canvas_url(item_id), {"code": "v1"}, HTTP_X_POSTHOG_TASK_ID=str(task.id))
+
+        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.

products/tasks/backend/facade/api.py

@@ -24,7 +24,7 @@
 from uuid import UUID, uuid4
 
 from django.conf import settings
-from django.db import IntegrityError, transaction
+from django.db import IntegrityError, close_old_connections, transaction
 from django.db.models import CharField, Count, Exists, F, Min, OuterRef, Q, QuerySet, Subquery
 from django.db.models.fields.json import KeyTextTransform
 from django.utils import timezone as django_timezone
@@ -49,7 +49,7 @@
     is_custom_images_enabled,
     read_spec_from_builder_sandbox,
 )
-from products.tasks.backend.mentions import resolve_mentioned_user_ids
+from products.tasks.backend.mentions import format_mention_token, resolve_mentioned_user_ids
 from products.tasks.backend.models import (
     Channel,
     ChannelFeedMessage,
@@ -4863,6 +4863,9 @@ def _thread_message_to_dto(message: TaskThreadMessage) -> contracts.TaskThreadMe
     return contracts.TaskThreadMessageDTO(
         id=message.id,
         task=message.task_id,
+        author_kind=message.author_kind,
+        event=message.event,
+        payload=message.payload or {},
         content=message.content,
         created_at=message.created_at,
         author=_user_basic_info(message.author if message.author_id else None),
@@ -5018,6 +5021,127 @@ def forward_thread_message(
     return "ok", _thread_message_to_dto(message)
 
 
+# Threads are a Channels (project-bluebird) surface, so agent-authored thread
+# updates are gated on the same flag — evaluated for the task creator.
+AGENT_THREAD_UPDATES_FLAG = "project-bluebird"
+
+# One turn-complete post per run within the window, so an SSE relay reconnect
+# replaying the tail of the stream can't double-post the same end-of-turn.
+_TURN_COMPLETE_COOLDOWN_SECONDS = 30
+
+# Cap the relayed final message so one agent essay can't dwarf the thread.
+_TURN_MESSAGE_MAX_CHARS = 4000
+
+
+def _create_agent_thread_message(task: Task, content: str, *, event: str, payload: dict | None = None) -> None:
+    """Write an agent-authored thread message and index its mentions.
+
+    ``content`` is the rendered text (older clients show it as-is); ``event`` +
+    ``payload`` are the structured record, mirroring ChannelFeedMessage, that
+    lets clients render agent rows natively and dedupe them against live views.
+    """
+    message = TaskThreadMessage.objects.create(
+        team_id=task.team_id,
+        task_id=task.id,
+        author_id=None,
+        author_kind=TaskThreadMessage.AuthorKind.AGENT,
+        event=event,
+        payload=payload or {},
+        content=content,
+    )
+    try:
+        _index_thread_message_mentions(message)
+    except Exception:
+        logger.exception("Failed to index thread message mentions", extra={"message_id": str(message.id)})
+
+
+def _agent_thread_updates_enabled(creator: User | None) -> bool:
+    """Fail closed: no creator to key the flag on, or a flag-service error, means no post."""
+    if creator is None:
+        return False
+    distinct_id = creator.distinct_id or f"user_{creator.id}"
+    try:
+        return bool(
+            posthoganalytics.feature_enabled(AGENT_THREAD_UPDATES_FLAG, distinct_id, send_feature_flag_events=False)
+        )
+    except Exception:
+        logger.warning("Agent thread update flag check failed", extra={"user_id": creator.id}, exc_info=True)
+        return False
+
+
+def post_canvas_created_thread_update(
+    task_id: str | UUID, team_id: int, *, acting_user_id: int | None, canvas_name: str, canvas_url: str | None
+) -> None:
+    """Announce a freshly created canvas in the generating task's thread.
+
+    Posts "[name](url) has been created" as an agent message. Called on a canvas's
+    first publish only — the caller owns that once-guard. ``acting_user_id`` must be
+    the task's creator: the sandbox publishes with the creator's credentials, so this
+    binds the attributed task to the caller's identity — a same-team caller can't
+    plant agent messages in someone else's task thread by naming its id. Best-effort
+    and never raises: the publish must not fail because its announcement couldn't
+    be written.
+    """
+    try:
+        task = Task.objects.select_related("created_by").filter(id=task_id, team_id=team_id).first()
+        if task is None or task.created_by_id is None or task.created_by_id != acting_user_id:
+            return
+        if not _agent_thread_updates_enabled(task.created_by):
+            return
+        # Brackets and newlines in the name would break the [label](url) token.
+        name = re.sub(r"[\[\]\n]", " ", canvas_name).strip() or "Canvas"
+        content = f"[{name}]({canvas_url}) has been created" if canvas_url else f"{name} has been created"
+        _create_agent_thread_message(
+            task,
+            content,
+            event="canvas_created",
+            payload={"canvas_name": name, "canvas_url": canvas_url},
+        )
+    except Exception:
+        logger.exception("Failed to post canvas-created thread update", extra={"task_id": str(task_id)})
+
+
+def post_turn_complete_thread_update(
+    run_id: str | UUID, task_id: str | UUID, team_id: int, *, message: str | None = None
+) -> None:
+    """Post the agent's final turn message into the task's thread, @-mentioning the task creator.
+
+    Fires from the sandbox event relay on every end-of-turn of a channel task's
+    background run, so the update lands even with no client open. ``message`` is
+    the agent's closing prose for the turn; when the relay captured none, a plain
+    "Turn complete." stands in. Best-effort and never raises — a failed post must
+    not disturb the relay.
+    """
+    try:
+        if not settings.TEST:
+            close_old_connections()
+        task = Task.objects.select_related("created_by").filter(id=task_id, team_id=team_id).first()
+        # Threads hang off a task's channel feed; a channel-less task has no audience.
+        if task is None or task.channel_id is None:
+            return
+        creator = task.created_by
+        if creator is None or not _agent_thread_updates_enabled(creator):
+            return
+        from products.tasks.backend.redis import get_tasks_cache  # noqa: PLC0415 — keep redis off the api import path
+
+        if not get_tasks_cache().add(f"thread_update:{run_id}:turn_complete", True, _TURN_COMPLETE_COOLDOWN_SECONDS):
+            return
+        body = (message or "").strip() or "Turn complete."
+        if len(body) > _TURN_MESSAGE_MAX_CHARS:
+            body = body[: _TURN_MESSAGE_MAX_CHARS - 1] + "…"
+        mention = format_mention_token(creator.get_full_name() or creator.email, creator.email)
+        # payload.run_id is the dedupe key: a client already rendering this run's
+        # live agent turns can suppress the durable row (or vice versa).
+        _create_agent_thread_message(
+            task,
+            f"{mention} {body}",
+            event="turn_complete",
+            payload={"run_id": str(run_id)},
+        )
+    except Exception:
+        logger.exception("Failed to post turn-complete thread update", extra={"task_id": str(task_id)})
+
+
 def respond_to_permission_request(
     run_id: str | UUID,
     task_id: str | UUID,

products/tasks/backend/facade/contracts.py

@@ -153,6 +153,9 @@ class TaskThreadMessageDTO:
 
     id: UUID
     task: UUID
+    author_kind: str
+    event: str
+    payload: dict
     content: str
     created_at: datetime
     author: "TaskUserBasicInfo | None" = None

products/tasks/backend/mentions.py

@@ -22,6 +22,16 @@ def extract_mention_emails(content: str) -> set[str]:
     return {match.group(1).lower() for match in MENTION_TOKEN_PATTERN.finditer(content)}
 
 
+def format_mention_token(name: str, email: str) -> str:
+    """Serialize a user reference into the inline mention token.
+
+    Brackets and newlines would break token parsing; the email is the identity,
+    so the name falls back to its local part when unusable.
+    """
+    safe_name = re.sub(r"[\[\]\n]", " ", name).strip() or email.split("@")[0] or email
+    return f"@[{safe_name}]({email})"
+
+
 def resolve_mentioned_user_ids(user_model: Any, content: str, *, team_id: int, author_id: int | None) -> list[int]:
     """Ids of the team's org members mentioned in the content, excluding the author.