pi-gui/test/pi-rpc-adapter.test.js

import assert from "node:assert/strict";
import { EventEmitter } from "node:events";
import { PassThrough } from "node:stream";
import test from "node:test";
import { startPiRpcAdapter } from "../src/bridge/pi-rpc-adapter.js";

class FakePiChild extends EventEmitter {
	constructor() {
		super();
		this.stdin = new PassThrough();
		this.stdout = new PassThrough();
		this.stderr = new PassThrough();
		this.killed = false;
	}

	kill(signal) {
		this.killed = true;
		this.emit("exit", null, signal);
		return true;
	}
}

function parseSent(stdin) {
	try {
		return stdin
			.split("\n")
			.filter(Boolean)
			.map((line) => JSON.parse(line));
	} catch (error) {
		throw new Error(
			`fake child received invalid JSON: ${error instanceof Error ? error.message : "unknown error"}`,
		);
	}
}

function createFixture() {
	const child = new FakePiChild();
	const calls = [];
	let stdin = "";
	child.stdin.on("data", (chunk) => {
		stdin += chunk;
	});
	return {
		child,
		calls,
		spawn: (command, args, options) => {
			calls.push({ command, args, options });
			return child;
		},
		sent: () => parseSent(stdin),
	};
}

test("starts Pi in RPC mode and correlates a command response", async () => {
	const fixture = createFixture();
	const events = [];
	const adapter = startPiRpcAdapter({
		cwd: "/workspace/home",
		sessionDir: "/workspace/sessions",
		spawnProcess: fixture.spawn,
		onEvent: (event) => events.push(event),
	});

	const response = adapter.send({
		type: "prompt",
		message: "Implement the bridge",
	});
	const [command] = fixture.sent();
	assert.deepEqual(fixture.calls[0], {
		command: "pi",
		args: ["--mode", "rpc", "--session-dir", "/workspace/sessions"],
		options: { cwd: "/workspace/home", stdio: ["pipe", "pipe", "pipe"] },
	});
	assert.equal(command.type, "prompt");
	assert.equal(command.message, "Implement the bridge");
	assert.match(command.id, /^bridge-/);

	fixture.child.stdout.write('{"type":"agent_start"}\n');
	fixture.child.stdout.write(
		`${JSON.stringify({ type: "response", id: command.id, command: "prompt", success: true })}\n`,
	);

	assert.deepEqual(await response, {
		type: "response",
		id: command.id,
		command: "prompt",
		success: true,
	});
	assert.deepEqual(events, [
		{
			seq: 1,
			type: "agent_state",
			data: { state: "streaming", event: { type: "agent_start" } },
		},
	]);
	await adapter.stop();
});

test("requests Pi session statistics for context and token status", async () => {
	const fixture = createFixture();
	const adapter = startPiRpcAdapter({
		cwd: "/workspace/home",
		sessionDir: "/workspace/sessions",
		spawnProcess: fixture.spawn,
	});

	const pending = adapter.send({ type: "get_session_stats" });
	const [command] = fixture.sent();
	assert.equal(command.type, "get_session_stats");
	fixture.child.stdout.write(
		`${JSON.stringify({ type: "response", id: command.id, command: "get_session_stats", success: true, data: { contextUsage: { tokens: 32000, contextWindow: 200000 } } })}\n`,
	);
	const stats = await pending;
	assert.equal(stats.data.contextUsage.tokens, 32000);
	await adapter.stop();
});

test("forwards extension responses without replacing Pi's request ID", async () => {
	const fixture = createFixture();
	const events = [];
	const adapter = startPiRpcAdapter({
		cwd: "/workspace/home",
		sessionDir: "/workspace/sessions",
		spawnProcess: fixture.spawn,
		onEvent: (event) => events.push(event),
	});

	const extensionRequest = {
		type: "extension_ui_request",
		id: "extension-request-1",
		method: "confirm",
		title: "Keep U+2028 here",
		message: "Approve?",
	};
	fixture.child.stdout.write(`${JSON.stringify(extensionRequest)}\n`);
	adapter.respondToExtension("extension-request-1", { confirmed: false });

	assert.deepEqual(events, [
		{ seq: 1, type: "extension_ui_request", data: { event: extensionRequest } },
	]);
	assert.deepEqual(fixture.sent(), [
		{
			type: "extension_ui_response",
			id: "extension-request-1",
			confirmed: false,
		},
	]);
	await adapter.stop();
});

test("reports invalid child output and rejects pending commands on child exit", async () => {
	const fixture = createFixture();
	const errors = [];
	const adapter = startPiRpcAdapter({
		cwd: "/workspace/home",
		sessionDir: "/workspace/sessions",
		spawnProcess: fixture.spawn,
		onError: (error) => errors.push(error),
	});

	const pending = adapter.send({ type: "get_state" });
	fixture.child.stdout.write("not-json\n");
	fixture.child.emit("exit", 1, null);

	await assert.rejects(pending, /exited before responding/);
	assert.equal(errors[0].code, "invalid_json");
	await adapter.stop();
});