mirror of
https://github.com/grp06/openclaw-studio.git
synced 2026-08-14 00:47:51 +00:00
fix: harden runtime stream replay-to-live handoff
This commit is contained in:
George Pickett
@@ -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<Uint8Array>({
|
||||
start(controller) {
|
||||
let closed = false;
|
||||
let unsubscribe: () => void = () => {};
|
||||
let heartbeat: ReturnType<typeof setInterval> | 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 });
|
||||
|
||||
@@ -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 <T>(modulePath: string, runtimeMock: RuntimeMock)
|
||||
return await import(modulePath) as T;
|
||||
};
|
||||
|
||||
const readStreamUntil = async (
|
||||
reader: ReadableStreamDefaultReader<Uint8Array>,
|
||||
predicate: (text: string) => boolean,
|
||||
maxReads: number = 20
|
||||
): Promise<string> => {
|
||||
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<Response>;
|
||||
}>("@/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<Response> }>(
|
||||
"@/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<Response> }>(
|
||||
"@/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<Response> }>(
|
||||
"@/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<Response> }>(
|
||||
"@/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");
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user