diff --git a/apps/p2/docs/p2-notes.md b/apps/p2/docs/p2-notes.md index f37341b1f..4f3c58810 100644 --- a/apps/p2/docs/p2-notes.md +++ b/apps/p2/docs/p2-notes.md @@ -29,7 +29,7 @@ P2_FAKE=1 npm start # http://127.0.0.1:8788, data in ~/.pi/p2 | --- | --- | --- | | `P2_DATA_DIR` | `~/.pi/p2` | data directory: `catalog.sqlite`, `sessions/.sqlite`, `settings.json`, `models.json`, `envs.json`, `mcp.json`, `mcp-auth.json`, `auth.json`, `radius.json`, `forwards.json` | | `PORT` | `8788` | listen port (loopback only), so p2 never collides with pi-mobile on 8787 | -| `P2_FAKE` | unset | `1`: the faux provider (`faux/faux-1`, echoes the prompt; `slow …` streams long; `rename-session ` calls the tool; `subagent`, `subagent-bad`, `subagent-answer`, `subagent-configure`, `subagent-send`, `subagent-archive`, `subagent-status` exercise subagents, `subagent <json>` calls the tool with those arguments, as do `team <json>`, `history_search|read|handoff <json>`, `location_envs|dirs|move <json>`, `tool_search <json>` and `mcp__<server>__<tool> <json>`; `codemode` runs a script calling bash and read, `codemode-restart` one whose slow middle call a restart interrupts, `codemode <script>` runs that script; `artifact <json>` calls the artifact tool with those arguments; `json <text>` answers exactly `<text>`; a one-shot agent whose task is not `json …`/`slow…` answers a small value matching its schema) | +| `P2_FAKE` | unset | `1`: the faux provider (`faux/faux-1`, echoes the prompt; `slow …` streams long; `rename-session <title>` calls the tool; `subagent`, `subagent-bad`, `subagent-answer`, `subagent-configure`, `subagent-send`, `subagent-archive`, `subagent-status` exercise subagents, `subagent <json>` calls the tool with those arguments, as do `team <json>`, `history_search|read|handoff <json>`, `location_envs|dirs|move <json>`, `tool_search <json>`, `terminal <json>` and `mcp__<server>__<tool> <json>`; `codemode` runs a script calling bash and read, `codemode-restart` one whose slow middle call a restart interrupts, `codemode <script>` runs that script; `artifact <json>` calls the artifact tool with those arguments; `json <text>` answers exactly `<text>`; a one-shot agent whose task is not `json …`/`slow…` answers a small value matching its schema) | | `P2_IDLE_MS` | `300000` | idle time before a session worker stops | | `P2_PROVIDER`, `P2_MODEL` | unset | default model for new sessions after settings.json's `defaultModel` (both or neither; Pi Mobile's `PI_MOBILE_PROVIDER`/`PI_MOBILE_MODEL` work too) | | `P2_RADIUS` | unset | `0`: disable the outbound Radius brain connection even when signed in | @@ -877,11 +877,49 @@ translated phone drawer and intercept chat taps. Terminal tab through the terminals plugin's `workspace-terminal` slot. Side and bottom panels carry across a session switch, showing that session's terminals without starting one. Only explicit terminal-button/tab taps request auto-start; attachment updates replace the route, never push or reopen a later destination. - Outdated hosts show Restart (ends all terminals), but idle outdated hosts are not automatically replaced yet. + Outdated hosts show Restart (ends all terminals), but idle outdated hosts are not automatically replaced yet; the + agent's calls to an outdated host fail with that advice (`OUTDATED_HOST`), and archiving cannot close its terminals. + Each terminal has a number in its session (`index`: the lowest free one from 1, kept while it lives), an optional + name and an `agent` mark; tabs show the name, else the title, else "Terminal N", agent terminals with a "Pi" badge. + Every connection that listed a session hears its list change (`terminals`: opened, closed, exited, renamed, by any + viewer or the agent); a panel showing a terminal keeps it, one showing none attaches the newest new one. Archiving a + session closes its terminals (`SessionArchived`, a host-lifetime contribution; no host is started for it). + **Agent terminal tool** (owner decisions): the agent drives the same terminals the owner sees (shared: the owner + watches and types; the agent may read, type into and close any terminal of its session, the owner's too). Guest + turns may use it, like bash. Code mode only, like MCP tools (`deferred`, `callers: ["tools"]`; scripts find it with + `searchTools()`); while code mode is off `tool_search` loads it (installed once by `durableRegistry`, not by each + plugin: two plugins contributing `tool-search` would be a registry conflict). The description says to use it only + when the user asks for a terminal or for driving an interactive program in one, never instead of bash; no prompt + section. Call: `tools.terminal({ command, args })`, tmux's argv after the command name, parsed getopt-style + (`commands.ts`: no string parsing, tmux-like errors such as `send-keys: unknown option -- x` or `usage: …`): + `new-window [-n name] [-c dir] [-w W] [-T s] [command ...]` (a shell at the calling conversation's location, + `environments.locate`; subagents cannot choose code mode or this tool yet; `-c` relative to it; `command` is typed into it, then Enter, and the shell stays), + `list-windows`, `send-keys -t T [-l] [-w W] [-T s] key ...` (tmux key names as keys, encoded by libghostty in the + terminal's modes, other arguments as text), `read -t T [-w W] [-T s]` (output since the agent's last read), + `capture-pane -t T [-S -N] [-e] [-p]` (the screen, N scrollback lines, `-e` with SGR colors), `kill-window -t T`. + Targets: the number (`2`, `:2`) or the name. Waits `W`: `quiet` (0.5 s without output; default of new-window and + send-keys), `exit`, `/regex/` (`/regex/i`) over the new plain text, `none` (read's default); `-T` seconds (default + 10, send-keys 5, max 120); a timeout is no error. Result (`structuredOutput`): `{ window, output, running, + foreground?, exitCode?, error?, screen?, skipped?, timedOut? }`, `{ windows }`, `{ window, killed }`; output is the + last 16 KiB of plain text (escape sequences removed, `\r` overwrites and backspaces applied, erase-in-line honored), + or a full-screen program's screen (`screen: true`). The host keeps each terminal's last MiB of output + (`OutputRing`) and the agent's read cursor, so reads survive server and worker restarts; it answers calls with ids + (`call`/`reply`, `cancel` ends a wait) and does keyed calls (open, send, close) once per key (the tool call's task + id; the last 1024 remembered). The tool is replay-safe: the read start and a killed window's id are memoized + (`api.memo`). Server: `AgentTerminals` (worker variant only) serves the session's running current generation (as + MCP's `currentWorker`), refuses archived sessions, opens only on environments the session works on (its own, its + subagents'), forwards cancellation. The agent may have 8 running terminals per session (the owner's do not count). + Web: the transcript card (code mode off) shows the command line and output with "Open terminal N"; code mode's + nested-call rows show the command line. TUI: a card with the command line and the output's last lines. `terminals-host.test.ts` covers shells, disconnect snapshots, foreground jobs, themes and flood resync; `terminals.test.ts` covers guest/control denial, one-use byte grants, session isolation, reload and server restart; - `ghostty.test.ts` and `terminal-input.test.ts` cover the copied browser primitives. PTY tests skip when no daemon build - exists. + `terminal-agent-host.test.ts` the host's calls (keys done once, cursor, waits, cancel, plain text, the alternate + screen, captures, ring overflow, list pushes, the cap, closing a session, an outdated host); `terminal-tool.test.ts` + the tool end to end (scripts, shared with the owner both ways, errors, guest turns, stale workers and foreign + environments refused, archiving, `tool_search` without code mode); `terminal-tool-unit.test.ts`, + `terminal-commands.test.ts`, `terminal-plain.test.ts` and `terminal-keys.test.ts` the pure parts; + `ghostty.test.ts` and `terminal-input.test.ts` cover the copied browser primitives. PTY tests skip when no daemon + build exists (`npm run build:daemon` in `packages/env` builds a debug one). - **Search** (`plugins/search`): `SearchIndex` (server local, `search.sqlite`: root/subagent user/assistant text in FTS5 and one cursor per conversation; derived, dropped on a schema change). Each worker's `Indexer` (lifetime "host") runs a task that reads the server's cursor (`SearchFeed`, worker principals only, session from the principal), scans each root/subagent @@ -1052,8 +1090,9 @@ translated phone drawer and intercept chat taps. refusal because a newer generation runs ends it), and contributes one Durable extension `mcp` (a `self.writable`, rebuilt only when the tool list changes): deferred tools named by `createMcpToolName` (every tool of a colliding name gets the hash suffix, as in pi), results through - `toLlmContent` and `limitMcpContent`, `structuredOutput` only for tools with an output schema. The worker facet also - installs `tool_search`. Tests: `mcp.test.ts` (manager and views on a bare host with `test/fixtures/mcp-server.mjs` and + `toLlmContent` and `limitMcpContent`, `structuredOutput` only for tools with an output schema. Without code mode, + `tool_search` is installed by the worker's registry (`durableRegistry`: once, while code mode is off and any extension + has a deferred tool the model may call; MCP's and the terminal tool share it). Tests: `mcp.test.ts` (manager and views on a bare host with `test/fixtures/mcp-server.mjs` and an HTTP fixture; projects, where sessions work, and an environment through the debug pi-env daemon, skipped without it: `npm run build:daemon` in `packages/env`, or `PI_ENV_DAEMON`), `mcp-instances.test.ts` (the supervisor with a short backoff), `mcp-leases.test.ts`, `mcp-environments.test.ts`, `mcp-project.test.ts`, diff --git a/apps/p2/plugins/chat/web.tsx b/apps/p2/plugins/chat/web.tsx index 8ce40c502..3173a5964 100644 --- a/apps/p2/plugins/chat/web.tsx +++ b/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> diff --git a/apps/p2/plugins/mcp/worker.ts b/apps/p2/plugins/mcp/worker.ts index d9e841bec..b266d09bf 100644 --- a/apps/p2/plugins/mcp/worker.ts +++ b/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) }; } /** diff --git a/apps/p2/plugins/models/fake.ts b/apps/p2/plugins/models/fake.ts index b7a133855..924397aa3 100644 --- a/apps/p2/plugins/models/fake.ts +++ b/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( diff --git a/apps/p2/plugins/terminals/Terminal.tsx b/apps/p2/plugins/terminals/Terminal.tsx index e4dc1eba2..a8e1ca14e 100644 --- a/apps/p2/plugins/terminals/Terminal.tsx +++ b/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 diff --git a/apps/p2/plugins/terminals/commands.ts b/apps/p2/plugins/terminals/commands.ts new file mode 100644 index 000000000..88e8a5941 --- /dev/null +++ b/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))); +} diff --git a/apps/p2/plugins/terminals/contract.ts b/apps/p2/plugins/terminals/contract.ts index 565ae20c8..572be2bb2 100644 --- a/apps/p2/plugins/terminals/contract.ts +++ b/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() })), + }, + }, + }, +}); diff --git a/apps/p2/plugins/terminals/detached.ts b/apps/p2/plugins/terminals/detached.ts index b644fe6de..1ee2a266f 100644 --- a/apps/p2/plugins/terminals/detached.ts +++ b/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); diff --git a/apps/p2/plugins/terminals/ghostty.ts b/apps/p2/plugins/terminals/ghostty.ts index 4a2f38fba..78cc5ff78 100644 --- a/apps/p2/plugins/terminals/ghostty.ts +++ b/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})`); diff --git a/apps/p2/plugins/terminals/keys.ts b/apps/p2/plugins/terminals/keys.ts new file mode 100644 index 000000000..26281e2a1 --- /dev/null +++ b/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 } : {}), + }; +} diff --git a/apps/p2/plugins/terminals/main.ts b/apps/p2/plugins/terminals/main.ts index 0d37b98b7..5174b32c1 100644 --- a/apps/p2/plugins/terminals/main.ts +++ b/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(); }); }); diff --git a/apps/p2/plugins/terminals/plain.ts b/apps/p2/plugins/terminals/plain.ts new file mode 100644 index 000000000..3ddea662d --- /dev/null +++ b/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); +} diff --git a/apps/p2/plugins/terminals/protocol.ts b/apps/p2/plugins/terminals/protocol.ts index ec4ec8669..78338daf7 100644 --- a/apps/p2/plugins/terminals/protocol.ts +++ b/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. diff --git a/apps/p2/plugins/terminals/server.ts b/apps/p2/plugins/terminals/server.ts index 8e8f6dcfe..a116f5127 100644 --- a/apps/p2/plugins/terminals/server.ts +++ b/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"); + }; +} diff --git a/apps/p2/plugins/terminals/service.ts b/apps/p2/plugins/terminals/service.ts index 17638fe64..df4481110 100644 --- a/apps/p2/plugins/terminals/service.ts +++ b/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())); } diff --git a/apps/p2/plugins/terminals/tool.ts b/apps/p2/plugins/terminals/tool.ts new file mode 100644 index 000000000..a9a2d2583 --- /dev/null +++ b/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}`}`; +} diff --git a/apps/p2/plugins/terminals/tui.ts b/apps/p2/plugins/terminals/tui.ts new file mode 100644 index 000000000..08db97bc5 --- /dev/null +++ b/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 }); + }), +); diff --git a/apps/p2/plugins/terminals/worker.ts b/apps/p2/plugins/terminals/worker.ts new file mode 100644 index 000000000..69bf17a58 --- /dev/null +++ b/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)); + }); + }); +}); diff --git a/apps/p2/src/runtime/durable.ts b/apps/p2/src/runtime/durable.ts index ee83a1c8a..9704ea64c 100644 --- a/apps/p2/src/runtime/durable.ts +++ b/apps/p2/src/runtime/durable.ts @@ -4,6 +4,7 @@ import type { RegistryValue } from "@earendil-works/chord"; import type { Context } from "@earendil-works/chord/context"; import { BACKGROUND_CONTEXT, withAbortSignal } from "@earendil-works/chord/context"; import { createRegistry, type Extension, type RegistryReader, type RegistrySnapshot } from "@earendil-works/pi-durable"; +import { createToolSearchExtension } from "@earendil-works/pi-durable/tool-search"; import { keyedRegistry } from "./chord.ts"; /** @@ -23,9 +24,14 @@ export const Extensions = keyedRegistry<Extension>("p2.durable-extensions"); */ export const ExtensionSwitches = keyedRegistry<boolean | undefined>("p2.extension-switches"); +/** The code mode plugin's extension: its switch is on while the `codemode` setting is, absent without the plugin. */ +const CODEMODE_EXTENSION = "codemode"; + /** * Durable's registry over the worker's `Extensions` and `ExtensionSwitches`: a snapshot per (extensions, switched-off - * set), made once and kept while the composition lives; `subscribe` observes both registries' compositions (its + * set), made once and kept while the composition lives. Without code mode, deferred tools the model may call (MCP + * tools, the terminal tool) are reachable only through `tool_search`, as pi falls back: it is installed then, once, + * whichever plugins offer such tools; `subscribe` observes both registries' compositions (its * listener called inside the Chord batch that composed, so it only wakes the scheduler). `context` ends the * observations. */ @@ -42,7 +48,14 @@ export function durableRegistry( let snapshot = byKey.get(key); if (snapshot === undefined) { const builder = createRegistry(); - for (const [name, extension] of table) if (!disabled.has(name)) builder.install(extension); + const installed = [...table].filter(([name]) => !disabled.has(name)).map(([, extension]) => extension); + for (const extension of installed) builder.install(extension); + const codemode = table.has(CODEMODE_EXTENSION) && !disabled.has(CODEMODE_EXTENSION); + const search = createToolSearchExtension(); + const deferred = installed.some((extension) => + extension.tools?.some((tool) => tool.deferred && (tool.callers?.includes("model") ?? true)), + ); + if (!codemode && deferred && !table.has(search.name)) builder.install(search); snapshot = builder.snapshot(); byKey.set(key, snapshot); } diff --git a/apps/p2/src/web/Chat.tsx b/apps/p2/src/web/Chat.tsx index 6f598e779..361161c31 100644 --- a/apps/p2/src/web/Chat.tsx +++ b/apps/p2/src/web/Chat.tsx @@ -89,6 +89,8 @@ export const OpenPeerConversation = createContext<((sessionId: string) => void) export const WorkingDirectory = createContext<string | undefined>(undefined); /** Opens a cwd-relative file in the workspace panel; absent when the viewer has no workspace access. */ export const OpenWorkspaceFile = createContext<((path: string, line?: number) => void) | undefined>(undefined); +/** Shows one of the session's terminals in the terminal panel (owner only: guests have none). */ +export const OpenTerminal = createContext<((terminalId: string) => void) | undefined>(undefined); /** Download and artifact locators are scoped to the containing session, including child views. */ export const SessionId = createContext<string | undefined>(undefined); /** Starts a reply to an entry; absent while the conversation accepts no replies (chat-mode turns them on). */ @@ -232,8 +234,14 @@ function callSummary(args: NestedCall["args"], cwd: string | undefined): string const key = ["command", "path", "pattern", "url", "query", "name", "label", "prompt"].find( (name) => typeof args[name] === "string", ); + // A command with an argv (the terminal tool's tmux commands): the whole command line. + const argv = key === "command" && Array.isArray(args.args) ? args.args.map(str) : []; const value = - key === undefined ? JSON.stringify(args) : key === "path" ? relativePath(cwd, str(args.path)) : str(args[key]); + key === undefined + ? JSON.stringify(args) + : key === "path" + ? relativePath(cwd, str(args.path)) + : [str(args[key]), ...argv].join(" "); return value === "{}" ? "" : value.replace(/\s+/g, " ").trim(); } @@ -401,6 +409,7 @@ const ToolCard = memo(function ToolCard({ tool }: { tool: ToolView }) { let body: React.ReactNode = null; let extra: React.ReactNode = null; const openWorkspace = useContext(OpenWorkspaceFile); + const openTerminal = useContext(OpenTerminal); // File tools link into the workspace panel when the path stays inside the working directory. const inside = sourcePath && !path.startsWith("/") && path !== "." && !path.split("/").includes(".."); const workspaceLink = (line?: number) => @@ -609,6 +618,28 @@ const ToolCard = memo(function ToolCard({ tool }: { tool: ToolView }) { ); break; } + case "terminal": { + icon = <TerminalIcon />; + title = "Terminal"; + const argv = Array.isArray(tool.args.args) ? tool.args.args.map(str) : []; + summary = [str(tool.args.command), ...argv].join(" "); + body = ( + <> + <pre className="tool-command">{summary}</pre> + {tool.output && ( + <pre className={`term ${tool.status === "error" ? "tool-error" : ""}`}>{tool.output}</pre> + )} + </> + ); + const terminalId = typeof details.terminalId === "string" ? details.terminalId : undefined; + if (openTerminal && terminalId && tool.status === "done" && str(tool.args.command) !== "kill-window") + extra = ( + <button type="button" className="tool-link" onClick={() => openTerminal(terminalId)}> + Open terminal{typeof details.window === "number" ? ` ${details.window}` : ""} → + </button> + ); + break; + } case "send_file": icon = <FileIcon />; title = "File"; diff --git a/apps/p2/src/web/style.css b/apps/p2/src/web/style.css index a3e244c93..dc348a9b7 100644 --- a/apps/p2/src/web/style.css +++ b/apps/p2/src/web/style.css @@ -1852,6 +1852,14 @@ audio.file-media { .terminal-tabs > button.exited { text-decoration: line-through; } +/* Opened by the agent (the terminal tool). */ +.terminal-agent { + margin-right: 0.4em; + padding: 0 0.3em; + border: 1px solid var(--line); + border-radius: 3px; + font-size: 0.85em; +} .terminal-tabs > .icon-button[aria-pressed="true"] { color: var(--text); } diff --git a/apps/p2/test/p2/durable-registry.test.ts b/apps/p2/test/p2/durable-registry.test.ts index 5d70d3ca4..ef9e07e2d 100644 --- a/apps/p2/test/p2/durable-registry.test.ts +++ b/apps/p2/test/p2/durable-registry.test.ts @@ -1,8 +1,9 @@ -// The Extensions registry as Durable's RegistryReader: one complete snapshot per table, swapped at reload. +// The Extensions registry as Durable's RegistryReader: one complete snapshot per table, swapped at reload, with +// `tool_search` installed once while code mode is off and deferred tools need it. import assert from "node:assert/strict"; import { describe, it } from "node:test"; -import { defineExtension, section } from "@earendil-works/pi-durable"; -import { bool } from "@earendil-works/pi-schema"; +import { defineExtension, defineTool, section } from "@earendil-works/pi-durable"; +import { bool, obj } from "@earendil-works/pi-schema"; import { BACKGROUND_CONTEXT, local, method, whenLive } from "@p2/app/v1"; import { defineFacet, type Facet, Session } from "../../src/runtime/chord.ts"; import { durableRegistry, ExtensionSwitches } from "../../src/runtime/durable.ts"; @@ -114,4 +115,45 @@ describe("durable registry bridge", () => { assert.deepEqual(names(), ["switched", "guard"]); assert.ok(published > before); }); + + it("installs tool_search once while code mode is off and deferred tools the model may call exist", async (t) => { + const deferredTool = (name: string, callers?: readonly ("model" | "tools")[]) => + defineFacet((_host, self) => + self.contribute(Extensions, { + [name]: defineExtension({ + name, + tools: [ + defineTool({ + name, + description: name, + parameters: obj({}), + deferred: true, + ...(callers ? { callers } : {}), + execute: async () => ({}), + }), + ], + }), + }), + ); + const codemode = (on: boolean) => + defineFacet((_host, self) => { + self.contribute(Extensions, { codemode: defineExtension({ name: "codemode" }) }); + self.contribute(ExtensionSwitches, { codemode: on }); + }); + const names = async (facets: Facet[]) => { + const fixture = await testHost({ address: Session({ id: "w" }), kind: "worker", facets }); + t.after(() => fixture.close()); + return (await readerOf(fixture)) + .snapshot() + .installed() + .map((extension) => extension.name); + }; + // Two plugins with deferred tools (MCP's, the terminal tool), no code mode: one tool_search. + assert.deepEqual(await names([deferredTool("a"), deferredTool("b")]), ["a", "b", "tool-search"]); + assert.deepEqual(await names([codemode(false), deferredTool("a")]), ["a", "tool-search"]); + // Code mode on, or only script-callable deferred tools, or none: no tool_search. + assert.deepEqual(await names([codemode(true), deferredTool("a")]), ["codemode", "a"]); + assert.deepEqual(await names([deferredTool("a", ["tools"])]), ["a"]); + assert.deepEqual(await names([facet("guard", "v1")]), ["guard"]); + }); }); diff --git a/apps/p2/test/p2/terminal-agent-host.test.ts b/apps/p2/test/p2/terminal-agent-host.test.ts new file mode 100644 index 000000000..72c24e092 --- /dev/null +++ b/apps/p2/test/p2/terminal-agent-host.test.ts @@ -0,0 +1,289 @@ +// The terminal host's calls for the agent's terminal tool, against a real host and pi-env daemon (skipped without a +// daemon build): opens and inputs done once per key, the read cursor and its ring overflow, waits (quiet, match, exit, +// timeout, cancel), plain text, the alternate screen as screen text, captures, session list pushes to the viewers, +// the agent's terminal cap, closing a session, and an outdated host refusing calls. +import assert from "node:assert/strict"; +import { mkdtemp, realpath, rm } from "node:fs/promises"; +import type { Socket } from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { test } from "node:test"; +import { localSystem, terminalDaemon } from "../../plugins/terminals/daemon.ts"; +import { connectTerminalHost } from "../../plugins/terminals/detached.ts"; +import { + encodeFrame, + type HostCall, + type HostResult, + readFrames, + type TerminalEvent, + type TerminalRead, + type TerminalRequest, +} from "../../plugins/terminals/protocol.ts"; +import { HostCalls, OUTDATED_HOST } from "../../plugins/terminals/service.ts"; +import { APP } from "./helpers.ts"; + +const ESC = "\x1b"; + +let unavailable: string | undefined; +try { + const system = localSystem(); + await terminalDaemon(system.platform, system.arch, true); +} catch (error) { + unavailable = error instanceof Error ? error.message : "No pi-env daemon with terminal support"; +} + +/** A host connection that records events and makes calls. */ +async function client(directory: string) { + const socket: Socket = await connectTerminalHost(directory, APP); + const events: TerminalEvent[] = []; + const replies = new Map<number, (event: Extract<TerminalEvent, { type: "reply" }>) => void>(); + let next = 1; + readFrames(socket, (frame) => { + if (frame instanceof Uint8Array) return; + const event = frame as TerminalEvent; + if (event.type === "reply") replies.get(event.id)?.(event); + else events.push(event); + }); + const send = (message: TerminalRequest) => socket.write(encodeFrame(message)); + /** Starts a call: its id, and its result (rejecting with the host's error). */ + const start = (call: HostCall) => { + const id = next++; + const result = new Promise<HostResult>((resolve, reject) => { + replies.set(id, (event) => ("error" in event ? reject(new Error(event.error)) : resolve(event.result))); + }); + send({ type: "call", id, call }); + return { id, result }; + }; + const call = (request: HostCall) => start(request).result; + const read = async (request: Omit<Extract<HostCall, { op: "read" }>, "op">): Promise<TerminalRead> => { + const result = await call({ op: "read", ...request }); + assert.equal(result.kind, "read"); + return result as Extract<HostResult, { kind: "read" }>; + }; + return { socket, events, send, start, call, read }; +} + +test("terminal host: the agent's calls", { timeout: 120_000, skip: unavailable }, async (t) => { + const directory = await mkdtemp(join(tmpdir(), "p2-agent-terminals-")); + const cwd = await realpath(await mkdtemp(join(tmpdir(), "p2-agent-terminal-cwd-"))); + const shell = process.env.SHELL; + // The host and its shells inherit this: a plain shell, without the developer's rc files. + process.env.SHELL = "/bin/sh"; + const viewer = await client(directory); + const agent = await client(directory); + const open = async (sessionId: string, key: string, command?: string, name?: string) => { + const result = await agent.call({ + op: "open", + sessionId, + key, + cwd, + env: null, + cols: 60, + rows: 10, + ...(command ? { command } : {}), + ...(name ? { name } : {}), + }); + assert.equal(result.kind, "terminal"); + return (result as Extract<HostResult, { kind: "terminal" }>).terminal; + }; + const send = (target: string, key: string, keys: string[], literal = false, sessionId = "s1") => + agent.call({ op: "send", sessionId, key, target, keys, literal }); + const prompt = { until: "match", pattern: "[$#] $", timeoutMs: 10_000 } as const; + try { + await t.test("opens once per key, types keys once per key, reads from the cursor", async () => { + viewer.send({ type: "list", sessionId: "s1" }); + const terminal = await open("s1", "open-1", "printf 'hi-%s\\n' there", "dev"); + assert.equal(terminal.index, 1); + assert.equal(terminal.agent, true); + assert.equal(terminal.name, "dev"); + assert.equal((await open("s1", "open-1")).id, terminal.id); + // The viewer that listed the session hears of it. + await poll(() => + viewer.events.some( + (event) => event.type === "terminals" && event.terminals.some((each) => each.id === terminal.id), + ), + ); + const first = await agent.read({ + sessionId: "s1", + target: "dev", + // The program's output, then the next prompt. + wait: { until: "match", pattern: "^hi-there\\n.*[$#] $", timeoutMs: 10_000 }, + }); + assert.equal(first.waited, "match", JSON.stringify(first.text)); + assert.equal(first.from, 0); + assert.match(first.text, /^hi-there$/m); + const cursor = await agent.call({ op: "cursor", sessionId: "s1", target: ":1" }); + assert.equal(cursor.kind, "cursor"); + const from = (cursor as Extract<HostResult, { kind: "cursor" }>).offset; + assert.ok(from > 0); + // The same key twice: typed once. + await send("1", "send-1", ["echo o''ne", "Enter"]); + await send("1", "send-1", ["echo o''ne", "Enter"]); + const typed = await agent.read({ + sessionId: "s1", + target: "1", + wait: { ...prompt, pattern: "one\\n.*[$#] $" }, + }); + assert.equal(typed.from, from); + assert.equal(typed.text.match(/^one$/gm)?.length, 1); + // A read repeated from the same start (a replayed call) gets the same output again. + const again = await agent.read({ sessionId: "s1", target: "1", from }); + assert.ok(again.text.startsWith(typed.text)); + // The cursor moved past it: nothing new. + assert.equal((await agent.read({ sessionId: "s1", target: "1" })).text, ""); + const listed = await agent.call({ op: "list", sessionId: "s1" }); + assert.deepEqual(listed.kind === "list" ? listed.windows.map((window) => window.terminal.id) : [], [ + terminal.id, + ]); + await assert.rejects(send("9", "send-2", ["x"]), /can't find window: 9/); + await assert.rejects(send("1", "send-3", ["x"], false, "s2"), /can't find window: 1/); + }); + + await t.test("waits: match, quiet, timeout with the foreground job, cancel, exit", async () => { + await send("1", "wait-1", ["sleep 0.3; echo la''te", "Enter"]); + const late = await agent.read({ + sessionId: "s1", + target: "1", + wait: { until: "match", pattern: "^late", timeoutMs: 5000 }, + }); + assert.equal(late.waited, "match"); + assert.equal( + ( + await agent.read({ + sessionId: "s1", + target: "1", + wait: { until: "quiet", quietMs: 200, timeoutMs: 5000 }, + }) + ).waited, + "quiet", + ); + await send("1", "wait-2", ["sleep 30", "Enter"]); + const timedOut = await agent.read({ + sessionId: "s1", + target: "1", + wait: { until: "match", pattern: "never", timeoutMs: 500 }, + }); + assert.equal(timedOut.waited, "timeout"); + assert.match(timedOut.foreground ?? "", /sleep/); + const waiting = agent.start({ + op: "read", + sessionId: "s1", + target: "1", + wait: { until: "exit", timeoutMs: 60_000 }, + }); + await new Promise((resolve) => setTimeout(resolve, 100)); + agent.send({ type: "cancel", id: waiting.id }); + await assert.rejects(waiting.result, /cancelled/); + await send("1", "wait-3", ["C-c"]); + await agent.read({ sessionId: "s1", target: "1", wait: prompt }); + await send("1", "wait-4", ["exit 3", "Enter"]); + const exited = await agent.read({ sessionId: "s1", target: "1", wait: { until: "exit", timeoutMs: 10_000 } }); + assert.equal(exited.waited, "exit"); + assert.deepEqual(exited.terminal.exited, { code: 3 }); + assert.equal(exited.foreground, undefined); + await assert.rejects(send("1", "wait-5", ["x"]), /window 1 has exited/); + }); + + await t.test("plain text, the alternate screen, captures and ring overflow", async () => { + const terminal = await open("s1", "open-2"); + // The exited window 1 keeps its number: this one is 2. + assert.equal(terminal.index, 2); + await agent.read({ sessionId: "s1", target: "2", wait: prompt }); + await send("2", "plain-1", ["printf 'a\\033[31mred\\033[0m\\r\\n10%%\\r50%%\\r100%%\\n'", "Enter"]); + const plain = await agent.read({ sessionId: "s1", target: "2", wait: { ...prompt, pattern: "100%\\n" } }); + assert.match(plain.text, /^ared$/m); + assert.match(plain.text, /^100%$/m); + assert.ok(!plain.text.includes("\x1b") && !plain.text.includes("10%\r")); + // A full-screen program: the read returns its screen. + await send("2", "plain-2", [ + "printf '\\033[?1049h\\033[H\\033[2JFULL SCREEN\\n'; sleep 1; printf '\\033[?1049l'", + "Enter", + ]); + const full = await agent.read({ + sessionId: "s1", + target: "2", + wait: { until: "match", pattern: "^FULL SCREEN$", timeoutMs: 5000 }, + }); + assert.equal(full.screen?.split("\n")[0], "FULL SCREEN"); + await agent.read({ sessionId: "s1", target: "2", wait: { ...prompt, timeoutMs: 5000 } }); + assert.equal((await agent.read({ sessionId: "s1", target: "2" })).screen, undefined); + await send("2", "plain-3", ["clear; seq 1 100; printf '\\033[32mgreen\\033[0m\\n'", "Enter"]); + await agent.read({ sessionId: "s1", target: "2", wait: { ...prompt, pattern: "green\\n.*[$#] $" } }); + const capture = async (scrollback: number, escapes = false) => { + const result = await agent.call({ op: "capture", sessionId: "s1", target: "2", scrollback, escapes }); + return result.kind === "capture" ? result.text : ""; + }; + const screen = (await capture(0)).split("\n"); + // Ten rows: the last numbers, the green line and the prompt. + assert.ok(screen.length <= 10); + assert.deepEqual(screen.slice(-3, -1), ["100", "green"]); + assert.match(screen.at(-1) ?? "", /[$#]$/); + const scrolled = (await capture(20)).split("\n"); + assert.equal(scrolled.length, screen.length + 20); + assert.equal(scrolled[0], String(Number(screen[0]) - 20)); + assert.match(await capture(0, true), new RegExp(`${ESC}\\[0;38;2;\\d+;\\d+;\\d+mgreen${ESC}\\[0m`)); + // More output than the ring keeps: the read reports what it lost and returns the last part. + await send("2", "plain-4", ["head -c 1500000 /dev/zero | tr '\\0' x; echo; echo E''ND", "Enter"]); + const flood = await agent.read({ + sessionId: "s1", + target: "2", + wait: { until: "match", pattern: "^END$", timeoutMs: 30_000 }, + }); + assert.ok(flood.skipped > 0, `skipped ${flood.skipped}`); + assert.ok(flood.truncated > 0, `truncated ${flood.truncated}`); + assert.match(flood.text, /^END$/m); + }); + + await t.test("the agent's cap, closing once per key, closing a session", async () => { + const other = await client(directory); + other.send({ type: "list", sessionId: "s2" }); + // The owner's own terminals do not count against the agent's cap. + other.send({ type: "open", sessionId: "s2", cwd, env: null, cols: 40, rows: 5 }); + await poll(() => other.events.some((event) => event.type === "attached")); + for (let index = 0; index < 8; index++) await open("s2", `cap-${index}`); + await assert.rejects(open("s2", "cap-8"), /the agent has 8 terminals running in this session already/); + const closed = await agent.call({ op: "close", sessionId: "s2", key: "close-1", target: "2" }); + assert.equal(closed.kind, "closed"); + assert.deepEqual(await agent.call({ op: "close", sessionId: "s2", key: "close-1", target: "2" }), closed); + await assert.rejects( + agent.call({ op: "close", sessionId: "s2", key: "close-2", target: "2" }), + /can't find window: 2/, + ); + await open("s2", "cap-8"); + assert.deepEqual(await agent.call({ op: "closeSession", sessionId: "s2" }), { kind: "done" }); + const list = await agent.call({ op: "list", sessionId: "s2" }); + assert.deepEqual(list, { kind: "list", windows: [] }); + await poll(() => { + const last = other.events.filter((event) => event.type === "terminals").at(-1); + return last?.type === "terminals" && last.terminals.length === 0; + }); + // Session s1 is untouched. + assert.equal((await agent.call({ op: "list", sessionId: "s1" })).kind, "list"); + other.socket.destroy(); + }); + + await t.test("a host running other code refuses calls", async () => { + const calls = new HostCalls(directory, APP, "another-version"); + await assert.rejects(calls.call({ op: "list", sessionId: "s1" }), new Error(OUTDATED_HOST)); + calls.close(); + }); + } finally { + viewer.socket.destroy(); + agent.socket.destroy(); + const last = await client(directory); + last.send({ type: "shutdown" }); + await new Promise((resolve) => last.socket.once("close", resolve)); + if (shell === undefined) delete process.env.SHELL; + else process.env.SHELL = shell; + await rm(directory, { recursive: true, force: true }); + await rm(cwd, { recursive: true, force: true }); + } +}); + +async function poll(check: () => boolean, timeout = 10_000): Promise<void> { + const deadline = Date.now() + timeout; + while (!check()) { + if (Date.now() > deadline) throw new Error("Timed out"); + await new Promise((resolve) => setTimeout(resolve, 50)); + } +} diff --git a/apps/p2/test/p2/terminal-commands.test.ts b/apps/p2/test/p2/terminal-commands.test.ts new file mode 100644 index 000000000..61065a721 --- /dev/null +++ b/apps/p2/test/p2/terminal-commands.test.ts @@ -0,0 +1,91 @@ +// The terminal tool's tmux commands: argv parsed getopt-style as tmux does, and tmux-like errors. +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { CommandError, parseCommand } from "../../plugins/terminals/commands.ts"; + +test("tmux argv: options first, combined flags, attached or separate values, `--`", () => { + assert.deepEqual(parseCommand("new-window", ["-n", "dev", "-c", "web", "npm", "run", "dev"]), { + command: "new-window", + name: "dev", + directory: "web", + shellCommand: "npm run dev", + }); + assert.deepEqual(parseCommand("new-window", []), { command: "new-window" }); + assert.deepEqual(parseCommand("send-keys", ["-t1", "-l", "npm test", "Enter"]), { + command: "send-keys", + target: "1", + literal: true, + keys: ["npm test", "Enter"], + }); + // Options end at the first argument that is not one, or at `--`: `-t` after a key is a key. + assert.deepEqual(parseCommand("send-keys", ["-t", "dev", "--", "-l", "q"]).command, "send-keys"); + assert.deepEqual(parseCommand("send-keys", ["-t", "dev", "--", "-l"]), { + command: "send-keys", + target: "dev", + literal: false, + keys: ["-l"], + }); + assert.deepEqual(parseCommand("send-keys", ["-t", "2", "C-c", "-t"]), { + command: "send-keys", + target: "2", + literal: false, + keys: ["C-c", "-t"], + }); + // A value may start with `-`: `-S -50`. + assert.deepEqual(parseCommand("capture-pane", ["-ep", "-t", "2", "-S", "-50"]), { + command: "capture-pane", + target: "2", + scrollback: 50, + escapes: true, + }); + assert.equal((parseCommand("capture-pane", ["-t", "2", "-S", "-"]) as { scrollback: number }).scrollback, 100_000); + assert.equal((parseCommand("capture-pane", ["-pt2"]) as { scrollback: number }).scrollback, 0); + assert.deepEqual(parseCommand("kill-window", ["-t", ":3"]), { command: "kill-window", target: ":3" }); + assert.deepEqual(parseCommand("list-windows", []), { command: "list-windows" }); +}); + +test("waits: quiet, exit, none, /regex/ with a timeout in seconds", () => { + assert.deepEqual(parseCommand("read", ["-t", "1", "-w", "/\\$ $/", "-T", "2.5"]), { + command: "read", + target: "1", + wait: { kind: "match", pattern: "\\$ $", ignoreCase: false }, + timeout: 2.5, + }); + assert.deepEqual(parseCommand("read", ["-t", "1", "-w", "/ready/i"]), { + command: "read", + target: "1", + wait: { kind: "match", pattern: "ready", ignoreCase: true }, + }); + for (const kind of ["quiet", "exit", "none"]) + assert.deepEqual((parseCommand("send-keys", ["-t", "1", "-w", kind, "q"]) as { wait: unknown }).wait, { kind }); +}); + +test("refusals read like tmux's", () => { + const refused = (command: Parameters<typeof parseCommand>[0], args: string[], message: string | RegExp) => + assert.throws( + () => parseCommand(command, args), + (error: unknown) => { + assert.ok(error instanceof CommandError); + if (typeof message === "string") assert.equal(error.message, message); + else assert.match(error.message, message); + return true; + }, + ); + refused( + "send-keys", + ["-x", "-t", "1", "a"], + "send-keys: unknown option -- x\nusage: send-keys [-l] [-w wait] [-T seconds] -t target-window key ...", + ); + refused("send-keys", ["-t"], /^send-keys: option requires an argument -- t\nusage: send-keys /); + refused("send-keys", ["Enter"], "usage: send-keys [-l] [-w wait] [-T seconds] -t target-window key ..."); + refused("send-keys", ["-t", "1"], /^usage: send-keys/); + refused("read", [], "usage: read [-w wait] [-T seconds] -t target-window"); + refused("read", ["-t", "1", "extra"], /^usage: read/); + refused("kill-window", [], "usage: kill-window -t target-window"); + refused("list-windows", ["-a"], /^list-windows: unknown option -- a/); + refused("capture-pane", ["-t", "1", "-S", "ten"], "capture-pane: invalid start line: ten"); + refused("read", ["-t", "1", "-w", "soon"], "read: invalid wait: soon (quiet, exit, none or /regex/)"); + refused("read", ["-t", "1", "-w", "/(/"], /^read: invalid regex: /); + refused("read", ["-t", "1", "-T", "500"], "read: invalid timeout: 500 (0 to 120 seconds)"); + refused("read", ["-t", "1", "-T", "x"], /^read: invalid timeout: x/); +}); diff --git a/apps/p2/test/p2/terminal-keys.test.ts b/apps/p2/test/p2/terminal-keys.test.ts new file mode 100644 index 000000000..9411b4123 --- /dev/null +++ b/apps/p2/test/p2/terminal-keys.test.ts @@ -0,0 +1,50 @@ +// tmux key names as libghostty key presses: the bytes programs receive, in the terminal's current modes. +import assert from "node:assert/strict"; +import { readFile } from "node:fs/promises"; +import { test } from "node:test"; +import { Ghostty } from "../../plugins/terminals/ghostty.ts"; +import { tmuxKey } from "../../plugins/terminals/keys.ts"; + +const ghostty = await Ghostty.load( + await readFile(new URL("../../plugins/terminals/vendor/ghostty/ghostty-vt.wasm", import.meta.url)), +); + +test("tmux key names encode as a terminal sends them; other words are text", () => { + const vt = ghostty.terminal({ cols: 20, rows: 4, scrollback: 10 }); + const encode = (name: string) => { + const key = tmuxKey(name); + const bytes = key && vt.encodeKey(key); + return bytes ? Buffer.from(bytes).toString("latin1") : undefined; + }; + const expected: Record<string, string> = { + Enter: "\r", + Escape: "\x1b", + "C-[": "\x1b", + Tab: "\t", + BTab: "\x1b[Z", + BSpace: "\x7f", + Space: " ", + "C-c": "\x03", + "C-C": "\x03", + "^D": "\x04", + "C-Space": "\x00", + "C-\\": "\x1c", + "C-_": "\x1f", + "M-x": "\x1bx", + "M-X": "\x1bX", + Up: "\x1b[A", + "S-Up": "\x1b[1;2A", + Home: "\x1b[H", + PPage: "\x1b[5~", + DC: "\x1b[3~", + F1: "\x1bOP", + F5: "\x1b[15~", + }; + for (const [name, bytes] of Object.entries(expected)) assert.equal(encode(name), bytes, name); + // Not key names: tmux types them. + for (const word of ["a", "npm test", "enter", "C-", "Ctrl-c", "x-y"]) assert.equal(tmuxKey(word), undefined, word); + // Application cursor mode (a full-screen program asked for it) changes the arrows. + vt.write(new TextEncoder().encode("\x1b[?1h")); + assert.equal(encode("Up"), "\x1bOA"); + vt.free(); +}); diff --git a/apps/p2/test/p2/terminal-plain.test.ts b/apps/p2/test/p2/terminal-plain.test.ts new file mode 100644 index 000000000..b9be86f82 --- /dev/null +++ b/apps/p2/test/p2/terminal-plain.test.ts @@ -0,0 +1,50 @@ +// Terminal output for the agent: the output ring's absolute offsets and overflow, and plain text as a terminal shows +// it line by line (escape sequences removed, carriage-return overwrites, backspaces, erase in line). +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { OutputRing, plainText } from "../../plugins/terminals/plain.ts"; + +const bytes = (text: string) => new TextEncoder().encode(text); +const text = (data: Uint8Array) => new TextDecoder().decode(data); + +test("the output ring keeps the last bytes at absolute offsets and counts what it lost", () => { + const ring = new OutputRing(8); + ring.append(bytes("abc")); + assert.deepEqual([ring.start, ring.end], [0, 3]); + assert.deepEqual({ ...ring.since(1), bytes: text(ring.since(1).bytes) }, { bytes: "bc", skipped: 0 }); + ring.append(bytes("defgh")); + ring.append(bytes("ij")); + assert.deepEqual([ring.start, ring.end], [2, 10]); + // Reading from 0: the first two bytes are gone. + assert.deepEqual({ ...ring.since(0), bytes: text(ring.since(0).bytes) }, { bytes: "cdefghij", skipped: 2 }); + assert.equal(text(ring.since(7).bytes), "hij"); + assert.equal(text(ring.since(10).bytes), ""); + // A chunk larger than the ring keeps its tail. + ring.append(bytes("0123456789AB")); + assert.deepEqual([ring.start, ring.end], [14, 22]); + assert.equal(text(ring.since(14).bytes), "456789AB"); + assert.equal(ring.since(12).skipped, 2); +}); + +test("plain text: escape sequences go, overwrites and backspaces apply", () => { + // Colors (CSI), a title (OSC with BEL and with ST), a DCS string, charset selection, a keypad mode. + assert.equal( + plainText("\x1b[1;31mred\x1b[0m \x1b]0;title\x07\x1b]2;t\x1b\\x\x1bPq#0\x1b\\y\x1b(Bz\x1b=!"), + "red xyz!", + ); + // A progress bar keeps its final state; CRLF is a newline. + assert.equal(plainText("10%\r50%\r100%\r\ndone\r\n"), "100%\ndone\n"); + // Overwriting a longer line keeps its tail unless the line is erased. + assert.equal(plainText("downloading\rdone"), "doneloading"); + assert.equal(plainText("downloading\r\x1b[Kdone"), "done"); + assert.equal(plainText("abc\x1b[2Kx"), " x"); + assert.equal(plainText("abc\b\bX"), "aXc"); + assert.equal(plainText("\b\bab"), "ab"); + // Other control characters go; tabs stay; trailing spaces are trimmed, except on the last line (a prompt). + assert.equal(plainText("a\x07b\x00c\td \n$ "), "abc\td\n$ "); + // C1 forms of CSI and OSC. + assert.equal(plainText("a\x9b31mb\x9d0;t\x9cc"), "abc"); + // A sequence cut off at the end leaves nothing behind. + assert.equal(plainText("ok\x1b[3"), "ok"); + assert.equal(plainText("ok\x1b]0;unterminated"), "ok"); +}); diff --git a/apps/p2/test/p2/terminal-tool-unit.test.ts b/apps/p2/test/p2/terminal-tool-unit.test.ts new file mode 100644 index 000000000..1b6d4753f --- /dev/null +++ b/apps/p2/test/p2/terminal-tool-unit.test.ts @@ -0,0 +1,235 @@ +// The terminal tool against a recording backend: what each command asks the server for (locations, keys, waits, +// read starts), its results, and replay safety (a rerun of the same tool call reads from the same start and closes +// the same window; the host does keyed calls once). +import assert from "node:assert/strict"; +import { test } from "node:test"; +import type { ToolExecutionApi, ToolExecutionResult } from "@earendil-works/pi-durable"; +import { BACKGROUND_CONTEXT, type JsonValue } from "@p2/app/v1"; +import { createTerminalExtension, type TerminalBackend } from "../../plugins/terminals/tool.ts"; + +type Info = Awaited<ReturnType<TerminalBackend["open"]>>; +type Read = Awaited<ReturnType<TerminalBackend["read"]>>; + +const info = (index: number, extra: Partial<Info> = {}): Info => ({ + id: `id-${index}`, + sessionId: "s", + index, + cwd: "/work", + env: null, + title: "", + createdAt: 0, + cols: 80, + rows: 24, + ...extra, +}); + +/** A backend recording its calls; `read` answers with `reads` in turn (or empty output). */ +function recorder(reads: Partial<Read>[] = []) { + const calls: [string, unknown][] = []; + let cursor = 100; + const backend: TerminalBackend = { + list: async () => { + calls.push(["list", null]); + return [ + { terminal: info(1, { agent: true, name: "dev" }), foreground: "vim" }, + { terminal: info(2, { env: "gpu", cwd: "/srv", exited: { code: 2 } }) }, + ]; + }, + open: async (args) => { + calls.push(["open", args]); + return info(3, { cwd: args.location.cwd, env: args.location.env }); + }, + send: async (args) => { + calls.push(["send", args]); + return info(1); + }, + cursor: async (target) => { + calls.push(["cursor", target]); + cursor += 10; + return { terminal: info(Number(target) || 1), offset: cursor }; + }, + read: async (args) => { + calls.push(["read", args]); + return { + terminal: info(1), + from: args.from ?? 0, + to: 0, + text: "", + skipped: 0, + truncated: 0, + ...reads.shift(), + }; + }, + capture: async (args) => { + calls.push(["capture", args]); + return { terminal: info(1), text: "screen" }; + }, + close: async (args) => { + calls.push(["close", args]); + return { terminalId: args.target }; + }, + }; + return { backend, calls }; +} + +/** One tool call's API: its task id and memo store (shared by its reruns). */ +function callApi(taskId: number): ToolExecutionApi { + const memos = new Map<string, JsonValue>(); + const api = { + taskId, + conversationId: 7, + memo: async (name: string, candidate: JsonValue | typeof BACKGROUND_CONTEXT, _context?: unknown) => { + if (!memos.has(name) && candidate !== BACKGROUND_CONTEXT) memos.set(name, candidate as JsonValue); + return memos.get(name); + }, + }; + // The tool uses only these members of the API. + return api as unknown as ToolExecutionApi; +} + +function terminal(backend: TerminalBackend, reach: "codemode" | "tool_search" = "codemode") { + const extension = createTerminalExtension(reach, backend, async (api) => { + assert.equal(api.conversationId, 7); + return { cwd: "/work/sub", env: "gpu" }; + }); + return extension.tools![0]!; +} + +const run = async ( + backend: TerminalBackend, + command: string, + args: string[], + api = callApi(42), +): Promise<ToolExecutionResult & { structuredOutput?: JsonValue }> => + terminal(backend).execute({ command, args }, api, BACKGROUND_CONTEXT); + +test("code mode only while code mode is on; tool_search reaches it otherwise", () => { + const { backend } = recorder(); + assert.deepEqual(terminal(backend).callers, ["tools"]); + assert.equal(terminal(backend).deferred, true); + assert.equal(terminal(backend).replay, "safe"); + assert.equal(terminal(backend, "tool_search").callers, undefined); + assert.equal(terminal(backend, "tool_search").deferred, true); + assert.match(terminal(backend).description, /Only when the user asks for a terminal/); +}); + +test("new-window opens at the calling conversation's location, keyed by the call, and reads from its start", async () => { + const { backend, calls } = recorder([{ text: "$ ", terminal: info(3) }]); + const result = await run(backend, "new-window", ["-n", "dev", "-c", "../web", "npm", "run", "dev"]); + assert.deepEqual(calls, [ + ["open", { key: "42", location: { cwd: "/work/web", env: "gpu" }, name: "dev", command: "npm run dev" }], + ["read", { target: "id-3", from: 0, wait: { until: "quiet", timeoutMs: 10_000 } }], + ]); + assert.deepEqual(result.structuredOutput, { window: 3, running: true, output: "$ " }); + assert.equal((result.output?.[0] as { text: string }).text, "window 3 · at the prompt\n$ "); + const absolute = recorder(); + await run(absolute.backend, "new-window", ["-c", "/tmp", "-w", "none"]); + assert.deepEqual(absolute.calls, [ + ["open", { key: "42", location: { cwd: "/tmp", env: "gpu" } }], + ["read", { target: "id-3", from: 0 }], + ]); +}); + +test("send-keys and read wait by default as documented; -w and -T override", async () => { + const { backend, calls } = recorder(); + await run(backend, "send-keys", ["-t", "1", "q"], callApi(1)); + await run(backend, "send-keys", ["-t", "1", "-w", "none", "-l", "C-c"], callApi(2)); + await run(backend, "read", ["-t", "1"], callApi(3)); + await run(backend, "read", ["-t", "1", "-w", "/x/i", "-T", "2"], callApi(4)); + assert.deepEqual(calls, [ + ["cursor", "1"], + ["send", { key: "1", target: "id-1", keys: ["q"], literal: false }], + ["read", { target: "id-1", from: 110, wait: { until: "quiet", timeoutMs: 5000 } }], + ["cursor", "1"], + ["send", { key: "2", target: "id-1", keys: ["C-c"], literal: true }], + ["read", { target: "id-1", from: 120 }], + ["cursor", "1"], + ["read", { target: "id-1", from: 130 }], + ["cursor", "1"], + [ + "read", + { target: "id-1", from: 140, wait: { until: "match", pattern: "x", ignoreCase: true, timeoutMs: 2000 } }, + ], + ]); +}); + +test("a rerun of the same call reads from the same start and closes the same window", async () => { + const { backend, calls } = recorder(); + const api = callApi(9); + await run(backend, "send-keys", ["-t", "1", "ls", "Enter"], api); + await run(backend, "send-keys", ["-t", "1", "ls", "Enter"], api); + // One cursor: the rerun's start is the memoized one; the send repeats with the same key (the host does it once). + assert.deepEqual( + calls.filter(([name]) => name !== "read"), + [ + ["cursor", "1"], + ["send", { key: "9", target: "id-1", keys: ["ls", "Enter"], literal: false }], + ["send", { key: "9", target: "id-1", keys: ["ls", "Enter"], literal: false }], + ], + ); + assert.deepEqual( + calls.filter(([name]) => name === "read").map(([, args]) => (args as { from: number }).from), + [110, 110], + ); + const killing = recorder(); + const kill = callApi(10); + assert.deepEqual((await run(killing.backend, "kill-window", ["-t", "2"], kill)).structuredOutput, { + window: 2, + killed: true, + }); + await run(killing.backend, "kill-window", ["-t", "2"], kill); + assert.deepEqual(killing.calls, [ + ["cursor", "2"], + ["close", { key: "10", target: "id-2" }], + ["close", { key: "10", target: "id-2" }], + ]); +}); + +test("results: the screen of full-screen programs, skipped output, timeouts, windows, errors", async () => { + const long = "x".repeat(20_000); + const { backend } = recorder([ + { text: "ignored", screen: "VIM", foreground: "vim", waited: "timeout" }, + { text: long, skipped: 5, truncated: 7, terminal: info(1, { exited: { code: 1 } }) }, + ]); + const screen = await run(backend, "read", ["-t", "1"]); + assert.deepEqual(screen.structuredOutput, { + window: 1, + running: true, + foreground: "vim", + output: "VIM", + screen: true, + timedOut: true, + }); + assert.equal( + (screen.output?.[0] as { text: string }).text, + "window 1 · running vim · full-screen program\nVIM\n(timed out: still running (foreground: vim))", + ); + const skipped = (await run(backend, "read", ["-t", "1"])).structuredOutput as { output: string; skipped: number }; + assert.equal(skipped.output.length, 16 * 1024); + assert.equal(skipped.skipped, 5 + 7 + 20_000 - 16 * 1024); + const listed = await run(backend, "list-windows", []); + assert.deepEqual(listed.structuredOutput, { + windows: [ + { window: 1, name: "dev", location: "/work", opener: "agent", running: true, foreground: "vim" }, + { window: 2, location: "gpu:/srv", opener: "user", running: false, exitCode: 2 }, + ], + }); + assert.equal( + (listed.output?.[0] as { text: string }).text, + "1: dev · /work · opened by agent · running vim\n2: gpu:/srv · opened by user · exited with code 2", + ); + assert.deepEqual(await run(backend, "kill-window", []), { + output: [{ type: "text", text: "usage: kill-window -t target-window" }], + isError: true, + }); + const failing: TerminalBackend = { + ...backend, + cursor: async () => { + throw new Error("can't find window: 9"); + }, + }; + assert.deepEqual(await run(failing, "read", ["-t", "9"]), { + output: [{ type: "text", text: "can't find window: 9" }], + isError: true, + }); +}); diff --git a/apps/p2/test/p2/terminal-tool.test.ts b/apps/p2/test/p2/terminal-tool.test.ts new file mode 100644 index 000000000..f2dc4d115 --- /dev/null +++ b/apps/p2/test/p2/terminal-tool.test.ts @@ -0,0 +1,254 @@ +// The agent's terminal tool end to end with the fake provider and a real terminal host (skipped without a pi-env +// daemon build): code mode scripts drive the session's terminals, which are the owner's (listed, labeled, typed into +// by both); guest turns may use it; tmux errors come back as tool errors; the server serves only the session's current worker, on the session's environments, and archiving closes +// the session's terminals. Without code mode, `tool_search` loads the tool. +import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import { realpathSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { test } from "node:test"; +import { setTimeout as sleep } from "node:timers/promises"; +import type { ToolResultMessage } from "@earendil-works/pi-ai"; +import type { ConversationView } from "@earendil-works/pi-durable"; +import { BACKGROUND_CONTEXT, whenLive } from "@p2/app/v1"; +import { WebSocket } from "ws"; +import { Chat } from "../../plugins/chat/contract.ts"; +import { Sessions } from "../../plugins/sessions/contract.ts"; +import { Sharing } from "../../plugins/sharing/contract.ts"; +import { AgentTerminals, TerminalServices, Terminals } from "../../plugins/terminals/contract.ts"; +import { localSystem, terminalDaemon } from "../../plugins/terminals/daemon.ts"; +import { connectTerminalHost } from "../../plugins/terminals/detached.ts"; +import { encodeFrame, type TerminalEvent, type TerminalInfo } from "../../plugins/terminals/protocol.ts"; +import { Hosts } from "../../src/runtime/server.ts"; +import { APP, expectChord, holdIn, http, poll, startFixture, until } from "./helpers.ts"; + +let unavailable: string | undefined; +try { + const system = localSystem(); + await terminalDaemon(system.platform, system.arch, true); +} catch (error) { + unavailable = error instanceof Error ? error.message : "No pi-env daemon with terminal support"; +} + +const results = (view: ConversationView | null | undefined): ToolResultMessage[] => + (view?.entries ?? []).flatMap((entry) => + (entry.model ?? []).filter((message): message is ToolResultMessage => message.role === "toolResult"), + ); +const text = (result: ToolResultMessage | undefined) => + (result?.content ?? []).flatMap((part) => (part.type === "text" ? [part.text] : [])).join(""); +/** A completed code mode result's returned value. */ +const returned = (result: ToolResultMessage): unknown => { + const match = /^Script completed\nWall time [\d.]+ seconds\nOutput:\n\n?/.exec(text(result)); + assert.ok(match, `a completed script's result: ${text(result)}`); + return JSON.parse(text(result).slice(match[0].length)); +}; + +test( + "the agent's terminal tool on the session's shared terminals", + { timeout: 180_000, skip: unavailable }, + async (t) => { + const shell = process.env.SHELL; + // The terminal host and its shells inherit this: a plain shell, without the developer's rc files. + process.env.SHELL = "/bin/sh"; + const fixture = await startFixture({ allowedHosts: ["share.test"] }); + t.after(async () => { + await fixture.close(); + const host = await connectTerminalHost(fixture.dataDir, APP); + host.resume(); + host.write(encodeFrame({ type: "shutdown" })); + await new Promise((resolve) => host.once("close", resolve)); + if (shell === undefined) delete process.env.SHELL; + else process.env.SHELL = shell; + }); + const workspace = realpathSync(fixture.workspace); + const owner = fixture.browser(); + const sessions = await holdIn(owner, Sessions, {}, "owner"); + const session = await sessions.create({ cwd: workspace, title: "Terminals" }, BACKGROUND_CONTEXT); + const chat = await holdIn(owner, Chat, { session: session.id, conversation: 1 }, "participant"); + let serial = 0; + const call = async (prompt: string, target = chat) => { + const count = results(await until(target.view, Boolean)).length; + await target.send({ text: prompt, requestId: `terminal-${++serial}` }, BACKGROUND_CONTEXT); + return results( + await until( + target.view, + (view) => results(view).length > count && !(view?.docs["pi.live"] as { run?: unknown })?.run, + ), + ).at(-1)!; + }; + const script = async (code: string, target = chat) => { + const result = await call(`codemode ${code}`, target); + assert.equal(result.isError, false, text(result)); + return returned(result); + }; + // The owner's terminal panel: its byte channel hears the session's terminals change. + const view = await holdIn(owner, Terminals, { session: session.id, view: randomUUID() }, "owner"); + const bytes = await byteChannel(fixture.server.port, (await view.channel({}, BACKGROUND_CONTEXT)).token); + t.after(() => bytes.socket.terminate()); + await view.list({}, BACKGROUND_CONTEXT); + const listed = () => + bytes.events.findLast( + (event): event is Extract<TerminalEvent, { type: "terminals" | "list" }> => + event.type === "terminals" || event.type === "list", + )?.terminals ?? []; + + await t.test("a script opens a window, types into it and reads its output", async () => { + const value = (await script( + [ + 'const opened = await tools.terminal({ command: "new-window", args: ["-n", "dev", "-w", "/^hello-agent$/", "echo hello-age\'\'nt"] });', + 'const sent = await tools.terminal({ command: "send-keys", args: ["-t", "dev", "-w", "/^two$/", "echo t\'\'wo", "Enter"] });', + 'const windows = await tools.terminal({ command: "list-windows" });', + "return { opened, sent, windows };", + ].join("\n"), + )) as { + opened: { window: number; output: string; running: boolean }; + sent: { window: number; output: string; timedOut?: boolean }; + windows: { windows: { window: number; name?: string; opener: string; location: string }[] }; + }; + assert.equal(value.opened.window, 1); + assert.equal(value.opened.running, true); + assert.match(value.opened.output, /^hello-agent$/m); + assert.match(value.sent.output, /^two$/m); + assert.doesNotMatch(value.sent.output, /hello-agent/); + assert.deepEqual( + value.windows.windows.map(({ window, name, opener, location }) => ({ window, name, opener, location })), + [{ window: 1, name: "dev", opener: "agent", location: workspace }], + ); + // The owner's panel lists it, marked as the agent's. + await poll(() => listed().some((terminal) => terminal.name === "dev" && terminal.agent === true)); + }); + + await t.test("the owner types into the agent's window, the agent into the owner's", async () => { + const dev = listed().find((terminal) => terminal.name === "dev") as TerminalInfo; + await view.attach({ terminalId: dev.id }, BACKGROUND_CONTEXT); + await poll(() => bytes.events.some((event) => event.type === "attached")); + bytes.socket.send(Buffer.from("echo from-own''er\n")); + const read = (await script( + 'return await tools.terminal({ command: "read", args: ["-t", "1", "-w", "/^from-owner$/"] });', + )) as { output: string; timedOut?: boolean }; + assert.equal(read.timedOut, undefined); + assert.match(read.output, /^from-owner$/m); + // The owner opens a terminal; the agent sees and drives it. + await view.open({ cols: 80, rows: 10, conversation: 1 }, BACKGROUND_CONTEXT); + await poll(() => listed().length === 2); + const value = (await script( + [ + `const read = await tools.terminal({ command: "send-keys", args: ["-t", "2", "-w", "/^owned-by-user$/", "-l", "echo own''ed-by-user\\n"] });`, + 'const { windows } = await tools.terminal({ command: "list-windows" });', + "return { read, windows };", + ].join("\n"), + )) as { read: { output: string; timedOut?: boolean }; windows: { window: number; opener: string }[] }; + assert.equal(value.read.timedOut, undefined); + assert.match(value.read.output, /^owned-by-user$/m); + assert.deepEqual( + value.windows.map(({ window, opener }) => [window, opener]), + [ + [1, "agent"], + [2, "user"], + ], + ); + }); + + await t.test("tmux errors and a missing window are tool errors", async () => { + const value = (await script( + [ + "const errors = [];", + 'for (const [command, args] of [["send-keys", ["-x"]], ["send-keys", ["Enter"]], ["read", ["-t", "9"]], ["capture-pane", ["-t", "1", "-S", "x"]]])', + " try { await tools.terminal({ command, args }); errors.push('no error'); } catch (error) { errors.push(String(error.message ?? error)); }", + "return errors;", + ].join("\n"), + )) as string[]; + assert.equal(value.length, 4); + assert.match(value[0]!, /send-keys: unknown option -- x/); + assert.match(value[1]!, /usage: send-keys/); + assert.match(value[2]!, /can't find window: 9/); + assert.match(value[3]!, /capture-pane: invalid start line: x/); + }); + + await t.test("guest turns may use it, like bash", async () => { + const sharing = owner.hold(Sharing, { session: session.id }, BACKGROUND_CONTEXT); + await sharing.link({ mode: "full" }, BACKGROUND_CONTEXT); + const links = await until(sharing.links, (value) => Boolean(value?.length)); + const headers = { Host: "share.test", Origin: "https://share.test", "Content-Type": "application/json" }; + const joined = await http(fixture.server.port, "/join", { + method: "POST", + headers, + body: JSON.stringify({ token: links?.[0]?.token, name: "Guest" }), + }); + const guest = fixture.client({ ...headers, Cookie: String(joined.headers["set-cookie"]).split(";")[0]! }).host; + const guestChat = await holdIn(guest, Chat, { session: session.id, conversation: 1 }, "participant"); + const value = (await script( + 'return await tools.terminal({ command: "capture-pane", args: ["-p", "-t", "dev"] });', + guestChat, + )) as { window: number; output: string }; + assert.equal(value.window, 1); + assert.match(value.output, /from-owner/); + // Guests still have no terminal panel. + await expectChord( + whenLive(guest.hold(Terminals, { session: session.id, view: randomUUID() }, BACKGROUND_CONTEXT)), + "forbidden", + ); + }); + + await t.test("the server serves the session's current worker only, on its environments", async () => { + const state = await fixture.server.host.root( + async (ctx) => (await ctx.get(Hosts, {})).workers.value[session.id], + BACKGROUND_CONTEXT, + ); + assert.ok(state?.running); + const as = (generation: number) => + holdIn(fixture.as({ kind: "worker", session: session.id, generation }), AgentTerminals, {}, "worker"); + const stale = await as(state.generation - 1); + await expectChord(stale.list({}, BACKGROUND_CONTEXT), "forbidden"); + const current = await as(state.generation); + assert.equal((await current.list({}, BACKGROUND_CONTEXT)).windows.length, 2); + await expectChord( + current.open({ key: "foreign", location: { cwd: "/", env: "gpu" } }, BACKGROUND_CONTEXT), + "forbidden", + ); + // The owner is no worker. + await expectChord(whenLive(owner.hold(AgentTerminals, {}, BACKGROUND_CONTEXT)), "forbidden"); + }); + + await t.test("archiving the session closes its terminals", async () => { + await sessions.archive({ id: session.id }, BACKGROUND_CONTEXT); + const service = (await fixture.server.host.registry(TerminalServices, BACKGROUND_CONTEXT)).value!.service; + assert.deepEqual(await service.call({ op: "list", sessionId: session.id }), { kind: "list", windows: [] }); + await poll(() => listed().length === 0); + }); + + await t.test("without code mode, tool_search loads the tool for the model", async () => { + const other = await sessions.create({ cwd: workspace, title: "No code mode" }, BACKGROUND_CONTEXT); + const otherChat = await holdIn(owner, Chat, { session: other.id, conversation: 1 }, "participant"); + // Code mode on: the model is not offered it. + assert.equal((await call(`terminal ${JSON.stringify({ command: "list-windows" })}`, otherChat)).isError, true); + writeFileSync(join(fixture.dataDir, "settings.json"), JSON.stringify({ codemode: false })); + let found: ToolResultMessage | undefined; + for (let attempt = 0; attempt < 20 && !(found && !found.isError); attempt++) { + if (attempt > 0) await sleep(200); + found = await call(`tool_search ${JSON.stringify({ query: "terminal tmux windows" })}`, otherChat); + } + assert.equal(found?.isError, false, text(found)); + assert.match(text(found), /- terminal: /); + const direct = await call(`terminal ${JSON.stringify({ command: "list-windows" })}`, otherChat); + assert.equal(direct.isError, false, text(direct)); + assert.equal(text(direct), "No windows."); + }); + }, +); + +async function byteChannel(port: number, token: string) { + const socket = new WebSocket(`ws://127.0.0.1:${port}/terminal?${new URLSearchParams({ token })}`); + const events: TerminalEvent[] = []; + socket.on("message", (data, binary) => { + if (!binary) { + const event = JSON.parse(data.toString()) as TerminalEvent | { type: "ping" }; + if (event.type !== "ping") events.push(event); + } + }); + await new Promise<void>((resolve, reject) => { + socket.once("open", resolve); + socket.once("error", reject); + }); + return { socket, events }; +}