feat(p2): terminal tool for the agent, tmux-style argv, code mode only

The agent drives the session's terminals (shared with the owner's panel) with
tools.terminal({ command, args }): new-window, list-windows, send-keys, read,
capture-pane, kill-window, with waits (quiet, exit, /regex/) and plain-text
output. The terminal host keeps each terminal's last MiB of output and the
agent's read cursor, answers calls with ids, and does keyed calls once, so the
tool is replay-safe. AgentTerminals serves the session's current worker on its
environments; archiving closes a session's terminals. Without code mode the
worker registry installs tool_search once for all deferred tools.

apps/p2/plugins/chat/web.tsx

@@ -42,6 +42,7 @@ import {
 	navigate,
 	OneShotAgents,
 	OpenConversation,
+	OpenTerminal,
 	OpenWorkspaceFile,
 	openLayer,
 	type PickedFile,
@@ -483,6 +484,18 @@ function SessionChat({ sessionId, conversationId }: { sessionId: string; convers
 		},
 		[nav],
 	);
+	// The terminal panel where it is open (the workspace's tab), else below the chat.
+	const openTerminal = useCallback(
+		(terminalId: string) => {
+			const current = nav.route;
+			void nav.push(
+				current.workspace?.tab === "terminal"
+					? { ...current, workspace: { ...current.workspace, terminal: terminalId, startTerminal: undefined } }
+					: { ...current, terminal: terminalId, startTerminal: undefined },
+			);
+		},
+		[nav],
+	);
 	const connected = useConnected();
 	const host = useHost();
 	// The catalog as the owner or a guest sees it (`worker` is a session worker's).
@@ -998,33 +1011,35 @@ function SessionChat({ sessionId, conversationId }: { sessionId: string; convers
 									<OpenConversation.Provider value={openConversation}>
 										<OneShotAgents.Provider value={oneShots}>
 											<OpenWorkspaceFile.Provider value={canWrite ? openWorkspaceFile : undefined}>
-												<ReplyTo.Provider value={participant && canReply ? startReply : undefined}>
-													<BranchFrom.Provider value={canBranch ? branchFrom : undefined}>
-														<RevealEntry.Provider value={reveal}>
-															<FillGap.Provider value={fillGap}>
-																<Messages
-																	transcript={shown ?? transcript}
-																	rows={rows}
-																	list={follow.list}
-																	range={follow.range}
-																	stopped={
-																		stopped?.sessionId === sessionId &&
-																		stopped?.conversationId === conversationId &&
-																		!view?.entries.some((entry) =>
-																			entry.model?.some(
-																				(message) =>
-																					message.role === "assistant" &&
-																					message.stopReason === "aborted",
-																			),
-																		)
-																	}
-																	earlier={earlier}
-																	onEarlier={() => void loadEarlier()}
-																/>
-															</FillGap.Provider>
-														</RevealEntry.Provider>
-													</BranchFrom.Provider>
-												</ReplyTo.Provider>
+												<OpenTerminal.Provider value={owner ? openTerminal : undefined}>
+													<ReplyTo.Provider value={participant && canReply ? startReply : undefined}>
+														<BranchFrom.Provider value={canBranch ? branchFrom : undefined}>
+															<RevealEntry.Provider value={reveal}>
+																<FillGap.Provider value={fillGap}>
+																	<Messages
+																		transcript={shown ?? transcript}
+																		rows={rows}
+																		list={follow.list}
+																		range={follow.range}
+																		stopped={
+																			stopped?.sessionId === sessionId &&
+																			stopped?.conversationId === conversationId &&
+																			!view?.entries.some((entry) =>
+																				entry.model?.some(
+																					(message) =>
+																						message.role === "assistant" &&
+																						message.stopReason === "aborted",
+																				),
+																			)
+																		}
+																		earlier={earlier}
+																		onEarlier={() => void loadEarlier()}
+																	/>
+																</FillGap.Provider>
+															</RevealEntry.Provider>
+														</BranchFrom.Provider>
+													</ReplyTo.Provider>
+												</OpenTerminal.Provider>
 											</OpenWorkspaceFile.Provider>
 										</OneShotAgents.Provider>
 									</OpenConversation.Provider>

apps/p2/plugins/mcp/worker.ts

@@ -2,12 +2,11 @@
 // conversations work at (the session's, active subagents' own), with the project `.pi/mcp.json` read at each through
 // its environment, as its leases, again whenever they may have changed and whenever the view (re)activates, and offers
 // the servers' tools as one Durable extension that follows the view's `servers` and the code mode switch: as in pi,
-// tools are for code mode scripts, and only without code mode does `tool_search` load them for the model. No MCP
-// process or credential is ever in a worker.
+// tools are for code mode scripts, and only without code mode does `tool_search` (installed by the worker's registry)
+// load them for the model. No MCP process or credential is ever in a worker.
 
 import { setTimeout as sleep } from "node:timers/promises";
 import { type ConversationId, type Extension, type Harness, ROOT_CONVERSATION_ID } from "@earendil-works/pi-durable";
-import { createToolSearchExtension } from "@earendil-works/pi-durable/tool-search";
 import {
 	type Context,
 	type Ctx,
@@ -81,14 +80,12 @@ function mcpWorker(self: StoreSelf): void {
 	leaseLocations(self, view, harness, worker);
 }
 
-/** The extensions offering `offer`: the servers' tools and section, and `tool_search` while it reaches the tools. */
+/**
+ * The extension offering `offer`: the servers' tools and section. `tool_search`, which reaches them without code mode,
+ * is the worker's registry's (`durableRegistry`), shared with other plugins' deferred tools.
+ */
 function extensionsFor(offer: McpOffer, call: McpCall): Readonly<Record<string, Extension>> {
-	const extensions: Record<string, Extension> = { [MCP_EXTENSION]: createMcpExtension(offer, call) };
-	if (offer.reach === "tool_search" && offer.servers.length > 0) {
-		const search = createToolSearchExtension();
-		extensions[search.name] = search;
-	}
-	return extensions;
+	return { [MCP_EXTENSION]: createMcpExtension(offer, call) };
 }
 
 /**

apps/p2/plugins/models/fake.ts

@@ -187,7 +187,7 @@ export function createFakeModels(options: CreateModelsOptions = {}) {
 												? fauxToolCall("subagent", JSON.parse(text.slice("subagent ".length)))
 												: text.startsWith("team ")
 													? fauxToolCall("team", JSON.parse(text.slice("team ".length)))
-													: /^(artifact|history_(search|read|handoff)|location_(envs|dirs|move)|tool_search|mcp__\w+) /.test(
+													: /^(artifact|history_(search|read|handoff)|location_(envs|dirs|move)|tool_search|terminal|mcp__\w+) /.test(
 																text,
 															)
 														? fauxToolCall(

apps/p2/plugins/terminals/Terminal.tsx

@@ -905,6 +905,17 @@ class TerminalSession {
 				this.#autoOpen = false;
 				return;
 			}
+			case "terminals": {
+				// Another viewer or the agent opened, closed or renamed one. The shown terminal stays; only a panel showing
+				// none attaches the newest that runs (the agent just opened it).
+				const known = new Set(this.#state.terminals.map((terminal) => terminal.id));
+				this.#emit({ terminals: event.terminals });
+				const added = event.terminals.filter((terminal) => !known.has(terminal.id) && !terminal.exited);
+				if (!this.#state.current && added.length > 0) this.attach((added.at(-1) as TerminalInfo).id);
+				return;
+			}
+			case "reply":
+				return;
 			case "attached": {
 				const terminal = event.terminal;
 				const loaded = await loadGhostty();
@@ -1115,8 +1126,10 @@ const KEYS: { label: string; input?: KeyInput; text?: string }[] = [
 	{ label: "-", text: "-" },
 ];
 
-function label(terminal: TerminalInfo, index: number): string {
-	const name = terminal.title || `Terminal ${index + 1}`;
+/** A tab's text: the name it was opened with, else its program's title, else its number (the agent's `-t`). */
+function label(terminal: TerminalInfo): string {
+	// A host running older code numbers none.
+	const name = terminal.name || terminal.title || (terminal.index ? `Terminal ${terminal.index}` : "Terminal");
 	return terminal.exited ? `${name} (exited)` : name;
 }
 
@@ -1237,17 +1250,18 @@ export function TerminalView({
 	return (
 		<div className="terminal">
 			<div className="terminal-tabs">
-				{state.terminals.map((terminal, index) => (
+				{state.terminals.map((terminal) => (
 					<button
 						type="button"
 						key={terminal.id}
 						aria-pressed={terminal.id === current?.id}
 						className={terminal.exited ? "exited" : ""}
-						title={terminal.env ? `${terminal.env}:${terminal.cwd}` : terminal.cwd}
+						title={`${terminal.index}: ${terminal.env ? `${terminal.env}:${terminal.cwd}` : terminal.cwd}${terminal.agent ? " (opened by the agent)" : ""}`}
 						onPointerDown={keepFocus}
 						onClick={() => session.current?.attach(terminal.id)}
 					>
-						{label(terminal, index)}
+						{terminal.agent && <span className="terminal-agent">Pi</span>}
+						{label(terminal)}
 					</button>
 				))}
 				<button

apps/p2/plugins/terminals/commands.ts

@@ -0,0 +1,162 @@
+// The terminal tool's commands (worker): a tmux command name and its argv, parsed getopt-style as tmux parses them
+// (options first, flags combinable as `-lp`, values attached or separate, `--` ends options), into a typed command.
+// Nothing is split or unquoted: every argument arrives as the script passed it.
+
+/** What a wait waits for: no output for a while, the program's exit, a regular expression over the new output. */
+export type WaitFor =
+	| { kind: "none" }
+	| { kind: "quiet" }
+	| { kind: "exit" }
+	| { kind: "match"; pattern: string; ignoreCase: boolean };
+
+/** A wait as asked: what for, and the timeout in seconds when given. */
+export type WaitOptions = { wait?: WaitFor; timeout?: number };
+
+export type TerminalCommand =
+	| ({ command: "new-window"; name?: string; directory?: string; shellCommand?: string } & WaitOptions)
+	| { command: "list-windows" }
+	| ({ command: "send-keys"; target: string; literal: boolean; keys: string[] } & WaitOptions)
+	| ({ command: "read"; target: string } & WaitOptions)
+	| { command: "capture-pane"; target: string; scrollback: number; escapes: boolean }
+	| { command: "kill-window"; target: string };
+
+export const COMMANDS = ["new-window", "list-windows", "send-keys", "read", "capture-pane", "kill-window"] as const;
+export type CommandName = (typeof COMMANDS)[number];
+
+/** Longest timeout a wait may ask for, in seconds. */
+export const MAX_TIMEOUT_S = 120;
+
+/** A command line tmux would refuse: the message says why, tmux-style (`usage: …`). */
+export class CommandError extends Error {}
+
+/** Each command's options (a letter, with `:` when it takes a value) and usage line, as tmux prints it. */
+const SPECS: Record<CommandName, { options: string; usage: string }> = {
+	"new-window": {
+		options: "n:c:w:T:",
+		usage: "new-window [-n window-name] [-c start-directory] [-w wait] [-T seconds] [shell-command ...]",
+	},
+	"list-windows": { options: "", usage: "list-windows" },
+	"send-keys": { options: "lt:w:T:", usage: "send-keys [-l] [-w wait] [-T seconds] -t target-window key ..." },
+	read: { options: "t:w:T:", usage: "read [-w wait] [-T seconds] -t target-window" },
+	"capture-pane": { options: "ept:S:", usage: "capture-pane [-ep] [-S start-line] -t target-window" },
+	"kill-window": { options: "t:", usage: "kill-window -t target-window" },
+};
+
+/** Lines of scrollback `capture-pane -S -` asks for: all the host returns. */
+const ALL_SCROLLBACK = 100_000;
+
+/** Parses one command; throws a `CommandError` for anything tmux would refuse. */
+export function parseCommand(command: CommandName, args: readonly string[]): TerminalCommand {
+	const spec = SPECS[command];
+	const { options, rest } = getopt(command, spec, args);
+	const usage = () => new CommandError(`usage: ${spec.usage}`);
+	const target = () => {
+		const value = options.get("t");
+		if (value === undefined || value === "") throw usage();
+		return value;
+	};
+	const waits = (): WaitOptions => {
+		const wait = options.get("w");
+		const timeout = options.get("T");
+		return {
+			...(wait === undefined ? {} : { wait: parseWait(command, wait) }),
+			...(timeout === undefined ? {} : { timeout: parseTimeout(command, timeout) }),
+		};
+	};
+	switch (command) {
+		case "new-window": {
+			const name = options.get("n");
+			const directory = options.get("c");
+			return {
+				command,
+				...(name ? { name } : {}),
+				...(directory ? { directory } : {}),
+				...(rest.length > 0 ? { shellCommand: rest.join(" ") } : {}),
+				...waits(),
+			};
+		}
+		case "list-windows":
+			if (rest.length > 0) throw usage();
+			return { command };
+		case "send-keys":
+			if (rest.length === 0) throw usage();
+			return { command, target: target(), literal: options.has("l"), keys: [...rest], ...waits() };
+		case "read":
+			if (rest.length > 0) throw usage();
+			return { command, target: target(), ...waits() };
+		case "capture-pane": {
+			if (rest.length > 0) throw usage();
+			const start = options.get("S");
+			return { command, target: target(), scrollback: parseStart(command, start), escapes: options.has("e") };
+		}
+		case "kill-window":
+			if (rest.length > 0) throw usage();
+			return { command, target: target() };
+	}
+}
+
+/**
+ * tmux's getopt: options until the first argument that is not one (or `--`); flags combine (`-ep`); an option's value
+ * is the rest of its argument or the next argument, even one starting with `-` (`-S -50`).
+ */
+function getopt(
+	command: string,
+	spec: { options: string; usage: string },
+	args: readonly string[],
+): { options: Map<string, string>; rest: string[] } {
+	const options = new Map<string, string>();
+	let index = 0;
+	for (; index < args.length; index++) {
+		const arg = args[index] as string;
+		if (arg === "--") {
+			index++;
+			break;
+		}
+		if (!arg.startsWith("-") || arg === "-") break;
+		for (let at = 1; at < arg.length; at++) {
+			const letter = arg[at] as string;
+			const position = spec.options.indexOf(letter);
+			if (position < 0 || letter === ":")
+				throw new CommandError(`${command}: unknown option -- ${letter}\nusage: ${spec.usage}`);
+			if (spec.options[position + 1] !== ":") {
+				options.set(letter, "");
+				continue;
+			}
+			const value = at + 1 < arg.length ? arg.slice(at + 1) : args[++index];
+			if (value === undefined)
+				throw new CommandError(`${command}: option requires an argument -- ${letter}\nusage: ${spec.usage}`);
+			options.set(letter, value);
+			break;
+		}
+	}
+	return { options, rest: args.slice(index) };
+}
+
+/** `quiet`, `exit`, `none` or `/regex/` (`/regex/i` ignores case). */
+function parseWait(command: string, value: string): WaitFor {
+	if (value === "quiet" || value === "exit" || value === "none") return { kind: value };
+	const match = /^\/(.*)\/(i?)$/s.exec(value);
+	if (!match) throw new CommandError(`${command}: invalid wait: ${value} (quiet, exit, none or /regex/)`);
+	const pattern = match[1] as string;
+	try {
+		new RegExp(pattern);
+	} catch (error) {
+		throw new CommandError(`${command}: invalid regex: ${error instanceof Error ? error.message : String(error)}`);
+	}
+	return { kind: "match", pattern, ignoreCase: match[2] === "i" };
+}
+
+function parseTimeout(command: string, value: string): number {
+	const seconds = Number(value);
+	if (value.trim() === "" || !Number.isFinite(seconds) || seconds < 0 || seconds > MAX_TIMEOUT_S)
+		throw new CommandError(`${command}: invalid timeout: ${value} (0 to ${MAX_TIMEOUT_S} seconds)`);
+	return seconds;
+}
+
+/** `-S`: the first line, `-N` for N lines of scrollback, `-` for all of it; 0 or none for the screen only. */
+function parseStart(command: string, value: string | undefined): number {
+	if (value === undefined) return 0;
+	if (value === "-") return ALL_SCROLLBACK;
+	if (!/^-?\d+$/.test(value)) throw new CommandError(`${command}: invalid start line: ${value}`);
+	return Math.min(ALL_SCROLLBACK, Math.max(0, -Number(value)));
+}

apps/p2/plugins/terminals/contract.ts

@@ -1,5 +1,15 @@
-import { bool, int, obj, opt, str, union } from "@earendil-works/pi-schema";
-import { defineRegistry, literal, local, method, param, propertiesOf, remote, serverHost } from "@p2/app/v1";
+import { arr, bool, int, nul, obj, opt, str, union } from "@earendil-works/pi-schema";
+import {
+	defineRegistry,
+	LocationSchema,
+	literal,
+	local,
+	method,
+	param,
+	propertiesOf,
+	remote,
+	serverHost,
+} from "@p2/app/v1";
 import type { TerminalService } from "./service.ts";
 
 const Empty = obj({});
@@ -47,3 +57,86 @@ export const Terminals = remote("p2.terminals", {
 		},
 	},
 });
+
+const Target = str({ minLength: 1, maxLength: 100 });
+/** A key making a call that changes something idempotent: the host does a call with a key it has seen only once. */
+const CallKey = str({ minLength: 1, maxLength: 200 });
+export const TerminalInfoSchema = obj({
+	id: str(),
+	sessionId: str(),
+	index: int(),
+	name: opt(str()),
+	agent: opt(bool()),
+	cwd: str(),
+	env: union([str(), nul()]),
+	title: str(),
+	createdAt: int(),
+	cols: int(),
+	rows: int(),
+	exited: opt(obj({ code: opt(int()), error: opt(str()) })),
+});
+export const TerminalWaitSchema = obj({
+	until: union([literal("quiet"), literal("exit"), literal("match")]),
+	pattern: opt(str({ maxLength: 1000 })),
+	ignoreCase: opt(bool()),
+	quietMs: opt(int({ minimum: 50, maximum: 60_000 })),
+	timeoutMs: int({ minimum: 0, maximum: 120_000 }),
+});
+export const TerminalReadSchema = obj({
+	terminal: TerminalInfoSchema,
+	from: int({ minimum: 0 }),
+	to: int({ minimum: 0 }),
+	text: str(),
+	skipped: int({ minimum: 0 }),
+	truncated: int({ minimum: 0 }),
+	screen: opt(str()),
+	waited: opt(union([literal("quiet"), literal("exit"), literal("match"), literal("timeout"), literal("closed")])),
+	foreground: opt(str()),
+});
+
+/**
+ * The terminal tool's calls from a session worker (the agent): its session's terminals, the same the owner sees in
+ * the terminal panel. The session comes from the worker's identity; only its current generation is served, never an
+ * archived session, and new terminals only on environments the session works on.
+ */
+export const AgentTerminals = remote("p2.agent-terminals", {
+	version: 1,
+	params: {},
+	hostAddress: serverHost,
+	variants: {
+		worker: {
+			methods: {
+				/** The session's terminals, each with the job in front of its shell while one runs. */
+				list: method(Empty, obj({ windows: arr(obj({ terminal: TerminalInfoSchema, foreground: opt(str()) })) })),
+				/** A new terminal at `location`, a shell; `command` is typed into it. */
+				open: method(
+					obj({
+						key: CallKey,
+						location: LocationSchema,
+						name: opt(str({ minLength: 1, maxLength: 60 })),
+						command: opt(str({ maxLength: 64 * 1024 })),
+					}),
+					TerminalInfoSchema,
+				),
+				/** tmux `send-keys`: key names as keys, other arguments as text; all text when `literal`. */
+				send: method(
+					obj({ key: CallKey, target: Target, keys: arr(str({ maxLength: 64 * 1024 })), literal: bool() }),
+					TerminalInfoSchema,
+				),
+				/** Where the next read without `from` starts. */
+				cursor: method(obj({ target: Target }), obj({ terminal: TerminalInfoSchema, offset: int({ minimum: 0 }) })),
+				/** Output since `from` (default: the read cursor) as plain text, after an optional wait. */
+				read: method(
+					obj({ target: Target, from: opt(int({ minimum: 0 })), wait: opt(TerminalWaitSchema) }),
+					TerminalReadSchema,
+				),
+				/** tmux `capture-pane`: the screen as plain text, up to `scrollback` lines before it; styled with `escapes`. */
+				capture: method(
+					obj({ target: Target, scrollback: int({ minimum: 0, maximum: 100_000 }), escapes: opt(bool()) }),
+					obj({ terminal: TerminalInfoSchema, text: str() }),
+				),
+				close: method(obj({ key: CallKey, target: Target }), obj({ terminalId: str() })),
+			},
+		},
+	},
+});

apps/p2/plugins/terminals/detached.ts

@@ -31,6 +31,8 @@ export async function terminalHostVersion(app: string): Promise<string> {
 		"detached.ts",
 		"envs.ts",
 		"ghostty.ts",
+		"keys.ts",
+		"plain.ts",
 		"vendor/ghostty/ghostty-vt.wasm",
 	]) {
 		hash.update(file);
@@ -50,6 +52,15 @@ function connectOnce(address: string): Promise<Socket> {
 	});
 }
 
+/** Connects to the data directory's terminal host when one runs; undefined when none does (none is started). */
+export async function connectRunningTerminalHost(directory: string): Promise<Socket | undefined> {
+	try {
+		return await connectOnce(await terminalHostAddress(directory));
+	} catch {
+		return undefined;
+	}
+}
+
 /** Connects to the data directory's terminal host, starting it from the p2 app directory `app` when none runs. */
 export async function connectTerminalHost(directory: string, app: string): Promise<Socket> {
 	const address = await terminalHostAddress(directory);

apps/p2/plugins/terminals/ghostty.ts

@@ -38,6 +38,8 @@ const CELL_GRAPHEMES_BUF = 4;
 const CELL_BG = 5;
 const CELL_FG = 6;
 const CELL_HAS_STYLING = 8;
+// GhosttyFormatterFormat
+const FORMAT_PLAIN = 0;
 const FORMAT_VT = 1;
 // GhosttyCellData, GhosttyCellWide
 const CELL_DATA_WIDE = 3;
@@ -556,6 +558,18 @@ export class VtTerminal {
 	 * size reproduces this one, for a browser that attaches later.
 	 */
 	snapshot(): Uint8Array {
+		return this.#format(FORMAT_VT);
+	}
+
+	/**
+	 * The active screen as plain text, with the scrollback before it (the alternate screen of a full-screen program has
+	 * none): soft-wrapped lines joined, trailing spaces trimmed.
+	 */
+	text(): string {
+		return new TextDecoder().decode(this.#format(FORMAT_PLAIN));
+	}
+
+	#format(emit: number): Uint8Array {
 		const ghostty = this.#ghostty;
 		const type = "GhosttyFormatterTerminalOptions";
 		const size = ghostty.size(type);
@@ -568,7 +582,7 @@ export class VtTerminal {
 			const flag = (base: number, struct: string, name: string) =>
 				view.setUint8(base + ghostty.field(struct, name), 1);
 			view.setUint32(options + ghostty.field(type, "size"), size, true);
-			view.setUint32(options + ghostty.field(type, "emit"), FORMAT_VT, true);
+			view.setUint32(options + ghostty.field(type, "emit"), emit, true);
 			flag(options, type, "unwrap");
 			flag(options, type, "trim");
 			view.setUint32(
@@ -576,17 +590,19 @@ export class VtTerminal {
 				ghostty.size("GhosttyFormatterTerminalExtra"),
 				true,
 			);
-			// Not `tabstops`: the formatter emits them after the cursor position, moving the cursor (programs rarely
-			// change the default stops anyway).
-			for (const name of ["modes", "scrolling_region", "pwd", "keyboard"])
-				flag(extra, "GhosttyFormatterTerminalExtra", name);
 			view.setUint32(
 				screen + ghostty.field("GhosttyFormatterScreenExtra", "size"),
 				ghostty.size("GhosttyFormatterScreenExtra"),
 				true,
 			);
-			for (const name of ["cursor", "style", "hyperlink", "protection", "kitty_keyboard", "charsets"])
-				flag(screen, "GhosttyFormatterScreenExtra", name);
+			if (emit === FORMAT_VT) {
+				// Not `tabstops`: the formatter emits them after the cursor position, moving the cursor (programs rarely
+				// change the default stops anyway).
+				for (const name of ["modes", "scrolling_region", "pwd", "keyboard"])
+					flag(extra, "GhosttyFormatterTerminalExtra", name);
+				for (const name of ["cursor", "style", "hyperlink", "protection", "kitty_keyboard", "charsets"])
+					flag(screen, "GhosttyFormatterScreenExtra", name);
+			}
 			formatter = ghostty.create("ghostty_formatter_terminal_new", this.handle, options);
 			const result = ghostty.call("ghostty_formatter_format_alloc", formatter, 0, this.#scratch, this.#scratch + 4);
 			if (result !== SUCCESS) throw new Error(`Formatting the terminal failed (${result})`);

apps/p2/plugins/terminals/keys.ts

@@ -0,0 +1,102 @@
+// tmux key names (`send-keys`) as key presses for libghostty's encoder (terminal host): `Enter`, `C-c`, `M-x`, `S-Up`,
+// `F5`, `^D`. The host encodes them in the terminal's current modes (application cursor keys, the kitty protocol), as a
+// browser's key presses are.
+
+import type { KeyInput } from "./ghostty.ts";
+
+/** tmux's names for keys without a character (key-string.c), by `KeyboardEvent.code`. */
+const NAMED: Record<string, { code: string; key: string; shift?: boolean }> = {
+	Enter: { code: "Enter", key: "Enter" },
+	Escape: { code: "Escape", key: "Escape" },
+	Tab: { code: "Tab", key: "Tab" },
+	BTab: { code: "Tab", key: "Tab", shift: true },
+	BSpace: { code: "Backspace", key: "Backspace" },
+	Space: { code: "Space", key: " " },
+	Up: { code: "ArrowUp", key: "ArrowUp" },
+	Down: { code: "ArrowDown", key: "ArrowDown" },
+	Left: { code: "ArrowLeft", key: "ArrowLeft" },
+	Right: { code: "ArrowRight", key: "ArrowRight" },
+	Home: { code: "Home", key: "Home" },
+	End: { code: "End", key: "End" },
+	PageUp: { code: "PageUp", key: "PageUp" },
+	PgUp: { code: "PageUp", key: "PageUp" },
+	PPage: { code: "PageUp", key: "PageUp" },
+	PageDown: { code: "PageDown", key: "PageDown" },
+	PgDn: { code: "PageDown", key: "PageDown" },
+	NPage: { code: "PageDown", key: "PageDown" },
+	IC: { code: "Insert", key: "Insert" },
+	Insert: { code: "Insert", key: "Insert" },
+	DC: { code: "Delete", key: "Delete" },
+	Delete: { code: "Delete", key: "Delete" },
+	...Object.fromEntries(
+		Array.from({ length: 12 }, (_, index) => [`F${index + 1}`, { code: `F${index + 1}`, key: `F${index + 1}` }]),
+	),
+};
+
+/** Physical keys of the punctuation characters (a US layout), so Ctrl and Alt combine with them as typed. */
+const PUNCTUATION: Record<string, string> = {
+	"`": "Backquote",
+	"\\": "Backslash",
+	"[": "BracketLeft",
+	"]": "BracketRight",
+	",": "Comma",
+	"=": "Equal",
+	"-": "Minus",
+	".": "Period",
+	"'": "Quote",
+	";": "Semicolon",
+	"/": "Slash",
+};
+
+/** The key press a tmux key name stands for; undefined when it names no key (tmux then sends it as text). */
+export function tmuxKey(name: string): KeyInput | undefined {
+	let rest = name;
+	const mods = { ctrl: false, alt: false, shift: false };
+	// Modifier prefixes in any order: `C-M-x`; `^x` is Ctrl too.
+	for (;;) {
+		const prefix = /^([CMS])-(?=.)/.exec(rest);
+		if (prefix) {
+			if (prefix[1] === "C") mods.ctrl = true;
+			else if (prefix[1] === "M") mods.alt = true;
+			else mods.shift = true;
+			rest = rest.slice(2);
+			continue;
+		}
+		if (rest.length === 2 && rest.startsWith("^")) {
+			mods.ctrl = true;
+			rest = rest.slice(1);
+		}
+		break;
+	}
+	// C-[ is Escape in tmux (and every terminal); libghostty sends it as a distinct key.
+	if (mods.ctrl && rest === "[") {
+		mods.ctrl = false;
+		rest = "Escape";
+	}
+	const named = NAMED[rest];
+	if (named)
+		return {
+			code: named.code,
+			key: named.key,
+			...(named.shift || mods.shift ? { shift: true } : {}),
+			...(mods.ctrl ? { ctrl: true } : {}),
+			...(mods.alt ? { alt: true } : {}),
+		};
+	// A single character only names a key with a modifier: plain `a` is text.
+	if ([...rest].length !== 1 || rest === name) return undefined;
+	const lower = rest.toLowerCase();
+	const code = /^[a-z]$/.test(lower)
+		? `Key${lower.toUpperCase()}`
+		: /^[0-9]$/.test(rest)
+			? `Digit${rest}`
+			: (PUNCTUATION[rest] ?? (rest === " " ? "Space" : ""));
+	// tmux's C-A is C-a: Ctrl ignores case. An uppercase letter otherwise means Shift.
+	const shift = mods.shift || (!mods.ctrl && rest !== lower);
+	return {
+		code,
+		key: shift ? rest.toUpperCase() : lower,
+		...(shift ? { shift: true } : {}),
+		...(mods.ctrl ? { ctrl: true } : {}),
+		...(mods.alt ? { alt: true } : {}),
+	};
+}

apps/p2/plugins/terminals/main.ts

@@ -7,15 +7,34 @@ import { takeProcessLock } from "../../src/runtime/process-lock.ts";
 import { localSystem, requireTerminalSupport, terminalDaemon } from "./daemon.ts";
 import { terminalHostAddress, terminalHostVersion } from "./detached.ts";
 import { envRevision, loadEnvs, sshOptions } from "./envs.ts";
-import { Ghostty, type VtTerminal } from "./ghostty.ts";
+import {
+	BLINK,
+	BOLD,
+	type Cell,
+	FAINT,
+	Ghostty,
+	INVERSE,
+	INVISIBLE,
+	ITALIC,
+	STRIKETHROUGH,
+	UNDERLINE,
+	type VtTerminal,
+} from "./ghostty.ts";
+import { tmuxKey } from "./keys.ts";
+import { OutputRing, plainText } from "./plain.ts";
 import {
 	encodeFrame,
+	type HostCall,
+	type HostResult,
+	READ_TEXT_LIMIT,
 	readFrames,
 	TERMINAL_OUTPUT_WINDOW,
 	type TerminalEvent,
 	type TerminalInfo,
+	type TerminalRead,
 	type TerminalRequest,
 	type TerminalTheme,
+	type TerminalWait,
 } from "./protocol.ts";
 
 // The terminal host: owns the pi-env connections and the programs running in terminals, independent of the server and
@@ -24,6 +43,10 @@ import {
 // and no connections for a while. One per data directory. It is its own program, started by path (not a facet built by
 // the plugin loader), so it and the files only it runs (daemon.ts, envs.ts, log-preload.ts) import p2's base code and
 // Chord directly.
+//
+// The agent's terminal tool reaches the same terminals through calls with replies (`call`, see `HostCall`): each
+// terminal keeps its last MiB of output and a read cursor for it here, so reads continue where the last one ended,
+// also across server and worker restarts.
 
 const IDLE_EXIT_MS = 10 * 60_000;
 const SCROLLBACK = 10_000;
@@ -31,6 +54,18 @@ const SCROLLBACK = 10_000;
 const MAX_BACKLOG = 8 * 1024 * 1024;
 /** The colors programs see until a viewer reports its theme: the browser's dark theme. */
 const DEFAULT_THEME: TerminalTheme = { foreground: 0xe6e6e6, background: 0x141414, scheme: "dark" };
+/** Output each terminal keeps for the agent's reads. */
+const RING_BYTES = 1024 * 1024;
+/** Running terminals the agent may have opened in one session. */
+const MAX_AGENT_TERMINALS = 8;
+/** Results of calls with a key the host remembers, so a repeated call (a replayed tool call) is not done twice. */
+const REMEMBERED_CALLS = 1024;
+/** A `match` wait tests the plain text of at most this much of the latest output. */
+const MATCH_WINDOW = 64 * 1024;
+/** How long without output a `quiet` wait waits by default. */
+const DEFAULT_QUIET_MS = 500;
+/** Lines of scrollback `capture` returns at most. */
+const MAX_CAPTURE_SCROLLBACK = 5000;
 
 const directory = process.argv[2];
 if (!directory) throw new Error("usage: main.ts <data directory>");
@@ -69,6 +104,10 @@ type Client = {
 	/** `sent` right after the latest snapshot. */
 	snapshotEnd: number;
 	behind: boolean;
+	/** The session it listed last: it hears about that session's terminals changing (`terminals`). */
+	session?: string;
+	/** Cancels of its calls still running, by call id. */
+	calls: Map<number, () => void>;
 };
 type Hosted = {
 	info: TerminalInfo;
@@ -81,6 +120,12 @@ type Hosted = {
 	pendingBytes: number;
 	clients: Set<Client>;
 	connection: string;
+	/** The last MiB of output, for the agent's reads. */
+	ring: OutputRing;
+	/** Where the agent's next read starts (an offset into `ring`). */
+	cursor: number;
+	/** Waits for output, called on each output, the exit and the close. */
+	waiters: Set<() => void>;
 };
 
 const terminals = new Map<string, Hosted>();
@@ -122,6 +167,18 @@ function resync(client: Client, hosted: Hosted): void {
 
 function changed(hosted: Hosted): void {
 	for (const client of hosted.clients) send(client, { type: "changed", terminal: hosted.info });
+	listed(hosted.info.sessionId);
+}
+
+/** The session's terminals, oldest first. */
+function sessionTerminals(sessionId: string): Hosted[] {
+	return [...terminals.values()].filter((hosted) => hosted.info.sessionId === sessionId);
+}
+
+/** Tells every connection that listed the session (its viewers) its terminals now. */
+function listed(sessionId: string): void {
+	const list = sessionTerminals(sessionId).map((hosted) => hosted.info);
+	for (const client of clients) if (client.session === sessionId) send(client, { type: "terminals", terminals: list });
 }
 
 async function connection(env: string | null): Promise<{ connection: Connection; shellEnv: Record<string, string> }> {
@@ -198,11 +255,28 @@ function attach(client: Client, hosted: Hosted): void {
 	resync(client, hosted);
 }
 
-async function open(client: Client, request: Extract<TerminalRequest, { type: "open" }>): Promise<void> {
+type Opening = {
+	sessionId: string;
+	cwd: string;
+	env: string | null;
+	cols: number;
+	rows: number;
+	name?: string;
+	agent?: true;
+};
+
+/** A new terminal, listed at once; `started` settles once its program started or failed to. */
+function create(request: Opening, theme: TerminalTheme): { hosted: Hosted; started: Promise<void> } {
 	const id = randomBytes(6).toString("base64url");
+	const taken = new Set(sessionTerminals(request.sessionId).map((each) => each.info.index));
+	let index = 1;
+	while (taken.has(index)) index++;
 	const info: TerminalInfo = {
 		id,
 		sessionId: request.sessionId,
+		index,
+		...(request.name ? { name: request.name } : {}),
+		...(request.agent ? { agent: true as const } : {}),
 		cwd: request.cwd,
 		env: request.env,
 		title: "",
@@ -211,76 +285,102 @@ async function open(client: Client, request: Extract<TerminalRequest, { type: "o
 		rows: request.rows,
 	};
 	let pty: RemoteTerminal | undefined;
-	let hosted: Hosted | undefined;
 	const vt = ghostty.terminal({
 		cols: request.cols,
 		rows: request.rows,
 		scrollback: SCROLLBACK,
-		colors: client.theme,
-		colorScheme: () => hosted?.theme.scheme,
+		colors: theme,
+		colorScheme: () => hosted.theme.scheme,
 		// Answers to queries (cursor position, device attributes) come from here, never from the browsers.
 		onReply: (bytes) => {
 			if (terminals.has(id)) pty?.write(bytes);
 		},
 	});
-	hosted = {
+	const hosted: Hosted = {
 		info,
 		vt,
-		theme: client.theme,
+		theme,
 		pending: [],
 		pendingBytes: 0,
 		clients: new Set(),
 		connection: request.env ?? "",
+		ring: new OutputRing(RING_BYTES),
+		cursor: 0,
+		waiters: new Set(),
 	};
 	terminals.set(id, hosted);
-	attach(client, hosted);
+	listed(request.sessionId);
 	const end = (exited: TerminalInfo["exited"]) => {
 		if (terminals.get(id) !== hosted) return;
 		info.exited = exited;
 		changed(hosted);
+		poke(hosted);
 		release(hosted.connection);
 	};
-	try {
-		const { connection: daemon, shellEnv } = await connection(request.env);
-		pty = await openTerminal(daemon, {
-			cwd: request.cwd,
-			// COLORFGBG: light or dark for programs that only look at the environment.
-			env: {
-				...shellEnv,
-				TERM: "xterm-256color",
-				COLORTERM: "truecolor",
-				COLORFGBG: client.theme.scheme === "dark" ? "15;0" : "0;15",
-			},
-			cols: request.cols,
-			rows: request.rows,
-			onData: (bytes) => {
-				// Output racing a close: the headless terminal is already freed.
-				if (!terminals.has(id)) return;
-				vt.write(bytes);
-				for (const viewer of hosted.clients) output(viewer, bytes);
-				const title = vt.title();
-				if (title !== info.title) {
-					info.title = title;
-					changed(hosted);
-				}
-			},
-		});
-		hosted.pty = pty;
-		// Closed meanwhile: the program must not outlive its terminal.
-		if (!terminals.has(id)) pty.close();
-		if (info.cols !== request.cols || info.rows !== request.rows) pty.resize(info.cols, info.rows);
-		for (const bytes of hosted.pending.splice(0)) pty.write(bytes);
-		hosted.pendingBytes = 0;
-		pty.exited.then(
-			(code) => end({ code }),
-			(error: unknown) => end({ error: error instanceof Error ? error.message : String(error) }),
-		);
-	} catch (error) {
-		end({ error: error instanceof Error ? error.message : String(error) });
+	const started = (async () => {
+		try {
+			const { connection: daemon, shellEnv } = await connection(request.env);
+			pty = await openTerminal(daemon, {
+				cwd: request.cwd,
+				// COLORFGBG: light or dark for programs that only look at the environment.
+				env: {
+					...shellEnv,
+					TERM: "xterm-256color",
+					COLORTERM: "truecolor",
+					COLORFGBG: theme.scheme === "dark" ? "15;0" : "0;15",
+				},
+				cols: request.cols,
+				rows: request.rows,
+				onData: (bytes) => {
+					// Output racing a close: the headless terminal is already freed.
+					if (!terminals.has(id)) return;
+					vt.write(bytes);
+					hosted.ring.append(bytes);
+					for (const viewer of hosted.clients) output(viewer, bytes);
+					const title = vt.title();
+					if (title !== info.title) {
+						info.title = title;
+						changed(hosted);
+					}
+					poke(hosted);
+				},
+			});
+			hosted.pty = pty;
+			// Closed meanwhile: the program must not outlive its terminal.
+			if (!terminals.has(id)) pty.close();
+			if (info.cols !== request.cols || info.rows !== request.rows) pty.resize(info.cols, info.rows);
+			for (const bytes of hosted.pending.splice(0)) pty.write(bytes);
+			hosted.pendingBytes = 0;
+			pty.exited.then(
+				(code) => end({ code }),
+				(error: unknown) => end({ error: error instanceof Error ? error.message : String(error) }),
+			);
+		} catch (error) {
+			end({ error: error instanceof Error ? error.message : String(error) });
+		}
+	})();
+	return { hosted, started };
+}
+
+/** Wakes the terminal's waits. */
+function poke(hosted: Hosted): void {
+	for (const waiter of hosted.waiters) waiter();
+}
+
+/** Input for the program; held until it started. False when too much is held already. */
+function input(hosted: Hosted, bytes: Uint8Array): boolean {
+	if (hosted.pty) {
+		hosted.pty.write(bytes);
+		return true;
 	}
+	if (hosted.pendingBytes + bytes.length > MAX_BACKLOG) return false;
+	hosted.pending.push(bytes);
+	hosted.pendingBytes += bytes.length;
+	return true;
 }
 
 function close(hosted: Hosted): void {
+	if (terminals.get(hosted.info.id) !== hosted) return;
 	terminals.delete(hosted.info.id);
 	hosted.pty?.close();
 	for (const client of hosted.clients) {
@@ -290,37 +390,31 @@ function close(hosted: Hosted): void {
 	hosted.clients.clear();
 	hosted.vt.free();
 	release(hosted.connection);
+	poke(hosted);
+	listed(hosted.info.sessionId);
 }
 
 function handle(client: Client, message: unknown): void {
 	if (message instanceof Uint8Array) {
 		const hosted = client.attached;
 		if (!hosted || hosted.info.exited) return;
-		if (hosted.pty) hosted.pty.write(message);
-		else {
-			if (hosted.pendingBytes + message.length > MAX_BACKLOG) {
-				client.socket.destroy();
-				return;
-			}
-			hosted.pending.push(message);
-			hosted.pendingBytes += message.length;
-		}
+		if (!input(hosted, message)) client.socket.destroy();
 		return;
 	}
 	const request = message as TerminalRequest;
 	switch (request.type) {
 		case "list":
-			send(client, {
-				type: "list",
-				terminals: [...terminals.values()]
-					.filter((hosted) => hosted.info.sessionId === request.sessionId)
-					.map((hosted) => hosted.info),
-			});
+			client.session = request.sessionId;
+			send(client, { type: "list", terminals: sessionTerminals(request.sessionId).map((hosted) => hosted.info) });
 			return;
 		case "open":
-			void open(client, request).catch((error: unknown) =>
-				send(client, { type: "error", message: error instanceof Error ? error.message : String(error) }),
-			);
+			attach(client, create(request, client.theme).hosted);
+			return;
+		case "call":
+			call(client, request.id, request.call);
+			return;
+		case "cancel":
+			client.calls.get(request.id)?.();
 			return;
 		case "attach": {
 			const hosted = terminals.get(request.terminalId);
@@ -385,8 +479,292 @@ function handle(client: Client, message: unknown): void {
 	}
 }
 
+/** Results of calls with a key, by session, operation and key; the oldest go first. */
+const remembered = new Map<string, Promise<HostResult>>();
+
+/** Runs a call and replies; the caller may cancel it while it waits. */
+function call(client: Client, id: number, request: HostCall): void {
+	const cancelled = Promise.withResolvers<never>();
+	cancelled.promise.catch(() => {});
+	client.calls.set(id, () => cancelled.reject(new Error("cancelled")));
+	const result = "key" in request ? once(request) : perform(request, cancelled.promise);
+	Promise.race([result, cancelled.promise])
+		.then(
+			(value) => send(client, { type: "reply", id, result: value }),
+			(error: unknown) =>
+				send(client, { type: "reply", id, error: error instanceof Error ? error.message : String(error) }),
+		)
+		.finally(() => client.calls.delete(id));
+}
+
+/** A call with a key, done once: a call with a key the host has seen gets the first one's result. */
+function once(request: Extract<HostCall, { key: string }>): Promise<HostResult> {
+	const key = JSON.stringify([request.sessionId, request.op, request.key]);
+	const known = remembered.get(key);
+	if (known) return known;
+	// Calls with a key never wait, so nothing cancels them.
+	const first = perform(request, new Promise<never>(() => {}));
+	remembered.set(key, first);
+	// A failed call may be tried again.
+	first.catch(() => {
+		if (remembered.get(key) === first) remembered.delete(key);
+	});
+	for (const old of remembered.keys()) {
+		if (remembered.size <= REMEMBERED_CALLS) break;
+		remembered.delete(old);
+	}
+	return first;
+}
+
+/** One call; `cancelled` rejects when the caller gives up (a wait then ends). */
+async function perform(request: HostCall, cancelled: Promise<never>): Promise<HostResult> {
+	switch (request.op) {
+		case "list":
+			return {
+				kind: "list",
+				windows: await Promise.all(
+					sessionTerminals(request.sessionId).map(async (hosted) => {
+						const job = await foreground(hosted);
+						return { terminal: hosted.info, ...(job ? { foreground: job } : {}) };
+					}),
+				),
+			};
+		case "open":
+			return { kind: "terminal", terminal: await openForAgent(request) };
+		case "send": {
+			const hosted = target(request.sessionId, request.target);
+			if (hosted.info.exited) throw new Error(`window ${hosted.info.index} has exited`);
+			const parts: Uint8Array[] = [];
+			for (const key of request.keys) {
+				const press = request.literal ? undefined : tmuxKey(key);
+				const bytes = press ? hosted.vt.encodeKey(press) : new TextEncoder().encode(key);
+				if (!bytes) throw new Error(`unknown key: ${key}`);
+				parts.push(bytes);
+			}
+			if (!input(hosted, Buffer.concat(parts))) throw new Error("too much input before the program started");
+			return { kind: "terminal", terminal: hosted.info };
+		}
+		case "cursor": {
+			const hosted = target(request.sessionId, request.target);
+			return { kind: "cursor", terminal: hosted.info, offset: hosted.cursor };
+		}
+		case "read":
+			return { kind: "read", ...(await read(request, cancelled)) };
+		case "capture": {
+			const hosted = target(request.sessionId, request.target);
+			return {
+				kind: "capture",
+				terminal: hosted.info,
+				text: capture(hosted.vt, request.scrollback, request.escapes === true),
+			};
+		}
+		case "close": {
+			const hosted = target(request.sessionId, request.target);
+			close(hosted);
+			return { kind: "closed", terminalId: hosted.info.id };
+		}
+		case "closeSession":
+			for (const hosted of sessionTerminals(request.sessionId)) close(hosted);
+			return { kind: "done" };
+	}
+}
+
+/** The session's terminal `name`: its id, its number (`2` or tmux's `:2`), or its name. */
+function target(sessionId: string, name: string): Hosted {
+	const own = sessionTerminals(sessionId);
+	const bare = name.startsWith(":") ? name.slice(1) : name;
+	const found =
+		own.find((hosted) => hosted.info.id === name) ??
+		own.find((hosted) => String(hosted.info.index) === bare) ??
+		own.find((hosted) => hosted.info.name === bare);
+	if (!found) throw new Error(`can't find window: ${name}`);
+	return found;
+}
+
+/** A terminal the agent opens; its program has started when this returns, or it is gone with the error. */
+async function openForAgent(request: Extract<HostCall, { op: "open" }>): Promise<TerminalInfo> {
+	const running = sessionTerminals(request.sessionId).filter((hosted) => hosted.info.agent && !hosted.info.exited);
+	if (running.length >= MAX_AGENT_TERMINALS)
+		throw new Error(
+			`the agent has ${MAX_AGENT_TERMINALS} terminals running in this session already: close one with kill-window`,
+		);
+	const { hosted, started } = create(
+		{
+			sessionId: request.sessionId,
+			cwd: request.cwd,
+			env: request.env,
+			cols: request.cols,
+			rows: request.rows,
+			agent: true,
+			...(request.name ? { name: request.name } : {}),
+		},
+		DEFAULT_THEME,
+	);
+	// The shell reads it once it is ready, as typed ahead.
+	if (request.command) input(hosted, new TextEncoder().encode(`${request.command}\r`));
+	await started;
+	const failed = hosted.info.exited?.error;
+	if (failed) {
+		close(hosted);
+		throw new Error(failed);
+	}
+	return hosted.info;
+}
+
+/** Output since `from` (default: the read cursor) as plain text, after the wait; moves the read cursor past it. */
+async function read(request: Extract<HostCall, { op: "read" }>, cancelled: Promise<never>): Promise<TerminalRead> {
+	const hosted = target(request.sessionId, request.target);
+	const from = request.from ?? hosted.cursor;
+	const waited = request.wait ? await Promise.race([waitFor(hosted, from, request.wait), cancelled]) : undefined;
+	const { bytes, skipped } = hosted.ring.since(from);
+	const to = hosted.ring.end;
+	hosted.cursor = Math.max(hosted.cursor, to);
+	const plain = plainText(new TextDecoder().decode(bytes));
+	const truncated = Math.max(0, plain.length - READ_TEXT_LIMIT);
+	const open = terminals.get(hosted.info.id) === hosted;
+	const screen = open && hosted.vt.alternateScreen() ? capture(hosted.vt, 0, false) : undefined;
+	const job = await foreground(hosted);
+	return {
+		terminal: hosted.info,
+		from,
+		to,
+		text: truncated ? plain.slice(truncated) : plain,
+		skipped,
+		truncated,
+		...(screen === undefined ? {} : { screen }),
+		...(waited ? { waited } : {}),
+		...(job ? { foreground: job } : {}),
+	};
+}
+
+/** The name of the job in front of the terminal's shell, while one runs. */
+async function foreground(hosted: Hosted): Promise<string | undefined> {
+	if (terminals.get(hosted.info.id) !== hosted || !hosted.pty || hosted.info.exited) return undefined;
+	const job = await hosted.pty.foreground().catch(() => undefined);
+	return job ? (job.name ?? `process ${job.pid}`) : undefined;
+}
+
+/**
+ * Waits for the terminal: until its output is quiet, matches, or it exits, whichever was asked, or the timeout. It
+ * also ends when the program exits or the terminal closes.
+ */
+function waitFor(hosted: Hosted, from: number, wait: TerminalWait): Promise<NonNullable<TerminalRead["waited"]>> {
+	// A bad pattern fails the read before it waits.
+	const pattern = wait.until === "match" ? new RegExp(wait.pattern ?? "", wait.ignoreCase ? "mi" : "m") : undefined;
+	return new Promise((resolve) => {
+		let quiet: ReturnType<typeof setTimeout> | undefined;
+		let check: ReturnType<typeof setTimeout> | undefined;
+		const matches = () => {
+			const { bytes } = hosted.ring.since(Math.max(from, hosted.ring.end - MATCH_WINDOW));
+			return pattern?.test(plainText(new TextDecoder().decode(bytes))) === true;
+		};
+		const finish = (why: NonNullable<TerminalRead["waited"]>) => {
+			clearTimeout(timeout);
+			clearTimeout(quiet);
+			clearTimeout(check);
+			hosted.waiters.delete(waiter);
+			resolve(why);
+		};
+		const waiter = () => {
+			if (terminals.get(hosted.info.id) !== hosted) finish(pattern && matches() ? "match" : "closed");
+			else if (hosted.info.exited) finish(pattern && matches() ? "match" : "exit");
+			else if (wait.until === "quiet") {
+				clearTimeout(quiet);
+				quiet = setTimeout(() => finish("quiet"), wait.quietMs ?? DEFAULT_QUIET_MS);
+			} else if (pattern && check === undefined)
+				// Output comes in many small chunks: test once they settled a little.
+				check = setTimeout(() => {
+					check = undefined;
+					if (matches()) finish("match");
+				}, 50);
+		};
+		const timeout = setTimeout(() => finish("timeout"), wait.timeoutMs);
+		hosted.waiters.add(waiter);
+		if (pattern && matches()) finish("match");
+		else waiter();
+	});
+}
+
+/**
+ * The screen as plain text, with up to `scrollback` lines before it; trailing blank lines left out. `escapes` keeps
+ * colors and attributes as SGR sequences (tmux's `capture-pane -e`).
+ */
+function capture(vt: VtTerminal, scrollback: number, escapes: boolean): string {
+	const { total, length } = vt.scrollbar();
+	const first = Math.max(0, total - length - Math.min(scrollback, MAX_CAPTURE_SCROLLBACK));
+	const lines: string[] = [];
+	// The viewport shows `length` rows at a time: page from `first` to the bottom, then scroll back down.
+	vt.scroll("bottom");
+	vt.scroll(first - (total - length));
+	let next = first;
+	for (;;) {
+		const shown = vt.scrollbar();
+		const rows = vt.frame(true).changed.map((row) => (escapes ? styledRow(row.cells) : plainRow(row.cells)));
+		lines.push(...rows.slice(next - shown.offset));
+		next = shown.offset + rows.length;
+		if (shown.offset + shown.length >= shown.total) break;
+		vt.scroll(shown.length);
+	}
+	vt.scroll("bottom");
+	while (lines.length > 0 && plainText(lines.at(-1) as string) === "") lines.pop();
+	return lines.join("\n");
+}
+
+function plainRow(cells: readonly Cell[]): string {
+	return cells
+		.filter((cell) => cell.width !== 0)
+		.map((cell) => cell.text || " ")
+		.join("")
+		.replace(/\s+$/, "");
+}
+
+/** SGR parameters of cell flags, in `Cell.flags` bit order. */
+const SGR_FLAGS: [number, number][] = [
+	[BOLD, 1],
+	[FAINT, 2],
+	[ITALIC, 3],
+	[UNDERLINE, 4],
+	[BLINK, 5],
+	[INVERSE, 7],
+	[INVISIBLE, 8],
+	[STRIKETHROUGH, 9],
+];
+
+/** A row with its colors and attributes as SGR sequences, reset at its end. */
+function styledRow(cells: readonly Cell[]): string {
+	const color = (base: number, value: number) =>
+		value < 0 ? [] : [base, 2, (value >> 16) & 0xff, (value >> 8) & 0xff, value & 0xff];
+	let row = "";
+	let style = "0";
+	for (const cell of cells) {
+		if (cell.width === 0) continue;
+		const params = [
+			0,
+			...SGR_FLAGS.filter(([flag]) => cell.flags & flag).map(([, sgr]) => sgr),
+			...color(38, cell.fg),
+			...color(48, cell.bg),
+		].join(";");
+		if (params !== style) {
+			row += `\x1b[${params}m`;
+			style = params;
+		}
+		row += cell.text || " ";
+	}
+	// Unstyled trailing blanks carry nothing.
+	const trimmed = style === "0" ? row.replace(/ +$/, "") : row;
+	return `${trimmed}${style === "0" ? "" : "\x1b[0m"}`;
+}
+
 const server = createServer((socket) => {
-	const client: Client = { socket, theme: DEFAULT_THEME, sent: 0, acked: 0, snapshotEnd: 0, behind: false };
+	const client: Client = {
+		socket,
+		theme: DEFAULT_THEME,
+		sent: 0,
+		acked: 0,
+		snapshotEnd: 0,
+		behind: false,
+		calls: new Map(),
+	};
 	clients.add(client);
 	send(client, {
 		type: "hello",
@@ -405,6 +783,7 @@ const server = createServer((socket) => {
 	socket.on("close", () => {
 		clients.delete(client);
 		client.attached?.clients.delete(client);
+		for (const cancel of client.calls.values()) cancel();
 		idleSince = Date.now();
 	});
 });

apps/p2/plugins/terminals/plain.ts

@@ -0,0 +1,148 @@
+// Terminal output for the agent (terminal host): a bounded ring of each terminal's raw output at absolute offsets, and
+// output turned into plain text the way a terminal would show it line by line, without escape sequences.
+
+/** The last `capacity` bytes of a terminal's output, addressed by absolute offsets since it started. */
+export class OutputRing {
+	readonly #buffer: Uint8Array;
+	/** Bytes written in total: the absolute offset after the newest byte. */
+	#end = 0;
+
+	constructor(capacity: number) {
+		this.#buffer = new Uint8Array(capacity);
+	}
+
+	/** The offset after the newest byte. */
+	get end(): number {
+		return this.#end;
+	}
+
+	/** The offset of the oldest byte still kept. */
+	get start(): number {
+		return Math.max(0, this.#end - this.#buffer.length);
+	}
+
+	append(bytes: Uint8Array): void {
+		const capacity = this.#buffer.length;
+		// Only the newest `capacity` bytes of a large chunk can survive.
+		const kept = bytes.length > capacity ? bytes.subarray(bytes.length - capacity) : bytes;
+		const lost = bytes.length - kept.length;
+		const at = (this.#end + lost) % capacity;
+		const first = Math.min(kept.length, capacity - at);
+		this.#buffer.set(kept.subarray(0, first), at);
+		if (first < kept.length) this.#buffer.set(kept.subarray(first), 0);
+		this.#end += bytes.length;
+	}
+
+	/** The bytes from `from` to the end; `skipped` counts those after `from` the ring no longer has. */
+	since(from: number): { bytes: Uint8Array; skipped: number } {
+		const begin = Math.min(Math.max(from, this.start), this.#end);
+		const skipped = Math.max(0, this.start - from);
+		const length = this.#end - begin;
+		const capacity = this.#buffer.length;
+		const at = begin % capacity;
+		const bytes = new Uint8Array(length);
+		const first = Math.min(length, capacity - at);
+		bytes.set(this.#buffer.subarray(at, at + first));
+		if (first < length) bytes.set(this.#buffer.subarray(0, length - first), first);
+		return { bytes, skipped };
+	}
+}
+
+const ESC = "\x1b";
+const BEL = "\x07";
+
+/**
+ * Terminal output as plain text: escape sequences removed (CSI, OSC, DCS, SOS, PM, APC, two- and three-byte escapes,
+ * C1 controls), and each line as it would end up on screen: a carriage return starts the line over (so a progress bar
+ * keeps its final state), a backspace steps back, erase-in-line clears. Tabs and line feeds stay; other control
+ * characters go. Trailing spaces of finished lines are trimmed.
+ */
+export function plainText(input: string): string {
+	const lines: string[] = [];
+	let line: string[] = [];
+	let column = 0;
+	const put = (char: string) => {
+		line[column] = char;
+		column++;
+	};
+	const endLine = () => {
+		lines.push(line.join("").replace(/\s+$/, ""));
+		line = [];
+		column = 0;
+	};
+	const chars = [...input];
+	for (let index = 0; index < chars.length; index++) {
+		const char = chars[index] as string;
+		const code = char.codePointAt(0) as number;
+		if (char === ESC) {
+			index = skipEscape(chars, index, (final, params) => {
+				// Erase in line: 0 (or none) to the end, 1 to the start, 2 the whole line.
+				if (final !== "K") return;
+				if (params === "" || params === "0") line.length = column;
+				else if (params === "1") for (let at = 0; at < column && at < line.length; at++) line[at] = " ";
+				else if (params === "2") line = new Array<string>(column).fill(" ");
+			});
+			continue;
+		}
+		if (char === "\n") endLine();
+		else if (char === "\r") column = 0;
+		else if (char === "\b") column = Math.max(0, column - 1);
+		else if (char === "\t") put(char);
+		else if (code < 0x20 || code === 0x7f) continue;
+		else if (code >= 0x80 && code < 0xa0) {
+			// C1 controls: a CSI, OSC or string introducer starts a sequence like its escape form; others carry nothing.
+			if (code === 0x9b || code === 0x9d || code === 0x90 || code === 0x98 || code === 0x9e || code === 0x9f) {
+				chars[index] = String.fromCharCode(0x40 + (code - 0x80));
+				index = skipEscape(chars, index - 1, () => {}, true);
+			}
+		} else put(char);
+	}
+	// The last line is where the cursor is (a prompt): its trailing spaces stay.
+	lines.push(line.join(""));
+	return lines.join("\n");
+}
+
+/**
+ * Skips the escape sequence starting at `chars[index]` (ESC) and returns the index of its last character; `onCsi`
+ * hears each CSI's final character and parameters. `introduced`: the ESC is implied (a C1 control at `index + 1`).
+ */
+function skipEscape(
+	chars: string[],
+	index: number,
+	onCsi: (final: string, params: string) => void,
+	introduced = false,
+): number {
+	const kind = chars[index + 1];
+	if (kind === undefined) return index;
+	if (kind === "[") {
+		let at = index + 2;
+		let params = "";
+		for (; at < chars.length; at++) {
+			const code = (chars[at] as string).codePointAt(0) as number;
+			if (code >= 0x40 && code <= 0x7e) {
+				onCsi(chars[at] as string, params);
+				return at;
+			}
+			params += chars[at];
+		}
+		return chars.length - 1;
+	}
+	if (kind === "]" || kind === "P" || kind === "X" || kind === "^" || kind === "_") {
+		// A string ends with ST (ESC \ or 0x9c); OSC also with BEL.
+		for (let at = index + 2; at < chars.length; at++) {
+			const char = chars[at];
+			if (char === "\x9c" || (kind === "]" && char === BEL)) return at;
+			if (char === ESC && chars[at + 1] === "\\") return at + 1;
+		}
+		return chars.length - 1;
+	}
+	if (introduced) return index + 1;
+	// ESC with intermediates (charset selection `ESC ( B`) then a final character; or ESC and one character.
+	let at = index + 1;
+	while (at < chars.length) {
+		const code = (chars[at] as string).codePointAt(0) as number;
+		if (code < 0x20 || code > 0x2f) break;
+		at++;
+	}
+	return Math.min(at, chars.length - 1);
+}

apps/p2/plugins/terminals/protocol.ts

@@ -4,6 +4,12 @@ import type { Socket } from "node:net";
 export type TerminalInfo = {
 	id: string;
 	sessionId: string;
+	/** Its number in the session, stable while it lives: the lowest free one from 1 (the agent's `-t 2`). */
+	index: number;
+	/** A name given when it was opened (`new-window -n`). */
+	name?: string;
+	/** The agent opened it (the terminal tool); the owner's own have no mark. */
+	agent?: true;
 	cwd: string;
 	env: string | null;
 	title: string;
@@ -21,7 +27,84 @@ export type TerminalRequest =
 	| { type: "resize"; cols: number; rows: number }
 	| { type: "ack"; bytes: number }
 	| { type: "close"; force?: boolean }
-	| { type: "shutdown" };
+	| { type: "shutdown" }
+	/** A request with a reply (`reply` with the same id): the agent's terminal tool and the server's own requests. */
+	| { type: "call"; id: number; call: HostCall }
+	/** Ends a call still waiting (a wait for output): it replies with an error. */
+	| { type: "cancel"; id: number };
+
+/** How long a read waits for output, and for what. A timeout is no error: the read returns what is there. */
+export type TerminalWait = {
+	until: "quiet" | "exit" | "match";
+	/** `match`: a regular expression (JavaScript, multiline) tested against the plain text read. */
+	pattern?: string;
+	ignoreCase?: boolean;
+	/** `quiet`: how long without output counts as quiet. */
+	quietMs?: number;
+	timeoutMs: number;
+};
+
+/**
+ * The host's calls. Each names its session: a terminal of another session is unknown to it. `key` makes a call that
+ * changes something idempotent: a call with a key the host has seen returns the first call's result, done once.
+ */
+export type HostCall =
+	| { op: "list"; sessionId: string }
+	| {
+			op: "open";
+			sessionId: string;
+			key: string;
+			cwd: string;
+			env: string | null;
+			name?: string;
+			/** Typed into the new shell, then Enter. */
+			command?: string;
+			cols: number;
+			rows: number;
+	  }
+	/** tmux `send-keys`: key names (`Enter`, `C-c`) as keys, other arguments as text; all text when `literal`. */
+	| { op: "send"; sessionId: string; key: string; target: string; keys: string[]; literal: boolean }
+	/** The terminal's read cursor: where the next read without `from` starts. */
+	| { op: "cursor"; sessionId: string; target: string }
+	/** Output since `from` (default: the read cursor) as plain text, after an optional wait; moves the read cursor. */
+	| { op: "read"; sessionId: string; target: string; from?: number; wait?: TerminalWait }
+	/** tmux `capture-pane`: the screen as plain text, with up to `scrollback` lines before it; styled with `escapes`. */
+	| { op: "capture"; sessionId: string; target: string; scrollback: number; escapes?: boolean }
+	| { op: "close"; sessionId: string; key: string; target: string }
+	/** Closes every terminal of the session (archived). */
+	| { op: "closeSession"; sessionId: string };
+
+/** What a read returns. */
+export type TerminalRead = {
+	terminal: TerminalInfo;
+	/** The output read: `from` to `to`, absolute byte offsets of the terminal's output. */
+	from: number;
+	to: number;
+	/** Plain text: escape sequences removed, carriage-return overwrites applied. */
+	text: string;
+	/** Bytes after `from` the host no longer kept (its ring holds the last MiB). */
+	skipped: number;
+	/** Characters of `text` left out at its start (it keeps the last `READ_TEXT_LIMIT`). */
+	truncated: number;
+	/** A full-screen program (alternate screen) runs: its screen as plain text, which says more than its output. */
+	screen?: string;
+	/** Why a wait ended. */
+	waited?: "quiet" | "exit" | "match" | "timeout" | "closed";
+	/** The job in front of the shell (an editor, a build), when one runs; absent at the prompt or once exited. */
+	foreground?: string;
+};
+
+export type HostResult =
+	| { kind: "list"; windows: { terminal: TerminalInfo; foreground?: string }[] }
+	| { kind: "terminal"; terminal: TerminalInfo }
+	| { kind: "cursor"; terminal: TerminalInfo; offset: number }
+	| ({ kind: "read" } & TerminalRead)
+	| { kind: "capture"; terminal: TerminalInfo; text: string }
+	| { kind: "closed"; terminalId: string }
+	| { kind: "done" };
+
+/** Plain text a read returns at most: its last characters. */
+export const READ_TEXT_LIMIT = 256 * 1024;
 export type TerminalEvent =
 	| { type: "list"; terminals: TerminalInfo[] }
 	| { type: "attached"; terminal: TerminalInfo }
@@ -30,7 +113,11 @@ export type TerminalEvent =
 	| { type: "busy"; terminalId: string; program?: string }
 	| { type: "error"; message: string }
 	| { type: "hello"; version: string; running: number; pid: number }
-	| { type: "outdated"; running?: number };
+	| { type: "outdated"; running?: number }
+	/** The session's terminals changed (opened, closed, exited, renamed): sent to every connection that listed it. */
+	| { type: "terminals"; terminals: TerminalInfo[] }
+	| { type: "reply"; id: number; result: HostResult }
+	| { type: "reply"; id: number; error: string };
 /**
  * Output a viewer may have unacknowledged. It bounds how long a keystroke's echo can wait behind earlier output: at
  * 5 Mbit/s, 256 KiB is 0.4 s. Viewers acknowledge every 64 KiB (Terminal.tsx), so four batches are in flight.

apps/p2/plugins/terminals/server.ts

@@ -1,13 +1,44 @@
-import { defineFacet, forbidden, isOwner, nativeOf, serveNative } from "@p2/app/v1";
-import { Server, serverOf, Upgrades } from "@p2/app/v1/server";
-import { Catalog } from "../sessions/contract.ts";
+import {
+	type Ctx,
+	contribute,
+	defineFacet,
+	forbidden,
+	isOwner,
+	isWorker,
+	nativeOf,
+	type Ref,
+	serveNative,
+	signalOf,
+	until,
+	type WorkerPrincipal,
+} from "@p2/app/v1";
+import { Hosts, Server, serverOf, Upgrades, type WorkerState } from "@p2/app/v1/server";
+import { Catalog, SessionArchived } from "../sessions/contract.ts";
 import { ChildStore, conversationLocation } from "../subagents/contract.ts";
-import { TerminalServiceObject, TerminalServices, Terminals } from "./contract.ts";
-import type { TerminalRequest } from "./protocol.ts";
+import { AgentTerminals, TerminalServiceObject, TerminalServices, Terminals } from "./contract.ts";
+import type { HostCall, HostResult, TerminalRequest } from "./protocol.ts";
 import { TerminalService } from "./service.ts";
 
+/** The size of a terminal the agent opens; the owner's panel resizes it to fit once it shows it. */
+const AGENT_COLS = 120;
+const AGENT_ROWS = 30;
+
 export default defineFacet((host) => {
 	serveNative(host, TerminalServices);
+	// Archiving a session ends its programs, whether or not anyone holds its terminals now (the service connects to
+	// the host only when it runs); archiving goes on when the host cannot be reached.
+	contribute(host, "p2.terminals-archived", (self) => {
+		self.get(TerminalServiceObject, {});
+		const services = nativeOf(self, TerminalServices, "The terminal service");
+		self.contribute(SessionArchived, {
+			terminals: (session) =>
+				services()
+					.service.closeSession(session)
+					.catch((error: unknown) =>
+						console.error(`p2 terminals: closing session ${session}'s terminals:`, error),
+					),
+		});
+	});
 	host.serve(TerminalServiceObject, (_params, self) => {
 		const server = self.get(Server, {});
 		const serviceResource = self.open(
@@ -75,4 +106,104 @@ export default defineFacet((host) => {
 			},
 		};
 	});
+
+	host.serve(AgentTerminals, (_params, principal, self) => {
+		if (!isWorker(principal)) throw forbidden("agent terminals are for session workers");
+		self.get(TerminalServiceObject, {});
+		const services = nativeOf(self, TerminalServices, "The terminal service");
+		const catalog = self.get(Catalog, {});
+		const children = self.get(ChildStore, {});
+		const current = currentWorker(self.get(Hosts, {}), principal);
+		const sessionId = principal.session;
+		/** The session's live, current worker only, never in an archived session. */
+		const admitted = async (ctx: Ctx) => {
+			await current(ctx);
+			const session = await catalog.get({ id: sessionId }, ctx);
+			if (session.archivedAt !== null) throw forbidden("Session is archived");
+			return session;
+		};
+		const call = async <K extends HostResult["kind"]>(
+			kind: K,
+			request: HostCall,
+			ctx: Ctx,
+		): Promise<Omit<Extract<HostResult, { kind: K }>, "kind">> => {
+			await admitted(ctx);
+			const result = await services().service.call(request, signalOf(ctx));
+			if (result.kind !== kind) throw new Error(`The terminal host answered ${result.kind} to ${request.op}`);
+			const { kind: _kind, ...rest } = result as Extract<HostResult, { kind: K }>;
+			return rest;
+		};
+		return {
+			variant: "worker",
+			members: {
+				list: (_args, ctx) => call("list", { op: "list", sessionId }, ctx),
+				open: async ({ key, location, name, command }, ctx) => {
+					const session = await admitted(ctx);
+					// Only where the session's conversations work: the session's environment or a subagent's own.
+					const own = children.children.value[sessionId] ?? [];
+					const envs = new Set([
+						session.env ?? null,
+						...own.map((child) => conversationLocation(session, child)?.env),
+					]);
+					if (!envs.has(location.env))
+						throw forbidden(
+							`This session does not work on ${location.env ? `"${location.env}"` : "this device"}`,
+						);
+					const opened = await call(
+						"terminal",
+						{
+							op: "open",
+							sessionId,
+							key,
+							cwd: location.cwd,
+							env: location.env,
+							...(name ? { name } : {}),
+							...(command ? { command } : {}),
+							cols: AGENT_COLS,
+							rows: AGENT_ROWS,
+						},
+						ctx,
+					);
+					return opened.terminal;
+				},
+				send: async ({ key, target, keys, literal }, ctx) =>
+					(await call("terminal", { op: "send", sessionId, key, target, keys, literal }, ctx)).terminal,
+				cursor: ({ target }, ctx) => call("cursor", { op: "cursor", sessionId, target }, ctx),
+				read: ({ target, from, wait }, ctx) =>
+					call(
+						"read",
+						{
+							op: "read",
+							sessionId,
+							target,
+							...(from === undefined ? {} : { from }),
+							...(wait === undefined ? {} : { wait }),
+						},
+						ctx,
+					),
+				capture: ({ target, scrollback, escapes }, ctx) =>
+					call("capture", { op: "capture", sessionId, target, scrollback, ...(escapes ? { escapes } : {}) }, ctx),
+				close: ({ key, target }, ctx) => call("closed", { op: "close", sessionId, key, target }, ctx),
+			},
+		};
+	});
 });
+
+/**
+ * Refuses all but the session's running current generation, as MCP's worker view does. A worker may ask before the
+ * server published its start: the check waits for that (the wait ends with the caller's call).
+ */
+function currentWorker(
+	hosts: Ref<typeof Hosts>,
+	{ session, generation }: WorkerPrincipal,
+): (ctx: Ctx) => Promise<void> {
+	const known = (state: WorkerState | undefined) =>
+		state !== undefined && (state.generation > generation || (state.generation === generation && state.running));
+	return async (ctx) => {
+		const published = hosts.workers.value[session];
+		const state = known(published)
+			? published
+			: (await until(hosts.workers, (workers) => known(workers[session]), ctx))[session];
+		if (!state?.running || state.generation !== generation) throw forbidden("stale worker");
+	};
+}

apps/p2/plugins/terminals/service.ts

@@ -3,8 +3,20 @@ import type { IncomingMessage } from "node:http";
 import type { Duplex } from "node:stream";
 import { forbidden } from "@p2/app/v1";
 import { WebSocket, WebSocketServer } from "ws";
-import { attachTerminalHost, type TerminalAttachment, terminalHostVersion } from "./detached.ts";
-import { MAX_BACKLOG, type TerminalEvent, type TerminalInfo, type TerminalRequest } from "./protocol.ts";
+import {
+	attachTerminalHost,
+	connectRunningTerminalHost,
+	type TerminalAttachment,
+	terminalHostVersion,
+} from "./detached.ts";
+import {
+	type HostCall,
+	type HostResult,
+	MAX_BACKLOG,
+	type TerminalEvent,
+	type TerminalInfo,
+	type TerminalRequest,
+} from "./protocol.ts";
 
 /**
  * How often a viewer's socket gets `{"type":"ping"}`. An idle terminal sends nothing, so without it a dead connection
@@ -170,6 +182,99 @@ export class TerminalView {
 	}
 }
 
+/** What a call to a terminal host running older code fails with: it does not know the calls. */
+export const OUTDATED_HOST =
+	"The terminal host runs older code: restart it from the terminal panel (this ends its terminals)";
+
+/**
+ * The server's connection for calls with replies (the agent's terminal tool, archiving): one attachment, made when
+ * the first call needs it and again after the host went away. Cancelling a call tells the host, which ends its wait.
+ */
+export class HostCalls {
+	readonly #directory: string;
+	readonly #app: string;
+	readonly #version: string;
+	#attachment: TerminalAttachment | undefined;
+	/** The attached host's version, from its hello. */
+	#hostVersion: string | undefined;
+	#next = 1;
+	readonly #pending = new Map<number, { resolve: (result: HostResult) => void; reject: (error: Error) => void }>();
+	#closed = false;
+
+	constructor(directory: string, app: string, version: string) {
+		this.#directory = directory;
+		this.#app = app;
+		this.#version = version;
+	}
+
+	async call(call: HostCall, signal?: AbortSignal): Promise<HostResult> {
+		signal?.throwIfAborted();
+		if (this.#closed) throw new Error("The terminal service is closed");
+		const attachment = this.#attach();
+		await attachment.ready;
+		if (this.#hostVersion !== this.#version) throw new Error(OUTDATED_HOST);
+		const id = this.#next++;
+		const reply = Promise.withResolvers<HostResult>();
+		this.#pending.set(id, reply);
+		const abort = () => {
+			if (!this.#pending.delete(id)) return;
+			try {
+				attachment.write({ type: "cancel", id });
+			} catch {}
+			reply.reject(signal?.reason instanceof Error ? signal.reason : new Error("Cancelled"));
+		};
+		signal?.addEventListener("abort", abort, { once: true });
+		try {
+			attachment.write({ type: "call", id, call });
+			return await reply.promise;
+		} finally {
+			this.#pending.delete(id);
+			signal?.removeEventListener("abort", abort);
+		}
+	}
+
+	#attach(): TerminalAttachment {
+		if (this.#attachment) return this.#attachment;
+		const attachment = attachTerminalHost(
+			this.#directory,
+			this.#app,
+			(frame) => this.#frame(frame),
+			() => {
+				if (this.#attachment !== attachment) return;
+				this.#attachment = undefined;
+				this.#fail(new Error("Terminal host disconnected"));
+			},
+		);
+		this.#attachment = attachment;
+		this.#hostVersion = undefined;
+		return attachment;
+	}
+
+	#frame(frame: unknown): void {
+		if (frame instanceof Uint8Array) return;
+		const event = frame as TerminalEvent;
+		if (event.type === "hello") this.#hostVersion = event.version;
+		if (event.type !== "reply") return;
+		const pending = this.#pending.get(event.id);
+		this.#pending.delete(event.id);
+		if ("error" in event) pending?.reject(new Error(event.error));
+		else pending?.resolve(event.result);
+	}
+
+	#fail(error: Error): void {
+		for (const pending of this.#pending.values()) pending.reject(error);
+		this.#pending.clear();
+	}
+
+	close(): void {
+		this.#closed = true;
+		const attachment = this.#attachment;
+		this.#attachment = undefined;
+		attachment?.detach();
+		this.#fail(new Error("The terminal service is closed"));
+	}
+}
+
 export class TerminalService {
 	readonly directory: string;
 	/** The p2 app directory (`Server.info.app`): where the terminal host's code is. */
@@ -177,12 +282,27 @@ export class TerminalService {
 	readonly #sockets = new WebSocketServer({ noServer: true, maxPayload: 1024 * 1024 });
 	readonly #views = new Set<TerminalView>();
 	readonly #tokens = new Map<string, { view: TerminalView; expires: number }>();
+	#calls: Promise<HostCalls> | undefined;
 
 	constructor(directory: string, app: string) {
 		this.directory = directory;
 		this.app = app;
 	}
 
+	/** One call to the terminal host (starting it when none runs); `signal` cancels it. */
+	async call(call: HostCall, signal?: AbortSignal): Promise<HostResult> {
+		this.#calls ??= terminalHostVersion(this.app).then((version) => new HostCalls(this.directory, this.app, version));
+		return (await this.#calls).call(call, signal);
+	}
+
+	/** Closes the session's terminals; without a running host there are none, and none is started. */
+	async closeSession(sessionId: string): Promise<void> {
+		const probe = await connectRunningTerminalHost(this.directory);
+		if (!probe) return;
+		probe.destroy();
+		await this.call({ op: "closeSession", sessionId });
+	}
+
 	async open(session: string): Promise<TerminalView> {
 		const view = new TerminalView(this.directory, this.app, session, await terminalHostVersion(this.app));
 		this.#views.add(view);
@@ -222,6 +342,7 @@ export class TerminalService {
 	}
 
 	async close(): Promise<void> {
+		void this.#calls?.then((calls) => calls.close());
 		await Promise.all([...this.#views].map((view) => this.release(view)));
 		await new Promise<void>((resolve) => this.#sockets.close(() => resolve()));
 	}

apps/p2/plugins/terminals/tool.ts

@@ -0,0 +1,316 @@
+// The agent's `terminal` tool (worker): tmux commands as argv on the session's terminals, the ones the owner sees in
+// the terminal panel. Code mode scripts call it (`tools.terminal({ command, args })`); without code mode `tool_search`
+// loads it, as MCP tools. Replay-safe: the host does a call with the tool call's key once, and reads start where the
+// first attempt's did (memoized), so a call a restart interrupted runs again without typing or opening twice.
+
+import {
+	defineExtension,
+	defineTool,
+	type Extension,
+	type ToolExecutionApi,
+	type ToolExecutionResult,
+} from "@earendil-works/pi-durable";
+import { arr, bool, int, obj, opt, type Static, str, union } from "@earendil-works/pi-schema";
+import { type Context, isRooted, joinPath, type Location, literal } from "@p2/app/v1";
+import {
+	COMMANDS,
+	CommandError,
+	parseCommand,
+	type TerminalCommand,
+	type WaitFor,
+	type WaitOptions,
+} from "./commands.ts";
+import type { TerminalInfoSchema, TerminalReadSchema, TerminalWaitSchema } from "./contract.ts";
+
+type Info = Static<typeof TerminalInfoSchema>;
+type Read = Static<typeof TerminalReadSchema>;
+type Wait = Static<typeof TerminalWaitSchema>;
+
+/** The server's terminal calls for this session (`AgentTerminals`), each cancelled with `context`. */
+export type TerminalBackend = {
+	list(context: Context): Promise<{ terminal: Info; foreground?: string }[]>;
+	open(args: { key: string; location: Location; name?: string; command?: string }, context: Context): Promise<Info>;
+	send(args: { key: string; target: string; keys: string[]; literal: boolean }, context: Context): Promise<Info>;
+	cursor(target: string, context: Context): Promise<{ terminal: Info; offset: number }>;
+	read(args: { target: string; from?: number; wait?: Wait }, context: Context): Promise<Read>;
+	capture(
+		args: { target: string; scrollback: number; escapes: boolean },
+		context: Context,
+	): Promise<{ terminal: Info; text: string }>;
+	close(args: { key: string; target: string }, context: Context): Promise<{ terminalId: string }>;
+};
+
+/**
+ * How the model reaches the tool, as MCP tools: from code mode scripts (`searchTools()` finds it), or, without code
+ * mode, by loading it with `tool_search`.
+ */
+export type TerminalReach = "codemode" | "tool_search";
+
+export const TERMINAL_EXTENSION = "p2-terminals";
+
+/** Characters of output a call returns at most: the last ones. */
+const OUTPUT_CHARS = 16 * 1024;
+/** Timeouts of waits by default, in seconds: send-keys returns sooner, its program usually answers at once. */
+const DEFAULT_TIMEOUT_S = 10;
+const SEND_TIMEOUT_S = 5;
+
+const DESCRIPTION = `Use the session's terminals, the ones the user sees in p2's terminal panel, like tmux. Only when the user asks for a terminal, or to drive an interactive program in one; never instead of bash.
+Call tools.terminal({ command, args }), args being tmux's argv after the command name (no quoting). Windows are numbered per session (1, 2, ...); -t takes a number or a name.
+- new-window [-n name] [-c dir] [-w W] [-T s] [command ...]: a shell at your cwd (-c: relative to it); command is typed into it, then Enter.
+- list-windows
+- send-keys -t T [-l] [-w W] [-T s] key ...: tmux key names (Enter, C-c, Up, Escape, Tab, BSpace, M-x, F1) are keys, other args text; -l: all text.
+- read -t T [-w W] [-T s]: output since your last read.
+- capture-pane -t T [-S -N] [-e]: the screen, plus N scrollback lines; -e keeps colors.
+- kill-window -t T
+new-window, send-keys and read return the output since your last read of that window, after waiting: W is quiet (0.5 s without output; the default of new-window and send-keys), exit, /regex/ or none (read's default); -T is the timeout in seconds (default 10, send-keys 5, max 120). A timeout is no error.
+Result: { window, output, running, foreground?, exitCode?, screen?, skipped?, timedOut? }; output is plain text, or a full-screen program's screen (screen: true); list-windows: { windows }.
+const { window } = await tools.terminal({ command: "new-window", args: ["-n", "dev", "npm run dev"] });
+await tools.terminal({ command: "send-keys", args: ["-t", String(window), "C-c"] });`;
+
+const WindowSchema = obj({
+	window: int(),
+	name: opt(str()),
+	title: opt(str()),
+	location: str(),
+	opener: union([literal("user"), literal("agent")]),
+	running: bool(),
+	foreground: opt(str()),
+	exitCode: opt(int()),
+	error: opt(str()),
+});
+/** What scripts get: one shape for every command, its fields as they apply. */
+const ResultSchema = obj({
+	window: opt(int()),
+	output: opt(str()),
+	running: opt(bool()),
+	foreground: opt(str()),
+	exitCode: opt(int()),
+	error: opt(str()),
+	screen: opt(bool()),
+	skipped: opt(int()),
+	timedOut: opt(bool()),
+	killed: opt(bool()),
+	windows: opt(arr(WindowSchema)),
+});
+type Result = Static<typeof ResultSchema>;
+
+/** The extension offering the `terminal` tool, reached as `reach` says. */
+export function createTerminalExtension(
+	reach: TerminalReach,
+	backend: TerminalBackend,
+	locate: (api: ToolExecutionApi, context: Context) => Promise<Location>,
+): Extension {
+	return defineExtension({
+		name: TERMINAL_EXTENSION,
+		tools: [
+			defineTool({
+				name: "terminal",
+				description: DESCRIPTION,
+				parameters: obj({
+					command: union(COMMANDS.map((name) => literal(name))),
+					args: opt(arr(str(), { description: "tmux's arguments after the command name, one per element." })),
+				}),
+				structuredOutputSchema: ResultSchema,
+				// Not offered to the model; only scripts call it while code mode is on.
+				deferred: true,
+				...(reach === "codemode" ? { callers: ["tools" as const] } : {}),
+				outputLimits: { maxBytes: OUTPUT_CHARS + 1024, retain: "tail" },
+				replay: "safe",
+				execute: async (args, api, context) => {
+					let command: TerminalCommand;
+					try {
+						command = parseCommand(args.command, args.args ?? []);
+					} catch (error) {
+						if (error instanceof CommandError) return failure(error.message);
+						throw error;
+					}
+					try {
+						const { result, terminalId } = await run(command, backend, locate, api, context);
+						return {
+							output: [{ type: "text", text: describe(result) }],
+							structuredOutput: result,
+							// The card's link to the window.
+							details: {
+								command: args.command,
+								args: args.args ?? [],
+								...(result.window === undefined ? {} : { window: result.window }),
+								...(terminalId === undefined ? {} : { terminalId }),
+							},
+						};
+					} catch (error) {
+						if (context.abortSignal?.aborted) throw error;
+						return failure(error instanceof Error ? error.message : String(error));
+					}
+				},
+			}),
+		],
+	});
+}
+
+function failure(text: string): ToolExecutionResult {
+	return { output: [{ type: "text", text }], isError: true };
+}
+
+async function run(
+	command: TerminalCommand,
+	backend: TerminalBackend,
+	locate: (api: ToolExecutionApi, context: Context) => Promise<Location>,
+	api: ToolExecutionApi,
+	context: Context,
+): Promise<{ result: Result; terminalId?: string }> {
+	const key = String(api.taskId);
+	const read = async (args: Parameters<TerminalBackend["read"]>[0]) => {
+		const value = await backend.read(args, context);
+		return { result: readResult(value), terminalId: value.terminal.id };
+	};
+	switch (command.command) {
+		case "new-window": {
+			const here = await locate(api, context);
+			const directory = command.directory;
+			const location =
+				directory === undefined
+					? here
+					: { env: here.env, cwd: isRooted(directory) ? directory : joinPath(here.cwd, directory) };
+			const opened = await backend.open(
+				{
+					key,
+					location,
+					...(command.name ? { name: command.name } : {}),
+					...(command.shellCommand ? { command: command.shellCommand } : {}),
+				},
+				context,
+			);
+			const wait = waitOf(command, { kind: "quiet" }, DEFAULT_TIMEOUT_S);
+			return read({ target: opened.id, from: 0, ...(wait ? { wait } : {}) });
+		}
+		case "list-windows": {
+			const windows = await backend.list(context);
+			return { result: { windows: windows.map(({ terminal, foreground }) => windowOf(terminal, foreground)) } };
+		}
+		case "send-keys": {
+			const start = await readStart(command.target, backend, api, context);
+			await backend.send({ key, target: start.id, keys: command.keys, literal: command.literal }, context);
+			const wait = waitOf(command, { kind: "quiet" }, SEND_TIMEOUT_S);
+			return read({ target: start.id, from: start.offset, ...(wait ? { wait } : {}) });
+		}
+		case "read": {
+			const start = await readStart(command.target, backend, api, context);
+			const wait = waitOf(command, { kind: "none" }, DEFAULT_TIMEOUT_S);
+			return read({ target: start.id, from: start.offset, ...(wait ? { wait } : {}) });
+		}
+		case "capture-pane": {
+			const { target, scrollback, escapes } = command;
+			const { terminal, text } = await backend.capture({ target, scrollback, escapes }, context);
+			return { result: { ...stateOf(terminal), ...cut(text, 0) }, terminalId: terminal.id };
+		}
+		case "kill-window": {
+			// A rerun closes the window the first attempt found, never another that took its number since.
+			const known = await api.memo<{ id: string; window: number }>("window", context);
+			const target = known ?? (await backend.cursor(command.target, context)).terminal;
+			const window = "window" in target ? target.window : target.index;
+			await api.memo("window", { id: target.id, window }, context);
+			await backend.close({ key, target: target.id }, context);
+			return { result: { window, killed: true } };
+		}
+	}
+}
+
+/** Where this call's read starts: the window's read cursor when the call first ran (memoized for a rerun). */
+async function readStart(
+	target: string,
+	backend: TerminalBackend,
+	api: ToolExecutionApi,
+	context: Context,
+): Promise<{ id: string; offset: number }> {
+	const known = await api.memo<{ id: string; offset: number }>("start", context);
+	if (known) return known;
+	const { terminal, offset } = await backend.cursor(target, context);
+	return api.memo("start", { id: terminal.id, offset }, context);
+}
+
+/** The wait a command asks for, or its default; none for `-w none`. */
+function waitOf(options: WaitOptions, fallback: WaitFor, timeout: number): Wait | undefined {
+	const wait = options.wait ?? fallback;
+	const timeoutMs = Math.round((options.timeout ?? timeout) * 1000);
+	switch (wait.kind) {
+		case "none":
+			return undefined;
+		case "quiet":
+		case "exit":
+			return { until: wait.kind, timeoutMs };
+		case "match":
+			return { until: "match", pattern: wait.pattern, ...(wait.ignoreCase ? { ignoreCase: true } : {}), timeoutMs };
+	}
+}
+
+function readResult(read: Read): Result {
+	const screen = read.screen !== undefined;
+	return {
+		...stateOf(read.terminal, read.foreground),
+		...cut(read.screen ?? read.text, screen ? 0 : read.skipped + read.truncated),
+		...(screen ? { screen: true } : {}),
+		...(read.waited === "timeout" ? { timedOut: true } : {}),
+	};
+}
+
+/** The output's last `OUTPUT_CHARS`, with what was left out before it counted in `skipped`. */
+function cut(text: string, skipped: number): { output: string; skipped?: number } {
+	const over = Math.max(0, text.length - OUTPUT_CHARS);
+	const total = skipped + over;
+	return { output: over ? text.slice(over) : text, ...(total ? { skipped: total } : {}) };
+}
+
+function stateOf(terminal: Info, foreground?: string): Result {
+	return {
+		window: terminal.index,
+		running: terminal.exited === undefined,
+		...(foreground ? { foreground } : {}),
+		...(terminal.exited?.code === undefined ? {} : { exitCode: terminal.exited.code }),
+		...(terminal.exited?.error === undefined ? {} : { error: terminal.exited.error }),
+	};
+}
+
+function windowOf(terminal: Info, foreground: string | undefined): Static<typeof WindowSchema> {
+	return {
+		window: terminal.index,
+		...(terminal.name ? { name: terminal.name } : {}),
+		...(terminal.title ? { title: terminal.title } : {}),
+		location: terminal.env ? `${terminal.env}:${terminal.cwd}` : terminal.cwd,
+		opener: terminal.agent ? "agent" : "user",
+		...stateOf(terminal, foreground),
+		running: terminal.exited === undefined,
+	};
+}
+
+/** The result as the transcript and the model (without code mode) read it. */
+function describe(result: Result): string {
+	if (result.windows)
+		return result.windows.length === 0
+			? "No windows."
+			: result.windows
+					.map(
+						(window) =>
+							`${window.window}: ${[
+								window.name ?? window.title ?? "",
+								window.location,
+								`opened by ${window.opener}`,
+								status(window),
+							]
+								.filter(Boolean)
+								.join(" · ")}`,
+					)
+					.join("\n");
+	if (result.killed) return `Killed window ${result.window}.`;
+	const lines = [`window ${result.window} · ${status(result)}${result.screen ? " · full-screen program" : ""}`];
+	if (result.skipped) lines.push(`… ${result.skipped} characters skipped`);
+	if (result.output) lines.push(result.output);
+	if (result.timedOut && result.running)
+		lines.push(`(timed out: still running${result.foreground ? ` (foreground: ${result.foreground})` : ""})`);
+	return lines.join("\n");
+}
+
+function status(result: Pick<Result, "running" | "foreground" | "exitCode" | "error">): string {
+	if (result.running) return result.foreground ? `running ${result.foreground}` : "at the prompt";
+	if (result.error) return `failed: ${result.error}`;
+	return `exited${result.exitCode === undefined ? "" : ` with code ${result.exitCode}`}`;
+}

apps/p2/plugins/terminals/tui.ts

@@ -0,0 +1,47 @@
+// The terminal tool's card in the TUI: the tmux command line, then the output it returned (the last lines; expanded,
+// all of it). The TUI has no terminal panel: the owner watches agent terminals in the web.
+
+import { Text } from "@earendil-works/pi-tui";
+import { contribute, defineFacet } from "@p2/app/v1";
+import { getTextOutput, lineText, str, type ToolRenderers, TuiToolRenderers } from "@p2/app/v1/tui";
+
+/** Collapsed cards show the last lines of the output; expanded ones all. */
+const PREVIEW_LINES = 8;
+
+const terminalRenderers: ToolRenderers = {
+	renderCall(args, theme, context) {
+		const text = context.lastComponent instanceof Text ? context.lastComponent : new Text("", 0, 0);
+		const argv = Array.isArray(args.args) ? args.args.map(str) : [];
+		const line = [str(args.command), ...argv].join(" ") || "...";
+		text.setText(`${theme.fg("toolTitle", theme.bold("terminal"))} ${theme.fg("accent", lineText(line))}`);
+		return text;
+	},
+	renderResult(result, options, theme, context) {
+		const text = context.lastComponent instanceof Text ? context.lastComponent : new Text("", 0, 0);
+		const all = getTextOutput(result)
+			.split("\n")
+			.map((line) => theme.fg(context.isError ? "error" : "toolOutput", lineText(line)));
+		const shown = options.expanded ? all : all.slice(-PREVIEW_LINES);
+		const hidden = all.length - shown.length;
+		text.setText(
+			`\n${[
+				...(hidden > 0
+					? [
+							theme.fg(
+								"muted",
+								`... (${hidden} earlier lines, ${context.ui.keyText("app.tools.expand")} to expand)`,
+							),
+						]
+					: []),
+				...shown,
+			].join("\n")}`,
+		);
+		return text;
+	},
+};
+
+export default defineFacet((host) =>
+	contribute(host, "p2.tui-terminals", (self) => {
+		self.contribute(TuiToolRenderers, { terminal: terminalRenderers });
+	}),
+);

apps/p2/plugins/terminals/worker.ts

@@ -0,0 +1,50 @@
+// The terminals plugin's worker facet: the agent's `terminal` tool, calling the server's `AgentTerminals` as this
+// session's worker, at the calling conversation's location. It follows the code mode switch as MCP tools do: for
+// scripts while code mode is on, loaded with `tool_search` while it is off.
+
+import type { Extension, ToolExecutionApi } from "@earendil-works/pi-durable";
+import { type Context, type Ctx, contribute, defineFacet, expectVariant } from "@p2/app/v1";
+import { ExtensionSwitches } from "@p2/app/v1/node";
+import { Extensions, harnessOf, workerOf } from "@p2/app/v1/worker";
+import { AgentTerminals } from "./contract.ts";
+import { createTerminalExtension, TERMINAL_EXTENSION, type TerminalBackend, type TerminalReach } from "./tool.ts";
+
+/** The code mode plugin's extension: its switch is on while the `codemode` setting is, absent without the plugin. */
+const CODEMODE_EXTENSION = "codemode";
+
+export default defineFacet((host) => {
+	contribute(host, "p2.terminal-tools", (self) => {
+		const harness = harnessOf(self);
+		const worker = workerOf(self);
+		const terminals = async (ctx: Ctx) => expectVariant(await ctx.get(AgentTerminals, {}), "worker");
+		/** One call of the server's object, as this worker, cancelled with `context`. */
+		const via = <T>(
+			call: (server: Awaited<ReturnType<typeof terminals>>, ctx: Ctx) => Promise<T>,
+			context: Context,
+		): Promise<T> => worker().root(async (ctx) => call(await terminals(ctx), ctx), context.abortSignal);
+		const backend: TerminalBackend = {
+			list: async (context) => (await via((server, ctx) => server.list({}, ctx), context)).windows,
+			open: (args, context) => via((server, ctx) => server.open(args, ctx), context),
+			send: (args, context) => via((server, ctx) => server.send(args, ctx), context),
+			cursor: (target, context) => via((server, ctx) => server.cursor({ target }, ctx), context),
+			read: (args, context) => via((server, ctx) => server.read(args, ctx), context),
+			capture: (args, context) => via((server, ctx) => server.capture(args, ctx), context),
+			close: (args, context) => via((server, ctx) => server.close(args, ctx), context),
+		};
+		const locate = (api: ToolExecutionApi, context: Context) =>
+			harness().environments.locate(api.conversationId, context, api);
+		const offer = (reach: TerminalReach) => ({
+			[TERMINAL_EXTENSION]: createTerminalExtension(reach, backend, locate),
+		});
+		let offered: TerminalReach = "codemode";
+		const extension = self.writable<Readonly<Record<string, Extension>>>(offer(offered));
+		self.contribute(Extensions, extension);
+		// Code mode switched off or not loaded: scripts cannot reach the tool, so `tool_search` loads it.
+		self.on(self.registry(ExtensionSwitches), ({ value }) => {
+			const reach: TerminalReach = value.get(CODEMODE_EXTENSION) === true ? "codemode" : "tool_search";
+			if (reach === offered) return;
+			offered = reach;
+			extension.set(offer(reach));
+		});
+	});
+});