Add shell input to the composer

CONTEXT.md

@@ -83,6 +83,9 @@ The counter distinguishing multiple ephemeral deltas that share the same anchor
 **Chat chunk**:
 A live conversation event delivered to watching peers as a turn runs.
 
+**Shell execution**:
+The conversation entry persisted for a **Shell input** run: command, stream-tagged lines (stdout/stderr), exit code, and start/complete times. A dedicated entry kind alongside message/thought/tool/error, not a synthetic tool call. Identified by its own execution id — distinct from a **Turn** id — since a shell execution runs independently of any turn and is cancelled independently (its own Stop control, not the composer's turn-scoped Stop button).
+
 **Ephemeral delta**:
 A streamed fragment (token or thought), ordered by its anchor seq plus sub — broadcast live but never persisted, and never held in the replay buffer; the completed message is persisted once, in full.
 _Avoid_: partial message
@@ -151,6 +154,10 @@ The health check that spawns an enabled agent and verifies the ACP handshake com
 **Composer**:
 The chat input area: prompt box, agent/model picker, mode/effort/persona controls, prompt queue, and the panel where approvals and elicitations are answered.
 
+**Shell input**:
+Composer state entered when `!` is the first character typed in a thread's composer; the text is sent as a shell command instead of a prompt. Thread-scoped only — requires an existing thread. Distinct from an agent's **mode** (`modeId`).
+_Avoid_: shell mode (collides with modeId)
+
 **Prompt queue**:
 Messages queued per thread while a turn is active, sent sequentially once the turn completes.
 

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

@@ -11,6 +11,7 @@ import { fsHandlers } from "./fs";
 import { gitHandlers } from "./git";
 import { interactiveHandlers } from "./interactive";
 import { projectsHandlers } from "./projects";
+import { shellHandlers } from "./shell";
 import { threadsHandlers } from "./threads";
 
 export function createControllerRouter(runtime: WorkerRuntime) {
@@ -29,6 +30,7 @@ export function createControllerRouter(runtime: WorkerRuntime) {
 		...personaHandlers(deps),
 		...contextHandlers(deps),
 		...chatHandlers(deps),
+		...shellHandlers(deps),
 		...interactiveHandlers(deps),
 	};
 }

apps/cli/src/handlers/controller/shell.test.ts

@@ -0,0 +1,66 @@
+import { describe, expect, test } from "bun:test";
+import type { ShellExecutionLine } from "@cyrus/schemas/rtc/chat";
+import { pumpLines } from "./shell";
+
+function streamFromChunks(chunks: Uint8Array[]): ReadableStream<Uint8Array> {
+	return new ReadableStream({
+		start(controller) {
+			for (const chunk of chunks) controller.enqueue(chunk);
+			controller.close();
+		},
+	});
+}
+
+function encode(text: string): Uint8Array {
+	return new TextEncoder().encode(text);
+}
+
+async function collectLines(
+	stream: ReadableStream<Uint8Array> | null
+): Promise<ShellExecutionLine[]> {
+	const lines: ShellExecutionLine[] = [];
+	await pumpLines(stream, "stdout", (line) => lines.push(line));
+	return lines;
+}
+
+describe("pumpLines", () => {
+	test("splits a single chunk into its newline-terminated lines", async () => {
+		const lines = await collectLines(streamFromChunks([encode("a\nb\nc\n")]));
+		expect(lines).toEqual([
+			{ stream: "stdout", text: "a" },
+			{ stream: "stdout", text: "b" },
+			{ stream: "stdout", text: "c" },
+		]);
+	});
+
+	test("joins a line split across two read chunks", async () => {
+		const lines = await collectLines(
+			streamFromChunks([encode("hello wor"), encode("ld\n")])
+		);
+		expect(lines).toEqual([{ stream: "stdout", text: "hello world" }]);
+	});
+
+	test("emits trailing text with no final newline as its own line", async () => {
+		const lines = await collectLines(streamFromChunks([encode("no newline")]));
+		expect(lines).toEqual([{ stream: "stdout", text: "no newline" }]);
+	});
+
+	test("decodes a multi-byte UTF-8 character split across chunk boundaries", async () => {
+		// "🎉" is 4 bytes in UTF-8 — split it in the middle of the codepoint.
+		const bytes = encode("party 🎉\n");
+		const lines = await collectLines(
+			streamFromChunks([bytes.slice(0, 7), bytes.slice(7)])
+		);
+		expect(lines).toEqual([{ stream: "stdout", text: "party 🎉" }]);
+	});
+
+	test("emits nothing for an empty stream", async () => {
+		const lines = await collectLines(streamFromChunks([]));
+		expect(lines).toEqual([]);
+	});
+
+	test("resolves immediately for a null stream (stdout/stderr absent)", async () => {
+		const lines = await collectLines(null);
+		expect(lines).toEqual([]);
+	});
+});

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

@@ -0,0 +1,199 @@
+import { appendConversation } from "@cyrus/database/repositories/conversations";
+import { resolveThreadGitCwd } from "@cyrus/database/repositories/git";
+import { orpcOk, throwOrpc } from "@cyrus/errors/orpc";
+import type {
+	ChatChunk,
+	ShellExecutionEndStatus,
+	ShellExecutionLine,
+} from "@cyrus/schemas/rtc/chat";
+import { randomId } from "@cyrus/utils/identity";
+import { Result } from "better-result";
+import { log } from "evlog";
+import { env } from "@/lib/env";
+import type { ControllerDeps } from "./deps";
+
+function toError(cause: unknown): Error {
+	return cause instanceof Error ? cause : new Error(String(cause));
+}
+
+type EventBusPublisher = { eventBus: { publish: (chunk: ChatChunk) => void } };
+
+type ActiveShellExecution = {
+	threadId: string;
+	kill: (status: ShellExecutionEndStatus) => void;
+};
+
+const activeShellExecutions = new Map<string, ActiveShellExecution>();
+
+export async function pumpLines(
+	stream: ReadableStream<Uint8Array> | null,
+	streamName: ShellExecutionLine["stream"],
+	onLine: (line: ShellExecutionLine) => void
+): Promise<void> {
+	if (!stream) return;
+	const decoder = new TextDecoder();
+	const reader = stream.getReader();
+	let buffer = "";
+	for (;;) {
+		const { done, value } = await reader.read();
+		if (done) break;
+		buffer += decoder.decode(value, { stream: true });
+		const parts = buffer.split("\n");
+		buffer = parts.pop() ?? "";
+		for (const text of parts) onLine({ stream: streamName, text });
+	}
+	buffer += decoder.decode();
+	if (buffer.length > 0) onLine({ stream: streamName, text: buffer });
+}
+
+async function runShellExecution(options: {
+	command: string;
+	cwd: string;
+	context: EventBusPublisher;
+	shellExecutionId: string;
+	threadId: string;
+}): Promise<void> {
+	const { command, cwd, context, shellExecutionId, threadId } = options;
+	const lines: ShellExecutionLine[] = [];
+
+	function emitLine(line: ShellExecutionLine): void {
+		lines.push(line);
+		context.eventBus.publish({
+			threadId,
+			shellExecutionId,
+			seq: 0,
+			event: { type: "shell_execution_line", lines: [line] },
+		});
+	}
+
+	async function finish(
+		status: ShellExecutionEndStatus,
+		exitCode: number | null
+	): Promise<void> {
+		const event = {
+			type: "shell_execution_end" as const,
+			status,
+			exitCode,
+			lines,
+		};
+		const entry = await appendConversation(threadId, {
+			threadId,
+			shellExecutionId,
+			event,
+		});
+		if (entry.isErr()) {
+			log.error({
+				kind: "shell_execution_persist_failed",
+				error: entry.error,
+				threadId,
+				shellExecutionId,
+			});
+			context.eventBus.publish({ threadId, shellExecutionId, seq: 0, event });
+			return;
+		}
+		context.eventBus.publish(entry.value.chunk);
+	}
+
+	const spawned = Result.try({
+		try: () =>
+			Bun.spawn(["sh", "-c", command], {
+				cwd,
+				stderr: "pipe",
+				stdout: "pipe",
+			}),
+		catch: toError,
+	});
+	if (spawned.isErr()) {
+		log.error({
+			kind: "shell_execution_spawn_failed",
+			error: spawned.error,
+			threadId,
+		});
+		await finish("spawn_error", null);
+		return;
+	}
+	const subprocess = spawned.value;
+
+	let killedFor: ShellExecutionEndStatus | null = null;
+	function kill(status: ShellExecutionEndStatus): void {
+		if (killedFor) return;
+		killedFor = status;
+		subprocess.kill();
+	}
+
+	activeShellExecutions.set(shellExecutionId, { threadId, kill });
+	const timeoutHandle = setTimeout(() => {
+		kill("timeout");
+	}, env.CYRUS_SHELL_INPUT_TIMEOUT_MS);
+
+	const streamed = await Result.tryPromise({
+		try: async () => {
+			await Promise.all([
+				pumpLines(subprocess.stdout, "stdout", emitLine),
+				pumpLines(subprocess.stderr, "stderr", emitLine),
+			]);
+			return await subprocess.exited;
+		},
+		catch: toError,
+	});
+
+	clearTimeout(timeoutHandle);
+	activeShellExecutions.delete(shellExecutionId);
+
+	if (streamed.isErr()) {
+		log.error({
+			kind: "shell_execution_stream_failed",
+			error: streamed.error,
+			shellExecutionId,
+			threadId,
+		});
+		await finish(killedFor ?? "exited", null);
+		return;
+	}
+	await finish(killedFor ?? "exited", killedFor ? null : streamed.value);
+}
+
+export function shellHandlers({ os }: ControllerDeps) {
+	return {
+		executeShellInput: os.executeShellInput.handler(
+			async ({ input, context }) => {
+				const { threadId, command } = input;
+				const cwd = orpcOk(await resolveThreadGitCwd(threadId));
+				const shellExecutionId = randomId();
+
+				const started = await appendConversation(threadId, {
+					threadId,
+					shellExecutionId,
+					event: { type: "shell_execution_start", command },
+				});
+				if (started.isErr()) throwOrpc(started.error);
+				context.eventBus.publish(started.value.chunk);
+
+				runShellExecution({
+					command,
+					context,
+					cwd,
+					shellExecutionId,
+					threadId,
+				}).catch((error) => {
+					log.error({
+						kind: "shell_execution_failed",
+						error,
+						shellExecutionId,
+						threadId,
+					});
+				});
+
+				return { shellExecutionId };
+			}
+		),
+
+		cancelShellExecution: os.cancelShellExecution.handler(({ input }) => {
+			const active = activeShellExecutions.get(input.shellExecutionId);
+			if (active && active.threadId === input.threadId) {
+				active.kill("cancelled");
+			}
+			return {};
+		}),
+	};
+}

apps/cli/src/lib/env.ts

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

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

@@ -20,6 +20,18 @@ function tokenChunk(turnId = "turn-1"): ChatChunk {
 	};
 }
 
+function shellPersistedChunk(
+	seq: number,
+	shellExecutionId = "shell-1"
+): ChatChunk {
+	return {
+		threadId: "thread-1",
+		shellExecutionId,
+		seq,
+		event: { type: "shell_execution_start", command: "ls" },
+	};
+}
+
 function createReader(iterator: AsyncGenerator<ChatChunk>) {
 	let pending = iterator.next();
 	return {
@@ -116,4 +128,37 @@ describe("thread event bus", () => {
 		expect(await second.next()).toMatchObject({ seq: 3 });
 		expect(await second.next()).toMatchObject({ seq: 4 });
 	});
+
+	test("fans out shell execution chunks live, same as turn chunks", async () => {
+		const bus = createThreadEventBus();
+		bus.watch("peer-1", "thread-1");
+		const reader = createReader(bus.subscribe("peer-1"));
+
+		bus.publish(persistedChunk(1));
+		bus.publish(shellPersistedChunk(2));
+
+		expect(await reader.next()).toMatchObject({ seq: 1, turnId: "turn-1" });
+		expect(await reader.next()).toMatchObject({
+			seq: 2,
+			shellExecutionId: "shell-1",
+		});
+	});
+
+	test("does not replay shell execution chunks to a late-joining watcher", async () => {
+		const bus = createThreadEventBus();
+		bus.watch("peer-1", "thread-1");
+		const first = createReader(bus.subscribe("peer-1"));
+
+		bus.publish(persistedChunk(1));
+		bus.publish(shellPersistedChunk(2));
+		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: 1 });
+
+		bus.publish(tokenChunk());
+		expect(await second.next()).toMatchObject({ event: { type: "token" } });
+	});
 });

apps/cli/src/queue/bus.ts

@@ -99,7 +99,7 @@ export function createThreadEventBus(
 		while (log.length > maxChunksPerTurn) log.shift();
 	}
 
-	function appendToTurnLog(chunk: ChatChunk): void {
+	function appendToTurnLog(chunk: ChatChunk & { turnId: string }): void {
 		const { turnId } = chunk;
 		let log = activeTurnLogs.get(turnId);
 		if (!log) {
@@ -137,11 +137,11 @@ export function createThreadEventBus(
 			const stamped = stampChunk(chunk);
 			const terminal = isTerminalEvent(stamped.event);
 
-			if (!terminal && stamped.sub === undefined) {
-				appendToTurnLog(stamped);
+			if (!terminal && stamped.sub === undefined && stamped.turnId) {
+				appendToTurnLog(stamped as ChatChunk & { turnId: string });
 			}
 			fanOut(stamped);
-			if (terminal) {
+			if (terminal && stamped.turnId) {
 				evictTurnLog(stamped.turnId);
 			}
 		},