fix(server): the latest task list survives a thread's activity window

apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts

@@ -1291,22 +1291,41 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
           )
         `;
       }
+      // Only the newest plan update outlives the window: it is what the
+      // activity popover renders as the task list.
+      for (const [id, createdAt] of [
+        ["plan-stale", "2026-02-01T00:00:00.000Z"],
+        ["plan-latest", "2026-02-02T00:00:00.000Z"],
+      ] as const) {
+        yield* sql`
+          INSERT INTO projection_thread_activities (
+            activity_id, thread_id, tone, kind, summary, payload_json, sequence, created_at
+          ) VALUES (
+            ${id}, 'thread-activity-cap', 'info', 'turn.plan.updated', 'Plan updated',
+            '{"plan":[{"step":"Ship","status":"inProgress"}]}', 0, ${createdAt}
+          )
+        `;
+      }
       const retainedSnapshot = yield* snapshotQuery.getSnapshot();
       const retainedDetail = yield* snapshotQuery.getThreadDetailById(
         ThreadId.make("thread-activity-cap"),
       );
       assert.equal(retainedDetail._tag, "Some");
       if (retainedDetail._tag === "Some") {
         const activities = retainedDetail.value.activities;
-        assert.equal(activities.length, MAX_THREAD_ACTIVITIES + 2);
+        assert.equal(activities.length, MAX_THREAD_ACTIVITIES + 3);
         assert.deepEqual(
-          activities.slice(0, 2).map((activity) => activity.id),
-          ["open-question", "open-approval"],
+          activities.slice(0, 3).map((activity) => activity.id),
+          ["plan-latest", "open-question", "open-approval"],
         );
         assert.equal(
           activities.some((activity) => activity.id === "closed-question"),
           false,
         );
+        assert.equal(
+          activities.some((activity) => activity.id === "plan-stale"),
+          false,
+        );
         assert.deepEqual(retainedSnapshot.threads[0]?.activities, activities);
       }
 

apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts

@@ -39,7 +39,7 @@ import * as Struct from "effect/Struct";
 import * as SqlClient from "effect/unstable/sql/SqlClient";
 import * as SqlSchema from "effect/unstable/sql/SqlSchema";
 import { MAX_THREAD_ACTIVITIES, MAX_THREAD_MESSAGES } from "@threadlines/shared/threadLimits";
-import { retainRecentActivitiesAndOpenRequests } from "@threadlines/shared/pendingRequests";
+import { retainThreadActivities } from "@threadlines/shared/threadActivityRetention";
 
 import {
   isPersistenceError,
@@ -719,6 +719,25 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
                 activity.activity_id DESC
             ) AS activity_rank
           FROM projection_thread_activities AS activity
+        ), latest_plan_activities AS (
+          -- Ranking only the plan rows keeps this a cheap side scan instead of
+          -- a second sort of the whole table.
+          SELECT activity_id
+          FROM (
+            SELECT
+              activity.activity_id,
+              ROW_NUMBER() OVER (
+                PARTITION BY activity.thread_id
+                ORDER BY
+                  activity.event_sequence DESC,
+                  activity.sequence DESC,
+                  activity.created_at DESC,
+                  activity.activity_id DESC
+              ) AS plan_rank
+            FROM projection_thread_activities AS activity
+            WHERE activity.kind = 'turn.plan.updated'
+          )
+          WHERE plan_rank = 1
         )
         SELECT
           activity_id AS "activityId",
@@ -737,6 +756,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
             'approval.requested', 'approval.resolved', 'provider.approval.respond.failed',
             'user-input.requested', 'user-input.resolved', 'provider.user-input.respond.failed'
           )
+          -- The newest plan update keeps the task list alive past the window
+          -- (see retainThreadActivities).
+          OR activity_id IN (SELECT activity_id FROM latest_plan_activities)
         ORDER BY
           thread_id ASC,
           event_sequence ASC,
@@ -1278,6 +1300,19 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
               'approval.requested', 'approval.resolved', 'provider.approval.respond.failed',
               'user-input.requested', 'user-input.resolved', 'provider.user-input.respond.failed'
             )
+          UNION
+          SELECT * FROM (
+            SELECT *
+            FROM projection_thread_activities
+            WHERE thread_id = ${threadId}
+              AND kind = 'turn.plan.updated'
+            ORDER BY
+              event_sequence DESC,
+              sequence DESC,
+              created_at DESC,
+              activity_id DESC
+            LIMIT 1
+          )
         )
         SELECT
           activity_id AS "activityId",
@@ -1757,7 +1792,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
                 messages: messagesByThread.get(row.threadId) ?? [],
                 proposedPlans: proposedPlansByThread.get(row.threadId) ?? [],
                 activities: dropStaleContextWindowActivities(
-                  retainRecentActivitiesAndOpenRequests(
+                  retainThreadActivities(
                     activitiesByThread.get(row.threadId) ?? [],
                     MAX_THREAD_ACTIVITIES,
                   ),
@@ -2692,10 +2727,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
         messages: messageRows.map(mapThreadMessageRow),
         proposedPlans: proposedPlanRows.map(mapProposedPlanRow),
         activities: dropStaleContextWindowActivities(
-          retainRecentActivitiesAndOpenRequests(
-            activityRows.map(mapThreadActivityRow),
-            MAX_THREAD_ACTIVITIES,
-          ),
+          retainThreadActivities(activityRows.map(mapThreadActivityRow), MAX_THREAD_ACTIVITIES),
         ),
         subagents: subagentRows.map(mapThreadSubagentRow),
         checkpoints: checkpointRows.map((row) => ({

apps/server/src/orchestration/projector.ts

@@ -14,7 +14,7 @@ import {
 } from "@threadlines/shared/threadLimits";
 import * as Effect from "effect/Effect";
 import * as Schema from "effect/Schema";
-import { retainRecentActivitiesAndOpenRequests } from "@threadlines/shared/pendingRequests";
+import { retainThreadActivities } from "@threadlines/shared/threadActivityRetention";
 
 import { toProjectorDecodeError, type OrchestrationProjectorDecodeError } from "./Errors.ts";
 import {
@@ -930,7 +930,7 @@ export function projectEvent(
           const existingActivity = thread.activities.find(
             (entry) => entry.id === payload.activity.id,
           );
-          const activities = retainRecentActivitiesAndOpenRequests(
+          const activities = retainThreadActivities(
             [
               ...thread.activities.filter((entry) => entry.id !== payload.activity.id),
               {

apps/server/src/persistence/Layers/StorageMaintenance.ts

@@ -130,7 +130,10 @@ const makeStorageMaintenance = Effect.gen(function* () {
   /**
    * Trims `thread.activity-appended` events to the newest
    * `MAX_THREAD_ACTIVITIES` per thread. The activity projection already
-   * trims to the same cap, so pruned events are invisible to rebuilds.
+   * trims to the same cap, so pruned events are invisible to rebuilds. The
+   * few older rows the projection keeps on purpose (open prompts, the latest
+   * plan update; see `retainThreadActivities`) can still be pruned here, so
+   * a full rebuild may drop them; the live projection table never does.
    */
   const pruneActivityEventsBeyondCap = (minAppliedSequence: number) =>
     Effect.gen(function* () {

apps/server/src/persistence/Services/StorageMaintenance.ts

@@ -9,8 +9,9 @@
  * - Command receipts older than the idempotency window.
  * - Events of deleted threads (their projections are already gone).
  * - `thread.activity-appended` events beyond the per-thread projection cap
- *   (`MAX_THREAD_ACTIVITIES`) — projections never surface more than the cap,
- *   so older activity events cannot influence a rebuild.
+ *   (`MAX_THREAD_ACTIVITIES`) — projections surface at most the cap plus a
+ *   few pinned rows (see `retainThreadActivities`), so older activity events
+ *   cannot influence a rebuild beyond those.
  *
  * Event pruning never crosses the minimum projector checkpoint, so a
  * lagging or newly added projector can still replay everything it has not

apps/web/src/store.ts

@@ -24,7 +24,7 @@ import { isProviderDriverKind, ProviderDriverKind } from "@threadlines/contracts
 import type { ThreadId, TurnId } from "@threadlines/contracts";
 import * as Schema from "effect/Schema";
 import { resolveModelSlugForProvider } from "@threadlines/shared/model";
-import { retainRecentActivitiesAndOpenRequests } from "@threadlines/shared/pendingRequests";
+import { retainThreadActivities } from "@threadlines/shared/threadActivityRetention";
 import {
   MAX_THREAD_ACTIVITIES,
   MAX_THREAD_CHECKPOINTS,
@@ -1146,10 +1146,7 @@ function upsertThreadActivity(
     (activities.length === 0 ||
       compareActivities(activities[activities.length - 1]!, activity) <= 0)
   ) {
-    return retainRecentActivitiesAndOpenRequests(
-      [...activities, nextActivity],
-      MAX_THREAD_ACTIVITIES,
-    );
+    return retainThreadActivities([...activities, nextActivity], MAX_THREAD_ACTIVITIES);
   }
 
   const nextActivities =
@@ -1160,10 +1157,7 @@ function upsertThreadActivity(
           nextActivity,
           ...activities.slice(existingIndex + 1),
         ];
-  return retainRecentActivitiesAndOpenRequests(
-    nextActivities.toSorted(compareActivities),
-    MAX_THREAD_ACTIVITIES,
-  );
+  return retainThreadActivities(nextActivities.toSorted(compareActivities), MAX_THREAD_ACTIVITIES);
 }
 
 function buildLatestTurn(params: {

packages/shared/package.json

@@ -163,6 +163,10 @@
       "types": "./src/pendingRequests.ts",
       "import": "./src/pendingRequests.ts"
     },
+    "./threadActivityRetention": {
+      "types": "./src/threadActivityRetention.ts",
+      "import": "./src/threadActivityRetention.ts"
+    },
     "./keybindings": {
       "types": "./src/keybindings.ts",
       "import": "./src/keybindings.ts"

packages/shared/src/pendingRequests.ts

@@ -144,21 +144,3 @@ export function countPendingUserInputs(
   }).length;
   return { pendingUserInputCount: requests.length, blockingUserInputCount };
 }
-
-/** Keep open prompts answerable even after their activity leaves the recent log. */
-export function retainRecentActivitiesAndOpenRequests<A extends PendingRequestActivityLike>(
-  orderedActivities: ReadonlyArray<A>,
-  recentLimit: number,
-): A[] {
-  if (orderedActivities.length <= recentLimit) return [...orderedActivities];
-  const openActivities = new Set(
-    [
-      ...collectOpenPendingRequests(orderedActivities, APPROVAL_ACTIVITY_KINDS),
-      ...collectOpenPendingRequests(orderedActivities, USER_INPUT_ACTIVITY_KINDS),
-    ].map(({ activity }) => activity),
-  );
-  const recentStart = orderedActivities.length - recentLimit;
-  return orderedActivities.filter(
-    (activity, index) => index >= recentStart || openActivities.has(activity),
-  );
-}

packages/shared/src/threadActivityRetention.test.ts

@@ -0,0 +1,28 @@
+import { describe, expect, it } from "vite-plus/test";
+
+import { retainThreadActivities } from "./threadActivityRetention.ts";
+
+describe("retainThreadActivities", () => {
+  it("keeps only the latest plan update from before the recent window", () => {
+    const activities = [
+      { id: "plan-1", kind: "turn.plan.updated", payload: { plan: [] } },
+      { id: "plan-2", kind: "turn.plan.updated", payload: { plan: [] } },
+      { id: "open", kind: "approval.requested", payload: { requestId: "r1" } },
+      { id: "old-note", kind: "runtime.note" },
+      { id: "recent-1", kind: "tool.started" },
+      { id: "recent-2", kind: "tool.completed" },
+    ];
+
+    expect(retainThreadActivities(activities, 2).map((activity) => activity.id)).toEqual([
+      "plan-2",
+      "open",
+      "recent-1",
+      "recent-2",
+    ]);
+  });
+
+  it("returns the log untouched while it fits the window", () => {
+    const activities = [{ id: "a", kind: "runtime.note" }];
+    expect(retainThreadActivities(activities, 5)).toEqual(activities);
+  });
+});

packages/shared/src/threadActivityRetention.ts

@@ -0,0 +1,45 @@
+/**
+ * Which activities a thread keeps once its log outgrows the recent window.
+ *
+ * Clients only ever see the newest `MAX_THREAD_ACTIVITIES` per thread, plus
+ * a few older rows that still drive UI state: open approvals and questions
+ * (so their prompts stay answerable) and the latest plan update (so the task
+ * list in the activity popover survives a long, chatty turn). The server
+ * projector, the snapshot SQL queries, and the web store all apply this same
+ * rule; if they disagree, a reload shows different state than the live feed.
+ */
+import {
+  APPROVAL_ACTIVITY_KINDS,
+  USER_INPUT_ACTIVITY_KINDS,
+  collectOpenPendingRequests,
+  type PendingRequestActivityLike,
+} from "./pendingRequests.ts";
+
+/** Activity kind carrying a thread's task list (TodoWrite / task tracker). */
+export const PLAN_ACTIVITY_KIND = "turn.plan.updated";
+
+/**
+ * Keeps the newest `recentLimit` of `orderedActivities` (callers pass them in
+ * thread order) plus any older activity the UI still needs: open prompts and
+ * the latest plan update.
+ */
+export function retainThreadActivities<A extends PendingRequestActivityLike>(
+  orderedActivities: ReadonlyArray<A>,
+  recentLimit: number,
+): A[] {
+  if (orderedActivities.length <= recentLimit) return [...orderedActivities];
+  const retained = new Set<A>(
+    [
+      ...collectOpenPendingRequests(orderedActivities, APPROVAL_ACTIVITY_KINDS),
+      ...collectOpenPendingRequests(orderedActivities, USER_INPUT_ACTIVITY_KINDS),
+    ].map(({ activity }) => activity),
+  );
+  const latestPlan = orderedActivities.findLast((activity) => activity.kind === PLAN_ACTIVITY_KIND);
+  if (latestPlan) {
+    retained.add(latestPlan);
+  }
+  const recentStart = orderedActivities.length - recentLimit;
+  return orderedActivities.filter(
+    (activity, index) => index >= recentStart || retained.has(activity),
+  );
+}