feat(dashboard): cli-agent session transport — WS attach, tickets, SSE state, output hardening (U10)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
14
.changeset/cli-agent-session-transport.md
Normal file
14
.changeset/cli-agent-session-transport.md
Normal file
@@ -0,0 +1,14 @@
|
||||
---
|
||||
"@runfusion/fusion": minor
|
||||
---
|
||||
|
||||
CLI agent session transport (U10): authenticated cli-sessions REST routes
|
||||
(list, single-use session-scoped attach tickets, inject, confirm-advance), a
|
||||
distinct `/api/cli-sessions/ws` WebSocket attach handler (daemon-token + Origin
|
||||
allowlist + single-use ticket gate, scrollback replay then live byte frames,
|
||||
ACK-credit flow control driving engine pause/resume, latest-active-client
|
||||
resize, server-side read-only enforcement, input-source attribution), a
|
||||
streaming-safe outbound output filter (`neutralizeTerminalOutput`) that strips
|
||||
OSC 52 clipboard writes, non-http(s) OSC 8 hyperlink URIs, and device-status /
|
||||
query sequences, and a throttled `cli:session:state` SSE event with
|
||||
Last-Event-ID replay.
|
||||
@@ -0,0 +1,134 @@
|
||||
// @vitest-environment node
|
||||
|
||||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
neutralizeTerminalOutput,
|
||||
flushTerminalOutput,
|
||||
MAX_CARRY_LENGTH,
|
||||
} from "../cli-session-output-filter.js";
|
||||
|
||||
const ESC = "\x1b";
|
||||
const BEL = "\x07";
|
||||
const ST = "\x1b\\";
|
||||
|
||||
/** Run a single full chunk and return the output (asserting no carry leftover). */
|
||||
function run(chunk: string): string {
|
||||
const { output, carry } = neutralizeTerminalOutput(chunk, "");
|
||||
return output + flushTerminalOutput(carry);
|
||||
}
|
||||
|
||||
describe("neutralizeTerminalOutput", () => {
|
||||
it("passes plain text through untouched", () => {
|
||||
expect(run("hello world\n")).toBe("hello world\n");
|
||||
});
|
||||
|
||||
it("strips OSC 52 clipboard-write sequences (BEL-terminated)", () => {
|
||||
const evil = `before${ESC}]52;c;ZXZpbA==${BEL}after`;
|
||||
const out = run(evil);
|
||||
expect(out).toBe("beforeafter");
|
||||
expect(out).not.toContain("52;");
|
||||
});
|
||||
|
||||
it("strips OSC 52 clipboard-write sequences (ST-terminated)", () => {
|
||||
const evil = `x${ESC}]52;c;ZGF0YQ==${ST}y`;
|
||||
expect(run(evil)).toBe("xy");
|
||||
});
|
||||
|
||||
it("strips the URI of an OSC 8 javascript: hyperlink but keeps the text", () => {
|
||||
const link = `${ESC}]8;;javascript:alert(1)${BEL}Click me${ESC}]8;;${BEL}`;
|
||||
const out = run(link);
|
||||
expect(out).not.toContain("javascript:");
|
||||
expect(out).toContain("Click me");
|
||||
// The opening link should be present but with an empty URI.
|
||||
expect(out).toContain(`${ESC}]8;;`);
|
||||
});
|
||||
|
||||
it("passes through OSC 8 https hyperlinks verbatim", () => {
|
||||
const link = `${ESC}]8;;https://example.com${BEL}Link${ESC}]8;;${BEL}`;
|
||||
const out = run(link);
|
||||
expect(out).toContain("https://example.com");
|
||||
expect(out).toContain("Link");
|
||||
});
|
||||
|
||||
it("strips DSR (device status report) query sequences", () => {
|
||||
// ESC [ 6 n is a cursor-position report query; the terminal would answer it.
|
||||
const out = run(`a${ESC}[6nb`);
|
||||
expect(out).toBe("ab");
|
||||
});
|
||||
|
||||
it("strips DA (device attributes) query sequences", () => {
|
||||
expect(run(`a${ESC}[cb`)).toBe("ab");
|
||||
expect(run(`a${ESC}[>cb`)).toBe("ab");
|
||||
});
|
||||
|
||||
it("strips DECRQSS (DCS query) sequences", () => {
|
||||
const out = run(`a${ESC}P$qm${ST}b`);
|
||||
expect(out).toBe("ab");
|
||||
});
|
||||
|
||||
it("preserves benign CSI sequences like SGR color", () => {
|
||||
const colored = `${ESC}[31mred${ESC}[0m`;
|
||||
expect(run(colored)).toBe(colored);
|
||||
});
|
||||
|
||||
it("preserves cursor movement CSI (not a query)", () => {
|
||||
const moved = `${ESC}[2J${ESC}[H`;
|
||||
expect(run(moved)).toBe(moved);
|
||||
});
|
||||
|
||||
it("handles an OSC 52 sequence split across two chunks", () => {
|
||||
const first = `before${ESC}]52;c;ZXZ`;
|
||||
const second = `pbA==${BEL}after`;
|
||||
const r1 = neutralizeTerminalOutput(first, "");
|
||||
// The unterminated OSC should be withheld in carry, not emitted.
|
||||
expect(r1.output).toBe("before");
|
||||
expect(r1.carry).toContain("52;");
|
||||
const r2 = neutralizeTerminalOutput(second, r1.carry);
|
||||
expect(r2.output).toBe("after");
|
||||
expect(r2.output + flushTerminalOutput(r2.carry)).not.toContain("52;");
|
||||
});
|
||||
|
||||
it("handles a DSR query split across two chunks", () => {
|
||||
const r1 = neutralizeTerminalOutput(`a${ESC}[6`, "");
|
||||
expect(r1.output).toBe("a");
|
||||
const r2 = neutralizeTerminalOutput(`nb`, r1.carry);
|
||||
expect(r2.output).toBe("b");
|
||||
});
|
||||
|
||||
it("withholds a lone trailing ESC as carry", () => {
|
||||
const r1 = neutralizeTerminalOutput(`hi${ESC}`, "");
|
||||
expect(r1.output).toBe("hi");
|
||||
expect(r1.carry).toBe(ESC);
|
||||
const r2 = neutralizeTerminalOutput(`[31mred`, r1.carry);
|
||||
expect(r2.output).toBe(`${ESC}[31mred`);
|
||||
});
|
||||
|
||||
it("flushes an unterminated sequence at stream end (no infinite withhold)", () => {
|
||||
const r = neutralizeTerminalOutput(`text${ESC}]52;c;partial`, "");
|
||||
expect(r.output).toBe("text");
|
||||
// flush emits the residual literally rather than losing it forever.
|
||||
expect(flushTerminalOutput(r.carry)).toContain("partial");
|
||||
});
|
||||
|
||||
it("bounds the carry so an unterminated sequence cannot grow unbounded", () => {
|
||||
const huge = `${ESC}]52;c;` + "A".repeat(MAX_CARRY_LENGTH + 100);
|
||||
const r = neutralizeTerminalOutput(huge, "");
|
||||
expect(r.carry.length).toBeLessThanOrEqual(MAX_CARRY_LENGTH);
|
||||
});
|
||||
|
||||
it("neutralizes a stream with OSC 52, OSC 8 js link, and a DSR query together", () => {
|
||||
const stream =
|
||||
`start${ESC}]52;c;ZXZpbA==${BEL}` +
|
||||
`${ESC}]8;;javascript:x${BEL}danger${ESC}]8;;${BEL}` +
|
||||
`${ESC}[6n` +
|
||||
`end`;
|
||||
const out = run(stream);
|
||||
expect(out).not.toContain("52;");
|
||||
expect(out).not.toContain("javascript:");
|
||||
expect(out).toContain("start");
|
||||
expect(out).toContain("danger");
|
||||
expect(out).toContain("end");
|
||||
// No bare query passthrough.
|
||||
expect(out).not.toContain(`${ESC}[6n`);
|
||||
});
|
||||
});
|
||||
580
packages/dashboard/src/__tests__/cli-session-ws.test.ts
Normal file
580
packages/dashboard/src/__tests__/cli-session-ws.test.ts
Normal file
@@ -0,0 +1,580 @@
|
||||
// @vitest-environment node
|
||||
|
||||
import http from "node:http";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { WebSocket } from "ws";
|
||||
import type { CliSession } from "@fusion/core";
|
||||
import type { CliSessionAttachment } from "@fusion/engine";
|
||||
import {
|
||||
setupCliSessionWebSocket,
|
||||
CLI_SESSION_WS_PATH,
|
||||
} from "../cli-session-ws.js";
|
||||
import {
|
||||
AttachTicketStore,
|
||||
CliInputAttributionLog,
|
||||
bridgeCliStateToSse,
|
||||
type CliSessionManagerLike,
|
||||
} from "../cli-session-transport.js";
|
||||
import {
|
||||
emitCliSessionStateSseEvent,
|
||||
getCliSessionStateEventsSince,
|
||||
resetCliSessionStateBufferForTests,
|
||||
} from "../sse.js";
|
||||
import { CliSessionStateMachine } from "@fusion/engine";
|
||||
|
||||
const DAEMON_TOKEN = "test-daemon-token";
|
||||
|
||||
// ── A fake PTY-backed session (no real node-pty) ─────────────────────────────
|
||||
|
||||
class FakeAttachment implements CliSessionAttachment {
|
||||
scrollback: Uint8Array;
|
||||
private queue: Uint8Array[] = [];
|
||||
private waiters: ((r: IteratorResult<Uint8Array>) => void)[] = [];
|
||||
private closed = false;
|
||||
writes: string[] = [];
|
||||
resizes: { cols: number; rows: number }[] = [];
|
||||
detached = false;
|
||||
|
||||
constructor(
|
||||
scrollback: string,
|
||||
private readonly onWrite: (data: string) => void,
|
||||
private readonly onResize: (cols: number, rows: number) => void,
|
||||
) {
|
||||
this.scrollback = Buffer.from(scrollback, "utf8");
|
||||
}
|
||||
|
||||
pushLive(text: string): void {
|
||||
const chunk = new Uint8Array(Buffer.from(text, "utf8"));
|
||||
const waiter = this.waiters.shift();
|
||||
if (waiter) waiter({ value: chunk, done: false });
|
||||
else this.queue.push(chunk);
|
||||
}
|
||||
|
||||
close(): void {
|
||||
if (this.closed) return;
|
||||
this.closed = true;
|
||||
while (this.waiters.length) this.waiters.shift()!({ value: undefined, done: true });
|
||||
}
|
||||
|
||||
get stream(): AsyncIterable<Uint8Array> {
|
||||
return {
|
||||
[Symbol.asyncIterator]: () => ({
|
||||
next: (): Promise<IteratorResult<Uint8Array>> => {
|
||||
const q = this.queue.shift();
|
||||
if (q !== undefined) return Promise.resolve({ value: q, done: false });
|
||||
if (this.closed) return Promise.resolve({ value: undefined, done: true });
|
||||
return new Promise((resolve) => this.waiters.push(resolve));
|
||||
},
|
||||
return: (): Promise<IteratorResult<Uint8Array>> => {
|
||||
this.close();
|
||||
return Promise.resolve({ value: undefined, done: true });
|
||||
},
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
write(data: string): void {
|
||||
this.onWrite(data);
|
||||
this.writes.push(data);
|
||||
}
|
||||
resize(cols: number, rows: number): void {
|
||||
this.onResize(cols, rows);
|
||||
this.resizes.push({ cols, rows });
|
||||
}
|
||||
detach(): void {
|
||||
this.detached = true;
|
||||
this.close();
|
||||
}
|
||||
}
|
||||
|
||||
class FakeManager implements CliSessionManagerLike {
|
||||
attachments = new Map<string, FakeAttachment[]>();
|
||||
live = new Set<string>();
|
||||
ptyInput: string[] = [];
|
||||
ptyResizes: { cols: number; rows: number }[] = [];
|
||||
paused = 0;
|
||||
resumed = 0;
|
||||
|
||||
constructor(liveIds: string[]) {
|
||||
for (const id of liveIds) this.live.add(id);
|
||||
}
|
||||
|
||||
isLive(id: string): boolean {
|
||||
return this.live.has(id);
|
||||
}
|
||||
|
||||
makeAttachment(id: string, scrollback: string): FakeAttachment {
|
||||
const att = new FakeAttachment(
|
||||
scrollback,
|
||||
(data) => this.ptyInput.push(data),
|
||||
(cols, rows) => this.ptyResizes.push({ cols, rows }),
|
||||
);
|
||||
const list = this.attachments.get(id) ?? [];
|
||||
list.push(att);
|
||||
this.attachments.set(id, list);
|
||||
return att;
|
||||
}
|
||||
|
||||
attach(id: string): CliSessionAttachment {
|
||||
// Each attach gets its own attachment, but they share the PTY input sink.
|
||||
return this.makeAttachment(id, this.scrollbackFor(id));
|
||||
}
|
||||
|
||||
private scrollbackFor(id: string): string {
|
||||
return this.scrollbackById.get(id) ?? "";
|
||||
}
|
||||
scrollbackById = new Map<string, string>();
|
||||
|
||||
/** Broadcast live bytes to every attachment of a session. */
|
||||
broadcast(id: string, text: string): void {
|
||||
for (const att of this.attachments.get(id) ?? []) att.pushLive(text);
|
||||
}
|
||||
|
||||
requestPause(): void {
|
||||
this.paused += 1;
|
||||
}
|
||||
requestResume(): void {
|
||||
this.resumed += 1;
|
||||
}
|
||||
}
|
||||
|
||||
function makeSession(overrides: Partial<CliSession> = {}): CliSession {
|
||||
return {
|
||||
id: "cli-1",
|
||||
taskId: "FN-1",
|
||||
chatSessionId: null,
|
||||
purpose: "execute",
|
||||
projectId: "proj-a",
|
||||
adapterId: "claude-code",
|
||||
agentState: "busy",
|
||||
terminationReason: null,
|
||||
nativeSessionId: null,
|
||||
resumeAttempts: 0,
|
||||
autonomyPosture: null,
|
||||
worktreePath: "/tmp/wt",
|
||||
createdAt: "2026-06-04T00:00:00.000Z",
|
||||
updatedAt: "2026-06-04T00:00:00.000Z",
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function makeStore(sessions: CliSession[]) {
|
||||
const map = new Map(sessions.map((s) => [s.id, s]));
|
||||
return {
|
||||
getSession: (id: string) => map.get(id),
|
||||
listSessions: () => [...map.values()],
|
||||
};
|
||||
}
|
||||
|
||||
// ── Harness: a real http.Server on an EPHEMERAL port (0 — never 4040) ────────
|
||||
|
||||
interface Harness {
|
||||
port: number;
|
||||
manager: FakeManager;
|
||||
ticketStore: AttachTicketStore;
|
||||
attributionLog: CliInputAttributionLog;
|
||||
store: ReturnType<typeof makeStore>;
|
||||
close: () => Promise<void>;
|
||||
}
|
||||
|
||||
async function startHarness(opts?: {
|
||||
sessions?: CliSession[];
|
||||
liveIds?: string[];
|
||||
highWatermarkBytes?: number;
|
||||
lowWatermarkBytes?: number;
|
||||
noAuth?: boolean;
|
||||
}): Promise<Harness> {
|
||||
const sessions = opts?.sessions ?? [makeSession()];
|
||||
const store = makeStore(sessions);
|
||||
const manager = new FakeManager(opts?.liveIds ?? sessions.map((s) => s.id));
|
||||
const ticketStore = new AttachTicketStore();
|
||||
const attributionLog = new CliInputAttributionLog();
|
||||
|
||||
const server = http.createServer((_req, res) => {
|
||||
res.statusCode = 426;
|
||||
res.end();
|
||||
});
|
||||
|
||||
setupCliSessionWebSocket(server, {
|
||||
manager,
|
||||
store,
|
||||
ticketStore,
|
||||
attributionLog,
|
||||
daemonToken: DAEMON_TOKEN,
|
||||
noAuth: opts?.noAuth ?? false,
|
||||
highWatermarkBytes: opts?.highWatermarkBytes,
|
||||
lowWatermarkBytes: opts?.lowWatermarkBytes,
|
||||
});
|
||||
|
||||
await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
|
||||
const addr = server.address();
|
||||
const port = typeof addr === "object" && addr ? addr.port : 0;
|
||||
expect(port).not.toBe(4040);
|
||||
|
||||
return {
|
||||
port,
|
||||
manager,
|
||||
ticketStore,
|
||||
attributionLog,
|
||||
store,
|
||||
close: () =>
|
||||
new Promise<void>((resolve) => {
|
||||
server.close(() => resolve());
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
function wsUrl(
|
||||
port: number,
|
||||
params: Record<string, string>,
|
||||
): string {
|
||||
const qs = new URLSearchParams(params).toString();
|
||||
return `ws://127.0.0.1:${port}${CLI_SESSION_WS_PATH}?${qs}`;
|
||||
}
|
||||
|
||||
/** Connect a WS, supplying an Origin (same-host by default) and token via query. */
|
||||
function connect(
|
||||
port: number,
|
||||
params: Record<string, string>,
|
||||
headers: Record<string, string> = {},
|
||||
): WebSocket {
|
||||
return new WebSocket(wsUrl(port, { fn_token: DAEMON_TOKEN, ...params }), {
|
||||
headers: { origin: `http://127.0.0.1:${port}`, ...headers },
|
||||
});
|
||||
}
|
||||
|
||||
function nextMessage(ws: WebSocket, predicate?: (m: any) => boolean): Promise<any> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const onMsg = (raw: Buffer) => {
|
||||
const msg = JSON.parse(raw.toString());
|
||||
if (!predicate || predicate(msg)) {
|
||||
ws.off("message", onMsg);
|
||||
resolve(msg);
|
||||
}
|
||||
};
|
||||
ws.on("message", onMsg);
|
||||
ws.once("error", reject);
|
||||
});
|
||||
}
|
||||
|
||||
function waitClose(ws: WebSocket): Promise<{ code: number; reason: string }> {
|
||||
return new Promise((resolve) => {
|
||||
ws.once("close", (code, reason) => resolve({ code, reason: reason.toString() }));
|
||||
});
|
||||
}
|
||||
|
||||
function decode(b64: string): string {
|
||||
return Buffer.from(b64, "base64").toString("utf8");
|
||||
}
|
||||
|
||||
describe("cli-session WS attach", () => {
|
||||
let h: Harness;
|
||||
|
||||
afterEach(async () => {
|
||||
if (h) await h.close();
|
||||
});
|
||||
|
||||
async function mintTicket(sessionId: string): Promise<string> {
|
||||
const { ticket } = h.ticketStore.mint({
|
||||
sessionId,
|
||||
projectId: h.store.getSession(sessionId)!.projectId,
|
||||
readOnly: false,
|
||||
});
|
||||
return ticket;
|
||||
}
|
||||
|
||||
it("two concurrent attaches both receive live bytes; input from either reaches the PTY; detach of one keeps the session", async () => {
|
||||
h = await startHarness();
|
||||
const t1 = await mintTicket("cli-1");
|
||||
const t2 = await mintTicket("cli-1");
|
||||
|
||||
const a = connect(h.port, { sessionId: "cli-1", ticket: t1 });
|
||||
const b = connect(h.port, { sessionId: "cli-1", ticket: t2 });
|
||||
await Promise.all([
|
||||
nextMessage(a, (m) => m.type === "scrollback"),
|
||||
nextMessage(b, (m) => m.type === "scrollback"),
|
||||
]);
|
||||
|
||||
const aData = nextMessage(a, (m) => m.type === "data");
|
||||
const bData = nextMessage(b, (m) => m.type === "data");
|
||||
h.manager.broadcast("cli-1", "live-output");
|
||||
expect(decode((await aData).data)).toBe("live-output");
|
||||
expect(decode((await bData).data)).toBe("live-output");
|
||||
|
||||
a.send(JSON.stringify({ type: "input", data: "from-a" }));
|
||||
b.send(JSON.stringify({ type: "input", data: "from-b" }));
|
||||
await vi.waitFor(() => {
|
||||
expect(h.manager.ptyInput).toContain("from-a");
|
||||
expect(h.manager.ptyInput).toContain("from-b");
|
||||
});
|
||||
|
||||
// Detach a — session must stay live and b keeps receiving.
|
||||
a.close();
|
||||
await waitClose(a);
|
||||
expect(h.manager.isLive("cli-1")).toBe(true);
|
||||
const bData2 = nextMessage(b, (m) => m.type === "data");
|
||||
h.manager.broadcast("cli-1", "still-here");
|
||||
expect(decode((await bData2).data)).toBe("still-here");
|
||||
b.close();
|
||||
});
|
||||
|
||||
it("rejects upgrade without a daemon token", async () => {
|
||||
h = await startHarness();
|
||||
const t = await mintTicket("cli-1");
|
||||
const ws = new WebSocket(wsUrl(h.port, { sessionId: "cli-1", ticket: t }), {
|
||||
headers: { origin: `http://127.0.0.1:${h.port}` },
|
||||
});
|
||||
const err = await new Promise<Error>((resolve) => ws.once("error", resolve));
|
||||
expect(err.message).toMatch(/401/);
|
||||
});
|
||||
|
||||
it("rejects a foreign Origin", async () => {
|
||||
h = await startHarness();
|
||||
const t = await mintTicket("cli-1");
|
||||
const ws = connect(
|
||||
h.port,
|
||||
{ sessionId: "cli-1", ticket: t },
|
||||
{ origin: "http://evil.example.com" },
|
||||
);
|
||||
const err = await new Promise<Error>((resolve) => ws.once("error", resolve));
|
||||
expect(err.message).toMatch(/403/);
|
||||
});
|
||||
|
||||
it("rejects an absent browser Origin (Sec-Fetch-Site present)", async () => {
|
||||
h = await startHarness();
|
||||
const t = await mintTicket("cli-1");
|
||||
const ws = new WebSocket(wsUrl(h.port, { fn_token: DAEMON_TOKEN, sessionId: "cli-1", ticket: t }), {
|
||||
headers: { "sec-fetch-site": "cross-site" },
|
||||
});
|
||||
const err = await new Promise<Error>((resolve) => ws.once("error", resolve));
|
||||
expect(err.message).toMatch(/403/);
|
||||
});
|
||||
|
||||
it("rejects a replayed / consumed ticket", async () => {
|
||||
h = await startHarness();
|
||||
const t = await mintTicket("cli-1");
|
||||
const first = connect(h.port, { sessionId: "cli-1", ticket: t });
|
||||
await nextMessage(first, (m) => m.type === "scrollback");
|
||||
// Reuse the same ticket on a new connection — must be rejected.
|
||||
const second = connect(h.port, { sessionId: "cli-1", ticket: t });
|
||||
const closed = await waitClose(second);
|
||||
expect(closed.code).toBe(4401);
|
||||
first.close();
|
||||
});
|
||||
|
||||
it("rejects a ticket for session A used to attach session B", async () => {
|
||||
h = await startHarness({
|
||||
sessions: [makeSession(), makeSession({ id: "cli-2", taskId: "FN-2" })],
|
||||
});
|
||||
const tA = await mintTicket("cli-1");
|
||||
const ws = connect(h.port, { sessionId: "cli-2", ticket: tA });
|
||||
const closed = await waitClose(ws);
|
||||
expect(closed.code).toBe(4401);
|
||||
});
|
||||
|
||||
it("rejects a cross-project session id (ticket project mismatch handled; unknown session)", async () => {
|
||||
h = await startHarness();
|
||||
const t = await mintTicket("cli-1");
|
||||
const ws = connect(h.port, { sessionId: "cli-unknown", ticket: t });
|
||||
const closed = await waitClose(ws);
|
||||
expect(closed.code).toBe(4004);
|
||||
});
|
||||
|
||||
it("late attacher gets scrollback then live with no duplication", async () => {
|
||||
h = await startHarness();
|
||||
h.manager.scrollbackById.set("cli-1", "PRIOR-OUTPUT");
|
||||
const t = await mintTicket("cli-1");
|
||||
const ws = connect(h.port, { sessionId: "cli-1", ticket: t });
|
||||
const sb = await nextMessage(ws, (m) => m.type === "scrollback");
|
||||
expect(decode(sb.data)).toBe("PRIOR-OUTPUT");
|
||||
const live = nextMessage(ws, (m) => m.type === "data");
|
||||
h.manager.broadcast("cli-1", "NEW");
|
||||
expect(decode((await live).data)).toBe("NEW");
|
||||
ws.close();
|
||||
});
|
||||
|
||||
it("flow control: slow consumer (no acks) triggers pause at high watermark; acks resume at low watermark", async () => {
|
||||
h = await startHarness({ highWatermarkBytes: 20, lowWatermarkBytes: 10 });
|
||||
const t = await mintTicket("cli-1");
|
||||
const ws = connect(h.port, { sessionId: "cli-1", ticket: t });
|
||||
await nextMessage(ws, (m) => m.type === "scrollback");
|
||||
|
||||
// Push 25 bytes without acking — crosses the 20-byte high watermark.
|
||||
const d1 = nextMessage(ws, (m) => m.type === "data");
|
||||
h.manager.broadcast("cli-1", "x".repeat(25));
|
||||
await d1;
|
||||
await vi.waitFor(() => expect(h.manager.paused).toBe(1));
|
||||
expect(h.manager.resumed).toBe(0);
|
||||
|
||||
// Ack 20 bytes — outstanding drops to 5 (<= low watermark 10) → resume.
|
||||
ws.send(JSON.stringify({ type: "ack", bytes: 20 }));
|
||||
await vi.waitFor(() => expect(h.manager.resumed).toBe(1));
|
||||
ws.close();
|
||||
});
|
||||
|
||||
it("resize is forwarded (latest-active-client) to the manager", async () => {
|
||||
h = await startHarness();
|
||||
const t = await mintTicket("cli-1");
|
||||
const ws = connect(h.port, { sessionId: "cli-1", ticket: t });
|
||||
await nextMessage(ws, (m) => m.type === "scrollback");
|
||||
ws.send(JSON.stringify({ type: "resize", cols: 120, rows: 40 }));
|
||||
await vi.waitFor(() =>
|
||||
expect(h.manager.ptyResizes).toContainEqual({ cols: 120, rows: 40 }),
|
||||
);
|
||||
ws.close();
|
||||
});
|
||||
|
||||
it("output hardening: OSC 52, OSC 8 javascript: link, and a DSR query are neutralized (incl. split across chunks)", async () => {
|
||||
h = await startHarness();
|
||||
const t = await mintTicket("cli-1");
|
||||
const ws = connect(h.port, { sessionId: "cli-1", ticket: t });
|
||||
await nextMessage(ws, (m) => m.type === "scrollback");
|
||||
|
||||
const ESC = "\x1b";
|
||||
const BEL = "\x07";
|
||||
// OSC 52 + OSC 8 js link + DSR — all in one chunk.
|
||||
const d1 = nextMessage(ws, (m) => m.type === "data");
|
||||
h.manager.broadcast(
|
||||
"cli-1",
|
||||
`A${ESC}]52;c;ZXZpbA==${BEL}${ESC}]8;;javascript:x${BEL}T${ESC}]8;;${BEL}${ESC}[6nB`,
|
||||
);
|
||||
const out1 = decode((await d1).data);
|
||||
expect(out1).not.toContain("52;");
|
||||
expect(out1).not.toContain("javascript:");
|
||||
expect(out1).not.toContain(`${ESC}[6n`);
|
||||
expect(out1).toContain("A");
|
||||
expect(out1).toContain("T");
|
||||
expect(out1).toContain("B");
|
||||
|
||||
// A sequence split across two broadcasts (chunks).
|
||||
h.manager.broadcast("cli-1", `C${ESC}]52;c;ZXZ`);
|
||||
h.manager.broadcast("cli-1", `pbA==${BEL}D`);
|
||||
// Collect data frames until we see the trailing "D".
|
||||
let collected = "";
|
||||
await vi.waitFor(
|
||||
() =>
|
||||
new Promise<void>((resolve, reject) => {
|
||||
const onMsg = (raw: Buffer) => {
|
||||
const m = JSON.parse(raw.toString());
|
||||
if (m.type === "data") collected += decode(m.data);
|
||||
if (collected.includes("D")) {
|
||||
ws.off("message", onMsg);
|
||||
resolve();
|
||||
}
|
||||
};
|
||||
ws.on("message", onMsg);
|
||||
setTimeout(() => reject(new Error("timeout")), 1000);
|
||||
}),
|
||||
);
|
||||
expect(collected).not.toContain("52;");
|
||||
expect(collected).toContain("C");
|
||||
expect(collected).toContain("D");
|
||||
ws.close();
|
||||
});
|
||||
|
||||
it("read-only session rejects input with an error frame (server-side)", async () => {
|
||||
h = await startHarness({
|
||||
sessions: [makeSession({ id: "cli-ro", purpose: "validator" })],
|
||||
});
|
||||
const t = await mintTicket("cli-ro");
|
||||
const ws = connect(h.port, { sessionId: "cli-ro", ticket: t });
|
||||
await nextMessage(ws, (m) => m.type === "scrollback");
|
||||
const errFrame = nextMessage(ws, (m) => m.type === "error");
|
||||
ws.send(JSON.stringify({ type: "input", data: "should-be-blocked" }));
|
||||
const err = await errFrame;
|
||||
expect(err.code).toBe("READ_ONLY");
|
||||
expect(h.manager.ptyInput).not.toContain("should-be-blocked");
|
||||
ws.close();
|
||||
});
|
||||
});
|
||||
|
||||
// ── SSE cli:session:state ─────────────────────────────────────────────────────
|
||||
|
||||
/** Minimal in-memory store satisfying what the state machine + bridge touch. */
|
||||
function makeStateStore(session: CliSession) {
|
||||
let current = { ...session };
|
||||
return {
|
||||
getSession: (id: string) => (id === current.id ? current : undefined),
|
||||
listSessions: () => [current],
|
||||
updateSession: (id: string, input: Partial<CliSession>) => {
|
||||
if (id !== current.id) return undefined;
|
||||
current = { ...current, ...input } as CliSession;
|
||||
return current;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
describe("cli:session:state SSE bridge", () => {
|
||||
beforeEach(() => resetCliSessionStateBufferForTests());
|
||||
|
||||
it("forwards a state transition as one event carrying owning entity + redacted preview", () => {
|
||||
const session = makeSession({
|
||||
id: "s-1",
|
||||
taskId: "FN-9",
|
||||
projectId: "proj-a",
|
||||
agentState: "starting",
|
||||
});
|
||||
const store = makeStateStore(session);
|
||||
const machine = new CliSessionStateMachine({
|
||||
sessionId: "s-1",
|
||||
store: store as never,
|
||||
});
|
||||
const captured: { id: number; payload: any; projectId?: string }[] = [];
|
||||
const unbridge = bridgeCliStateToSse(machine, {
|
||||
store: store as never,
|
||||
getRecentOutput: () => "bearer sk-secret-1234567890 \x1b[31mfoo\x1b[0m",
|
||||
});
|
||||
// Seed machine into busy via the legal path, then trigger a transition.
|
||||
machine.markReady(); // starting → ready
|
||||
machine.injectPrompt(); // ready → busy
|
||||
const before = getCliSessionStateEventsSince(0).length;
|
||||
machine.signalDone(); // busy → done
|
||||
const events = getCliSessionStateEventsSince(0);
|
||||
expect(events.length).toBeGreaterThan(before);
|
||||
const last = events[events.length - 1];
|
||||
expect(last.payload.sessionId).toBe("s-1");
|
||||
expect(last.payload.taskId).toBe("FN-9");
|
||||
expect(last.payload.state).toBe("done");
|
||||
expect(last.projectId).toBe("proj-a");
|
||||
// Preview is ANSI-stripped and bounded.
|
||||
expect(last.payload.lastOutputPreview).not.toContain("\x1b");
|
||||
unbridge();
|
||||
void captured;
|
||||
});
|
||||
|
||||
it("engine throttle coalesces rapid transitions into fewer emitted events", async () => {
|
||||
const session = makeSession({ id: "s-2", projectId: "proj-a", agentState: "starting" });
|
||||
const store = makeStateStore(session);
|
||||
const machine = new CliSessionStateMachine({
|
||||
sessionId: "s-2",
|
||||
store: store as never,
|
||||
stateChangeThrottleMs: 500,
|
||||
});
|
||||
bridgeCliStateToSse(machine, { store: store as never });
|
||||
machine.markReady();
|
||||
machine.injectPrompt();
|
||||
// Rapid waiting↔busy churn within the throttle window.
|
||||
machine.signalWaitingOnInput();
|
||||
machine.signalBusy();
|
||||
machine.signalWaitingOnInput();
|
||||
machine.signalBusy();
|
||||
const emitted = getCliSessionStateEventsSince(0).length;
|
||||
// Leading-edge + coalesced trailing means far fewer than the ~6 raw transitions.
|
||||
expect(emitted).toBeLessThan(6);
|
||||
expect(emitted).toBeGreaterThanOrEqual(1);
|
||||
});
|
||||
|
||||
it("getCliSessionStateEventsSince replays only transitions after a lastEventId", () => {
|
||||
const a = emitCliSessionStateSseEvent(
|
||||
{ sessionId: "x", taskId: null, chatSessionId: null, state: "ready", at: "t0" },
|
||||
"proj-a",
|
||||
);
|
||||
const b = emitCliSessionStateSseEvent(
|
||||
{ sessionId: "x", taskId: null, chatSessionId: null, state: "busy", at: "t1" },
|
||||
"proj-a",
|
||||
);
|
||||
const replay = getCliSessionStateEventsSince(a);
|
||||
expect(replay.map((e) => e.id)).toEqual([b]);
|
||||
expect(replay[0].payload.state).toBe("busy");
|
||||
});
|
||||
});
|
||||
211
packages/dashboard/src/__tests__/cli-sessions-routes.test.ts
Normal file
211
packages/dashboard/src/__tests__/cli-sessions-routes.test.ts
Normal file
@@ -0,0 +1,211 @@
|
||||
// @vitest-environment node
|
||||
|
||||
import express from "express";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { request } from "../test-request.js";
|
||||
import type { CliSession } from "@fusion/core";
|
||||
|
||||
type App = (req: import("http").IncomingMessage, res: import("http").ServerResponse) => void;
|
||||
|
||||
function getJson(app: App, path: string) {
|
||||
return request(app, "GET", path);
|
||||
}
|
||||
function postJson(app: App, path: string, body: unknown) {
|
||||
return request(app, "POST", path, JSON.stringify(body), {
|
||||
"content-type": "application/json",
|
||||
});
|
||||
}
|
||||
import { createCliSessionsRouter } from "../routes/cli-sessions.js";
|
||||
import {
|
||||
AttachTicketStore,
|
||||
CliInputAttributionLog,
|
||||
CliConfirmAdvanceRegistry,
|
||||
type CliSessionManagerLike,
|
||||
} from "../cli-session-transport.js";
|
||||
|
||||
function makeSession(overrides: Partial<CliSession> = {}): CliSession {
|
||||
return {
|
||||
id: "cli-1",
|
||||
taskId: "FN-1",
|
||||
chatSessionId: null,
|
||||
purpose: "execute",
|
||||
projectId: "proj-a",
|
||||
adapterId: "claude-code",
|
||||
agentState: "busy",
|
||||
terminationReason: null,
|
||||
nativeSessionId: null,
|
||||
resumeAttempts: 0,
|
||||
autonomyPosture: null,
|
||||
worktreePath: "/tmp/wt",
|
||||
createdAt: "2026-06-04T00:00:00.000Z",
|
||||
updatedAt: "2026-06-04T00:00:00.000Z",
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function makeStore(sessions: CliSession[]) {
|
||||
const map = new Map(sessions.map((s) => [s.id, s]));
|
||||
return {
|
||||
getSession: (id: string) => map.get(id),
|
||||
listSessions: (opts?: {
|
||||
projectId?: string;
|
||||
taskId?: string;
|
||||
chatSessionId?: string;
|
||||
}) =>
|
||||
[...map.values()].filter(
|
||||
(s) =>
|
||||
(opts?.projectId === undefined || s.projectId === opts.projectId) &&
|
||||
(opts?.taskId === undefined || s.taskId === opts.taskId) &&
|
||||
(opts?.chatSessionId === undefined || s.chatSessionId === opts.chatSessionId),
|
||||
),
|
||||
_map: map,
|
||||
};
|
||||
}
|
||||
|
||||
function buildApp(opts: {
|
||||
store: ReturnType<typeof makeStore>;
|
||||
manager: CliSessionManagerLike;
|
||||
ticketStore: AttachTicketStore;
|
||||
attributionLog: CliInputAttributionLog;
|
||||
confirmAdvance: CliConfirmAdvanceRegistry;
|
||||
}): express.Express {
|
||||
const app = express();
|
||||
app.use(express.json());
|
||||
app.use(
|
||||
"/api/cli-sessions",
|
||||
createCliSessionsRouter({
|
||||
store: opts.store,
|
||||
manager: opts.manager,
|
||||
ticketStore: opts.ticketStore,
|
||||
attributionLog: opts.attributionLog,
|
||||
confirmAdvance: opts.confirmAdvance,
|
||||
}),
|
||||
);
|
||||
return app;
|
||||
}
|
||||
|
||||
describe("cli-sessions routes", () => {
|
||||
let store: ReturnType<typeof makeStore>;
|
||||
let manager: CliSessionManagerLike;
|
||||
let injectSpy: ReturnType<typeof vi.fn>;
|
||||
let ticketStore: AttachTicketStore;
|
||||
let attributionLog: CliInputAttributionLog;
|
||||
let confirmAdvance: CliConfirmAdvanceRegistry;
|
||||
let app: express.Express;
|
||||
|
||||
beforeEach(() => {
|
||||
store = makeStore([
|
||||
makeSession(),
|
||||
makeSession({ id: "cli-2", projectId: "proj-b", taskId: "FN-2" }),
|
||||
makeSession({ id: "cli-ro", purpose: "validator", projectId: "proj-a", taskId: "FN-3" }),
|
||||
]);
|
||||
injectSpy = vi.fn().mockResolvedValue(undefined);
|
||||
manager = {
|
||||
isLive: () => true,
|
||||
attach: () => {
|
||||
throw new Error("not used in route tests");
|
||||
},
|
||||
inject: injectSpy,
|
||||
requestPause: vi.fn(),
|
||||
requestResume: vi.fn(),
|
||||
};
|
||||
ticketStore = new AttachTicketStore();
|
||||
attributionLog = new CliInputAttributionLog();
|
||||
confirmAdvance = new CliConfirmAdvanceRegistry();
|
||||
app = buildApp({ store, manager, ticketStore, attributionLog, confirmAdvance });
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
it("lists sessions filtered by project / task / chat", async () => {
|
||||
const res = await getJson(app, "/api/cli-sessions?projectId=proj-a");
|
||||
expect(res.status).toBe(200);
|
||||
const ids = res.body.sessions.map((s: CliSession) => s.id).sort();
|
||||
expect(ids).toEqual(["cli-1", "cli-ro"]);
|
||||
|
||||
const byTask = await getJson(app, "/api/cli-sessions?taskId=FN-2");
|
||||
expect(byTask.body.sessions.map((s: CliSession) => s.id)).toEqual(["cli-2"]);
|
||||
});
|
||||
|
||||
it("returns a single session record", async () => {
|
||||
const res = await getJson(app, "/api/cli-sessions/cli-1");
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.body.session.id).toBe("cli-1");
|
||||
});
|
||||
|
||||
it("404s for unknown session", async () => {
|
||||
const res = await getJson(app, "/api/cli-sessions/nope");
|
||||
expect(res.status).toBe(404);
|
||||
});
|
||||
|
||||
it("rejects cross-project access (project scope check)", async () => {
|
||||
const res = await getJson(app, "/api/cli-sessions/cli-1?projectId=proj-b");
|
||||
expect(res.status).toBe(403);
|
||||
});
|
||||
|
||||
it("mints a single-use attach ticket with expiry", async () => {
|
||||
const res = await postJson(app, "/api/cli-sessions/cli-1/attach-ticket", {});
|
||||
expect(res.status).toBe(200);
|
||||
expect(typeof res.body.ticket).toBe("string");
|
||||
expect(res.body.ticket.length).toBeGreaterThan(20);
|
||||
expect(new Date(res.body.expiresAt).getTime()).toBeGreaterThan(Date.now());
|
||||
expect(res.body.readOnly).toBe(false);
|
||||
|
||||
// The ticket consumes exactly once and is bound to its session.
|
||||
const entry = ticketStore.consume(res.body.ticket, "cli-1");
|
||||
expect(entry).not.toBeNull();
|
||||
expect(ticketStore.consume(res.body.ticket, "cli-1")).toBeNull(); // single-use
|
||||
});
|
||||
|
||||
it("marks a read-only (validator) session's ticket as readOnly", async () => {
|
||||
const res = await postJson(app, "/api/cli-sessions/cli-ro/attach-ticket", {});
|
||||
expect(res.body.readOnly).toBe(true);
|
||||
});
|
||||
|
||||
it("injects text and records attribution", async () => {
|
||||
const res = await postJson(app, "/api/cli-sessions/cli-1/inject", { text: "hello agent" });
|
||||
expect(res.status).toBe(200);
|
||||
expect(injectSpy).toHaveBeenCalledWith("cli-1", "hello agent");
|
||||
const log = attributionLog.list("cli-1");
|
||||
expect(log).toHaveLength(1);
|
||||
expect(log[0].source).toBe("inject");
|
||||
});
|
||||
|
||||
it("rejects inject on a read-only session (server-side enforcement)", async () => {
|
||||
const res = await postJson(app, "/api/cli-sessions/cli-ro/inject", { text: "nope" });
|
||||
expect(res.status).toBe(403);
|
||||
expect(injectSpy).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("rejects empty inject body", async () => {
|
||||
const res = await postJson(app, "/api/cli-sessions/cli-1/inject", { text: "" });
|
||||
expect(res.status).toBe(400);
|
||||
});
|
||||
|
||||
it("409s inject when session is not live", async () => {
|
||||
manager.isLive = () => false;
|
||||
const res = await postJson(app, "/api/cli-sessions/cli-1/inject", { text: "hi" });
|
||||
expect(res.status).toBe(409);
|
||||
});
|
||||
|
||||
it("records a confirm-advance decision and emits an event", async () => {
|
||||
const seen: string[] = [];
|
||||
confirmAdvance.on((info) => seen.push(`${info.sessionId}:${info.decision}`));
|
||||
const res = await postJson(app, "/api/cli-sessions/cli-1/confirm-advance", {
|
||||
decision: "advance",
|
||||
});
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.body.decision).toBe("advance");
|
||||
expect(confirmAdvance.getLatest("cli-1")).toBe("advance");
|
||||
expect(seen).toContain("cli-1:advance");
|
||||
});
|
||||
|
||||
it("rejects an invalid confirm-advance decision", async () => {
|
||||
const res = await postJson(app, "/api/cli-sessions/cli-1/confirm-advance", {
|
||||
decision: "maybe",
|
||||
});
|
||||
expect(res.status).toBe(400);
|
||||
});
|
||||
});
|
||||
244
packages/dashboard/src/cli-session-output-filter.ts
Normal file
244
packages/dashboard/src/cli-session-output-filter.ts
Normal file
@@ -0,0 +1,244 @@
|
||||
/**
|
||||
* cli-session-output-filter — outbound terminal-output hardening
|
||||
* (CLI Agent Executor, U10).
|
||||
*
|
||||
* The CLI agent's terminal byte stream is UNTRUSTED: the agent (or anything it
|
||||
* runs) controls the bytes, and a co-driving surface (browser xterm, TUI host
|
||||
* TTY) honors escape sequences. Some sequences let an attacker exfiltrate or
|
||||
* forge input, so the server-side WS bridge neutralizes them BEFORE forwarding
|
||||
* a data frame. `neutralizeTerminalOutput` is the single, pure, streaming-safe
|
||||
* implementation; the TUI (U14) re-imports it so every passthrough applies the
|
||||
* identical set.
|
||||
*
|
||||
* The neutralized set (KTD — output hardening):
|
||||
* - OSC 52 (clipboard write): `ESC ] 52 ; … BEL|ST` — a remote write to the
|
||||
* user's clipboard. Stripped entirely.
|
||||
* - OSC 8 hyperlinks with a non-http/https scheme (e.g. `javascript:`,
|
||||
* `file:`): `ESC ] 8 ; params ; URI BEL|ST`. The URI is stripped (replaced
|
||||
* with an empty URI so the hyperlink is closed/neutral) while the visible
|
||||
* link TEXT that follows is kept. http/https links pass through untouched.
|
||||
* - Device-status / query sequences whose auto-responses would forge INPUT back
|
||||
* into the PTY: DSR (`ESC [ … n`), DA1/DA2 (`ESC [ c`, `ESC [ > c`), and
|
||||
* DECRQSS (`ESC P $ q … ESC \`). A terminal answers these by writing bytes to
|
||||
* stdin — which converge on the agent's input FIFO — so a malicious stream
|
||||
* could smuggle keystrokes. Stripped.
|
||||
*
|
||||
* Streaming-safe: a sequence may be split across two chunks. The function takes
|
||||
* (and returns) a small carry buffer holding a trailing partial sequence; the
|
||||
* caller threads the carry across calls. `flush` emits any residual carry on
|
||||
* stream end. The carry is bounded so a never-terminated sequence can't grow
|
||||
* without limit.
|
||||
*/
|
||||
|
||||
const ESC = "\x1b";
|
||||
const BEL = "\x07";
|
||||
const ST = "\x1b\\"; // String Terminator (ESC \)
|
||||
|
||||
/**
|
||||
* Maximum bytes held in the carry buffer for an in-progress sequence. A
|
||||
* sequence longer than this is almost certainly malformed/hostile (a real OSC /
|
||||
* CSI is short); once exceeded we flush the carry as literal text rather than
|
||||
* buffer unboundedly.
|
||||
*/
|
||||
export const MAX_CARRY_LENGTH = 8 * 1024;
|
||||
|
||||
/** Allowed URI schemes for OSC 8 hyperlinks. Everything else has its URI stripped. */
|
||||
const SAFE_LINK_SCHEME = /^(https?):/i;
|
||||
|
||||
export interface NeutralizeResult {
|
||||
/** The sanitized, safe-to-forward text. */
|
||||
output: string;
|
||||
/**
|
||||
* Trailing bytes withheld because they may be the prefix of a sequence that
|
||||
* continues in the next chunk. Pass this back in as the next call's `carry`.
|
||||
*/
|
||||
carry: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Does `text` (already known to start at an ESC) look like it COULD be the start
|
||||
* of a longer sequence we care about, if more bytes arrive? Used to decide
|
||||
* whether to withhold a trailing partial as carry. Returns true when the buffer
|
||||
* is a strict, still-growing prefix of a recognized-but-unterminated sequence.
|
||||
*/
|
||||
function isIncompleteSequence(buf: string): boolean {
|
||||
if (buf === ESC) return true; // lone ESC — could begin anything
|
||||
// OSC: ESC ] … (terminated by BEL or ST). Incomplete until terminator seen.
|
||||
if (buf.startsWith(`${ESC}]`)) {
|
||||
return !buf.includes(BEL) && !buf.includes(ST);
|
||||
}
|
||||
// DCS (DECRQSS uses DCS): ESC P … ESC \ (terminated by ST).
|
||||
if (buf.startsWith(`${ESC}P`)) {
|
||||
return !buf.includes(ST);
|
||||
}
|
||||
// CSI: ESC [ params/intermediates, terminated by a final byte @-~.
|
||||
if (buf.startsWith(`${ESC}[`)) {
|
||||
// Final byte is in 0x40–0x7e. Incomplete while we've not seen one yet.
|
||||
return !/[@-~]/.test(buf.slice(2));
|
||||
}
|
||||
// ESC followed by exactly nothing-meaningful-yet: ESC + a single byte that
|
||||
// could still become "]" / "[" / "P". `${ESC}` alone handled above; a 2-char
|
||||
// ESC + intermediate is its own complete 2-byte sequence, not incomplete.
|
||||
if (buf.length === 1) return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Process a fully-terminated OSC body (the text between `ESC ]` and its
|
||||
* terminator, terminator excluded). Returns the replacement text to emit in
|
||||
* place of the WHOLE OSC sequence (including terminator handling done by the
|
||||
* caller).
|
||||
*/
|
||||
function neutralizeOsc(body: string): string {
|
||||
// OSC 52 (clipboard): drop entirely.
|
||||
if (body.startsWith("52;") || body === "52") {
|
||||
return "";
|
||||
}
|
||||
// OSC 8 hyperlink: ESC ] 8 ; params ; URI
|
||||
if (body.startsWith("8;")) {
|
||||
const rest = body.slice(2); // params ; URI
|
||||
const sep = rest.indexOf(";");
|
||||
if (sep === -1) {
|
||||
// Malformed — strip the whole thing to be safe.
|
||||
return "";
|
||||
}
|
||||
const params = rest.slice(0, sep);
|
||||
const uri = rest.slice(sep + 1);
|
||||
// The closing OSC 8 (empty URI) is always safe — keep it so link state is
|
||||
// balanced.
|
||||
if (uri === "") {
|
||||
return `${ESC}]8;${params};${ST}`;
|
||||
}
|
||||
if (SAFE_LINK_SCHEME.test(uri)) {
|
||||
// Safe scheme — pass the hyperlink through verbatim.
|
||||
return `${ESC}]8;${params};${uri}${ST}`;
|
||||
}
|
||||
// Unsafe scheme — strip the URI (emit an opening link with empty URI so the
|
||||
// following visible text still renders, just not as a live link).
|
||||
return `${ESC}]8;${params};${ST}`;
|
||||
}
|
||||
// Any other OSC — pass through verbatim (re-add ST terminator).
|
||||
return `${ESC}]${body}${ST}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Is this a CSI device-status / query sequence whose auto-response forges input?
|
||||
* `seq` is the full CSI starting at `ESC [` and ending at its final byte.
|
||||
*/
|
||||
function isForgingQueryCsi(seq: string): boolean {
|
||||
const body = seq.slice(2); // params + intermediates + final
|
||||
const final = body.slice(-1);
|
||||
const params = body.slice(0, -1);
|
||||
// DSR — Device Status Report: ESC [ … n (e.g. 5n, 6n cursor-position report).
|
||||
if (final === "n") return true;
|
||||
// DA1 — Primary Device Attributes: ESC [ c or ESC [ 0 c
|
||||
if (final === "c" && !params.startsWith(">") && !params.startsWith("=")) return true;
|
||||
// DA2/DA3 — Secondary/Tertiary DA: ESC [ > c , ESC [ = c
|
||||
if (final === "c" && (params.startsWith(">") || params.startsWith("="))) return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Neutralize one chunk of (possibly mid-sequence) terminal output, threading a
|
||||
* carry buffer so sequences split across chunks are handled. Pure: no I/O, no
|
||||
* shared state.
|
||||
*/
|
||||
export function neutralizeTerminalOutput(chunk: string, carry = ""): NeutralizeResult {
|
||||
let input = carry + chunk;
|
||||
let out = "";
|
||||
let i = 0;
|
||||
|
||||
while (i < input.length) {
|
||||
const ch = input[i];
|
||||
if (ch !== ESC) {
|
||||
out += ch;
|
||||
i += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
// We're at an ESC. Examine the remainder.
|
||||
const rest = input.slice(i);
|
||||
|
||||
// ── OSC: ESC ] … BEL|ST ──
|
||||
if (rest.startsWith(`${ESC}]`)) {
|
||||
const belIdx = rest.indexOf(BEL);
|
||||
const stIdx = rest.indexOf(ST);
|
||||
let endIdx = -1;
|
||||
let termLen = 0;
|
||||
if (belIdx !== -1 && (stIdx === -1 || belIdx < stIdx)) {
|
||||
endIdx = belIdx;
|
||||
termLen = BEL.length;
|
||||
} else if (stIdx !== -1) {
|
||||
endIdx = stIdx;
|
||||
termLen = ST.length;
|
||||
}
|
||||
if (endIdx === -1) {
|
||||
// Unterminated — withhold as carry (bounded).
|
||||
break;
|
||||
}
|
||||
const body = rest.slice(2, endIdx);
|
||||
out += neutralizeOsc(body);
|
||||
i += endIdx + termLen;
|
||||
continue;
|
||||
}
|
||||
|
||||
// ── DCS (DECRQSS): ESC P … ESC \ ──
|
||||
if (rest.startsWith(`${ESC}P`)) {
|
||||
const stIdx = rest.indexOf(ST, 2);
|
||||
if (stIdx === -1) {
|
||||
break; // unterminated — carry
|
||||
}
|
||||
// DECRQSS is a query (ESC P $ q …). Strip the whole DCS.
|
||||
i += stIdx + ST.length;
|
||||
continue;
|
||||
}
|
||||
|
||||
// ── CSI: ESC [ … final(@-~) ──
|
||||
if (rest.startsWith(`${ESC}[`)) {
|
||||
const finalMatch = rest.slice(2).search(/[@-~]/);
|
||||
if (finalMatch === -1) {
|
||||
break; // unterminated — carry
|
||||
}
|
||||
const seq = rest.slice(0, 2 + finalMatch + 1);
|
||||
if (isForgingQueryCsi(seq)) {
|
||||
// Strip the query.
|
||||
} else {
|
||||
out += seq;
|
||||
}
|
||||
i += seq.length;
|
||||
continue;
|
||||
}
|
||||
|
||||
// ── Lone ESC at end of buffer: could begin a sequence next chunk. ──
|
||||
if (rest.length === 1) {
|
||||
break; // carry the ESC
|
||||
}
|
||||
|
||||
// ── Some other 2-byte ESC sequence (ESC <byte>) — pass through. ──
|
||||
out += rest.slice(0, 2);
|
||||
i += 2;
|
||||
}
|
||||
|
||||
let newCarry = input.slice(i);
|
||||
// Bound the carry: if a "sequence" never terminates, don't buffer forever —
|
||||
// flush it as literal output (it isn't a recognized hazard if it's this long).
|
||||
if (newCarry.length > MAX_CARRY_LENGTH) {
|
||||
out += newCarry;
|
||||
newCarry = "";
|
||||
} else if (newCarry.length > 0 && !isIncompleteSequence(newCarry)) {
|
||||
// The residual isn't actually a growing prefix — emit it.
|
||||
out += newCarry;
|
||||
newCarry = "";
|
||||
}
|
||||
|
||||
return { output: out, carry: newCarry };
|
||||
}
|
||||
|
||||
/**
|
||||
* Flush a residual carry at stream end. The held bytes were an unterminated
|
||||
* sequence; emit them as literal text (we never got a terminator, so there's no
|
||||
* safe interpretation to apply — but withholding forever would lose output).
|
||||
*/
|
||||
export function flushTerminalOutput(carry: string): string {
|
||||
return carry;
|
||||
}
|
||||
343
packages/dashboard/src/cli-session-transport.ts
Normal file
343
packages/dashboard/src/cli-session-transport.ts
Normal file
@@ -0,0 +1,343 @@
|
||||
/**
|
||||
* cli-session-transport — shared transport primitives for CLI agent sessions
|
||||
* (CLI Agent Executor, U10).
|
||||
*
|
||||
* Holds the pieces the REST router (cli-sessions.ts) and the WS attach handler
|
||||
* (server.ts wiring) both depend on:
|
||||
* - the narrow manager/store/hub interfaces the transport needs (so tests can
|
||||
* supply a fake CliSessionManager without a real PTY),
|
||||
* - the short-lived, single-use, session-scoped ATTACH TICKET store,
|
||||
* - input-source ATTRIBUTION bookkeeping (ticket id → session input log),
|
||||
* - the generic-tier CONFIRM-ADVANCE flag/event seam,
|
||||
* - the ORIGIN allowlist check for the WS upgrade.
|
||||
*
|
||||
* Attach auth (KTD): the long-lived daemon token alone never authorizes PTY
|
||||
* WRITE access. A surface must first call the authenticated attach-ticket route
|
||||
* (daemon-token gated), then present the single-use ticket on the WS upgrade —
|
||||
* which ALSO re-checks the daemon token and an Origin allowlist. Keystroke
|
||||
* injection into a privileged agent PTY warrants the stronger posture.
|
||||
*/
|
||||
|
||||
import { randomBytes } from "node:crypto";
|
||||
import type { CliSession, CliSessionStore } from "@fusion/core";
|
||||
import {
|
||||
stripAnsiControl,
|
||||
type CliSessionAttachment,
|
||||
type CliStateChange,
|
||||
} from "@fusion/engine";
|
||||
import { emitCliSessionStateSseEvent } from "./sse.js";
|
||||
|
||||
// ── Narrow interfaces the transport needs ────────────────────────────────────
|
||||
|
||||
/**
|
||||
* The subset of the engine `CliSessionManager` the transport touches. The real
|
||||
* manager satisfies this; WS tests supply a fake (no PTY) implementing exactly
|
||||
* these members.
|
||||
*/
|
||||
export interface CliSessionManagerLike {
|
||||
/** Whether a session id is currently live (attachable). */
|
||||
isLive(sessionId: string): boolean;
|
||||
/** Attach a client: scrollback + live stream + write/resize/detach. */
|
||||
attach(sessionId: string): CliSessionAttachment;
|
||||
/** Inject a composed/engine prompt (neutralized) onto the shared FIFO. */
|
||||
inject(sessionId: string, text: string): Promise<void>;
|
||||
/** High-watermark backpressure: pause the PTY. */
|
||||
requestPause(sessionId: string): void;
|
||||
/** Low-watermark backpressure release: resume the PTY. */
|
||||
requestResume(sessionId: string): void;
|
||||
}
|
||||
|
||||
/** Dependencies the transport binds to (engine-owned, supplied at setup). */
|
||||
export interface CliSessionTransportDeps {
|
||||
manager: CliSessionManagerLike;
|
||||
store: Pick<CliSessionStore, "getSession" | "listSessions">;
|
||||
}
|
||||
|
||||
// ── Attach tickets ───────────────────────────────────────────────────────────
|
||||
|
||||
/** Default attach-ticket TTL (ms). Short-lived: enough to open a WS, no more. */
|
||||
export const DEFAULT_ATTACH_TICKET_TTL_MS = 60_000;
|
||||
|
||||
interface AttachTicketEntry {
|
||||
ticket: string;
|
||||
sessionId: string;
|
||||
projectId: string;
|
||||
/** Whether write (input/inject over WS) is permitted (read-only sessions: false). */
|
||||
readOnly: boolean;
|
||||
expiresAt: number;
|
||||
consumed: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* In-memory, single-use, session-scoped attach-ticket store. A ticket is minted
|
||||
* by the authenticated REST route, then consumed exactly once on the WS upgrade.
|
||||
* Expired/consumed tickets never validate, and a ticket for session A can never
|
||||
* attach session B (the session id is bound into the ticket).
|
||||
*/
|
||||
export class AttachTicketStore {
|
||||
private readonly tickets = new Map<string, AttachTicketEntry>();
|
||||
private readonly ttlMs: number;
|
||||
private readonly now: () => number;
|
||||
|
||||
constructor(opts?: { ttlMs?: number; now?: () => number }) {
|
||||
this.ttlMs = opts?.ttlMs ?? DEFAULT_ATTACH_TICKET_TTL_MS;
|
||||
this.now = opts?.now ?? (() => Date.now());
|
||||
}
|
||||
|
||||
/** Mint a ticket for a session. Returns the opaque ticket + its expiry. */
|
||||
mint(input: {
|
||||
sessionId: string;
|
||||
projectId: string;
|
||||
readOnly: boolean;
|
||||
}): { ticket: string; expiresAt: number } {
|
||||
const ticket = randomBytes(24).toString("hex");
|
||||
const expiresAt = this.now() + this.ttlMs;
|
||||
this.tickets.set(ticket, {
|
||||
ticket,
|
||||
sessionId: input.sessionId,
|
||||
projectId: input.projectId,
|
||||
readOnly: input.readOnly,
|
||||
expiresAt,
|
||||
consumed: false,
|
||||
});
|
||||
return { ticket, expiresAt };
|
||||
}
|
||||
|
||||
/**
|
||||
* Consume a ticket for a specific session (single-use). Returns the entry on
|
||||
* success, or null when: unknown, already consumed, expired, or bound to a
|
||||
* DIFFERENT session than `sessionId`.
|
||||
*/
|
||||
consume(ticket: string | null | undefined, sessionId: string): AttachTicketEntry | null {
|
||||
if (!ticket) return null;
|
||||
const entry = this.tickets.get(ticket);
|
||||
if (!entry) return null;
|
||||
if (entry.consumed) return null;
|
||||
if (this.now() > entry.expiresAt) {
|
||||
this.tickets.delete(ticket);
|
||||
return null;
|
||||
}
|
||||
if (entry.sessionId !== sessionId) return null;
|
||||
entry.consumed = true;
|
||||
// Keep briefly for attribution, but it can never be reused.
|
||||
return entry;
|
||||
}
|
||||
|
||||
/** Sweep expired tickets (best-effort housekeeping). */
|
||||
sweepExpired(): void {
|
||||
const now = this.now();
|
||||
for (const [ticket, entry] of this.tickets) {
|
||||
if (now > entry.expiresAt) this.tickets.delete(ticket);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Input-source attribution ─────────────────────────────────────────────────
|
||||
|
||||
export interface CliInputAttributionEntry {
|
||||
/** The ticket id the input arrived under (the accountability floor in v1). */
|
||||
ticketId: string;
|
||||
/** "ws" for WS input frames, "inject" for the REST inject route. */
|
||||
source: "ws" | "inject";
|
||||
/** Byte length of the input (not the content — content is not retained). */
|
||||
byteLength: number;
|
||||
at: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* In-memory per-session input-attribution log. v1 has no per-user arbitration,
|
||||
* so logging which ticket each input frame arrived under is the accountability
|
||||
* floor for post-incident attribution.
|
||||
*/
|
||||
export class CliInputAttributionLog {
|
||||
private readonly bySession = new Map<string, CliInputAttributionEntry[]>();
|
||||
private readonly cap: number;
|
||||
|
||||
constructor(opts?: { capPerSession?: number }) {
|
||||
this.cap = opts?.capPerSession ?? 1000;
|
||||
}
|
||||
|
||||
record(sessionId: string, entry: CliInputAttributionEntry): void {
|
||||
let list = this.bySession.get(sessionId);
|
||||
if (!list) {
|
||||
list = [];
|
||||
this.bySession.set(sessionId, list);
|
||||
}
|
||||
list.push(entry);
|
||||
if (list.length > this.cap) list.splice(0, list.length - this.cap);
|
||||
}
|
||||
|
||||
list(sessionId: string): CliInputAttributionEntry[] {
|
||||
return [...(this.bySession.get(sessionId) ?? [])];
|
||||
}
|
||||
}
|
||||
|
||||
// ── Confirm-advance (generic-tier R20 affordance) ────────────────────────────
|
||||
|
||||
export type CliConfirmAdvanceListener = (info: {
|
||||
sessionId: string;
|
||||
projectId: string;
|
||||
decision: "advance" | "not-yet";
|
||||
}) => void;
|
||||
|
||||
/**
|
||||
* The generic-tier "this session looks idle — advance to review?" affordance.
|
||||
* The engine pipeline layer acts on the event later; for now the transport
|
||||
* persists the latest decision per session and emits to subscribers (the engine
|
||||
* seam wires a listener in a later unit).
|
||||
*/
|
||||
export class CliConfirmAdvanceRegistry {
|
||||
private readonly latest = new Map<string, "advance" | "not-yet">();
|
||||
private readonly listeners = new Set<CliConfirmAdvanceListener>();
|
||||
|
||||
on(listener: CliConfirmAdvanceListener): () => void {
|
||||
this.listeners.add(listener);
|
||||
return () => this.listeners.delete(listener);
|
||||
}
|
||||
|
||||
record(sessionId: string, projectId: string, decision: "advance" | "not-yet"): void {
|
||||
this.latest.set(sessionId, decision);
|
||||
for (const listener of this.listeners) {
|
||||
listener({ sessionId, projectId, decision });
|
||||
}
|
||||
}
|
||||
|
||||
getLatest(sessionId: string): "advance" | "not-yet" | undefined {
|
||||
return this.latest.get(sessionId);
|
||||
}
|
||||
}
|
||||
|
||||
// ── Read-only enforcement ────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Whether a session is read-only (one-shot validator/planning, U9). Server-side
|
||||
* enforcement (not just client) — input/inject is rejected for these.
|
||||
*
|
||||
* U9's one-shot sessions are not yet wired, so the current signal is the
|
||||
* autonomy posture `readOnly` flag (forward-compatible) plus validator/planning
|
||||
* purposes which are inherently read-only.
|
||||
*/
|
||||
export function isReadOnlySession(session: CliSession): boolean {
|
||||
if (session.autonomyPosture && session.autonomyPosture.readOnly === true) return true;
|
||||
return session.purpose === "validator" || session.purpose === "planning";
|
||||
}
|
||||
|
||||
// ── Origin allowlist ─────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Origin allowlist check for the WS upgrade. Rejects a foreign or absent Origin
|
||||
* (a privileged keystroke channel must not be CSRF-attachable from another
|
||||
* page). Allowed:
|
||||
* - no Origin header AND not a browser context (native clients: TUI, our own WS
|
||||
* client set no Origin) — these are allowed because they aren't subject to the
|
||||
* browser same-origin model and authenticate via the daemon token + ticket.
|
||||
* We detect "browser context" by the presence of `Sec-Fetch-Site` / a
|
||||
* `User-Agent` claiming a browser; absent those, an absent Origin is a native
|
||||
* client.
|
||||
* - an Origin whose host matches the request Host (same-host), or
|
||||
* - an Origin in the configured extras allowlist.
|
||||
*
|
||||
* Per the plan, a FOREIGN or ABSENT Origin from a browser is rejected.
|
||||
*/
|
||||
export interface OriginCheckInput {
|
||||
origin: string | undefined;
|
||||
host: string | undefined;
|
||||
/** Sec-Fetch-Site header (present on browser-issued requests). */
|
||||
secFetchSite?: string | undefined;
|
||||
/** Extra allowed origins (exact, scheme+host[:port]) from config. */
|
||||
extraAllowedOrigins?: string[];
|
||||
}
|
||||
|
||||
// ── SSE state bridge ─────────────────────────────────────────────────────────
|
||||
|
||||
/** Default cap (chars) on the SSE last-output preview. */
|
||||
export const DEFAULT_OUTPUT_PREVIEW_CHARS = 200;
|
||||
|
||||
/**
|
||||
* Bound + sanitize a last-output preview for the SSE `cli:session:state` event:
|
||||
* ANSI/control-stripped, redacted, and capped (~200 chars). The text is already
|
||||
* expected to be recent output (e.g. the scrollback tail); we strip first, then
|
||||
* the engine's redaction is applied by the supplier — but we strip here defensively.
|
||||
*/
|
||||
export function buildOutputPreview(
|
||||
raw: string | undefined,
|
||||
maxChars = DEFAULT_OUTPUT_PREVIEW_CHARS,
|
||||
): string | undefined {
|
||||
if (!raw) return undefined;
|
||||
const stripped = stripAnsiControl(raw).replace(/\s+/g, " ").trim();
|
||||
if (stripped.length === 0) return undefined;
|
||||
return stripped.length > maxChars ? stripped.slice(-maxChars) : stripped;
|
||||
}
|
||||
|
||||
/**
|
||||
* Subscribe to an engine state machine's throttled `onStateChange` and forward
|
||||
* each transition onto the SSE bus as a `cli:session:state` event. The engine
|
||||
* applies the ~500ms throttle (via `stateChangeThrottleMs`); this bridge only
|
||||
* shapes the payload (owning entity + bounded redacted preview) and routes it
|
||||
* project-scoped.
|
||||
*
|
||||
* Returns an unsubscribe handle.
|
||||
*/
|
||||
export function bridgeCliStateToSse(
|
||||
machine: { onStateChange(listener: (change: CliStateChange) => void): () => void },
|
||||
deps: {
|
||||
store: Pick<CliSessionStore, "getSession">;
|
||||
/** Recent (raw) output supplier for the bounded preview (e.g. scrollback tail). */
|
||||
getRecentOutput?: (sessionId: string) => string | undefined;
|
||||
},
|
||||
): () => void {
|
||||
return machine.onStateChange((change) => {
|
||||
const session = deps.store.getSession(change.sessionId);
|
||||
const preview = buildOutputPreview(deps.getRecentOutput?.(change.sessionId));
|
||||
emitCliSessionStateSseEvent(
|
||||
{
|
||||
sessionId: change.sessionId,
|
||||
taskId: session?.taskId ?? null,
|
||||
chatSessionId: session?.chatSessionId ?? null,
|
||||
state: change.state,
|
||||
terminationReason: change.terminationReason,
|
||||
lastOutputPreview: preview,
|
||||
at: change.at,
|
||||
},
|
||||
session?.projectId,
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
export function isOriginAllowed(input: OriginCheckInput): boolean {
|
||||
const { origin, host, secFetchSite, extraAllowedOrigins } = input;
|
||||
|
||||
// No Origin header.
|
||||
if (!origin) {
|
||||
// A browser ALWAYS sends Origin on a cross-site WS and sets Sec-Fetch-Site;
|
||||
// an absent Origin together with a browser signal is suspicious → reject.
|
||||
if (secFetchSite && secFetchSite !== "none" && secFetchSite !== "same-origin") {
|
||||
return false;
|
||||
}
|
||||
// Native client (TUI / our WS client) — allowed (token + ticket gated).
|
||||
return true;
|
||||
}
|
||||
|
||||
// Parse the Origin.
|
||||
let originUrl: URL;
|
||||
try {
|
||||
originUrl = new URL(origin);
|
||||
} catch {
|
||||
return false; // malformed Origin → reject
|
||||
}
|
||||
|
||||
// Same-host: the Origin's host:port matches the request Host header.
|
||||
if (host && originUrl.host === host) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// Extras allowlist (exact scheme+host[:port]).
|
||||
if (extraAllowedOrigins && extraAllowedOrigins.length > 0) {
|
||||
const normalized = `${originUrl.protocol}//${originUrl.host}`;
|
||||
if (extraAllowedOrigins.some((o) => o === normalized || o === origin)) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
325
packages/dashboard/src/cli-session-ws.ts
Normal file
325
packages/dashboard/src/cli-session-ws.ts
Normal file
@@ -0,0 +1,325 @@
|
||||
/**
|
||||
* cli-session-ws — distinct WebSocket attach handler for CLI agent sessions
|
||||
* (CLI Agent Executor, U10).
|
||||
*
|
||||
* This is a SEPARATE connection handler from the existing terminal WS
|
||||
* (`/api/terminal/ws`): it shares only the upgrade gate shape (daemon-token
|
||||
* auth) and the JSON frame protocol style. The connection body resolves the
|
||||
* session from the engine's `CliSessionManager` (the explicit async attach
|
||||
* interface), never the dashboard-local terminal service.
|
||||
*
|
||||
* Path: /api/cli-sessions/ws?sessionId=<id>&ticket=<ticket>[&fn_token=<token>]
|
||||
*
|
||||
* Upgrade gate (stronger than the terminal WS — this channel injects keystrokes
|
||||
* into privileged agent PTYs):
|
||||
* 1. daemon-token auth (authenticateUpgradeRequest), AND
|
||||
* 2. an Origin allowlist check (reject foreign/absent browser Origin), AND
|
||||
* 3. a valid, unconsumed, single-use ticket bound to this exact session.
|
||||
*
|
||||
* Frame protocol (JSON, mirrors the terminal WS shape):
|
||||
* server→client: {type:"scrollback", data} (base64 bytes) — once, on connect
|
||||
* {type:"data", data} (base64 bytes) — live, neutralized
|
||||
* {type:"state", ...} (optional state hints)
|
||||
* {type:"error", message} (e.g. read-only input rejected)
|
||||
* client→server: {type:"input", data} (utf8 keystrokes → PTY write)
|
||||
* {type:"resize", cols, rows} (latest-active-client resize)
|
||||
* {type:"ack", bytes} (ACK-credit flow control)
|
||||
*
|
||||
* Flow control: the client ACKs bytes it has consumed; the server tracks
|
||||
* outstanding unacked bytes and, above the high watermark, calls
|
||||
* `manager.requestPause`; back below the low watermark, `manager.requestResume`.
|
||||
* The server never buffers PTY bytes — the engine owns the ring + pause/resume.
|
||||
*
|
||||
* Output hardening: every live `data` frame is passed through
|
||||
* `neutralizeTerminalOutput` (carry threaded across chunks) BEFORE send.
|
||||
*/
|
||||
|
||||
import { WebSocketServer, type WebSocket } from "ws";
|
||||
import type { IncomingMessage, Server as HttpServer } from "node:http";
|
||||
import { authenticateUpgradeRequest } from "./auth-middleware.js";
|
||||
import { neutralizeTerminalOutput, flushTerminalOutput } from "./cli-session-output-filter.js";
|
||||
import {
|
||||
isOriginAllowed,
|
||||
isReadOnlySession,
|
||||
type AttachTicketStore,
|
||||
type CliInputAttributionLog,
|
||||
type CliSessionTransportDeps,
|
||||
} from "./cli-session-transport.js";
|
||||
|
||||
/** Default high/low watermark (bytes) for ACK-credit backpressure. */
|
||||
export const DEFAULT_HIGH_WATERMARK_BYTES = 128 * 1024;
|
||||
export const DEFAULT_LOW_WATERMARK_BYTES = 16 * 1024;
|
||||
|
||||
export const CLI_SESSION_WS_PATH = "/api/cli-sessions/ws";
|
||||
|
||||
export interface CliSessionWebSocketOptions extends CliSessionTransportDeps {
|
||||
ticketStore: AttachTicketStore;
|
||||
attributionLog: CliInputAttributionLog;
|
||||
/** Daemon token; when set (and not noAuth) the upgrade requires it. */
|
||||
daemonToken?: string;
|
||||
noAuth?: boolean;
|
||||
/** Extra allowed WS Origins (scheme+host[:port]). */
|
||||
extraAllowedOrigins?: string[];
|
||||
highWatermarkBytes?: number;
|
||||
lowWatermarkBytes?: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Attach the CLI-session WebSocket handler to an HTTP server. Adds its OWN
|
||||
* `upgrade` listener filtered to the cli-sessions path, so it coexists with the
|
||||
* terminal/badge WS upgrade listeners (each ignores non-matching paths).
|
||||
*/
|
||||
export function setupCliSessionWebSocket(
|
||||
server: HttpServer,
|
||||
options: CliSessionWebSocketOptions,
|
||||
): WebSocketServer {
|
||||
const wss = new WebSocketServer({ noServer: true });
|
||||
const highWatermark = options.highWatermarkBytes ?? DEFAULT_HIGH_WATERMARK_BYTES;
|
||||
const lowWatermark = options.lowWatermarkBytes ?? DEFAULT_LOW_WATERMARK_BYTES;
|
||||
|
||||
server.on("upgrade", (req, socket, head) => {
|
||||
const pathname = new URL(req.url || "", `http://${req.headers.host}`).pathname;
|
||||
if (pathname !== CLI_SESSION_WS_PATH) return;
|
||||
|
||||
const reject = (code: number, reason: string) => {
|
||||
socket.write(`HTTP/1.1 ${code} ${reason}\r\nConnection: close\r\n\r\n`);
|
||||
socket.destroy();
|
||||
};
|
||||
|
||||
// 1. Daemon-token auth at the upgrade.
|
||||
if (options.daemonToken && !options.noAuth) {
|
||||
if (!authenticateUpgradeRequest(options.daemonToken, req)) {
|
||||
reject(401, "Unauthorized");
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// 2. Origin allowlist.
|
||||
const originOk = isOriginAllowed({
|
||||
origin: headerStr(req, "origin"),
|
||||
host: headerStr(req, "host"),
|
||||
secFetchSite: headerStr(req, "sec-fetch-site"),
|
||||
extraAllowedOrigins: options.extraAllowedOrigins,
|
||||
});
|
||||
if (!originOk) {
|
||||
reject(403, "Forbidden");
|
||||
return;
|
||||
}
|
||||
|
||||
wss.handleUpgrade(req, socket, head, (upgraded) => {
|
||||
wss.emit("connection", upgraded, req);
|
||||
});
|
||||
});
|
||||
|
||||
wss.on("connection", (ws: WebSocket, req: IncomingMessage) => {
|
||||
const url = new URL(req.url || "", `http://${req.headers.host}`);
|
||||
const sessionId = url.searchParams.get("sessionId");
|
||||
const ticket = url.searchParams.get("ticket");
|
||||
|
||||
if (!sessionId) {
|
||||
ws.close(4000, "Missing sessionId");
|
||||
return;
|
||||
}
|
||||
|
||||
const session = options.store.getSession(sessionId);
|
||||
if (!session) {
|
||||
ws.close(4004, "Session not found");
|
||||
return;
|
||||
}
|
||||
|
||||
// 3. Single-use, session-scoped ticket. Consume here (after upgrade) so a
|
||||
// replayed/consumed ticket, or a ticket for a different session, is rejected.
|
||||
const consumed = options.ticketStore.consume(ticket, sessionId);
|
||||
if (!consumed) {
|
||||
ws.close(4401, "Invalid or expired attach ticket");
|
||||
return;
|
||||
}
|
||||
if (consumed.projectId !== session.projectId) {
|
||||
ws.close(4503, "Ticket does not match session project");
|
||||
return;
|
||||
}
|
||||
|
||||
if (!options.manager.isLive(sessionId)) {
|
||||
ws.close(4409, "Session is not live");
|
||||
return;
|
||||
}
|
||||
|
||||
const readOnly = isReadOnlySession(session) || consumed.readOnly;
|
||||
const ticketId = consumed.ticket;
|
||||
|
||||
// Attach to the engine manager.
|
||||
let attachment;
|
||||
try {
|
||||
attachment = options.manager.attach(sessionId);
|
||||
} catch {
|
||||
ws.close(4500, "Failed to attach");
|
||||
return;
|
||||
}
|
||||
|
||||
// ── Flow control state ──
|
||||
let unacked = 0;
|
||||
let paused = false;
|
||||
const onSentBytes = (n: number) => {
|
||||
unacked += n;
|
||||
if (!paused && unacked >= highWatermark) {
|
||||
paused = true;
|
||||
try {
|
||||
options.manager.requestPause(sessionId);
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
}
|
||||
};
|
||||
const onAck = (n: number) => {
|
||||
if (!Number.isFinite(n) || n <= 0) return;
|
||||
unacked = Math.max(0, unacked - n);
|
||||
if (paused && unacked <= lowWatermark) {
|
||||
paused = false;
|
||||
try {
|
||||
options.manager.requestResume(sessionId);
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// ── Outbound: scrollback (neutralized) then live stream (neutralized) ──
|
||||
let carry = "";
|
||||
const sendData = (bytes: Uint8Array) => {
|
||||
if (ws.readyState !== ws.OPEN) return;
|
||||
const text = Buffer.from(bytes).toString("utf8");
|
||||
const result = neutralizeTerminalOutput(text, carry);
|
||||
carry = result.carry;
|
||||
if (result.output.length === 0) return;
|
||||
const payload = Buffer.from(result.output, "utf8");
|
||||
try {
|
||||
ws.send(
|
||||
JSON.stringify({ type: "data", data: payload.toString("base64") }),
|
||||
);
|
||||
onSentBytes(payload.byteLength);
|
||||
} catch {
|
||||
/* socket closing */
|
||||
}
|
||||
};
|
||||
|
||||
// Scrollback replay (run through the same neutralizer so a hostile sequence
|
||||
// recorded in scrollback is also stripped). Sent as its own frame so the
|
||||
// client can clear before replay.
|
||||
{
|
||||
const scrollText = Buffer.from(attachment.scrollback).toString("utf8");
|
||||
const result = neutralizeTerminalOutput(scrollText, "");
|
||||
// Flush carry into the scrollback frame (replay is a complete snapshot).
|
||||
const full = result.output + flushTerminalOutput(result.carry);
|
||||
try {
|
||||
ws.send(
|
||||
JSON.stringify({
|
||||
type: "scrollback",
|
||||
data: Buffer.from(full, "utf8").toString("base64"),
|
||||
}),
|
||||
);
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
}
|
||||
|
||||
ws.send(JSON.stringify({ type: "state", state: session.agentState, readOnly }));
|
||||
|
||||
// Pump the live byte stream.
|
||||
let streamClosed = false;
|
||||
(async () => {
|
||||
try {
|
||||
for await (const chunk of attachment.stream) {
|
||||
if (streamClosed || ws.readyState !== ws.OPEN) break;
|
||||
sendData(chunk);
|
||||
}
|
||||
} catch {
|
||||
/* stream error — close below */
|
||||
}
|
||||
if (!streamClosed && ws.readyState === ws.OPEN) {
|
||||
try {
|
||||
ws.send(JSON.stringify({ type: "exit" }));
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
}
|
||||
})();
|
||||
|
||||
// ── Inbound frames ──
|
||||
ws.on("message", (raw: Buffer) => {
|
||||
let msg: Record<string, unknown>;
|
||||
try {
|
||||
msg = JSON.parse(raw.toString());
|
||||
} catch {
|
||||
return; // ignore malformed
|
||||
}
|
||||
switch (msg.type) {
|
||||
case "input": {
|
||||
if (typeof msg.data !== "string") return;
|
||||
if (readOnly) {
|
||||
try {
|
||||
ws.send(
|
||||
JSON.stringify({
|
||||
type: "error",
|
||||
code: "READ_ONLY",
|
||||
message: "Session is read-only — input is not permitted",
|
||||
}),
|
||||
);
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
return;
|
||||
}
|
||||
attachment.write(msg.data);
|
||||
options.attributionLog.record(sessionId, {
|
||||
ticketId,
|
||||
source: "ws",
|
||||
byteLength: Buffer.byteLength(msg.data, "utf8"),
|
||||
at: new Date().toISOString(),
|
||||
});
|
||||
break;
|
||||
}
|
||||
case "resize": {
|
||||
if (typeof msg.cols === "number" && typeof msg.rows === "number") {
|
||||
// Latest-active-client policy is enforced by the manager (latest wins).
|
||||
attachment.resize(msg.cols, msg.rows);
|
||||
}
|
||||
break;
|
||||
}
|
||||
case "ack": {
|
||||
if (typeof msg.bytes === "number") onAck(msg.bytes);
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
const teardown = () => {
|
||||
if (streamClosed) return;
|
||||
streamClosed = true;
|
||||
try {
|
||||
attachment.detach(); // NEVER kills the session (other clients keep it).
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
// Release backpressure so a remaining client isn't stuck paused.
|
||||
if (paused) {
|
||||
try {
|
||||
options.manager.requestResume(sessionId);
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
ws.on("close", teardown);
|
||||
ws.on("error", teardown);
|
||||
});
|
||||
|
||||
return wss;
|
||||
}
|
||||
|
||||
function headerStr(req: IncomingMessage, name: string): string | undefined {
|
||||
const v = req.headers[name];
|
||||
if (typeof v === "string") return v;
|
||||
if (Array.isArray(v)) return v[0];
|
||||
return undefined;
|
||||
}
|
||||
162
packages/dashboard/src/routes/cli-sessions.ts
Normal file
162
packages/dashboard/src/routes/cli-sessions.ts
Normal file
@@ -0,0 +1,162 @@
|
||||
/**
|
||||
* cli-sessions routes — authenticated REST surface for CLI agent sessions
|
||||
* (CLI Agent Executor, U10).
|
||||
*
|
||||
* Daemon-token gated like every other /api route (the caller mounts this router
|
||||
* behind the auth middleware). Routes:
|
||||
* - GET /api/cli-sessions list (by project/task/chat)
|
||||
* - GET /api/cli-sessions/:id one session record
|
||||
* - POST /api/cli-sessions/:id/attach-ticket mint a short-lived single-use
|
||||
* session-scoped attach ticket
|
||||
* - POST /api/cli-sessions/:id/inject inject text onto the session FIFO
|
||||
* - POST /api/cli-sessions/:id/confirm-advance generic-tier R20 affordance
|
||||
*
|
||||
* Attach tickets (KTD — attach auth): the long-lived daemon token never
|
||||
* authorizes PTY write access by itself. A surface mints a ticket here (gated by
|
||||
* the daemon token), then presents it on the WS upgrade (which re-checks the
|
||||
* token AND an Origin allowlist). Tickets are single-use, ~60s TTL, and bound to
|
||||
* their session id.
|
||||
*/
|
||||
|
||||
import { Router, type Request, type Response } from "express";
|
||||
import { badRequest, notFound, ApiError, catchHandler } from "../api-error.js";
|
||||
import {
|
||||
type AttachTicketStore,
|
||||
type CliInputAttributionLog,
|
||||
type CliConfirmAdvanceRegistry,
|
||||
type CliSessionTransportDeps,
|
||||
isReadOnlySession,
|
||||
} from "../cli-session-transport.js";
|
||||
|
||||
export interface CliSessionRoutesOptions extends CliSessionTransportDeps {
|
||||
ticketStore: AttachTicketStore;
|
||||
attributionLog: CliInputAttributionLog;
|
||||
confirmAdvance: CliConfirmAdvanceRegistry;
|
||||
/** Max inject body length (chars). Bounds a hostile body. */
|
||||
maxInjectChars?: number;
|
||||
}
|
||||
|
||||
const DEFAULT_MAX_INJECT_CHARS = 64 * 1024;
|
||||
|
||||
/** Coerce a route param (may be string | string[]) to a single string. */
|
||||
function paramId(value: unknown): string {
|
||||
return Array.isArray(value) ? String(value[0]) : String(value);
|
||||
}
|
||||
|
||||
/** Optional project scoping: when a `projectId` is provided it must match. */
|
||||
function assertProjectScope(sessionProjectId: string, requested: unknown): void {
|
||||
if (requested === undefined || requested === null || requested === "") return;
|
||||
if (typeof requested !== "string" || requested !== sessionProjectId) {
|
||||
// Cross-project access is a hard rejection (the session id is not enough).
|
||||
throw new ApiError(403, "Session does not belong to this project");
|
||||
}
|
||||
}
|
||||
|
||||
export function createCliSessionsRouter(options: CliSessionRoutesOptions): Router {
|
||||
const { manager, store, ticketStore, attributionLog, confirmAdvance } = options;
|
||||
const maxInjectChars = options.maxInjectChars ?? DEFAULT_MAX_INJECT_CHARS;
|
||||
const router = Router();
|
||||
|
||||
// ── List ────────────────────────────────────────────────────────────────
|
||||
router.get(
|
||||
"/",
|
||||
catchHandler(async (req: Request, res: Response) => {
|
||||
const { projectId, taskId, chatSessionId } = req.query;
|
||||
const sessions = store.listSessions({
|
||||
projectId: typeof projectId === "string" ? projectId : undefined,
|
||||
taskId: typeof taskId === "string" ? taskId : undefined,
|
||||
chatSessionId: typeof chatSessionId === "string" ? chatSessionId : undefined,
|
||||
});
|
||||
res.json({ sessions });
|
||||
}),
|
||||
);
|
||||
|
||||
// ── One ─────────────────────────────────────────────────────────────────
|
||||
router.get(
|
||||
"/:id",
|
||||
catchHandler(async (req: Request, res: Response) => {
|
||||
const session = store.getSession(paramId(req.params.id));
|
||||
if (!session) throw notFound("Session not found");
|
||||
assertProjectScope(session.projectId, req.query.projectId);
|
||||
res.json({ session });
|
||||
}),
|
||||
);
|
||||
|
||||
// ── Attach ticket ─────────────────────────────────────────────────────────
|
||||
router.post(
|
||||
"/:id/attach-ticket",
|
||||
catchHandler(async (req: Request, res: Response) => {
|
||||
const session = store.getSession(paramId(req.params.id));
|
||||
if (!session) throw notFound("Session not found");
|
||||
assertProjectScope(session.projectId, req.body?.projectId ?? req.query.projectId);
|
||||
const readOnly = isReadOnlySession(session);
|
||||
const { ticket, expiresAt } = ticketStore.mint({
|
||||
sessionId: session.id,
|
||||
projectId: session.projectId,
|
||||
readOnly,
|
||||
});
|
||||
res.json({
|
||||
ticket,
|
||||
expiresAt: new Date(expiresAt).toISOString(),
|
||||
readOnly,
|
||||
});
|
||||
}),
|
||||
);
|
||||
|
||||
// ── Inject ────────────────────────────────────────────────────────────────
|
||||
router.post(
|
||||
"/:id/inject",
|
||||
catchHandler(async (req: Request, res: Response) => {
|
||||
const session = store.getSession(paramId(req.params.id));
|
||||
if (!session) throw notFound("Session not found");
|
||||
assertProjectScope(session.projectId, req.body?.projectId ?? req.query.projectId);
|
||||
|
||||
const text = req.body?.text;
|
||||
if (typeof text !== "string" || text.length === 0) {
|
||||
throw badRequest("Missing or empty `text`");
|
||||
}
|
||||
if (text.length > maxInjectChars) {
|
||||
throw badRequest(`\`text\` exceeds max length (${maxInjectChars})`);
|
||||
}
|
||||
// Server-side read-only enforcement (not just client).
|
||||
if (isReadOnlySession(session)) {
|
||||
throw new ApiError(403, "Session is read-only — input is not permitted");
|
||||
}
|
||||
if (!manager.isLive(session.id)) {
|
||||
throw new ApiError(409, "Session is not live");
|
||||
}
|
||||
|
||||
await manager.inject(session.id, text);
|
||||
|
||||
// Attribution: record the input source on the session's input log. The
|
||||
// REST inject path has no attach ticket; attribute it to the daemon token.
|
||||
attributionLog.record(session.id, {
|
||||
ticketId: "rest:inject",
|
||||
source: "inject",
|
||||
byteLength: Buffer.byteLength(text, "utf8"),
|
||||
at: new Date().toISOString(),
|
||||
});
|
||||
|
||||
res.json({ ok: true });
|
||||
}),
|
||||
);
|
||||
|
||||
// ── Confirm-advance (generic-tier R20 affordance) ──────────────────────────
|
||||
router.post(
|
||||
"/:id/confirm-advance",
|
||||
catchHandler(async (req: Request, res: Response) => {
|
||||
const session = store.getSession(paramId(req.params.id));
|
||||
if (!session) throw notFound("Session not found");
|
||||
assertProjectScope(session.projectId, req.body?.projectId ?? req.query.projectId);
|
||||
|
||||
const decisionRaw = req.body?.decision ?? "advance";
|
||||
if (decisionRaw !== "advance" && decisionRaw !== "not-yet") {
|
||||
throw badRequest("`decision` must be 'advance' or 'not-yet'");
|
||||
}
|
||||
confirmAdvance.record(session.id, session.projectId, decisionRaw);
|
||||
res.json({ ok: true, decision: decisionRaw });
|
||||
}),
|
||||
);
|
||||
|
||||
return router;
|
||||
}
|
||||
@@ -59,6 +59,8 @@ import { ChatManager } from "./chat.js";
|
||||
import { stopAllDevServers } from "./dev-server-routes.js";
|
||||
import type { SkillsAdapter } from "./skills-adapter.js";
|
||||
import { createAuthMiddleware, authenticateUpgradeRequest, getDaemonToken } from "./auth-middleware.js";
|
||||
import { setupCliSessionWebSocket } from "./cli-session-ws.js";
|
||||
import { createCliSessionsRouter } from "./routes/cli-sessions.js";
|
||||
import { validateRemoteAuthToken } from "./remote-auth.js";
|
||||
import { getCliPackageVersion } from "./cli-package-version.js";
|
||||
import {
|
||||
@@ -223,6 +225,19 @@ export interface ServerOptions {
|
||||
};
|
||||
/** Optional AiSessionStore — if not provided, one is created from the default store's database */
|
||||
aiSessionStore?: AiSessionStore;
|
||||
/**
|
||||
* Optional CLI agent session transport dependencies (CLI Agent Executor, U10).
|
||||
* When provided, the server mounts the cli-sessions REST routes and the
|
||||
* distinct `/api/cli-sessions/ws` attach handler. Wiring the engine-owned
|
||||
* CliSessionManager/store into this dep happens in a later unit; until then
|
||||
* the transport is inert unless explicitly supplied (e.g. in tests).
|
||||
*/
|
||||
cliSessionTransport?: import("./cli-session-transport.js").CliSessionTransportDeps & {
|
||||
ticketStore: import("./cli-session-transport.js").AttachTicketStore;
|
||||
attributionLog: import("./cli-session-transport.js").CliInputAttributionLog;
|
||||
confirmAdvance: import("./cli-session-transport.js").CliConfirmAdvanceRegistry;
|
||||
extraAllowedOrigins?: string[];
|
||||
};
|
||||
/** Optional MissionAutopilot for autonomous mission progression */
|
||||
missionAutopilot?: {
|
||||
watchMission(missionId: string): void;
|
||||
@@ -1435,6 +1450,21 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
|
||||
});
|
||||
app.use("/api", apiRouter);
|
||||
|
||||
// CLI agent session REST routes (U10). Daemon-token gated by the app-level
|
||||
// auth middleware. Mounted only when transport deps are supplied.
|
||||
if (options?.cliSessionTransport) {
|
||||
app.use(
|
||||
"/api/cli-sessions",
|
||||
createCliSessionsRouter({
|
||||
manager: options.cliSessionTransport.manager,
|
||||
store: options.cliSessionTransport.store,
|
||||
ticketStore: options.cliSessionTransport.ticketStore,
|
||||
attributionLog: options.cliSessionTransport.attributionLog,
|
||||
confirmAdvance: options.cliSessionTransport.confirmAdvance,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
// API 404 Handler - Return JSON for unmatched API routes (instead of falling through to SPA)
|
||||
app.use("/api", (_req: express.Request, res: express.Response) => {
|
||||
sendErrorResponse(res, 404, "Not found");
|
||||
@@ -1548,6 +1578,20 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
|
||||
const websocketOptions = { ...options, runtimeLogger };
|
||||
setupTerminalWebSocket(dashboardApp, server as HttpServer, store, websocketOptions);
|
||||
setupBadgeWebSocket(dashboardApp, server as HttpServer, store, websocketOptions);
|
||||
// CLI agent session attach WS (U10) — distinct handler, shares only the
|
||||
// upgrade-gate shape with the terminal WS. Mounted only when transport
|
||||
// deps are supplied (engine wiring lands in a later unit).
|
||||
if (options?.cliSessionTransport) {
|
||||
setupCliSessionWebSocket(server as HttpServer, {
|
||||
manager: options.cliSessionTransport.manager,
|
||||
store: options.cliSessionTransport.store,
|
||||
ticketStore: options.cliSessionTransport.ticketStore,
|
||||
attributionLog: options.cliSessionTransport.attributionLog,
|
||||
daemonToken: getDaemonToken(options),
|
||||
noAuth: options?.noAuth,
|
||||
extraAllowedOrigins: options.cliSessionTransport.extraAllowedOrigins,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
return server as HttpServer;
|
||||
|
||||
@@ -281,6 +281,72 @@ export function emitPluginCustomSseEvent(
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* CLI agent session state transitions (CLI Agent Executor, U10). The engine
|
||||
* state machine's `onStateChange` (throttled ~500ms in the engine) is bridged
|
||||
* into this seam by `setupCliSessionTransport`; every open SSE stream forwards
|
||||
* a matching (project-scoped) `cli:session:state` event so cards / banners
|
||||
* update without touching the byte stream.
|
||||
*
|
||||
* Events are also appended to a small module-level ring buffer with monotonic
|
||||
* ids so a client reconnecting with `Last-Event-ID` replays the transitions it
|
||||
* missed (the byte stream is on a separate WS channel; this is state only).
|
||||
*/
|
||||
export interface CliSessionStateSsePayload {
|
||||
sessionId: string;
|
||||
taskId: string | null;
|
||||
chatSessionId: string | null;
|
||||
/** Machine state (may be the transient "resuming"). */
|
||||
state: string;
|
||||
terminationReason?: string | null;
|
||||
/** Bounded (~200 chars), ANSI-stripped, redacted preview of recent output. */
|
||||
lastOutputPreview?: string;
|
||||
at: string;
|
||||
}
|
||||
|
||||
type CliSessionStateSseListener = (
|
||||
id: number,
|
||||
payload: CliSessionStateSsePayload,
|
||||
projectId?: string,
|
||||
) => void;
|
||||
|
||||
const cliSessionStateSseListeners = new Set<CliSessionStateSseListener>();
|
||||
|
||||
/** Module-level ring buffer of cli-session-state events for lastEventId replay. */
|
||||
const cliSessionStateBuffer: { id: number; payload: CliSessionStateSsePayload; projectId?: string }[] =
|
||||
[];
|
||||
let cliSessionStateNextId = 1;
|
||||
const CLI_SESSION_STATE_BUFFER_CAP = 200;
|
||||
|
||||
export function emitCliSessionStateSseEvent(
|
||||
payload: CliSessionStateSsePayload,
|
||||
projectId?: string,
|
||||
): number {
|
||||
const id = cliSessionStateNextId++;
|
||||
cliSessionStateBuffer.push({ id, payload, projectId });
|
||||
if (cliSessionStateBuffer.length > CLI_SESSION_STATE_BUFFER_CAP) {
|
||||
cliSessionStateBuffer.splice(0, cliSessionStateBuffer.length - CLI_SESSION_STATE_BUFFER_CAP);
|
||||
}
|
||||
for (const listener of cliSessionStateSseListeners) {
|
||||
listener(id, payload, projectId);
|
||||
}
|
||||
return id;
|
||||
}
|
||||
|
||||
/** Buffered cli-session-state events with id > lastEventId (for reconnect replay). */
|
||||
export function getCliSessionStateEventsSince(
|
||||
lastEventId: number,
|
||||
): { id: number; payload: CliSessionStateSsePayload; projectId?: string }[] {
|
||||
if (!Number.isFinite(lastEventId)) return [...cliSessionStateBuffer];
|
||||
return cliSessionStateBuffer.filter((entry) => entry.id > lastEventId);
|
||||
}
|
||||
|
||||
/** Test seam: reset the cli-session-state buffer between tests. */
|
||||
export function resetCliSessionStateBufferForTests(): void {
|
||||
cliSessionStateBuffer.length = 0;
|
||||
cliSessionStateNextId = 1;
|
||||
}
|
||||
|
||||
/**
|
||||
* Normalized plugin lifecycle payload emitted via SSE.
|
||||
* This is the stable contract the UI can reconcile.
|
||||
@@ -650,6 +716,11 @@ export function createSSE(
|
||||
send(`event: plugin:custom\ndata: ${JSON.stringify({ pluginId, event, payload })}\n\n`);
|
||||
};
|
||||
|
||||
const onCliSessionStateEvent: CliSessionStateSseListener = (id, payload, eventProjectId) => {
|
||||
if (projectId && eventProjectId && eventProjectId !== projectId) return;
|
||||
send(`id: ${id}\nevent: cli:session:state\ndata: ${JSON.stringify(payload)}\n\n`);
|
||||
};
|
||||
|
||||
// --- Chat store event handlers ---
|
||||
const onChatSessionCreated = (session: unknown) => {
|
||||
send(`event: chat:session:created\ndata: ${JSON.stringify(session)}\n\n`);
|
||||
@@ -801,6 +872,7 @@ export function createSSE(
|
||||
approvalSseListeners.delete(onApprovalEvent);
|
||||
workflowSseListeners.delete(onWorkflowEvent);
|
||||
pluginCustomSseListeners.delete(onPluginCustomEvent);
|
||||
cliSessionStateSseListeners.delete(onCliSessionStateEvent);
|
||||
if (chatStore) {
|
||||
chatStore.off("chat:session:created", onChatSessionCreated);
|
||||
chatStore.off("chat:session:updated", onChatSessionUpdated);
|
||||
@@ -949,6 +1021,30 @@ export function createSSE(
|
||||
approvalSseListeners.add(onApprovalEvent);
|
||||
workflowSseListeners.add(onWorkflowEvent);
|
||||
pluginCustomSseListeners.add(onPluginCustomEvent);
|
||||
cliSessionStateSseListeners.add(onCliSessionStateEvent);
|
||||
|
||||
// Replay any cli-session-state transitions missed since the client's
|
||||
// Last-Event-ID (reconnect recovery — the byte stream is on a separate WS
|
||||
// channel, so only state events are replayed here).
|
||||
{
|
||||
const headerVal = _req.headers?.["last-event-id"];
|
||||
const lastEventIdRaw =
|
||||
(typeof headerVal === "string"
|
||||
? headerVal
|
||||
: Array.isArray(headerVal)
|
||||
? headerVal[0]
|
||||
: undefined) ??
|
||||
(typeof _req.query?.lastEventId === "string" ? _req.query.lastEventId : undefined);
|
||||
const lastEventId = lastEventIdRaw !== undefined ? Number(lastEventIdRaw) : NaN;
|
||||
if (Number.isFinite(lastEventId)) {
|
||||
for (const entry of getCliSessionStateEventsSince(lastEventId)) {
|
||||
if (projectId && entry.projectId && entry.projectId !== projectId) continue;
|
||||
send(
|
||||
`id: ${entry.id}\nevent: cli:session:state\ndata: ${JSON.stringify(entry.payload)}\n\n`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
registerManagedConnection({
|
||||
id: connectionId,
|
||||
|
||||
@@ -623,3 +623,33 @@ export {
|
||||
getNativePrebuildName,
|
||||
resetPtyModuleCacheForTests,
|
||||
} from "./pty-native.js";
|
||||
// CLI agent executor (U2/U3) — transport surfaces (U10) consume these.
|
||||
export {
|
||||
CliSessionManager,
|
||||
CliConcurrencyLimitError,
|
||||
UnknownCliSessionError,
|
||||
neutralizeInjection,
|
||||
DEFAULT_SCROLLBACK_BYTES,
|
||||
DEFAULT_CONCURRENCY_CEILING,
|
||||
type CliSessionAttachment,
|
||||
type CliSessionManagerOptions,
|
||||
type SpawnCliSessionOptions,
|
||||
} from "./cli-agent/session-manager.js";
|
||||
export {
|
||||
TelemetryHub,
|
||||
stripAnsiControl,
|
||||
type TelemetryHubOptions,
|
||||
type TelemetryEvent,
|
||||
type TelemetryEventKind,
|
||||
type SanitizedTelemetryEvent,
|
||||
} from "./cli-agent/telemetry-hub.js";
|
||||
export {
|
||||
CliSessionStateMachine,
|
||||
classifyTermination,
|
||||
isResumeEligible,
|
||||
toPersistedState,
|
||||
type CliMachineState,
|
||||
type CliStateChange,
|
||||
type CliStateChangeListener,
|
||||
type CliStateMachineOptions,
|
||||
} from "./cli-agent/state-machine.js";
|
||||
|
||||
Reference in New Issue
Block a user