diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index daa6562c..da61c84a 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 8a00caea..f7ef3e0e 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 8ea815f7..909232ec 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 08b30fa7..4d93cba5 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 c285f445..553cdba3 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 75156e59..d43a5d65 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 09928718..a59be9d9 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 a7308159..19fc7763 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 00000000..9bdedecc --- /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 00000000..ac0cd169 --- /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), + ); +}