diff --git a/.semgrep/rules/security/idor-team-scoped-models.yaml b/.semgrep/rules/security/idor-team-scoped-models.yaml
index 8ab093dd0016..3e1ed02c74a7 100644
--- a/.semgrep/rules/security/idor-team-scoped-models.yaml
+++ b/.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
diff --git a/posthog/api/file_system/file_system.py b/posthog/api/file_system/file_system.py
index 84c81ddd4b61..e79f4d0e6e74 100644
--- a/posthog/api/file_system/file_system.py
+++ b/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:
diff --git a/posthog/api/file_system/test/test_canvas_publish.py b/posthog/api/file_system/test/test_canvas_publish.py
index 791c929bdce0..d918e22b940d 100644
--- a/posthog/api/file_system/test/test_canvas_publish.py
+++ b/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.
diff --git a/products/tasks/backend/facade/api.py b/products/tasks/backend/facade/api.py
index 5a9327a6f418..696dfd28ba6d 100644
--- a/products/tasks/backend/facade/api.py
+++ b/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,
diff --git a/products/tasks/backend/facade/contracts.py b/products/tasks/backend/facade/contracts.py
index 1cb022e31fe3..29b2007bf035 100644
--- a/products/tasks/backend/facade/contracts.py
+++ b/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
diff --git a/products/tasks/backend/mentions.py b/products/tasks/backend/mentions.py
index a65daf954d2a..f0ee4cad5204 100644
--- a/products/tasks/backend/mentions.py
+++ b/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.
 
diff --git a/products/tasks/backend/migrations/0057_taskthreadmessage_agent_fields.py b/products/tasks/backend/migrations/0057_taskthreadmessage_agent_fields.py
new file mode 100644
index 000000000000..6ccad078fb3d
--- /dev/null
+++ b/products/tasks/backend/migrations/0057_taskthreadmessage_agent_fields.py
@@ -0,0 +1,29 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("tasks", "0056_channelfeedmessage"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="taskthreadmessage",
+            name="author_kind",
+            field=models.CharField(
+                choices=[("human", "Human"), ("system", "System"), ("agent", "Agent")],
+                default="human",
+                max_length=16,
+            ),
+        ),
+        migrations.AddField(
+            model_name="taskthreadmessage",
+            name="event",
+            field=models.CharField(blank=True, default="", max_length=64),
+        ),
+        migrations.AddField(
+            model_name="taskthreadmessage",
+            name="payload",
+            field=models.JSONField(blank=True, default=dict),
+        ),
+    ]
diff --git a/products/tasks/backend/migrations/max_migration.txt b/products/tasks/backend/migrations/max_migration.txt
index f14ffd9a113d..5c64038bac7f 100644
--- a/products/tasks/backend/migrations/max_migration.txt
+++ b/products/tasks/backend/migrations/max_migration.txt
@@ -1 +1 @@
-0056_channelfeedmessage
+0057_taskthreadmessage_agent_fields
diff --git a/products/tasks/backend/models.py b/products/tasks/backend/models.py
index 648835496914..12abdc851f5a 100644
--- a/products/tasks/backend/models.py
+++ b/products/tasks/backend/models.py
@@ -812,9 +812,18 @@ def _dispatch() -> None:
 
 
 class TaskThreadMessage(TeamScopedRootMixin):
-    """One human message in a task's thread — the side conversation channel members
-    have around a task. Messages never reach the agent unless the task author
-    forwards one (send_to_agent), which stamps the forwarded_* fields."""
+    """One message in a task's thread — the side conversation channel members have
+    around a task. Human messages never reach the agent unless the task author
+    forwards one (send_to_agent), which stamps the forwarded_* fields. Agent rows
+    (``author_kind=AGENT``, no ``author``) are server-emitted announcements carrying
+    a stable ``event`` key + ``payload`` — the same shape as ``ChannelFeedMessage`` —
+    so clients can render them structurally and dedupe them against live
+    session-derived views (e.g. ``turn_complete`` carries the run id)."""
+
+    class AuthorKind(models.TextChoices):
+        HUMAN = "human", "Human"
+        SYSTEM = "system", "System"
+        AGENT = "agent", "Agent"
 
     # nosemgrep: prefer-uuid7-django-pk -- mirrors sibling task models in this app
     id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
@@ -826,6 +835,11 @@ class TaskThreadMessage(TeamScopedRootMixin):
     author = models.ForeignKey(
         "posthog.User", on_delete=models.SET_NULL, null=True, blank=True, related_name="+", db_constraint=False
     )
+    author_kind = models.CharField(max_length=16, choices=AuthorKind, default=AuthorKind.HUMAN)
+    # Stable event key + structured payload for non-human rows (empty for human
+    # messages); `content` stays the rendered text so older clients degrade cleanly.
+    event = models.CharField(max_length=64, blank=True, default="")
+    payload = models.JSONField(default=dict, blank=True)
     content = models.TextField()
     forwarded_to_agent_at = models.DateTimeField(null=True, blank=True)
     forwarded_by = models.ForeignKey(
diff --git a/products/tasks/backend/presentation/serializers.py b/products/tasks/backend/presentation/serializers.py
index 529fe7c12304..5bda37632614 100644
--- a/products/tasks/backend/presentation/serializers.py
+++ b/products/tasks/backend/presentation/serializers.py
@@ -1274,7 +1274,18 @@ class TaskThreadMessageSerializer(DataclassSerializer):
 
     class Meta:
         dataclass = TaskThreadMessageDTO
-        fields = ["id", "task", "content", "created_at", "author", "forwarded_to_agent_at", "forwarded_by"]
+        fields = [
+            "id",
+            "task",
+            "author_kind",
+            "event",
+            "payload",
+            "content",
+            "created_at",
+            "author",
+            "forwarded_to_agent_at",
+            "forwarded_by",
+        ]
 
 
 class TaskThreadMessageWriteSerializer(serializers.Serializer):
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 0808bfb9c66f..52259ca2ad05 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
@@ -19,6 +19,7 @@
 
 from posthog.temporal.common.utils import close_db_connections
 
+from products.tasks.backend.facade import api as tasks_facade
 from products.tasks.backend.logic.services.agent_command import validate_sandbox_url
 from products.tasks.backend.logic.services.connection_token import create_sandbox_connection_token
 from products.tasks.backend.logic.services.permission_broker import (
@@ -303,6 +304,10 @@ async def _relay_loop(
     last_audit_ts_ns: list[int] = [0]  # track last agentsh audit timestamp
     # Brackets turn_started / turn_completed signals to the parent.
     slack_turn_active: list[bool] = [False]
+    # The agent's in-progress closing message for the current turn. Chunks
+    # accumulate; a new tool call or user message resets it, so at end-of-turn
+    # it holds the prose after the last tool call — what the thread update posts.
+    final_message_parts: list[str] = []
     # ACP emits one tool_call + N tool_call_update per id; only render the start.
     emitted_tool_call_ids: set[str] = set()
 
@@ -374,12 +379,31 @@ async def _relay_loop(
                                     # does sync Redis (cache.add) and a potential network call to
                                     # the feature-flag service.
                                     asyncio.create_task(asyncio.to_thread(_safe_dispatch_awaiting_input, task_run))
+                                if task_run is not None and task_run.mode != "interactive":
+                                    # Background run finished a turn — post its closing message
+                                    # into the task's thread so teammates following it see the
+                                    # outcome without a client open. Guards (flag, channel,
+                                    # cooldown) live in the facade; same thread hop as above
+                                    # for its sync I/O.
+                                    asyncio.create_task(
+                                        asyncio.to_thread(
+                                            tasks_facade.post_turn_complete_thread_update,
+                                            str(task_run.id),
+                                            str(task_run.task_id),
+                                            task_run.team_id,
+                                            message="".join(final_message_parts).strip() or None,
+                                        )
+                                    )
+                                final_message_parts.clear()
                                 if is_agent_design_enabled and slack_turn_active[0] and workflow_handle is not None:
                                     slack_turn_active[0] = False
                                     asyncio.create_task(_signal_safely(workflow_handle, "turn_completed"))
                             elif not agent_active[0] and _is_active_agent_update(event_data):
                                 agent_active[0] = True
 
+                            if task_run is not None and task_run.mode != "interactive":
+                                _track_final_message(event_data, final_message_parts)
+
                             # Agent-design signal fan-out: first session/update opens the
                             # child relay; tool_call → step, agent_message_chunk → markdown.
                             if is_agent_design_enabled and workflow_handle is not None:
@@ -432,6 +456,7 @@ async def _relay_loop(
                 reconnect_count += 1
                 # May have missed an end_of_turn on the dropped stream — assume idle until re-confirmed.
                 agent_active[0] = False
+                final_message_parts.clear()
                 logger.warning(
                     "relay_sandbox_events_read_timeout",
                     run_id=run_id,
@@ -453,6 +478,7 @@ async def _relay_loop(
                 # 5xx — transient server error, worth retrying
                 reconnect_count += 1
                 agent_active[0] = False  # missed-end_of_turn guard (see ReadTimeout above)
+                final_message_parts.clear()
                 logger.warning(
                     "relay_sandbox_events_http_error",
                     run_id=run_id,
@@ -465,6 +491,7 @@ async def _relay_loop(
             except (httpx.TransportError, httpx_sse.SSEError) as e:
                 reconnect_count += 1
                 agent_active[0] = False  # missed-end_of_turn guard (see ReadTimeout above)
+                final_message_parts.clear()
                 logger.warning(
                     "relay_sandbox_events_connection_error",
                     run_id=run_id,
@@ -594,6 +621,20 @@ def _tool_args_preview(raw_input: Any) -> str | None:
     return one_line
 
 
+def _track_final_message(event_data: dict, parts: list[str]) -> None:
+    """Accumulate agent_message_chunk text; a new tool call or user message resets,
+    so `parts` ends the turn holding only the agent's closing prose."""
+    text = _extract_agent_message_text(event_data)
+    if text:
+        parts.append(text)
+        return
+    if not _is_session_update(event_data):
+        return
+    update = (event_data.get("notification", {}).get("params") or {}).get("update") or {}
+    if update.get("sessionUpdate") in ("tool_call", "user_message", "user_message_chunk"):
+        parts.clear()
+
+
 def _extract_agent_message_text(event_data: dict) -> str | None:
     """Text delta from an ACP agent_message_chunk session/update, else None."""
     notification = event_data.get("notification", {})
diff --git a/products/tasks/backend/tests/test_thread_updates.py b/products/tasks/backend/tests/test_thread_updates.py
new file mode 100644
index 000000000000..4f197b309a34
--- /dev/null
+++ b/products/tasks/backend/tests/test_thread_updates.py
@@ -0,0 +1,186 @@
+from unittest.mock import patch
+
+from django.core.cache import cache
+from django.test import SimpleTestCase, TestCase
+
+from parameterized import parameterized
+
+from posthog.models import Organization, OrganizationMembership, Team, User
+
+from products.tasks.backend.facade.api import post_canvas_created_thread_update, post_turn_complete_thread_update
+from products.tasks.backend.models import Channel, Task, TaskRun, TaskThreadMessage, TaskThreadMessageMention
+from products.tasks.backend.temporal.process_task.activities.relay_sandbox_events import _track_final_message
+
+_FLAG_TARGET = "products.tasks.backend.facade.api.posthoganalytics.feature_enabled"
+
+
+class TestAgentThreadUpdates(TestCase):
+    def setUp(self) -> None:
+        cache.clear()
+        self.organization = Organization.objects.create(name="Test Org")
+        self.team = Team.objects.create(organization=self.organization, name="Test Team")
+        self.user = User.objects.create_user(
+            email="creator@example.com", first_name="Casey", last_name="Creator", password="password"
+        )
+        OrganizationMembership.objects.create(user=self.user, organization=self.organization)
+        self.channel = Channel.objects.create(team=self.team, name="general")
+        self.task = Task.objects.create(
+            team=self.team,
+            title="Build canvas",
+            description="",
+            origin_product=Task.OriginProduct.USER_CREATED,
+            created_by=self.user,
+            channel=self.channel,
+        )
+        self.task_run = TaskRun.objects.create(task=self.task, team=self.team)
+
+    def _messages(self, task: Task) -> list[TaskThreadMessage]:
+        return list(TaskThreadMessage.objects.for_team(self.team.id).filter(task=task).order_by("created_at"))
+
+    @parameterized.expand(
+        [
+            (
+                "relays_final_message",
+                "Shipped the canvas with three charts.",
+                "@[Casey Creator](creator@example.com) Shipped the canvas with three charts.",
+            ),
+            ("falls_back_without_message", None, "@[Casey Creator](creator@example.com) Turn complete."),
+        ]
+    )
+    @patch(_FLAG_TARGET, return_value=True)
+    def test_turn_complete_posts_authorless_message_mentioning_creator(self, _name, message, expected, _flag) -> None:
+        post_turn_complete_thread_update(str(self.task_run.id), str(self.task.id), self.team.id, message=message)
+
+        messages = self._messages(self.task)
+        self.assertEqual(len(messages), 1)
+        self.assertIsNone(messages[0].author_id)
+        self.assertEqual(messages[0].author_kind, TaskThreadMessage.AuthorKind.AGENT)
+        self.assertEqual(messages[0].event, "turn_complete")
+        # run_id is the client's key for deduping this durable row against
+        # live session-derived agent turns.
+        self.assertEqual(messages[0].payload, {"run_id": str(self.task_run.id)})
+        self.assertEqual(messages[0].content, expected)
+        # The creator's mention is indexed so it lands in their mentions feed.
+        self.assertTrue(
+            TaskThreadMessageMention.objects.for_team(self.team.id)
+            .filter(message=messages[0], mentioned_user=self.user)
+            .exists()
+        )
+
+    @patch(_FLAG_TARGET, return_value=True)
+    def test_turn_complete_truncates_oversized_message(self, _flag) -> None:
+        post_turn_complete_thread_update(str(self.task_run.id), str(self.task.id), self.team.id, message="x" * 5000)
+
+        content = self._messages(self.task)[0].content
+        self.assertTrue(content.endswith("…"))
+        self.assertLess(len(content), 4100)
+
+    @patch(_FLAG_TARGET, return_value=True)
+    def test_turn_complete_cooldown_collapses_duplicate_end_of_turn_events(self, _flag) -> None:
+        post_turn_complete_thread_update(str(self.task_run.id), str(self.task.id), self.team.id)
+        post_turn_complete_thread_update(str(self.task_run.id), str(self.task.id), self.team.id)
+
+        self.assertEqual(len(self._messages(self.task)), 1)
+
+    @parameterized.expand(
+        [
+            ("flag_off", False, True, True),
+            ("no_channel", True, False, True),
+            ("no_creator", True, True, False),
+        ]
+    )
+    @patch(_FLAG_TARGET)
+    def test_turn_complete_skips(self, _name, flag_on, has_channel, has_creator, flag_mock) -> None:
+        flag_mock.return_value = flag_on
+        task = Task.objects.create(
+            team=self.team,
+            title="Other task",
+            description="",
+            origin_product=Task.OriginProduct.USER_CREATED,
+            created_by=self.user if has_creator else None,
+            channel=self.channel if has_channel else None,
+        )
+        run = TaskRun.objects.create(task=task, team=self.team)
+
+        post_turn_complete_thread_update(str(run.id), str(task.id), self.team.id)
+
+        self.assertEqual(self._messages(task), [])
+
+    @parameterized.expand(
+        [
+            (
+                "with_link",
+                "Signups overview",
+                "https://us.posthog.com/code/canvas/c/d",
+                "[Signups overview](https://us.posthog.com/code/canvas/c/d) has been created",
+            ),
+            (
+                "name_sanitized_for_link_token",
+                "[Q3] KPIs",
+                "https://us.posthog.com/code/canvas/c/d",
+                "[Q3  KPIs](https://us.posthog.com/code/canvas/c/d) has been created",
+            ),
+            ("without_link", "Signups overview", None, "Signups overview has been created"),
+        ]
+    )
+    @patch(_FLAG_TARGET, return_value=True)
+    def test_canvas_created_message_content(self, _name, canvas_name, canvas_url, expected, _flag) -> None:
+        post_canvas_created_thread_update(
+            self.task.id, self.team.id, acting_user_id=self.user.id, canvas_name=canvas_name, canvas_url=canvas_url
+        )
+
+        messages = self._messages(self.task)
+        self.assertEqual(len(messages), 1)
+        self.assertIsNone(messages[0].author_id)
+        self.assertEqual(messages[0].author_kind, TaskThreadMessage.AuthorKind.AGENT)
+        self.assertEqual(messages[0].event, "canvas_created")
+        self.assertEqual(messages[0].content, expected)
+
+    @patch(_FLAG_TARGET, return_value=True)
+    def test_canvas_created_requires_creator_match(self, _flag) -> None:
+        other = User.objects.create_user(email="other@example.com", first_name="Other", password="password")
+
+        post_canvas_created_thread_update(
+            self.task.id, self.team.id, acting_user_id=other.id, canvas_name="Canvas", canvas_url=None
+        )
+
+        self.assertEqual(self._messages(self.task), [])
+
+    @patch(_FLAG_TARGET, return_value=False)
+    def test_canvas_created_skips_when_flag_off(self, _flag) -> None:
+        post_canvas_created_thread_update(
+            self.task.id, self.team.id, acting_user_id=self.user.id, canvas_name="Canvas", canvas_url=None
+        )
+
+        self.assertEqual(self._messages(self.task), [])
+
+
+def _session_update(update: dict) -> dict:
+    return {"type": "notification", "notification": {"method": "session/update", "params": {"update": update}}}
+
+
+def _chunk(text: str) -> dict:
+    return _session_update({"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": text}})
+
+
+class TestTrackFinalMessage(SimpleTestCase):
+    def test_holds_only_prose_after_last_tool_call(self) -> None:
+        parts: list[str] = []
+        for event in [
+            _chunk("Let me look at the data first. "),
+            _session_update({"sessionUpdate": "tool_call", "toolCallId": "t1"}),
+            _session_update({"sessionUpdate": "tool_call_update", "toolCallId": "t1"}),
+            _chunk("Done. The canvas "),
+            _chunk("shows signups by week."),
+        ]:
+            _track_final_message(event, parts)
+
+        self.assertEqual("".join(parts), "Done. The canvas shows signups by week.")
+
+    @parameterized.expand([("user_message",), ("user_message_chunk",), ("tool_call",)])
+    def test_resets_on(self, session_update: str) -> None:
+        parts = ["stale narration"]
+
+        _track_final_message(_session_update({"sessionUpdate": session_update}), parts)
+
+        self.assertEqual(parts, [])
diff --git a/products/tasks/frontend/generated/api.schemas.ts b/products/tasks/frontend/generated/api.schemas.ts
index aa994c41c87c..47f59325908b 100644
--- a/products/tasks/frontend/generated/api.schemas.ts
+++ b/products/tasks/frontend/generated/api.schemas.ts
@@ -414,6 +414,57 @@ export interface ChannelWriteApi {
     name: string
 }
 
+export type ChannelFeedMessageDTOApiPayload = { [key: string]: unknown }
+
+/**
+ * Response shape for one system announcement in a channel's feed.
+ */
+export interface ChannelFeedMessageDTOApi {
+    id: string
+    channel: string
+    author?: TaskUserBasicInfoApi | null
+    author_kind: string
+    event: string
+    payload: ChannelFeedMessageDTOApiPayload
+    content: string
+    created_at: string
+}
+
+export interface PaginatedChannelFeedMessageDTOListApi {
+    count: number
+    /** @nullable */
+    next?: string | null
+    /** @nullable */
+    previous?: string | null
+    results: ChannelFeedMessageDTOApi[]
+}
+
+/**
+ * * `context_created` - context_created
+ * * `context_md_building` - context_md_building
+ */
+export type EventEnumApi = (typeof EventEnumApi)[keyof typeof EventEnumApi]
+
+export const EventEnumApi = {
+    ContextCreated: 'context_created',
+    ContextMdBuilding: 'context_md_building',
+} as const
+
+/**
+ * Request body for posting a system announcement into a channel's feed.
+ */
+export interface ChannelFeedMessageWriteApi {
+    /** Lifecycle event key.
+     *
+     * * `context_created` - context_created
+     * * `context_md_building` - context_md_building */
+    event: EventEnumApi
+    /** Structured event data, e.g. {"context_name": "mobile"}. */
+    payload?: unknown
+    /** Optional explicit timestamp (within 10 minutes of now), so a client can order a burst of announcements. */
+    created_at?: string
+}
+
 /**
  * Request body for creating (resolve-or-create) or renaming a public channel.
  */
@@ -2183,12 +2234,17 @@ export interface TaskRunLivingArtifactEditRequestApi {
     metadata?: TaskRunLivingArtifactEditRequestApiMetadata
 }
 
+export type TaskThreadMessageDTOApiPayload = { [key: string]: unknown }
+
 /**
  * Response shape for one message in a task's thread.
  */
 export interface TaskThreadMessageDTOApi {
     id: string
     task: string
+    author_kind: string
+    event: string
+    payload: TaskThreadMessageDTOApiPayload
     content: string
     created_at: string
     author?: TaskUserBasicInfoApi | null
@@ -2618,6 +2674,17 @@ export type TaskChannelsListParams = {
     offset?: number
 }
 
+export type TaskChannelsFeedListParams = {
+    /**
+     * Number of results to return per page.
+     */
+    limit?: number
+    /**
+     * The initial index from which to return the results.
+     */
+    offset?: number
+}
+
 export type TaskMentionsListParams = {
     /**
      * Maximum number of mentions to return (newest first).
diff --git a/products/tasks/frontend/generated/api.ts b/products/tasks/frontend/generated/api.ts
index 6fdc10d93dda..ed261a55b7d3 100644
--- a/products/tasks/frontend/generated/api.ts
+++ b/products/tasks/frontend/generated/api.ts
@@ -10,10 +10,13 @@ import { apiMutator } from '../../../../frontend/src/lib/api-orval-mutator'
  */
 import type {
     ChannelDTOApi,
+    ChannelFeedMessageDTOApi,
+    ChannelFeedMessageWriteApi,
     ChannelWriteApi,
     CodeInviteRedeemRequestApi,
     ConnectionTokenResponseApi,
     PaginatedChannelDTOListApi,
+    PaginatedChannelFeedMessageDTOListApi,
     PaginatedSandboxCustomImageDTOListApi,
     PaginatedSandboxEnvironmentDTOListApi,
     PaginatedTaskAutomationDTOListApi,
@@ -41,6 +44,7 @@ import type {
     TaskAutomationDTOApi,
     TaskAutomationWriteApi,
     TaskAutomationsListParams,
+    TaskChannelsFeedListParams,
     TaskChannelsListParams,
     TaskDetailDTOApi,
     TaskMentionsListParams,
@@ -533,6 +537,66 @@ export const taskChannelsCreate = async (
     })
 }
 
+export const getTaskChannelsFeedListUrl = (
+    projectId: string,
+    channelId: string,
+    params?: TaskChannelsFeedListParams
+) => {
+    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_channels/${channelId}/feed/?${stringifiedParams}`
+        : `/api/projects/${projectId}/task_channels/${channelId}/feed/`
+}
+
+/**
+ * A channel's system announcements in chronological order.
+ * @summary List channel feed messages
+ */
+export const taskChannelsFeedList = async (
+    projectId: string,
+    channelId: string,
+    params?: TaskChannelsFeedListParams,
+    options?: RequestInit
+): Promise<PaginatedChannelFeedMessageDTOListApi> => {
+    return apiMutator<PaginatedChannelFeedMessageDTOListApi>(getTaskChannelsFeedListUrl(projectId, channelId, params), {
+        ...options,
+        method: 'GET',
+    })
+}
+
+export const getTaskChannelsFeedCreateUrl = (projectId: string, channelId: string) => {
+    return `/api/projects/${projectId}/task_channels/${channelId}/feed/`
+}
+
+/**
+ * API for a channel's system-announcement feed — durable "PostHog agent" rows
+ * (context created, CONTEXT.md being built) rendered alongside the channel's task
+ * cards. Read by any team member for a public channel; personal channels are owner-only.
+ * @summary Post a channel feed message
+ */
+export const taskChannelsFeedCreate = async (
+    projectId: string,
+    channelId: string,
+    channelFeedMessageWriteApi: ChannelFeedMessageWriteApi,
+    options?: RequestInit
+): Promise<ChannelFeedMessageDTOApi> => {
+    return apiMutator<ChannelFeedMessageDTOApi>(getTaskChannelsFeedCreateUrl(projectId, channelId), {
+        ...options,
+        method: 'POST',
+        headers: { 'Content-Type': 'application/json', ...options?.headers },
+        body: JSON.stringify(channelFeedMessageWriteApi),
+    })
+}
+
 export const getTaskChannelsPartialUpdateUrl = (projectId: string, id: string) => {
     return `/api/projects/${projectId}/task_channels/${id}/`
 }
diff --git a/products/tasks/frontend/generated/api.zod.ts b/products/tasks/frontend/generated/api.zod.ts
index 4039b5334460..1620c2a7f8ce 100644
--- a/products/tasks/frontend/generated/api.zod.ts
+++ b/products/tasks/frontend/generated/api.zod.ts
@@ -294,6 +294,30 @@ export const TaskChannelsCreateBody = /* @__PURE__ */ zod
     })
     .describe('Request body for creating (resolve-or-create) or renaming a public channel.')
 
+/**
+ * API for a channel's system-announcement feed — durable "PostHog agent" rows
+ * (context created, CONTEXT.md being built) rendered alongside the channel's task
+ * cards. Read by any team member for a public channel; personal channels are owner-only.
+ * @summary Post a channel feed message
+ */
+export const TaskChannelsFeedCreateBody = /* @__PURE__ */ zod
+    .object({
+        event: zod
+            .enum(['context_created', 'context_md_building'])
+            .describe('\* `context_created` - context_created\n\* `context_md_building` - context_md_building')
+            .describe(
+                'Lifecycle event key.\n\n\* `context_created` - context_created\n\* `context_md_building` - context_md_building'
+            ),
+        payload: zod.unknown().optional().describe('Structured event data, e.g. {\"context_name\": \"mobile\"}.'),
+        created_at: zod.iso
+            .datetime({ offset: true })
+            .optional()
+            .describe(
+                'Optional explicit timestamp (within 10 minutes of now), so a client can order a burst of announcements.'
+            ),
+    })
+    .describe("Request body for posting a system announcement into a channel's feed.")
+
 /**
  * API for task channels — the shared feeds tasks are kicked off in. Listing lazily
  * provisions the requester's personal "#me" channel; creation is resolve-or-create
@@ -1841,6 +1865,9 @@ export const TasksThreadMessagesSendToAgentCreateBody = /* @__PURE__ */ zod
     .object({
         id: zod.uuid(),
         task: zod.uuid(),
+        author_kind: zod.string(),
+        event: zod.string(),
+        payload: zod.record(zod.string(), zod.unknown()),
         content: zod.string(),
         created_at: zod.iso.datetime({ offset: true }),
         author: zod
diff --git a/products/tasks/mcp/tools.yaml b/products/tasks/mcp/tools.yaml
index f51e0102ed8e..40d8d18c226d 100644
--- a/products/tasks/mcp/tools.yaml
+++ b/products/tasks/mcp/tools.yaml
@@ -72,6 +72,12 @@ tools:
     task-channels-destroy:
         operation: task_channels_destroy
         enabled: false
+    task-channels-feed-create:
+        operation: task_channels_feed_create
+        enabled: false
+    task-channels-feed-list:
+        operation: task_channels_feed_list
+        enabled: false
     task-channels-list:
         operation: task_channels_list
         enabled: false
diff --git a/services/mcp/src/api/generated.ts b/services/mcp/src/api/generated.ts
index ba4aa022c4d4..e0a6acaee5b0 100644
--- a/services/mcp/src/api/generated.ts
+++ b/services/mcp/src/api/generated.ts
@@ -12939,6 +12939,49 @@ export namespace Schemas {
       GithubIssue: 'github_issue',
     } as const;
 
+    export type ChannelFeedMessageDTOPayload = { [key: string]: unknown };
+
+    /**
+     * Response shape for one system announcement in a channel's feed.
+     */
+    export interface ChannelFeedMessageDTO {
+      id: string;
+      channel: string;
+      author?: TaskUserBasicInfo | null;
+      author_kind: string;
+      event: string;
+      payload: ChannelFeedMessageDTOPayload;
+      content: string;
+      created_at: string;
+    }
+
+    /**
+     * * `context_created` - context_created
+     * * `context_md_building` - context_md_building
+     */
+    export type EventEnum = typeof EventEnum[keyof typeof EventEnum];
+
+
+    export const EventEnum = {
+      ContextCreated: 'context_created',
+      ContextMdBuilding: 'context_md_building',
+    } as const;
+
+    /**
+     * Request body for posting a system announcement into a channel's feed.
+     */
+    export interface ChannelFeedMessageWrite {
+      /** Lifecycle event key.
+       *
+       * * `context_created` - context_created
+       * * `context_md_building` - context_md_building */
+      event: EventEnum;
+      /** Structured event data, e.g. {"context_name": "mobile"}. */
+      payload?: unknown;
+      /** Optional explicit timestamp (within 10 minutes of now), so a client can order a burst of announcements. */
+      created_at?: string;
+    }
+
     /**
      * * `widget` - Widget
      * * `email` - Email
@@ -33783,6 +33826,15 @@ export namespace Schemas {
       results: ChannelDTO[];
     }
 
+    export interface PaginatedChannelFeedMessageDTOList {
+      count: number;
+      /** @nullable */
+      next?: string | null;
+      /** @nullable */
+      previous?: string | null;
+      results: ChannelFeedMessageDTO[];
+    }
+
     export interface PaginatedClickhouseEventList {
       /** @nullable */
       next?: string | null;
@@ -37261,12 +37313,17 @@ export namespace Schemas {
       results: TaskSummaryDTO[];
     }
 
+    export type TaskThreadMessageDTOPayload = { [key: string]: unknown };
+
     /**
      * Response shape for one message in a task's thread.
      */
     export interface TaskThreadMessageDTO {
       id: string;
       task: string;
+      author_kind: string;
+      event: string;
+      payload: TaskThreadMessageDTOPayload;
       content: string;
       created_at: string;
       author?: TaskUserBasicInfo | null;
@@ -72221,6 +72278,17 @@ export namespace Schemas {
     offset?: number;
     };
 
+    export type TaskChannelsFeedListParams = {
+    /**
+     * Number of results to return per page.
+     */
+    limit?: number;
+    /**
+     * The initial index from which to return the results.
+     */
+    offset?: number;
+    };
+
     export type TaskMentionsListParams = {
     /**
      * Maximum number of mentions to return (newest first).