/** * F4 — the WS agent gateway round-trip against the B6 frame contract: * `agent.launch` → unified `terminal.attached` (sessionId + assignedConversationId * consumed), then output/input/resize reuse the terminal mechanics; agent * re-attach uses `terminal.attach` (no relaunch) and returns scrollback; a * structured agent (`UNSUPPORTED`) and a dead session (`NOT_FOUND`) reject clearly. */ import { describe, it, expect, vi } from "vitest"; import { HttpAgentGateway } from "./streamGateways"; import { HttpInvoker, type FetchLike } from "./httpInvoker"; import { WsLiveClient, type WebSocketLike } from "./wsLiveClient"; import { bytesToBase64 } from "./frames"; class FakeSocket implements WebSocketLike { sent: string[] = []; onopen: (() => void) | null = null; onmessage: ((event: { data: string }) => void) | null = null; onerror: ((event: unknown) => void) | null = null; onclose: (() => void) | null = null; send(data: string): void { this.sent.push(data); } close(): void { this.onclose?.(); } receive(frame: unknown): void { this.onmessage?.({ data: JSON.stringify(frame) }); } } function gateway(): { gw: HttpAgentGateway; sockets: FakeSocket[] } { const sockets: FakeSocket[] = []; const ws = new WsLiveClient({ wsUrl: "wss://host", socketFactory: () => { const s = new FakeSocket(); sockets.push(s); queueMicrotask(() => s.onopen?.()); return s; }, }); // The HTTP invoker is unused by the WS paths under test. const gw = new HttpAgentGateway(new HttpInvoker({ baseUrl: "https://host" }), ws); return { gw, sockets }; } function httpAgentGateway( fetchImpl: FetchLike, ): { gw: HttpAgentGateway; calls: { url: string; init: unknown }[] } { const calls: { url: string; init: unknown }[] = []; const recordingFetch: FetchLike = async (url, init) => { calls.push({ url, init }); return fetchImpl(url, init); }; const ws = new WsLiveClient({ wsUrl: "wss://host", socketFactory: () => new FakeSocket(), }); return { gw: new HttpAgentGateway( new HttpInvoker({ baseUrl: "https://host", fetchImpl: recordingFetch }), ws, ), calls, }; } async function replyToLast( socket: FakeSocket, index: number, reply: (id: string) => unknown, ): Promise> { await vi.waitFor(() => expect(socket.sent.length).toBeGreaterThan(index)); const sent = JSON.parse(socket.sent[index]); socket.receive(reply(sent.id)); return sent; } function attachedAck( id: string, sessionId: string, opts: { scroll?: Uint8Array; assignedConversationId?: string } = {}, ) { return { kind: "terminal.attached", replyTo: id, payload: { session: { sessionId, rows: 24, cols: 80 }, scrollback: opts.scroll ? [{ seq: 0, bytesBase64: bytesToBase64(opts.scroll) }] : [], nextSeq: opts.scroll ? 1 : 0, status: "running", gap: false, assignedConversationId: opts.assignedConversationId, }, }; } const OPTS = { cwd: "/srv/app", rows: 24, cols: 80, nodeId: "node-1" }; describe("HttpAgentGateway WS round-trip (B6 frames)", () => { it("readContext unwraps the read_agent_context DTO content over HTTP", async () => { const { gw, calls } = httpAgentGateway(async () => ({ ok: true, status: 200, json: async () => ({ content: "## Restored context" }), text: async () => JSON.stringify({ content: "## Restored context" }), })); await expect(gw.readContext("proj-1", "agent-1")).resolves.toBe( "## Restored context", ); expect(calls).toHaveLength(1); const init = calls[0].init as { body: string }; expect(JSON.parse(init.body)).toEqual({ command: "read_agent_context", args: { projectId: "proj-1", agentId: "agent-1" }, }); }); it("updates agent contactability through the dedicated request DTO", async () => { const response = { id: "agent-1", contactable: true }; const { gw, calls } = httpAgentGateway(async () => ({ ok: true, status: 200, json: async () => response, text: async () => JSON.stringify(response), })); await expect( gw.updateAgentContactable("proj-1", "agent-1", true), ).resolves.toEqual(response); const init = calls[0].init as { body: string }; expect(JSON.parse(init.body)).toEqual({ command: "update_agent_contactable", args: { request: { projectId: "proj-1", agentId: "agent-1", contactable: true }, }, }); }); it("restarts the exact cell runtime through the HTTP command", async () => { const response = { restartId: "restart-7", status: "scheduled", agentId: "agent-1", nodeId: "node-2", kind: "pty", presentation: "tui", layoutId: null, }; const { gw, calls } = httpAgentGateway(async () => ({ ok: true, status: 200, json: async () => response, text: async () => JSON.stringify(response), })); await expect( gw.restartCellAgentRuntime({ projectId: "proj-1", nodeId: "node-2", agentId: "agent-1", presentation: "tui", }), ).resolves.toEqual(response); const init = calls[0].init as { body: string }; expect(JSON.parse(init.body)).toEqual({ command: "restart_cell_agent_runtime", args: { request: { projectId: "proj-1", nodeId: "node-2", agentId: "agent-1", presentation: "tui", }, }, }); }); it("updates the project orchestrator through the dedicated request DTO", async () => { const response = { orchestratorAgentId: "agent-2" }; const { gw, calls } = httpAgentGateway(async () => ({ ok: true, status: 200, json: async () => response, text: async () => JSON.stringify(response), })); await expect( gw.updateProjectOrchestrator("proj-1", "agent-2"), ).resolves.toEqual(response); const init = calls[0].init as { body: string }; expect(JSON.parse(init.body)).toEqual({ command: "update_project_orchestrator", args: { request: { projectId: "proj-1", agentId: "agent-2" } }, }); }); it("launch → attached: conforming agent.launch frame, assignedConversationId consumed", async () => { const { gw, sockets } = gateway(); const chunks: Uint8Array[] = []; const handlePromise = gw.launchAgent("proj-1", "agent-1", { ...OPTS, conversationId: "resume-9" }, (b) => chunks.push(b), ); const sent = await replyToLast(sockets[0], 0, (id) => attachedAck(id, "sess-1", { assignedConversationId: "conv-1" }), ); const handle = await handlePromise; // B6 payload: flat camelCase, no cwd (agent launch resolves the project). expect(sent.kind).toBe("agent.launch"); expect(sent.payload).toEqual({ projectId: "proj-1", agentId: "agent-1", nodeId: "node-1", rows: 24, cols: 80, conversationId: "resume-9", freshStart: false, }); expect(handle.sessionId).toBe("sess-1"); // The conversation id minted by the launch is surfaced on the handle. expect(handle.assignedConversationId).toBe("conv-1"); // Fresh launch: nothing repainted. expect(chunks).toHaveLength(0); // Output/input/resize reuse the terminal mechanics. sockets[0].receive({ kind: "terminal.output", payload: { sessionId: "sess-1", seq: 1, bytesBase64: bytesToBase64(new Uint8Array([65])) }, }); expect(chunks).toEqual([new Uint8Array([65])]); const base = sockets[0].sent.length; await handle.write(new Uint8Array([13])); await handle.resize(30, 100); expect(JSON.parse(sockets[0].sent[base]).kind).toBe("terminal.input"); expect(JSON.parse(sockets[0].sent[base + 1])).toMatchObject({ kind: "terminal.resize", payload: { sessionId: "sess-1", rows: 30, cols: 100 }, }); }); it("omits assignedConversationId on the handle when the launch minted none", async () => { const { gw, sockets } = gateway(); const handlePromise = gw.launchAgent("p", "a", OPTS, () => {}); await replyToLast(sockets[0], 0, (id) => attachedAck(id, "sess-x")); const handle = await handlePromise; expect(handle.assignedConversationId).toBeUndefined(); }); it("re-attaches to an existing agent via terminal.attach (no relaunch) and returns scrollback", async () => { const { gw, sockets } = gateway(); const chunks: Uint8Array[] = []; const scroll = new Uint8Array([104, 105]); const resPromise = gw.reattach("sess-7", (b) => chunks.push(b)); const sent = await replyToLast(sockets[0], 0, (id) => attachedAck(id, "sess-7", { scroll })); const res = await resPromise; // Re-attach must NOT relaunch: the frame is terminal.attach, not agent.launch. expect(sent.kind).toBe("terminal.attach"); expect(sent.payload).toMatchObject({ sessionId: "sess-7" }); expect(res.scrollback).toEqual(scroll); // Scrollback returned for the view to repaint, not double-pushed via onData. expect(chunks).toHaveLength(0); }); it("rejects a structured agent with UNSUPPORTED (no crash)", async () => { const { gw, sockets } = gateway(); const handlePromise = gw.launchAgent("p", "a", OPTS, () => {}); await replyToLast(sockets[0], 0, (id) => ({ kind: "error", replyTo: id, payload: { code: "UNSUPPORTED", message: "structured agent sessions do not stream over the PTY websocket" }, })); await expect(handlePromise).rejects.toMatchObject({ code: "UNSUPPORTED" }); }); it("rejects re-attach to a missing/exited session with NOT_FOUND", async () => { const { gw, sockets } = gateway(); const resPromise = gw.reattach("gone", () => {}); await replyToLast(sockets[0], 0, (id) => ({ kind: "error", replyTo: id, payload: { code: "NOT_FOUND", message: "terminal session not found" }, })); await expect(resPromise).rejects.toMatchObject({ code: "NOT_FOUND" }); }); it("routes structured chat through chat.* frames and HTTP close/cancel commands", async () => { const calls: { url: string; init: unknown }[] = []; const socket = new FakeSocket(); const ws = new WsLiveClient({ wsUrl: "wss://host", socketFactory: () => { queueMicrotask(() => socket.onopen?.()); return socket; }, }); const gw = new HttpAgentGateway( new HttpInvoker({ baseUrl: "https://host", fetchImpl: async (url, init) => { calls.push({ url, init }); return { ok: true, status: 200, json: async () => undefined, text: async () => "", }; }, }), ws, ); const launched = gw.launchAgentChat("proj-1", "agent-1", { cwd: "/srv/app", nodeId: "node-1", rows: 24, cols: 80, conversationId: "resume-9", }); const launch = await replyToLast(socket, 0, (id) => ({ kind: "chat.attached", replyTo: id, payload: { sessionId: "chat-1", cellKind: "chat", assignedConversationId: "conv-1", transcriptUnavailableReason: "emptyConversation", }, })); await expect(launched).resolves.toEqual({ sessionId: "chat-1", cellKind: "chat", assignedConversationId: "conv-1", transcriptUnavailableReason: "emptyConversation", }); expect(launch.kind).toBe("agent.launch"); expect(launch.payload).toEqual({ projectId: "proj-1", agentId: "agent-1", nodeId: "node-1", rows: 24, cols: 80, cellKind: "chat", conversationId: "resume-9", freshStart: false, }); const reattached = gw.reattachAgentChat("chat-1", () => {}, { projectId: "proj-1", agentId: "agent-1", }); const attach = await replyToLast(socket, 1, (id) => ({ kind: "chat.attached", replyTo: id, payload: { sessionId: "chat-1", conversationId: "conv-replacement", scrollback: [{ kind: "final", content: "Recovered reply" }], transcriptUnavailableReason: "missingTranscript", }, })); await expect(reattached).resolves.toEqual({ sessionId: "chat-1", conversationId: "conv-replacement", scrollback: [{ kind: "final", content: "Recovered reply" }], transcriptUnavailableReason: "missingTranscript", }); expect(attach.kind).toBe("chat.attach"); expect(attach.payload).toEqual({ sessionId: "chat-1", projectId: "proj-1", agentId: "agent-1", }); const chunks: unknown[] = []; const sent = gw.sendAgentChat("chat-1", "Hello", (chunk) => chunks.push(chunk), { attachments: [{ path: "/tmp/note.txt", filename: "note.txt" }], }); const chatSend = await replyToLast(socket, 2, (id) => ({ kind: "chat.output", replyTo: id, payload: { sessionId: "chat-1", chunk: { kind: "final", content: "Hello back" }, }, })); await sent; expect(chatSend.kind).toBe("chat.send"); expect(chatSend.payload).toEqual({ sessionId: "chat-1", prompt: "Hello", attachments: [{ path: "/tmp/note.txt", filename: "note.txt" }], }); expect(chunks).toEqual([{ kind: "final", content: "Hello back" }]); await gw.cancelAgentChat("chat-1"); await gw.closeAgentChat("chat-1"); await gw.detachAgentChat("chat-1"); expect(calls).toHaveLength(2); expect(JSON.parse((calls[0].init as { body: string }).body)).toEqual({ command: "cancel_agent_chat", args: { sessionId: "chat-1" }, }); expect(JSON.parse((calls[1].init as { body: string }).body)).toEqual({ command: "close_agent_session", args: { sessionId: "chat-1" }, }); expect(JSON.parse(socket.sent[3])).toMatchObject({ kind: "chat.detach", payload: { sessionId: "chat-1" }, }); }); });