Merge branch 'main' into fix/704

Author: igennova

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

@@ -0,0 +1,101 @@
+import {
+  latestActivityForChannel,
+  unreadChannelIds,
+} from "@posthog/core/canvas/channelUnread";
+import type { MentionActivityItem } from "@posthog/core/canvas/mentionActivity";
+import { describe, expect, it } from "vitest";
+
+function mention(
+  overrides: Partial<MentionActivityItem> & { createdAt: string },
+): MentionActivityItem {
+  return {
+    messageId: `m-${overrides.createdAt}`,
+    taskId: "t1",
+    taskTitle: "Task",
+    channelId: "c1",
+    channelName: "mobile",
+    author: null,
+    content: "hey @adam",
+    ...overrides,
+  };
+}
+
+describe("unreadChannelIds", () => {
+  const cases: {
+    name: string;
+    lastSeen: Record<string, string>;
+    expected: string[];
+  }[] = [
+    {
+      name: "a channel never seen is unread",
+      lastSeen: {},
+      expected: ["c1"],
+    },
+    {
+      name: "activity newer than the last visit is unread",
+      lastSeen: { c1: "2026-07-16T09:00:00.000Z" },
+      expected: ["c1"],
+    },
+    {
+      name: "activity older than the last visit is read",
+      lastSeen: { c1: "2026-07-16T11:00:00.000Z" },
+      expected: [],
+    },
+    {
+      name: "activity exactly at the last visit is read",
+      lastSeen: { c1: "2026-07-16T10:00:00.000Z" },
+      expected: [],
+    },
+  ];
+  it.each(cases)("$name", ({ lastSeen, expected }) => {
+    const items = [mention({ createdAt: "2026-07-16T10:00:00.000Z" })];
+    expect([...unreadChannelIds(items, lastSeen)]).toEqual(expected);
+  });
+
+  it("compares each channel against its own last visit", () => {
+    const items = [
+      mention({ channelId: "c1", createdAt: "2026-07-16T10:00:00.000Z" }),
+      mention({ channelId: "c2", createdAt: "2026-07-16T10:00:00.000Z" }),
+    ];
+    const unread = unreadChannelIds(items, {
+      c1: "2026-07-16T11:00:00.000Z",
+      c2: "2026-07-16T09:00:00.000Z",
+    });
+    expect([...unread]).toEqual(["c2"]);
+  });
+
+  it("uses the newest item in a channel, whatever the order", () => {
+    const items = [
+      mention({ messageId: "old", createdAt: "2026-07-16T08:00:00.000Z" }),
+      mention({ messageId: "new", createdAt: "2026-07-16T12:00:00.000Z" }),
+    ];
+    expect([
+      ...unreadChannelIds(items, { c1: "2026-07-16T10:00:00.000Z" }),
+    ]).toEqual(["c1"]);
+  });
+
+  it("ignores channel-less mentions", () => {
+    const items = [
+      mention({ channelId: null, createdAt: "2026-07-16T10:00Z" }),
+    ];
+    expect([...unreadChannelIds(items, {})]).toEqual([]);
+  });
+});
+
+describe("latestActivityForChannel", () => {
+  it("returns the newest timestamp for that channel only", () => {
+    const items = [
+      mention({ channelId: "c1", createdAt: "2026-07-16T08:00:00.000Z" }),
+      mention({ channelId: "c1", createdAt: "2026-07-16T12:00:00.000Z" }),
+      mention({ channelId: "c2", createdAt: "2026-07-16T13:00:00.000Z" }),
+    ];
+    expect(latestActivityForChannel(items, "c1")).toBe(
+      "2026-07-16T12:00:00.000Z",
+    );
+  });
+
+  it("is undefined for a channel with no activity, or no channel", () => {
+    expect(latestActivityForChannel([], "c1")).toBeUndefined();
+    expect(latestActivityForChannel([], undefined)).toBeUndefined();
+  });
+});

packages/core/src/canvas/channelUnread.ts

@@ -0,0 +1,66 @@
+import type { MentionActivityItem } from "@posthog/core/canvas/mentionActivity";
+
+/**
+ * Which channels have activity the viewer hasn't seen — the signal behind the
+ * sidebar's bold channel names.
+ *
+ * "Activity" is currently an @-mention: that's the only cross-channel,
+ * all-users feed the client has (the mentions index), and it's what
+ * "notification" means elsewhere in the app. The backend exposes no per-channel
+ * activity timestamp, so a broader "any new message" signal would mean polling
+ * every user's full task list — the app's heaviest poll, deliberately retired.
+ * If that timestamp lands, only `latestActivityByChannel` changes shape; the
+ * unread comparison and the seen bookkeeping stay as they are.
+ *
+ * Keyed by backend channel id rather than name, so renaming a channel doesn't
+ * silently mark it unread again.
+ */
+
+/** Newest activity per channel id. Ignores items with no channel. */
+export function latestActivityByChannel(
+  items: readonly MentionActivityItem[],
+): Map<string, string> {
+  const latest = new Map<string, string>();
+  for (const item of items) {
+    if (!item.channelId) continue;
+    const current = latest.get(item.channelId);
+    if (!current || item.createdAt > current) {
+      latest.set(item.channelId, item.createdAt);
+    }
+  }
+  return latest;
+}
+
+/**
+ * Channel ids whose newest activity postdates the viewer's last visit. A
+ * channel never visited is unread as soon as it has any activity.
+ */
+export function unreadChannelIds(
+  items: readonly MentionActivityItem[],
+  lastSeenByChannel: Readonly<Record<string, string>>,
+): Set<string> {
+  const unread = new Set<string>();
+  for (const [channelId, activityAt] of latestActivityByChannel(items)) {
+    const seenAt = lastSeenByChannel[channelId];
+    if (!seenAt || activityAt > seenAt) unread.add(channelId);
+  }
+  return unread;
+}
+
+/**
+ * The newest activity in one channel, for stamping it seen while it's open.
+ * Scans for the one channel rather than reusing `latestActivityByChannel`,
+ * which would build (and throw away) a map of every other channel to answer.
+ */
+export function latestActivityForChannel(
+  items: readonly MentionActivityItem[],
+  channelId: string | undefined,
+): string | undefined {
+  if (!channelId) return undefined;
+  let latest: string | undefined;
+  for (const item of items) {
+    if (item.channelId !== channelId) continue;
+    if (!latest || item.createdAt > latest) latest = item.createdAt;
+  }
+  return latest;
+}

packages/core/src/sessions/contextUsage.test.ts

@@ -17,6 +17,31 @@ function usageUpdateEvent(used: number, size: number): AcpMessage {
   };
 }
 
+function costUsageUpdateEvent(
+  used: number,
+  size: number,
+  amount: number,
+  currency = "USD",
+): AcpMessage {
+  return {
+    type: "acp_message",
+    ts: 1,
+    message: {
+      jsonrpc: "2.0",
+      method: "session/update",
+      params: {
+        sessionId: "s1",
+        update: {
+          sessionUpdate: "usage_update",
+          used,
+          size,
+          cost: { amount, currency },
+        },
+      },
+    },
+  };
+}
+
 function sizelessUsageUpdateEvent(used: number): AcpMessage {
   return {
     type: "acp_message",
@@ -119,6 +144,30 @@ describe("extractContextUsage", () => {
     expect(result?.breakdown?.conversation).toBe(45_500);
   });
 
+  it("reports null cost when no update carries a cost", () => {
+    const result = extractContextUsage([usageUpdateEvent(50_000, 200_000)]);
+    expect(result?.cost).toBeNull();
+  });
+
+  it("surfaces the cost from a single turn", () => {
+    const result = extractContextUsage([
+      costUsageUpdateEvent(50_000, 200_000, 0.42),
+    ]);
+    expect(result?.cost).toEqual({ amount: 0.42, currency: "USD" });
+  });
+
+  it("sums cost across turns since each result reports only its own spend", () => {
+    const result = extractContextUsage([
+      costUsageUpdateEvent(40_000, 200_000, 0.4),
+      costUsageUpdateEvent(90_000, 200_000, 0.35),
+      costUsageUpdateEvent(120_000, 200_000, 0.25),
+    ]);
+    // Context occupancy tracks the newest turn; cost accrues across all of them.
+    expect(result?.used).toBe(120_000);
+    expect(result?.cost?.amount).toBeCloseTo(1.0, 10);
+    expect(result?.cost?.currency).toBe("USD");
+  });
+
   it("tolerates the double-underscore method prefix from extNotification", () => {
     const result = extractContextUsage([
       usageUpdateEvent(50_000, 200_000),
@@ -180,6 +229,28 @@ describe("createContextUsageTracker", () => {
     expect(tracker.update([earlier])?.used).toBe(50_000);
   });
 
+  it("accumulates cost only over newly appended turns", () => {
+    const tracker = createContextUsageTracker();
+    const first = costUsageUpdateEvent(40_000, 200_000, 0.4);
+
+    expect(tracker.update([first])?.cost?.amount).toBeCloseTo(0.4, 10);
+
+    const result = tracker.update([
+      first,
+      costUsageUpdateEvent(90_000, 200_000, 0.35),
+    ]);
+    expect(result?.cost?.amount).toBeCloseTo(0.75, 10);
+  });
+
+  it("matches the batch extractor for a cost-bearing log", () => {
+    const tracker = createContextUsageTracker();
+    const events = [
+      costUsageUpdateEvent(40_000, 200_000, 0.4),
+      costUsageUpdateEvent(90_000, 200_000, 0.35),
+    ];
+    expect(tracker.update(events)).toEqual(extractContextUsage(events));
+  });
+
   it("rebuilds when the tail changes at the same length", () => {
     const tracker = createContextUsageTracker();
     const first = usageUpdateEvent(50_000, 200_000);

packages/core/src/sessions/contextUsage.ts

@@ -15,18 +15,28 @@ export interface ContextUsage {
   used: number;
   size: number;
   percentage: number;
+  /** Cumulative estimated session cost, summed across turns; `null` if none reported (e.g. codex). */
   cost: { amount: number; currency: string } | null;
   breakdown: ContextBreakdown | null;
 }
 
-type ContextUsageAggregate = Omit<ContextUsage, "breakdown">;
+type ContextUsageAggregate = Omit<ContextUsage, "breakdown" | "cost">;
 
 export function extractContextUsage(events: AcpMessage[]): ContextUsage | null {
   let aggregate: ContextUsageAggregate | null = null;
   let breakdown: ContextBreakdown | null = null;
+  let costAmount: number | null = null;
+  let costCurrency = "USD";
 
+  // Cost sums over every turn, so this can't early-break once the newest
+  // aggregate/breakdown is found — it walks the full log.
   for (let i = events.length - 1; i >= 0; i--) {
     const msg = events[i].message;
+    const cost = extractCost(msg);
+    if (cost) {
+      costAmount = (costAmount ?? 0) + cost.amount;
+      costCurrency = cost.currency;
+    }
     if (!aggregate) {
       aggregate = extractAggregate(msg);
     } else if (aggregate.size <= 0) {
@@ -37,36 +47,55 @@ export function extractContextUsage(events: AcpMessage[]): ContextUsage | null {
     if (!breakdown) {
       breakdown = extractBreakdown(msg);
     }
-    if (aggregate && aggregate.size > 0 && breakdown) break;
   }
 
   if (!aggregate) return null;
-  return { ...aggregate, breakdown };
+  return { ...aggregate, cost: toCost(costAmount, costCurrency), breakdown };
 }
 
 interface ContextUsageState {
   aggregate: ContextUsageAggregate | null;
+  costAmount: number | null;
+  costCurrency: string;
   breakdown: ContextBreakdown | null;
 }
 
 export function createContextUsageTracker() {
   return createAppendOnlyTracker<ContextUsageState, ContextUsage | null>({
-    init: () => ({ aggregate: null, breakdown: null }),
+    init: () => ({
+      aggregate: null,
+      costAmount: null,
+      costCurrency: "USD",
+      breakdown: null,
+    }),
     processEvent: (state, event) => {
       const msg = event.message;
       const next = extractAggregate(msg);
       if (next) {
         state.aggregate = withCarriedSize(next, state.aggregate);
       }
+      const cost = extractCost(msg);
+      if (cost) {
+        state.costAmount = (state.costAmount ?? 0) + cost.amount;
+        state.costCurrency = cost.currency;
+      }
       state.breakdown = extractBreakdown(msg) ?? state.breakdown;
     },
     getResult: (state) =>
       state.aggregate
-        ? { ...state.aggregate, breakdown: state.breakdown }
+        ? {
+            ...state.aggregate,
+            cost: toCost(state.costAmount, state.costCurrency),
+            breakdown: state.breakdown,
+          }
         : null,
   });
 }
 
+function toCost(amount: number | null, currency: string): ContextUsage["cost"] {
+  return amount != null ? { amount, currency } : null;
+}
+
 /**
  * An update that omits `size` must not wipe a previously known context window
  * (codex reports `modelContextWindow` intermittently), so keep the last known
@@ -116,11 +145,38 @@ function extractAggregate(
       const size = typeof update.size === "number" ? update.size : 0;
       const percentage =
         size > 0 ? Math.min(100, Math.round((update.used / size) * 100)) : 0;
+      return { used: update.used, size, percentage };
+    }
+  }
+  return null;
+}
+
+function extractCost(
+  msg: AcpMessage["message"],
+): { amount: number; currency: string } | null {
+  if (
+    "method" in msg &&
+    msg.method === "session/update" &&
+    !("id" in msg) &&
+    "params" in msg
+  ) {
+    const params = msg.params as
+      | {
+          update?: {
+            sessionUpdate?: string;
+            cost?: { amount: number; currency: string } | null;
+          };
+        }
+      | undefined;
+    const update = params?.update;
+    if (
+      update?.sessionUpdate === "usage_update" &&
+      update.cost &&
+      typeof update.cost.amount === "number"
+    ) {
       return {
-        used: update.used,
-        size,
-        percentage,
-        cost: update.cost ?? null,
+        amount: update.cost.amount,
+        currency: update.cost.currency ?? "USD",
       };
     }
   }