diff --git a/platform_web/api/client.test.ts b/platform_web/api/client.test.ts index 7889b6a..dd3091e 100644 --- a/platform_web/api/client.test.ts +++ b/platform_web/api/client.test.ts @@ -778,6 +778,30 @@ describe("PlatformApiClient AI providers", () => { expect(client.serverLogEventsUrl("server-1")).toBe("/api/v1/server-instances/server-1/logs/events"); }); + it("streams terminal log SSE with bearer authorization and dispatches live events", async () => { + const payload = "event: stream\ndata: {\"id\":\"log-1\",\"serverInstanceId\":\"server-1\"}\n\nevent: log\ndata: {\"streamId\":\"log-1\",\"entry\":{\"seq\":3,\"line\":\"live line\"}}\n\nevent: ready\ndata: {\"serverInstanceId\":\"server-1\"}\n\n"; + const fetchMock = vi.fn(async (_input: RequestInfo | URL, init?: RequestInit) => { + expect(new Headers(init?.headers).get("Authorization")).toBe("Bearer terminal-session"); + expect(new Headers(init?.headers).get("Accept")).toBe("text/event-stream"); + return new Response(new ReadableStream({ start(controller) { controller.enqueue(new TextEncoder().encode(payload)); controller.close(); } }), { status: 200, headers: { "Content-Type": "text/event-stream" } }); + }); + vi.stubGlobal("fetch", fetchMock); + const client = new PlatformApiClient("/api/v1", () => "terminal-session"); + const events: string[] = []; + const stream = client.openServerLogEvents("server-1", { historyLimit: 2 }); + stream.addEventListener("stream", (event) => events.push(`stream:${event.data}`)); + stream.addEventListener("log", (event) => events.push(`log:${event.data}`)); + stream.addEventListener("ready", (event) => events.push(`ready:${event.data}`)); + + await vi.waitFor(() => expect(events).toEqual([ + 'stream:{"id":"log-1","serverInstanceId":"server-1"}', + 'log:{"streamId":"log-1","entry":{"seq":3,"line":"live line"}}', + 'ready:{"serverInstanceId":"server-1"}' + ])); + expect(String(fetchMock.mock.calls[0]?.[0])).toBe("/api/v1/server-instances/server-1/logs/events?historyLimit=2"); + stream.close(); + }); + it("surfaces password confirmation denials without exposing generic forbidden text", async () => { vi.stubGlobal("fetch", vi.fn(async () => new Response(JSON.stringify({ code: "forbidden", diff --git a/platform_web/api/client.ts b/platform_web/api/client.ts index 9b6e806..ecb4b29 100644 --- a/platform_web/api/client.ts +++ b/platform_web/api/client.ts @@ -152,6 +152,13 @@ export class PlatformApiError extends Error { } } +export interface PlatformEventStream { + onerror: ((event: Event) => void) | null; + addEventListener(type: string, listener: (event: MessageEvent) => void): void; + removeEventListener(type: string, listener: (event: MessageEvent) => void): void; + close(): void; +} + export class PlatformApiClient { constructor(private readonly baseUrl = "/api/v1", private readonly sessionTokenProvider: () => string | null = () => platformApiSessionToken) {} @@ -638,8 +645,13 @@ export class PlatformApiClient { return this.request("/log-streams"); } - openServerLogEvents(id: string, options: LogStreamEventOptions = {}): EventSource { - return new EventSource(this.serverLogEventsUrl(id, options), { withCredentials: true }); + openServerLogEvents(id: string, options: LogStreamEventOptions = {}): PlatformEventStream { + const url = this.serverLogEventsUrl(id, options); + const sessionToken = this.sessionTokenProvider(); + if (!sessionToken) { + return new EventSource(url, { withCredentials: true }); + } + return new FetchServerSentEventStream(url, sessionToken); } serverLogEventsUrl(id: string, options: LogStreamEventOptions = {}): string { @@ -777,6 +789,89 @@ async function responseError(response: Response): Promise { return error; } +class FetchServerSentEventStream implements PlatformEventStream { + onerror: ((event: Event) => void) | null = null; + private readonly controller = new AbortController(); + private readonly listeners = new Map void>>(); + + constructor(private readonly url: string, private readonly sessionToken: string) { + void this.connect(); + } + + addEventListener(type: string, listener: (event: MessageEvent) => void): void { + const listeners = this.listeners.get(type) ?? new Set<(event: MessageEvent) => void>(); + listeners.add(listener); + this.listeners.set(type, listeners); + } + + removeEventListener(type: string, listener: (event: MessageEvent) => void): void { + this.listeners.get(type)?.delete(listener); + } + + close(): void { + this.controller.abort(); + this.listeners.clear(); + } + + private async connect(): Promise { + try { + const response = await fetch(this.url, { method: "GET", credentials: "include", headers: { Accept: "text/event-stream", Authorization: `Bearer ${this.sessionToken}` }, signal: this.controller.signal }); + if (!response.ok || !response.body) { + throw new Error(`event stream failed: ${response.status}`); + } + await this.read(response.body); + } catch { + if (!this.controller.signal.aborted) { + this.onerror?.(new Event("error")); + } + } + } + + private async read(body: ReadableStream): Promise { + const reader = body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + let eventName = "message"; + let eventId = ""; + let dataLines: string[] = []; + const processLine = (rawLine: string) => { + const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine; + if (line === "") { + if (dataLines.length > 0) { + this.dispatch(eventName, dataLines.join("\n"), eventId); + } + eventName = "message"; + eventId = ""; + dataLines = []; + return; + } + if (line.startsWith(":")) return; + const separator = line.indexOf(":"); + const field = separator === -1 ? line : line.slice(0, separator); + const value = separator === -1 ? "" : line.slice(separator + 1).replace(/^ /, ""); + if (field === "event") eventName = value || "message"; + if (field === "id") eventId = value; + if (field === "data") dataLines.push(value); + }; + while (!this.controller.signal.aborted) { + const { value, done } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split("\n"); + buffer = lines.pop() ?? ""; + lines.forEach(processLine); + } + } + + private dispatch(type: string, data: string, lastEventId: string): void { + const event = new MessageEvent(type, { data, lastEventId }); + this.listeners.get(type)?.forEach((listener) => listener(event)); + if (type !== "message") { + this.listeners.get("message")?.forEach((listener) => listener(event)); + } + } +} + function safeValidationMessage(apiError?: ApiErrorResponse | null): string { const details = apiError?.details ?.map((detail) => safeValidationDetail(detail))