Deepen conversation fold/feed/bus modules from architecture review

apps/cli/src/queue/bus.test.ts

@@ -94,4 +94,26 @@ describe("thread event bus", () => {
 		expect(b.sub).toBeUndefined();
 		expect(await reader.next()).toMatchObject({ seq: 6, sub: 1 });
 	});
+
+	test("evicts the oldest persisted chunk once a turn's log exceeds its bound", async () => {
+		const bus = createThreadEventBus({ maxChunksPerTurn: 3 });
+		bus.watch("peer-1", "thread-1");
+		const first = createReader(bus.subscribe("peer-1"));
+
+		bus.publish(persistedChunk(1));
+		bus.publish(persistedChunk(2));
+		bus.publish(persistedChunk(3));
+		bus.publish(persistedChunk(4));
+		await first.next();
+		await first.next();
+		await first.next();
+		await first.next();
+
+		bus.watch("peer-2", "thread-1");
+		const second = createReader(bus.subscribe("peer-2"));
+
+		expect(await second.next()).toMatchObject({ seq: 2 });
+		expect(await second.next()).toMatchObject({ seq: 3 });
+		expect(await second.next()).toMatchObject({ seq: 4 });
+	});
 });

apps/web/src/components/chat/work-log/diff-row.tsx

@@ -11,10 +11,7 @@ export function DiffRow({ diff }: { diff: DiffView }) {
 		<div className="overflow-hidden rounded-md border border-border/60 bg-card text-xs">
 			<button
 				className="flex w-full items-center gap-2 px-2.5 py-1.5 text-left"
-				onClick={(event) => {
-					event.stopPropagation();
-					setOpen((v) => !v);
-				}}
+				onClick={() => setOpen((v) => !v)}
 				type="button"
 			>
 				<Show fallback={<ChevronRightIcon className="size-3" />} when={open}>

apps/web/src/components/chat/work-log/tool-row.test.tsx

@@ -36,4 +36,18 @@ describe("ToolRow", () => {
 
 		expect(screen.getByText("a.ts")).toBeInTheDocument();
 	});
+
+	test("clicking inside an open diff body does not close the parent tool row", async () => {
+		const user = userEvent.setup();
+		render(<ToolRow tool={tool} />);
+
+		await user.click(screen.getByText("Edit a.ts"));
+		await user.click(screen.getByText("a.ts"));
+
+		const diffBody = document.querySelector(".diff-render-surface");
+		expect(diffBody).not.toBeNull();
+		if (diffBody) await user.click(diffBody);
+
+		expect(screen.getByText("a.ts")).toBeInTheDocument();
+	});
 });

apps/web/src/components/chat/work-log/tool-row.tsx

@@ -61,27 +61,27 @@ export function ToolRow({ tool }: { tool: ToolCallView }) {
 		tool.status === "pending" || tool.status === "in_progress";
 
 	return (
-		<div
-			className={cn(
-				"flex flex-col rounded-md px-0.5 py-0.5 text-xs transition-colors",
-				canExpand &&
-					"cursor-pointer hover:bg-accent/20 focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-ring/70 focus-visible:ring-inset"
-			)}
-			{...(canExpand
-				? {
-						role: "button" as const,
-						tabIndex: 0,
-						onClick: () => setOpen((value) => !value),
-						onKeyDown: (event: KeyboardEvent<HTMLDivElement>) => {
-							if (event.key === "Enter" || event.key === " ") {
-								event.preventDefault();
-								setOpen((value) => !value);
-							}
-						},
-					}
-				: {})}
-		>
-			<div className="flex select-none items-center gap-1.5">
+		<div className="flex flex-col rounded-md px-0.5 py-0.5 text-xs transition-colors">
+			<div
+				className={cn(
+					"flex select-none items-center gap-1.5 rounded-sm",
+					canExpand &&
+						"cursor-pointer hover:bg-accent/20 focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-ring/70 focus-visible:ring-inset"
+				)}
+				{...(canExpand
+					? {
+							role: "button" as const,
+							tabIndex: 0,
+							onClick: () => setOpen((value) => !value),
+							onKeyDown: (event: KeyboardEvent<HTMLDivElement>) => {
+								if (event.key === "Enter" || event.key === " ") {
+									event.preventDefault();
+									setOpen((value) => !value);
+								}
+							},
+						}
+					: {})}
+			>
 				<span className="flex size-5 shrink-0 items-center justify-center text-muted-foreground/65">
 					<Icon className="size-3.5 stroke-[1.8] opacity-80" />
 				</span>

shared/hooks/src/conversation/conversation-cache.test.ts

@@ -1,7 +1,11 @@
 import { RTC_OPERATION_KEYS } from "@cyrus/constants/operation-keys";
+import { waitForTurnEnd } from "@cyrus/utils/conversations/turn-waiters";
 import { QueryClient } from "@tanstack/react-query";
 import { describe, expect, test } from "vitest";
-import { appendOptimisticUserMessage } from "./conversation-cache";
+import {
+	appendOptimisticUserMessage,
+	appendTurnTerminal,
+} from "./conversation-cache";
 
 describe("conversation cache", () => {
 	test("appends an optimistic user message to the conversations cache", () => {
@@ -23,4 +27,16 @@ describe("conversation cache", () => {
 			content: "hello",
 		});
 	});
+
+	test("appendTurnTerminal resolves a pending waitForTurnEnd without a separate settle call", async () => {
+		const queryClient = new QueryClient();
+		const threadId = "thread-2";
+		const turnId = "turn-2";
+
+		const waiting = waitForTurnEnd(threadId, turnId);
+		appendTurnTerminal(queryClient, threadId, turnId, "turn_interrupted");
+
+		const result = await waiting;
+		expect(result.isErr()).toBe(true);
+	});
 });

shared/hooks/src/conversation/conversation-cache.ts

@@ -8,6 +8,10 @@ import {
 	currentMaxPersistedSeq,
 	sortConversationEntries,
 } from "@cyrus/utils/conversations/entries";
+import {
+	isTerminalEvent,
+	settleTurnWaiter,
+} from "@cyrus/utils/conversations/turn-waiters";
 import { Throttler } from "@tanstack/pacer";
 import type { QueryClient } from "@tanstack/react-query";
 
@@ -26,10 +30,6 @@ function turnKey(threadId: string, turnId: string): string {
 	return `${threadId}:${turnId}`;
 }
 
-function isTerminalEvent(event: ChatChunk["event"]): boolean {
-	return event.type === "turn_completed" || event.type === "turn_interrupted";
-}
-
 function isStreamingDeltaChunk(chunk: ChatChunk): boolean {
 	return (
 		chunk.sub !== undefined &&
@@ -183,6 +183,7 @@ function commitChunk(queryClient: QueryClient, chunk: ChatChunk): void {
 	updateCache(queryClient, chunk.threadId, (entries) =>
 		applyChunkToEntries(entries, chunk)
 	);
+	settleTurnWaiter(chunk.threadId, chunk.turnId, chunk.event);
 }
 
 function flushPendingDeltas(queryClient: QueryClient): void {

shared/hooks/src/conversation/use-thread-conversation.ts

@@ -3,6 +3,7 @@ import type { GetConversationsOutput } from "@cyrus/schemas/rtc/threads";
 import type { ThreadConversation } from "@cyrus/schemas/view";
 import {
 	currentMaxPersistedSeq,
+	isSnapshotBehindWatermark,
 	mergeConversationEntries,
 } from "@cyrus/utils/conversations/entries";
 import { fold } from "@cyrus/utils/conversations/fold";
@@ -69,6 +70,24 @@ export function useThreadConversation(
 		log.error({ kind: "unwatch_thread", error, threadId: tid });
 	});
 
+	const onWatched = useEffectEvent(
+		(snapshotHighWaterMark: number, tid: string) => {
+			const cached = queryClient.getQueryData<GetConversationsOutput>(
+				RTC_OPERATION_KEYS.getConversations(tid)
+			);
+			if (
+				isSnapshotBehindWatermark(
+					cached?.conversations ?? [],
+					snapshotHighWaterMark
+				)
+			) {
+				queryClient.invalidateQueries({
+					queryKey: RTC_OPERATION_KEYS.getConversations(tid),
+				});
+			}
+		}
+	);
+
 	useEffect(() => {
 		if (!threadId) return;
 
@@ -79,7 +98,8 @@ export function useThreadConversation(
 		).then((result) => {
 			if (abort.signal.aborted) return;
 			result.match({
-				ok: () => undefined,
+				ok: ({ snapshotHighWaterMark }) =>
+					onWatched(snapshotHighWaterMark, threadId),
 				err: (error) => onWatchError(error, threadId),
 			});
 		});

shared/hooks/src/conversation/use-thread-turns.ts

@@ -3,10 +3,7 @@ import { isTurnInterruptedError } from "@cyrus/errors/turn";
 import type { ChatMessage } from "@cyrus/schemas/rtc/chat";
 import { formatPromptBlocks } from "@cyrus/schemas/rtc/chat";
 import type { GetConversationsOutput } from "@cyrus/schemas/rtc/threads";
-import {
-	settleTurnWaiter,
-	waitForTurnEnd,
-} from "@cyrus/utils/conversations/turn-waiters";
+import { waitForTurnEnd } from "@cyrus/utils/conversations/turn-waiters";
 import { randomId } from "@cyrus/utils/identity";
 import { useQueryClient } from "@tanstack/react-query";
 import { Result } from "better-result";
@@ -246,7 +243,6 @@ export function useThreadTurns() {
 
 			for (const turnId of activeTurnIds) {
 				appendTurnTerminal(queryClient, threadId, turnId, "turn_interrupted");
-				settleTurnWaiter(threadId, turnId, { type: "turn_interrupted" });
 			}
 
 			const result = await Result.tryPromise({

shared/hooks/src/conversation/use-worker-conversation-sync.ts

@@ -1,6 +1,6 @@
 import { RTC_OPERATION_KEYS } from "@cyrus/constants/operation-keys";
 import type { ChatChunk } from "@cyrus/schemas/rtc/chat";
-import { settleTurnWaiter } from "@cyrus/utils/conversations/turn-waiters";
+import { isTerminalEvent } from "@cyrus/utils/conversations/turn-waiters";
 import { useQueryClient } from "@tanstack/react-query";
 import { Result } from "better-result";
 import { log } from "evlog";
@@ -12,13 +12,6 @@ import { applyChunkToCache } from "./conversation-cache";
 const SUBSCRIBE_RETRY_MS = 1000;
 const SUBSCRIBE_RETRY_MAX_MS = 8000;
 
-function isTerminalChunk(chunk: ChatChunk): boolean {
-	return (
-		chunk.event.type === "turn_completed" ||
-		chunk.event.type === "turn_interrupted"
-	);
-}
-
 function syncCatalogFromChunk(chunk: ChatChunk): void {
 	if (chunk.event.type !== "session_update") return;
 
@@ -60,9 +53,8 @@ export function useWorkerConversationSync(): void {
 		syncCatalogFromChunk(chunk);
 		applyChunkToCache(queryClient, chunk);
 
-		if (!isTerminalChunk(chunk)) return;
+		if (!isTerminalEvent(chunk.event)) return;
 
-		settleTurnWaiter(chunk.threadId, chunk.turnId, chunk.event);
 		if (chunk.event.type === "turn_completed") {
 			queryClient.invalidateQueries({
 				predicate: (query) =>

shared/utils/src/conversations/entries.test.ts

@@ -2,6 +2,7 @@ import type { ConversationEntry } from "@cyrus/schemas/rtc/threads";
 import { describe, expect, test } from "vitest";
 import {
 	currentMaxPersistedSeq,
+	isSnapshotBehindWatermark,
 	mergeConversationEntries,
 	sortConversationEntries,
 } from "./entries";
@@ -126,3 +127,25 @@ describe("mergeConversationEntries", () => {
 		expect(merged.map((e) => e.id).sort()).toEqual(["delta-1", "persisted"]);
 	});
 });
+
+describe("isSnapshotBehindWatermark", () => {
+	test("is true when the watermark is ahead of the cached snapshot's max persisted seq", () => {
+		const cached: ConversationEntry[] = [
+			entry("persisted", 2, "turn-1", { type: "turn_completed" }),
+		];
+
+		expect(isSnapshotBehindWatermark(cached, 5)).toBe(true);
+	});
+
+	test("is false once the cached snapshot already covers the watermark", () => {
+		const cached: ConversationEntry[] = [
+			entry("persisted", 5, "turn-1", { type: "turn_completed" }),
+		];
+
+		expect(isSnapshotBehindWatermark(cached, 5)).toBe(false);
+	});
+
+	test("is false for an empty snapshot with no durable watermark yet", () => {
+		expect(isSnapshotBehindWatermark([], 0)).toBe(false);
+	});
+});

shared/utils/src/conversations/entries.ts

@@ -1,4 +1,5 @@
 import type { ConversationEntry } from "@cyrus/schemas/rtc/threads";
+import { compareBySeqSub } from "./order-key";
 
 function isPersisted(entry: ConversationEntry): boolean {
 	return entry.sub === undefined;
@@ -32,10 +33,8 @@ export function sortConversationEntries(
 	entries: ConversationEntry[]
 ): ConversationEntry[] {
 	return [...entries].sort((left, right) => {
-		if (left.seq !== right.seq) return left.seq - right.seq;
-		const leftSub = left.sub ?? 0;
-		const rightSub = right.sub ?? 0;
-		if (leftSub !== rightSub) return leftSub - rightSub;
+		const bySeqSub = compareBySeqSub(left, right);
+		if (bySeqSub !== 0) return bySeqSub;
 		return left.createdAt.localeCompare(right.createdAt);
 	});
 }
@@ -47,6 +46,13 @@ export function currentMaxPersistedSeq(entries: ConversationEntry[]): number {
 	);
 }
 
+export function isSnapshotBehindWatermark(
+	entries: ConversationEntry[],
+	watermark: number
+): boolean {
+	return watermark > currentMaxPersistedSeq(entries);
+}
+
 export function mergeConversationEntries(
 	cached: ConversationEntry[],
 	fetched: ConversationEntry[]

shared/utils/src/conversations/order-key.test.ts

@@ -0,0 +1,37 @@
+import { describe, expect, test } from "vitest";
+import { compareBySeqSub, compareOrderKey, orderKey } from "./order-key";
+
+describe("orderKey", () => {
+	test("defaults a missing sub to 0", () => {
+		expect(orderKey({ seq: 3 })).toEqual({ seq: 3, sub: 0 });
+	});
+});
+
+describe("compareOrderKey", () => {
+	test("orders by seq first, then sub", () => {
+		expect(
+			compareOrderKey({ seq: 1, sub: 5 }, { seq: 2, sub: 0 })
+		).toBeLessThan(0);
+		expect(
+			compareOrderKey({ seq: 2, sub: 0 }, { seq: 2, sub: 1 })
+		).toBeLessThan(0);
+		expect(compareOrderKey({ seq: 2, sub: 1 }, { seq: 2, sub: 1 })).toBe(0);
+	});
+});
+
+describe("compareBySeqSub", () => {
+	test("ignores createdAt — a reversed createdAt does not flip seq/sub order", () => {
+		const earlierBySeq = {
+			seq: 1,
+			sub: 0,
+			createdAt: "2026-01-01T00:00:05.000Z",
+		};
+		const laterBySeq = {
+			seq: 2,
+			sub: 0,
+			createdAt: "2026-01-01T00:00:00.000Z",
+		};
+
+		expect(compareBySeqSub(earlierBySeq, laterBySeq)).toBeLessThan(0);
+	});
+});

shared/utils/src/conversations/order-key.ts

@@ -0,0 +1,16 @@
+export type OrderKey = { seq: number; sub: number };
+
+export function orderKey(entity: { seq: number; sub?: number }): OrderKey {
+	return { seq: entity.seq, sub: entity.sub ?? 0 };
+}
+
+export function compareOrderKey(left: OrderKey, right: OrderKey): number {
+	return left.seq - right.seq || left.sub - right.sub;
+}
+
+export function compareBySeqSub(
+	left: { seq: number; sub?: number },
+	right: { seq: number; sub?: number }
+): number {
+	return compareOrderKey(orderKey(left), orderKey(right));
+}