feat(activity): use authoritative task activity state

Generated-By: PostHog Code
Task-Id: c4f5025f-8edf-40d4-a163-7174899045d1

Author: k11kirky

packages/api-client/src/posthog-client.ts

@@ -84,7 +84,7 @@ import type {
   SuggestedReviewersArtefact,
   SuggestedReviewerWriteEntry,
   Task,
-  TaskActivity,
+  TaskActivityPage,
   TaskChannel,
   TaskMention,
   TaskRun,
@@ -2634,13 +2634,10 @@ export class PostHogAPIClient {
 
   // Tasks the current user is involved in (created, mentioned, or messaged),
   // one row per task, newest activity first.
-  async getTaskActivity(options?: { since?: string }): Promise<TaskActivity[]> {
+  async getTaskActivity(): Promise<TaskActivityPage> {
     const teamId = await this.getTeamId();
     const urlPath = `/api/projects/${teamId}/task_activity/`;
     const url = new URL(`${this.api.baseUrl}${urlPath}`);
-    if (options?.since) {
-      url.searchParams.set("since", options.since);
-    }
     const response = await this.api.fetcher.fetch({
       method: "get",
       url,
@@ -2649,7 +2646,22 @@ export class PostHogAPIClient {
     if (!response.ok) {
       throw new Error(`Failed to fetch task activity: ${response.statusText}`);
     }
-    return (await response.json()) as TaskActivity[];
+    return (await response.json()) as TaskActivityPage;
+  }
+
+  async markTaskActivityRead(): Promise<void> {
+    const teamId = await this.getTeamId();
+    const urlPath = `/api/projects/${teamId}/task_activity/mark_read/`;
+    const response = await this.api.fetcher.fetch({
+      method: "post",
+      url: new URL(`${this.api.baseUrl}${urlPath}`),
+      path: urlPath,
+    });
+    if (!response.ok) {
+      throw new Error(
+        `Failed to mark task activity read: ${response.statusText}`,
+      );
+    }
   }
 
   async getTaskThreadMessages(taskId: string): Promise<TaskThreadMessage[]> {

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

@@ -1,10 +1,6 @@
 import type { TaskActivity, UserBasic } from "@posthog/shared/domain-types";
 import { describe, expect, it } from "vitest";
-import {
-  countUnseenActivity,
-  mergeTaskActivity,
-  toTaskActivityItems,
-} from "./taskActivity";
+import { toTaskActivityItems } from "./taskActivity";
 
 const ann: UserBasic = {
   id: 2,
@@ -15,6 +11,7 @@ const ann: UserBasic = {
 
 function activity(overrides: Partial<TaskActivity> = {}): TaskActivity {
   return {
+    id: "activity-1",
     task_id: "t1",
     task_title: "Task t1",
     channel_id: "c1",
@@ -24,14 +21,16 @@ function activity(overrides: Partial<TaskActivity> = {}): TaskActivity {
     snippet: "ping @[Me](me@posthog.com)",
     latest_author: ann,
     latest_message_id: "m1",
+    is_unread: true,
     ...overrides,
   };
 }
 
 describe("toTaskActivityItems", () => {
-  it("maps activity DTOs to feed items", () => {
+  it("maps the authoritative activity and unread state", () => {
     expect(toTaskActivityItems([activity()])).toEqual([
       {
+        id: "activity-1",
         taskId: "t1",
         taskTitle: "Task t1",
         channelId: "c1",
@@ -41,12 +40,13 @@ describe("toTaskActivityItems", () => {
         snippet: "ping @[Me](me@posthog.com)",
         author: ann,
         messageId: "m1",
+        isUnread: true,
       },
     ]);
   });
 
-  it("labels untitled tasks and tolerates missing channel/author/message", () => {
-    const items = toTaskActivityItems([
+  it("labels untitled tasks and tolerates missing optional values", () => {
+    const [item] = toTaskActivityItems([
       activity({
         task_title: "",
         channel_id: null,
@@ -57,7 +57,7 @@ describe("toTaskActivityItems", () => {
         snippet: "",
       }),
     ]);
-    expect(items[0]).toMatchObject({
+    expect(item).toMatchObject({
       taskTitle: "Untitled task",
       channelId: null,
       channelName: null,
@@ -66,87 +66,3 @@ describe("toTaskActivityItems", () => {
     });
   });
 });
-
-describe("countUnseenActivity", () => {
-  const items = toTaskActivityItems([
-    activity({ task_id: "t2", activity_at: "2026-07-03T10:00:00Z" }),
-    activity({ task_id: "t1", activity_at: "2026-07-01T10:00:00Z" }),
-  ]);
-
-  it("counts everything when never seen", () => {
-    expect(countUnseenActivity(items, null)).toBe(2);
-  });
-
-  it("counts only rows with activity after the last-seen timestamp", () => {
-    expect(countUnseenActivity(items, "2026-07-02T00:00:00Z")).toBe(1);
-    expect(countUnseenActivity(items, "2026-07-04T00:00:00Z")).toBe(0);
-  });
-});
-
-describe("mergeTaskActivity", () => {
-  it("prepends newly-active tasks ahead of the previous page", () => {
-    const previous = [
-      activity({ task_id: "t1", activity_at: "2026-07-01T10:00:00Z" }),
-    ];
-    const incoming = [
-      activity({ task_id: "t2", activity_at: "2026-07-02T10:00:00Z" }),
-    ];
-    expect(mergeTaskActivity(previous, incoming).map((r) => r.task_id)).toEqual(
-      ["t2", "t1"],
-    );
-  });
-
-  it("replaces a task's row when its activity advances instead of duplicating it", () => {
-    const previous = [
-      activity({
-        task_id: "t1",
-        activity_kind: "mention",
-        activity_at: "2026-07-01T10:00:00Z",
-      }),
-    ];
-    const incoming = [
-      activity({
-        task_id: "t1",
-        activity_kind: "message",
-        snippet: "replied",
-        activity_at: "2026-07-02T10:00:00Z",
-      }),
-    ];
-    const merged = mergeTaskActivity(previous, incoming);
-    expect(merged).toHaveLength(1);
-    expect(merged[0].activity_kind).toBe("message");
-    expect(merged[0].activity_at).toBe("2026-07-02T10:00:00Z");
-  });
-
-  it("keeps the newer row when an older duplicate arrives out of order", () => {
-    const previous = [
-      activity({ task_id: "t1", activity_at: "2026-07-05T10:00:00Z" }),
-    ];
-    const incoming = [
-      activity({ task_id: "t1", activity_at: "2026-07-01T10:00:00Z" }),
-    ];
-    expect(mergeTaskActivity(previous, incoming)[0].activity_at).toBe(
-      "2026-07-05T10:00:00Z",
-    );
-  });
-
-  it("returns the previous page unchanged when there is nothing new", () => {
-    const previous = [activity({ task_id: "t1" })];
-    expect(mergeTaskActivity(previous, [])).toEqual(previous);
-  });
-
-  it("caps the merged result so a long session can't grow it unbounded", () => {
-    const previous = Array.from({ length: 300 }, (_, i) =>
-      activity({
-        task_id: `old-${i}`,
-        activity_at: `2026-06-01T${String(i % 24).padStart(2, "0")}:00:00Z`,
-      }),
-    );
-    const incoming = [
-      activity({ task_id: "newest", activity_at: "2026-07-05T10:00:00Z" }),
-    ];
-    const merged = mergeTaskActivity(previous, incoming);
-    expect(merged).toHaveLength(300);
-    expect(merged[0].task_id).toBe("newest");
-  });
-});

packages/core/src/canvas/taskActivity.ts

@@ -12,6 +12,7 @@ import type {
  */
 
 export interface TaskActivityItem {
+  id: string;
   taskId: string;
   taskTitle: string;
   /** Backend channel (tasks product Channel UUID); null for channel-less tasks. */
@@ -24,13 +25,15 @@ export interface TaskActivityItem {
   snippet: string;
   author: UserBasic | null;
   messageId: string | null;
+  isUnread: boolean;
 }
 
 /** Map activity DTOs (already newest-first from the backend) to feed items. */
 export function toTaskActivityItems(
   activity: readonly TaskActivity[],
 ): TaskActivityItem[] {
   return activity.map((row) => ({
+    id: row.id,
     taskId: row.task_id,
     taskTitle: row.task_title || "Untitled task",
     channelId: row.channel_id ?? null,
@@ -40,42 +43,6 @@ export function toTaskActivityItems(
     snippet: row.snippet,
     author: row.latest_author ?? null,
     messageId: row.latest_message_id ?? null,
+    isUnread: row.is_unread,
   }));
 }
-
-/** How many rows have activity after the viewer last opened the Activity page. */
-export function countUnseenActivity(
-  items: readonly TaskActivityItem[],
-  lastSeenAt: string | null,
-): number {
-  if (!lastSeenAt) return items.length;
-  return items.filter((item) => item.activityAt > lastSeenAt).length;
-}
-
-// Bounds the cache so a long-running session's accumulated feed can't grow
-// without limit.
-const MAX_CACHED_ACTIVITY = 300;
-
-/**
- * Fold a page of freshly-fetched activity into the previously cached set —
- * dedupe by task (newest activity wins), keep newest first. Lets repolls fetch
- * only what's new (via `since`) instead of re-fetching the whole top page every
- * time. A task whose activity advanced comes back on the next poll and replaces
- * its stale row rather than duplicating it.
- */
-export function mergeTaskActivity(
-  previous: readonly TaskActivity[],
-  incoming: readonly TaskActivity[],
-): TaskActivity[] {
-  if (incoming.length === 0) return [...previous];
-  const byTaskId = new Map(previous.map((row) => [row.task_id, row]));
-  for (const row of incoming) {
-    const existing = byTaskId.get(row.task_id);
-    if (!existing || row.activity_at > existing.activity_at) {
-      byTaskId.set(row.task_id, row);
-    }
-  }
-  return Array.from(byTaskId.values())
-    .sort((a, b) => (a.activity_at < b.activity_at ? 1 : -1))
-    .slice(0, MAX_CACHED_ACTIVITY);
-}

packages/shared/src/domain-types.ts

@@ -151,6 +151,7 @@ export type TaskActivityKind =
  * `TaskActivityDTO`.
  */
 export interface TaskActivity {
+  id: string;
   task_id: string;
   task_title: string;
   channel_id?: string | null;
@@ -160,6 +161,12 @@ export interface TaskActivity {
   snippet: string;
   latest_author?: UserBasic | null;
   latest_message_id?: string | null;
+  is_unread: boolean;
+}
+
+export interface TaskActivityPage {
+  results: TaskActivity[];
+  unread_count: number;
 }
 
 export type TaskRunStatus =

packages/ui/src/features/canvas/components/ActivityView.tsx

@@ -18,9 +18,9 @@ import { useCurrentUser } from "@posthog/ui/features/auth/useCurrentUser";
 import { getUserInitials } from "@posthog/ui/features/auth/userInitials";
 import { MentionText } from "@posthog/ui/features/canvas/components/MentionText";
 import { useChannels } from "@posthog/ui/features/canvas/hooks/useChannels";
+import { useMarkTaskActivityRead } from "@posthog/ui/features/canvas/hooks/useMarkTaskActivityRead";
 import { useTaskActivity } from "@posthog/ui/features/canvas/hooks/useTaskActivity";
 import { normalizeChannelName } from "@posthog/ui/features/canvas/hooks/useTaskChannels";
-import { useActivitySeenStore } from "@posthog/ui/features/canvas/stores/activitySeenStore";
 import { copyChannelLink } from "@posthog/ui/features/canvas/utils/copyChannelLink";
 import { userDisplayName } from "@posthog/ui/features/canvas/utils/userDisplay";
 import {
@@ -30,7 +30,7 @@ import {
 import { track } from "@posthog/ui/shell/analytics";
 import { Text } from "@radix-ui/themes";
 import type { ReactNode } from "react";
-import { useEffect, useMemo, useState } from "react";
+import { useEffect, useMemo } from "react";
 
 function ChannelSuffix({ channelName }: { channelName: string | null }) {
   if (!channelName) return null;
@@ -168,6 +168,7 @@ export function ActivityView() {
   const client = useOptionalAuthenticatedClient();
   const { data: currentUser } = useCurrentUser({ client });
   const { items, isLoading } = useTaskActivity();
+  const { mutate: markRead } = useMarkTaskActivityRead();
   // Items carry backend channel names only; the desktop folder-channel id
   // (needed for /website navigation and copy-link) is resolved here, where
   // the single useChannels subscription lives.
@@ -186,25 +187,16 @@ export function ActivityView() {
     channelName
       ? (folderIdByName.get(normalizeChannelName(channelName)) ?? null)
       : null;
-  const markSeen = useActivitySeenStore((s) => s.markSeen);
-  // Snapshot before marking seen so rows that were new on arrival keep their
-  // dot for this visit.
-  const [seenAtOpen] = useState(
-    () => useActivitySeenStore.getState().lastSeenAt,
-  );
-
   useEffect(() => {
     track(ANALYTICS_EVENTS.CHANNEL_ACTION, {
       action_type: "view_activity",
       surface: "activity",
     });
   }, []);
 
-  // Re-mark as items stream in so the badge stays cleared while reading.
-  // biome-ignore lint/correctness/useExhaustiveDependencies: re-run per new item
   useEffect(() => {
-    markSeen();
-  }, [markSeen, items.length]);
+    if (items.some((item) => item.isUnread)) markRead();
+  }, [items, markRead]);
 
   return (
     <div className="h-full overflow-y-auto bg-gray-1">
@@ -240,7 +232,7 @@ export function ActivityView() {
                   key={item.taskId}
                   item={item}
                   folderChannelId={folderChannelIdFor(item.channelName)}
-                  isNew={!seenAtOpen || item.activityAt > seenAtOpen}
+                  isNew={item.isUnread}
                   currentUserEmail={currentUser?.email}
                 />
               ))}

packages/ui/src/features/canvas/hooks/useMarkTaskActivityRead.ts

@@ -0,0 +1,32 @@
+import type { TaskActivityPage } from "@posthog/shared/domain-types";
+import { useOptionalAuthenticatedClient } from "@posthog/ui/features/auth/authClient";
+import { useMutation, useQueryClient } from "@tanstack/react-query";
+
+const TASK_ACTIVITY_QUERY_KEY = ["task-activity"] as const;
+
+export function useMarkTaskActivityRead() {
+  const client = useOptionalAuthenticatedClient();
+  const queryClient = useQueryClient();
+  return useMutation({
+    mutationFn: async () => {
+      if (!client) throw new Error("Not authenticated");
+      await client.markTaskActivityRead();
+    },
+    onSuccess: () => {
+      queryClient.setQueryData<TaskActivityPage>(
+        TASK_ACTIVITY_QUERY_KEY,
+        (page) =>
+          page
+            ? {
+                ...page,
+                unread_count: 0,
+                results: page.results.map((row) => ({
+                  ...row,
+                  is_unread: false,
+                })),
+              }
+            : page,
+      );
+    },
+  });
+}