From cb6972ec551cd2cfdeec0795ece24dd45e60ab86 Mon Sep 17 00:00:00 2001 From: George Pickett Date: Sun, 1 Mar 2026 23:18:41 -0800 Subject: [PATCH] fix: harden runtime stream replay-to-live handoff --- src/app/api/runtime/stream/route.ts | 94 +++- tests/unit/runtimeRoutes.test.ts | 659 +++++++++++++++++++++++++++- 2 files changed, 717 insertions(+), 36 deletions(-) diff --git a/src/app/api/runtime/stream/route.ts b/src/app/api/runtime/stream/route.ts index f257e70..34fbee2 100644 --- a/src/app/api/runtime/stream/route.ts +++ b/src/app/api/runtime/stream/route.ts @@ -1,5 +1,5 @@ import type { ControlPlaneOutboxEntry } from "@/lib/controlplane/contracts"; -import { getControlPlaneRuntime, isStudioDomainApiModeEnabled } from "@/lib/controlplane/runtime"; +import { bootstrapDomainRuntime } from "@/lib/controlplane/runtime-route-bootstrap"; export const runtime = "nodejs"; @@ -26,34 +26,44 @@ const toSseFrame = (entry: ControlPlaneOutboxEntry): Uint8Array => { const heartbeatFrame = (): Uint8Array => encoder.encode(": heartbeat\n\n"); export async function GET(request: Request) { - if (!isStudioDomainApiModeEnabled()) { + const bootstrap = await bootstrapDomainRuntime(); + if (bootstrap.kind === "mode-disabled") { return new Response( JSON.stringify({ enabled: false, error: "domain_api_mode_disabled" }), { status: 404, headers: { "content-type": "application/json; charset=utf-8" } } ); } - - const controlPlane = getControlPlaneRuntime(); - try { - await controlPlane.ensureStarted(); - } catch (err) { + if (bootstrap.kind === "runtime-init-failed") { return new Response( JSON.stringify({ enabled: true, - error: err instanceof Error ? err.message : "controlplane_start_failed", + error: bootstrap.message, + code: "CONTROLPLANE_RUNTIME_INIT_FAILED", + reason: "runtime_init_failed", }), { status: 503, headers: { "content-type": "application/json; charset=utf-8" } } ); } - + if (bootstrap.kind === "start-failed") { + return new Response( + JSON.stringify({ + enabled: true, + error: bootstrap.message, + }), + { status: 503, headers: { "content-type": "application/json; charset=utf-8" } } + ); + } + const controlPlane = bootstrap.runtime; const lastSeenId = parseLastEventId(request); - const replayEntries = controlPlane.eventsAfter(lastSeenId, REPLAY_LIMIT); const stream = new ReadableStream({ start(controller) { let closed = false; let unsubscribe: () => void = () => {}; let heartbeat: ReturnType | null = null; + let startupPhase = true; + let lastDeliveredId = lastSeenId; + const startupLiveBuffer: ControlPlaneOutboxEntry[] = []; const close = () => { if (closed) return; closed = true; @@ -64,21 +74,67 @@ export async function GET(request: Request) { } try { controller.close(); - } catch {} + } catch (err) { + console.error("Failed to close runtime stream controller.", err); + } + }; + const enqueueFrame = (frame: Uint8Array): boolean => { + if (closed) return false; + try { + controller.enqueue(frame); + return true; + } catch (err) { + console.error("Failed to enqueue runtime stream frame.", err); + close(); + return false; + } + }; + const emitEntry = (entry: ControlPlaneOutboxEntry): boolean => { + if (entry.id <= lastDeliveredId) { + return true; + } + if (!enqueueFrame(toSseFrame(entry))) { + return false; + } + lastDeliveredId = entry.id; + return true; }; - for (const entry of replayEntries) { - controller.enqueue(toSseFrame(entry)); - } - unsubscribe = controlPlane.subscribe((entry) => { - if (closed) return; - controller.enqueue(toSseFrame(entry)); + if (closed) { + return; + } + if (startupPhase) { + startupLiveBuffer.push(entry); + return; + } + emitEntry(entry); }); + if (lastSeenId <= 0) { + const snapshot = controlPlane.snapshot(); + lastDeliveredId = Math.max(0, snapshot.outboxHead - REPLAY_LIMIT); + } + + const replayEntries = controlPlane.eventsAfter(lastDeliveredId, REPLAY_LIMIT); + for (const entry of replayEntries) { + if (!emitEntry(entry)) { + return; + } + } + + startupLiveBuffer.sort((left, right) => left.id - right.id); + for (const entry of startupLiveBuffer) { + if (!emitEntry(entry)) { + return; + } + } + startupLiveBuffer.length = 0; + + startupPhase = false; + heartbeat = setInterval(() => { - if (closed) return; - controller.enqueue(heartbeatFrame()); + enqueueFrame(heartbeatFrame()); }, HEARTBEAT_INTERVAL_MS); request.signal.addEventListener("abort", close, { once: true }); diff --git a/tests/unit/runtimeRoutes.test.ts b/tests/unit/runtimeRoutes.test.ts index caced4d..bd3bab3 100644 --- a/tests/unit/runtimeRoutes.test.ts +++ b/tests/unit/runtimeRoutes.test.ts @@ -10,6 +10,20 @@ type RuntimeMock = { event: unknown; createdAt: string; }>; + eventsBefore: (beforeOutboxId: number, limit?: number) => Array<{ + id: number; + event: unknown; + createdAt: string; + }>; + eventsBeforeForAgent: (agentId: string, beforeOutboxId: number, limit?: number) => Array<{ + id: number; + event: unknown; + createdAt: string; + }>; + backfillAgentHistoryIndex: ( + beforeOutboxId: number, + limit?: number + ) => { scannedRows: number; updatedRows: number; exhausted: boolean }; subscribe: (handler: (entry: { id: number; event: unknown; createdAt: string }) => void) => () => void; }; @@ -22,6 +36,22 @@ const loadRouteModule = async (modulePath: string, runtimeMock: RuntimeMock) return await import(modulePath) as T; }; +const readStreamUntil = async ( + reader: ReadableStreamDefaultReader, + predicate: (text: string) => boolean, + maxReads: number = 20 +): Promise => { + let output = ""; + const decoder = new TextDecoder(); + for (let index = 0; index < maxReads; index += 1) { + const chunk = await reader.read(); + if (chunk.done) break; + output += decoder.decode(chunk.value); + if (predicate(output)) break; + } + return output; +}; + describe("runtime routes", () => { afterEach(() => { vi.restoreAllMocks(); @@ -38,6 +68,9 @@ describe("runtime routes", () => { outboxHead: 12, }), eventsAfter: () => [], + eventsBefore: () => [], + eventsBeforeForAgent: () => [], + backfillAgentHistoryIndex: () => ({ scannedRows: 0, updatedRows: 0, exhausted: true }), subscribe: () => () => {}, }; @@ -74,6 +107,9 @@ describe("runtime routes", () => { outboxHead: 9, }), eventsAfter: () => [], + eventsBefore: () => [], + eventsBeforeForAgent: () => [], + backfillAgentHistoryIndex: () => ({ scannedRows: 0, updatedRows: 0, exhausted: true }), subscribe: () => () => {}, }), })); @@ -104,39 +140,221 @@ describe("runtime routes", () => { expect(body.freshness.reason).toBe("gateway_unavailable"); }); - it("agent history route filters by agent id", async () => { + it("summary route returns 503 when runtime initialization fails", async () => { + vi.resetModules(); + vi.doMock("@/lib/controlplane/runtime", () => ({ + isStudioDomainApiModeEnabled: () => true, + getControlPlaneRuntime: () => { + throw new Error("runtime init failed"); + }, + })); + + const mod = await import("@/app/api/runtime/summary/route"); + const response = await mod.GET(); + expect(response.status).toBe(503); + const body = await response.json() as { + enabled: boolean; + error: string; + code: string; + reason: string; + }; + expect(body.enabled).toBe(true); + expect(body.error).toBe("runtime init failed"); + expect(body.code).toBe("CONTROLPLANE_RUNTIME_INIT_FAILED"); + expect(body.reason).toBe("runtime_init_failed"); + }); + + it("summary route returns 404 when domain mode is disabled", async () => { + vi.resetModules(); + vi.doMock("@/lib/controlplane/runtime", () => ({ + isStudioDomainApiModeEnabled: () => false, + getControlPlaneRuntime: vi.fn(), + })); + + const mod = await import("@/app/api/runtime/summary/route"); + const response = await mod.GET(); + expect(response.status).toBe(404); + const body = await response.json() as { enabled: boolean; error: string }; + expect(body.enabled).toBe(false); + expect(body.error).toBe("domain_api_mode_disabled"); + }); + + it("agent history route returns newest window with beforeOutboxId cursor metadata", async () => { const runtimeMock: RuntimeMock = { ensureStarted: async () => {}, snapshot: () => ({ status: "connected", reason: null, asOf: "2026-02-28T02:40:00.000Z", - outboxHead: 3, + outboxHead: 5, }), - eventsAfter: () => [ + eventsAfter: () => [], + eventsBefore: () => { + throw new Error("legacy scan path used"); + }, + eventsBeforeForAgent: (_agentId: string, beforeOutboxId: number) => { + if (beforeOutboxId === 6) { + return [ + { + id: 1, + event: { + type: "gateway.event", + event: "runtime.delta", + seq: 10, + payload: { sessionKey: "agent:alpha:main", delta: "a" }, + asOf: "2026-02-28T02:40:01.000Z", + }, + createdAt: "2026-02-28T02:40:01.000Z", + }, + { + id: 3, + event: { + type: "gateway.event", + event: "runtime.delta", + seq: 12, + payload: { sessionKey: "agent:alpha:main", delta: "c" }, + asOf: "2026-02-28T02:40:03.000Z", + }, + createdAt: "2026-02-28T02:40:03.000Z", + }, + { + id: 5, + event: { + type: "gateway.event", + event: "runtime.delta", + seq: 14, + payload: { sessionKey: "agent:alpha:main", delta: "e" }, + asOf: "2026-02-28T02:40:05.000Z", + }, + createdAt: "2026-02-28T02:40:05.000Z", + }, + ]; + } + if (beforeOutboxId === 3) { + return [ + { + id: 1, + event: { + type: "gateway.event", + event: "runtime.delta", + seq: 10, + payload: { sessionKey: "agent:alpha:main", delta: "a" }, + asOf: "2026-02-28T02:40:01.000Z", + }, + createdAt: "2026-02-28T02:40:01.000Z", + }, + ]; + } + return []; + }, + backfillAgentHistoryIndex: () => ({ scannedRows: 0, updatedRows: 0, exhausted: true }), + subscribe: () => () => {}, + }; + + const mod = await loadRouteModule<{ + GET: ( + request: Request, + context: { params: Promise<{ agentId: string }> } + ) => Promise; + }>("@/app/api/runtime/agents/[agentId]/history/route", runtimeMock); + + const firstResponse = await mod.GET( + new Request("http://localhost/api/runtime/agents/alpha/history?limit=2"), + { params: Promise.resolve({ agentId: "alpha" }) } + ); + expect(firstResponse.status).toBe(200); + const firstBody = (await firstResponse.json()) as { + entries: Array<{ id: number }>; + hasMore: boolean; + nextBeforeOutboxId: number | null; + }; + expect(firstBody.entries.map((entry) => entry.id)).toEqual([3, 5]); + expect(firstBody.hasMore).toBe(true); + expect(firstBody.nextBeforeOutboxId).toBe(3); + + const secondResponse = await mod.GET( + new Request("http://localhost/api/runtime/agents/alpha/history?limit=2&beforeOutboxId=3"), + { params: Promise.resolve({ agentId: "alpha" }) } + ); + expect(secondResponse.status).toBe(200); + const secondBody = (await secondResponse.json()) as { + entries: Array<{ id: number }>; + hasMore: boolean; + nextBeforeOutboxId: number | null; + }; + expect(secondBody.entries.map((entry) => entry.id)).toEqual([1]); + expect(secondBody.hasMore).toBe(false); + expect(secondBody.nextBeforeOutboxId).toBeNull(); + }); + + it("agent history route backfills legacy rows in bounded batches", async () => { + const eventsBeforeForAgent = vi + .fn() + .mockReturnValueOnce([ { - id: 1, + id: 7, event: { type: "gateway.event", event: "runtime.delta", - seq: 10, - payload: { sessionKey: "agent:alpha:main", delta: "a" }, - asOf: "2026-02-28T02:40:01.000Z", + seq: 20, + payload: { sessionKey: "agent:alpha:main", delta: "newest" }, + asOf: "2026-02-28T02:40:07.000Z", }, - createdAt: "2026-02-28T02:40:01.000Z", + createdAt: "2026-02-28T02:40:07.000Z", }, + ]) + .mockReturnValueOnce([ { - id: 2, + id: 3, event: { type: "gateway.event", event: "runtime.delta", - seq: 11, - payload: { sessionKey: "agent:beta:main", delta: "b" }, - asOf: "2026-02-28T02:40:02.000Z", + seq: 18, + payload: { sessionKey: "agent:alpha:main", delta: "older" }, + asOf: "2026-02-28T02:40:03.000Z", }, - createdAt: "2026-02-28T02:40:02.000Z", + createdAt: "2026-02-28T02:40:03.000Z", }, - ], + { + id: 5, + event: { + type: "gateway.event", + event: "runtime.delta", + seq: 19, + payload: { sessionKey: "agent:alpha:main", delta: "middle" }, + asOf: "2026-02-28T02:40:05.000Z", + }, + createdAt: "2026-02-28T02:40:05.000Z", + }, + { + id: 7, + event: { + type: "gateway.event", + event: "runtime.delta", + seq: 20, + payload: { sessionKey: "agent:alpha:main", delta: "newest" }, + asOf: "2026-02-28T02:40:07.000Z", + }, + createdAt: "2026-02-28T02:40:07.000Z", + }, + ]); + const backfillAgentHistoryIndex = vi + .fn() + .mockReturnValue({ scannedRows: 500, updatedRows: 221, exhausted: false }); + const runtimeMock: RuntimeMock = { + ensureStarted: async () => {}, + snapshot: () => ({ + status: "connected", + reason: null, + asOf: "2026-02-28T02:40:00.000Z", + outboxHead: 10, + }), + eventsAfter: () => [], + eventsBefore: () => { + throw new Error("legacy scan path used"); + }, + eventsBeforeForAgent, + backfillAgentHistoryIndex, subscribe: () => () => {}, }; @@ -148,12 +366,21 @@ describe("runtime routes", () => { }>("@/app/api/runtime/agents/[agentId]/history/route", runtimeMock); const response = await mod.GET( - new Request("http://localhost/api/runtime/agents/alpha/history?limit=50"), + new Request("http://localhost/api/runtime/agents/alpha/history?limit=2"), { params: Promise.resolve({ agentId: "alpha" }) } ); expect(response.status).toBe(200); - const body = await response.json() as { entries: Array<{ id: number }> }; - expect(body.entries.map((entry) => entry.id)).toEqual([1]); + const body = (await response.json()) as { + entries: Array<{ id: number }>; + hasMore: boolean; + nextBeforeOutboxId: number | null; + }; + expect(eventsBeforeForAgent).toHaveBeenNthCalledWith(1, "alpha", 11, 3); + expect(backfillAgentHistoryIndex).toHaveBeenCalledWith(11, 500); + expect(eventsBeforeForAgent).toHaveBeenNthCalledWith(2, "alpha", 11, 3); + expect(body.entries.map((entry) => entry.id)).toEqual([5, 7]); + expect(body.hasMore).toBe(true); + expect(body.nextBeforeOutboxId).toBe(5); }); it("stream route replays from Last-Event-ID and emits live updates", async () => { @@ -182,6 +409,9 @@ describe("runtime routes", () => { }, ]; }, + eventsBefore: () => [], + eventsBeforeForAgent: () => [], + backfillAgentHistoryIndex: () => ({ scannedRows: 0, updatedRows: 0, exhausted: true }), subscribe: (handler) => { subscriber = handler; return () => { @@ -230,6 +460,346 @@ describe("runtime routes", () => { await reader.cancel(); }); + it("stream route does not drop rows committed between reconnect replay and live subscribe", async () => { + const subscriberRef: { + current: ((entry: { id: number; event: unknown; createdAt: string }) => void) | null; + } = { current: null }; + const replayBackedOutbox: Array<{ id: number; event: unknown; createdAt: string }> = [ + { + id: 3, + event: { + type: "gateway.event", + event: "runtime.delta", + seq: 20, + payload: { sessionKey: "agent:alpha:main", delta: "replay" }, + asOf: "2026-02-28T02:40:03.000Z", + }, + createdAt: "2026-02-28T02:40:03.000Z", + }, + ]; + const runtimeMock: RuntimeMock = { + ensureStarted: async () => {}, + snapshot: () => ({ + status: "connected", + reason: null, + asOf: "2026-02-28T02:40:00.000Z", + outboxHead: 4, + }), + eventsAfter: (lastSeenId: number) => { + expect(lastSeenId).toBe(2); + return replayBackedOutbox.filter((entry) => entry.id > lastSeenId); + }, + eventsBefore: () => [], + eventsBeforeForAgent: () => [], + backfillAgentHistoryIndex: () => ({ scannedRows: 0, updatedRows: 0, exhausted: true }), + subscribe: (handler) => { + replayBackedOutbox.push({ + id: 4, + event: { + type: "gateway.event", + event: "runtime.delta", + seq: 21, + payload: { sessionKey: "agent:alpha:main", delta: "committed-during-startup" }, + asOf: "2026-02-28T02:40:04.000Z", + }, + createdAt: "2026-02-28T02:40:04.000Z", + }); + subscriberRef.current = handler; + return () => { + subscriberRef.current = null; + }; + }, + }; + + const mod = await loadRouteModule<{ GET: (request: Request) => Promise }>( + "@/app/api/runtime/stream/route", + runtimeMock + ); + const response = await mod.GET( + new Request("http://localhost/api/runtime/stream", { + headers: { "Last-Event-ID": "2" }, + }) + ); + + expect(response.status).toBe(200); + expect(response.body).toBeTruthy(); + + const emit = subscriberRef.current; + if (!emit) { + throw new Error("expected runtime stream subscriber to be attached"); + } + emit({ + id: 5, + event: { + type: "runtime.status", + status: "connected", + reason: null, + asOf: "2026-02-28T02:40:05.000Z", + }, + createdAt: "2026-02-28T02:40:05.000Z", + }); + + const reader = response.body!.getReader(); + const output = await readStreamUntil(reader, (text) => text.includes("id: 5")); + expect(output).toContain("id: 3"); + expect(output).toContain("id: 4"); + expect(output).toContain("id: 5"); + expect(output.indexOf("id: 3")).toBeLessThan(output.indexOf("id: 4")); + expect(output.indexOf("id: 4")).toBeLessThan(output.indexOf("id: 5")); + await reader.cancel(); + }); + + it("stream route deduplicates rows that arrive in both replay and live startup buffer", async () => { + const subscriberRef: { + current: ((entry: { id: number; event: unknown; createdAt: string }) => void) | null; + } = { current: null }; + const overlapEntry = { + id: 4, + event: { + type: "gateway.event", + event: "runtime.delta", + seq: 21, + payload: { sessionKey: "agent:alpha:main", delta: "overlap" }, + asOf: "2026-02-28T02:40:04.000Z", + }, + createdAt: "2026-02-28T02:40:04.000Z", + }; + const runtimeMock: RuntimeMock = { + ensureStarted: async () => {}, + snapshot: () => ({ + status: "connected", + reason: null, + asOf: "2026-02-28T02:40:00.000Z", + outboxHead: 4, + }), + eventsAfter: (lastSeenId: number) => { + expect(lastSeenId).toBe(2); + return [ + { + id: 3, + event: { + type: "gateway.event", + event: "runtime.delta", + seq: 20, + payload: { sessionKey: "agent:alpha:main", delta: "replay" }, + asOf: "2026-02-28T02:40:03.000Z", + }, + createdAt: "2026-02-28T02:40:03.000Z", + }, + overlapEntry, + ]; + }, + eventsBefore: () => [], + eventsBeforeForAgent: () => [], + backfillAgentHistoryIndex: () => ({ scannedRows: 0, updatedRows: 0, exhausted: true }), + subscribe: (handler) => { + subscriberRef.current = handler; + handler(overlapEntry); + return () => { + subscriberRef.current = null; + }; + }, + }; + + const mod = await loadRouteModule<{ GET: (request: Request) => Promise }>( + "@/app/api/runtime/stream/route", + runtimeMock + ); + const response = await mod.GET( + new Request("http://localhost/api/runtime/stream", { + headers: { "Last-Event-ID": "2" }, + }) + ); + + expect(response.status).toBe(200); + expect(response.body).toBeTruthy(); + + const emit = subscriberRef.current; + if (!emit) { + throw new Error("expected runtime stream subscriber to be attached"); + } + emit({ + id: 5, + event: { + type: "runtime.status", + status: "connected", + reason: null, + asOf: "2026-02-28T02:40:05.000Z", + }, + createdAt: "2026-02-28T02:40:05.000Z", + }); + + const reader = response.body!.getReader(); + const output = await readStreamUntil(reader, (text) => text.includes("id: 5")); + const duplicateCount = (output.match(/(^|\n)id: 4\n/gm) ?? []).length; + expect(output).toContain("id: 3"); + expect(output).toContain("id: 4"); + expect(output).toContain("id: 5"); + expect(duplicateCount).toBe(1); + await reader.cancel(); + }); + + it("stream route does not lose newest head rows during fresh-connect capped replay startup", async () => { + const subscriberRef: { + current: ((entry: { id: number; event: unknown; createdAt: string }) => void) | null; + } = { current: null }; + let outboxHead = 5_000; + const outboxRows: Array<{ id: number; event: unknown; createdAt: string }> = [ + { + id: 4_999, + event: { + type: "gateway.event", + event: "runtime.delta", + seq: 19, + payload: { sessionKey: "agent:alpha:main", delta: "n-1" }, + asOf: "2026-02-28T02:40:03.000Z", + }, + createdAt: "2026-02-28T02:40:03.000Z", + }, + { + id: 5_000, + event: { + type: "gateway.event", + event: "runtime.delta", + seq: 20, + payload: { sessionKey: "agent:alpha:main", delta: "n" }, + asOf: "2026-02-28T02:40:04.000Z", + }, + createdAt: "2026-02-28T02:40:04.000Z", + }, + ]; + const runtimeMock: RuntimeMock = { + ensureStarted: async () => {}, + snapshot: () => ({ + status: "connected", + reason: null, + asOf: "2026-02-28T02:40:00.000Z", + outboxHead, + }), + eventsAfter: (lastSeenId: number, limit?: number) => { + expect(limit).toBe(2_000); + return outboxRows.filter((entry) => entry.id > lastSeenId).slice(0, limit); + }, + eventsBefore: () => [], + eventsBeforeForAgent: () => [], + backfillAgentHistoryIndex: () => ({ scannedRows: 0, updatedRows: 0, exhausted: true }), + subscribe: (handler) => { + outboxHead = 5_001; + outboxRows.push({ + id: 5_001, + event: { + type: "gateway.event", + event: "runtime.delta", + seq: 21, + payload: { sessionKey: "agent:alpha:main", delta: "n+1 boundary" }, + asOf: "2026-02-28T02:40:05.000Z", + }, + createdAt: "2026-02-28T02:40:05.000Z", + }); + subscriberRef.current = handler; + return () => { + subscriberRef.current = null; + }; + }, + }; + + const mod = await loadRouteModule<{ GET: (request: Request) => Promise }>( + "@/app/api/runtime/stream/route", + runtimeMock + ); + const response = await mod.GET(new Request("http://localhost/api/runtime/stream")); + expect(response.status).toBe(200); + expect(response.body).toBeTruthy(); + + const emit = subscriberRef.current; + if (!emit) { + throw new Error("expected runtime stream subscriber to be attached"); + } + emit({ + id: 5_002, + event: { + type: "runtime.status", + status: "connected", + reason: null, + asOf: "2026-02-28T02:40:06.000Z", + }, + createdAt: "2026-02-28T02:40:06.000Z", + }); + + const reader = response.body!.getReader(); + const output = await readStreamUntil(reader, (text) => text.includes("id: 5002")); + expect(output).toContain("id: 4999"); + expect(output).toContain("id: 5000"); + expect(output).toContain("id: 5001"); + expect(output).toContain("id: 5002"); + expect(output.indexOf("id: 5000")).toBeLessThan(output.indexOf("id: 5001")); + expect(output.indexOf("id: 5001")).toBeLessThan(output.indexOf("id: 5002")); + await reader.cancel(); + }); + + it("stream route replays newest window when Last-Event-ID is absent", async () => { + const runtimeMock: RuntimeMock = { + ensureStarted: async () => {}, + snapshot: () => ({ + status: "connected", + reason: null, + asOf: "2026-02-28T02:40:00.000Z", + outboxHead: 5_000, + }), + eventsAfter: (lastSeenId: number) => { + expect(lastSeenId).toBe(3_000); + return [ + { + id: 4_999, + event: { + type: "gateway.event", + event: "runtime.delta", + seq: 20, + payload: { sessionKey: "agent:alpha:main", delta: "replay" }, + asOf: "2026-02-28T02:40:03.000Z", + }, + createdAt: "2026-02-28T02:40:03.000Z", + }, + ]; + }, + eventsBefore: () => [], + eventsBeforeForAgent: () => [], + backfillAgentHistoryIndex: () => ({ scannedRows: 0, updatedRows: 0, exhausted: true }), + subscribe: () => () => {}, + }; + + const mod = await loadRouteModule<{ GET: (request: Request) => Promise }>( + "@/app/api/runtime/stream/route", + runtimeMock + ); + const response = await mod.GET(new Request("http://localhost/api/runtime/stream")); + expect(response.status).toBe(200); + const reader = response.body!.getReader(); + const first = await reader.read(); + const firstChunk = new TextDecoder().decode(first.value); + expect(firstChunk).toContain("id: 4999"); + await reader.cancel(); + }); + + it("stream route returns 503 when runtime cannot start", async () => { + vi.resetModules(); + vi.doMock("@/lib/controlplane/runtime", () => ({ + isStudioDomainApiModeEnabled: () => true, + getControlPlaneRuntime: () => ({ + ensureStarted: async () => { + throw new Error("gateway unavailable"); + }, + }), + })); + + const mod = await import("@/app/api/runtime/stream/route"); + const response = await mod.GET(new Request("http://localhost/api/runtime/stream")); + expect(response.status).toBe(503); + const body = await response.json() as { enabled: boolean; error: string }; + expect(body.enabled).toBe(true); + expect(body.error).toBe("gateway unavailable"); + }); + it("agent-rename and agent-delete intent routes forward to runtime", async () => { const callGateway = vi.fn(async () => ({ ok: true })); vi.doMock("@/lib/controlplane/runtime", () => ({ @@ -270,6 +840,30 @@ describe("runtime routes", () => { }); }); + it("intent routes return 503 when runtime initialization fails", async () => { + vi.resetModules(); + vi.doMock("@/lib/controlplane/runtime", () => ({ + isStudioDomainApiModeEnabled: () => true, + getControlPlaneRuntime: () => { + throw new Error("runtime init failed"); + }, + })); + + const renameRoute = await import("@/app/api/intents/agent-rename/route"); + const renameRes = await renameRoute.POST( + new Request("http://localhost/api/intents/agent-rename", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ agentId: "agent-1", name: "Agent One Renamed" }), + }) + ); + expect(renameRes.status).toBe(503); + const body = await renameRes.json() as { error: string; code: string; reason: string }; + expect(body.error).toBe("runtime init failed"); + expect(body.code).toBe("CONTROLPLANE_RUNTIME_INIT_FAILED"); + expect(body.reason).toBe("runtime_init_failed"); + }); + it("runtime fleet route hydrates through control-plane runtime", async () => { const callGateway = vi.fn(async () => ({ ok: true })); vi.doMock("@/lib/controlplane/runtime", () => ({ @@ -310,4 +904,35 @@ describe("runtime routes", () => { const body = await response.json() as { result: { seeds: Array<{ agentId: string }> } }; expect(body.result.seeds[0]?.agentId).toBe("agent-1"); }); + + it("runtime fleet route returns 503 gateway_unavailable when runtime cannot start", async () => { + vi.resetModules(); + vi.doMock("@/lib/controlplane/runtime", () => ({ + isStudioDomainApiModeEnabled: () => true, + getControlPlaneRuntime: () => ({ + ensureStarted: async () => { + throw new Error("gateway unavailable"); + }, + }), + })); + const route = await import("@/app/api/runtime/fleet/route"); + const response = await route.POST( + new Request("http://localhost/api/runtime/fleet", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ cachedConfigSnapshot: null }), + }) + ); + expect(response.status).toBe(503); + const body = await response.json() as { + enabled: boolean; + error: string; + code: string; + reason: string; + }; + expect(body.enabled).toBe(true); + expect(body.error).toBe("gateway unavailable"); + expect(body.code).toBe("GATEWAY_UNAVAILABLE"); + expect(body.reason).toBe("gateway_unavailable"); + }); });