diff --git a/extensions/remote-control/server/routes/stream.ts b/extensions/remote-control/server/routes/stream.ts index 2fad941..3fd3dd7 100644 --- a/extensions/remote-control/server/routes/stream.ts +++ b/extensions/remote-control/server/routes/stream.ts @@ -22,6 +22,7 @@ import { readChunks } from "../../buffer/reader.js"; import type { StateEvent } from "../../pi/events.js"; import { SequenceCounter } from "../../sequence.js"; import { ControlClient } from "../../tmux/control.js"; +import { sendKey, sendKeys, sendPaste } from "../../tmux/input.js"; import { resizeSession } from "../../tmux/manager.js"; import { capturePane } from "../../tmux/snapshot.js"; import type { WsClient, WsServer } from "../types.js"; @@ -162,6 +163,54 @@ function handleStreamConnection( const cols = typeof m.cols === "number" ? m.cols : 80; const rows = typeof m.rows === "number" ? m.rows : 24; resizeSession(sessionId, cols, rows).catch(() => {}); + } else if (m.type === "keys") { + if (typeof m.data !== "string") { + sendJson(ws, { + type: "error", + code: "bad_input", + message: "keys.data must be a string", + }); + return; + } + sendKeys(sessionId, m.data).catch((err) => { + sendJson(ws, { + type: "error", + code: "bad_input", + message: `Failed to send keys: ${String(err)}`, + }); + }); + } else if (m.type === "key") { + if (typeof m.name !== "string") { + sendJson(ws, { + type: "error", + code: "bad_input", + message: "key.name must be a string", + }); + return; + } + sendKey(sessionId, m.name).catch((err) => { + sendJson(ws, { + type: "error", + code: "bad_input", + message: `Failed to send key: ${String(err)}`, + }); + }); + } else if (m.type === "paste") { + if (typeof m.data !== "string") { + sendJson(ws, { + type: "error", + code: "bad_input", + message: "paste.data must be a string", + }); + return; + } + sendPaste(sessionId, m.data).catch((err) => { + sendJson(ws, { + type: "error", + code: "bad_input", + message: `Failed to send paste: ${String(err)}`, + }); + }); } else if (m.type === "snapshot-request") { capturePane({ session: sessionId, escapes: true }) .then((text) => { diff --git a/scripts/smoke/stream.test.mjs b/scripts/smoke/stream.test.mjs index 243bff0..1411bbe 100644 --- a/scripts/smoke/stream.test.mjs +++ b/scripts/smoke/stream.test.mjs @@ -173,6 +173,40 @@ describe("T-1.8 stream integration", () => { assert.ok(found, `Should observe marker "${marker}" in stream output`); }); + it("WS stream: send keys via WS, observe in stream output", async () => { + const ws = await openWebSocket( + `${WS_BASE}/sessions/${sessionId}/stream${AUTH}`, + ); + ws.send(JSON.stringify({ type: "resume", lastSeq: null })); + + const marker = `ws-smoke-${Date.now()}`; + + // Wait a tick to let resume complete, then send keys via WS + await new Promise((r) => setTimeout(r, 200)); + ws.send(JSON.stringify({ type: "keys", data: `echo ${marker}` })); + ws.send(JSON.stringify({ type: "key", name: "enter" })); + + const found = await new Promise((resolve) => { + const timeout = setTimeout(() => resolve(false), 6000); + let accumulated = ""; + ws.on("message", (data) => { + if (Buffer.isBuffer(data) && data.length > 8) { + accumulated += data.slice(8).toString("utf8"); + if (accumulated.includes(marker)) { + clearTimeout(timeout); + resolve(true); + } + } + }); + }); + + await closeWebSocket(ws); + assert.ok( + found, + `Should observe marker "${marker}" sent via WS keys message in stream output`, + ); + }); + it("WS stream: reconnect with lastSeq → receives only delta", async () => { // First pass: collect frames and note highest seq const ws1 = await openWebSocket(