feat(channels): stream agent turns into threads, read Activity per mention

apps/code/src/renderer/desktop-services.ts

@@ -70,6 +70,7 @@ import {
   type IAuthSideEffects,
 } from "@posthog/ui/features/auth/identifiers";
 import { authKeys } from "@posthog/ui/features/auth/useCurrentUser";
+import { useThreadPanelStore } from "@posthog/ui/features/canvas/stores/threadPanelStore";
 import {
   FEATURE_FLAGS,
   type FeatureFlags,
@@ -85,6 +86,7 @@ import {
 } from "@posthog/ui/features/integrations/integrationsClientImpl";
 import { NAVIGATION_TASK_BINDER } from "@posthog/ui/features/navigation/taskBinder";
 import { navigationTaskBinder } from "@posthog/ui/features/navigation/taskBinderImpl";
+import { activeNotificationTarget } from "@posthog/ui/features/notifications/activeTarget";
 import {
   ACTIVE_VIEW_PROVIDER,
   type IActiveView,
@@ -324,28 +326,18 @@ container.bind<IActiveView>(ACTIVE_VIEW_PROVIDER).toConstantValue({
   hasFocus: () => document.hasFocus(),
   // Read the active leaf route directly: AppView collapses the channel routes
   // and drops channelId/dashboardId, which we need to identify a canvas target.
+  // What counts as "viewing" is decided in ui — a task can be on screen in a
+  // thread panel without owning the route.
   getActiveTarget: (): NotificationTarget | undefined => {
     const matches = getCurrentMatches();
     const last = matches[matches.length - 1];
     if (!last) return undefined;
-    const params = last.params as Record<string, string | undefined>;
-    switch (last.routeId) {
-      case "/code/tasks/$taskId":
-      case "/website/$channelId/tasks/$taskId":
-        return params.taskId
-          ? { kind: "task", taskId: params.taskId }
-          : undefined;
-      case "/website/$channelId/dashboards/$dashboardId":
-        return params.channelId && params.dashboardId
-          ? {
-              kind: "canvas",
-              channelId: params.channelId,
-              dashboardId: params.dashboardId,
-            }
-          : undefined;
-      default:
-        return undefined;
-    }
+    const { openByChannel, collapsed } = useThreadPanelStore.getState();
+    return activeNotificationTarget({
+      routeId: last.routeId,
+      params: last.params as Record<string, string | undefined>,
+      threadPanel: { openByChannel, collapsed },
+    });
   },
 });
 

packages/core/src/canvas/mentionActivity.test.ts

@@ -1,7 +1,10 @@
 import type { TaskMention, UserBasic } from "@posthog/shared/domain-types";
 import { describe, expect, it } from "vitest";
 import {
+  countUnreadMentions,
   countUnseenActivity,
+  isMentionUnread,
+  type MentionActivityItem,
   mergeTaskMentions,
   toMentionActivityItems,
 } from "./mentionActivity";
@@ -133,3 +136,52 @@ describe("mergeTaskMentions", () => {
     expect(merged[0].message_id).toBe("newest");
   });
 });
+
+describe("isMentionUnread", () => {
+  const item = (over: Partial<MentionActivityItem> = {}) =>
+    ({
+      messageId: "m1",
+      taskId: "t1",
+      taskTitle: "Task",
+      channelId: "c1",
+      channelName: "mobile",
+      author: null,
+      content: "@you",
+      createdAt: "2026-07-17T10:00:00.000Z",
+      ...over,
+    }) as MentionActivityItem;
+
+  const none: ReadonlySet<string> = new Set();
+
+  it("is unread until its thread is opened", () => {
+    expect(isMentionUnread(item(), null, none)).toBe(true);
+    expect(isMentionUnread(item(), null, new Set(["m1"]))).toBe(false);
+  });
+
+  it("reading one mention leaves the others unread", () => {
+    const read = new Set(["m1"]);
+    expect(isMentionUnread(item({ messageId: "m2" }), null, read)).toBe(true);
+  });
+
+  it("treats anything before the legacy seen watermark as read", () => {
+    const old = item({ createdAt: "2026-07-01T00:00:00.000Z" });
+    expect(isMentionUnread(old, "2026-07-10T00:00:00.000Z", none)).toBe(false);
+  });
+
+  it("keeps mentions after the watermark unread until opened", () => {
+    const fresh = item({ createdAt: "2026-07-17T10:00:00.000Z" });
+    const seen = "2026-07-10T00:00:00.000Z";
+    expect(isMentionUnread(fresh, seen, none)).toBe(true);
+    expect(isMentionUnread(fresh, seen, new Set(["m1"]))).toBe(false);
+  });
+
+  it("counts only what's unread", () => {
+    const items = [
+      item({ messageId: "a" }),
+      item({ messageId: "b" }),
+      item({ messageId: "c" }),
+    ];
+    expect(countUnreadMentions(items, null, new Set(["b"]))).toBe(2);
+    expect(countUnreadMentions(items, null, new Set(["a", "b", "c"]))).toBe(0);
+  });
+});

packages/core/src/canvas/mentionActivity.ts

@@ -44,6 +44,36 @@ export function countUnseenActivity(
   return items.filter((item) => item.createdAt > lastSeenAt).length;
 }
 
+/**
+ * Has the viewer read this mention?
+ *
+ * Read is per mention: it's earned by opening that mention's thread, not by
+ * glancing at the list. `lastSeenAt` stays on as a historical watermark —
+ * everything from before the viewer's last "seen the page" sweep counts as
+ * read, so switching to per-mention tracking doesn't resurface years of old
+ * mentions as unread. Anything after it is unread until opened.
+ */
+export function isMentionUnread(
+  item: MentionActivityItem,
+  lastSeenAt: string | null,
+  readMessageIds: ReadonlySet<string>,
+): boolean {
+  if (readMessageIds.has(item.messageId)) return false;
+  if (!lastSeenAt) return true;
+  return item.createdAt > lastSeenAt;
+}
+
+/** How many mentions the viewer hasn't read. Drives the sidebar's badge. */
+export function countUnreadMentions(
+  items: readonly MentionActivityItem[],
+  lastSeenAt: string | null,
+  readMessageIds: ReadonlySet<string>,
+): number {
+  return items.filter((item) =>
+    isMentionUnread(item, lastSeenAt, readMessageIds),
+  ).length;
+}
+
 // Bounds the cache so a long-running session's accumulated feed can't grow
 // without limit.
 const MAX_CACHED_MENTIONS = 300;

packages/core/src/canvas/threadTimeline.test.ts

@@ -3,8 +3,10 @@ import {
   buildThreadTimeline,
   deriveThreadAgentStatus,
   hasAgentMention,
+  isAgentThreadMessage,
   normalizeAgentPromptText,
   shouldSuspendThreadSession,
+  visibleThreadMessages,
 } from "./threadTimeline";
 
 describe("hasAgentMention", () => {
@@ -167,3 +169,75 @@ describe("shouldSuspendThreadSession", () => {
     expect(shouldSuspendThreadSession(input)).toBe(false);
   });
 });
+
+describe("isAgentThreadMessage", () => {
+  it.each([
+    { author_kind: "agent", expected: true },
+    { author_kind: "human", expected: false },
+    { author_kind: "system", expected: false },
+    // Older payloads predate agent rows; absent means human.
+    { author_kind: undefined, expected: false },
+  ])("author_kind $author_kind → $expected", ({ author_kind, expected }) => {
+    expect(isAgentThreadMessage({ author_kind })).toBe(expected);
+  });
+});
+
+describe("visibleThreadMessages", () => {
+  const human = { id: "h", author_kind: "human" as const };
+  const turn = (runId: string, id = `turn-${runId}`) => ({
+    id,
+    author_kind: "agent" as const,
+    event: "turn_complete",
+    payload: { run_id: runId },
+  });
+
+  it("keeps everything when no run is streaming", () => {
+    expect(
+      visibleThreadMessages([human, turn("run-1")], undefined).map((m) => m.id),
+    ).toEqual(["h", "turn-run-1"]);
+  });
+
+  it("drops the durable turn for the run being streamed", () => {
+    expect(
+      visibleThreadMessages([human, turn("run-1")], "run-1").map((m) => m.id),
+    ).toEqual(["h"]);
+  });
+
+  it("keeps durable turns from other runs", () => {
+    expect(
+      visibleThreadMessages([turn("run-1"), turn("run-2")], "run-1").map(
+        (m) => m.id,
+      ),
+    ).toEqual(["turn-run-2"]);
+  });
+
+  it("never drops human messages", () => {
+    expect(visibleThreadMessages([human], "run-1").map((m) => m.id)).toEqual([
+      "h",
+    ]);
+  });
+
+  it("keeps other agent rows — only turn_complete collides with a live turn", () => {
+    const ask = {
+      id: "ask",
+      author_kind: "agent" as const,
+      event: "permission_request",
+      payload: { run_id: "run-1" },
+    };
+    expect(visibleThreadMessages([ask], "run-1").map((m) => m.id)).toEqual([
+      "ask",
+    ]);
+  });
+
+  it("keeps a turn row with no run id rather than dropping the only copy", () => {
+    const orphan = {
+      id: "orphan",
+      author_kind: "agent" as const,
+      event: "turn_complete",
+      payload: {},
+    };
+    expect(visibleThreadMessages([orphan], "run-1").map((m) => m.id)).toEqual([
+      "orphan",
+    ]);
+  });
+});

packages/core/src/canvas/threadTimeline.ts

@@ -142,3 +142,43 @@ export function shouldSuspendThreadSession({
 }): boolean {
   return !isCloud && !hasRun && !hasSession;
 }
+
+/**
+ * Agent rows are authorless by design, so anything reading `author` alone sees
+ * nobody and calls them "Unknown". `author_kind` is the only thing that says
+ * the agent wrote it.
+ */
+export function isAgentThreadMessage(message: {
+  author_kind?: string;
+}): boolean {
+  return message.author_kind === "agent";
+}
+
+/**
+ * The durable thread messages worth rendering, given the run whose turns the
+ * viewer is already watching stream.
+ *
+ * The backend posts every agent turn as a durable `turn_complete` row carrying
+ * `payload.run_id` so a client streaming that run can drop one copy. We drop
+ * the durable row rather than the live turn: the streamed one is what the
+ * reader watched arrive, and swapping it for the server's copy at the end would
+ * make the message jump. Anyone without that session — a teammate, or a run on
+ * another machine — has no live turn to collide with, so they keep the durable
+ * row and see the same conversation from the other side.
+ */
+export function visibleThreadMessages<
+  T extends {
+    author_kind?: string;
+    event?: string;
+    payload?: Record<string, unknown>;
+  },
+>(messages: readonly T[], streamingRunId: string | undefined): T[] {
+  if (!streamingRunId) return [...messages];
+  return messages.filter((message) => {
+    if (message.event !== "turn_complete") return true;
+    const runId = message.payload?.run_id;
+    // A row with no run id can't be matched to the live turn; keeping a
+    // possible duplicate beats silently dropping the only copy.
+    return typeof runId !== "string" || runId !== streamingRunId;
+  });
+}

packages/shared/src/domain-types.ts

@@ -100,16 +100,36 @@ export interface ChannelFeedMessage {
   created_at: string;
 }
 
+/** Server-emitted agent rows in a task's thread. */
+export type TaskThreadMessageEvent =
+  /** The agent's final turn, made durable. `payload.run_id` identifies the run,
+   *  so a client already streaming that run can drop this copy. */
+  | "turn_complete"
+  /** The agent is asking to do something before it proceeds. */
+  | "permission_request";
+
 /**
- * One human message in a task's thread. Thread messages never reach the agent
- * unless the task author forwards one, which stamps the forwarded_* fields.
+ * One message in a task's thread. Human messages never reach the agent unless
+ * the task author forwards one, which stamps the forwarded_* fields.
+ *
+ * Agent rows are authorless (`author_kind: "agent"`, no `author`) and carry a
+ * stable `event` + structured `payload`, so they can be rendered from those
+ * rather than by parsing `content` — which stays the rendered text, so a client
+ * that doesn't know the event still shows something sensible.
+ *
+ * `author_kind` is optional only so older payloads (and test fixtures) parse;
+ * absent means human, which is what every message was before agents could post.
  */
 export interface TaskThreadMessage {
   id: string;
   task: string;
   content: string;
   created_at: string;
   author?: UserBasic | null;
+  author_kind?: "human" | "system" | "agent";
+  /** Empty for human messages. */
+  event?: TaskThreadMessageEvent | string;
+  payload?: Record<string, unknown>;
   forwarded_to_agent_at?: string | null;
   forwarded_by?: UserBasic | null;
 }