Skip to content

fix(web): switching threads no longer delays the next thread's visit by up to 10s #25368

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

Merged
juliusmarminge merged 1 commit into from
Oct 26, 2026
Merged

fix(web): switching threads no longer delays the next thread's visit by up to 10s #25368

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
82 changes: 82 additions & 0 deletions apps/web/src/components/chat/ThreadVisitWatermark.browser.test.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
import { claimThreadOpen } from "@t3tools/client-runtime/state/thread-search";
import { EnvironmentId, type ScopedThreadRef, ThreadId } from "@t3tools/contracts";
import { AsyncResult, Atom } from "effect/reactivity";
import { expect, it, vi } from "vite-plus/test";

import { fakeTimers, render } from "../../../test/render";
import { AppAtomRegistryProvider, appAtomRegistry } from "../../rpc/atomRegistry";

const mocks = vi.hoisted(() => ({
visitRun: vi.fn(),
}));

const shellKey = (ref: ScopedThreadRef) => `${ref.environmentId}:${ref.threadId}`;

vi.mock(import("../../state/threads"), async (original) => {
const actual = await original();
return {
...actual,
environmentThreadShells: {
...actual.environmentThreadShells,
threadShellAtom: ((ref: ScopedThreadRef) => shellAtom(shellKey(ref))) as never,
},
threadEnvironment: {
...actual.threadEnvironment,
visit: { label: "visit", run: mocks.visitRun } as never,
},
};
});

import { ThreadVisitWatermark } from "./ThreadVisitWatermark";

// Kept alive so a shell set before its tracker mounts is still there when it does.
const shellAtom = Atom.family((_key: string) => Atom.make<unknown>(null).pipe(Atom.keepAlive));

const NOW = Date.parse("2026-01-04T00:00:00.000Z");
const ENVIRONMENT = EnvironmentId.make("environment-visit-watermark");
const threadA: ScopedThreadRef = { environmentId: ENVIRONMENT, threadId: ThreadId.make("a") };
const threadB: ScopedThreadRef = { environmentId: ENVIRONMENT, threadId: ThreadId.make("b") };

function setShell(ref: ScopedThreadRef, updatedAt: string) {
appAtomRegistry.set(shellAtom(shellKey(ref)), {
environmentId: ref.environmentId,
id: ref.threadId,
updatedAt,
lastVisitedAt: "2026-01-01T00:00:00.000Z",
latestRun: null,
});
}

function visitedThreadIds() {
return mocks.visitRun.mock.calls.map(
([, input]) => (input as { input: { threadId: string } }).input.threadId,
);
}

it("records the visit of the next thread at once after switching threads", async () => {
mocks.visitRun.mockReset();
mocks.visitRun.mockResolvedValue(AsyncResult.success(undefined));
setShell(threadA, "2026-01-02T00:00:00.000Z");
setShell(threadB, "2026-01-03T00:00:00.000Z");
using _timers = fakeTimers();
vi.setSystemTime(NOW);
// A palette pick already sent B's open with its query, so B's visit is a plain watermark advance.
claimThreadOpen(threadB.environmentId, threadB.threadId, NOW);

await using screen = await render(
<AppAtomRegistryProvider>
<ThreadVisitWatermark threadRef={threadA} />
</AppAtomRegistryProvider>,
);
expect(visitedThreadIds()).toEqual(["a"]);

await screen.rerender(
<AppAtomRegistryProvider>
<ThreadVisitWatermark threadRef={threadB} />
</AppAtomRegistryProvider>,
);

// Thread A's throttle must not hold back B's first watermark: leaving B
// within the throttle window would otherwise keep it unread.
expect(visitedThreadIds()).toEqual(["a", "b"]);
});
123 changes: 14 additions & 109 deletions apps/web/src/components/chat/ThreadVisitWatermark.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,117 +1,22 @@
import { useAtomMount } from "@effect/atom-react";
import { scopedThreadKey } from "@t3tools/client-runtime/environment";
import { createThreadVisitTracker } from "@t3tools/client-runtime/state/thread-visits";
import { type ScopedThreadRef } from "@t3tools/contracts";
import { resolveThreadLastVisitedAt } from "../Sidebar.logic";
import { scopedThreadKey, scopeThreadRef } from "@t3tools/client-runtime/environment";
import { claimThreadOpen } from "@t3tools/client-runtime/state/thread-search";
import { useEffect, useRef } from "react";

import { useUiStateStore } from "../../uiStateStore";
import { threadEnvironment } from "../../state/threads";
import { useThreadShell } from "../../state/entities";
import { useAtomCommand } from "../../state/use-atom-command";
import { environmentThreadShells, threadEnvironment } from "../../state/threads";

// During an active turn the thread's updatedAt advances several times per
// second, and every server-side visit is a full command dispatch plus a
// broadcast to all shell subscribers.
const VISIT_DISPATCH_THROTTLE_MS = 10_000;
const threadVisitTracker = createThreadVisitTracker({
threadShellAtom: environmentThreadShells.threadShellAtom,
visit: threadEnvironment.visit,
markLocalVisit: (ref, visitedAt) =>
useUiStateStore.getState().markThreadVisited(scopedThreadKey(ref), visitedAt),
});

// The visit watermark is the one part of the chat view that follows the
// shell's updatedAt, which ticks while a run streams. Keeping it in a leaf lets
// ChatView read the quiet shell.
// shell's updatedAt, which ticks while a run streams. The tracker subscribes to
// it outside React, so ChatView keeps reading the quiet shell.
export function ThreadVisitWatermark(props: { readonly threadRef: ScopedThreadRef }) {
const serverThread = useThreadShell(props.threadRef);
const threadKey = scopedThreadKey(props.threadRef);
const markThreadVisited = useUiStateStore((store) => store.markThreadVisited);
const activeThreadLocalLastVisitedAt = useUiStateStore(
(store) => store.threadLastVisitedAtById[threadKey],
);
const visitThreadMutation = useAtomCommand(threadEnvironment.visit, { reportFailure: false });
const lastDispatchedVisitRef = useRef<string | null>(null);
const lastVisitDispatchAtRef = useRef(0);
const recordedOpenRef = useRef<string | null>(null);

// Opening a thread is one visit that also counts toward search ranking. A
// palette pick already sent it with the query. It runs before the watermark
// effect, which then sees this watermark as sent.
useEffect(() => {
if (!serverThread?.id) return;
if (recordedOpenRef.current === threadKey) return;
recordedOpenRef.current = threadKey;
// Recorded here even when the server cannot take opens, so recents follow it.
const claimed = claimThreadOpen(serverThread.environmentId, serverThread.id, Date.now());
if (!claimed || serverThread.lastVisitedAt === undefined) return;
lastDispatchedVisitRef.current = `${threadKey}:${serverThread.updatedAt}`;
lastVisitDispatchAtRef.current = Date.now();
void visitThreadMutation({
environmentId: serverThread.environmentId,
input: {
threadId: serverThread.id,
visitedAt: serverThread.updatedAt,
opened: true,
},
});
}, [
threadKey,
serverThread?.environmentId,
serverThread?.id,
serverThread?.lastVisitedAt,
serverThread?.updatedAt,
visitThreadMutation,
]);

useEffect(() => {
if (!serverThread?.id) return;
const threadUpdatedAt = Date.parse(serverThread.updatedAt);
if (Number.isNaN(threadUpdatedAt)) return;
const effectiveLastVisitedAt = resolveThreadLastVisitedAt(
serverThread.lastVisitedAt,
activeThreadLocalLastVisitedAt,
);
const lastVisitedAt = effectiveLastVisitedAt ? Date.parse(effectiveLastVisitedAt) : NaN;
if (!Number.isNaN(lastVisitedAt) && lastVisitedAt >= threadUpdatedAt) return;

if (serverThread.lastVisitedAt !== undefined) {
// Dedupe per watermark — the effect re-runs before the command's echo
// lands. The dedupe also keeps a mark-unread on the open thread sticky:
// the rewind leaves updatedAt untouched, so the already-dispatched key
// skips a fresh visit until new activity lands or the thread is reopened.
const dispatchKey = `${threadKey}:${serverThread.updatedAt}`;
if (lastDispatchedVisitRef.current === dispatchKey) return;
const dispatch = () => {
lastDispatchedVisitRef.current = dispatchKey;
lastVisitDispatchAtRef.current = Date.now();
void visitThreadMutation({
environmentId: serverThread.environmentId,
input: { threadId: serverThread.id, visitedAt: serverThread.updatedAt },
});
};
const latestRunCompletedAtMs = serverThread.latestRun?.completedAt
? Date.parse(serverThread.latestRun.completedAt)
: NaN;
const hasUnseenCompletion =
!Number.isNaN(latestRunCompletedAtMs) &&
(Number.isNaN(lastVisitedAt) || latestRunCompletedAtMs > lastVisitedAt);
const elapsed = Date.now() - lastVisitDispatchAtRef.current;
if (hasUnseenCompletion || elapsed >= VISIT_DISPATCH_THROTTLE_MS) {
dispatch();
return;
}
const timer = setTimeout(dispatch, VISIT_DISPATCH_THROTTLE_MS - elapsed);
return () => clearTimeout(timer);
}

markThreadVisited(
scopedThreadKey(scopeThreadRef(serverThread.environmentId, serverThread.id)),
serverThread.updatedAt,
);
}, [
activeThreadLocalLastVisitedAt,
markThreadVisited,
threadKey,
serverThread?.environmentId,
serverThread?.id,
serverThread?.lastVisitedAt,
serverThread?.latestRun?.completedAt,
serverThread?.updatedAt,
visitThreadMutation,
]);
useAtomMount(threadVisitTracker(props.threadRef));
return null;
}
Loading