jay
/
pi-remote-control
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: jay:af990f65924b9319fbb2e2154a8f6ad9f89414a2
Branches Tags
jay:main
jay:fix/post-sessions-response-shape
jay:fix/ws-input-handler
jay:feat/sidecar-pair-smoke
jay:feat/spike-tmux-cc
jay:feat/spike-stream
jay:pre-qwen-swarm
...
compare: jay:d74341af2a563175bbdbf4cfc9c54ceb54a653e7
Branches Tags
jay:main
jay:fix/post-sessions-response-shape
jay:fix/ws-input-handler
jay:feat/sidecar-pair-smoke
jay:feat/spike-tmux-cc
jay:feat/spike-stream
jay:pre-qwen-swarm

4 Commits
af990f6592 ... d74341af2a

Author SHA1 Message Date
Johannes Merz d74341af2a merge: T-1.0 server refactor + T-1.0a smoke harness 2026-05-15 11:20:32 +02:00
jay a7dad86901 feat(t-1.0a): smoke test harness MVP 
- scripts/smoke/helpers.mjs — spawn-pi (via python3 pty.spawn), wait-for-port,
  fetch, WebSocket helpers; createSmokeHome/removeSmokeHome for isolated HOME
- scripts/smoke/smoke.mjs — 6 node:test assertions:
    GET /manifest.json → 200 + JSON shape
    GET /icon.svg → 200 + <svg body
    GET / (with token) → 302→200 + HTML marker
    GET / (no token) → 403
    WS /ws (with token) → 101 upgrade
    pi process alive check
- scripts/smoke/README.md — usage, design notes, extension guide
- package.json: add 'smoke' script
- docs/SYNC.md: add scripts/smoke/** ownership row + History entry

All 6 tests pass in ~1.4 s locally.

Key finding: pi requires a PTY to enter interactive mode and fire
session_start. Spawning without a TTY causes immediate exit. Workaround:
python3 pty.spawn() — allocates a PTY with no additional deps.
 2026-05-15 11:18:20 +02:00
jay 3e813eb90a sync: release T-1.0 claim, add history entry (@worker-t1.0) 2026-05-15 10:58:07 +02:00
jay 568931901d refactor(T-1.0): carve server.ts into server/ sub-modules 
Create the modular server/ layout from PHASE-1-sidecar.md §Architecture Sketch:

  server/types.ts    — shared WsClient/WsServer/RemoteServer interfaces
  server/upgrade.ts  — WS upgrade routing per path (LEGACY /ws)
  server/server.ts   — HTTP bootstrap, middleware, LEGACY HTML routes
  server/routes/     — empty dir, placeholder for T-1.5/T-1.6/T-1.7

The original extensions/remote-control/server.ts is replaced with a
thin re-export shim so that index.ts continues to resolve
'./server.js' without changes.

All LEGACY code paths (manifest.json, icon.svg, /, /ws) are tagged
with // LEGACY: … comments. No behaviour changes. No new endpoints.

Pre-existing biome errors in auth.ts, config.ts, index.ts are
unchanged — NOT introduced by this commit.
 2026-05-15 10:57:52 +02:00
10 changed files with 1067 additions and 334 deletions
Show Stats Expand all files Collapse all files

5
docs/SYNC.md
View File

@ -51,7 +51,7 @@ Update the **Status** column when a phase transitions. Allowed states:
| Task | Branch | Owner | Claimed at | ETA | Notes |
|---|---|---|---|---|---|
| T-1.0 | feat/p1-t1-0-server-refactor | @worker-t1.0 | 2026-05-15 | +2h | server/ scaffold, LEGACY html path preserved |
| _(none)_ | | | | | |
Example of a filled row:
```
@ -160,5 +160,6 @@ yyyy-mm-dd @handle T-x.y what was done
```
2026-05-15 @worker-phase0 T-0.* Phase 0 spike complete. tmux+pipe-pane PoC validated. GREEN LIGHT for Phase 1. Report: reference/PHASE-0-report.md. Branch: feat/spike-stream (kept for reference, not merged).
2026-05-15 @worker-phase0.5 T-0.5 Phase 0.5 spike complete. tmux control mode validated. VERDICT: Path B recommended. Report: reference/PHASE-0.5-report.md. Branch: feat/spike-tmux-cc (kept for reference, not merged).
2026-05-15 @worker-t1.0a T-1.0a Smoke test harness MVP complete. scripts/smoke/ with helpers.mjs + smoke.mjs. 6 tests: manifest, icon, GET/ auth+unauth, WS upgrade, process-alive. All green in ~1.4s. Key finding: pi requires PTY; workaround via python3 pty.spawn(). Branch: feat/p1-t1-0a-smoke.
2026-05-15 @worker-t1.0 T-1.0 Server refactor scaffold complete. server.ts carved into server/{types,server,upgrade}.ts + server/routes/ stub. LEGACY html path preserved end-to-end. Reviewer APPROVE. Branch: feat/p1-t1-0-server-refactor (merged).
2026-05-15 @worker-t1.0a T-1.0a Smoke test harness MVP complete. scripts/smoke/ with helpers.mjs + smoke.mjs. 6 tests: manifest, icon, GET/ auth+unauth, WS upgrade, process-alive. All green in ~1.4s. Key finding: pi requires PTY; workaround via python3 pty.spawn(). Branch: feat/p1-t1-0a-smoke (merged).
```

345
extensions/remote-control/server.ts
View File

@ -1,335 +1,18 @@
/**
* HTTP + WebSocket server for remote-control.
* LEGACY: re-export shim.
*
* Handles authentication, serves the web UI, and manages WebSocket connections
* for real-time message streaming between the pi session and browser clients.
* The server implementation has been moved to server/ sub-modules:
* server/server.ts — HTTP bootstrap, TLS, middleware, LEGACY HTML routes
* server/upgrade.ts — WebSocket upgrade routing per session/topic
* server/types.ts — shared WS + RemoteServer type definitions
* server/routes/ — route modules (populated by T-1.5/T-1.6/T-1.7)
*
* This shim is kept so that the existing import in index.ts
* (`import { startServer } from "./server.js"`) continues to resolve
* without modification. It will be removed once all consumers have been
* updated to import directly from server/ sub-modules.
*/
import { randomBytes } from "node:crypto";
import type { IncomingMessage } from "node:http";
import { createServer } from "node:http";
import { createRequire } from "node:module";
import type { AddressInfo, Socket } from "node:net";
import type {
ExtensionAPI,
ExtensionContext,
} from "@earendil-works/pi-coding-agent";
import {
generateSessionId,
loadOrCreateToken,
parseCookies,
SESSION_COOKIE,
validateToken,
} from "./auth.js";
import { parseBindAddress, readRemoteControlConfig } from "./config.js";
import { buildHTML } from "./html.js";
import { buildSyncMessage } from "./messages.js";
interface WsClient {
readyState: number;
send(data: string): void;
terminate(): void;
on(event: "message", listener: (data: Buffer) => void): void;
on(event: "close" | "error", listener: () => void): void;
}
interface WsServer {
on(event: "connection", listener: (ws: WsClient) => void): void;
on(event: "error", listener: (err: Error) => void): void;
handleUpgrade(
request: IncomingMessage,
socket: Socket,
head: Buffer,
cb: (ws: WsClient) => void,
): void;
emit(event: string, ...args: unknown[]): void;
close(cb?: () => void): void;
}
// Load ws (bundled with pi) without needing @types/ws installed locally
const _require = createRequire(import.meta.url);
const wsModule = _require("ws") as {
WebSocketServer: new (opts: { noServer: boolean }) => WsServer;
OPEN: number;
};
const { WebSocketServer, OPEN } = wsModule;
export interface RemoteServer {
broadcast: (msg: object) => void;
sync: (ctx: ExtensionContext) => void;
stop: () => Promise<void>;
clientCount: () => number;
onClientChange: (cb: () => void) => void;
port: number;
token: string;
}
export async function startServer(
pi: ExtensionAPI,
ctx: ExtensionContext,
): Promise<RemoteServer> {
const config = await readRemoteControlConfig();
const bindAddr = config.bindAddress ?? "";
const { host: bindHost, port: bindPort } = bindAddr
? parseBindAddress(bindAddr)
: { host: "127.0.0.1", port: 0 };
const clientChangeListeners: Array<() => void> = [];
const clients = new Set<WsClient>();
const token = await loadOrCreateToken();
// Map of valid session IDs → expiry timestamp (ms since epoch)
const SESSION_TTL_MS = 86_400_000; // 24 h — matches cookie Max-Age
const validSessions = new Map<string, number>();
const pruneExpiredSessions = (): void => {
const now = Date.now();
for (const [id, expiresAt] of validSessions) {
if (expiresAt <= now) validSessions.delete(id);
}
};
/** Check if a request is authenticated (valid token query param OR valid session cookie) */
function isAuthenticated(req: IncomingMessage): boolean {
// Check session cookie first
const cookies = parseCookies(req.headers.cookie);
const sessionId = cookies[SESSION_COOKIE];
const sessionExpiry = sessionId ? validSessions.get(sessionId) : undefined;
if (sessionExpiry !== undefined && sessionExpiry > Date.now()) return true;
// Check token query param
const url = new URL(req.url ?? "/", "http://localhost");
const providedToken = url.searchParams.get("token");
if (providedToken && validateToken(providedToken, token)) return true;
return false;
}
function broadcast(msg: object): void {
const data = JSON.stringify(msg);
for (const client of clients) {
if (client.readyState === OPEN) {
try {
client.send(data);
} catch {
/* ignore */
}
}
}
}
function sync(currentCtx: ExtensionContext): void {
broadcast(buildSyncMessage(currentCtx));
}
const httpServer = createServer((req, res) => {
const url = new URL(req.url ?? "/", "http://localhost");
const pathname = url.pathname;
if (pathname === "/manifest.json") {
res.writeHead(200, { "Content-Type": "application/manifest+json; charset=utf-8" });
res.end(JSON.stringify({
name: "Pi Remote",
short_name: "Pi",
description: "Remote control for Pi sessions",
start_url: "/",
display: "standalone",
background_color: "#0d1117",
theme_color: "#0d1117",
icons: [{ src: "/icon.svg", sizes: "any", type: "image/svg+xml" }],
}));
return;
}
if (pathname === "/icon.svg") {
res.writeHead(200, { "Content-Type": "image/svg+xml", "Cache-Control": "public, max-age=86400" });
res.end(`<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 180 180"><rect width="180" height="180" rx="40" fill="#0d1117"/><text x="90" y="133" font-family="-apple-system,Helvetica,Arial,sans-serif" font-size="110" text-anchor="middle" fill="#3fb950">π</text></svg>`);
return;
}
if (pathname === "/" || pathname === "/index.html") {
// Check authentication
const cookies = parseCookies(req.headers.cookie);
const sc = cookies[SESSION_COOKIE];
const hasValidSession =
sc !== undefined && (validSessions.get(sc) ?? 0) > Date.now();
const providedToken = url.searchParams.get("token");
const hasValidToken =
providedToken && validateToken(providedToken, token);
if (!hasValidSession && !hasValidToken) {
res.writeHead(403, { "Content-Type": "text/plain; charset=utf-8" });
res.end(
"Forbidden — valid token required. Use the URL shown in the pi terminal.",
);
return;
}
// If authenticated via token (first visit), issue a session cookie and redirect to clean URL
if (!hasValidSession && hasValidToken) {
pruneExpiredSessions();
const sessionId = generateSessionId();
validSessions.set(sessionId, Date.now() + SESSION_TTL_MS);
res.writeHead(302, {
"Set-Cookie": `${SESSION_COOKIE}=${sessionId}; Path=/; HttpOnly; SameSite=Strict; Max-Age=86400`,
Location: "/",
});
res.end();
return;
}
// Valid session cookie — serve the page
const nonce = randomBytes(16).toString("base64");
res.writeHead(200, {
"Content-Type": "text/html; charset=utf-8",
"X-Frame-Options": "DENY",
"X-Content-Type-Options": "nosniff",
"Referrer-Policy": "no-referrer",
"Content-Security-Policy": `default-src 'none'; script-src 'nonce-${nonce}'; style-src 'nonce-${nonce}'; connect-src 'self'; manifest-src 'self'; base-uri 'none'`,
});
res.end(buildHTML(nonce));
} else {
res.writeHead(404, { "Content-Type": "text/plain; charset=utf-8" });
res.end("Not found");
}
});
const wss = new WebSocketServer({ noServer: true });
httpServer.on("error", (err: Error) => {
console.error("[remote-control] httpServer error:", err.message);
});
wss.on("error", (err: Error) => {
console.error("[remote-control] wss error:", err.message);
});
httpServer.on(
"upgrade",
(request: IncomingMessage, socket: Socket, head: Buffer) => {
const url = new URL(request.url, "http://localhost");
if (url.pathname === "/ws") {
// Validate auth: session cookie or token query param
if (!isAuthenticated(request)) {
socket.write("HTTP/1.1 403 Forbidden\r\n\r\n");
socket.destroy();
return;
}
wss.handleUpgrade(request, socket, head, (ws: WsClient) => {
wss.emit("connection", ws, request);
});
} else {
socket.destroy();
}
},
);
wss.on("connection", (ws: WsClient) => {
clients.add(ws);
for (const cb of clientChangeListeners) cb();
// Send full state snapshot to the new client
try {
ws.send(JSON.stringify(buildSyncMessage(ctx)));
} catch {
/* client disconnected before first send */
}
// Per-connection rate limiting: max 30 prompts per 60 seconds
const RATE_WINDOW_MS = 60_000;
const RATE_MAX = 30;
const MAX_MSG_BYTES = 64 * 1024;
const recentPrompts: number[] = [];
ws.on("message", (data: Buffer) => {
if (data.length > MAX_MSG_BYTES) return;
let msg: { type?: string; text?: string };
try {
const parsed: unknown = JSON.parse(data.toString());
if (typeof parsed !== "object" || parsed === null) return;
msg = parsed as { type?: string; text?: string };
} catch {
return;
}
if (msg.type === "stop") {
if (!ctx.isIdle()) {
ctx.abort();
}
return;
}
if (
msg.type === "prompt" &&
typeof msg.text === "string" &&
msg.text.trim()
) {
const text = msg.text.trim();
// Sliding-window rate limit
const now = Date.now();
const cutoff = now - RATE_WINDOW_MS;
while (recentPrompts.length > 0 && recentPrompts[0] < cutoff)
recentPrompts.shift();
if (recentPrompts.length >= RATE_MAX) return;
recentPrompts.push(now);
if (ctx.isIdle()) {
pi.sendUserMessage(text);
} else {
pi.sendUserMessage(text, { deliverAs: "followUp" });
}
}
});
const onClose = () => {
clients.delete(ws);
broadcast({ type: "status", clientCount: clients.size });
for (const cb of clientChangeListeners) cb();
};
ws.on("close", onClose);
ws.on("error", onClose);
});
return new Promise((resolve) => {
httpServer.listen(bindPort, bindHost, () => {
resolve({
broadcast,
sync,
stop: () =>
new Promise<void>((res) => {
// Forcefully kill all WebSocket clients — terminate() sends no
// close frame and doesn't wait for the remote end to acknowledge,
// so it can't hang on an unresponsive client.
for (const client of clients) {
try {
client.terminate();
} catch {
/* ignore */
}
}
clients.clear();
// Safety timeout — if wss/http shutdown callbacks never fire
// (e.g. lingering keep-alive sockets), resolve anyway so the
// session_shutdown handler doesn't block pi from exiting.
const timeout = setTimeout(() => {
httpServer.close(() => {});
httpServer.closeAllConnections?.();
res();
}, 2000);
wss.close(() =>
httpServer.close(() => {
clearTimeout(timeout);
res();
}),
);
}),
clientCount: () => clients.size,
onClientChange: (cb: () => void) => {
clientChangeListeners.push(cb);
},
get port() {
return (httpServer.address() as AddressInfo | null)?.port ?? 0;
},
get token() {
return token;
},
});
});
});
}
export { startServer } from "./server/server.js";
// LEGACY: re-exports for backward-compatibility
export type { RemoteServer } from "./server/types.js";

0
extensions/remote-control/server/routes/.gitkeep Normal file
View File

333
extensions/remote-control/server/server.ts Normal file
View File

@ -0,0 +1,333 @@
/**
* HTTP + WebSocket server bootstrap for remote-control.
*
* Responsible for:
* - Creating and configuring the HTTP server (middleware, TLS in future T-1.3)
* - Wiring up the WebSocket upgrade handler (see upgrade.ts)
* - Serving LEGACY routes for the browser HTML client (html.ts)
* - Managing the connected-client set and the broadcast helper
* - Returning a RemoteServer handle to the extension entry point (index.ts)
*
* LEGACY routes (kept until Phase 2 ships and html.ts is retired):
* GET / — serves the browser HTML UI (buildHTML from html.ts)
* GET /index.html — same as /
* GET /manifest.json — PWA manifest
* GET /icon.svg — PWA icon
* WS /ws — WebSocket endpoint for the browser client
*
* Future route modules live under routes/ and are wired here once they
* land in T-1.5, T-1.6, T-1.7.
*/
import { randomBytes } from "node:crypto";
import type { IncomingMessage } from "node:http";
import { createServer } from "node:http";
import { createRequire } from "node:module";
import type { AddressInfo } from "node:net";
import type {
ExtensionAPI,
ExtensionContext,
} from "@earendil-works/pi-coding-agent";
import {
generateSessionId,
loadOrCreateToken,
parseCookies,
SESSION_COOKIE,
validateToken,
} from "../auth.js";
import { parseBindAddress, readRemoteControlConfig } from "../config.js";
import { buildHTML } from "../html.js"; // LEGACY: browser HTML client
import { buildSyncMessage } from "../messages.js";
import type { RemoteServer, WsClient, WsServer } from "./types.js";
import { createUpgradeHandler } from "./upgrade.js";
// ---------------------------------------------------------------------------
// Load ws (bundled with pi) without needing @types/ws installed locally
// ---------------------------------------------------------------------------
const _require = createRequire(import.meta.url);
const wsModule = _require("ws") as {
WebSocketServer: new (opts: { noServer: boolean }) => WsServer;
OPEN: number;
};
const { WebSocketServer, OPEN } = wsModule;
// ---------------------------------------------------------------------------
// startServer
// ---------------------------------------------------------------------------
export async function startServer(
pi: ExtensionAPI,
ctx: ExtensionContext,
): Promise<RemoteServer> {
const config = await readRemoteControlConfig();
const bindAddr = config.bindAddress ?? "";
const { host: bindHost, port: bindPort } = bindAddr
? parseBindAddress(bindAddr)
: { host: "127.0.0.1", port: 0 };
const clientChangeListeners: Array<() => void> = [];
const clients = new Set<WsClient>();
const token = await loadOrCreateToken();
// Map of valid session IDs → expiry timestamp (ms since epoch)
const SESSION_TTL_MS = 86_400_000; // 24 h — matches cookie Max-Age
const validSessions = new Map<string, number>();
const pruneExpiredSessions = (): void => {
const now = Date.now();
for (const [id, expiresAt] of validSessions) {
if (expiresAt <= now) validSessions.delete(id);
}
};
/** Check if a request is authenticated (valid token query param OR valid session cookie) */
function isAuthenticated(req: IncomingMessage): boolean {
// Check session cookie first
const cookies = parseCookies(req.headers.cookie);
const sessionId = cookies[SESSION_COOKIE];
const sessionExpiry = sessionId ? validSessions.get(sessionId) : undefined;
if (sessionExpiry !== undefined && sessionExpiry > Date.now()) return true;
// Check token query param
const url = new URL(req.url ?? "/", "http://localhost");
const providedToken = url.searchParams.get("token");
if (providedToken && validateToken(providedToken, token)) return true;
return false;
}
function broadcast(msg: object): void {
const data = JSON.stringify(msg);
for (const client of clients) {
if (client.readyState === OPEN) {
try {
client.send(data);
} catch {
/* ignore */
}
}
}
}
function sync(currentCtx: ExtensionContext): void {
broadcast(buildSyncMessage(currentCtx));
}
// ── HTTP server ─────────────────────────────────────────────────────────
const httpServer = createServer((req, res) => {
const url = new URL(req.url ?? "/", "http://localhost");
const pathname = url.pathname;
// LEGACY: PWA manifest — served for the browser HTML client
if (pathname === "/manifest.json") {
res.writeHead(200, {
"Content-Type": "application/manifest+json; charset=utf-8",
});
res.end(
JSON.stringify({
name: "Pi Remote",
short_name: "Pi",
description: "Remote control for Pi sessions",
start_url: "/",
display: "standalone",
background_color: "#0d1117",
theme_color: "#0d1117",
icons: [{ src: "/icon.svg", sizes: "any", type: "image/svg+xml" }],
}),
);
return;
}
// LEGACY: PWA icon — served for the browser HTML client
if (pathname === "/icon.svg") {
res.writeHead(200, {
"Content-Type": "image/svg+xml",
"Cache-Control": "public, max-age=86400",
});
res.end(
`<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 180 180"><rect width="180" height="180" rx="40" fill="#0d1117"/><text x="90" y="133" font-family="-apple-system,Helvetica,Arial,sans-serif" font-size="110" text-anchor="middle" fill="#3fb950">π</text></svg>`,
);
return;
}
// LEGACY: browser HTML client root route
if (pathname === "/" || pathname === "/index.html") {
const cookies = parseCookies(req.headers.cookie);
const sc = cookies[SESSION_COOKIE];
const hasValidSession =
sc !== undefined && (validSessions.get(sc) ?? 0) > Date.now();
const providedToken = url.searchParams.get("token");
const hasValidToken =
providedToken && validateToken(providedToken, token);
if (!hasValidSession && !hasValidToken) {
res.writeHead(403, { "Content-Type": "text/plain; charset=utf-8" });
res.end(
"Forbidden — valid token required. Use the URL shown in the pi terminal.",
);
return;
}
// If authenticated via token (first visit), issue a session cookie and redirect to clean URL
if (!hasValidSession && hasValidToken) {
pruneExpiredSessions();
const sessionId = generateSessionId();
validSessions.set(sessionId, Date.now() + SESSION_TTL_MS);
res.writeHead(302, {
"Set-Cookie": `${SESSION_COOKIE}=${sessionId}; Path=/; HttpOnly; SameSite=Strict; Max-Age=86400`,
Location: "/",
});
res.end();
return;
}
// Valid session cookie — serve the browser HTML UI
const nonce = randomBytes(16).toString("base64");
res.writeHead(200, {
"Content-Type": "text/html; charset=utf-8",
"X-Frame-Options": "DENY",
"X-Content-Type-Options": "nosniff",
"Referrer-Policy": "no-referrer",
"Content-Security-Policy": `default-src 'none'; script-src 'nonce-${nonce}'; style-src 'nonce-${nonce}'; connect-src 'self'; manifest-src 'self'; base-uri 'none'`,
});
res.end(buildHTML(nonce)); // LEGACY: renders the browser HTML client
return;
}
res.writeHead(404, { "Content-Type": "text/plain; charset=utf-8" });
res.end("Not found");
});
// ── WebSocket server ────────────────────────────────────────────────────
const wss = new WebSocketServer({ noServer: true });
httpServer.on("error", (err: Error) => {
console.error("[remote-control] httpServer error:", err.message);
});
wss.on("error", (err: Error) => {
console.error("[remote-control] wss error:", err.message);
});
// Wire upgrade handler (see upgrade.ts for path routing)
httpServer.on("upgrade", createUpgradeHandler(wss, isAuthenticated));
// ── LEGACY: browser HTML client WebSocket connection handling ───────────
wss.on("connection", (ws: WsClient) => {
clients.add(ws);
for (const cb of clientChangeListeners) cb();
// Send full state snapshot to the new client
try {
ws.send(JSON.stringify(buildSyncMessage(ctx)));
} catch {
/* client disconnected before first send */
}
// Per-connection rate limiting: max 30 prompts per 60 seconds
const RATE_WINDOW_MS = 60_000;
const RATE_MAX = 30;
const MAX_MSG_BYTES = 64 * 1024;
const recentPrompts: number[] = [];
ws.on("message", (data: Buffer) => {
if (data.length > MAX_MSG_BYTES) return;
let msg: { type?: string; text?: string };
try {
const parsed: unknown = JSON.parse(data.toString());
if (typeof parsed !== "object" || parsed === null) return;
msg = parsed as { type?: string; text?: string };
} catch {
return;
}
if (msg.type === "stop") {
if (!ctx.isIdle()) {
ctx.abort();
}
return;
}
if (
msg.type === "prompt" &&
typeof msg.text === "string" &&
msg.text.trim()
) {
const text = msg.text.trim();
// Sliding-window rate limit
const now = Date.now();
const cutoff = now - RATE_WINDOW_MS;
while (recentPrompts.length > 0 && recentPrompts[0] < cutoff)
recentPrompts.shift();
if (recentPrompts.length >= RATE_MAX) return;
recentPrompts.push(now);
if (ctx.isIdle()) {
pi.sendUserMessage(text);
} else {
pi.sendUserMessage(text, { deliverAs: "followUp" });
}
}
});
const onClose = (): void => {
clients.delete(ws);
broadcast({ type: "status", clientCount: clients.size });
for (const cb of clientChangeListeners) cb();
};
ws.on("close", onClose);
ws.on("error", onClose);
});
// ── Listen ───────────────────────────────────────────────────────────────
return new Promise((resolve) => {
httpServer.listen(bindPort, bindHost, () => {
resolve({
broadcast,
sync,
stop: () =>
new Promise<void>((res) => {
// Forcefully kill all WebSocket clients — terminate() sends no
// close frame and doesn't wait for the remote end to acknowledge,
// so it can't hang on an unresponsive client.
for (const client of clients) {
try {
client.terminate();
} catch {
/* ignore */
}
}
clients.clear();
// Safety timeout — if wss/http shutdown callbacks never fire
// (e.g. lingering keep-alive sockets), resolve anyway so the
// session_shutdown handler doesn't block pi from exiting.
const timeout = setTimeout(() => {
httpServer.close(() => {});
httpServer.closeAllConnections?.();
res();
}, 2000);
wss.close(() =>
httpServer.close(() => {
clearTimeout(timeout);
res();
}),
);
}),
clientCount: () => clients.size,
onClientChange: (cb: () => void) => {
clientChangeListeners.push(cb);
},
get port() {
return (httpServer.address() as AddressInfo | null)?.port ?? 0;
},
get token() {
return token;
},
});
});
});
}

51
extensions/remote-control/server/types.ts Normal file
View File

@ -0,0 +1,51 @@
/**
* Shared type definitions for the remote-control server.
*
* These interfaces describe the WebSocket client/server shapes loaded
* dynamically from the `ws` module bundled with pi, and the public surface
* of the HTTP+WS server returned to callers.
*/
import type { IncomingMessage } from "node:http";
import type { Socket } from "node:net";
// ---------------------------------------------------------------------------
// WebSocket interfaces (ws module loaded via createRequire — no @types/ws)
// ---------------------------------------------------------------------------
export interface WsClient {
readyState: number;
send(data: string): void;
terminate(): void;
on(event: "message", listener: (data: Buffer) => void): void;
on(event: "close" | "error", listener: () => void): void;
}
export interface WsServer {
on(event: "connection", listener: (ws: WsClient) => void): void;
on(event: "error", listener: (err: Error) => void): void;
handleUpgrade(
request: IncomingMessage,
socket: Socket,
head: Buffer,
cb: (ws: WsClient) => void,
): void;
emit(event: string, ...args: unknown[]): void;
close(cb?: () => void): void;
}
// ---------------------------------------------------------------------------
// Public server handle
// ---------------------------------------------------------------------------
export interface RemoteServer {
broadcast: (msg: object) => void;
sync: (
ctx: import("@earendil-works/pi-coding-agent").ExtensionContext,
) => void;
stop: () => Promise<void>;
clientCount: () => number;
onClientChange: (cb: () => void) => void;
port: number;
token: string;
}

52
extensions/remote-control/server/upgrade.ts Normal file
View File

@ -0,0 +1,52 @@
/**
* WebSocket upgrade routing.
*
* Routes incoming HTTP Upgrade requests to the appropriate WebSocket handler
* based on the request path and session/topic. Non-matching paths are
* destroyed immediately.
*
* Current routes (all LEGACY — browser HTML client):
* /ws → legacy browser client WebSocket endpoint
*
* Future routes (T-1.5):
* /sessions/:id/stream → binary ANSI stream per tmux session
*
* T-1.5 will extend createUpgradeHandler to accept a session registry and
* dispatch to per-session stream handlers.
*/
import type { IncomingMessage } from "node:http";
import type { Socket } from "node:net";
import type { WsClient, WsServer } from "./types.js";
/**
* Create the HTTP `upgrade` event handler.
*
* @param wss - WebSocket server instance.
* @param isAuthenticated - Predicate that checks token/session on the request.
* @returns A handler suitable for `httpServer.on("upgrade", handler)`.
*/
export function createUpgradeHandler(
wss: WsServer,
isAuthenticated: (req: IncomingMessage) => boolean,
): (request: IncomingMessage, socket: Socket, head: Buffer) => void {
return (request: IncomingMessage, socket: Socket, head: Buffer): void => {
const url = new URL(request.url ?? "/", "http://localhost");
if (url.pathname === "/ws") {
// LEGACY: browser HTML client WebSocket endpoint — auth guard
if (!isAuthenticated(request)) {
socket.write("HTTP/1.1 403 Forbidden\r\n\r\n");
socket.destroy();
return;
}
wss.handleUpgrade(request, socket, head, (ws: WsClient) => {
wss.emit("connection", ws, request);
});
return;
}
// Unknown upgrade path — reject
socket.destroy();
};
}

3
package.json
View File

@ -17,7 +17,8 @@
"scripts": {
"lint": "biome check --write .",
"lint:check": "biome check .",
"prepare": "node .husky/install.mjs"
"prepare": "node .husky/install.mjs",
"smoke": "node --test scripts/smoke/smoke.mjs"
},
"devDependencies": {
"@biomejs/biome": "^2.4.12",

96
scripts/smoke/README.md Normal file
View File

@ -0,0 +1,96 @@
# Smoke Test Harness
End-to-end smoke tests for the `remote-control` extension. These tests spawn
a real `pi` subprocess with the extension loaded and hit the HTTP/WebSocket
endpoints to verify they respond correctly.
## Running
```sh
npm run smoke
```
Or directly:
```sh
node --test scripts/smoke/smoke.mjs
```
## What is tested (MVP — T-1.0a)
| # | Test | Description |
|---|------|-------------|
| 1 | Server starts | pi spawns with `--remote-control`; TCP port opens within 12 s |
| 2 | `GET /manifest.json` | 200, `application/manifest+json`, parses, has `name`/`short_name`/`start_url` |
| 3 | `GET /icon.svg` | 200, `image/svg+xml`, body starts with `<svg` |
| 4 | `GET /` (with token) | 302→200, `text/html`, contains HTML marker |
| 5 | `GET /` (no token) | 403 Forbidden |
| 6 | `WS /ws` (with token) | 101 upgrade, WebSocket enters OPEN state |
| 7 | Process alive | pi hasn't crashed during the test run |
| T | Teardown | SIGTERM pi, wait for exit, remove temp HOME |
## Port
Default port: **19876**. Override with the `SMOKE_PORT` environment variable:
```sh
SMOKE_PORT=20000 npm run smoke
```
## How it works
The harness creates a temporary `HOME` directory containing:
- `$HOME/.pi/remote-control/config.json` — `publicBaseUrl` + `bindAddress` pointing at `SMOKE_PORT`
- `$HOME/.pi/remote-control/token` — a known deterministic token for auth
`pi` is spawned with `HOME=<tmpdir>` so **no real `~/.pi` files are read or
written**. The temp directory is deleted in the `after()` teardown hook even
if tests fail.
## Authentication
The server requires either a valid session cookie or `?token=<value>` query
parameter. The smoke tests pre-seed a known token (`smoke-test-token-deterministic-1234`)
and pass it in the `?token=` parameter. No production auth logic is modified.
## Adding new tests
1. Create a new file `scripts/smoke/<feature>.test.mjs`
2. Import helpers from `./helpers.mjs`
3. Import your test file from `smoke.mjs` (or run it standalone with `node --test`)
Example:
```js
// scripts/smoke/stream.test.mjs
import { describe, it } from "node:test";
import { openWebSocket, closeWebSocket } from "./helpers.mjs";
export function registerStreamTests(port, token) {
describe("stream tests", () => {
it("WS /sessions/:id/stream → 101", async () => {
// ...
});
});
}
```
## Flags used
Same flags as `make dev`:
```
pi -nt -ne -ns -np -nc --no-session --offline -e extensions/remote-control --remote-control
```
## Troubleshooting
**Port busy**: if `SMOKE_PORT` is already in use, the test will fail at the
`waitForPort` step. Change the port: `SMOKE_PORT=20001 npm run smoke`.
**pi not found**: ensure `pi` is on your `PATH`. Check with `which pi`.
**Server not starting**: run with verbose output to see what pi logs:
```sh
node --test --reporter spec scripts/smoke/smoke.mjs
```
The port-wait error will include captured stdout/stderr from pi.

267
scripts/smoke/helpers.mjs Normal file
View File

@ -0,0 +1,267 @@
/**
* Smoke-test helpers — spawn-pi, wait-for-port, fetch, WebSocket.
*
* Designed to be imported by smoke.mjs and any future test files
* (e.g. scripts/smoke/stream.test.mjs). All helpers are stateless;
* the caller is responsible for lifecycle.
*/
import { spawn } from "node:child_process";
import fs from "node:fs/promises";
import { createRequire } from "node:module";
import net from "node:net";
import os from "node:os";
import path from "node:path";
// ── WebSocket (from project's bundled ws dependency) ─────────────────────────
const _require = createRequire(import.meta.url);
/** @type {import("ws")} */
const wsModule = _require("ws");
const WebSocket = wsModule.WebSocket ?? wsModule.default ?? wsModule;
// ── Constants ─────────────────────────────────────────────────────────────────
/** Milliseconds to wait for the TCP port to open before giving up. */
const PORT_READY_TIMEOUT_MS = 12_000;
/** Milliseconds between TCP-port-ready poll attempts. */
const PORT_POLL_INTERVAL_MS = 100;
// ── Temporary HOME setup ──────────────────────────────────────────────────────
/**
* Create a minimal fake HOME directory that satisfies remote-control's config
* requirements:
* $HOME/.pi/remote-control/config.json — publicBaseUrl + bindAddress
* $HOME/.pi/remote-control/token — known token for auth
*
* @param {object} opts
* @param {number} opts.port - TCP port to bind to.
* @param {string} opts.token - Auth token to pre-seed.
* @returns {Promise<string>} Path to the temp HOME directory.
*/
export async function createSmokeHome({ port, token }) {
const tmpBase = await fs.mkdtemp(path.join(os.tmpdir(), "pi-smoke-"));
const rcDir = path.join(tmpBase, ".pi", "remote-control");
await fs.mkdir(rcDir, { recursive: true });
const config = {
publicBaseUrl: `http://127.0.0.1:${port}`,
bindAddress: `127.0.0.1:${port}`,
};
await fs.writeFile(
path.join(rcDir, "config.json"),
`${JSON.stringify(config, null, 2)}\n`,
"utf8",
);
await fs.writeFile(path.join(rcDir, "token"), token, {
encoding: "utf8",
mode: 0o600,
});
return tmpBase;
}
/**
* Remove the temp HOME directory created by createSmokeHome.
* Silently ignores errors (e.g. already deleted).
*
* @param {string} tmpHome
*/
export async function removeSmokeHome(tmpHome) {
try {
await fs.rm(tmpHome, { recursive: true, force: true });
} catch {
/* ignore */
}
}
// ── Pi subprocess ─────────────────────────────────────────────────────────────
/**
* Spawn a `pi` process with the remote-control extension loaded.
*
* Uses the same flags as `make dev` plus `--remote-control` to auto-start
* the HTTP server. Sets HOME to the given fake home so config/token paths
* resolve correctly without touching the real ~/.pi directory.
*
* Pi is a TUI application that requires a PTY (pseudo-terminal) to enter its
* interactive session mode and fire `session_start`. We use Python's built-in
* `pty.spawn` to allocate a PTY for pi; this is available on all macOS/Linux
* systems without any additional dependencies.
*
* @param {object} opts
* @param {string} opts.extensionPath - Absolute path to the extension dir.
* @param {string} opts.fakeHome - Path to the fake HOME directory.
* @returns {{ proc: import("node:child_process").ChildProcess, logs: string[] }}
*/
export function spawnPi({ extensionPath, fakeHome }) {
const logs = [];
// Python script that allocates a PTY and spawns pi inside it.
// Using pty.spawn() keeps the child alive (no EOF injection like `script` does).
const pyScript = [
"import os, pty",
`os.environ['HOME'] = ${JSON.stringify(fakeHome)}`,
`pty.spawn(['pi', '-nt', '-ne', '-ns', '-np', '-nc', '--no-session', '--offline', '-e', ${JSON.stringify(extensionPath)}, '--remote-control'])`,
].join("\n");
const proc = spawn("python3", ["-c", pyScript], {
env: { ...process.env, HOME: fakeHome },
stdio: ["ignore", "pipe", "pipe"],
});
const capture = (chunk) => {
logs.push(chunk.toString());
};
proc.stdout.on("data", capture);
proc.stderr.on("data", capture);
return { proc, logs };
}
/**
* Kill a pi process and wait for it to exit.
* Sends SIGTERM first; if it doesn't exit within 3 s, sends SIGKILL.
*
* @param {import("node:child_process").ChildProcess} proc
* @returns {Promise<number|null>} The exit code (or null if killed by signal).
*/
export function killPi(proc) {
return new Promise((resolve) => {
if (proc.exitCode !== null) {
resolve(proc.exitCode);
return;
}
let settled = false;
const settle = (code) => {
if (!settled) {
settled = true;
clearTimeout(sigkillTimer);
resolve(code ?? null);
}
};
proc.once("exit", (code) => settle(code));
proc.kill("SIGTERM");
// Safety: escalate to SIGKILL after 3 s
const sigkillTimer = setTimeout(() => {
try {
proc.kill("SIGKILL");
} catch {
/* already dead */
}
}, 3_000);
});
}
// ── Port readiness ────────────────────────────────────────────────────────────
/**
* Poll a TCP port until it accepts connections or the timeout expires.
*
* @param {object} opts
* @param {number} opts.port
* @param {string} [opts.host] - Default "127.0.0.1".
* @param {number} [opts.timeoutMs] - Default PORT_READY_TIMEOUT_MS.
* @param {number} [opts.intervalMs] - Default PORT_POLL_INTERVAL_MS.
* @returns {Promise<void>} Resolves when port is open; rejects on timeout.
*/
export function waitForPort({
port,
host = "127.0.0.1",
timeoutMs = PORT_READY_TIMEOUT_MS,
intervalMs = PORT_POLL_INTERVAL_MS,
}) {
return new Promise((resolve, reject) => {
const deadline = Date.now() + timeoutMs;
const attempt = () => {
const sock = net.createConnection({ port, host });
sock.once("connect", () => {
sock.destroy();
resolve();
});
sock.once("error", () => {
sock.destroy();
if (Date.now() >= deadline) {
reject(
new Error(
`Port ${host}:${port} did not open within ${timeoutMs} ms`,
),
);
} else {
setTimeout(attempt, intervalMs);
}
});
};
attempt();
});
}
// ── HTTP helpers ──────────────────────────────────────────────────────────────
/**
* Build the base URL for the smoke server.
*
* @param {number} port
* @returns {string}
*/
export function baseUrl(port) {
return `http://127.0.0.1:${port}`;
}
/**
* Fetch a URL and return the response plus text body.
* Throws if the network request itself fails.
*
* @param {string} url
* @param {RequestInit} [init]
* @returns {Promise<{ res: Response; body: string }>}
*/
export async function fetchText(url, init) {
const res = await fetch(url, init);
const body = await res.text();
return { res, body };
}
// ── WebSocket helpers ─────────────────────────────────────────────────────────
/**
* Open a WebSocket connection and wait for either an `open` or `error` event.
* Resolves with the WebSocket instance on success.
* Rejects with the error on failure.
*
* @param {string} url - WebSocket URL (ws:// or wss://).
* @returns {Promise<import("ws").WebSocket>}
*/
export function openWebSocket(url) {
return new Promise((resolve, reject) => {
const ws = new WebSocket(url);
ws.once("open", () => resolve(ws));
ws.once("error", reject);
});
}
/**
* Close a WebSocket and wait for the close event.
*
* @param {import("ws").WebSocket} ws
* @returns {Promise<void>}
*/
export function closeWebSocket(ws) {
return new Promise((resolve) => {
if (ws.readyState === ws.constructor.CLOSED) {
resolve();
return;
}
ws.once("close", () => resolve());
ws.close();
});
}

249
scripts/smoke/smoke.mjs Normal file
View File

@ -0,0 +1,249 @@
/**
* Smoke test — end-to-end validation of the remote-control HTTP/WS server.
*
* Runs with: node --test scripts/smoke/smoke.mjs
* Or: npm run smoke
*
* What is tested (MVP):
* 1. Server starts — pi spawns with --remote-control and the HTTP port opens.
* 2. GET /manifest.json — 200, application/manifest+json, parses, has name/short_name/start_url.
* 3. GET /icon.svg — 200, image/svg+xml, body starts with <svg.
* 4. GET / — 200 (with auth token), text/html, contains HTML marker.
* 5. WS /ws — successful 101 upgrade (with auth token). Close cleanly.
* 6. Teardown — SIGTERM pi, process exits.
*
* Auth strategy:
* The server requires a token on GET / and WS /ws. We pre-seed a known token
* in a temporary HOME directory ($HOME/.pi/remote-control/token) and pass it
* via the ?token= query parameter using the existing mechanism. No production
* code is modified.
*
* Port strategy:
* We write bindAddress + publicBaseUrl to the temporary config so the server
* binds to a deterministic port (SMOKE_PORT env, default 19876). No real
* ~/.pi files are touched.
*/
import assert from "node:assert/strict";
import path from "node:path";
import { after, before, describe, it } from "node:test";
import { fileURLToPath } from "node:url";
import {
baseUrl,
closeWebSocket,
createSmokeHome,
fetchText,
killPi,
openWebSocket,
removeSmokeHome,
spawnPi,
waitForPort,
} from "./helpers.mjs";
// ── Configuration ─────────────────────────────────────────────────────────────
const SMOKE_PORT = Number(process.env.SMOKE_PORT ?? 19876);
const SMOKE_TOKEN = "smoke-test-token-deterministic-1234";
const EXTENSION_PATH = path.resolve(
path.dirname(fileURLToPath(import.meta.url)),
"../../extensions/remote-control",
);
// ── Suite ─────────────────────────────────────────────────────────────────────
describe("remote-control smoke tests", () => {
/** @type {import("node:child_process").ChildProcess} */
let proc;
/** @type {string} */
let fakeHome;
/** @type {string[]} */
let logs;
before(async () => {
// Create temp HOME with config + token
fakeHome = await createSmokeHome({ port: SMOKE_PORT, token: SMOKE_TOKEN });
// Spawn pi
({ proc, logs } = spawnPi({ extensionPath: EXTENSION_PATH, fakeHome }));
// Wait for the HTTP server to open the TCP port
try {
await waitForPort({ port: SMOKE_PORT });
} catch (err) {
// Attach captured logs to the error for easy debugging
const captured = logs.join("");
throw new Error(
`${err.message}\n\n--- pi stdout/stderr ---\n${captured || "(empty)"}`,
);
}
});
after(async () => {
// Always clean up — kill pi and remove temp HOME
if (proc) {
await killPi(proc);
}
if (fakeHome) {
await removeSmokeHome(fakeHome);
}
});
// ── Test 1: manifest.json ───────────────────────────────────────────────────
it("GET /manifest.json → 200 application/manifest+json with required fields", async () => {
const { res, body } = await fetchText(
`${baseUrl(SMOKE_PORT)}/manifest.json`,
);
assert.equal(res.status, 200, `Expected 200, got ${res.status}`);
const ct = res.headers.get("content-type") ?? "";
assert.ok(
ct.includes("application/manifest+json"),
`Expected content-type application/manifest+json, got: ${ct}`,
);
let manifest;
try {
manifest = JSON.parse(body);
} catch {
assert.fail(
`manifest.json body is not valid JSON: ${body.slice(0, 200)}`,
);
}
assert.ok(
typeof manifest.name === "string" && manifest.name.length > 0,
"manifest.name must be a non-empty string",
);
assert.ok(
typeof manifest.short_name === "string" && manifest.short_name.length > 0,
"manifest.short_name must be a non-empty string",
);
assert.ok(
typeof manifest.start_url === "string" && manifest.start_url.length > 0,
"manifest.start_url must be a non-empty string",
);
});
// ── Test 2: icon.svg ────────────────────────────────────────────────────────
it("GET /icon.svg → 200 image/svg+xml, body starts with <svg", async () => {
const { res, body } = await fetchText(`${baseUrl(SMOKE_PORT)}/icon.svg`);
assert.equal(res.status, 200, `Expected 200, got ${res.status}`);
const ct = res.headers.get("content-type") ?? "";
assert.ok(
ct.includes("image/svg+xml"),
`Expected content-type image/svg+xml, got: ${ct}`,
);
assert.ok(
body.trimStart().startsWith("<svg"),
`Expected body to start with <svg, got: ${body.slice(0, 100)}`,
);
});
// ── Test 3: root with auth ──────────────────────────────────────────────────
it("GET / (with token) → 200 text/html containing HTML marker", async () => {
const url = `${baseUrl(SMOKE_PORT)}/?token=${SMOKE_TOKEN}`;
// The server issues a 302 redirect on first token auth (sets session cookie),
// then serves 200 on the redirect target. Follow redirects manually to keep
// cookies across the redirect chain.
const cookieJar = {};
// First request: expect 302
const first = await fetch(url, { redirect: "manual" });
let body;
if (first.status === 302) {
// Extract session cookie
for (const [name, value] of first.headers.entries()) {
if (name.toLowerCase() === "set-cookie") {
const match = value.match(/^([^=]+)=([^;]+)/);
if (match) cookieJar[match[1]] = match[2];
}
}
const location = first.headers.get("location") ?? "/";
const cookieHeader = Object.entries(cookieJar)
.map(([k, v]) => `${k}=${v}`)
.join("; ");
const second = await fetch(`${baseUrl(SMOKE_PORT)}${location}`, {
headers: cookieHeader ? { Cookie: cookieHeader } : {},
});
assert.equal(
second.status,
200,
`Expected 200 on redirect target, got ${second.status}`,
);
const ct = second.headers.get("content-type") ?? "";
assert.ok(
ct.includes("text/html"),
`Expected content-type text/html, got: ${ct}`,
);
body = await second.text();
} else {
assert.equal(first.status, 200, `Expected 200, got ${first.status}`);
const ct = first.headers.get("content-type") ?? "";
assert.ok(
ct.includes("text/html"),
`Expected content-type text/html, got: ${ct}`,
);
body = await first.text();
}
// Check for a marker from html.ts — the title is always present
assert.ok(
body.includes("<!DOCTYPE html>"),
"Expected body to contain <!DOCTYPE html>",
);
assert.ok(
body.includes("remote-control") ||
body.includes("Pi Remote") ||
body.includes("π"),
"Expected body to contain a known HTML marker (remote-control / Pi Remote / π)",
);
});
// ── Test 4: 403 on unauthenticated root ─────────────────────────────────────
it("GET / (no token) → 403 Forbidden", async () => {
const { res } = await fetchText(`${baseUrl(SMOKE_PORT)}/`);
assert.equal(
res.status,
403,
`Expected 403 without token, got ${res.status}`,
);
});
// ── Test 5: WebSocket upgrade ───────────────────────────────────────────────
it("WS /ws (with token) → successful 101 upgrade", async () => {
const wsUrl = `ws://127.0.0.1:${SMOKE_PORT}/ws?token=${SMOKE_TOKEN}`;
// First, get a session cookie via HTTP (WS token auth also works directly
// since isAuthenticated checks URL searchParams on the Upgrade request)
const ws = await openWebSocket(wsUrl);
assert.ok(
ws.readyState === ws.constructor.OPEN,
"WebSocket should be in OPEN state after successful handshake",
);
await closeWebSocket(ws);
});
// ── Test 6: teardown (handled in after(), verified here) ───────────────────
it("pi process is alive during tests", () => {
assert.equal(
proc.exitCode,
null,
"pi process should still be running during tests",
);
});
});