Scope live chat to watched threads with overlay-based streaming

apps/cli/package.json

@@ -26,6 +26,7 @@
     "better-result": "catalog:",
     "commander": "^15.0.0",
     "diff": "^7.0.0",
+    "evlog": "catalog:",
     "zod": "catalog:"
   },
   "devDependencies": {

apps/cli/src/commands/agents/doctor.ts

@@ -18,9 +18,7 @@ async function checkAgent(entry: AgentEntry): Promise<HealthResult> {
 function printHealth(name: string, result: HealthResult): void {
 	const status = result.healthy ? green("healthy") : red("unhealthy");
 	print.line`${name}: ${status}`;
-	if (result.error) {
-		print.line`  ${result.error}`;
-	}
+	if (result.error) print.line`  ${result.error}`;
 }
 
 async function checkSingleAgentHealth(name: string) {

apps/cli/src/commands/agents/rm.ts

@@ -4,9 +4,7 @@ import { print } from "@/utils/style";
 export async function rm(name: string): Promise<void> {
 	const removed = await removeAgent(name);
 	removed.match({
-		ok: () => {
-			print.success`✓ removed agent "${name}"`;
-		},
+		ok: () => print.success`✓ removed agent "${name}"`,
 		err: (message) => {
 			print.error`${message}`;
 			process.exit(1);

apps/cli/src/commands/auth/login.ts

@@ -4,7 +4,7 @@ import { authClient } from "@/lib/auth";
 import { getOrCreate, set } from "@/store/config";
 import { blue, bold, cyan, print, underline } from "@/utils/style";
 
-export const CLIENT_ID = "cyrus-cli";
+export const CLIENT_ID = "cyrusd";
 export const GRANT_TYPE = "urn:ietf:params:oauth:grant-type:device_code";
 
 export async function login(): Promise<void> {

apps/cli/src/commands/auth/whoami.ts

@@ -24,11 +24,8 @@ export async function whoami(opts: WhoamiOptions): Promise<void> {
 	}
 
 	const name = await get("name");
+
 	print.line`${field("User")}${data.user.name}`;
-	if (name) {
-		print.line`${field("Device")}${name}`;
-	}
-	if (opts.email) {
-		print.line`${field("Email")}${data.user.email}`;
-	}
+	if (name) print.line`${field("Device")}${name}`;
+	if (opts.email) print.line`${field("Email")}${data.user.email}`;
 }

apps/cli/src/commands/service/start.ts

@@ -19,20 +19,15 @@ async function runWorker(): Promise<void> {
 	process.on("exit", () => {
 		Result.try(() => {
 			const stored = Number.parseInt(readFileSync(PID_PATH, "utf8").trim(), 10);
-			if (stored === ownPid) {
-				unlinkSync(PID_PATH);
-			}
+			if (stored === ownPid) unlinkSync(PID_PATH);
 		});
 	});
 	const { worker } = await import("./worker");
 	await worker();
 }
 
 export async function start(opts: StartOptions): Promise<void> {
-	if (env.CYRUS_DAEMON) {
-		await runWorker();
-		return;
-	}
+	if (env.CYRUS_DAEMON) return await runWorker();
 
 	const running = await runningPid();
 	if (running !== null) {
@@ -50,9 +45,7 @@ export async function start(opts: StartOptions): Promise<void> {
 		});
 		closeSync(log);
 		child.unref();
-		if (child.pid) {
-			await writePid(child.pid);
-		}
+		if (child.pid) await writePid(child.pid);
 
 		// brief wait to catch immediate startup failures (e.g. not logged in)
 		await Bun.sleep(300);

apps/cli/src/commands/service/stop.ts

@@ -11,9 +11,8 @@ export async function stop(): Promise<void> {
 
 	Result.try(() => process.kill(pid, "SIGTERM"));
 
-	for (let i = 0; i < 20 && isAlive(pid); i++) {
-		await Bun.sleep(50);
-	}
+	for (let i = 0; i < 20 && isAlive(pid); i++) await Bun.sleep(50);
+
 	if (isAlive(pid)) {
 		print.error`Worker (pid ${pid}) did not stop in time.`;
 		process.exit(1);

apps/cli/src/commands/service/worker.ts

@@ -8,6 +8,7 @@ import { createControllerRouter } from "@/handlers/controller";
 import { workerRouter } from "@/handlers/worker";
 import { authClient } from "@/lib/auth";
 import { env } from "@/lib/env";
+import { createThreadEventBus } from "@/queue/bus";
 import { get, getOrCreate } from "@/store/config";
 import { initDatabase } from "@/store/database";
 import { print } from "@/utils/style";
@@ -48,9 +49,12 @@ export async function worker(): Promise<void> {
 	});
 	print.success`✓ connected — waiting for message…`;
 
+	const eventBus = createThreadEventBus();
+
 	const device = serveWorker({
 		signaling: signalingSession.signaling,
 		events: signalingSession.events,
+		eventBus,
 		routers: {
 			controller: createControllerRouter(runtime),
 			worker: workerRouter,
@@ -64,6 +68,7 @@ export async function worker(): Promise<void> {
 		await connection.close();
 		process.exit(0);
 	};
+
 	process.on("SIGINT", shutdown);
 	process.on("SIGTERM", shutdown);
 	process.on("SIGBREAK", shutdown);

apps/cli/src/handlers/controller/chat.ts

@@ -2,7 +2,8 @@ import { appendConversation } from "@cyrus/database/repositories/conversations";
 import { ensureThread } from "@cyrus/database/repositories/threads";
 import type { ChatChunk } from "@cyrus/schemas/rtc/chat";
 import { randomId } from "@cyrus/utils/identity";
-import { env } from "@/lib/env";
+import { Result } from "better-result";
+import { log } from "evlog";
 import { throwOrpcFromRepositoryError } from "@/utils/error";
 import {
 	isStreamingDelta,
@@ -11,29 +12,88 @@ import {
 } from "@/utils/streams";
 import type { ControllerDeps } from "./deps";
 
+type RunTurnOptions = {
+	agentName: string;
+	threadId: string;
+	projectId: string;
+	message: string;
+	turnId: string;
+	emit: (event: ChatChunk["event"]) => Promise<void>;
+	runtime: ControllerDeps["runtime"];
+};
+
+async function runTurn({
+	agentName,
+	threadId,
+	projectId,
+	message,
+	emit,
+	emitTerminal,
+	runtime,
+}: Omit<RunTurnOptions, "turnId" | "emitTerminal"> & {
+	emitTerminal: (
+		event: Extract<
+			ChatChunk["event"],
+			{ type: "turn_completed" | "turn_interrupted" }
+		>
+	) => Promise<void>;
+}): Promise<Result<void, unknown>> {
+	const started = await Result.tryPromise(async () => {
+		await emit({ type: "user_message", content: message });
+		await emit({ type: "thread_started", threadId });
+	});
+	if (started.isErr()) return started;
+
+	const streamed = await Result.tryPromise(async () => {
+		const gen = runtime.threadCoordinator.prompt(
+			agentName,
+			threadId,
+			projectId,
+			message
+		);
+		for await (const event of gen) await emit(event);
+	});
+
+	if (streamed.isErr()) {
+		await emitTerminal({ type: "turn_interrupted" });
+		return streamed;
+	}
+
+	await emitTerminal({ type: "turn_completed" });
+	return Result.ok(undefined);
+}
+
 export function chatHandlers({ os, runtime }: ControllerDeps) {
 	return {
-		chat: os.chat.handler(async function* ({ input, context }) {
-			const { agentName, threadId = randomId(), message, projectId } = input;
+		chat: os.chat.handler(async ({ input, context }) => {
+			const {
+				agentName,
+				threadId = randomId(),
+				turnId = randomId(),
+				message,
+				projectId,
+			} = input;
 
 			const thread = await ensureThread(threadId, projectId, {
 				agentName,
 				firstMessage: message,
 			});
 			if (thread.isErr()) throwOrpcFromRepositoryError(thread.error);
 
-			const turnId = randomId();
 			const messageBuffers = new Map<string, string>();
 			const thoughtBuffers = new Map<string, string>();
 
-			async function emit(event: ChatChunk["event"]): Promise<ChatChunk> {
+			context.eventBus.ensureWatch(context.peerId, threadId);
+
+			function publishChunk(chunk: ChatChunk): void {
+				context.eventBus.publish(chunk);
+			}
+
+			async function emit(event: ChatChunk["event"]): Promise<void> {
 				trackDelta(event, messageBuffers, thoughtBuffers);
 
-				if (isStreamingDelta(event)) {
-					const chunk: ChatChunk = { threadId, turnId, seq: 0, event };
-					context.broadcaster.broadcast(chunk, context.peerId);
-					return chunk;
-				}
+				if (isStreamingDelta(event))
+					return publishChunk({ threadId, turnId, seq: 0, event });
 
 				const persistEvent = resolvePersistEvent(
 					event,
@@ -46,39 +106,92 @@ export function chatHandlers({ os, runtime }: ControllerDeps) {
 					event: persistEvent,
 				});
 				if (entry.isErr()) throwOrpcFromRepositoryError(entry.error);
-				const chunk = entry.value.chunk;
-				context.broadcaster.broadcast(chunk, context.peerId);
-				return chunk;
+				publishChunk(entry.value.chunk);
 			}
 
-			yield await emit({ type: "user_message", content: message });
-			yield await emit({ type: "thread_started", threadId });
+			async function emitTerminal(
+				event: Extract<
+					ChatChunk["event"],
+					{ type: "turn_completed" | "turn_interrupted" }
+				>
+			): Promise<void> {
+				trackDelta(event, messageBuffers, thoughtBuffers);
 
-			const gen = runtime.threadCoordinator.prompt(
+				const persistEvent = resolvePersistEvent(
+					event,
+					messageBuffers,
+					thoughtBuffers
+				);
+				const entry = await appendConversation(threadId, {
+					threadId,
+					turnId,
+					event: persistEvent,
+				});
+
+				if (entry.isOk()) {
+					publishChunk(entry.value.chunk);
+					return;
+				}
+
+				log.error({
+					kind: "terminal_event_persist",
+					error: entry.error,
+					threadId,
+					turnId,
+					event: event.type,
+				});
+				publishChunk({ threadId, turnId, seq: 0, event });
+			}
+
+			runTurn({
 				agentName,
 				threadId,
 				projectId,
-				message
-			);
-			try {
-				for await (const event of gen) {
-					yield await emit(event);
-					await Bun.sleep(env.CYRUS_STREAM_THROTTLING_MS);
-				}
-				yield await emit({ type: "turn_completed" });
-			} catch (error) {
-				yield await emit({ type: "turn_interrupted" });
-				throw error;
-			}
+				message,
+				emit,
+				emitTerminal,
+				runtime,
+			})
+				.then((result) => {
+					result.tapError((error) => {
+						log.error({ kind: "chat_turn_failed", error, threadId, turnId });
+					});
+				})
+				.catch((error) => {
+					log.error({ kind: "chat_turn_failed", error, threadId, turnId });
+				});
+
+			return { threadId, turnId };
 		}),
 
 		subscribe: os.subscribe.handler(async function* ({ context }) {
-			for await (const chunk of context.broadcaster.subscribe(context.peerId))
+			for await (const chunk of context.eventBus.subscribe(context.peerId))
 				yield chunk;
 		}),
 
-		cancel: os.cancel.handler(async ({ input }) => {
+		cancel: os.cancel.handler(async ({ input, context }) => {
+			// Snapshot before awaiting — a follow-up chat() can start while cancel
+			// is in flight and must not be interrupted by this request.
+			const snapshottedTurnIds = context.eventBus.getActiveTurnIdsForThread(
+				input.threadId
+			);
+
 			await runtime.threadCoordinator.cancel(input.agentName, input.threadId);
+
+			const stillActiveTurnIds = new Set(
+				context.eventBus.getActiveTurnIdsForThread(input.threadId)
+			);
+
+			for (const turnId of snapshottedTurnIds) {
+				if (!stillActiveTurnIds.has(turnId)) continue;
+				context.eventBus.publish({
+					threadId: input.threadId,
+					turnId,
+					seq: 0,
+					event: { type: "turn_interrupted" },
+				});
+			}
+
 			return {};
 		}),
 	};

apps/cli/src/handlers/controller/threads.ts

@@ -1,4 +1,7 @@
-import { getConversations } from "@cyrus/database/repositories/conversations";
+import {
+	getConversations,
+	getSnapshotHighWaterMark,
+} from "@cyrus/database/repositories/conversations";
 import {
 	createThread as createStoredThread,
 	deleteThread,
@@ -62,5 +65,29 @@ export function threadsHandlers(os: ControllerOs) {
 
 			return {};
 		}),
+
+		watchThread: os.watchThread.handler(async ({ input, context }) => {
+			const thread = await getThread(input.threadId);
+			if (thread.isErr()) throwOrpcFromRepositoryError(thread.error);
+			if (!thread.value) {
+				throwOrpcFromRepositoryError({
+					type: "not_found",
+					entity: "thread",
+					id: input.threadId,
+				});
+			}
+
+			context.eventBus.watch(context.peerId, input.threadId);
+
+			return (await getSnapshotHighWaterMark(input.threadId)).match({
+				ok: (snapshotHighWaterMark) => ({ snapshotHighWaterMark }),
+				err: throwOrpcFromRepositoryError,
+			});
+		}),
+
+		unwatchThread: os.unwatchThread.handler(({ input, context }) => {
+			context.eventBus.unwatch(context.peerId, input.threadId);
+			return {};
+		}),
 	};
 }

apps/cli/src/lib/env.ts

@@ -25,7 +25,6 @@ export const env = createEnv({
 			.int()
 			.positive()
 			.default(30 * 60 * 1000),
-		CYRUS_STREAM_THROTTLING_MS: z.coerce.number().positive().default(25),
 	},
 	runtimeEnv: process.env,
 	emptyStringAsUndefined: true,