From 36ebc6ad9e06ae62279b296cc9e228ecd07822db Mon Sep 17 00:00:00 2001
From: badcuban <108198679+badcuban@users.noreply.github.com>
Date: Sun, 6 Sep 2026 04:56:10 -0400
Subject: [PATCH] fix(server): the latest task list survives a thread's
activity window
Clients only receive the newest 500 activities per thread. The task list in
the activity popover is itself an activity (`turn.plan.updated`), so a plan
written early in a long, chatty turn could fall out of the window and the
popover would show no tasks after a reload, while the live feed still had
them.
Move the retention rule into one shared helper, `retainThreadActivities`,
that keeps the recent window plus open prompts and the newest plan update,
and use it from the projector, both snapshot queries, and the web store so
a reload and the live feed agree. The SQL side ranks only the plan rows in
a side scan; on a 233k-row activity table that adds about 0.2 s to the full
snapshot query instead of the 2.7 s a second whole-table window cost.
---
.../Layers/ProjectionSnapshotQuery.test.ts | 25 +++++++++--
.../Layers/ProjectionSnapshotQuery.ts | 44 +++++++++++++++---
apps/server/src/orchestration/projector.ts | 4 +-
.../persistence/Layers/StorageMaintenance.ts | 5 ++-
.../Services/StorageMaintenance.ts | 5 ++-
apps/web/src/store.ts | 12 ++---
packages/shared/package.json | 4 ++
packages/shared/src/pendingRequests.ts | 18 --------
.../src/threadActivityRetention.test.ts | 28 ++++++++++++
.../shared/src/threadActivityRetention.ts | 45 +++++++++++++++++++
10 files changed, 149 insertions(+), 41 deletions(-)
create mode 100644 packages/shared/src/threadActivityRetention.test.ts
create mode 100644 packages/shared/src/threadActivityRetention.ts
diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts
index daa6562c2..da61c84a5 100644
--- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts
+++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts
@@ -1291,6 +1291,21 @@ 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"),
@@ -1298,15 +1313,19 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
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);
}
diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts
index 8a00caea0..f7ef3e0e1 100644
--- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts
+++ b/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) => ({
diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts
index 8ea815f7e..909232ecf 100644
--- a/apps/server/src/orchestration/projector.ts
+++ b/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),
{
diff --git a/apps/server/src/persistence/Layers/StorageMaintenance.ts b/apps/server/src/persistence/Layers/StorageMaintenance.ts
index 08b30fa72..4d93cba57 100644
--- a/apps/server/src/persistence/Layers/StorageMaintenance.ts
+++ b/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* () {
diff --git a/apps/server/src/persistence/Services/StorageMaintenance.ts b/apps/server/src/persistence/Services/StorageMaintenance.ts
index c285f445c..553cdba3a 100644
--- a/apps/server/src/persistence/Services/StorageMaintenance.ts
+++ b/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
diff --git a/apps/web/src/store.ts b/apps/web/src/store.ts
index 75156e593..d43a5d657 100644
--- a/apps/web/src/store.ts
+++ b/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: {
diff --git a/packages/shared/package.json b/packages/shared/package.json
index 099287188..a59be9d9a 100644
--- a/packages/shared/package.json
+++ b/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"
diff --git a/packages/shared/src/pendingRequests.ts b/packages/shared/src/pendingRequests.ts
index a73081596..19fc77637 100644
--- a/packages/shared/src/pendingRequests.ts
+++ b/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(
- orderedActivities: ReadonlyArray,
- 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),
- );
-}
diff --git a/packages/shared/src/threadActivityRetention.test.ts b/packages/shared/src/threadActivityRetention.test.ts
new file mode 100644
index 000000000..9bdedecca
--- /dev/null
+++ b/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);
+ });
+});
diff --git a/packages/shared/src/threadActivityRetention.ts b/packages/shared/src/threadActivityRetention.ts
new file mode 100644
index 000000000..ac0cd169b
--- /dev/null
+++ b/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(
+ orderedActivities: ReadonlyArray,
+ recentLimit: number,
+): A[] {
+ if (orderedActivities.length <= recentLimit) return [...orderedActivities];
+ const retained = new Set(
+ [
+ ...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),
+ );
+}