Skip to content

feat(server): sync thread read status across clients #5652

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
t3dotgg wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(server): sync thread read status across clients #5652

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 28 additions & 0 deletions apps/mobile/src/features/threads/ThreadRouteScreen.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
},
});
Comment thread
cursor[bot] marked this conversation as resolved.
}, [
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);
Expand Down
4 changes: 3 additions & 1 deletion apps/mobile/src/state/threads.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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"),
Expand Down
1 change: 1 addition & 0 deletions apps/server/src/environment/ServerEnvironment.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 } : {}),
Expand Down
34 changes: 34 additions & 0 deletions apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
},
]);
}),
);
});
Expand Down
13 changes: 13 additions & 0 deletions apps/server/src/orchestration/Layers/ProjectionPipeline.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
2 changes: 2 additions & 0 deletions apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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",
Expand Down
10 changes: 10 additions & 0 deletions apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
2 changes: 2 additions & 0 deletions apps/server/src/orchestration/Schemas.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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;
Expand Down
63 changes: 63 additions & 0 deletions apps/server/src/orchestration/decider.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
},
Expand Down Expand Up @@ -445,6 +446,68 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"
};
}

case "thread.mark-viewed": {
Comment thread
macroscopeapp[bot] marked this conversation as resolved.
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 },
};
Comment thread
cursor[bot] marked this conversation as resolved.
}

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,
Expand Down
Loading
Loading