From d82b34c5144a291acd1da6242575baa6d54f1377 Mon Sep 17 00:00:00 2001 From: Maksym Yezhov Date: Thu, 15 Oct 2026 18:19:08 -0700 Subject: [PATCH] feat: local agent host package --- packages/local-agent-host/README.md | 132 ++++++++ packages/local-agent-host/eslint.config.js | 3 + packages/local-agent-host/package.json | 29 ++ .../resources/proxyProvider.ts | 38 +++ .../resources/sessionResources.ts | 187 +++++++++++ packages/local-agent-host/src/bridge.test.ts | 167 ++++++++++ packages/local-agent-host/src/bridge.ts | 293 ++++++++++++++++++ packages/local-agent-host/src/fileToolOps.ts | 153 +++++++++ packages/local-agent-host/src/hostApi.test.ts | 87 ++++++ packages/local-agent-host/src/hostApi.ts | 113 +++++++ .../local-agent-host/src/piAdapter.test.ts | 74 +++++ packages/local-agent-host/src/piAdapter.ts | 189 +++++++++++ .../local-agent-host/src/piEvents.test.ts | 225 ++++++++++++++ packages/local-agent-host/src/piEvents.ts | 214 +++++++++++++ .../src/remoteEnvToken.test.ts | 83 +++++ .../local-agent-host/src/remoteEnvToken.ts | 98 ++++++ .../src/remoteFileTools.test.ts | 170 ++++++++++ .../local-agent-host/src/remoteFileTools.ts | 127 ++++++++ .../src/scopedConnection.test.ts | 116 +++++++ .../local-agent-host/src/scopedConnection.ts | 78 +++++ .../local-agent-host/src/shellUrl.test.ts | 21 ++ packages/local-agent-host/src/shellUrl.ts | 26 ++ packages/local-agent-host/src/toolArgs.ts | 40 +++ packages/local-agent-host/src/workspaceFs.ts | 126 ++++++++ packages/local-agent-host/tsconfig.json | 8 + pnpm-lock.yaml | 25 ++ 26 files changed, 2822 insertions(+) create mode 100644 packages/local-agent-host/README.md create mode 100644 packages/local-agent-host/eslint.config.js create mode 100644 packages/local-agent-host/package.json create mode 100644 packages/local-agent-host/resources/proxyProvider.ts create mode 100644 packages/local-agent-host/resources/sessionResources.ts create mode 100644 packages/local-agent-host/src/bridge.test.ts create mode 100644 packages/local-agent-host/src/bridge.ts create mode 100644 packages/local-agent-host/src/fileToolOps.ts create mode 100644 packages/local-agent-host/src/hostApi.test.ts create mode 100644 packages/local-agent-host/src/hostApi.ts create mode 100644 packages/local-agent-host/src/piAdapter.test.ts create mode 100644 packages/local-agent-host/src/piAdapter.ts create mode 100644 packages/local-agent-host/src/piEvents.test.ts create mode 100644 packages/local-agent-host/src/piEvents.ts create mode 100644 packages/local-agent-host/src/remoteEnvToken.test.ts create mode 100644 packages/local-agent-host/src/remoteEnvToken.ts create mode 100644 packages/local-agent-host/src/remoteFileTools.test.ts create mode 100644 packages/local-agent-host/src/remoteFileTools.ts create mode 100644 packages/local-agent-host/src/scopedConnection.test.ts create mode 100644 packages/local-agent-host/src/scopedConnection.ts create mode 100644 packages/local-agent-host/src/shellUrl.test.ts create mode 100644 packages/local-agent-host/src/shellUrl.ts create mode 100644 packages/local-agent-host/src/toolArgs.ts create mode 100644 packages/local-agent-host/src/workspaceFs.ts create mode 100644 packages/local-agent-host/tsconfig.json diff --git a/packages/local-agent-host/README.md b/packages/local-agent-host/README.md new file mode 100644 index 00000000..dd0582d8 --- /dev/null +++ b/packages/local-agent-host/README.md @@ -0,0 +1,132 @@ +# @tangent/local-agent-host + +The engine that hosts Tangent **sub-agents as local `pi` processes**, used by the +`tangent-local-agent` CLI (`apps/local-agent-cli`). + +It connects to a Tangent Shell over the `/remote-env` Socket.IO namespace using +[`@tangent/remote-subagent`](../remote-subagent/README.md), maps each +`spawn` / `message` / `kill` command onto a local `pi --mode rpc` child process +working in the user's open folder, and streams that child's output back as the +same agent events a server-hosted sub-agent produces. It also offers Prime a set +of read-only workspace tools and lets a hosted sub-agent contribute files back +into the Tangent session. + +This package is the **host runtime**; the thin CLI that parses flags, resolves a +workspace, and keeps a token fresh lives in `apps/local-agent-cli`. + +## Architecture + +```mermaid +flowchart LR + shell["Tangent Shell (/remote-env)"] <-->|socket.io| bridge["LocalAgentBridge"] + bridge -->|spawn/message/kill| pi["PiAgent (pi --mode rpc child)"] + pi -->|stdout events| bridge + bridge -->|"agentEvent / subagentUpdate"| shell + pi -->|"put/read session resource"| hostApi["loopback host API"] + hostApi -->|"addResource / readResource"| bridge +``` + +- **`LocalAgentBridge`** ([src/bridge.ts](src/bridge.ts)) — the core. Holds the + remote-env connection, a roster of hosted `pi` children, and the loopback host + API. `connect(token, environmentId)` (re)connects with a fresh scoped token + without interrupting running sub-agents; `dispose()` tears everything down. +- **`startScopedConnection`** ([src/scopedConnection.ts](src/scopedConnection.ts)) + — keeps the bridge connected with a self-renewing token: mints, connects, and + schedules a refresh before expiry, pinning the environment id across refreshes. +- **`mintRemoteEnvToken`** ([src/remoteEnvToken.ts](src/remoteEnvToken.ts)) — + mints a selector/session/user-scoped remote-env token from the Shell. +- **`PiAgent`** ([src/piAdapter.ts](src/piAdapter.ts)) — one long-lived + `pi --mode rpc` child driven over stdin/stdout JSONL. `buildPiArgs` and + `piEnv` assemble its launch; `src/piEvents.ts` translates pi's stdout into + remote-env agent events. +- **Remote file tools** ([src/remoteFileTools.ts](src/remoteFileTools.ts)) — + `createFileTools(getRoot)` returns the read-only workspace tools + (`read_file` / `list_dir` / `find_files` / `search_text`) Prime calls over + `/remote-env` without spawning a sub-agent. All paths resolve under the open + folder, with the guards in [src/workspaceFs.ts](src/workspaceFs.ts). + +## Usage + +```ts +import { LocalAgentBridge } from "@tangent/local-agent-host/bridge.ts"; +import { createFileTools } from "@tangent/local-agent-host/remoteFileTools.ts"; +import { startScopedConnection } from "@tangent/local-agent-host/scopedConnection.ts"; + +const bridge = new LocalAgentBridge({ + url: "http://localhost:8787", + selector: "projectId:abc", + metadata: { editor: "my-editor", workspacePath: "/Users/me/proj" }, + tools: createFileTools(() => "/Users/me/proj"), + piLaunchOptions: async () => ({ + bin: "pi", + cwd: "/Users/me/proj", + proxyExtensionPath: "/path/to/proxyProvider.ts", + sessionResourceExtensionPath: "/path/to/sessionResources.ts", + }), + log: console.log, + onChange: () => {}, + onSpawned: (agent) => console.log(`started ${agent.name}`), +}); + +// Keep a selector-scoped host connected with a self-renewing token: +const stop = startScopedConnection({ + shellUrl: "http://localhost:8787", + selector: "projectId:abc", + email: "me@example.com", + connect: (token, environmentId) => bridge.connect(token, environmentId), + onError: console.error, +}); +``` + +## Session resources + +A hosted `pi` sub-agent works in the user's **local** workspace, which is not the +Tangent session. So the host gives each sub-agent a small set of tools to push +files/blobs into the session — where Prime and every other agent (including +agents hosted on a different environment) can see and fetch them. + +The chain: + +1. The sub-agent calls `put_session_resource` / `list_session_resources` / + `read_session_resource`, registered by the bundled pi extension + [resources/sessionResources.ts](resources/sessionResources.ts) (loaded via + `--extension` when a host API is configured). +2. The extension calls back into the host's **loopback HTTP API** + ([src/hostApi.ts](src/hostApi.ts)) — bound to `127.0.0.1` on an ephemeral + port and guarded by a random per-process bearer token, so only this host's own + sub-agents can reach it. +3. The bridge forwards each call to the live connection's + `addResource` / `listResources` / `readResource`, which the Shell's gateway + services against the session catalog. + +`put_session_resource` takes either inline `content` (for small text/blobs, with +`encoding: 'base64'` for binary) or a `sourcePath` — a local file path the tool +reads off disk itself, resolved against the sub-agent's working folder — so large +files transfer without the agent inlining their bytes. Provide exactly one; a +`sourcePath` read is sent base64 and rejected locally when it exceeds 1 MiB. + +`buildPiArgs` loads the extension and `piEnv` injects `TANGENT_HOST_API_URL`, +`TANGENT_HOST_API_TOKEN`, `TANGENT_SESSION_ID`, and `TANGENT_AGENT_ID` only when a +`hostApiUrl` is present on the launch options; the bridge starts the loopback +lazily on the first spawn and closes it on `dispose()`. A `file` resource lands +under the session's `uploads/` folder (binary rides base64), catalogued and +surfaced to Prime; fetches are capped at 1 MiB. See the protocol and server side +in [`@tangent/remote-subagent`](../remote-subagent/README.md#session-resources). + +## Bundled extensions (`resources/`) + +`pi` loads these from disk, so a host copies them next to its bundle (the CLI's +`esbuild.mjs` does this). They are authored against pi's extension runtime, not +this repo's `node_modules`, so they are excluded from type-check and lint: + +- [resources/proxyProvider.ts](resources/proxyProvider.ts) — points pi's model + providers at the Shell's pi proxy (loaded only when a `proxyUrl` is set). +- [resources/sessionResources.ts](resources/sessionResources.ts) — the session + resource tools described above (loaded only when a host API is configured). + +## Exports + +Consumers import modules directly, e.g. `@tangent/local-agent-host/bridge.ts`. +Key entry points: `bridge.ts` (`LocalAgentBridge`), `scopedConnection.ts`, +`remoteEnvToken.ts`, `piAdapter.ts`, `remoteFileTools.ts`, `hostApi.ts`, +`shellUrl.ts`, and the `resources/*` extensions. diff --git a/packages/local-agent-host/eslint.config.js b/packages/local-agent-host/eslint.config.js new file mode 100644 index 00000000..ed1a71a7 --- /dev/null +++ b/packages/local-agent-host/eslint.config.js @@ -0,0 +1,3 @@ +import node from "@tangent/build/eslint/node"; + +export default [{ ignores: ["node_modules", "resources"] }, ...node]; diff --git a/packages/local-agent-host/package.json b/packages/local-agent-host/package.json new file mode 100644 index 00000000..d3b787b9 --- /dev/null +++ b/packages/local-agent-host/package.json @@ -0,0 +1,29 @@ +{ + "name": "@tangent/local-agent-host", + "version": "0.0.0", + "private": true, + "type": "module", + "exports": { + "./resources/*": "./resources/*", + "./*.ts": "./src/*.ts", + "./*": "./src/*.ts" + }, + "prettier": "@tangent/build/prettier", + "scripts": { + "lint": "eslint .", + "test": "tsx --test \"src/**/*.test.ts\"", + "typecheck": "tsc --noEmit", + "format": "prettier --write ." + }, + "dependencies": { + "@tangent/remote-subagent": "workspace:*", + "@tangent/shared": "workspace:*" + }, + "devDependencies": { + "@tangent/build": "workspace:*", + "@types/node": "^25.9.1", + "eslint": "^10.4.0", + "tsx": "^4.22.3", + "typescript": "^6.0.3" + } +} diff --git a/packages/local-agent-host/resources/proxyProvider.ts b/packages/local-agent-host/resources/proxyProvider.ts new file mode 100644 index 00000000..803074b6 --- /dev/null +++ b/packages/local-agent-host/resources/proxyProvider.ts @@ -0,0 +1,38 @@ +// @ts-nocheck +/** + * Loaded into local pi sub-agents via `--extension` when a pi proxy URL is set. + * Hosts copy it next to their bundle. Authored against pi's extension runtime + * (loaded with jiti), so it is excluded from this package's type-check and + * lint. Mirrors the server's `apps/server/src/pi/extensions/proxyProvider.ts`. + */ + +import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; + +const PROXY_URL = (process.env.PI_PROXY_URL ?? "").replace(/\/+$/, ""); + +export default function (pi: ExtensionAPI) { + pi.registerProvider("anthropic", { + baseUrl: `${PROXY_URL}/apis/anthropic`, + apiKey: "$PI_PROXY_API_KEY", + }); + + pi.registerProvider("openai", { + baseUrl: `${PROXY_URL}/v1`, + apiKey: "$PI_PROXY_API_KEY", + }); + + pi.registerProvider("google", { + baseUrl: `${PROXY_URL}/apis/google/v1`, + apiKey: "$PI_PROXY_API_KEY", + }); + + pi.registerProvider("groq", { + baseUrl: `${PROXY_URL}/vendors/groq/openai/v1`, + apiKey: "$PI_PROXY_API_KEY", + }); + + pi.registerProvider("xai", { + baseUrl: `${PROXY_URL}/vendors/xai/v1`, + apiKey: "$PI_PROXY_API_KEY", + }); +} diff --git a/packages/local-agent-host/resources/sessionResources.ts b/packages/local-agent-host/resources/sessionResources.ts new file mode 100644 index 00000000..bdf76d17 --- /dev/null +++ b/packages/local-agent-host/resources/sessionResources.ts @@ -0,0 +1,187 @@ +// @ts-nocheck +/** + * Loaded into local pi sub-agents via `--extension` so an agent working in the + * user's local workspace can contribute files/blobs to the Tangent *session* — + * where Prime and other agents (including those on other environments) can see + * and fetch them. Hosts copy it next to their bundle. Authored against pi's + * extension runtime (loaded with jiti), so it is excluded from this package's + * type-check and lint. Mirrors the server's host-resource extension, but calls + * the host's loopback API instead of the server's internal API. + */ + +import { readFile } from "node:fs/promises"; +import { basename, resolve } from "node:path"; + +import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; +import { Type } from "typebox"; + +const API_URL = (process.env.TANGENT_HOST_API_URL ?? "").replace(/\/+$/, ""); +const API_TOKEN = process.env.TANGENT_HOST_API_TOKEN ?? ""; +const SESSION_ID = process.env.TANGENT_SESSION_ID ?? ""; +const AGENT_ID = process.env.TANGENT_AGENT_ID ?? ""; + +/** Matches the server's per-file resource cap (MAX_FILE_RESOURCE_BYTES). */ +const MAX_FILE_BYTES = 1024 * 1024; + +async function callHostApi(route: string, payload: unknown): Promise { + if (!API_URL) throw new Error("Session resources are unavailable here."); + const response = await fetch(`${API_URL}/${route}`, { + method: "POST", + headers: { + "content-type": "application/json", + authorization: `Bearer ${API_TOKEN}`, + }, + body: JSON.stringify(payload), + }); + const data = (await response.json().catch(() => ({}))) as { + ok?: boolean; + result?: unknown; + error?: string; + }; + if (!response.ok || !data.ok) { + throw new Error( + data.error ?? `host API ${route} failed (${response.status})`, + ); + } + return data.result; +} + +function textResult(text: string) { + return { content: [{ type: "text", text }], details: {} }; +} + +function renderList(resources: Array<{ name: string; uri: string }>): string { + if (!resources || resources.length === 0) return "(no session resources)"; + return resources.map((r) => `- ${r.name} (${r.uri})`).join("\n"); +} + +export default function (pi: ExtensionAPI) { + pi.registerTool({ + name: "put_session_resource", + label: "Put Session Resource", + description: + "Write a file/blob into the Tangent session so Prime and other agents " + + "(including those on other environments) can fetch it. Pass 'content' " + + "inline for small pieces (binary must be base64 with encoding: 'base64'), " + + "or pass 'sourcePath' to upload a local file the tool reads itself without " + + "inlining its bytes. Provide exactly one of content/sourcePath. The decoded " + + "size limit is 1 MiB.", + promptSnippet: "Contribute a file to the Tangent session", + parameters: Type.Object({ + path: Type.String({ + description: "Destination path inside the session's uploads/ folder.", + }), + content: Type.Optional( + Type.String({ + description: + "Inline file contents (text, or base64). Use for small pieces.", + }), + ), + sourcePath: Type.Optional( + Type.String({ + description: + "Local file path the tool reads itself (preferred for large files). " + + "Resolved against the sub-agent's working folder.", + }), + ), + encoding: Type.Optional( + Type.Union([Type.Literal("utf8"), Type.Literal("base64")], { + description: + "How inline content is decoded. Defaults to utf8. Ignored for sourcePath.", + }), + ), + name: Type.Optional( + Type.String({ description: "Catalog display name; defaults to path." }), + ), + contentType: Type.Optional( + Type.String({ description: "MIME type recorded in the catalog meta." }), + ), + }), + async execute( + _toolCallId: string, + args: { + path: string; + content?: string; + sourcePath?: string; + encoding?: "utf8" | "base64"; + name?: string; + contentType?: string; + }, + ) { + const hasContent = args.content !== undefined; + const hasSource = args.sourcePath !== undefined && args.sourcePath !== ""; + if (hasContent === hasSource) { + throw new Error("Provide exactly one of 'content' or 'sourcePath'."); + } + + let content = args.content ?? ""; + let encoding: "utf8" | "base64" = args.encoding ?? "utf8"; + let defaultName = args.path; + if (hasSource) { + const absolute = resolve(process.cwd(), args.sourcePath as string); + const bytes = await readFile(absolute); + if (bytes.byteLength > MAX_FILE_BYTES) { + throw new Error( + `File is ${bytes.byteLength} bytes; the session resource limit is 1 MiB.`, + ); + } + content = bytes.toString("base64"); + encoding = "base64"; + defaultName = basename(absolute); + } + + const input = { + kind: "file", + path: args.path, + content, + encoding, + name: args.name ?? defaultName, + ...(args.contentType + ? { meta: { contentType: args.contentType } } + : {}), + }; + const resource = (await callHostApi("put", { + sessionId: SESSION_ID, + agentId: AGENT_ID, + input, + })) as { uri: string }; + return textResult(`Added session resource ${resource.uri}`); + }, + }); + + pi.registerTool({ + name: "list_session_resources", + label: "List Session Resources", + description: + "List the files and resources catalogued in the Tangent session, " + + "including those contributed by Prime and agents on other environments.", + promptSnippet: "List the Tangent session's resources", + parameters: Type.Object({}), + async execute() { + const resources = (await callHostApi("list", { + sessionId: SESSION_ID, + })) as Array<{ name: string; uri: string }>; + return textResult(`## Session resources\n\n${renderList(resources)}`); + }, + }); + + pi.registerTool({ + name: "read_session_resource", + label: "Read Session Resource", + description: + "Read a Tangent session resource's contents by its uri (e.g. " + + "'uploads/report.csv'), including content another environment contributed.", + promptSnippet: "Read a Tangent session resource", + parameters: Type.Object({ + uri: Type.String({ description: "Session-root-relative resource uri." }), + }), + async execute(_toolCallId: string, args: { uri: string }) { + const result = (await callHostApi("read", { + sessionId: SESSION_ID, + uri: args.uri, + })) as { content: string; contentType: string }; + const text = Buffer.from(result.content, "base64").toString("utf8"); + return textResult(text); + }, + }); +} diff --git a/packages/local-agent-host/src/bridge.test.ts b/packages/local-agent-host/src/bridge.test.ts new file mode 100644 index 00000000..064295c6 --- /dev/null +++ b/packages/local-agent-host/src/bridge.test.ts @@ -0,0 +1,167 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; + +import type { + connectRemoteEnvironment, + RemoteEnvironmentClient, +} from "@tangent/remote-subagent"; + +import { type BridgeOptions, LocalAgentBridge } from "./bridge.ts"; +import { createFileTools } from "./remoteFileTools.ts"; + +type ConnectConfig = Parameters[0]; + +class FakeSocket { + connected = false; + private readonly handlers = new Map< + string, + Array<(...args: unknown[]) => void> + >(); + + on(event: string, handler: (...args: unknown[]) => void): this { + const list = this.handlers.get(event) ?? []; + list.push(handler); + this.handlers.set(event, list); + return this; + } + + emit(): this { + return this; + } + + fire(event: string, ...args: unknown[]): void { + for (const handler of this.handlers.get(event) ?? []) handler(...args); + } +} + +interface FakeClient extends RemoteEnvironmentClient { + socket: RemoteEnvironmentClient["socket"]; + disconnected: boolean; +} + +function fakeClient(socket: FakeSocket): FakeClient { + const client = { + socket: socket as unknown as RemoteEnvironmentClient["socket"], + disconnected: false, + agentEvent() {}, + subagentUpdate() {}, + report() {}, + readRoom: async () => [], + addResource: async () => { + throw new Error("not implemented"); + }, + listResources: async () => [], + readResource: async () => ({ content: "", contentType: "" }), + disconnect() { + socket.connected = false; + client.disconnected = true; + }, + }; + return client; +} + +function buildBridge( + selector = "projectId:project-a", + metadata?: Record, +): { + bridge: LocalAgentBridge; + configs: ConnectConfig[]; + clients: FakeClient[]; +} { + const configs: ConnectConfig[] = []; + const clients: FakeClient[] = []; + const options: BridgeOptions = { + url: "http://shell.test", + selector, + ...(metadata ? { metadata } : {}), + tools: createFileTools(() => "/workspace"), + piLaunchOptions: async () => ({ + bin: "pi", + proxyExtensionPath: "/dev/null", + }), + log() {}, + onChange() {}, + onSpawned() {}, + }; + const bridge = new LocalAgentBridge(options, (config) => { + configs.push(config); + const client = fakeClient(new FakeSocket()); + clients.push(client); + return client; + }); + return { bridge, configs, clients }; +} + +test("connect wires the scoped token and environment into the SDK", () => { + const { bridge, configs } = buildBridge("projectId:project-a"); + + bridge.connect("scoped-token", "vscode-123"); + + assert.equal(configs.length, 1); + const config = configs[0]; + assert.equal(config.url, "http://shell.test"); + assert.equal(config.token, "scoped-token"); + assert.equal(config.environmentId, "vscode-123"); + assert.equal(config.sessionId, undefined); + assert.deepEqual(Object.keys(config.tools ?? {}).sort(), [ + "find_files", + "list_dir", + "read_file", + "search_text", + ]); + assert.equal(bridge.environmentId, "vscode-123"); + assert.equal(bridge.state, "connecting"); +}); + +test("connect forwards the host self-description metadata to the SDK", () => { + const metadata = { + description: "Tangent Local Agent in /workspace", + capabilities: "edits files", + }; + const { bridge, configs } = buildBridge("projectId:project-a", metadata); + + bridge.connect("scoped-token", "vscode-123"); + + assert.deepEqual(configs[0].metadata, metadata); +}); + +test("connect omits metadata when none is advertised", () => { + const { bridge, configs } = buildBridge(); + + bridge.connect("scoped-token", "vscode-123"); + + assert.equal("metadata" in configs[0], false); +}); + +test("the bridge reports connected once the socket connects", () => { + const { bridge, clients } = buildBridge(); + + bridge.connect("scoped-token", "vscode-123"); + const socket = clients[0].socket as unknown as FakeSocket; + socket.connected = true; + socket.fire("connect"); + + assert.equal(bridge.state, "connected"); +}); + +test("a token refresh reconnects and disconnects the stale client", () => { + const { bridge, configs, clients } = buildBridge(); + + bridge.connect("token-1", "vscode-123"); + bridge.connect("token-2", "vscode-123"); + + assert.equal(configs.length, 2); + assert.equal(configs[1].token, "token-2"); + assert.equal(clients[0].disconnected, true); + assert.equal(clients[1].disconnected, false); +}); + +test("dispose closes the connection and reports disconnected", () => { + const { bridge, clients } = buildBridge(); + + bridge.connect("scoped-token", "vscode-123"); + bridge.dispose(); + + assert.equal(clients[0].disconnected, true); + assert.equal(bridge.state, "disconnected"); +}); diff --git a/packages/local-agent-host/src/bridge.ts b/packages/local-agent-host/src/bridge.ts new file mode 100644 index 00000000..7962be17 --- /dev/null +++ b/packages/local-agent-host/src/bridge.ts @@ -0,0 +1,293 @@ +import { + connectRemoteEnvironment, + type RemoteEnvironmentClient, + type RemoteKillCommand, + type RemoteMessageCommand, + type RemoteResourceInput, + type RemoteSpawnCommand, + type RemoteToolMap, +} from "@tangent/remote-subagent"; +import type { RunId } from "@tangent/shared/contracts.ts"; + +import { type HostApi, type HostApiRoutes, startHostApi } from "./hostApi.ts"; +import { PiAgent, type PiLaunchOptions } from "./piAdapter.ts"; + +export type BridgeState = "connecting" | "connected" | "disconnected"; + +export interface BridgeOptions { + url: string; + /** The project selector this host is scoped to, e.g. `projectId:`. */ + selector: string; + /** + * Self-description advertised in the handshake (e.g. `description`, + * `capabilities`, `editor`, `workspacePath`) so Prime can tell this host apart + * from others and match work to it. Free string->string pairs. + */ + metadata?: Record; + /** RPC tools Prime can call on this window without spawning a sub-agent. */ + tools?: RemoteToolMap; + /** Resolved per spawn so setting changes apply without reconnecting. */ + piLaunchOptions(): Promise; + log(line: string): void; + /** Fires on any connection or roster change, for the status bar. */ + onChange(): void; + /** Fires when Prime starts a local sub-agent, so the user can see and stop it. */ + onSpawned(agent: HostedAgent): void; +} + +/** A sub-agent this window hosts. */ +export interface HostedAgent { + sessionId: string; + agentId: string; + name: string; + /** The Run of the latest message, echoed on every event it produces. */ + runId?: RunId; + /** Set once the server asked for the kill, so the exit is not an error. */ + stopping: boolean; + agent?: PiAgent; +} + +/** + * Hosts Tangent sub-agents as local pi processes. Connects to the Shell's + * `/remote-env` namespace and maps each spawn/message/kill command onto a + * {@link PiAgent}, streaming its events back under the server-assigned id. + */ +export class LocalAgentBridge { + environmentId: string | undefined; + private readonly options: BridgeOptions; + private readonly connectFactory: typeof connectRemoteEnvironment; + private readonly startApi: typeof startHostApi; + private client: RemoteEnvironmentClient | undefined; + private readonly agents = new Map(); + private connectionState: BridgeState = "disconnected"; + private connectionError: string | undefined; + /** Memoized loopback API the hosted sub-agents' resource extension calls. */ + private hostApi: Promise | undefined; + + constructor( + options: BridgeOptions, + connect: typeof connectRemoteEnvironment = connectRemoteEnvironment, + startApi: typeof startHostApi = startHostApi, + ) { + this.options = options; + this.connectFactory = connect; + this.startApi = startApi; + } + + get state(): BridgeState { + return this.connectionState; + } + + get lastError(): string | undefined { + return this.connectionError; + } + + listAgents(): HostedAgent[] { + return [...this.agents.values()]; + } + + /** + * (Re)connects with a freshly minted scoped token, keeping hosted sub-agents + * so a token refresh does not interrupt them. The SDK (re)registers the tool + * catalog on every connect; the server binds it to the token's selector, so it + * is offered to every session in the project rather than a single one. + */ + connect(token: string, environmentId: string): void { + this.client?.disconnect(); + this.environmentId = environmentId; + this.setState("connecting"); + this.client = this.connectFactory({ + url: this.options.url, + token, + environmentId, + ...(this.options.metadata ? { metadata: this.options.metadata } : {}), + tools: this.options.tools, + handlers: { + onSpawn: (command) => this.onSpawn(command), + onMessage: (command) => this.onMessage(command), + onKill: (command) => this.onKill(command), + }, + }); + this.watchConnection(this.client); + } + + /** Stops every local sub-agent and closes the connection. */ + dispose(): void { + for (const hosted of this.agents.values()) { + hosted.stopping = true; + hosted.agent?.kill(); + } + this.agents.clear(); + this.client?.disconnect(); + this.client = undefined; + void this.hostApi?.then((api) => api.close()); + this.hostApi = undefined; + this.setState("disconnected"); + } + + /** The loopback routes a hosted sub-agent's resource extension calls, each + * forwarding to the live connection; a dropped connection fails clearly. */ + private hostApiRoutes(): HostApiRoutes { + const requireClient = (): RemoteEnvironmentClient => { + if (!this.client) throw new Error("Not connected to the Shell."); + return this.client; + }; + return { + put: async (body) => { + const { sessionId, agentId, input } = body as { + sessionId: string; + agentId: string; + input: RemoteResourceInput; + }; + return requireClient().addResource(sessionId, agentId, input); + }, + list: async (body) => { + const { sessionId } = body as { sessionId: string }; + return requireClient().listResources(sessionId); + }, + read: async (body) => { + const { sessionId, uri } = body as { sessionId: string; uri: string }; + return requireClient().readResource(sessionId, uri); + }, + }; + } + + /** Starts the loopback API on first use, memoized for the host's lifetime. */ + private ensureHostApi(): Promise { + this.hostApi ??= this.startApi(this.hostApiRoutes()); + return this.hostApi; + } + + private watchConnection(client: RemoteEnvironmentClient): void { + const { socket } = client; + socket.on("connect", () => { + this.connectionError = undefined; + this.setState("connected"); + this.options.log( + `connected to ${this.options.url} as ${this.environmentId} (selector ${this.options.selector})`, + ); + this.reclaimAgents(client); + }); + socket.on("disconnect", (reason) => { + this.options.log(`disconnected: ${reason}`); + this.setState("connecting"); + }); + socket.on("connect_error", (err) => { + this.connectionError = err.message; + this.options.log(`connection failed: ${err.message}`); + this.setState("connecting"); + }); + } + + /** + * A dropped socket leaves this host's sub-agents `detached` on the server; + * declaring them active again is how the gateway learns they survived. + */ + private reclaimAgents(client: RemoteEnvironmentClient): void { + for (const hosted of this.agents.values()) { + client.subagentUpdate(hosted.sessionId, hosted.agentId, "active"); + } + } + + private setState(state: BridgeState): void { + this.connectionState = state; + this.options.onChange(); + } + + private async onSpawn(command: RemoteSpawnCommand): Promise { + const hosted: HostedAgent = { + sessionId: command.sessionId, + agentId: command.agentId, + name: command.name, + stopping: false, + }; + this.agents.set(command.agentId, hosted); + this.options.onChange(); + + const base = await this.options.piLaunchOptions(); + if (!base.cwd) { + this.fail( + hosted, + "No workspace folder is open to host local sub-agents.", + ); + return; + } + const hostApi = await this.ensureHostApi(); + const launch = { + ...base, + hostApiUrl: hostApi.url, + hostApiToken: hostApi.token, + }; + this.options.log( + `spawning "${command.name}" (${command.agentId}) in ${launch.cwd}`, + ); + hosted.agent = new PiAgent(command, launch, { + onEvent: (event) => + this.client?.agentEvent( + hosted.sessionId, + hosted.agentId, + event, + hosted.runId, + ), + onExit: (reason) => this.onAgentExit(hosted, reason), + onLog: (line) => this.options.log(`[${hosted.name}] ${line}`), + }); + this.options.onSpawned(hosted); + } + + /** Stops a local sub-agent from this side and tells the server it was killed. */ + stopAgent(agentId: string): void { + const hosted = this.agents.get(agentId); + if (!hosted) return; + this.onKill({ + sessionId: hosted.sessionId, + agentId, + completed: false, + }); + } + + private onMessage(command: RemoteMessageCommand): void { + const hosted = this.agents.get(command.agentId); + if (!hosted?.agent) { + this.options.log(`message for unknown sub-agent ${command.agentId}`); + return; + } + hosted.runId = command.runId; + hosted.agent.prompt(command.text, command.delivery); + } + + private onKill(command: RemoteKillCommand): void { + const hosted = this.agents.get(command.agentId); + if (!hosted) return; + hosted.stopping = true; + hosted.agent?.kill(); + this.agents.delete(command.agentId); + this.client?.subagentUpdate( + command.sessionId, + command.agentId, + command.completed ? "completed" : "killed", + ); + this.options.log(`stopped "${hosted.name}"`); + this.options.onChange(); + } + + private onAgentExit(hosted: HostedAgent, reason: string): void { + this.options.log(`[${hosted.name}] pi ${reason}`); + if (hosted.stopping || this.agents.get(hosted.agentId) !== hosted) return; + this.fail(hosted, `Local pi ${reason}.`); + } + + /** Reports a sub-agent that cannot run, and drops it from the roster. */ + private fail(hosted: HostedAgent, message: string): void { + this.agents.delete(hosted.agentId); + this.client?.agentEvent( + hosted.sessionId, + hosted.agentId, + { type: "error", message }, + hosted.runId, + ); + this.client?.subagentUpdate(hosted.sessionId, hosted.agentId, "error"); + this.options.log(`[${hosted.name}] ${message}`); + this.options.onChange(); + } +} diff --git a/packages/local-agent-host/src/fileToolOps.ts b/packages/local-agent-host/src/fileToolOps.ts new file mode 100644 index 00000000..126bf49e --- /dev/null +++ b/packages/local-agent-host/src/fileToolOps.ts @@ -0,0 +1,153 @@ +import { open, readdir, readFile, stat } from "node:fs/promises"; +import path from "node:path"; + +import { + boundedInt, + optionalString, + requiredString, + type ToolArgs, +} from "./toolArgs.ts"; +import { + globMatcher, + IGNORED_DIRS, + isBinary, + isDeniedReadPath, + resolveExisting, + walkFiles, +} from "./workspaceFs.ts"; + +export const DEFAULT_READ_BYTES = 256 * 1024; +export const MAX_READ_BYTES = 1024 * 1024; +export const DEFAULT_MAX_RESULTS = 200; +export const MAX_RESULTS = 1000; +/** Larger files are skipped by `search_text` rather than read whole. */ +const MAX_SEARCH_FILE_BYTES = 1024 * 1024; +const MAX_MATCH_LINE_CHARS = 300; + +function resultLimit(args: ToolArgs): number { + return boundedInt(args, "maxResults", DEFAULT_MAX_RESULTS, MAX_RESULTS); +} + +function limitNote(count: number, limit: number): string { + return count >= limit ? `\n… stopped at ${limit} results` : ""; +} + +export async function readTextFile( + root: string, + args: ToolArgs, +): Promise { + const relative = requiredString(args, "path"); + if (isDeniedReadPath(relative)) { + throw new Error(`"${relative}" is not readable.`); + } + const maxBytes = boundedInt( + args, + "maxBytes", + DEFAULT_READ_BYTES, + MAX_READ_BYTES, + ); + const file = await resolveExisting(root, relative); + const { size } = await stat(file); + const handle = await open(file, "r"); + try { + const buffer = Buffer.alloc(Math.min(size, maxBytes)); + const { bytesRead } = await handle.read(buffer, 0, buffer.length, 0); + const content = buffer.subarray(0, bytesRead); + if (isBinary(content)) throw new Error(`"${relative}" is a binary file.`); + const text = content.toString("utf8"); + if (size <= maxBytes) return text; + return `${text}\n… truncated (showing ${bytesRead} of ${size} bytes)`; + } finally { + await handle.close(); + } +} + +export async function listDirectory( + root: string, + args: ToolArgs, +): Promise { + const relative = optionalString(args, "path") ?? "."; + const dir = await resolveExisting(root, relative); + const entries = await readdir(dir, { withFileTypes: true }); + const lines = entries + .filter((entry) => !IGNORED_DIRS.has(entry.name)) + .sort((a, b) => a.name.localeCompare(b.name)) + .map((entry) => (entry.isDirectory() ? `${entry.name}/` : entry.name)); + return lines.length > 0 ? lines.join("\n") : "(empty directory)"; +} + +export async function findFiles(root: string, args: ToolArgs): Promise { + const matches = globMatcher(requiredString(args, "glob")); + const limit = resultLimit(args); + const found: string[] = []; + for await (const file of walkFiles(root)) { + if (!matches(file)) continue; + found.push(file); + if (found.length >= limit) break; + } + if (found.length === 0) return "No files matched."; + return found.join("\n") + limitNote(found.length, limit); +} + +function searchPattern(args: ToolArgs): RegExp { + const query = requiredString(args, "query"); + const source = + args.regex === true ? query : query.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"); + return new RegExp(source, args.ignoreCase === true ? "i" : ""); +} + +/** + * A file's text for searching, or `undefined` when it is too large, binary, or + * no longer readable — a file removed or permission-changed between the walk + * and the read skips rather than failing the whole search. + */ +async function searchableText(full: string): Promise { + try { + if ((await stat(full)).size > MAX_SEARCH_FILE_BYTES) return undefined; + const buffer = await readFile(full); + return isBinary(buffer) ? undefined : buffer.toString("utf8"); + } catch { + return undefined; + } +} + +/** Matching lines of one file as `path:line: text`, at most `limit`. */ +async function searchFile( + root: string, + file: string, + pattern: RegExp, + limit: number, +): Promise { + const text = await searchableText(path.join(root, file)); + if (text === undefined) return []; + const hits: string[] = []; + const lines = text.split("\n"); + for (let i = 0; i < lines.length && hits.length < limit; i++) { + const line = lines[i] ?? ""; + if (!pattern.test(line)) continue; + hits.push( + `${file}:${i + 1}: ${line.trim().slice(0, MAX_MATCH_LINE_CHARS)}`, + ); + } + return hits; +} + +export async function searchText( + root: string, + args: ToolArgs, +): Promise { + const pattern = searchPattern(args); + const glob = optionalString(args, "glob"); + const matches = glob ? globMatcher(glob) : () => true; + const limit = resultLimit(args); + const eligible = (file: string): boolean => + !isDeniedReadPath(file) && matches(file); + const hits: string[] = []; + for await (const file of walkFiles(root)) { + if (hits.length >= limit) break; + if (!eligible(file)) continue; + hits.push(...(await searchFile(root, file, pattern, limit - hits.length))); + } + if (hits.length === 0) return "No matches."; + return hits.join("\n") + limitNote(hits.length, limit); +} diff --git a/packages/local-agent-host/src/hostApi.test.ts b/packages/local-agent-host/src/hostApi.test.ts new file mode 100644 index 00000000..fa64c96b --- /dev/null +++ b/packages/local-agent-host/src/hostApi.test.ts @@ -0,0 +1,87 @@ +import assert from "node:assert/strict"; +import { after, test } from "node:test"; + +import { type HostApi, type HostApiRoutes, startHostApi } from "./hostApi.ts"; + +const started: HostApi[] = []; +after(async () => { + for (const api of started) await api.close(); +}); + +async function serve(routes: Partial): Promise { + const api = await startHostApi({ + put: async () => ({}), + list: async () => [], + read: async () => ({}), + ...routes, + }); + started.push(api); + return api; +} + +async function post( + api: HostApi, + route: string, + body: unknown, + token = api.token, +): Promise<{ status: number; json: Record }> { + const res = await fetch(`${api.url}/${route}`, { + method: "POST", + headers: { + "content-type": "application/json", + ...(token ? { authorization: `Bearer ${token}` } : {}), + }, + body: JSON.stringify(body), + }); + const text = await res.text(); + return { + status: res.status, + json: text ? (JSON.parse(text) as Record) : {}, + }; +} + +test("forwards a put body to the route and acks its result", async () => { + const seen: unknown[] = []; + const api = await serve({ + put: async (body) => { + seen.push(body); + return { uri: "uploads/x.txt" }; + }, + }); + + const res = await post(api, "put", { sessionId: "s", input: { a: 1 } }); + + assert.equal(res.status, 200); + assert.deepEqual(res.json, { ok: true, result: { uri: "uploads/x.txt" } }); + assert.deepEqual(seen, [{ sessionId: "s", input: { a: 1 } }]); +}); + +test("rejects a request without the bearer token", async () => { + const api = await serve({}); + + const res = await post(api, "put", {}, ""); + + assert.equal(res.status, 401); + assert.equal(res.json.ok, false); +}); + +test("404s an unknown route", async () => { + const api = await serve({}); + + const res = await post(api, "nope", {}); + + assert.equal(res.status, 404); +}); + +test("wraps a thrown route error as a 400 ack", async () => { + const api = await serve({ + read: async () => { + throw new Error("boom"); + }, + }); + + const res = await post(api, "read", { uri: "x" }); + + assert.equal(res.status, 400); + assert.deepEqual(res.json, { ok: false, error: "boom" }); +}); diff --git a/packages/local-agent-host/src/hostApi.ts b/packages/local-agent-host/src/hostApi.ts new file mode 100644 index 00000000..e208398c --- /dev/null +++ b/packages/local-agent-host/src/hostApi.ts @@ -0,0 +1,113 @@ +import { randomBytes } from "node:crypto"; +import { + createServer, + type IncomingMessage, + type Server, + type ServerResponse, +} from "node:http"; +import type { AddressInfo } from "node:net"; + +/** + * One loopback route: receives the parsed JSON body a pi extension posted and + * returns the JSON body to send back. A throw becomes a `{ ok: false, error }` + * response, so a route never has to format its own failure. + */ +export type HostApiRoute = (body: unknown) => Promise; + +/** The resource routes the session-resources pi extension calls. */ +export interface HostApiRoutes { + put: HostApiRoute; + list: HostApiRoute; + read: HostApiRoute; +} + +/** A running loopback API: its base url, bearer token, and a stop handle. */ +export interface HostApi { + url: string; + token: string; + close(): Promise; +} + +/** Caps a posted body so a runaway extension cannot exhaust host memory. */ +const MAX_BODY_BYTES = 8 * 1024 * 1024; + +/** Reads a request's JSON body, rejecting an oversized or malformed payload. */ +async function readBody(req: IncomingMessage): Promise { + const chunks: Buffer[] = []; + let size = 0; + for await (const chunk of req) { + size += (chunk as Buffer).length; + if (size > MAX_BODY_BYTES) throw new Error("Request body too large"); + chunks.push(chunk as Buffer); + } + const raw = Buffer.concat(chunks).toString("utf8"); + return raw ? JSON.parse(raw) : {}; +} + +/** Sends a JSON response with the given status. */ +function send(res: ServerResponse, status: number, body: unknown): void { + res.writeHead(status, { "content-type": "application/json" }); + res.end(JSON.stringify(body)); +} + +/** Whether the request carries the expected `Bearer ` authorization. */ +function authorized(req: IncomingMessage, token: string): boolean { + return req.headers.authorization === `Bearer ${token}`; +} + +/** The route a request targets (`POST /put|/list|/read`), or undefined. */ +function routeFor( + routes: HostApiRoutes, + req: IncomingMessage, +): HostApiRoute | undefined { + if (req.method !== "POST") return undefined; + if (req.url === "/put") return routes.put; + if (req.url === "/list") return routes.list; + if (req.url === "/read") return routes.read; + return undefined; +} + +/** Routes one request: authorize, parse, run, and ack — all errors as 4xx JSON. */ +async function dispatch( + routes: HostApiRoutes, + token: string, + req: IncomingMessage, + res: ServerResponse, +): Promise { + const route = routeFor(routes, req); + if (!route) return send(res, 404, { ok: false, error: "Not found" }); + if (!authorized(req, token)) { + return send(res, 401, { ok: false, error: "Unauthorized" }); + } + try { + const result = await route(await readBody(req)); + send(res, 200, { ok: true, result }); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + send(res, 400, { ok: false, error: message }); + } +} + +/** + * Starts a loopback HTTP API on `127.0.0.1` the host's session-resource pi + * extension calls back into. Bound to an ephemeral port and guarded by a random + * per-process bearer token, so only this host's own sub-agents can reach it. + * Resolves once listening, with the base url and token to inject into pi's env. + */ +export function startHostApi(routes: HostApiRoutes): Promise { + const token = randomBytes(24).toString("hex"); + const server: Server = createServer((req, res) => { + void dispatch(routes, token, req, res); + }); + return new Promise((resolve, reject) => { + server.on("error", reject); + server.listen(0, "127.0.0.1", () => { + const { port } = server.address() as AddressInfo; + resolve({ + url: `http://127.0.0.1:${port}`, + token, + close: () => new Promise((done) => server.close(() => done())), + }); + }); + }); +} diff --git a/packages/local-agent-host/src/piAdapter.test.ts b/packages/local-agent-host/src/piAdapter.test.ts new file mode 100644 index 00000000..9575fd0f --- /dev/null +++ b/packages/local-agent-host/src/piAdapter.test.ts @@ -0,0 +1,74 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; + +import type { RemoteSpawnCommand } from "@tangent/shared/remoteSubagent.ts"; + +import { buildPiArgs, piEnv, type PiLaunchOptions } from "./piAdapter.ts"; + +const command: RemoteSpawnCommand = { + sessionId: "session-1", + agentId: "agent-1", + name: "worker", + tools: [], + systemPrompt: "", + autoRelayToPrime: true, +}; + +const baseLaunch: PiLaunchOptions = { + bin: "pi", + cwd: "/workspace", + proxyExtensionPath: "/ext/proxy.ts", + sessionResourceExtensionPath: "/ext/sessionResources.ts", +}; + +test("buildPiArgs loads the session-resources extension when the host API is set", () => { + const args = buildPiArgs(command, { + ...baseLaunch, + hostApiUrl: "http://127.0.0.1:1234", + }); + + const idx = args.indexOf("--extension"); + assert.notEqual(idx, -1); + assert.ok(args.includes("/ext/sessionResources.ts")); +}); + +test("buildPiArgs omits the extension when no host API is set", () => { + const args = buildPiArgs(command, baseLaunch); + + assert.equal(args.includes("/ext/sessionResources.ts"), false); +}); + +test("buildPiArgs grants pi's full default toolset (no --tools allowlist)", () => { + const args = buildPiArgs(command, { + ...baseLaunch, + hostApiUrl: "http://127.0.0.1:1234", + }); + + assert.equal(args.includes("--tools"), false); +}); + +test("buildPiArgs does not disable skill discovery", () => { + const args = buildPiArgs(command, baseLaunch); + + assert.equal(args.includes("--no-skills"), false); +}); + +test("piEnv injects the host API and session/agent ids when the API is set", () => { + const env = piEnv(command, { + ...baseLaunch, + hostApiUrl: "http://127.0.0.1:1234", + hostApiToken: "secret", + }); + + assert.equal(env.TANGENT_HOST_API_URL, "http://127.0.0.1:1234"); + assert.equal(env.TANGENT_HOST_API_TOKEN, "secret"); + assert.equal(env.TANGENT_SESSION_ID, "session-1"); + assert.equal(env.TANGENT_AGENT_ID, "agent-1"); +}); + +test("piEnv omits the host API env when no API is set", () => { + const env = piEnv(command, baseLaunch); + + assert.equal("TANGENT_HOST_API_URL" in env, false); + assert.equal("TANGENT_SESSION_ID" in env, false); +}); diff --git a/packages/local-agent-host/src/piAdapter.ts b/packages/local-agent-host/src/piAdapter.ts new file mode 100644 index 00000000..599c1e38 --- /dev/null +++ b/packages/local-agent-host/src/piAdapter.ts @@ -0,0 +1,189 @@ +import { type ChildProcessWithoutNullStreams, spawn } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { StringDecoder } from "node:string_decoder"; + +import type { MessageDelivery } from "@tangent/shared/contracts.ts"; +import type { + RemoteAgentEvent, + RemoteSpawnCommand, +} from "@tangent/shared/remoteSubagent.ts"; + +import { + createStreamState, + parsePiLine, + translatePiEvent, +} from "./piEvents.ts"; + +/** How a local pi process is launched. */ +export interface PiLaunchOptions { + bin: string; + /** Working directory; sub-agents are refused when no folder is open. */ + cwd?: string; + /** Overrides the model Prime asked for. */ + model?: string; + proxyUrl?: string; + proxyApiKey?: string; + /** The bundled proxy-provider extension, loaded only with `proxyUrl`. */ + proxyExtensionPath: string; + /** The bundled session-resources extension, loaded only with `hostApiUrl`. */ + sessionResourceExtensionPath?: string; + /** Loopback host API the session-resources extension calls back into. */ + hostApiUrl?: string; + /** Bearer token guarding {@link hostApiUrl}. */ + hostApiToken?: string; +} + +export interface PiAgentCallbacks { + onEvent(event: RemoteAgentEvent): void; + /** Fires once, whether the process exited or failed to start. */ + onExit(reason: string): void; + onLog(line: string): void; +} + +function modelArgs( + command: RemoteSpawnCommand, + options: PiLaunchOptions, +): string[] { + const args: string[] = []; + const model = options.model ?? command.model; + if (model) args.push("--model", model); + if (command.thinkingDepth) args.push("--thinking", command.thinkingDepth); + return args; +} + +/** + * The `pi --mode rpc` arguments for a spawn. A local host runs on the user's + * own machine, so - unlike the server's sandboxed launcher - it grants pi its + * full default capabilities: no `--tools` allowlist (all built-in, MCP, and + * extension-registered tools stay enabled) and no `--no-skills` (skills, the + * user's installed extensions, prompt templates, themes, and context files all + * auto-discover). The Tangent proxy + session-resource extensions layer on top. + */ +export function buildPiArgs( + command: RemoteSpawnCommand, + options: PiLaunchOptions, +): string[] { + const args = [ + "--mode", + "rpc", + "--no-session", + ...modelArgs(command, options), + ]; + if (command.systemPrompt) { + args.push("--append-system-prompt", command.systemPrompt); + } + if (options.proxyUrl) args.push("--extension", options.proxyExtensionPath); + if (options.hostApiUrl && options.sessionResourceExtensionPath) { + args.push("--extension", options.sessionResourceExtensionPath); + } + return args; +} + +export function piEnv( + command: RemoteSpawnCommand, + options: PiLaunchOptions, +): NodeJS.ProcessEnv { + const env: NodeJS.ProcessEnv = { ...process.env }; + if (options.proxyUrl) env.PI_PROXY_URL = options.proxyUrl; + if (options.proxyApiKey) env.PI_PROXY_API_KEY = options.proxyApiKey; + if (options.hostApiUrl) { + env.TANGENT_HOST_API_URL = options.hostApiUrl; + env.TANGENT_HOST_API_TOKEN = options.hostApiToken ?? ""; + env.TANGENT_SESSION_ID = command.sessionId; + env.TANGENT_AGENT_ID = command.agentId; + } + return env; +} + +/** Splits a stream into LF-delimited lines (not `readline`, which also splits on U+2028). */ +function onLines( + stream: NodeJS.ReadableStream, + onLine: (line: string) => void, +): void { + const decoder = new StringDecoder("utf8"); + let buffer = ""; + stream.on("data", (chunk: Buffer) => { + buffer += decoder.write(chunk); + const lines = buffer.split("\n"); + buffer = lines.pop() ?? ""; + for (const line of lines) { + const trimmed = line.replace(/\r$/, ""); + if (trimmed) onLine(trimmed); + } + }); +} + +/** One long-lived local `pi --mode rpc` child driven over stdin/stdout JSONL. */ +export class PiAgent { + private readonly child: ChildProcessWithoutNullStreams; + private readonly callbacks: PiAgentCallbacks; + private readonly state = createStreamState(); + private busy = false; + private exited = false; + + constructor( + command: RemoteSpawnCommand, + options: PiLaunchOptions, + callbacks: PiAgentCallbacks, + ) { + this.callbacks = callbacks; + this.child = spawn(options.bin, buildPiArgs(command, options), { + cwd: options.cwd, + env: piEnv(command, options), + }); + onLines(this.child.stdout, (line) => this.handleLine(line)); + onLines(this.child.stderr, (line) => callbacks.onLog(`stderr ${line}`)); + this.child.stdin.on("error", (err) => + callbacks.onLog(`stdin ${err.message}`), + ); + this.child.on("error", (err) => this.finish(`failed: ${err.message}`)); + this.child.on("exit", (code, signal) => + this.finish(`exited (${signal ?? `code ${code}`})`), + ); + } + + /** Sends a prompt; mid-run it steers or queues according to `delivery`. */ + prompt(text: string, delivery: MessageDelivery): void { + const streamingBehavior = delivery === "steer" ? "steer" : "followUp"; + this.write({ + id: randomUUID(), + type: "prompt", + message: text, + ...(this.busy ? { streamingBehavior } : {}), + }); + this.busy = true; + } + + kill(): void { + if (!this.exited) this.child.kill("SIGTERM"); + } + + private write(command: Record): void { + if (this.exited) return; + this.child.stdin.write(`${JSON.stringify(command)}\n`); + } + + private handleLine(line: string): void { + const event = parsePiLine(line); + if (!event) { + this.callbacks.onLog(`unparseable ${line}`); + return; + } + if (event.type === "agent_end") this.busy = false; + try { + for (const out of translatePiEvent(this.state, event, randomUUID)) { + this.callbacks.onEvent(out); + } + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + this.callbacks.onLog(`translate failed ${message}`); + } + } + + private finish(reason: string): void { + if (this.exited) return; + this.exited = true; + this.busy = false; + this.callbacks.onExit(reason); + } +} diff --git a/packages/local-agent-host/src/piEvents.test.ts b/packages/local-agent-host/src/piEvents.test.ts new file mode 100644 index 00000000..ed8a6881 --- /dev/null +++ b/packages/local-agent-host/src/piEvents.test.ts @@ -0,0 +1,225 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; + +import type { RemoteSpawnCommand } from "@tangent/shared/remoteSubagent.ts"; + +import { buildPiArgs } from "./piAdapter.ts"; +import { + createStreamState, + parsePiLine, + type PiStdoutEvent, + translatePiEvent, +} from "./piEvents.ts"; + +const newId = () => "msg-1"; + +function run(events: PiStdoutEvent[]) { + const state = createStreamState(); + return events.flatMap((event) => translatePiEvent(state, event, newId)); +} + +const assistant = (text: string) => ({ + role: "assistant", + content: [{ type: "text", text }], +}); + +test("streams one assistant message as start, deltas, then end", () => { + const out = run([ + { type: "agent_start" }, + { type: "message_start", message: { role: "assistant" } }, + { + type: "message_update", + assistantMessageEvent: { type: "thinking_delta", delta: "hmm" }, + }, + { + type: "message_update", + assistantMessageEvent: { type: "text_delta", delta: "Hel" }, + }, + { + type: "message_update", + assistantMessageEvent: { type: "text_delta", delta: "lo" }, + }, + { type: "message_end", message: assistant("Hello") }, + { type: "agent_end" }, + ]); + + assert.deepEqual(out, [ + { type: "activity", activity: { kind: "thinking", label: "Thinking..." } }, + { type: "start", messageId: "msg-1" }, + { type: "activity", activity: null }, + { type: "thinking", messageId: "msg-1", delta: "hmm" }, + { type: "delta", messageId: "msg-1", delta: "Hel" }, + { type: "delta", messageId: "msg-1", delta: "lo" }, + { type: "end", messageId: "msg-1", content: "Hello", thinking: "hmm" }, + { type: "activity", activity: { kind: "thinking", label: "Thinking..." } }, + { type: "activity", activity: null }, + ]); +}); + +test("a malformed content array does not crash the translation", () => { + const out = run([ + { type: "message_start", message: { role: "assistant" } }, + { + type: "message_update", + assistantMessageEvent: { type: "text_delta", delta: "hi" }, + }, + { + type: "message_end", + message: { + role: "assistant", + content: [null, { type: "text", text: "ok" }, { type: "image" }], + } as unknown as PiStdoutEvent["message"], + }, + ]); + + const end = out.find((event) => event.type === "end"); + assert.deepEqual(end, { + type: "end", + messageId: "msg-1", + content: "ok", + thinking: "", + }); +}); + +test("a non-object stdout line parses to null", () => { + assert.equal(parsePiLine("[1,2,3]"), null); + assert.equal(parsePiLine('"a string"'), null); + assert.equal(parsePiLine("42"), null); +}); + +test("a tool-only assistant message opens no bubble", () => { + const out = run([ + { type: "message_start", message: { role: "assistant" } }, + { type: "message_end", message: { role: "assistant", content: [] } }, + ]); + + assert.equal( + out.some((event) => event.type === "start" || event.type === "end"), + false, + ); +}); + +test("falls back to streamed text when the final message has none", () => { + const out = run([ + { type: "message_start", message: { role: "assistant" } }, + { + type: "message_update", + assistantMessageEvent: { type: "text_delta", delta: "partial" }, + }, + { type: "message_end", message: { role: "assistant", content: [] } }, + ]); + + const end = out.find((event) => event.type === "end"); + assert.deepEqual(end, { + type: "end", + messageId: "msg-1", + content: "partial", + thinking: "", + }); +}); + +test("ignores user and tool-result messages", () => { + const out = run([ + { type: "message_start", message: { role: "user" } }, + { + type: "message_update", + assistantMessageEvent: { type: "text_delta", delta: "x" }, + }, + { type: "message_end", message: { role: "user" } }, + ]); + + assert.deepEqual(out, []); +}); + +test("labels a running tool with its target", () => { + const out = run([ + { + type: "tool_execution_start", + toolName: "bash", + args: { command: "pnpm test" }, + }, + { type: "tool_execution_start", toolName: "custom_tool", args: {} }, + ]); + + assert.deepEqual(out, [ + { + type: "activity", + activity: { kind: "tool", label: "Running pnpm test", toolName: "bash" }, + }, + { + type: "activity", + activity: { + kind: "tool", + label: "Running custom_tool", + toolName: "custom_tool", + }, + }, + ]); +}); + +test("relays queue updates, dropping non-string entries", () => { + const out = run([ + { type: "queue_update", steering: ["steer", 3], followUp: "nope" }, + ]); + + assert.deepEqual(out, [{ type: "queue", steering: ["steer"], followUp: [] }]); +}); + +test("unknown event types and bad lines yield nothing", () => { + assert.deepEqual(run([{ type: "turn_start" }, {}]), []); + assert.equal(parsePiLine("not json"), null); + assert.equal(parsePiLine("42"), null); + assert.deepEqual(parsePiLine('{"type":"agent_end"}'), { type: "agent_end" }); +}); + +const spawnCommand: RemoteSpawnCommand = { + sessionId: "s1", + agentId: "a1", + name: "worker", + tools: ["read", "bash", "message_prime", "read_room"], + systemPrompt: "Be brief.", + model: "openai/gpt-5", + thinkingDepth: "low", + autoRelayToPrime: true, +}; + +const launch = { bin: "pi", proxyExtensionPath: "/ext/proxyProvider.ts" }; + +test("pi args keep built-in tools, add read access, and pass the resolved config", () => { + assert.deepEqual(buildPiArgs(spawnCommand, launch), [ + "--mode", + "rpc", + "--no-session", + "--no-skills", + "--model", + "openai/gpt-5", + "--thinking", + "low", + "--tools", + "read,bash,grep,find,ls", + "--append-system-prompt", + "Be brief.", + ]); +}); + +test("pi args honor a local model override, default tools, and the proxy", () => { + const args = buildPiArgs( + { ...spawnCommand, tools: ["message_prime"], thinkingDepth: undefined }, + { ...launch, model: "anthropic/claude", proxyUrl: "https://proxy" }, + ); + + assert.deepEqual(args, [ + "--mode", + "rpc", + "--no-session", + "--no-skills", + "--model", + "anthropic/claude", + "--tools", + "read,write,edit,bash,grep,find,ls", + "--append-system-prompt", + "Be brief.", + "--extension", + "/ext/proxyProvider.ts", + ]); +}); diff --git a/packages/local-agent-host/src/piEvents.ts b/packages/local-agent-host/src/piEvents.ts new file mode 100644 index 00000000..470cac03 --- /dev/null +++ b/packages/local-agent-host/src/piEvents.ts @@ -0,0 +1,214 @@ +import type { AgentActivity } from "@tangent/shared/contracts.ts"; +import type { RemoteAgentEvent } from "@tangent/shared/remoteSubagent.ts"; + +/** + * One parsed line of `pi --mode rpc` stdout. Flat rather than a union because + * `type` comes straight from JSON and may name events we don't model. + */ +export interface PiStdoutEvent { + type?: string; + assistantMessageEvent?: { type?: string; delta?: unknown }; + message?: { role?: string; content?: unknown }; + toolName?: string; + args?: unknown; + steering?: unknown; + followUp?: unknown; +} + +/** The in-flight assistant message of one pi process. */ +export interface PiStreamState { + messageId: string | null; + /** `start` is deferred to the first delta so tool-only messages open no bubble. */ + startEmitted: boolean; + accum: string; + thinkingAccum: string; +} + +export function createStreamState(): PiStreamState { + return { messageId: null, startEmitted: false, accum: "", thinkingAccum: "" }; +} + +function resetMessage(state: PiStreamState): void { + state.messageId = null; + state.startEmitted = false; + state.accum = ""; + state.thinkingAccum = ""; +} + +/** Parses one stdout line, or `null` when it is not a JSON object. */ +export function parsePiLine(line: string): PiStdoutEvent | null { + try { + const parsed: unknown = JSON.parse(line); + if (typeof parsed !== "object" || parsed === null) return null; + if (Array.isArray(parsed)) return null; + return parsed as PiStdoutEvent; + } catch { + return null; + } +} + +const THINKING: AgentActivity = { kind: "thinking", label: "Thinking..." }; + +const activity = (value: AgentActivity | null): RemoteAgentEvent => ({ + type: "activity", + activity: value, +}); + +function isAssistant(event: PiStdoutEvent): boolean { + return event.message?.role === "assistant"; +} + +/** A `{ type: "text", text }` part of an assistant message's content array. */ +function isTextPart(part: unknown): part is { type: "text"; text: string } { + if (typeof part !== "object" || part === null) return false; + const candidate = part as { type?: unknown; text?: unknown }; + return candidate.type === "text" && typeof candidate.text === "string"; +} + +/** Concatenated text parts of a finalized assistant message. */ +function assistantText(message: PiStdoutEvent["message"]): string { + if (!Array.isArray(message?.content)) return ""; + return message.content + .filter(isTextPart) + .map((part) => part.text) + .join(""); +} + +function readDelta( + raw: PiStdoutEvent["assistantMessageEvent"], +): { kind: "delta" | "thinking"; text: string } | null { + if (!raw || typeof raw.delta !== "string") return null; + if (raw.type === "text_delta") return { kind: "delta", text: raw.delta }; + if (raw.type === "thinking_delta") { + return { kind: "thinking", text: raw.delta }; + } + return null; +} + +function stringArg(args: unknown, key: string): string | undefined { + if (typeof args !== "object" || args === null) return undefined; + const value = (args as Record)[key]; + if (typeof value !== "string" || !value.trim()) return undefined; + return value.replace(/\s+/g, " ").trim().slice(0, 80); +} + +const TOOL_VERBS: Record = { + read: ["Reading", "path"], + write: ["Writing", "path"], + edit: ["Editing", "path"], + ls: ["Listing", "path"], + bash: ["Running", "command"], + grep: ["Searching for", "pattern"], + find: ["Finding", "pattern"], +}; + +export function toolActivityLabel(toolName: string, args: unknown): string { + const [verb, argKey] = TOOL_VERBS[toolName] ?? [`Running ${toolName}`, ""]; + const target = argKey ? stringArg(args, argKey) : undefined; + return target ? `${verb} ${target}` : verb; +} + +function toStringArray(value: unknown): string[] { + if (!Array.isArray(value)) return []; + return value.filter((item): item is string => typeof item === "string"); +} + +type Translator = ( + state: PiStreamState, + event: PiStdoutEvent, + newId: () => string, +) => RemoteAgentEvent[]; + +function onMessageStart( + state: PiStreamState, + event: PiStdoutEvent, + newId: () => string, +): RemoteAgentEvent[] { + if (!isAssistant(event)) return []; + resetMessage(state); + state.messageId = newId(); + return []; +} + +function onMessageUpdate( + state: PiStreamState, + event: PiStdoutEvent, +): RemoteAgentEvent[] { + const delta = readDelta(event.assistantMessageEvent); + const messageId = state.messageId; + if (!delta || !messageId) return []; + + const out: RemoteAgentEvent[] = []; + if (!state.startEmitted) { + state.startEmitted = true; + out.push({ type: "start", messageId }, activity(null)); + } + if (delta.kind === "delta") state.accum += delta.text; + else state.thinkingAccum += delta.text; + out.push({ type: delta.kind, messageId, delta: delta.text }); + return out; +} + +function onMessageEnd( + state: PiStreamState, + event: PiStdoutEvent, +): RemoteAgentEvent[] { + if (!isAssistant(event)) return []; + const out: RemoteAgentEvent[] = []; + if (state.messageId && state.startEmitted) { + out.push({ + type: "end", + messageId: state.messageId, + content: assistantText(event.message) || state.accum, + thinking: state.thinkingAccum, + }); + } + resetMessage(state); + out.push(activity(THINKING)); + return out; +} + +const TRANSLATORS: Record = { + agent_start: (state) => { + resetMessage(state); + return [activity(THINKING)]; + }, + message_start: onMessageStart, + message_update: onMessageUpdate, + message_end: onMessageEnd, + tool_execution_start: (_state, event) => { + const toolName = event.toolName ?? "tool"; + return [ + activity({ + kind: "tool", + label: toolActivityLabel(toolName, event.args), + toolName, + }), + ]; + }, + queue_update: (_state, event) => [ + { + type: "queue", + steering: toStringArray(event.steering), + followUp: toStringArray(event.followUp), + }, + ], + agent_end: (state) => { + resetMessage(state); + return [activity(null)]; + }, +}; + +/** + * Translates one pi RPC event into the remote-env events it implies, updating + * `state`. Mirrors the server's local pi relay so a sub-agent hosted here + * renders the same as one hosted on the server. Unknown types yield nothing. + */ +export function translatePiEvent( + state: PiStreamState, + event: PiStdoutEvent, + newId: () => string, +): RemoteAgentEvent[] { + const translate = TRANSLATORS[event.type ?? ""]; + return translate ? translate(state, event, newId) : []; +} diff --git a/packages/local-agent-host/src/remoteEnvToken.test.ts b/packages/local-agent-host/src/remoteEnvToken.test.ts new file mode 100644 index 00000000..c68c1cc9 --- /dev/null +++ b/packages/local-agent-host/src/remoteEnvToken.test.ts @@ -0,0 +1,83 @@ +import assert from "node:assert/strict"; +import { afterEach, test } from "node:test"; + +import { createUnsignedJwt, mintRemoteEnvToken } from "./remoteEnvToken.ts"; + +const realFetch = globalThis.fetch; + +afterEach(() => { + globalThis.fetch = realFetch; +}); + +function decodeSegment(segment: string): unknown { + return JSON.parse(Buffer.from(segment, "base64url").toString("utf8")); +} + +test("createUnsignedJwt encodes claims with an alg:none header", () => { + const jwt = createUnsignedJwt({ email: "dev@localhost" }); + const [header, payload, signature] = jwt.split("."); + + assert.deepEqual(decodeSegment(header), { alg: "none", typ: "JWT" }); + assert.deepEqual(decodeSegment(payload), { email: "dev@localhost" }); + assert.equal(signature, ""); +}); + +test("mintRemoteEnvToken posts an identity JWT and parses the response", async () => { + const calls: Array<{ url: string; init: RequestInit }> = []; + globalThis.fetch = (async (url: string, init: RequestInit) => { + calls.push({ url, init }); + return new Response( + JSON.stringify({ + token: "scoped-token", + environmentId: "env-minted", + expiresAt: "2026-10-15T12:00:00.000Z", + }), + { status: 200, headers: { "Content-Type": "application/json" } }, + ); + }) as typeof fetch; + + const result = await mintRemoteEnvToken({ + shellUrl: "http://shell.test/", + selector: "projectId:project-a", + email: "dev@localhost", + environmentId: "env-pin", + }); + + assert.deepEqual(result, { + token: "scoped-token", + environmentId: "env-minted", + expiresAtMs: Date.parse("2026-10-15T12:00:00.000Z"), + }); + + assert.equal(calls.length, 1); + assert.equal(calls[0].url, "http://shell.test/api/embed/remote-env-token"); + const headers: Record = calls[0].init.headers as Record< + string, + string + >; + assert.equal( + headers.Authorization, + `Bearer ${createUnsignedJwt({ email: "dev@localhost" })}`, + ); + assert.deepEqual(JSON.parse(String(calls[0].init.body)), { + selector: "projectId:project-a", + environmentId: "env-pin", + }); +}); + +test("mintRemoteEnvToken throws on a non-ok response", async () => { + globalThis.fetch = (async () => + new Response("nope", { + status: 401, + statusText: "Unauthorized", + })) as typeof fetch; + + await assert.rejects( + mintRemoteEnvToken({ + shellUrl: "http://shell.test", + selector: "projectId:project-a", + email: "dev@localhost", + }), + /Failed to mint remote-env token \(401 Unauthorized\)/, + ); +}); diff --git a/packages/local-agent-host/src/remoteEnvToken.ts b/packages/local-agent-host/src/remoteEnvToken.ts new file mode 100644 index 00000000..5e0d89e8 --- /dev/null +++ b/packages/local-agent-host/src/remoteEnvToken.ts @@ -0,0 +1,98 @@ +import type { + RemoteEnvTokenRequest, + RemoteEnvTokenResponse, +} from "@tangent/shared/contracts.ts"; + +const REMOTE_ENV_TOKEN_PATH = "/api/embed/remote-env-token"; + +export interface ScopedEnvToken { + token: string; + environmentId: string; + expiresAtMs?: number; +} + +export interface MintRemoteEnvTokenParams { + shellUrl: string; + /** Routes this host to every session whose tags include it, e.g. `projectId:`. */ + selector: string; + email: string; + environmentId?: string; +} + +function base64UrlEncode(value: string): string { + return Buffer.from(value, "utf8").toString("base64url"); +} + +/** + * A JWT with `alg: "none"` and an empty signature. The Shell's embed mint + * endpoint decodes the claims without verifying the signature, so this window + * can present its identity without a server-issued credential. + */ +export function createUnsignedJwt(claims: Record): string { + const header = base64UrlEncode(JSON.stringify({ alg: "none", typ: "JWT" })); + const payload = base64UrlEncode(JSON.stringify(claims)); + return `${header}.${payload}.`; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null; +} + +function isResponseBody(value: unknown): value is RemoteEnvTokenResponse { + return ( + isRecord(value) && + typeof value.token === "string" && + value.token.length > 0 && + typeof value.environmentId === "string" && + value.environmentId.length > 0 + ); +} + +function parseExpiresAtMs(value: string): number | undefined { + const parsed = Date.parse(value); + return Number.isNaN(parsed) ? undefined : parsed; +} + +/** + * Mints a selector-scoped `/remote-env` token as `email`. The server routes a + * session's spawns and tool calls to the prompting person's scoped hosts, so + * this host reaches every session whose tags include `selector` — but only when + * the token's `sub` matches the prompting person. `environmentId` pins this + * host's place across refreshes; the server honors it when free, so the + * response's id is authoritative. + */ +export async function mintRemoteEnvToken({ + shellUrl, + selector, + email, + environmentId, +}: MintRemoteEnvTokenParams): Promise { + const url = `${shellUrl.replace(/\/+$/, "")}${REMOTE_ENV_TOKEN_PATH}`; + const body: RemoteEnvTokenRequest = { + selector, + ...(environmentId ? { environmentId } : {}), + }; + const response = await fetch(url, { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: `Bearer ${createUnsignedJwt({ email })}`, + }, + body: JSON.stringify(body), + }); + if (!response.ok) { + throw new Error( + `Failed to mint remote-env token (${response.status} ${response.statusText}).`, + ); + } + const parsed: unknown = await response.json(); + if (!isResponseBody(parsed)) { + throw new Error("Remote-env token response was malformed."); + } + const expiresAtMs = parseExpiresAtMs(parsed.expiresAt); + return { + token: parsed.token, + environmentId: parsed.environmentId, + ...(expiresAtMs ? { expiresAtMs } : {}), + }; +} diff --git a/packages/local-agent-host/src/remoteFileTools.test.ts b/packages/local-agent-host/src/remoteFileTools.test.ts new file mode 100644 index 00000000..a30c5aec --- /dev/null +++ b/packages/local-agent-host/src/remoteFileTools.test.ts @@ -0,0 +1,170 @@ +import assert from "node:assert/strict"; +import { mkdir, mkdtemp, rm, symlink, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { after, before, test } from "node:test"; + +import { createFileTools } from "./remoteFileTools.ts"; +import { + globMatcher, + isDeniedReadPath, + resolveWithinRoot, +} from "./workspaceFs.ts"; + +let root = ""; +let outside = ""; + +async function put(relative: string, content: string | Buffer): Promise { + const file = path.join(root, relative); + await mkdir(path.dirname(file), { recursive: true }); + await writeFile(file, content); +} + +before(async () => { + root = await mkdtemp(path.join(tmpdir(), "tangent-tools-")); + outside = await mkdtemp(path.join(tmpdir(), "tangent-outside-")); + await writeFile(path.join(outside, "secret.txt"), "secret"); + await put("README.md", "# Hello\n"); + await put("src/index.ts", "export const answer = 42;\n// TODO later\n"); + await put("src/lib/util.ts", "export function todo() {}\n"); + await put("src/lib/util.test.ts", "test('todo')\n"); + await put("node_modules/pkg/index.ts", "// TODO hidden\n"); + await put("image.bin", Buffer.from([0x89, 0x00, 0x01, 0x02])); + await put("big.txt", "x".repeat(2000)); + await symlink(path.join(outside, "secret.txt"), path.join(root, "link.txt")); +}); + +after(async () => { + await rm(root, { recursive: true, force: true }); + await rm(outside, { recursive: true, force: true }); +}); + +const tools = () => createFileTools(() => root); +const call = async (name: string, args: unknown): Promise => + tools()[name]?.execute(args); + +test("resolveWithinRoot rejects absolute and escaping paths", () => { + assert.equal( + resolveWithinRoot("/ws", "src/a.ts"), + path.resolve("/ws/src/a.ts"), + ); + assert.equal(resolveWithinRoot("/ws", "..file"), path.resolve("/ws/..file")); + assert.throws(() => resolveWithinRoot("/ws", "../etc/passwd"), /outside/); + assert.throws(() => resolveWithinRoot("/ws", "src/../../x"), /outside/); + assert.throws(() => resolveWithinRoot("/ws", "/etc/passwd"), /relative/); +}); + +test("globMatcher matches names anywhere and paths from the root", () => { + assert.ok(globMatcher("*.ts")("src/lib/util.ts")); + assert.ok(globMatcher("src/**/*.test.ts")("src/lib/util.test.ts")); + assert.ok(globMatcher("src/**/*.ts")("src/index.ts")); + assert.ok(globMatcher("*.{md,ts}")("README.md")); + assert.ok(!globMatcher("src/*.ts")("src/lib/util.ts")); + assert.ok(!globMatcher("*.ts")("README.md")); +}); + +test("read_file returns text and truncates past maxBytes", async () => { + assert.equal(await call("read_file", { path: "README.md" }), "# Hello\n"); + const truncated = (await call("read_file", { + path: "big.txt", + maxBytes: 10, + })) as string; + assert.match(truncated, /^x{10}\n… truncated \(showing 10 of 2000 bytes\)$/); +}); + +test("read_file refuses binary files, escapes, and outbound symlinks", async () => { + await assert.rejects(call("read_file", { path: "image.bin" }), /binary/); + await assert.rejects(call("read_file", { path: "../secret.txt" }), /outside/); + await assert.rejects(call("read_file", { path: "link.txt" }), /outside/); + await assert.rejects(call("read_file", {}), /"path" is required/); +}); + +test("list_dir marks directories and hides ignored folders", async () => { + const listing = (await call("list_dir", {})) as string; + assert.deepEqual(listing.split("\n"), [ + "big.txt", + "image.bin", + "link.txt", + "README.md", + "src/", + ]); + assert.equal( + await call("list_dir", { path: "src/lib" }), + "util.test.ts\nutil.ts", + ); +}); + +test("find_files walks the tree, skips ignored dirs, and honors maxResults", async () => { + assert.equal( + await call("find_files", { glob: "*.ts" }), + "src/index.ts\nsrc/lib/util.test.ts\nsrc/lib/util.ts", + ); + assert.equal( + await call("find_files", { glob: "*.ts", maxResults: 1 }), + "src/index.ts\n… stopped at 1 results", + ); + assert.equal(await call("find_files", { glob: "*.py" }), "No files matched."); +}); + +test("search_text finds literal and regex matches with line numbers", async () => { + assert.equal( + await call("search_text", { query: "TODO" }), + "src/index.ts:2: // TODO later", + ); + assert.equal( + await call("search_text", { + query: "todo", + ignoreCase: true, + glob: "src/lib/*.ts", + }), + "src/lib/util.test.ts:1: test('todo')\nsrc/lib/util.ts:1: export function todo() {}", + ); + assert.equal( + await call("search_text", { query: "answer = \\d+", regex: true }), + "src/index.ts:1: export const answer = 42;", + ); + assert.equal(await call("search_text", { query: "a.b" }), "No matches."); +}); + +test("isDeniedReadPath refuses VCS internals and dotenv secrets", () => { + assert.ok(isDeniedReadPath(".env")); + assert.ok(isDeniedReadPath(".env.local")); + assert.ok(isDeniedReadPath("./.git/config")); + assert.ok(isDeniedReadPath("nested/.git/HEAD")); + assert.ok(!isDeniedReadPath("src/env.ts")); + assert.ok(!isDeniedReadPath("README.md")); +}); + +test("read_file and search_text never surface secret or VCS paths", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "tangent-deny-")); + try { + await writeFile(path.join(dir, ".env"), "SECRET=topsecret\n"); + await mkdir(path.join(dir, ".git"), { recursive: true }); + await writeFile(path.join(dir, ".git", "config"), "token = abc\n"); + await writeFile(path.join(dir, "app.ts"), "const SECRET = 1;\n"); + const deny = createFileTools(() => dir); + + await assert.rejects( + Promise.resolve(deny.read_file?.execute({ path: ".env" })), + /not readable/, + ); + await assert.rejects( + Promise.resolve(deny.read_file?.execute({ path: ".git/config" })), + /not readable/, + ); + assert.equal( + await deny.search_text?.execute({ query: "SECRET" }), + "app.ts:1: const SECRET = 1;", + ); + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); + +test("tools fail clearly when no folder is open", async () => { + const closed = createFileTools(() => undefined); + await assert.rejects( + Promise.resolve(closed.list_dir?.execute({})), + /No folder is open/, + ); +}); diff --git a/packages/local-agent-host/src/remoteFileTools.ts b/packages/local-agent-host/src/remoteFileTools.ts new file mode 100644 index 00000000..734ebf54 --- /dev/null +++ b/packages/local-agent-host/src/remoteFileTools.ts @@ -0,0 +1,127 @@ +import type { RemoteTool, RemoteToolMap } from "@tangent/remote-subagent"; + +import { + DEFAULT_MAX_RESULTS, + DEFAULT_READ_BYTES, + findFiles, + listDirectory, + MAX_READ_BYTES, + MAX_RESULTS, + readTextFile, + searchText, +} from "./fileToolOps.ts"; +import { toArgs, type ToolArgs } from "./toolArgs.ts"; + +interface FileToolSpec extends Omit { + run(root: string, args: ToolArgs): Promise; +} + +const MAX_RESULTS_SCHEMA = { + type: "number", + description: `Default ${DEFAULT_MAX_RESULTS}, max ${MAX_RESULTS}.`, +}; + +const FILE_TOOLS: Record = { + read_file: { + description: + "Read a text file from the user's local workspace (read-only). " + + "Paths are relative to the workspace root. Large files are truncated.", + inputSchema: { + type: "object", + properties: { + path: { type: "string", description: "Workspace-relative file path." }, + maxBytes: { + type: "number", + description: `Bytes to read (default ${DEFAULT_READ_BYTES}, max ${MAX_READ_BYTES}).`, + }, + }, + required: ["path"], + }, + run: readTextFile, + }, + list_dir: { + description: + "List a directory in the user's local workspace (read-only). " + + "Directories end with '/'. Build and dependency folders are hidden.", + inputSchema: { + type: "object", + properties: { + path: { + type: "string", + description: "Workspace-relative directory; defaults to the root.", + }, + }, + }, + run: listDirectory, + }, + find_files: { + description: + "Find files in the user's local workspace by glob (read-only), " + + "e.g. '*.ts' or 'src/**/*.test.ts'. A glob without '/' matches file names anywhere.", + inputSchema: { + type: "object", + properties: { + glob: { + type: "string", + description: "Glob supporting *, **, ?, {a,b}.", + }, + maxResults: MAX_RESULTS_SCHEMA, + }, + required: ["glob"], + }, + run: findFiles, + }, + search_text: { + description: + "Search file contents in the user's local workspace (read-only). " + + "Returns 'path:line: text' for each matching line.", + inputSchema: { + type: "object", + properties: { + query: { + type: "string", + description: "Text to find, or a regex with regex: true.", + }, + regex: { + type: "boolean", + description: "Treat query as a JavaScript regex.", + }, + ignoreCase: { + type: "boolean", + description: "Match case-insensitively.", + }, + glob: { + type: "string", + description: "Only search files matching this glob.", + }, + maxResults: MAX_RESULTS_SCHEMA, + }, + required: ["query"], + }, + run: searchText, + }, +}; + +/** + * Read-only workspace tools Prime calls over `/remote-env` without spawning a + * local sub-agent. Every path is resolved against `getRoot()` at call time. + */ +export function createFileTools( + getRoot: () => string | undefined, +): RemoteToolMap { + const entries = Object.entries(FILE_TOOLS).map( + ([name, { run, ...spec }]): [string, RemoteTool] => [ + name, + { + ...spec, + async execute(args) { + const root = getRoot(); + if (!root) + throw new Error("No folder is open in the user's local workspace."); + return run(root, toArgs(args)); + }, + }, + ], + ); + return Object.fromEntries(entries); +} diff --git a/packages/local-agent-host/src/scopedConnection.test.ts b/packages/local-agent-host/src/scopedConnection.test.ts new file mode 100644 index 00000000..713b4dc6 --- /dev/null +++ b/packages/local-agent-host/src/scopedConnection.test.ts @@ -0,0 +1,116 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; + +import type { + MintRemoteEnvTokenParams, + ScopedEnvToken, +} from "./remoteEnvToken.ts"; +import { startScopedConnection } from "./scopedConnection.ts"; + +async function flush(): Promise { + for (let i = 0; i < 5; i += 1) await Promise.resolve(); +} + +const base = { + shellUrl: "http://shell.test", + selector: "projectId:project-a", + email: "dev@localhost", +}; + +test("mints, connects, and pins the minted environment id on refresh", async (t) => { + t.mock.timers.enable({ apis: ["setTimeout"] }); + const mintCalls: MintRemoteEnvTokenParams[] = []; + const connects: Array<{ token: string; environmentId: string }> = []; + let count = 0; + const mint = async ( + params: MintRemoteEnvTokenParams, + ): Promise => { + mintCalls.push(params); + count += 1; + return { + token: `token-${count}`, + environmentId: "env-minted", + expiresAtMs: Date.now() + 100_000, + }; + }; + + const stop = startScopedConnection({ + ...base, + initialEnvironmentId: "env-pin", + connect: (token, environmentId) => connects.push({ token, environmentId }), + onError: () => {}, + mint, + }); + await flush(); + + assert.equal(connects.length, 1); + assert.deepEqual(connects[0], { + token: "token-1", + environmentId: "env-minted", + }); + assert.equal(mintCalls[0].environmentId, "env-pin"); + + t.mock.timers.tick(80_000); + await flush(); + + assert.equal(connects.length, 2); + assert.equal(mintCalls[1].environmentId, "env-minted"); + stop(); +}); + +test("reports the error and retries after a failed mint", async (t) => { + t.mock.timers.enable({ apis: ["setTimeout"] }); + const errors: string[] = []; + let calls = 0; + const mint = async (): Promise => { + calls += 1; + if (calls === 1) throw new Error("mint failed"); + return { token: "token", environmentId: "env" }; + }; + const connects: number[] = []; + + const stop = startScopedConnection({ + ...base, + connect: () => connects.push(1), + onError: (message) => errors.push(message), + mint, + }); + await flush(); + + assert.deepEqual(errors, ["mint failed"]); + assert.equal(connects.length, 0); + + t.mock.timers.tick(5_000); + await flush(); + + assert.equal(connects.length, 1); + stop(); +}); + +test("stop cancels the pending refresh", async (t) => { + t.mock.timers.enable({ apis: ["setTimeout"] }); + let calls = 0; + const mint = async (): Promise => { + calls += 1; + return { + token: "token", + environmentId: "env", + expiresAtMs: Date.now() + 100_000, + }; + }; + + const stop = startScopedConnection({ + ...base, + connect: () => {}, + onError: () => {}, + mint, + }); + await flush(); + assert.equal(calls, 1); + + stop(); + t.mock.timers.tick(200_000); + await flush(); + + assert.equal(calls, 1); +}); diff --git a/packages/local-agent-host/src/scopedConnection.ts b/packages/local-agent-host/src/scopedConnection.ts new file mode 100644 index 00000000..e28e5219 --- /dev/null +++ b/packages/local-agent-host/src/scopedConnection.ts @@ -0,0 +1,78 @@ +import { mintRemoteEnvToken } from "./remoteEnvToken.ts"; + +/** Refresh this long before a token expires, and never sooner than the floor. */ +const TOKEN_REFRESH_BUFFER_MS = 30_000; +const MIN_RETRY_MS = 5_000; +const MAX_RETRY_MS = 30_000; + +export interface ScopedConnectionOptions { + shellUrl: string; + /** Routes this host to every session tagged with it, e.g. `projectId:`. */ + selector: string; + email: string; + initialEnvironmentId?: string; + connect(token: string, environmentId: string): void; + onError(message: string): void; + /** Test seam; defaults to the live mint endpoint. */ + mint?: typeof mintRemoteEnvToken; +} + +function errorMessage(error: unknown): string { + if (error instanceof Error) return error.message; + return typeof error === "string" ? error : "Unknown error."; +} + +/** + * Keeps a selector-scoped host connected with a self-renewing token: mints a + * token, calls {@link ScopedConnectionOptions.connect}, and schedules a refresh + * before it expires. The environment id is pinned across refreshes so the server + * keeps routing the project's spawns to the same host. Returns a stop function + * that cancels the loop; the caller still owns tearing down the connection. + */ +export function startScopedConnection( + options: ScopedConnectionOptions, +): () => void { + const mint = options.mint ?? mintRemoteEnvToken; + let cancelled = false; + let timer: ReturnType | undefined; + let environmentId = options.initialEnvironmentId; + let retryMs = MIN_RETRY_MS; + + function schedule(delay: number): void { + timer = setTimeout(() => void connectOnce(), delay); + } + + async function connectOnce(): Promise { + try { + const token = await mint({ + shellUrl: options.shellUrl, + selector: options.selector, + email: options.email, + ...(environmentId ? { environmentId } : {}), + }); + if (cancelled) return; + environmentId = token.environmentId; + options.connect(token.token, token.environmentId); + retryMs = MIN_RETRY_MS; + if (token.expiresAtMs) { + const delay = Math.max( + token.expiresAtMs - Date.now() - TOKEN_REFRESH_BUFFER_MS, + MIN_RETRY_MS, + ); + schedule(delay); + } + } catch (error) { + if (cancelled) return; + options.onError(errorMessage(error)); + schedule(retryMs); + retryMs = Math.min(retryMs * 2, MAX_RETRY_MS); + } + } + + void connectOnce(); + + return () => { + cancelled = true; + if (timer) clearTimeout(timer); + }; +} diff --git a/packages/local-agent-host/src/shellUrl.test.ts b/packages/local-agent-host/src/shellUrl.test.ts new file mode 100644 index 00000000..fc5387cb --- /dev/null +++ b/packages/local-agent-host/src/shellUrl.test.ts @@ -0,0 +1,21 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; + +import { normalizeShellUrl, parseSessionRef } from "./shellUrl.ts"; + +test("shell URLs drop a session path and trailing slash", () => { + assert.equal( + normalizeShellUrl("http://localhost:5173/sessions/abc?x=1"), + "http://localhost:5173", + ); + assert.equal( + normalizeShellUrl("https://host/tangent/"), + "https://host/tangent", + ); + assert.throws(() => normalizeShellUrl("ftp://host")); +}); + +test("session refs accept a pasted URL or a raw id", () => { + assert.equal(parseSessionRef("http://h/sessions/abc-123#chat"), "abc-123"); + assert.equal(parseSessionRef(" abc-123 "), "abc-123"); +}); diff --git a/packages/local-agent-host/src/shellUrl.ts b/packages/local-agent-host/src/shellUrl.ts new file mode 100644 index 00000000..fb1a34b9 --- /dev/null +++ b/packages/local-agent-host/src/shellUrl.ts @@ -0,0 +1,26 @@ +/** + * Normalizes a user-entered Shell URL to its origin plus any mount path, + * without a trailing slash. Throws on anything that is not http(s). + */ +export function normalizeShellUrl(input: string): string { + const url = new URL(input.trim()); + if (url.protocol !== "http:" && url.protocol !== "https:") { + throw new Error("The Shell URL must start with http:// or https://"); + } + const sessionPath = url.pathname.indexOf("/sessions"); + const mount = + sessionPath === -1 ? url.pathname : url.pathname.slice(0, sessionPath); + return `${url.origin}${mount}`.replace(/\/+$/, ""); +} + +/** The session id a pasted Shell URL points at, if it names one. */ +export function sessionIdFromUrl(input: string): string | undefined { + const match = /\/sessions\/([^/?#]+)/.exec(input); + return match?.[1] ? decodeURIComponent(match[1]) : undefined; +} + +/** Extracts a session id from a pasted Shell session URL, or returns the raw id. */ +export function parseSessionRef(input: string): string { + const trimmed = input.trim(); + return sessionIdFromUrl(trimmed) ?? trimmed; +} diff --git a/packages/local-agent-host/src/toolArgs.ts b/packages/local-agent-host/src/toolArgs.ts new file mode 100644 index 00000000..93910902 --- /dev/null +++ b/packages/local-agent-host/src/toolArgs.ts @@ -0,0 +1,40 @@ +export type ToolArgs = Record; + +export function toArgs(args: unknown): ToolArgs { + if (args === undefined || args === null) return {}; + if (typeof args !== "object" || Array.isArray(args)) { + throw new Error("Arguments must be a JSON object."); + } + return args as ToolArgs; +} + +export function optionalString( + args: ToolArgs, + key: string, +): string | undefined { + const value = args[key]; + if (value === undefined || value === null || value === "") return undefined; + if (typeof value !== "string") throw new Error(`"${key}" must be a string.`); + return value; +} + +export function requiredString(args: ToolArgs, key: string): string { + const value = optionalString(args, key); + if (value === undefined) throw new Error(`"${key}" is required.`); + return value; +} + +/** A positive integer argument, clamped to `max`. */ +export function boundedInt( + args: ToolArgs, + key: string, + fallback: number, + max: number, +): number { + const value = args[key]; + if (value === undefined || value === null) return fallback; + if (typeof value !== "number" || !Number.isFinite(value) || value < 1) { + throw new Error(`"${key}" must be a positive number.`); + } + return Math.min(Math.floor(value), max); +} diff --git a/packages/local-agent-host/src/workspaceFs.ts b/packages/local-agent-host/src/workspaceFs.ts new file mode 100644 index 00000000..47eae2e6 --- /dev/null +++ b/packages/local-agent-host/src/workspaceFs.ts @@ -0,0 +1,126 @@ +import { readdir, realpath } from "node:fs/promises"; +import path from "node:path"; + +export const IGNORED_DIRS = new Set([ + ".git", + "node_modules", + "dist", + "build", + "out", + "coverage", + ".turbo", + ".next", + ".venv", + "__pycache__", +]); + +const BINARY_SNIFF_BYTES = 8 * 1024; + +/** Path segments whose contents are VCS internals or secrets, never read. */ +const DENIED_READ_DIRS = new Set([".git"]); + +/** + * Whether a workspace-relative path must not have its contents read even though + * it lives inside the root: version-control internals (`.git/**`) and dotenv + * secret files (`.env`, `.env.local`, ...). Traversal already skips ignored + * directories, but `read_file`/`search_text` can still name such a path + * directly, so they consult this before returning any bytes to the remote peer. + */ +export function isDeniedReadPath(relative: string): boolean { + const segments = relative + .replace(/^\.\//, "") + .split(/[\\/]/) + .filter((segment) => segment && segment !== "."); + if (segments.some((segment) => DENIED_READ_DIRS.has(segment))) return true; + const name = segments.at(-1) ?? ""; + return name === ".env" || name.startsWith(".env."); +} + +const GLOB_TOKENS: Record = { + "**/": "(?:.*/)?", + "**": ".*", + "*": "[^/]*", + "?": "[^/]", + "{": "(?:", + "}": ")", + ",": "|", +}; + +function toPosix(relative: string): string { + return relative.split(path.sep).join("/"); +} + +function isInside(root: string, target: string): boolean { + const relative = path.relative(root, target); + return ( + relative !== ".." && + !relative.startsWith(`..${path.sep}`) && + !path.isAbsolute(relative) + ); +} + +/** Resolves a workspace-relative path, refusing anything that escapes `root`. */ +export function resolveWithinRoot(root: string, relative = "."): string { + if (path.isAbsolute(relative)) { + throw new Error("Paths must be relative to the workspace root."); + } + const resolved = path.resolve(root, relative); + if (!isInside(root, resolved)) { + throw new Error(`Path "${relative}" is outside the workspace.`); + } + return resolved; +} + +/** Like {@link resolveWithinRoot}, but also refuses symlinks that point outside. */ +export async function resolveExisting( + root: string, + relative = ".", +): Promise { + const resolved = resolveWithinRoot(root, relative); + const [realRoot, realTarget] = await Promise.all([ + realpath(root), + realpath(resolved).catch(() => { + throw new Error(`"${relative}" does not exist.`); + }), + ]); + if (!isInside(realRoot, realTarget)) { + throw new Error(`Path "${relative}" is outside the workspace.`); + } + return realTarget; +} + +/** + * Compiles a glob (`*`, `**`, `?`, `{a,b}`). A pattern without `/` matches a + * file's base name anywhere in the tree, so `*.ts` matches at any depth. + */ +export function globMatcher(glob: string): (relative: string) => boolean { + const pattern = glob.trim().replace(/^\.\//, ""); + const source = pattern.replace( + /\*\*\/|\*\*|\*|\?|\{|\}|,|[.+^$()|[\]\\]/g, + (token) => GLOB_TOKENS[token] ?? `\\${token}`, + ); + const regex = new RegExp(`^(?:${source})$`); + const byName = !pattern.includes("/"); + return (relative) => + regex.test(byName ? path.posix.basename(relative) : relative); +} + +export function isBinary(buffer: Buffer): boolean { + return buffer.subarray(0, BINARY_SNIFF_BYTES).includes(0); +} + +/** Yields workspace-relative posix paths of regular files, skipping ignored dirs and symlinks. */ +export async function* walkFiles( + root: string, + dir = root, +): AsyncGenerator { + const entries = await readdir(dir, { withFileTypes: true }).catch(() => []); + entries.sort((a, b) => a.name.localeCompare(b.name)); + for (const entry of entries) { + const full = path.join(dir, entry.name); + if (entry.isFile()) yield toPosix(path.relative(root, full)); + else if (entry.isDirectory() && !IGNORED_DIRS.has(entry.name)) { + yield* walkFiles(root, full); + } + } +} diff --git a/packages/local-agent-host/tsconfig.json b/packages/local-agent-host/tsconfig.json new file mode 100644 index 00000000..d0d3089e --- /dev/null +++ b/packages/local-agent-host/tsconfig.json @@ -0,0 +1,8 @@ +{ + "extends": "@tangent/build/tsconfig/node.json", + "compilerOptions": { + "tsBuildInfoFile": "./node_modules/.tmp/tsconfig.tsbuildinfo", + "types": ["node"] + }, + "include": ["src"] +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 129a032a..60f26914 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -343,6 +343,31 @@ importers: specifier: ^6.0.3 version: 6.0.3 + packages/local-agent-host: + dependencies: + '@tangent/remote-subagent': + specifier: workspace:* + version: link:../remote-subagent + '@tangent/shared': + specifier: workspace:* + version: link:../shared + devDependencies: + '@tangent/build': + specifier: workspace:* + version: link:../build + '@types/node': + specifier: ^25.9.1 + version: 25.9.1 + eslint: + specifier: ^10.4.0 + version: 10.4.0(jiti@2.7.0) + tsx: + specifier: ^4.22.3 + version: 4.22.3 + typescript: + specifier: ^6.0.3 + version: 6.0.3 + packages/remote-subagent: dependencies: '@tangent/shared':