From 0125c6f600f1842283a375d8e14fbed8bc527d16 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Fri, 7 Aug 2026 13:35:43 -0700 Subject: [PATCH] feat(server): sync thread read status across clients --- .../features/threads/ThreadRouteScreen.tsx | 28 +++ apps/mobile/src/state/threads.ts | 4 +- .../src/environment/ServerEnvironment.ts | 1 + .../Layers/ProjectionPipeline.test.ts | 34 +++ .../Layers/ProjectionPipeline.ts | 13 + .../Layers/ProjectionSnapshotQuery.test.ts | 2 + .../Layers/ProjectionSnapshotQuery.ts | 10 + apps/server/src/orchestration/Schemas.ts | 2 + apps/server/src/orchestration/decider.ts | 63 +++++ .../orchestration/decider.viewStatus.test.ts | 228 ++++++++++++++++++ .../src/orchestration/projector.test.ts | 1 + apps/server/src/orchestration/projector.ts | 17 ++ .../Layers/ProjectionRepositories.test.ts | 2 + .../persistence/Layers/ProjectionThreads.ts | 5 + apps/server/src/persistence/Migrations.ts | 2 + .../039_ProjectionThreadsLastViewedAt.test.ts | 49 ++++ .../039_ProjectionThreadsLastViewedAt.ts | 22 ++ .../persistence/Services/ProjectionThreads.ts | 1 + apps/web/src/components/ChatView.tsx | 80 +++--- apps/web/src/components/LegacySidebar.tsx | 111 ++++----- apps/web/src/components/Sidebar.logic.test.ts | 62 ++++- apps/web/src/components/Sidebar.logic.ts | 28 ++- apps/web/src/components/Sidebar.tsx | 103 ++++++-- .../src/components/ThreadStatusIndicators.tsx | 18 +- .../components/threadActionMenu.logic.test.ts | 23 +- .../src/components/threadActionMenu.logic.ts | 6 +- apps/web/src/hooks/useThreadActionMenu.ts | 17 +- apps/web/src/hooks/useThreadActions.ts | 36 ++- apps/web/src/state/entities.ts | 8 + apps/web/src/state/threads.ts | 4 +- apps/web/src/uiStateStore.test.ts | 42 +--- apps/web/src/uiStateStore.ts | 74 ------ .../src/operations/commands.test.ts | 37 +++ .../client-runtime/src/operations/commands.ts | 22 ++ .../src/state/threadCommands.ts | 69 +++++- .../client-runtime/src/state/threadDetail.ts | 1 + .../src/state/threadReducer.test.ts | 19 ++ .../client-runtime/src/state/threadReducer.ts | 7 + .../client-runtime/src/state/threadSettled.ts | 4 +- packages/contracts/src/environment.ts | 3 + packages/contracts/src/orchestration.ts | 41 ++++ 41 files changed, 1027 insertions(+), 272 deletions(-) create mode 100644 apps/server/src/orchestration/decider.viewStatus.test.ts create mode 100644 apps/server/src/persistence/Migrations/039_ProjectionThreadsLastViewedAt.test.ts create mode 100644 apps/server/src/persistence/Migrations/039_ProjectionThreadsLastViewedAt.ts diff --git a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx index d7754b7d78f..cc859e0c3e9 100644 --- a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx @@ -63,6 +63,7 @@ import { useSelectedThreadRequests } from "../../state/use-selected-thread-reque import { useSelectedThreadWorktree } from "../../state/use-selected-thread-worktree"; import { useThreadComposerState } from "../../state/use-thread-composer-state"; import { threadEnvironment } from "../../state/threads"; +import { useEnvironmentServerConfig } from "../../state/entities"; import { projectThreadContentPresentation } from "./threadContentPresentation"; import { useAdaptiveWorkspaceLayout, @@ -214,6 +215,33 @@ function ThreadRouteContent( const gitActions = useSelectedThreadGitActions(); const requests = useSelectedThreadRequests(); const interruptThreadTurn = useAtomCommand(threadEnvironment.interruptTurn, "thread interrupt"); + const markThreadViewed = useAtomCommand(threadEnvironment.markViewed, { reportFailure: false }); + const selectedServerConfig = useEnvironmentServerConfig(selectedThread?.environmentId ?? null); + useFocusEffect( + useCallback(() => { + const thread = selectedThread; + const completedAt = thread?.latestTurn?.completedAt; + if ( + thread == null || + completedAt == null || + selectedServerConfig?.environment.capabilities.threadViewStatus !== true + ) + return; + void markThreadViewed({ + environmentId: thread.environmentId, + input: { + threadId: thread.id, + viewedAt: completedAt, + }, + }); + }, [ + markThreadViewed, + selectedServerConfig?.environment.capabilities.threadViewStatus, + selectedThread?.environmentId, + selectedThread?.id, + selectedThread?.latestTurn?.completedAt, + ]), + ); const navigation = useNavigation(); const params = props.route.params; const environmentIdRaw = firstRouteParam(params.environmentId); diff --git a/apps/mobile/src/state/threads.ts b/apps/mobile/src/state/threads.ts index 7f247123051..6b08b44c0f7 100644 --- a/apps/mobile/src/state/threads.ts +++ b/apps/mobile/src/state/threads.ts @@ -15,7 +15,6 @@ import { environmentCatalog } from "../connection/catalog"; import { connectionAtomRuntime } from "../connection/runtime"; import { environmentSnapshotAtom } from "./shell"; -export const threadEnvironment = createThreadEnvironmentAtoms(connectionAtomRuntime); export const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime); export const environmentThreadDetails = createEnvironmentThreadDetailAtoms( environmentThreads.stateAtom, @@ -24,6 +23,9 @@ export const environmentThreadShells = createEnvironmentThreadShellAtoms({ catalogValueAtom: environmentCatalog.catalogValueAtom, snapshotAtom: environmentSnapshotAtom, }); +export const threadEnvironment = createThreadEnvironmentAtoms(connectionAtomRuntime, { + threadShellAtom: environmentThreadShells.threadShellAtom, +}); const EMPTY_THREAD_STATE_ATOM = Atom.make(AsyncResult.success(EMPTY_ENVIRONMENT_THREAD_STATE)).pipe( Atom.withLabel("mobile-environment-thread:empty"), diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index c697b4bd98f..e66ed6ed231 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -147,6 +147,7 @@ export const make = Effect.gen(function* () { threadSnooze: true, threadPinning: true, threadPinReorder: true, + threadViewStatus: true, threadTitleRegeneration: true, ...(serverSelfUpdate === null ? {} : { serverSelfUpdate }), ...(serverSelfUpdate === "boot-service" ? { serverSelfUpdateProgress: true } : {}), diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 8e65295b1ba..1b41e6371e9 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -237,6 +237,40 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { WHERE thread_id = 'thread-1' `; assert.deepEqual(unsettledRows, [{ settledOverride: "active", settledAt: null }]); + + yield* eventStore.append({ + type: "thread.view-status-updated", + eventId: EventId.make("evt-viewed-1"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + occurredAt: "2026-01-01T00:00:03.000Z", + commandId: CommandId.make("cmd-viewed-1"), + causationEventId: null, + correlationId: CommandId.make("cmd-viewed-1"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-1"), + lastViewedAt: "2026-01-01T00:00:03.000Z", + }, + }); + yield* projectionPipeline.bootstrap; + + const viewRows = yield* sql<{ + readonly lastViewedAt: string | null; + readonly updatedAt: string; + }>` + SELECT + last_viewed_at AS "lastViewedAt", + updated_at AS "updatedAt" + FROM projection_threads + WHERE thread_id = 'thread-1' + `; + assert.deepEqual(viewRows, [ + { + lastViewedAt: "2026-01-01T00:00:03.000Z", + updatedAt: "2026-01-01T00:00:02.000Z", + }, + ]); }), ); }); diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 38a70240d97..d98fd767e29 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -609,6 +609,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti archivedAt: null, settledOverride: null, settledAt: null, + lastViewedAt: event.payload.lastViewedAt ?? null, snoozedUntil: null, snoozedAt: null, pinnedAt: null, @@ -655,6 +656,18 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti return; } + case "thread.view-status-updated": { + const existingRow = yield* projectionThreadRepository.getById({ + threadId: event.payload.threadId, + }); + if (Option.isNone(existingRow)) return; + yield* projectionThreadRepository.upsert({ + ...existingRow.value, + lastViewedAt: event.payload.lastViewedAt, + }); + return; + } + case "thread.settled": { const existingRow = yield* projectionThreadRepository.getById({ threadId: event.payload.threadId, diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index c89124751b5..72841cfff00 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -319,6 +319,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { archivedAt: null, settledOverride: null, settledAt: null, + lastViewedAt: null, snoozedUntil: null, snoozedAt: null, pinnedAt: "2026-02-24T00:00:01.000Z", @@ -436,6 +437,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { archivedAt: null, settledOverride: null, settledAt: null, + lastViewedAt: null, snoozedUntil: null, snoozedAt: null, pinnedAt: "2026-02-24T00:00:01.000Z", diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index e744574a73c..a9816d5f53a 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -420,6 +420,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { archived_at AS "archivedAt", settled_override AS "settledOverride", settled_at AS "settledAt", + last_viewed_at AS "lastViewedAt", snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", pinned_at AS "pinnedAt", @@ -456,6 +457,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { archived_at AS "archivedAt", settled_override AS "settledOverride", settled_at AS "settledAt", + last_viewed_at AS "lastViewedAt", snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", pinned_at AS "pinnedAt", @@ -494,6 +496,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { archived_at AS "archivedAt", settled_override AS "settledOverride", settled_at AS "settledAt", + last_viewed_at AS "lastViewedAt", snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", pinned_at AS "pinnedAt", @@ -932,6 +935,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { archived_at AS "archivedAt", settled_override AS "settledOverride", settled_at AS "settledAt", + last_viewed_at AS "lastViewedAt", snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", pinned_at AS "pinnedAt", @@ -1563,6 +1567,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { archivedAt: row.archivedAt, settledOverride: row.settledOverride, settledAt: row.settledAt, + lastViewedAt: row.lastViewedAt, snoozedUntil: row.snoozedUntil, snoozedAt: row.snoozedAt, pinnedAt: row.pinnedAt, @@ -1768,6 +1773,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { archivedAt: row.archivedAt, settledOverride: row.settledOverride, settledAt: row.settledAt, + lastViewedAt: row.lastViewedAt, snoozedUntil: row.snoozedUntil, snoozedAt: row.snoozedAt, pinnedAt: row.pinnedAt, @@ -1904,6 +1910,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { archivedAt: row.archivedAt, settledOverride: row.settledOverride, settledAt: row.settledAt, + lastViewedAt: row.lastViewedAt, snoozedUntil: row.snoozedUntil, snoozedAt: row.snoozedAt, pinnedAt: row.pinnedAt, @@ -2049,6 +2056,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { archivedAt: row.archivedAt, settledOverride: row.settledOverride, settledAt: row.settledAt, + lastViewedAt: row.lastViewedAt, snoozedUntil: row.snoozedUntil, snoozedAt: row.snoozedAt, pinnedAt: row.pinnedAt, @@ -2326,6 +2334,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { archivedAt: threadRow.value.archivedAt, settledOverride: threadRow.value.settledOverride, settledAt: threadRow.value.settledAt, + lastViewedAt: threadRow.value.lastViewedAt, snoozedUntil: threadRow.value.snoozedUntil, snoozedAt: threadRow.value.snoozedAt, pinnedAt: threadRow.value.pinnedAt, @@ -2447,6 +2456,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { archivedAt: threadRow.value.archivedAt, settledOverride: threadRow.value.settledOverride, settledAt: threadRow.value.settledAt, + lastViewedAt: threadRow.value.lastViewedAt, snoozedUntil: threadRow.value.snoozedUntil, snoozedAt: threadRow.value.snoozedAt, pinnedAt: threadRow.value.pinnedAt, diff --git a/apps/server/src/orchestration/Schemas.ts b/apps/server/src/orchestration/Schemas.ts index 7e866cf8959..e52a55af805 100644 --- a/apps/server/src/orchestration/Schemas.ts +++ b/apps/server/src/orchestration/Schemas.ts @@ -10,6 +10,7 @@ import { ThreadInteractionModeSetPayload as ContractsThreadInteractionModeSetPayloadSchema, ThreadDeletedPayload as ContractsThreadDeletedPayloadSchema, ThreadUnarchivedPayload as ContractsThreadUnarchivedPayloadSchema, + ThreadViewStatusUpdatedPayload as ContractsThreadViewStatusUpdatedPayloadSchema, ThreadUnsettledPayload as ContractsThreadUnsettledPayloadSchema, ThreadSnoozedPayload as ContractsThreadSnoozedPayloadSchema, ThreadUnsnoozedPayload as ContractsThreadUnsnoozedPayloadSchema, @@ -42,6 +43,7 @@ export const ThreadRuntimeModeSetPayload = ContractsThreadRuntimeModeSetPayloadS export const ThreadInteractionModeSetPayload = ContractsThreadInteractionModeSetPayloadSchema; export const ThreadDeletedPayload = ContractsThreadDeletedPayloadSchema; export const ThreadUnarchivedPayload = ContractsThreadUnarchivedPayloadSchema; +export const ThreadViewStatusUpdatedPayload = ContractsThreadViewStatusUpdatedPayloadSchema; export const ThreadUnsettledPayload = ContractsThreadUnsettledPayloadSchema; export const ThreadSnoozedPayload = ContractsThreadSnoozedPayloadSchema; export const ThreadUnsnoozedPayload = ContractsThreadUnsnoozedPayloadSchema; diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 3de2592c884..ffcb6f0ff19 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -372,6 +372,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" interactionMode: command.interactionMode, branch: command.branch, worktreePath: command.worktreePath, + lastViewedAt: command.createdAt, createdAt: command.createdAt, updatedAt: command.createdAt, }, @@ -445,6 +446,68 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }; } + case "thread.mark-viewed": { + const thread = yield* requireThread({ readModel, command, threadId: command.threadId }); + const previousViewedAt = thread.lastViewedAt ?? null; + if ( + previousViewedAt !== command.expectedLastViewedAt && + previousViewedAt !== command.supersededViewedAt + ) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `thread ${command.threadId} view status changed after this view was queued`, + }); + } + const occurredAt = yield* nowIso; + const lastViewedAt = + previousViewedAt != null && Date.parse(previousViewedAt) > Date.parse(command.viewedAt) + ? previousViewedAt + : command.viewedAt; + return { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt, + commandId: command.commandId, + })), + type: "thread.view-status-updated", + payload: { threadId: command.threadId, lastViewedAt }, + }; + } + + case "thread.mark-unread": { + const thread = yield* requireThread({ readModel, command, threadId: command.threadId }); + const completedAt = thread.latestTurn?.completedAt; + if (completedAt == null) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `thread ${command.threadId} has no completed turn to mark unread`, + }); + } + const occurredAt = yield* nowIso; + const previousViewedAt = thread.lastViewedAt; + // Repeated unread actions still move the optimistic view token. + const unreadFrom = + previousViewedAt != null && Date.parse(previousViewedAt) < Date.parse(completedAt) + ? previousViewedAt + : completedAt; + return { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt, + commandId: command.commandId, + })), + type: "thread.view-status-updated", + payload: { + threadId: command.threadId, + lastViewedAt: DateTime.formatIso( + DateTime.subtract(DateTime.makeUnsafe(unreadFrom), { milliseconds: 1 }), + ), + }, + }; + } + case "thread.settle": { const thread = yield* requireThreadNotArchived({ readModel, diff --git a/apps/server/src/orchestration/decider.viewStatus.test.ts b/apps/server/src/orchestration/decider.viewStatus.test.ts new file mode 100644 index 00000000000..a324fbdd438 --- /dev/null +++ b/apps/server/src/orchestration/decider.viewStatus.test.ts @@ -0,0 +1,228 @@ +import { + CommandId, + ProjectId, + ProviderInstanceId, + ThreadId, + TurnId, + type OrchestrationReadModel, +} from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; + +import { decideOrchestrationCommand } from "./decider.ts"; + +const COMPLETED_AT = "1970-01-01T00:00:00.000Z"; + +function makeReadModel(completedAt: string | null = COMPLETED_AT): OrchestrationReadModel { + return { + snapshotSequence: 0, + projects: [], + threads: [ + { + id: ThreadId.make("thread-1"), + projectId: ProjectId.make("project-1"), + title: "Thread", + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + latestTurn: { + turnId: TurnId.make("turn-1"), + state: "completed", + requestedAt: COMPLETED_AT, + startedAt: COMPLETED_AT, + completedAt, + assistantMessageId: null, + }, + createdAt: COMPLETED_AT, + updatedAt: COMPLETED_AT, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastViewedAt: null, + deletedAt: null, + messages: [], + proposedPlans: [], + activities: [], + checkpoints: [], + session: null, + }, + ], + updatedAt: COMPLETED_AT, + }; +} + +it.layer(NodeServices.layer)("thread view-status decider", (it) => { + it.effect("uses server-owned timestamps when a thread is viewed", () => + Effect.gen(function* () { + const event = yield* decideOrchestrationCommand({ + command: { + type: "thread.mark-viewed", + commandId: CommandId.make("cmd-viewed"), + threadId: ThreadId.make("thread-1"), + viewedAt: COMPLETED_AT, + expectedLastViewedAt: null, + supersededViewedAt: null, + }, + readModel: makeReadModel(), + }); + const events = Array.isArray(event) ? event : [event]; + expect(events).toHaveLength(1); + expect(events[0]?.type).toBe("thread.view-status-updated"); + if (events[0]?.type === "thread.view-status-updated") { + expect(events[0].payload.lastViewedAt).toBe(COMPLETED_AT); + expect(Number.isFinite(Date.parse(events[0].payload.lastViewedAt))).toBe(true); + } + }), + ); + + it.effect("marks unread immediately before the latest server completion", () => + Effect.gen(function* () { + const event = yield* decideOrchestrationCommand({ + command: { + type: "thread.mark-unread", + commandId: CommandId.make("cmd-unread"), + threadId: ThreadId.make("thread-1"), + }, + readModel: makeReadModel(), + }); + const events = Array.isArray(event) ? event : [event]; + expect(events).toHaveLength(1); + expect(events[0]?.type).toBe("thread.view-status-updated"); + if (events[0]?.type === "thread.view-status-updated") { + expect(events[0].payload.lastViewedAt).toBe("1969-12-31T23:59:59.999Z"); + } + }), + ); + + it.effect("accepts a server-owned boundary ahead of the decider clock", () => + Effect.gen(function* () { + const viewedAt = "1970-01-01T00:00:00.001Z"; + const event = yield* decideOrchestrationCommand({ + command: { + type: "thread.mark-viewed", + commandId: CommandId.make("cmd-viewed-clock-skew"), + threadId: ThreadId.make("thread-1"), + viewedAt, + expectedLastViewedAt: null, + supersededViewedAt: null, + }, + readModel: makeReadModel(), + }); + const events = Array.isArray(event) ? event : [event]; + expect(events).toHaveLength(1); + expect(events[0]?.type).toBe("thread.view-status-updated"); + if (events[0]?.type === "thread.view-status-updated") { + expect(events[0].payload.lastViewedAt).toBe(viewedAt); + } + }), + ); + + it.effect("does not move viewed state backwards", () => + Effect.gen(function* () { + const readModel = makeReadModel(); + const previousViewedAt = "1970-01-01T00:00:10.000Z"; + const event = yield* decideOrchestrationCommand({ + command: { + type: "thread.mark-viewed", + commandId: CommandId.make("cmd-viewed-stale"), + threadId: ThreadId.make("thread-1"), + viewedAt: "1970-01-01T00:00:05.000Z", + expectedLastViewedAt: previousViewedAt, + supersededViewedAt: null, + }, + readModel: { + ...readModel, + threads: [{ ...readModel.threads[0]!, lastViewedAt: previousViewedAt }], + }, + }); + const events = Array.isArray(event) ? event : [event]; + expect(events).toHaveLength(1); + expect(events[0]?.type).toBe("thread.view-status-updated"); + if (events[0]?.type === "thread.view-status-updated") { + expect(events[0].payload.lastViewedAt).toBe(previousViewedAt); + } + }), + ); + + it.effect("does not let a queued view undo a newer unread action", () => + Effect.gen(function* () { + const readModel = makeReadModel(); + const unreadEvent = yield* decideOrchestrationCommand({ + command: { + type: "thread.mark-unread", + commandId: CommandId.make("cmd-unread-before-stale-view"), + threadId: ThreadId.make("thread-1"), + }, + readModel, + }); + const unreadEvents = Array.isArray(unreadEvent) ? unreadEvent : [unreadEvent]; + expect(unreadEvents[0]?.type).toBe("thread.view-status-updated"); + if (unreadEvents[0]?.type !== "thread.view-status-updated") return; + + const error = yield* decideOrchestrationCommand({ + command: { + type: "thread.mark-viewed", + commandId: CommandId.make("cmd-viewed-queued-before-unread"), + threadId: ThreadId.make("thread-1"), + viewedAt: COMPLETED_AT, + expectedLastViewedAt: null, + supersededViewedAt: null, + }, + readModel: { + ...readModel, + threads: [ + { + ...readModel.threads[0]!, + lastViewedAt: unreadEvents[0].payload.lastViewedAt, + }, + ], + }, + }).pipe(Effect.flip); + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + }), + ); + + it.effect("accepts a view that follows its queued predecessor", () => + Effect.gen(function* () { + const readModel = makeReadModel(); + const predecessorViewedAt = "1970-01-01T00:00:05.000Z"; + const viewedAt = "1970-01-01T00:00:10.000Z"; + const event = yield* decideOrchestrationCommand({ + command: { + type: "thread.mark-viewed", + commandId: CommandId.make("cmd-viewed-after-predecessor"), + threadId: ThreadId.make("thread-1"), + viewedAt, + expectedLastViewedAt: null, + supersededViewedAt: predecessorViewedAt, + }, + readModel: { + ...readModel, + threads: [{ ...readModel.threads[0]!, lastViewedAt: predecessorViewedAt }], + }, + }); + const events = Array.isArray(event) ? event : [event]; + expect(events[0]?.type).toBe("thread.view-status-updated"); + if (events[0]?.type === "thread.view-status-updated") { + expect(events[0].payload.lastViewedAt).toBe(viewedAt); + } + }), + ); + + it.effect("rejects mark-unread before any turn completes", () => + Effect.gen(function* () { + const error = yield* decideOrchestrationCommand({ + command: { + type: "thread.mark-unread", + commandId: CommandId.make("cmd-unread-empty"), + threadId: ThreadId.make("thread-1"), + }, + readModel: makeReadModel(null), + }).pipe(Effect.flip); + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + }), + ); +}); diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts index 9c07a312023..bc15bc955ad 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -91,6 +91,7 @@ describe("orchestration projector", () => { archivedAt: null, settledOverride: null, settledAt: null, + lastViewedAt: null, snoozedUntil: null, snoozedAt: null, deletedAt: null, diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index 5acf3ee6968..8e98eb53246 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -30,6 +30,7 @@ import { ThreadUnarchivedPayload, ThreadUnsettledPayload, ThreadUnsnoozedPayload, + ThreadViewStatusUpdatedPayload, ThreadRevertedPayload, ThreadSessionSetPayload, ThreadTurnDiffCompletedPayload, @@ -295,6 +296,7 @@ export function projectEvent( archivedAt: null, settledOverride: null, settledAt: null, + lastViewedAt: payload.lastViewedAt ?? null, snoozedUntil: null, snoozedAt: null, deletedAt: null, @@ -349,6 +351,21 @@ export function projectEvent( })), ); + case "thread.view-status-updated": + return decodeForEvent( + ThreadViewStatusUpdatedPayload, + event.payload, + event.type, + "payload", + ).pipe( + Effect.map((payload) => ({ + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + lastViewedAt: payload.lastViewedAt, + }), + })), + ); + case "thread.settled": return decodeForEvent(ThreadSettledPayload, event.payload, event.type, "payload").pipe( Effect.map((payload) => ({ diff --git a/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts b/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts index 71d7df566fd..e7476fa91cd 100644 --- a/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts +++ b/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts @@ -93,6 +93,7 @@ projectionRepositoriesLayer("Projection repositories", (it) => { archivedAt: null, settledOverride: null, settledAt: null, + lastViewedAt: null, snoozedUntil: null, snoozedAt: null, pinnedAt: null, @@ -156,6 +157,7 @@ projectionRepositoriesLayer("Projection repositories", (it) => { archivedAt: null, settledOverride: "settled", settledAt: "2026-03-25T00:00:00.000Z", + lastViewedAt: "2026-03-25T01:00:00.000Z", snoozedUntil: "2026-03-26T09:00:00.000Z", snoozedAt: "2026-03-25T00:00:00.000Z", pinnedAt: "2026-03-25T00:00:00.000Z", diff --git a/apps/server/src/persistence/Layers/ProjectionThreads.ts b/apps/server/src/persistence/Layers/ProjectionThreads.ts index b7d8ae13747..39721527439 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreads.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreads.ts @@ -45,6 +45,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { archived_at, settled_override, settled_at, + last_viewed_at, snoozed_until, snoozed_at, pinned_at, @@ -72,6 +73,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { ${row.archivedAt}, ${row.settledOverride}, ${row.settledAt}, + ${row.lastViewedAt}, ${row.snoozedUntil}, ${row.snoozedAt}, ${row.pinnedAt}, @@ -99,6 +101,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { archived_at = excluded.archived_at, settled_override = excluded.settled_override, settled_at = excluded.settled_at, + last_viewed_at = excluded.last_viewed_at, snoozed_until = excluded.snoozed_until, snoozed_at = excluded.snoozed_at, pinned_at = excluded.pinned_at, @@ -133,6 +136,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { archived_at AS "archivedAt", settled_override AS "settledOverride", settled_at AS "settledAt", + last_viewed_at AS "lastViewedAt", snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", pinned_at AS "pinnedAt", @@ -169,6 +173,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { archived_at AS "archivedAt", settled_override AS "settledOverride", settled_at AS "settledAt", + last_viewed_at AS "lastViewedAt", snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", pinned_at AS "pinnedAt", diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 733c52fab3e..2aa386b9054 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -51,6 +51,7 @@ import Migration0035 from "./Migrations/035_ProjectionThreadTitleRegeneration.ts import Migration0036 from "./Migrations/036_ProjectionThreadsPinned.ts"; import Migration0037 from "./Migrations/037_ProjectionTurnsKeysetIndex.ts"; import Migration0038 from "./Migrations/038_ProjectionThreadsPinOrderKey.ts"; +import Migration0039 from "./Migrations/039_ProjectionThreadsLastViewedAt.ts"; /** * Migration loader with all migrations defined inline. @@ -101,6 +102,7 @@ export const migrationEntries = [ [36, "ProjectionThreadsPinned", Migration0036], [37, "ProjectionTurnsKeysetIndex", Migration0037], [38, "ProjectionThreadsPinOrderKey", Migration0038], + [39, "ProjectionThreadsLastViewedAt", Migration0039], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/039_ProjectionThreadsLastViewedAt.test.ts b/apps/server/src/persistence/Migrations/039_ProjectionThreadsLastViewedAt.test.ts new file mode 100644 index 00000000000..cd6f0e55514 --- /dev/null +++ b/apps/server/src/persistence/Migrations/039_ProjectionThreadsLastViewedAt.test.ts @@ -0,0 +1,49 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { runMigrations } from "../Migrations.ts"; +import * as NodeSqliteClient from "../NodeSqliteClient.ts"; + +const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory())); + +layer("039_ProjectionThreadsLastViewedAt", (it) => { + it.effect("starts existing threads read at the upgrade boundary", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* runMigrations({ toMigrationInclusive: 38 }); + yield* sql` + INSERT INTO projection_threads ( + thread_id, + project_id, + title, + model_selection_json, + created_at, + updated_at + ) + VALUES ( + 'thread-1', + 'project-1', + 'Thread 1', + '{"provider":"codex","model":"gpt-5.4"}', + '2026-01-01T00:00:00.000Z', + '2026-01-01T00:00:00.000Z' + ) + `; + + yield* runMigrations({ toMigrationInclusive: 39 }); + + const rows = yield* sql<{ readonly lastViewedAt: string | null }>` + SELECT last_viewed_at AS "lastViewedAt" + FROM projection_threads + WHERE thread_id = 'thread-1' + `; + assert.strictEqual(rows.length, 1); + assert.ok(rows[0]?.lastViewedAt != null); + assert.ok(Number.isFinite(Date.parse(rows[0].lastViewedAt))); + assert.ok(rows[0].lastViewedAt > "2026-01-01T00:00:00.000Z"); + }), + ); +}); diff --git a/apps/server/src/persistence/Migrations/039_ProjectionThreadsLastViewedAt.ts b/apps/server/src/persistence/Migrations/039_ProjectionThreadsLastViewedAt.ts new file mode 100644 index 00000000000..046aaa1a9b9 --- /dev/null +++ b/apps/server/src/persistence/Migrations/039_ProjectionThreadsLastViewedAt.ts @@ -0,0 +1,22 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const columns = yield* sql<{ readonly name: string }>` + PRAGMA table_info(projection_threads) + `; + + if (!columns.some((column) => column.name === "last_viewed_at")) { + yield* sql` + ALTER TABLE projection_threads + ADD COLUMN last_viewed_at TEXT + `; + + // Existing threads start read, while wakes after the upgrade remain visible. + yield* sql` + UPDATE projection_threads + SET last_viewed_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + `; + } +}); diff --git a/apps/server/src/persistence/Services/ProjectionThreads.ts b/apps/server/src/persistence/Services/ProjectionThreads.ts index c572e1d11cc..d48ef607d88 100644 --- a/apps/server/src/persistence/Services/ProjectionThreads.ts +++ b/apps/server/src/persistence/Services/ProjectionThreads.ts @@ -39,6 +39,7 @@ export const ProjectionThread = Schema.Struct({ archivedAt: Schema.NullOr(IsoDateTime), settledOverride: Schema.NullOr(Schema.Literals(["settled", "active"])), settledAt: Schema.NullOr(IsoDateTime), + lastViewedAt: Schema.NullOr(IsoDateTime), snoozedUntil: Schema.NullOr(IsoDateTime), snoozedAt: Schema.NullOr(IsoDateTime), pinnedAt: Schema.NullOr(IsoDateTime), diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 8b510d457fd..5eb1ed2032a 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -100,7 +100,6 @@ import { togglePendingUserInputOptionSelection, type PendingUserInputDraftAnswer, } from "../pendingUserInput"; -import { useUiStateStore } from "../uiStateStore"; import { buildPlanImplementationThreadTitle, buildPlanImplementationPrompt, @@ -234,6 +233,7 @@ import { useEnvironments, usePrimaryEnvironment } from "../state/environments"; import { useProject, useProjects, + useServerConfigs, useThread, useThreadRefs, useThreadShell, @@ -262,6 +262,7 @@ import { } from "./chat/ProviderStatusBanner"; import { ThreadErrorBanner } from "./chat/ThreadErrorBanner"; import { resolveThreadPr } from "./ThreadStatusIndicators"; +import { hasUnseenWake } from "./Sidebar.logic"; import { ComposerBannerStack, type ComposerBannerStackItem } from "./chat/ComposerBannerStack"; import { ThreadSyncStatusPill } from "./chat/ThreadSyncStatusPill"; import { @@ -1196,6 +1197,9 @@ function ChatViewContent(props: ChatViewProps) { const updateThreadMetadata = useAtomCommand(threadEnvironment.updateMetadata, { reportFailure: false, }); + const markThreadViewed = useAtomCommand(threadEnvironment.markViewed, { + reportFailure: false, + }); const switchGitRef = useAtomCommand(vcsEnvironment.switchRef, { reportFailure: false }); const setThreadRuntimeMode = useAtomCommand(threadEnvironment.setRuntimeMode, { reportFailure: false, @@ -1219,6 +1223,7 @@ function ChatViewContent(props: ChatViewProps) { const openPreview = useAtomCommand(previewEnvironment.open, { reportFailure: false }); const closePreview = useAtomCommand(previewEnvironment.close, "preview close"); const { environments } = useEnvironments(); + const serverConfigs = useServerConfigs(); const primaryEnvironment = usePrimaryEnvironment(); const retryEnvironment = useAtomCommand(environmentCatalog.retryNow, { reportFailure: false }); const environmentById = useMemo( @@ -1261,7 +1266,6 @@ function ChatViewContent(props: ChatViewProps) { }, }; }, [routeKind, routeThreadRef, routeThreadState]); - const markThreadVisited = useUiStateStore((store) => store.markThreadVisited); const settings = useEnvironmentSettings(environmentId); // New-thread defaults live in the primary environment's settings.json (the // settings UI never writes to remote environments), so read them from the @@ -1617,23 +1621,29 @@ function ChatViewContent(props: ChatViewProps) { return openTerminalThreadKeys.filter((nextThreadKey) => existingThreadKeys.has(nextThreadKey)); }, [draftThreadKeys, openTerminalThreadKeys, serverThreadKeys]); const activeLatestTurn = activeThread?.latestTurn ?? null; + const serverThreadSupportsViewStatus = + serverThread != null && + serverConfigs.get(serverThread.environmentId)?.environment.capabilities.threadViewStatus === + true; // Reading a finished thread clears the sidebar's Done badge. The visit is - // stamped at the turn's completion time — not now/updatedAt — so it clears - // exactly the completion the user is looking at: a wake or completion that - // lands later still gets its signal (markThreadVisited never moves the - // timestamp backwards). + // recorded by the server. A later completion retriggers this effect, so + // work that lands while the thread stays open is acknowledged too. useEffect(() => { const completedAt = serverThread?.latestTurn?.completedAt; - if (!serverThread?.id || !completedAt) return; - markThreadVisited( - scopedThreadKey(scopeThreadRef(serverThread.environmentId, serverThread.id)), - completedAt, - ); + if (!serverThread?.id || !completedAt || !serverThreadSupportsViewStatus) return; + void markThreadViewed({ + environmentId: serverThread.environmentId, + input: { + threadId: serverThread.id, + viewedAt: completedAt, + }, + }); }, [ - markThreadVisited, + markThreadViewed, serverThread?.environmentId, serverThread?.id, serverThread?.latestTurn?.completedAt, + serverThreadSupportsViewStatus, ]); useEffect(() => { setMountedTerminalThreadKeys((currentThreadIds) => { @@ -4044,6 +4054,10 @@ function ChatViewContent(props: ChatViewProps) { activeThreadShell !== null && supportsSnooze ? threadWokeAt(activeThreadShell, { now: snoozeNow }) : null; + const activeThreadSupportsViewStatus = + activeThreadRef !== null && + serverConfigs.get(activeThreadRef.environmentId)?.environment.capabilities.threadViewStatus === + true; useEffect(() => { if (!activeThreadSnoozed) return; const wakeAtMs = Date.parse(activeThreadShell?.snoozedUntil ?? ""); @@ -4055,37 +4069,43 @@ function ChatViewContent(props: ChatViewProps) { return () => window.clearTimeout(id); }, [activeThreadShell?.snoozedUntil, activeThreadSnoozed, snoozeWakeTick]); const acknowledgeActiveThreadWoke = useCallback(() => { - if (activeThreadRef === null || activeThreadWokeAt === null) return; - markThreadVisited(scopedThreadKey(activeThreadRef), activeThreadWokeAt); - }, [activeThreadRef, activeThreadWokeAt, markThreadVisited]); + if (activeThreadRef === null || activeThreadWokeAt === null || !activeThreadSupportsViewStatus) + return; + void markThreadViewed({ + environmentId: activeThreadRef.environmentId, + input: { + threadId: activeThreadRef.threadId, + viewedAt: activeThreadWokeAt, + }, + }); + }, [activeThreadRef, activeThreadSupportsViewStatus, activeThreadWokeAt, markThreadViewed]); // Mirror of the sidebar's Woke pill for the open thread: same visit // comparison, same merged/closed-PR suppression (finished work needs no // wake-up call). Drives the dismissible composer banner below. - const activeThreadLastVisitedAt = useUiStateStore((store) => - activeThreadKey === null ? undefined : store.threadLastVisitedAtById[activeThreadKey], - ); + const activeThreadLastViewedAt = activeThreadShell?.lastViewedAt; const activeThreadWokeVisible = useMemo(() => { - if (activeThreadWokeAt === null) return false; if (activeThreadPr?.state === "merged" || activeThreadPr?.state === "closed") return false; + if ( + !hasUnseenWake({ + viewStatusSupported: activeThreadSupportsViewStatus, + wokeAt: activeThreadWokeAt, + lastViewedAt: activeThreadLastViewedAt, + }) + ) + return false; + if (activeThreadWokeAt === null) return false; const wokeAtMs = Date.parse(activeThreadWokeAt); - if (Number.isNaN(wokeAtMs)) return false; // Having the thread open counts as a visit at completedAt (the effect // above stamps it); folding that floor in here keeps a completion- - // triggered wake from flashing a banner for one frame before the stamp - // lands. An unparseable stored visit counts as never-visited: corrupt - // local data must not eat the wake signal. - const storedVisitMs = activeThreadLastVisitedAt ? Date.parse(activeThreadLastVisitedAt) : NaN; + // triggered wake from flashing a banner for one frame before the stamp lands. const completedAtMs = activeLatestTurn?.completedAt ? Date.parse(activeLatestTurn.completedAt) : NaN; - const lastVisitedMs = Math.max( - Number.isNaN(storedVisitMs) ? -Infinity : storedVisitMs, - Number.isNaN(completedAtMs) ? -Infinity : completedAtMs, - ); - return lastVisitedMs < wokeAtMs; + return Number.isNaN(completedAtMs) || completedAtMs < wokeAtMs; }, [ activeLatestTurn?.completedAt, - activeThreadLastVisitedAt, + activeThreadSupportsViewStatus, + activeThreadLastViewedAt, activeThreadPr?.state, activeThreadWokeAt, ]); diff --git a/apps/web/src/components/LegacySidebar.tsx b/apps/web/src/components/LegacySidebar.tsx index 2c1f99ffa0f..0caf1f11fe4 100644 --- a/apps/web/src/components/LegacySidebar.tsx +++ b/apps/web/src/components/LegacySidebar.tsx @@ -77,6 +77,7 @@ import { useOpenPrLink } from "../lib/openPullRequestLink"; import { isTerminalFocused } from "../lib/terminalFocus"; import { isMacPlatform } from "../lib/utils"; import { + readEnvironmentSupportsViewStatus, readThreadShell, useProject, useProjects, @@ -366,7 +367,6 @@ export const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThr } = props; const threadRef = scopeThreadRef(thread.environmentId, thread.id); const threadKey = scopedThreadKey(threadRef); - const lastVisitedAt = useUiStateStore((state) => state.threadLastVisitedAtById[threadKey]); const isSelected = useThreadSelectionStore((state) => state.selectedThreadKeys.has(threadKey)); const runningTerminalIds = useThreadRunningTerminalIds({ environmentId: thread.environmentId, @@ -442,12 +442,7 @@ export const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThr ); const isThreadRunning = thread.session?.status === "running" && thread.session.activeTurnId != null; - const threadStatus = resolveThreadStatusPill({ - thread: { - ...thread, - lastVisitedAt, - }, - }); + const threadStatus = resolveThreadStatusPill({ thread }); const pr = resolveThreadPr({ threadBranch: thread.branch, gitStatus: gitStatus.data, @@ -1112,7 +1107,9 @@ const SidebarProjectItem = memo(function SidebarProjectItem(props: SidebarProjec ); const router = useRouter(); const { isMobile, setOpenMobile } = useSidebar(); - const markThreadUnread = useUiStateStore((state) => state.markThreadUnread); + const markThreadUnread = useAtomCommand(threadEnvironment.markUnread, { + reportFailure: false, + }); const setProjectExpanded = useUiStateStore((state) => state.setProjectExpanded); const toggleThreadSelection = useThreadSelectionStore((state) => state.toggleThread); const rangeSelectTo = useThreadSelectionStore((state) => state.rangeSelectTo); @@ -1181,16 +1178,6 @@ const SidebarProjectItem = memo(function SidebarProjectItem(props: SidebarProjec const projectExpanded = useUiStateStore((state) => resolveProjectExpanded(state.projectExpandedById, projectPreferenceKeys), ); - const threadLastVisitedAts = useUiStateStore( - useShallow((state) => - projectThreads.map( - (thread) => - state.threadLastVisitedAtById[ - scopedThreadKey(scopeThreadRef(thread.environmentId, thread.id)) - ] ?? null, - ), - ), - ); const [renamingThreadKey, setRenamingThreadKey] = useState(null); const [renamingTitle, setRenamingTitle] = useState(""); const [confirmingArchiveThreadKey, setConfirmingArchiveThreadKey] = useState(null); @@ -1233,22 +1220,8 @@ const SidebarProjectItem = memo(function SidebarProjectItem(props: SidebarProjec }, [memberProjectByScopedKey, project.memberProjects, projectThreads]); const { projectStatus, visibleProjectThreads, orderedProjectThreadKeys } = useMemo(() => { - const lastVisitedAtByThreadKey = new Map( - projectThreads.map((thread, index) => [ - scopedThreadKey(scopeThreadRef(thread.environmentId, thread.id)), - threadLastVisitedAts[index] ?? null, - ]), - ); const resolveProjectThreadStatus = (thread: SidebarThreadSummary) => { - const lastVisitedAt = lastVisitedAtByThreadKey.get( - scopedThreadKey(scopeThreadRef(thread.environmentId, thread.id)), - ); - return resolveThreadStatusPill({ - thread: { - ...thread, - ...(lastVisitedAt !== null && lastVisitedAt !== undefined ? { lastVisitedAt } : {}), - }, - }); + return resolveThreadStatusPill({ thread }); }; const visibleProjectThreads = sortThreads( projectThreads.filter((thread) => thread.archivedAt === null), @@ -1264,7 +1237,7 @@ const SidebarProjectItem = memo(function SidebarProjectItem(props: SidebarProjec projectStatus, visibleProjectThreads, }; - }, [projectThreads, threadLastVisitedAts, threadSortOrder]); + }, [projectThreads, threadSortOrder]); const pinnedCollapsedThread = useMemo(() => { const activeThreadKey = activeRouteThreadKey ?? undefined; if (!activeThreadKey || projectExpanded) { @@ -1285,22 +1258,8 @@ const SidebarProjectItem = memo(function SidebarProjectItem(props: SidebarProjec showEmptyThreadState, shouldShowThreadPanel, } = useMemo(() => { - const lastVisitedAtByThreadKey = new Map( - projectThreads.map((thread, index) => [ - scopedThreadKey(scopeThreadRef(thread.environmentId, thread.id)), - threadLastVisitedAts[index] ?? null, - ]), - ); const resolveProjectThreadStatus = (thread: SidebarThreadSummary) => { - const lastVisitedAt = lastVisitedAtByThreadKey.get( - scopedThreadKey(scopeThreadRef(thread.environmentId, thread.id)), - ); - return resolveThreadStatusPill({ - thread: { - ...thread, - ...(lastVisitedAt !== null && lastVisitedAt !== undefined ? { lastVisitedAt } : {}), - }, - }); + return resolveThreadStatusPill({ thread }); }; const hasOverflowingThreads = visibleProjectThreads.length > sidebarThreadPreviewCount; const previewThreads = @@ -1336,7 +1295,6 @@ const SidebarProjectItem = memo(function SidebarProjectItem(props: SidebarProjec projectExpanded, projectThreads, sidebarThreadPreviewCount, - threadLastVisitedAts, visibleProjectThreads, ]); @@ -1767,15 +1725,42 @@ const SidebarProjectItem = memo(function SidebarProjectItem(props: SidebarProjec const hasRunningThread = selectedThreadEntries.some( ({ thread }) => thread.session?.status === "running" && thread.session.activeTurnId != null, ); + const canMarkUnread = + selectedThreadEntries.length > 0 && + selectedThreadEntries.every( + ({ thread }) => + readEnvironmentSupportsViewStatus(thread.environmentId) && + thread.latestTurn?.completedAt != null, + ); const clicked = await api.contextMenu.show( - buildMultiSelectThreadContextMenuItems({ count, hasRunningThread }), + buildMultiSelectThreadContextMenuItems({ count, hasRunningThread, canMarkUnread }), position, ); if (clicked === "mark-unread") { - for (const { threadKey, thread } of selectedThreadEntries) { - markThreadUnread(threadKey, thread.latestTurn?.completedAt); + for (const { thread } of selectedThreadEntries) { + if ( + !readEnvironmentSupportsViewStatus(thread.environmentId) || + thread.latestTurn?.completedAt == null + ) + continue; + const result = await markThreadUnread({ + environmentId: thread.environmentId, + input: { threadId: thread.id }, + }); + if (result._tag === "Success") continue; + if (!isAtomCommandInterrupted(result)) { + const error = squashAtomCommandFailure(result); + toastManager.add( + stackedThreadToast({ + type: "error", + title: "Failed to mark threads unread", + description: error instanceof Error ? error.message : "An error occurred.", + }), + ); + } + return; } clearSelection(); return; @@ -2114,7 +2099,10 @@ const SidebarProjectItem = memo(function SidebarProjectItem(props: SidebarProjec ? [{ id: "new-thread-on-branch", label: `New thread on ${thread.branch}` }] : []), { id: "rename", label: "Rename thread" }, - { id: "mark-unread", label: "Mark unread" }, + ...(readEnvironmentSupportsViewStatus(thread.environmentId) && + thread.latestTurn?.completedAt != null + ? [{ id: "mark-unread", label: "Mark unread" }] + : []), { id: "copy-path", label: "Copy Path" }, { id: "copy-thread-id", label: "Copy Thread ID" }, { id: "delete", label: "Delete", destructive: true, icon: "trash" }, @@ -2152,7 +2140,20 @@ const SidebarProjectItem = memo(function SidebarProjectItem(props: SidebarProjec } if (clicked === "mark-unread") { - markThreadUnread(threadKey, thread.latestTurn?.completedAt); + const result = await markThreadUnread({ + environmentId: thread.environmentId, + input: { threadId: thread.id }, + }); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + const error = squashAtomCommandFailure(result); + toastManager.add( + stackedThreadToast({ + type: "error", + title: "Failed to mark thread unread", + description: error instanceof Error ? error.message : "An error occurred.", + }), + ); + } return; } if (clicked === "copy-path") { diff --git a/apps/web/src/components/Sidebar.logic.test.ts b/apps/web/src/components/Sidebar.logic.test.ts index bfe4162cd20..7377384a4a5 100644 --- a/apps/web/src/components/Sidebar.logic.test.ts +++ b/apps/web/src/components/Sidebar.logic.test.ts @@ -11,6 +11,7 @@ import { getVisibleThreadsForProject, getProjectSortTimestamp, hasUnseenCompletion, + hasUnseenWake, isContextMenuPointerDown, isTrailingDoubleClick, orderItemsByPreferredIds, @@ -193,13 +194,21 @@ describe("buildBulkTitleRegenerationContextMenuItem", () => { describe("buildMultiSelectThreadContextMenuItems", () => { it("offers bulk archive with the selected count", () => { expect( - buildMultiSelectThreadContextMenuItems({ count: 3, hasRunningThread: false }), + buildMultiSelectThreadContextMenuItems({ + count: 3, + hasRunningThread: false, + canMarkUnread: true, + }), ).toContainEqual({ id: "archive", label: "Archive (3)", disabled: false }); }); it("disables bulk archive when a selected thread is running", () => { expect( - buildMultiSelectThreadContextMenuItems({ count: 2, hasRunningThread: true }), + buildMultiSelectThreadContextMenuItems({ + count: 2, + hasRunningThread: true, + canMarkUnread: true, + }), ).toContainEqual({ id: "archive", label: "Archive (2)", disabled: true }); }); }); @@ -267,13 +276,13 @@ describe("hasUnseenCompletion", () => { hasPendingUserInput: false, interactionMode: "default", latestTurn: makeLatestTurn(), - lastVisitedAt: "2026-03-09T10:04:00.000Z", + lastViewedAt: "2026-03-09T10:04:00.000Z", session: null, }), ).toBe(true); }); - it("treats a missing client visit marker as read", () => { + it("treats missing server view state as read", () => { expect( hasUnseenCompletion({ hasActionableProposedPlan: false, @@ -281,13 +290,52 @@ describe("hasUnseenCompletion", () => { hasPendingUserInput: false, interactionMode: "default", latestTurn: makeLatestTurn(), - lastVisitedAt: undefined, + lastViewedAt: undefined, session: null, }), ).toBe(false); }); }); +describe("hasUnseenWake", () => { + it("treats missing server view state as read", () => { + expect( + hasUnseenWake({ + viewStatusSupported: true, + wokeAt: "2026-03-09T10:05:00.000Z", + lastViewedAt: null, + }), + ).toBe(false); + }); + + it("suppresses wakes that an older server cannot acknowledge", () => { + expect( + hasUnseenWake({ + viewStatusSupported: false, + wokeAt: "2026-03-09T10:05:00.000Z", + lastViewedAt: undefined, + }), + ).toBe(false); + }); + + it("clears a wake only after its timestamp has been viewed", () => { + expect( + hasUnseenWake({ + viewStatusSupported: true, + wokeAt: "2026-03-09T10:05:00.000Z", + lastViewedAt: "2026-03-09T10:04:00.000Z", + }), + ).toBe(true); + expect( + hasUnseenWake({ + viewStatusSupported: true, + wokeAt: "2026-03-09T10:05:00.000Z", + lastViewedAt: "2026-03-09T10:05:00.000Z", + }), + ).toBe(false); + }); +}); + describe("createThreadJumpHintVisibilityController", () => { beforeEach(() => { vi.useFakeTimers(); @@ -1015,7 +1063,7 @@ describe("resolveThreadStatusPill", () => { hasPendingUserInput: false, interactionMode: "plan" as const, latestTurn: null, - lastVisitedAt: undefined, + lastViewedAt: undefined, session: { threadId: ThreadId.make("thread-1"), status: "running" as const, @@ -1099,7 +1147,7 @@ describe("resolveThreadStatusPill", () => { ...baseThread, interactionMode: "default", latestTurn: makeLatestTurn(), - lastVisitedAt: "2026-03-09T10:04:00.000Z", + lastViewedAt: "2026-03-09T10:04:00.000Z", session: { ...baseThread.session, status: "ready", diff --git a/apps/web/src/components/Sidebar.logic.ts b/apps/web/src/components/Sidebar.logic.ts index cae26f5d6bd..a7483cbab94 100644 --- a/apps/web/src/components/Sidebar.logic.ts +++ b/apps/web/src/components/Sidebar.logic.ts @@ -87,9 +87,12 @@ export async function archiveSelectedThreadEntries< export function buildMultiSelectThreadContextMenuItems(input: { count: number; hasRunningThread: boolean; + canMarkUnread: boolean; }): readonly ContextMenuItem<"mark-unread" | "archive" | "delete">[] { return [ - { id: "mark-unread", label: `Mark unread (${input.count})` }, + ...(input.canMarkUnread + ? [{ id: "mark-unread" as const, label: `Mark unread (${input.count})` }] + : []), { id: "archive", label: `Archive (${input.count})`, @@ -154,7 +157,7 @@ type ThreadStatusInput = Pick< | "session" | "backgroundLiveness" > & { - lastVisitedAt?: string | undefined; + lastViewedAt?: string | null | undefined; }; export interface ThreadJumpHintVisibilityController { @@ -253,11 +256,24 @@ export function hasUnseenCompletion(thread: ThreadStatusInput): boolean { if (!thread.latestTurn?.completedAt) return false; const completedAt = Date.parse(thread.latestTurn.completedAt); if (Number.isNaN(completedAt)) return false; - if (!thread.lastVisitedAt) return false; + if (!thread.lastViewedAt) return false; - const lastVisitedAt = Date.parse(thread.lastVisitedAt); - if (Number.isNaN(lastVisitedAt)) return true; - return completedAt > lastVisitedAt; + const lastViewedAt = Date.parse(thread.lastViewedAt); + if (Number.isNaN(lastViewedAt)) return true; + return completedAt > lastViewedAt; +} + +export function hasUnseenWake(input: { + readonly viewStatusSupported: boolean; + readonly wokeAt: string | null; + readonly lastViewedAt: string | null | undefined; +}): boolean { + if (!input.viewStatusSupported || input.wokeAt === null) return false; + const wokeAt = Date.parse(input.wokeAt); + if (Number.isNaN(wokeAt)) return false; + if (input.lastViewedAt == null) return false; + const lastViewedAt = Date.parse(input.lastViewedAt); + return Number.isNaN(lastViewedAt) || lastViewedAt < wokeAt; } export function shouldClearThreadSelectionOnMouseDown(target: HTMLElement | null): boolean { diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index 75c580402b1..9f8ae32ccbf 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -121,7 +121,7 @@ import { resolveActiveThreadRouteRef, resolveThreadRouteTarget, } from "../threadRoutes"; -import { formatRelativeTimeLabel, parseTimestampDate } from "../timestampFormat"; +import { formatRelativeTimeLabel } from "../timestampFormat"; import type { SidebarThreadSummary } from "../types"; import { cn } from "~/lib/utils"; import { buildThreadActionMenuItems } from "./threadActionMenu.logic"; @@ -130,6 +130,7 @@ import { formatWorkingDurationLabel, firstValidTimestampMs, hasUnseenCompletion, + hasUnseenWake, isTrailingDoubleClick, orderItemsByPreferredIds, planPinnedReorder, @@ -439,6 +440,8 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { settlementSupported: boolean; // Same contract for thread.snooze/unsnooze. snoozeSupported: boolean; + // Read/unread and Woke acknowledgement require server-owned view state. + viewStatusSupported: boolean; // Renders the pin glyph. Pinned cards keep the full settle/snooze quick // actions: settling clears the pin server-side, and snoozing hides the // card until wake with the pin intact underneath. The glyph is also the @@ -509,7 +512,6 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { ); const threadKey = scopedThreadKey(threadRef); const isRegeneratingTitle = thread.titleRegeneration != null; - const lastVisitedAt = useUiStateStore((state) => state.threadLastVisitedAtById[threadKey]); const isSelected = useThreadSelectionStore((state) => state.selectedThreadKeys.has(threadKey)); const openPrLink = useOpenPrLink(); const runningTerminalIds = useThreadRunningTerminalIds({ @@ -534,9 +536,9 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { }); const prState = pr?.state ?? null; - // Same semantics as the legacy sidebar (never-visited counts as read): - // switching sidebars must not light up every historical thread as unread. - const isUnread = hasUnseenCompletion({ ...thread, lastVisitedAt }); + // Missing server state counts as read so upgrading does not light up every + // historical thread as unread. + const isUnread = hasUnseenCompletion(thread); const status = resolveSidebarThreadStatus(thread); // A woken thread reappears at its original position (the sort is // deliberately static), so the pill has to carry the weight. Snoozing is @@ -544,13 +546,14 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { // reading a completion-triggered wake, clicking the pill, sending a // message, settling, archiving — or finishing the work outright (merged // or closed PR). Timer wakes survive a mere visit. An unparseable visit - // timestamp counts as never-visited — corrupt local data must not eat + // timestamp counts as never-viewed — corrupt server data must not eat // the wake signal. - const lastVisitedDate = lastVisitedAt === undefined ? null : parseTimestampDate(lastVisitedAt); - const wokeAtDate = props.wokeAt === null ? null : parseTimestampDate(props.wokeAt); const isWoke = - wokeAtDate !== null && - (lastVisitedDate === null || lastVisitedDate < wokeAtDate) && + hasUnseenWake({ + viewStatusSupported: props.viewStatusSupported, + wokeAt: props.wokeAt, + lastViewedAt: thread.lastViewedAt, + }) && prState !== "merged" && prState !== "closed"; // In-flight rows (working, or waiting on approval/input) fade as a whole: @@ -1433,18 +1436,31 @@ export default function Sidebar() { [], ); const { environments } = useEnvironments(); + const serverConfigs = useAtomValue(environmentServerConfigsAtom); const primaryEnvironmentId = usePrimaryEnvironmentId(); const clearSelection = useThreadSelectionStore((s) => s.clearSelection); const setSelectionAnchor = useThreadSelectionStore((s) => s.setAnchor); const toggleThreadSelection = useThreadSelectionStore((s) => s.toggleThread); const rangeSelectTo = useThreadSelectionStore((s) => s.rangeSelectTo); - const markThreadUnread = useUiStateStore((s) => s.markThreadUnread); - const markThreadVisited = useUiStateStore((s) => s.markThreadVisited); + const markThreadUnread = useAtomCommand(threadEnvironment.markUnread, { + reportFailure: false, + }); + const markThreadViewed = useAtomCommand(threadEnvironment.markViewed, { + reportFailure: false, + }); const acknowledgeWoke = useCallback( (threadRef: ScopedThreadRef, visitedAt: string) => { - markThreadVisited(scopedThreadKey(threadRef), visitedAt); + if ( + serverConfigs.get(threadRef.environmentId)?.environment.capabilities.threadViewStatus !== + true + ) + return; + void markThreadViewed({ + environmentId: threadRef.environmentId, + input: { threadId: threadRef.threadId, viewedAt: visitedAt }, + }); }, - [markThreadVisited], + [markThreadViewed, serverConfigs], ); const routeTarget = useParams({ strict: false, @@ -1758,7 +1774,6 @@ export default function Sidebar() { // the partition works directly off live shells: no archived-snapshot // merging, no optimistic holds. Archived threads remain hidden here — // archive keeps its original "remove from sidebar" meaning. - const serverConfigs = useAtomValue(environmentServerConfigsAtom); const { pinnedThreads, reorderablePinnedKeys, @@ -2529,6 +2544,11 @@ export default function Sidebar() { const regeneratableTitleThreads = titleRegenerationThreads.filter( (thread) => thread.titleRegeneration == null, ); + const markUnreadThreads = selectedThreads.filter( + (thread) => + serverConfigs.get(thread.environmentId)?.environment.capabilities.threadViewStatus === + true && thread.latestTurn?.completedAt != null, + ); const titleRegenerationMenuItem = buildBulkTitleRegenerationContextMenuItem({ supportedCount: titleRegenerationThreads.length, actionableCount: regeneratableTitleThreads.length, @@ -2551,7 +2571,9 @@ export default function Sidebar() { ] : []), ...(titleRegenerationMenuItem ? [titleRegenerationMenuItem] : []), - { id: "mark-unread", label: `Mark unread (${count})` }, + ...(markUnreadThreads.length > 0 + ? [{ id: "mark-unread", label: `Mark unread (${markUnreadThreads.length})` }] + : []), { id: "delete", label: `Delete (${count})`, destructive: true }, ], position, @@ -2656,9 +2678,25 @@ export default function Sidebar() { return; } if (clicked.value === "mark-unread") { - for (const threadKey of threadKeys) { - const thread = threadByKeyRef.current.get(threadKey); - markThreadUnread(threadKey, thread?.latestTurn?.completedAt); + for (const thread of markUnreadThreads) { + const result = await markThreadUnread({ + environmentId: thread.environmentId, + input: { threadId: thread.id }, + }); + if (result._tag === "Success") continue; + if (!isAtomCommandInterrupted(result)) { + const error = squashAtomCommandFailure(result); + toastManager.add( + stackedThreadToast({ + type: "error", + title: "Failed to mark threads unread", + description: error instanceof Error ? error.message : "An error occurred.", + }), + ); + } + // Keep the selection so a transient failure can be retried. Commands + // that already landed are idempotent. + return; } clearSelection(); return; @@ -2750,6 +2788,9 @@ export default function Sidebar() { const supportsTitleRegeneration = serverConfigs.get(thread.environmentId)?.environment.capabilities .threadTitleRegeneration === true; + const supportsViewStatus = + serverConfigs.get(thread.environmentId)?.environment.capabilities.threadViewStatus === + true; const isRegeneratingTitle = thread.titleRegeneration != null; const isSettled = settledThreadKeysRef.current.has(threadKey); const isSnoozed = snoozedThreadKeysRef.current.has(threadKey); @@ -2764,12 +2805,14 @@ export default function Sidebar() { isSettled, isSnoozed, canSnoozeNow: canSnooze(thread, { now: new Date().toISOString() }), + canMarkUnread: thread.latestTurn?.completedAt != null, isRegeneratingTitle, supports: { settlement: supportsSettlement, snooze: supportsSnooze, pinning: supportsPinning, titleRegeneration: supportsTitleRegeneration, + viewStatus: supportsViewStatus, }, snoozePresets, }), @@ -2844,9 +2887,23 @@ export default function Sidebar() { } return; } - case "mark-unread": - markThreadUnread(threadKey, thread.latestTurn?.completedAt); + case "mark-unread": { + const result = await markThreadUnread({ + environmentId: thread.environmentId, + input: { threadId: thread.id }, + }); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + const error = squashAtomCommandFailure(result); + toastManager.add( + stackedThreadToast({ + type: "error", + title: "Failed to mark thread unread", + description: error instanceof Error ? error.message : "An error occurred.", + }), + ); + } return; + } case "copy-path": if (!threadWorkspacePath) { toastManager.add( @@ -3305,6 +3362,10 @@ export default function Sidebar() { serverConfigs.get(thread.environmentId)?.environment.capabilities .threadPinning === true } + viewStatusSupported={ + serverConfigs.get(thread.environmentId)?.environment.capabilities + .threadViewStatus === true + } isPinned={section === "pinned"} sortable={sortable} snoozeWakeLabelText={ diff --git a/apps/web/src/components/ThreadStatusIndicators.tsx b/apps/web/src/components/ThreadStatusIndicators.tsx index af53d1a78b2..774a9e88292 100644 --- a/apps/web/src/components/ThreadStatusIndicators.tsx +++ b/apps/web/src/components/ThreadStatusIndicators.tsx @@ -1,8 +1,4 @@ -import { - scopeProjectRef, - scopedThreadKey, - scopeThreadRef, -} from "@t3tools/client-runtime/environment"; +import { scopeProjectRef } from "@t3tools/client-runtime/environment"; import type { VcsStatusResult } from "@t3tools/contracts"; import { CloudIcon, FolderGit2Icon, GitPullRequestIcon, TerminalIcon } from "lucide-react"; import { useMemo } from "react"; @@ -11,7 +7,6 @@ import { useProject } from "../state/entities"; import { useEnvironmentQuery } from "../state/query"; import { useThreadRunningTerminalIds } from "../state/terminalSessions"; import { vcsEnvironment } from "../state/vcs"; -import { useUiStateStore } from "../uiStateStore"; import { resolveChangeRequestPresentation } from "../sourceControlPresentation"; import { resolveThreadStatusPill, type ThreadStatusPill } from "./Sidebar.logic"; import type { SidebarThreadSummary } from "../types"; @@ -230,10 +225,6 @@ export function ThreadStatusLabel({ * thread status dot, matching the sidebar's leading indicators. */ export function ThreadRowLeadingStatus({ thread }: { thread: SidebarThreadSummary }) { - const threadRef = scopeThreadRef(thread.environmentId, thread.id); - const lastVisitedAt = useUiStateStore( - (state) => state.threadLastVisitedAtById[scopedThreadKey(threadRef)], - ); const threadProject = useProject( useMemo( () => scopeProjectRef(thread.environmentId, thread.projectId), @@ -255,12 +246,7 @@ export function ThreadRowLeadingStatus({ thread }: { thread: SidebarThreadSummar gitStatus: gitStatus.data, }); const prStatus = prStatusIndicator(pr, gitStatus.data?.sourceControlProvider); - const threadStatus = resolveThreadStatusPill({ - thread: { - ...thread, - lastVisitedAt, - }, - }); + const threadStatus = resolveThreadStatusPill({ thread }); if (!prStatus && !threadStatus) { return null; diff --git a/apps/web/src/components/threadActionMenu.logic.test.ts b/apps/web/src/components/threadActionMenu.logic.test.ts index a450b29f226..4bd927a8996 100644 --- a/apps/web/src/components/threadActionMenu.logic.test.ts +++ b/apps/web/src/components/threadActionMenu.logic.test.ts @@ -8,8 +8,15 @@ const baseState: ThreadActionMenuState = { isSettled: false, isSnoozed: false, canSnoozeNow: true, + canMarkUnread: true, isRegeneratingTitle: false, - supports: { settlement: true, snooze: true, pinning: true, titleRegeneration: true }, + supports: { + settlement: true, + snooze: true, + pinning: true, + titleRegeneration: true, + viewStatus: true, + }, snoozePresets: [ { id: "hour", label: "In 1 hour", whenLabel: "3:00 PM", snoozedUntil: "2026-08-07T15:00:00Z" }, ], @@ -24,9 +31,15 @@ describe("buildThreadActionMenuItems", () => { expect( ids({ ...baseState, - supports: { settlement: false, snooze: false, pinning: false, titleRegeneration: false }, + supports: { + settlement: false, + snooze: false, + pinning: false, + titleRegeneration: false, + viewStatus: false, + }, }), - ).toEqual(["rename", "mark-unread", "copy-path", "delete"]); + ).toEqual(["rename", "copy-path", "delete"]); }); it("includes branch items only for threads with a branch", () => { @@ -59,6 +72,10 @@ describe("buildThreadActionMenuItems", () => { expect(item).toMatchObject({ label: "Regenerating…", disabled: true }); }); + it("hides mark unread until the thread has a completed turn", () => { + expect(ids({ ...baseState, canMarkUnread: false })).not.toContain("mark-unread"); + }); + it("marks delete as destructive and keeps it last", () => { const items = buildThreadActionMenuItems({ ...baseState, branch: "main" }); expect(items.at(-1)).toMatchObject({ id: "delete", destructive: true }); diff --git a/apps/web/src/components/threadActionMenu.logic.ts b/apps/web/src/components/threadActionMenu.logic.ts index 66aaf3debf5..6a5167a5574 100644 --- a/apps/web/src/components/threadActionMenu.logic.ts +++ b/apps/web/src/components/threadActionMenu.logic.ts @@ -28,12 +28,14 @@ export interface ThreadActionMenuState { readonly isSettled: boolean; readonly isSnoozed: boolean; readonly canSnoozeNow: boolean; + readonly canMarkUnread: boolean; readonly isRegeneratingTitle: boolean; readonly supports: { readonly settlement: boolean; readonly snooze: boolean; readonly pinning: boolean; readonly titleRegeneration: boolean; + readonly viewStatus: boolean; }; readonly snoozePresets: ReadonlyArray; } @@ -97,7 +99,9 @@ export function buildThreadActionMenuItems( }, ] : []), - { id: "mark-unread", label: "Mark unread" }, + ...(state.supports.viewStatus && state.canMarkUnread + ? [{ id: "mark-unread" as const, label: "Mark unread" }] + : []), { id: "copy-path", label: "Copy path", icon: "copy" }, ...(state.branch ? [{ id: "copy-branch" as const, label: "Copy branch", icon: "copy" }] : []), { id: "delete", label: "Delete", destructive: true, icon: "trash" }, diff --git a/apps/web/src/hooks/useThreadActionMenu.ts b/apps/web/src/hooks/useThreadActionMenu.ts index 85ffde776b4..f00b3492b5f 100644 --- a/apps/web/src/hooks/useThreadActionMenu.ts +++ b/apps/web/src/hooks/useThreadActionMenu.ts @@ -1,4 +1,4 @@ -import { scopeProjectRef, scopedThreadKey } from "@t3tools/client-runtime/environment"; +import { scopeProjectRef } from "@t3tools/client-runtime/environment"; import { type AtomCommandResult, isAtomCommandInterrupted, @@ -27,10 +27,10 @@ import { readEnvironmentSupportsSettlement, readEnvironmentSupportsSnooze, readEnvironmentSupportsTitleRegeneration, + readEnvironmentSupportsViewStatus, readThreadShell, } from "../state/entities"; import { readLocalApi } from "../localApi"; -import { useUiStateStore } from "../uiStateStore"; import { useCopyToClipboard } from "./useCopyToClipboard"; import { useNewThreadHandler } from "./useHandleNewThread"; import { useClientSettings } from "./useSettings"; @@ -78,7 +78,9 @@ export function useThreadActionMenu(input: { reportFailure: false, }); const handleNewThread = useNewThreadHandler(); - const markThreadUnread = useUiStateStore((s) => s.markThreadUnread); + const markThreadUnread = useAtomCommand(threadEnvironment.markUnread, { + reportFailure: false, + }); const autoSettleAfterDays = useClientSettings((s) => s.sidebarAutoSettleAfterDays); const confirmThreadDelete = useClientSettings((s) => s.confirmThreadDelete); const timestampFormat = useClientSettings((s) => s.timestampFormat); @@ -112,6 +114,7 @@ export function useThreadActionMenu(input: { snooze: readEnvironmentSupportsSnooze(threadRef.environmentId), pinning: readEnvironmentSupportsPinning(threadRef.environmentId), titleRegeneration: readEnvironmentSupportsTitleRegeneration(threadRef.environmentId), + viewStatus: readEnvironmentSupportsViewStatus(threadRef.environmentId), }; const isRegeneratingTitle = thread.titleRegeneration != null; const snoozePresets = resolveSnoozePresets(now, timestampFormat); @@ -130,6 +133,7 @@ export function useThreadActionMenu(input: { }), isSnoozed: supports.snooze && effectiveSnoozed(thread, { now: now.toISOString() }), canSnoozeNow: canSnooze(thread, { now: now.toISOString() }), + canMarkUnread: thread.latestTurn?.completedAt != null, isRegeneratingTitle, supports, snoozePresets, @@ -220,7 +224,12 @@ export function useThreadActionMenu(input: { ); return; case "mark-unread": - markThreadUnread(scopedThreadKey(threadRef), thread.latestTurn?.completedAt); + await reportFailure("Failed to mark thread unread", () => + markThreadUnread({ + environmentId: threadRef.environmentId, + input: { threadId: threadRef.threadId }, + }), + ); return; case "copy-path": { const workspacePath = thread.worktreePath ?? projectCwd; diff --git a/apps/web/src/hooks/useThreadActions.ts b/apps/web/src/hooks/useThreadActions.ts index 22548b23360..0deefb292ad 100644 --- a/apps/web/src/hooks/useThreadActions.ts +++ b/apps/web/src/hooks/useThreadActions.ts @@ -2,7 +2,6 @@ import { parseScopedThreadKey, scopeProjectRef, scopeThreadRef, - scopedThreadKey, } from "@t3tools/client-runtime/environment"; import { settlePromise, squashAtomCommandFailure } from "@t3tools/client-runtime/state/runtime"; import { canSettle, canSnooze, threadWokeAt } from "@t3tools/client-runtime/state/thread-settled"; @@ -26,13 +25,13 @@ import { readEnvironmentSupportsPinReorder, readEnvironmentSupportsSettlement, readEnvironmentSupportsSnooze, + readEnvironmentSupportsViewStatus, readEnvironmentThreadRefs, readProject, readThreadShell, readThreadShells, } from "../state/entities"; import { useTerminalUiStateStore } from "../terminalUiStateStore"; -import { useUiStateStore } from "../uiStateStore"; import { buildThreadRouteParams, resolveThreadRouteRef } from "../threadRoutes"; import { formatWorktreePathForDisplay, getOrphanedWorktreePathForThread } from "../worktreeCleanup"; import { stackedThreadToast, toastManager } from "../components/ui/toast"; @@ -167,6 +166,9 @@ export function useThreadActions() { const unsnoozeThreadMutation = useAtomCommand(threadEnvironment.unsnooze, { reportFailure: false, }); + const markThreadViewed = useAtomCommand(threadEnvironment.markViewed, { + reportFailure: false, + }); const stopThreadSession = useAtomCommand(threadEnvironment.stopSession); const removeWorktree = useAtomCommand(vcsEnvironment.removeWorktree, { reportFailure: false, @@ -181,7 +183,6 @@ export function useThreadActions() { (store) => store.clearProjectDraftThreadById, ); const clearTerminalUiState = useTerminalUiStateStore((state) => state.clearTerminalUiState); - const markThreadVisited = useUiStateStore((state) => state.markThreadVisited); const router = useRouter(); const handleNewThread = useNewThreadHandler(); // Keep a ref so archiveThread can call handleNewThread without appearing in @@ -234,8 +235,15 @@ export function useThreadActions() { return archiveResult; } const wokeAt = threadWokeAt(thread, { now: new Date().toISOString() }); - if (wokeAt !== null) { - markThreadVisited(scopedThreadKey(threadRef), wokeAt); + if (wokeAt !== null && readEnvironmentSupportsViewStatus(threadRef.environmentId)) { + void markThreadViewed({ + environmentId: threadRef.environmentId, + input: { + threadId: threadRef.threadId, + viewedAt: wokeAt, + expectedLastViewedAt: thread.lastViewedAt ?? null, + }, + }); } refreshArchivedThreadsForEnvironment(threadRef.environmentId); opts.onArchived?.(); @@ -252,7 +260,7 @@ export function useThreadActions() { return archiveResult; }, - [archiveThreadMutation, getCurrentRouteThreadRef, markThreadVisited, resolveThreadTarget], + [archiveThreadMutation, getCurrentRouteThreadRef, markThreadViewed, resolveThreadTarget], ); const unarchiveThread = useCallback( @@ -504,12 +512,22 @@ export function useThreadActions() { environmentId: target.environmentId, input: { threadId: target.threadId }, }); - if (result._tag === "Success" && wokeAt !== null) { - markThreadVisited(scopedThreadKey(target), wokeAt); + if ( + result._tag === "Success" && + wokeAt !== null && + readEnvironmentSupportsViewStatus(target.environmentId) + ) { + void markThreadViewed({ + environmentId: target.environmentId, + input: { + threadId: target.threadId, + viewedAt: wokeAt, + }, + }); } return result; }, - [markThreadVisited, resolveThreadTarget, settleThreadMutation], + [markThreadViewed, resolveThreadTarget, settleThreadMutation], ); const unsettleThread = useCallback( diff --git a/apps/web/src/state/entities.ts b/apps/web/src/state/entities.ts index 7bca3118237..e3c4b71b254 100644 --- a/apps/web/src/state/entities.ts +++ b/apps/web/src/state/entities.ts @@ -268,6 +268,14 @@ export function readEnvironmentSupportsPinReorder(environmentId: EnvironmentId): ); } +/** Whether the environment server owns synchronized thread read state. */ +export function readEnvironmentSupportsViewStatus(environmentId: EnvironmentId): boolean { + return ( + appAtomRegistry.get(environmentServerConfigsAtom).get(environmentId)?.environment.capabilities + .threadViewStatus === true + ); +} + export function readThreadDetail(ref: ScopedThreadRef): EnvironmentThread | null { return appAtomRegistry.get(environmentThreadDetails.detailAtom(ref)); } diff --git a/apps/web/src/state/threads.ts b/apps/web/src/state/threads.ts index fd936f99ff2..c9b4687ca92 100644 --- a/apps/web/src/state/threads.ts +++ b/apps/web/src/state/threads.ts @@ -15,7 +15,6 @@ import { environmentCatalog } from "../connection/catalog"; import { connectionAtomRuntime } from "../connection/runtime"; import { environmentSnapshotAtom } from "./shell"; -export const threadEnvironment = createThreadEnvironmentAtoms(connectionAtomRuntime); export const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime); export const environmentThreadDetails = createEnvironmentThreadDetailAtoms( environmentThreads.stateAtom, @@ -24,6 +23,9 @@ export const environmentThreadShells = createEnvironmentThreadShellAtoms({ catalogValueAtom: environmentCatalog.catalogValueAtom, snapshotAtom: environmentSnapshotAtom, }); +export const threadEnvironment = createThreadEnvironmentAtoms(connectionAtomRuntime, { + threadShellAtom: environmentThreadShells.threadShellAtom, +}); const EMPTY_THREAD_STATE_ATOM = Atom.make(AsyncResult.success(EMPTY_ENVIRONMENT_THREAD_STATE)).pipe( Atom.withLabel("web-environment-thread:empty"), diff --git a/apps/web/src/uiStateStore.test.ts b/apps/web/src/uiStateStore.test.ts index 30450287353..ac21ae69745 100644 --- a/apps/web/src/uiStateStore.test.ts +++ b/apps/web/src/uiStateStore.test.ts @@ -3,8 +3,6 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test" import { legacyProjectCwdPreferenceKey, - markThreadUnread, - markThreadVisited, parsePersistedState, PERSISTED_STATE_KEY, type PersistedUiState, @@ -21,7 +19,6 @@ function makeUiState(overrides: Partial = {}): UiState { return { projectExpandedById: {}, projectOrder: [], - threadLastVisitedAtById: {}, threadChangedFilesExpandedById: {}, defaultAdvertisedEndpointKey: null, ...overrides, @@ -29,30 +26,6 @@ function makeUiState(overrides: Partial = {}): UiState { } describe("uiStateStore pure functions", () => { - it("stores server timestamps without moving visit state backwards", () => { - const threadId = ThreadId.make("thread-1"); - const initialState = makeUiState(); - const visited = markThreadVisited(initialState, threadId, "2026-02-25T12:30:00.700Z"); - - expect(visited.threadLastVisitedAtById[threadId]).toBe("2026-02-25T12:30:00.700Z"); - expect(markThreadVisited(visited, threadId, "2026-02-25T12:30:00.000Z")).toBe(visited); - expect(markThreadVisited(visited, threadId, "not-a-date")).toBe(visited); - }); - - it("marks a completed thread unread using the server completion timestamp", () => { - const threadId = ThreadId.make("thread-1"); - const initialState = makeUiState({ - threadLastVisitedAtById: { - [threadId]: "2026-02-25T12:35:00.000Z", - }, - }); - - const next = markThreadUnread(initialState, threadId, "2026-02-25T12:30:00.000Z"); - - expect(next.threadLastVisitedAtById[threadId]).toBe("2026-02-25T12:29:59.999Z"); - expect(markThreadUnread(next, threadId, null)).toBe(next); - }); - it("resolves project expansion from logical, physical, and legacy preference keys", () => { const physicalKey = "environment:/repo/project"; const legacyKey = legacyProjectCwdPreferenceKey("/repo/project"); @@ -154,10 +127,6 @@ describe("parsePersistedState", () => { invalid: "no" as unknown as boolean, }, projectOrder: ["physical-b", "", "physical-a", "physical-b"], - threadLastVisitedAtById: { - "environment:thread-1": "2026-02-25T12:35:00.000Z", - invalid: "not-a-date", - }, defaultAdvertisedEndpointKey: "desktop-core:lan:http", threadChangedFilesExpansionVersion: 1, threadChangedFilesExpandedById: { @@ -173,9 +142,6 @@ describe("parsePersistedState", () => { logical: false, }, projectOrder: ["physical-b", "physical-a"], - threadLastVisitedAtById: { - "environment:thread-1": "2026-02-25T12:35:00.000Z", - }, defaultAdvertisedEndpointKey: "desktop-core:lan:http", threadChangedFilesExpandedById: { "environment:thread-1": { @@ -264,15 +230,12 @@ describe("uiStateStore persistence", () => { vi.unstubAllGlobals(); }); - it("persists raw UI preferences including thread visit markers", () => { + it("persists raw UI preferences", () => { const state = makeUiState({ projectExpandedById: { logical: false, }, projectOrder: ["physical-b", "physical-a"], - threadLastVisitedAtById: { - "environment:thread-1": "2026-02-25T12:35:00.000Z", - }, threadChangedFilesExpandedById: { "environment:thread-1": { "turn-1": false, @@ -292,9 +255,6 @@ describe("uiStateStore persistence", () => { logical: false, }, projectOrder: ["physical-b", "physical-a"], - threadLastVisitedAtById: { - "environment:thread-1": "2026-02-25T12:35:00.000Z", - }, defaultAdvertisedEndpointKey: "desktop-core:lan:http", threadChangedFilesExpansionVersion: 1, threadChangedFilesExpandedById: { diff --git a/apps/web/src/uiStateStore.ts b/apps/web/src/uiStateStore.ts index 5d744d540a5..849ef347887 100644 --- a/apps/web/src/uiStateStore.ts +++ b/apps/web/src/uiStateStore.ts @@ -20,7 +20,6 @@ const LEGACY_PERSISTED_STATE_KEYS = [ export interface PersistedUiState { projectExpandedById?: Record; projectOrder?: string[]; - threadLastVisitedAtById?: Record; collapsedProjectCwds?: string[]; expandedProjectCwds?: string[]; projectOrderCwds?: string[]; @@ -35,7 +34,6 @@ export interface UiProjectState { } export interface UiThreadState { - threadLastVisitedAtById: Record; threadChangedFilesExpandedById: Record>; } @@ -48,7 +46,6 @@ export interface UiState extends UiProjectState, UiThreadState, UiEndpointState const initialState: UiState = { projectExpandedById: {}, projectOrder: [], - threadLastVisitedAtById: {}, threadChangedFilesExpandedById: {}, defaultAdvertisedEndpointKey: null, }; @@ -83,21 +80,6 @@ function sanitizeBooleanRecord(value: unknown): Record { ); } -function sanitizeTimestampRecord(value: unknown): Record { - if (!value || typeof value !== "object") { - return {}; - } - return Object.fromEntries( - Object.entries(value).filter( - (entry): entry is [string, string] => - entry[0].length > 0 && - typeof entry[1] === "string" && - entry[1].length > 0 && - Number.isFinite(Date.parse(entry[1])), - ), - ); -} - export function parsePersistedState(parsed: PersistedUiState): UiState { const projectExpandedById = parsed.projectExpandedById === undefined @@ -125,7 +107,6 @@ export function parsePersistedState(parsed: PersistedUiState): UiState { return { projectExpandedById, projectOrder, - threadLastVisitedAtById: sanitizeTimestampRecord(parsed.threadLastVisitedAtById), threadChangedFilesExpandedById: parsed.threadChangedFilesExpansionVersion === THREAD_CHANGED_FILES_EXPANSION_VERSION ? sanitizePersistedThreadChangedFilesExpanded(parsed.threadChangedFilesExpandedById) @@ -203,7 +184,6 @@ export function persistState(state: UiState): void { JSON.stringify({ projectExpandedById, projectOrder: state.projectOrder, - threadLastVisitedAtById: state.threadLastVisitedAtById, defaultAdvertisedEndpointKey: state.defaultAdvertisedEndpointKey, threadChangedFilesExpansionVersion: THREAD_CHANGED_FILES_EXPANSION_VERSION, threadChangedFilesExpandedById: state.threadChangedFilesExpandedById, @@ -222,54 +202,6 @@ export function persistState(state: UiState): void { const debouncedPersistState = new Debouncer(persistState, { wait: 500 }); -export function markThreadVisited(state: UiState, threadId: string, visitedAt: string): UiState { - const visitedAtMs = Date.parse(visitedAt); - if (!Number.isFinite(visitedAtMs)) { - return state; - } - const previousVisitedAt = state.threadLastVisitedAtById[threadId]; - const previousVisitedAtMs = previousVisitedAt ? Date.parse(previousVisitedAt) : NaN; - if ( - Number.isFinite(previousVisitedAtMs) && - Number.isFinite(visitedAtMs) && - previousVisitedAtMs >= visitedAtMs - ) { - return state; - } - return { - ...state, - threadLastVisitedAtById: { - ...state.threadLastVisitedAtById, - [threadId]: visitedAt, - }, - }; -} - -export function markThreadUnread( - state: UiState, - threadId: string, - latestTurnCompletedAt: string | null | undefined, -): UiState { - if (!latestTurnCompletedAt) { - return state; - } - const latestTurnCompletedAtMs = Date.parse(latestTurnCompletedAt); - if (Number.isNaN(latestTurnCompletedAtMs)) { - return state; - } - const unreadVisitedAt = new Date(latestTurnCompletedAtMs - 1).toISOString(); - if (state.threadLastVisitedAtById[threadId] === unreadVisitedAt) { - return state; - } - return { - ...state, - threadLastVisitedAtById: { - ...state.threadLastVisitedAtById, - [threadId]: unreadVisitedAt, - }, - }; -} - export function setThreadChangedFilesExpanded( state: UiState, threadId: string, @@ -382,8 +314,6 @@ export function reorderProjects( } interface UiStateStore extends UiState { - markThreadVisited: (threadId: string, visitedAt: string) => void; - markThreadUnread: (threadId: string, latestTurnCompletedAt: string | null | undefined) => void; setThreadChangedFilesExpanded: (threadId: string, turnId: string, expanded: boolean) => void; setDefaultAdvertisedEndpointKey: (key: string | null) => void; setProjectExpanded: (projectIds: string | readonly string[], expanded: boolean) => void; @@ -396,10 +326,6 @@ interface UiStateStore extends UiState { export const useUiStateStore = create((set) => ({ ...readPersistedState(), - markThreadVisited: (threadId, visitedAt) => - set((state) => markThreadVisited(state, threadId, visitedAt)), - markThreadUnread: (threadId, latestTurnCompletedAt) => - set((state) => markThreadUnread(state, threadId, latestTurnCompletedAt)), setThreadChangedFilesExpanded: (threadId, turnId, expanded) => set((state) => setThreadChangedFilesExpanded(state, threadId, turnId, expanded)), setDefaultAdvertisedEndpointKey: (key) => diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index 0cb1650066c..4883a5e11e1 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -24,6 +24,8 @@ import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; import { archiveThread, createProject, + markThreadUnread, + markThreadViewed, settleThread, stopThreadSession, unsettleThread, @@ -171,4 +173,39 @@ describe("environment commands", () => { ]); }).pipe(Effect.provide(TEST_CRYPTO_LAYER)), ); + + it.effect("dispatches server-owned thread view-status commands", () => + Effect.gen(function* () { + const dispatched: ClientOrchestrationCommand[] = []; + const supervisor = yield* makeSupervisor(dispatched); + + yield* markThreadViewed({ + commandId: CommandId.make("viewed-command"), + threadId: ThreadId.make("thread-1"), + viewedAt: "2026-08-07T12:00:00.000Z", + expectedLastViewedAt: null, + supersededViewedAt: null, + }).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor)); + yield* markThreadUnread({ + commandId: CommandId.make("unread-command"), + threadId: ThreadId.make("thread-1"), + }).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor)); + + expect(dispatched).toEqual([ + { + type: "thread.mark-viewed", + commandId: "viewed-command", + threadId: "thread-1", + viewedAt: "2026-08-07T12:00:00.000Z", + expectedLastViewedAt: null, + supersededViewedAt: null, + }, + { + type: "thread.mark-unread", + commandId: "unread-command", + threadId: "thread-1", + }, + ]); + }).pipe(Effect.provide(TEST_CRYPTO_LAYER)), + ); }); diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index cb74f117b77..f0d2d878889 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -35,6 +35,8 @@ export type CreateThreadInput = CommandInput<"thread.create">; export type DeleteThreadInput = CommandInput<"thread.delete">; export type ArchiveThreadInput = CommandInput<"thread.archive">; export type UnarchiveThreadInput = CommandInput<"thread.unarchive">; +export type MarkThreadViewedInput = CommandInput<"thread.mark-viewed">; +export type MarkThreadUnreadInput = CommandInput<"thread.mark-unread">; export type SettleThreadInput = CommandInput<"thread.settle">; export type UnsettleThreadInput = CommandInput<"thread.unsettle">; export type SnoozeThreadInput = CommandInput<"thread.snooze">; @@ -160,6 +162,26 @@ export const unarchiveThread: (input: UnarchiveThreadInput) => CommandEffect = E }); }); +export const markThreadViewed: (input: MarkThreadViewedInput) => CommandEffect = Effect.fn( + "EnvironmentCommands.markThreadViewed", +)(function* (input) { + return yield* dispatch({ + ...input, + type: "thread.mark-viewed", + commandId: yield* commandId(input), + }); +}); + +export const markThreadUnread: (input: MarkThreadUnreadInput) => CommandEffect = Effect.fn( + "EnvironmentCommands.markThreadUnread", +)(function* (input) { + return yield* dispatch({ + ...input, + type: "thread.mark-unread", + commandId: yield* commandId(input), + }); +}); + export const settleThread: (input: SettleThreadInput) => CommandEffect = Effect.fn( "EnvironmentCommands.settleThread", )(function* (input) { diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index ed3537e4f83..7fcb07f01e4 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -1,5 +1,6 @@ +import type { EnvironmentId, ScopedThreadRef } from "@t3tools/contracts"; import * as Crypto from "effect/Crypto"; -import { Atom } from "effect/unstable/reactivity"; +import { Atom, AtomRegistry } from "effect/unstable/reactivity"; import { createAtomCommandScheduler, createEnvironmentCommand } from "./runtime.ts"; import { @@ -7,6 +8,8 @@ import { type CreateThreadInput, type DeleteThreadInput, type InterruptThreadTurnInput, + type MarkThreadUnreadInput, + type MarkThreadViewedInput as DispatchMarkThreadViewedInput, type RespondToThreadApprovalInput, type RespondToThreadUserInputInput, type RevertThreadCheckpointInput, @@ -27,6 +30,8 @@ import { createThread, deleteThread, interruptThreadTurn, + markThreadUnread, + markThreadViewed, respondToThreadApproval, respondToThreadUserInput, revertThreadCheckpoint, @@ -45,12 +50,14 @@ import { updateThreadMetadata, } from "../operations/commands.ts"; import type { EnvironmentRegistry } from "../connection/registry.ts"; +import type { EnvironmentThreadShell } from "./models.ts"; export type { ArchiveThreadInput, CreateThreadInput, DeleteThreadInput, InterruptThreadTurnInput, + MarkThreadUnreadInput, RespondToThreadApprovalInput, RespondToThreadUserInputInput, RevertThreadCheckpointInput, @@ -69,8 +76,18 @@ export type { UpdateThreadMetadataInput, } from "../operations/commands.ts"; +export type MarkThreadViewedInput = Omit< + DispatchMarkThreadViewedInput, + "expectedLastViewedAt" | "supersededViewedAt" +> & { + readonly expectedLastViewedAt?: DispatchMarkThreadViewedInput["expectedLastViewedAt"]; +}; + export function createThreadEnvironmentAtoms( runtime: Atom.AtomRuntime, + options: { + readonly threadShellAtom: (ref: ScopedThreadRef) => Atom.Atom; + }, ) { const scheduler = createAtomCommandScheduler(); const concurrency = { @@ -78,6 +95,49 @@ export function createThreadEnvironmentAtoms( key: ({ environmentId, input }: { environmentId: string; input: { threadId: string } }) => JSON.stringify([environmentId, input.threadId]), }; + const queuedViewedAtByThread = new Map(); + const dispatchMarkViewed = createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:mark-viewed", + execute: (input: DispatchMarkThreadViewedInput) => markThreadViewed(input), + scheduler, + concurrency, + }); + const markViewed = { + label: dispatchMarkViewed.label, + run: ( + registry: AtomRegistry.AtomRegistry, + target: { readonly environmentId: EnvironmentId; readonly input: MarkThreadViewedInput }, + ) => { + const { expectedLastViewedAt: expectedOverride, ...input } = target.input; + const threadKey = JSON.stringify([target.environmentId, input.threadId]); + const supersededViewedAt = queuedViewedAtByThread.get(threadKey) ?? null; + if ( + supersededViewedAt === null || + Date.parse(input.viewedAt) > Date.parse(supersededViewedAt) + ) { + queuedViewedAtByThread.set(threadKey, input.viewedAt); + } + const expectedLastViewedAt = + expectedOverride !== undefined + ? expectedOverride + : (registry.get( + options.threadShellAtom({ + environmentId: target.environmentId, + threadId: input.threadId, + }), + )?.lastViewedAt ?? null); + return dispatchMarkViewed + .run(registry, { + environmentId: target.environmentId, + input: { ...input, expectedLastViewedAt, supersededViewedAt }, + }) + .finally(() => { + if (queuedViewedAtByThread.get(threadKey) === input.viewedAt) { + queuedViewedAtByThread.delete(threadKey); + } + }); + }, + }; return { create: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:create", @@ -103,6 +163,13 @@ export function createThreadEnvironmentAtoms( scheduler, concurrency, }), + markViewed, + markUnread: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:mark-unread", + execute: (input: MarkThreadUnreadInput) => markThreadUnread(input), + scheduler, + concurrency, + }), settle: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:settle", execute: (input: SettleThreadInput) => settleThread(input), diff --git a/packages/client-runtime/src/state/threadDetail.ts b/packages/client-runtime/src/state/threadDetail.ts index 5a2ffa442e0..94220137a01 100644 --- a/packages/client-runtime/src/state/threadDetail.ts +++ b/packages/client-runtime/src/state/threadDetail.ts @@ -59,6 +59,7 @@ export function mergeEnvironmentThread( archivedAt: shell.archivedAt, settledOverride: shell.settledOverride, settledAt: shell.settledAt, + lastViewedAt: shell.lastViewedAt, snoozedUntil: shell.snoozedUntil, snoozedAt: shell.snoozedAt, pinnedAt: shell.pinnedAt, diff --git a/packages/client-runtime/src/state/threadReducer.test.ts b/packages/client-runtime/src/state/threadReducer.test.ts index 8b2479c7a34..632981fb824 100644 --- a/packages/client-runtime/src/state/threadReducer.test.ts +++ b/packages/client-runtime/src/state/threadReducer.test.ts @@ -231,6 +231,25 @@ describe("applyThreadDetailEvent", () => { }); }); + it("applies synchronized thread view status without changing updatedAt", () => { + const lastViewedAt = "2026-04-01T06:30:00.000Z"; + const result = applyThreadDetailEvent(baseThread, { + ...baseEventFields, + sequence: 7, + occurredAt: lastViewedAt, + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + type: "thread.view-status-updated", + payload: { threadId: ThreadId.make("thread-1"), lastViewedAt }, + }); + + expect(result.kind).toBe("updated"); + if (result.kind === "updated") { + expect(result.thread.lastViewedAt).toBe(lastViewedAt); + expect(result.thread.updatedAt).toBe(baseThread.updatedAt); + } + }); + describe("thread.pinned / thread.unpinned", () => { it("sets pinnedAt", () => { const pinnedAt = "2026-04-01T05:00:00.000Z"; diff --git a/packages/client-runtime/src/state/threadReducer.ts b/packages/client-runtime/src/state/threadReducer.ts index 970fd94b1a1..5de9403ff7e 100644 --- a/packages/client-runtime/src/state/threadReducer.ts +++ b/packages/client-runtime/src/state/threadReducer.ts @@ -92,6 +92,7 @@ export function applyThreadDetailEvent( archivedAt: null, settledOverride: null, settledAt: null, + lastViewedAt: event.payload.lastViewedAt ?? null, snoozedUntil: null, snoozedAt: null, deletedAt: null, @@ -123,6 +124,12 @@ export function applyThreadDetailEvent( thread: { ...thread, archivedAt: null, updatedAt: event.payload.updatedAt }, }; + case "thread.view-status-updated": + return { + kind: "updated", + thread: { ...thread, lastViewedAt: event.payload.lastViewedAt }, + }; + case "thread.settled": return { kind: "updated", diff --git a/packages/client-runtime/src/state/threadSettled.ts b/packages/client-runtime/src/state/threadSettled.ts index 595b1303bea..34b9be61ab4 100644 --- a/packages/client-runtime/src/state/threadSettled.ts +++ b/packages/client-runtime/src/state/threadSettled.ts @@ -183,7 +183,7 @@ export function effectiveSnoozed( * still snoozed. Used for the "Woke" indicator: the thread reappears in its * original sort position (the inbox sort is deliberately static), so the * wake signal has to carry the weight. Compare against the client's - * lastVisitedAt — visiting clears the indicator like it clears unread. + * server-owned lastViewedAt — viewing clears the indicator like it clears unread. * * Timer wakes report the wake time itself; raised-hand wakes report the * triggering timestamp so a visit BEFORE the early wake doesn't suppress @@ -199,7 +199,7 @@ export function threadWokeAt( // An early hand-raise wake stays authoritative even after the scheduled // wake time passes: reporting snoozedUntil then would resurface a Woke // indicator the user already cleared by visiting (snoozedUntil is newer - // than that visit's lastVisitedAt). + // than that view's lastViewedAt). if (threadRaisedHandWhileSnoozed(shell)) { if ( shell.snoozedAt != null && diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 329ff911503..884a73e714d 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -53,6 +53,9 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ /** Server understands thread.pin.reorder (and orderKey on thread.pin). Same version-skew contract as threadSettlement. */ threadPinReorder: Schema.optionalKey(Schema.Boolean), + /** Server owns thread read/unread state and understands + thread.mark-viewed / thread.mark-unread commands. */ + threadViewStatus: Schema.optionalKey(Schema.Boolean), /** Server understands regenerateTitle on thread.meta.update. Absent on older servers, so clients hide the action instead of sending it. */ threadTitleRegeneration: Schema.optionalKey(Schema.Boolean), diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 87270d98c1f..3b73a8009fb 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -369,6 +369,9 @@ export const OrchestrationThread = Schema.Struct({ Schema.withDecodingDefault(Effect.succeed(null)), ), settledAt: Schema.NullOr(IsoDateTime).pipe(Schema.withDecodingDefault(Effect.succeed(null))), + // Optional so threads from servers predating synchronized read state still decode. + // Missing means read-by-default, avoiding an unread flood on upgrade. + lastViewedAt: Schema.optional(Schema.NullOr(IsoDateTime)), // Snooze is an overlay on the active lifecycle, not a fourth destination: // a snoozed thread stays "active" in the model and is only suppressed from // the inbox until snoozedUntil passes (or the thread raises its hand). @@ -436,6 +439,7 @@ export const OrchestrationThreadShell = Schema.Struct({ Schema.withDecodingDefault(Effect.succeed(null)), ), settledAt: Schema.NullOr(IsoDateTime).pipe(Schema.withDecodingDefault(Effect.succeed(null))), + lastViewedAt: Schema.optional(Schema.NullOr(IsoDateTime)), snoozedUntil: Schema.optional(Schema.NullOr(IsoDateTime)), snoozedAt: Schema.optional(Schema.NullOr(IsoDateTime)), pinnedAt: Schema.optional(Schema.NullOr(IsoDateTime)), @@ -667,6 +671,27 @@ const ThreadUnarchiveCommand = Schema.Struct({ threadId: ThreadId, }); +const ThreadMarkViewedCommand = Schema.Struct({ + type: Schema.Literal("thread.mark-viewed"), + commandId: CommandId, + threadId: ThreadId, + // A boundary copied from server-owned thread state (completion or wake). + // It is not the client's wall clock. + viewedAt: IsoDateTime, + // Captured when the client queues the view, so it cannot overwrite a newer + // explicit unread action. + expectedLastViewedAt: Schema.NullOr(IsoDateTime), + // The preceding queued view boundary. This lets serialized completion views + // advance in order without treating an explicit unread as their predecessor. + supersededViewedAt: Schema.NullOr(IsoDateTime), +}); + +const ThreadMarkUnreadCommand = Schema.Struct({ + type: Schema.Literal("thread.mark-unread"), + commandId: CommandId, + threadId: ThreadId, +}); + const ThreadSettleCommand = Schema.Struct({ type: Schema.Literal("thread.settle"), commandId: CommandId, @@ -880,6 +905,8 @@ const DispatchableClientOrchestrationCommand = Schema.Union([ ThreadDeleteCommand, ThreadArchiveCommand, ThreadUnarchiveCommand, + ThreadMarkViewedCommand, + ThreadMarkUnreadCommand, ThreadSettleCommand, ThreadUnsettleCommand, ThreadSnoozeCommand, @@ -908,6 +935,8 @@ export const ClientOrchestrationCommand = Schema.Union([ ThreadDeleteCommand, ThreadArchiveCommand, ThreadUnarchiveCommand, + ThreadMarkViewedCommand, + ThreadMarkUnreadCommand, ThreadSettleCommand, ThreadUnsettleCommand, ThreadSnoozeCommand, @@ -1026,6 +1055,7 @@ export const OrchestrationEventType = Schema.Literals([ "thread.deleted", "thread.archived", "thread.unarchived", + "thread.view-status-updated", "thread.settled", "thread.unsettled", "thread.snoozed", @@ -1092,6 +1122,7 @@ export const ThreadCreatedPayload = Schema.Struct({ ), branch: Schema.NullOr(TrimmedNonEmptyString), worktreePath: Schema.NullOr(TrimmedNonEmptyString), + lastViewedAt: Schema.optional(IsoDateTime), createdAt: IsoDateTime, updatedAt: IsoDateTime, }); @@ -1112,6 +1143,11 @@ export const ThreadUnarchivedPayload = Schema.Struct({ updatedAt: IsoDateTime, }); +export const ThreadViewStatusUpdatedPayload = Schema.Struct({ + threadId: ThreadId, + lastViewedAt: IsoDateTime, +}); + export const ThreadSettledPayload = Schema.Struct({ threadId: ThreadId, settledAt: IsoDateTime, @@ -1335,6 +1371,11 @@ export const OrchestrationEvent = Schema.Union([ type: Schema.Literal("thread.unarchived"), payload: ThreadUnarchivedPayload, }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("thread.view-status-updated"), + payload: ThreadViewStatusUpdatedPayload, + }), Schema.Struct({ ...EventBaseFields, type: Schema.Literal("thread.settled"),