From 0031a8d6d82848360036cbc03e4412337c62a39b Mon Sep 17 00:00:00 2001 From: flamboh Date: Sun, 25 Oct 2026 22:34:21 +0000 Subject: [PATCH 1/2] fix(server): automatic compactions no longer mark read threads unread --- .../IdleCompactionService.test.ts | 85 +++++++++++++++++++ .../orchestration-v2/IdleCompactionService.ts | 49 ++++++++++- 2 files changed, 132 insertions(+), 2 deletions(-) create mode 100644 apps/server/src/orchestration-v2/IdleCompactionService.test.ts diff --git a/apps/server/src/orchestration-v2/IdleCompactionService.test.ts b/apps/server/src/orchestration-v2/IdleCompactionService.test.ts new file mode 100644 index 0000000000000..3d305d17af126 --- /dev/null +++ b/apps/server/src/orchestration-v2/IdleCompactionService.test.ts @@ -0,0 +1,85 @@ +import { expect, it } from "@effect/vitest"; +import type { + OrchestrationV2DomainEvent, + OrchestrationV2Run, + OrchestrationV2ServerCommand, + OrchestrationV2ThreadProjection, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; + +import * as ServerSettings from "../serverSettings.ts"; +import * as IdleCompactionService from "./IdleCompactionService.ts"; +import * as Orchestrator from "./Orchestrator.ts"; +import * as ProjectionStore from "./ProjectionStore.ts"; + +const at = (minute: number) => DateTime.makeUnsafe(Date.UTC(2026, 9, 19, 12, minute)); + +function compactedThread(threadId: string, lastVisitedAt: DateTime.Utc | null) { + const run = (ordinal: number, userMessageId: string, completedAt: DateTime.Utc) => + ({ + id: `${threadId}:run-${ordinal}`, + threadId, + ordinal, + userMessageId, + status: "completed", + completedAt, + }) as unknown as OrchestrationV2Run; + const previous = run(1, "message", at(0)); + const compaction = run(2, `idle-compact:${threadId}:${previous.id}`, at(58)); + return { + records: { + thread: { id: threadId, lastVisitedAt }, + runs: [previous, compaction], + } as unknown as OrchestrationV2ThreadProjection, + completed: { + type: "run.updated", + threadId, + payload: compaction, + } as unknown as OrchestrationV2DomainEvent, + }; +} + +it.effect("marks a caught-up thread read when its automatic compaction completes", () => + Effect.gen(function* () { + const threads = [ + compactedThread("unseen", null), + compactedThread("behind", at(-1)), + compactedThread("caught-up", at(1)), + ]; + const events = yield* Queue.unbounded(); + const commands = yield* Queue.unbounded(); + const service = yield* IdleCompactionService.make.pipe( + Effect.provide( + Layer.mergeAll( + Layer.mock(Orchestrator.OrchestratorV2)({ + streamDomainEvents: Stream.fromQueue(events), + dispatch: (command) => + Queue.offer(commands, command).pipe(Effect.as({ sequence: 1, storedEvents: [] })), + }), + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getThreadRecords: (threadId) => + Effect.succeed(threads.find((t) => t.records.thread.id === threadId)!.records), + }), + ServerSettings.layerTest(), + ), + ), + ); + yield* service.start(); + + yield* Queue.offerAll( + events, + threads.map((thread) => thread.completed), + ); + + expect(yield* Queue.take(commands)).toMatchObject({ + type: "thread.visit", + threadId: "caught-up", + visitedAt: DateTime.formatIso(at(58)), + opened: false, + }); + }).pipe(Effect.scoped), +); diff --git a/apps/server/src/orchestration-v2/IdleCompactionService.ts b/apps/server/src/orchestration-v2/IdleCompactionService.ts index 2b16cb8c1d73d..798c53d6fdb9b 100644 --- a/apps/server/src/orchestration-v2/IdleCompactionService.ts +++ b/apps/server/src/orchestration-v2/IdleCompactionService.ts @@ -2,6 +2,7 @@ import { CommandId, IDLE_COMPACTION_TEXT, MessageId, + type OrchestrationV2Run, resolveIdleCompaction, resolveProjectProviderInstanceEnabled, type RunId, @@ -19,6 +20,8 @@ import { idleCompactionWindow } from "./IdleCompaction.ts"; import * as Orchestrator from "./Orchestrator.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; +const IDLE_COMPACTION_MESSAGE_PREFIX = "idle-compact:"; + /** * Beta: when a run completes on a Claude instance set to compact before the * cache expires, waits until shortly before its prompt cache expires and sends @@ -69,7 +72,7 @@ export const make = Effect.gen(function* () { type: "message.dispatch", // One compaction per completed run, even across duplicate events. commandId: CommandId.make(`server:idle-compact:${threadId}:${runId}`), - messageId: MessageId.make(`idle-compact:${threadId}:${runId}`), + messageId: MessageId.make(`${IDLE_COMPACTION_MESSAGE_PREFIX}${threadId}:${runId}`), threadId, text: IDLE_COMPACTION_TEXT, attachments: [], @@ -87,13 +90,55 @@ export const make = Effect.gen(function* () { ), ); + const markSeenIfCaughtUp = Effect.fn("IdleCompactionService.markSeenIfCaughtUp")( + function* (run: OrchestrationV2Run) { + if (run.completedAt === null || !run.userMessageId.startsWith(IDLE_COMPACTION_MESSAGE_PREFIX)) + return; + const { thread, runs } = yield* projections.getThreadRecords(run.threadId, ["runs"]); + const previous = runs.reduce( + (latest, candidate) => + candidate.ordinal < run.ordinal && (latest === null || candidate.ordinal > latest.ordinal) + ? candidate + : latest, + null, + ); + if ( + thread.lastVisitedAt === null || + previous?.completedAt == null || + DateTime.isGreaterThan(previous.completedAt, thread.lastVisitedAt) + ) + return; + yield* orchestrator.dispatch({ + type: "thread.visit", + commandId: CommandId.make(`server:idle-compact-visit:${run.threadId}:${run.id}`), + threadId: run.threadId, + visitedAt: DateTime.formatIso(run.completedAt), + opened: false, + }); + }, + (effect, run) => + effect.pipe( + Effect.catch((cause) => + Effect.logWarning("idle compaction visit skipped", { threadId: run.threadId, cause }), + ), + ), + ); + const start = Effect.fn("IdleCompactionService.start")(function* () { // Keyed by thread: a newer completed run replaces the waiting timer. const timers = yield* FiberMap.make(); yield* forkParked( Stream.runForEach(orchestrator.streamDomainEvents, (event) => event.type === "run.updated" && event.payload.status === "completed" - ? FiberMap.run(timers, event.threadId, compactWhenDue(event.threadId, event.payload.id)) + ? markSeenIfCaughtUp(event.payload).pipe( + Effect.andThen( + FiberMap.run( + timers, + event.threadId, + compactWhenDue(event.threadId, event.payload.id), + ), + ), + ) : Effect.void, ), ); From c979424161e1b0c250f7a0481d47b329a2cf7fc0 Mon Sep 17 00:00:00 2001 From: flamboh Date: Mon, 26 Oct 2026 03:01:01 +0000 Subject: [PATCH 2/2] fix(server): keep an explicit mark unread after an automatic compaction --- apps/server/src/orchestration-v2/IdleCompactionService.test.ts | 2 ++ apps/server/src/orchestration-v2/IdleCompactionService.ts | 3 ++- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/IdleCompactionService.test.ts b/apps/server/src/orchestration-v2/IdleCompactionService.test.ts index 3d305d17af126..c03ed1ccf3769 100644 --- a/apps/server/src/orchestration-v2/IdleCompactionService.test.ts +++ b/apps/server/src/orchestration-v2/IdleCompactionService.test.ts @@ -26,6 +26,7 @@ function compactedThread(threadId: string, lastVisitedAt: DateTime.Utc | null) { ordinal, userMessageId, status: "completed", + requestedAt: DateTime.subtract(completedAt, { minutes: 1 }), completedAt, }) as unknown as OrchestrationV2Run; const previous = run(1, "message", at(0)); @@ -48,6 +49,7 @@ it.effect("marks a caught-up thread read when its automatic compaction completes const threads = [ compactedThread("unseen", null), compactedThread("behind", at(-1)), + compactedThread("marked-unread", DateTime.subtract(at(58), { milliseconds: 1 })), compactedThread("caught-up", at(1)), ]; const events = yield* Queue.unbounded(); diff --git a/apps/server/src/orchestration-v2/IdleCompactionService.ts b/apps/server/src/orchestration-v2/IdleCompactionService.ts index 798c53d6fdb9b..d0737390ed091 100644 --- a/apps/server/src/orchestration-v2/IdleCompactionService.ts +++ b/apps/server/src/orchestration-v2/IdleCompactionService.ts @@ -105,7 +105,8 @@ export const make = Effect.gen(function* () { if ( thread.lastVisitedAt === null || previous?.completedAt == null || - DateTime.isGreaterThan(previous.completedAt, thread.lastVisitedAt) + DateTime.isGreaterThan(previous.completedAt, thread.lastVisitedAt) || + !DateTime.isLessThan(thread.lastVisitedAt, run.requestedAt) ) return; yield* orchestrator.dispatch({