plugin: welcome-timeout + stall watchdog (bulletproof review: fixes handshake hang + wedged-socket unbounded buffering); smoke 27 checks, e2e 13/13
This commit is contained in:
Raphael Westphal
+484
-266
@@ -17,312 +17,530 @@
|
||||
* 6. session_shutdown -> socket closed, no reconnect afterwards
|
||||
*/
|
||||
|
||||
import { check as rawCheck, sleep, waitFor, startMiniDaemon } from "./mini-daemon.ts";
|
||||
import {
|
||||
check as rawCheck,
|
||||
sleep,
|
||||
waitFor,
|
||||
startMiniDaemon,
|
||||
} from "./mini-daemon.ts";
|
||||
|
||||
const SESSION_ID = "sess-1";
|
||||
|
||||
let failures = 0;
|
||||
|
||||
function check(name: string, ok: boolean, detail = ""): void {
|
||||
rawCheck(name, ok, detail);
|
||||
if (!ok) failures++;
|
||||
rawCheck(name, ok, detail);
|
||||
if (!ok) failures++;
|
||||
}
|
||||
|
||||
interface FakePi {
|
||||
handlers: Map<string, (event: unknown, ctx: unknown) => unknown>;
|
||||
sentMessages: Array<{ message: string; options: unknown }>;
|
||||
aborted: number;
|
||||
on(event: string, handler: (event: unknown, ctx: unknown) => unknown): void;
|
||||
sendUserMessage(message: string, options?: unknown): void;
|
||||
abort(): void;
|
||||
handlers: Map<string, (event: unknown, ctx: unknown) => unknown>;
|
||||
sentMessages: Array<{ message: string; options: unknown }>;
|
||||
aborted: number;
|
||||
on(event: string, handler: (event: unknown, ctx: unknown) => unknown): void;
|
||||
sendUserMessage(message: string, options?: unknown): void;
|
||||
abort(): void;
|
||||
}
|
||||
|
||||
function makeFakePi(): FakePi {
|
||||
return {
|
||||
handlers: new Map(),
|
||||
sentMessages: [],
|
||||
aborted: 0,
|
||||
on(event, handler) {
|
||||
this.handlers.set(event, handler);
|
||||
},
|
||||
sendUserMessage(message, options) {
|
||||
this.sentMessages.push({ message, options });
|
||||
},
|
||||
abort() {
|
||||
this.aborted++;
|
||||
},
|
||||
};
|
||||
return {
|
||||
handlers: new Map(),
|
||||
sentMessages: [],
|
||||
aborted: 0,
|
||||
on(event, handler) {
|
||||
this.handlers.set(event, handler);
|
||||
},
|
||||
sendUserMessage(message, options) {
|
||||
this.sentMessages.push({ message, options });
|
||||
},
|
||||
abort() {
|
||||
this.aborted++;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function makeFakeCtx(): unknown {
|
||||
return {
|
||||
cwd: "/work/repo",
|
||||
model: { id: "glm-5.3", provider: "zai-renaud" },
|
||||
sessionManager: {
|
||||
getSessionId: () => SESSION_ID,
|
||||
getSessionName: () => undefined,
|
||||
getCwd: () => "/work/repo",
|
||||
getHeader: () => ({ timestamp: "2024-12-03T14:00:00.000Z", id: SESSION_ID }),
|
||||
},
|
||||
};
|
||||
return {
|
||||
cwd: "/work/repo",
|
||||
model: { id: "glm-5.3", provider: "zai-renaud" },
|
||||
sessionManager: {
|
||||
getSessionId: () => SESSION_ID,
|
||||
getSessionName: () => undefined,
|
||||
getCwd: () => "/work/repo",
|
||||
getHeader: () => ({
|
||||
timestamp: "2024-12-03T14:00:00.000Z",
|
||||
id: SESSION_ID,
|
||||
}),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function loadExtension(): Promise<(pi: unknown) => void> {
|
||||
const mod = (await import("./lvmh-agent.ts")) as { default: (pi: unknown) => void };
|
||||
return mod.default;
|
||||
const mod = (await import("./lvmh-agent.ts")) as {
|
||||
default: (pi: unknown) => void;
|
||||
};
|
||||
return mod.default;
|
||||
}
|
||||
|
||||
async function main(): Promise<void> {
|
||||
// --- Scenario 1: env unset -> inert --------------------------------
|
||||
delete process.env.LVMH_URL;
|
||||
delete process.env.LVMH_TOKEN;
|
||||
delete process.env.LVMH_AGENT;
|
||||
delete process.env.LVMH_REPO;
|
||||
const factory = await loadExtension();
|
||||
const inertPi = makeFakePi();
|
||||
inertPi.on = ((event: string) => {
|
||||
throw new Error(`inert extension registered handler ${event}`);
|
||||
}) as FakePi["on"];
|
||||
factory(inertPi); // must not touch pi at all
|
||||
check("1 inert: no side effects on pi", true);
|
||||
// --- Scenario 1: env unset -> inert --------------------------------
|
||||
delete process.env.LVMH_URL;
|
||||
delete process.env.LVMH_TOKEN;
|
||||
delete process.env.LVMH_AGENT;
|
||||
delete process.env.LVMH_REPO;
|
||||
const factory = await loadExtension();
|
||||
const inertPi = makeFakePi();
|
||||
inertPi.on = ((event: string) => {
|
||||
throw new Error(`inert extension registered handler ${event}`);
|
||||
}) as FakePi["on"];
|
||||
factory(inertPi); // must not touch pi at all
|
||||
check("1 inert: no side effects on pi", true);
|
||||
|
||||
// --- Scenario 2: live daemon ---------------------------------------
|
||||
let welcomeLastSeq = 0;
|
||||
const daemon = await startMiniDaemon(() => welcomeLastSeq);
|
||||
process.env.LVMH_URL = daemon.url;
|
||||
process.env.LVMH_TOKEN = "smoke-token";
|
||||
// --- Scenario 2: live daemon ---------------------------------------
|
||||
let welcomeLastSeq = 0;
|
||||
const daemon = await startMiniDaemon(() => welcomeLastSeq);
|
||||
process.env.LVMH_URL = daemon.url;
|
||||
process.env.LVMH_TOKEN = "smoke-token";
|
||||
|
||||
const pi = makeFakePi();
|
||||
factory(pi);
|
||||
const handlerNames = [
|
||||
"session_start",
|
||||
"session_shutdown",
|
||||
"session_info_changed",
|
||||
"model_select",
|
||||
"message_start",
|
||||
"message_update",
|
||||
"message_end",
|
||||
"tool_execution_start",
|
||||
"tool_execution_update",
|
||||
"tool_execution_end",
|
||||
"agent_start",
|
||||
"agent_end",
|
||||
"agent_settled",
|
||||
];
|
||||
check("2 all 13 handlers registered", handlerNames.every((n) => pi.handlers.has(n)));
|
||||
const pi = makeFakePi();
|
||||
factory(pi);
|
||||
const handlerNames = [
|
||||
"session_start",
|
||||
"session_shutdown",
|
||||
"session_info_changed",
|
||||
"model_select",
|
||||
"message_start",
|
||||
"message_update",
|
||||
"message_end",
|
||||
"tool_execution_start",
|
||||
"tool_execution_update",
|
||||
"tool_execution_end",
|
||||
"agent_start",
|
||||
"agent_end",
|
||||
"agent_settled",
|
||||
];
|
||||
check(
|
||||
"2 all 13 handlers registered",
|
||||
handlerNames.every((n) => pi.handlers.has(n)),
|
||||
);
|
||||
|
||||
let lastReturn: unknown = "sentinel";
|
||||
const fire = (name: string, event: unknown): void => {
|
||||
const h = pi.handlers.get(name);
|
||||
if (h === undefined) throw new Error(`missing handler ${name}`);
|
||||
lastReturn = h(event, makeFakeCtx());
|
||||
if (lastReturn instanceof Promise) lastReturn.catch(() => undefined);
|
||||
};
|
||||
let lastReturn: unknown = "sentinel";
|
||||
const fire = (name: string, event: unknown): void => {
|
||||
const h = pi.handlers.get(name);
|
||||
if (h === undefined) throw new Error(`missing handler ${name}`);
|
||||
lastReturn = h(event, makeFakeCtx());
|
||||
if (lastReturn instanceof Promise) lastReturn.catch(() => undefined);
|
||||
};
|
||||
|
||||
fire("session_start", { reason: "startup" });
|
||||
const helloSeen = await waitFor(() => daemon.frames.some((f) => f.type === "hello"), 5000);
|
||||
check("2 connects and sends hello", helloSeen);
|
||||
check("2 bearer auth on upgrade", daemon.authHeaders.at(-1) === "Bearer smoke-token", daemon.authHeaders.at(-1));
|
||||
fire("session_start", { reason: "startup" });
|
||||
const helloSeen = await waitFor(
|
||||
() => daemon.frames.some((f) => f.type === "hello"),
|
||||
5000,
|
||||
);
|
||||
check("2 connects and sends hello", helloSeen);
|
||||
check(
|
||||
"2 bearer auth on upgrade",
|
||||
daemon.authHeaders.at(-1) === "Bearer smoke-token",
|
||||
daemon.authHeaders.at(-1),
|
||||
);
|
||||
|
||||
const hello = daemon.frames.find((f) => f.type === "hello");
|
||||
const hs = (hello?.session ?? {}) as Record<string, unknown>;
|
||||
check(
|
||||
"2 hello snapshot",
|
||||
hs.id === SESSION_ID &&
|
||||
hs.name === null &&
|
||||
hs.cwd === "/work/repo" &&
|
||||
hs.model === "glm-5.3" &&
|
||||
hs.provider === "zai-renaud" &&
|
||||
hs.agent === false &&
|
||||
hs.repo === null &&
|
||||
typeof hs.startedAt === "number",
|
||||
JSON.stringify(hs),
|
||||
);
|
||||
const hello = daemon.frames.find((f) => f.type === "hello");
|
||||
const hs = (hello?.session ?? {}) as Record<string, unknown>;
|
||||
check(
|
||||
"2 hello snapshot",
|
||||
hs.id === SESSION_ID &&
|
||||
hs.name === null &&
|
||||
hs.cwd === "/work/repo" &&
|
||||
hs.model === "glm-5.3" &&
|
||||
hs.provider === "zai-renaud" &&
|
||||
hs.agent === false &&
|
||||
hs.repo === null &&
|
||||
typeof hs.startedAt === "number",
|
||||
JSON.stringify(hs),
|
||||
);
|
||||
|
||||
fire("message_start", { message: { role: "assistant", content: [], timestamp: 1000 } });
|
||||
fire("message_update", { message: { role: "assistant", content: [{ type: "text", text: "Hel" }], timestamp: 1000 } });
|
||||
fire("message_update", { message: { role: "assistant", content: [{ type: "text", text: "Hello world" }], timestamp: 1000 } });
|
||||
fire("message_end", {
|
||||
message: {
|
||||
role: "assistant",
|
||||
content: [
|
||||
{ type: "thinking", thinking: "hmm" },
|
||||
{ type: "text", text: "Hello world" },
|
||||
{ type: "toolCall", id: "tc1", name: "bash", arguments: { command: "ls" } },
|
||||
],
|
||||
timestamp: 1000,
|
||||
},
|
||||
});
|
||||
await waitFor(() => daemon.frames.some((f) => f.type === "message_end"), 2000);
|
||||
fire("message_start", {
|
||||
message: { role: "assistant", content: [], timestamp: 1000 },
|
||||
});
|
||||
fire("message_update", {
|
||||
message: {
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text: "Hel" }],
|
||||
timestamp: 1000,
|
||||
},
|
||||
});
|
||||
fire("message_update", {
|
||||
message: {
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text: "Hello world" }],
|
||||
timestamp: 1000,
|
||||
},
|
||||
});
|
||||
fire("message_end", {
|
||||
message: {
|
||||
role: "assistant",
|
||||
content: [
|
||||
{ type: "thinking", thinking: "hmm" },
|
||||
{ type: "text", text: "Hello world" },
|
||||
{
|
||||
type: "toolCall",
|
||||
id: "tc1",
|
||||
name: "bash",
|
||||
arguments: { command: "ls" },
|
||||
},
|
||||
],
|
||||
timestamp: 1000,
|
||||
},
|
||||
});
|
||||
await waitFor(
|
||||
() => daemon.frames.some((f) => f.type === "message_end"),
|
||||
2000,
|
||||
);
|
||||
|
||||
const n0 = daemon.frames.findIndex((f) => f.type === "hello");
|
||||
const mirrored = daemon.frames.slice(n0 + 1).map((f) => f.type);
|
||||
check(
|
||||
"2 mirror order",
|
||||
JSON.stringify(mirrored) === JSON.stringify(["message_start", "message_update", "message_update", "message_end"]),
|
||||
JSON.stringify(mirrored),
|
||||
);
|
||||
const deltas = daemon.frames.filter((f) => f.type === "message_update").map((f) => f.delta);
|
||||
check("2 delta-only updates", JSON.stringify(deltas) === JSON.stringify(["Hel", "lo world"]), JSON.stringify(deltas));
|
||||
const n0 = daemon.frames.findIndex((f) => f.type === "hello");
|
||||
const mirrored = daemon.frames.slice(n0 + 1).map((f) => f.type);
|
||||
check(
|
||||
"2 mirror order",
|
||||
JSON.stringify(mirrored) ===
|
||||
JSON.stringify([
|
||||
"message_start",
|
||||
"message_update",
|
||||
"message_update",
|
||||
"message_end",
|
||||
]),
|
||||
JSON.stringify(mirrored),
|
||||
);
|
||||
const deltas = daemon.frames
|
||||
.filter((f) => f.type === "message_update")
|
||||
.map((f) => f.delta);
|
||||
check(
|
||||
"2 delta-only updates",
|
||||
JSON.stringify(deltas) === JSON.stringify(["Hel", "lo world"]),
|
||||
JSON.stringify(deltas),
|
||||
);
|
||||
|
||||
const msg = daemon.frames.find((f) => f.type === "message_end")?.message as Record<string, unknown>;
|
||||
const tc = (msg?.toolCalls as Array<Record<string, unknown>> | undefined)?.[0];
|
||||
check(
|
||||
"2 message_end mapping",
|
||||
msg?.role === "assistant" &&
|
||||
msg?.text === "Hello world" &&
|
||||
msg?.thinking === "hmm" &&
|
||||
typeof msg?.id === "string" &&
|
||||
tc?.id === "tc1" &&
|
||||
tc?.name === "bash" &&
|
||||
tc?.argsJson === '{"command":"ls"}' &&
|
||||
msg?.toolCallId === null,
|
||||
JSON.stringify(msg),
|
||||
);
|
||||
const msg = daemon.frames.find((f) => f.type === "message_end")
|
||||
?.message as Record<string, unknown>;
|
||||
const tc = (
|
||||
msg?.toolCalls as Array<Record<string, unknown>> | undefined
|
||||
)?.[0];
|
||||
check(
|
||||
"2 message_end mapping",
|
||||
msg?.role === "assistant" &&
|
||||
msg?.text === "Hello world" &&
|
||||
msg?.thinking === "hmm" &&
|
||||
typeof msg?.id === "string" &&
|
||||
tc?.id === "tc1" &&
|
||||
tc?.name === "bash" &&
|
||||
tc?.argsJson === '{"command":"ls"}' &&
|
||||
msg?.toolCallId === null,
|
||||
JSON.stringify(msg),
|
||||
);
|
||||
|
||||
const big = "x".repeat(3000);
|
||||
fire("agent_start", {});
|
||||
fire("tool_execution_start", { toolCallId: "tc1", toolName: "bash", args: { command: "ls" } });
|
||||
fire("tool_execution_update", { toolCallId: "tc1", toolName: "bash", partialResult: { content: [{ type: "text", text: big }] } });
|
||||
fire("tool_execution_end", { toolCallId: "tc1", toolName: "bash", isError: true, result: { content: [{ type: "text", text: big }] } });
|
||||
fire("agent_end", { messages: [{ role: "assistant", usage: { input: 10, output: 5, cost: { total: 0.25 } } }] });
|
||||
fire("agent_settled", {});
|
||||
await waitFor(() => daemon.frames.some((f) => f.type === "agent_settled"), 2000);
|
||||
const big = "x".repeat(3000);
|
||||
fire("agent_start", {});
|
||||
fire("tool_execution_start", {
|
||||
toolCallId: "tc1",
|
||||
toolName: "bash",
|
||||
args: { command: "ls" },
|
||||
});
|
||||
fire("tool_execution_update", {
|
||||
toolCallId: "tc1",
|
||||
toolName: "bash",
|
||||
partialResult: { content: [{ type: "text", text: big }] },
|
||||
});
|
||||
fire("tool_execution_end", {
|
||||
toolCallId: "tc1",
|
||||
toolName: "bash",
|
||||
isError: true,
|
||||
result: { content: [{ type: "text", text: big }] },
|
||||
});
|
||||
fire("agent_end", {
|
||||
messages: [
|
||||
{
|
||||
role: "assistant",
|
||||
usage: { input: 10, output: 5, cost: { total: 0.25 } },
|
||||
},
|
||||
],
|
||||
});
|
||||
fire("agent_settled", {});
|
||||
await waitFor(
|
||||
() => daemon.frames.some((f) => f.type === "agent_settled"),
|
||||
2000,
|
||||
);
|
||||
|
||||
const partial = daemon.frames.find((f) => f.type === "tool_execution_update");
|
||||
const end = daemon.frames.find((f) => f.type === "tool_execution_end");
|
||||
const partialLen = (partial?.partial as string | undefined)?.length;
|
||||
const previewLen = (end?.resultPreview as string | undefined)?.length;
|
||||
check("2 partial truncated to 2000", partialLen === 2000, String(partialLen));
|
||||
check("2 resultPreview truncated + isError", previewLen === 2000 && end?.isError === true);
|
||||
const usage = daemon.frames.find((f) => f.type === "agent_end")?.usage as Record<string, number>;
|
||||
check(
|
||||
"2 agent_end usage",
|
||||
usage?.inputTokens === 10 && usage?.outputTokens === 5 && usage?.totalCost === 0.25,
|
||||
JSON.stringify(usage),
|
||||
);
|
||||
const partial = daemon.frames.find((f) => f.type === "tool_execution_update");
|
||||
const end = daemon.frames.find((f) => f.type === "tool_execution_end");
|
||||
const partialLen = (partial?.partial as string | undefined)?.length;
|
||||
const previewLen = (end?.resultPreview as string | undefined)?.length;
|
||||
check("2 partial truncated to 2000", partialLen === 2000, String(partialLen));
|
||||
check(
|
||||
"2 resultPreview truncated + isError",
|
||||
previewLen === 2000 && end?.isError === true,
|
||||
);
|
||||
const usage = daemon.frames.find((f) => f.type === "agent_end")
|
||||
?.usage as Record<string, number>;
|
||||
check(
|
||||
"2 agent_end usage",
|
||||
usage?.inputTokens === 10 &&
|
||||
usage?.outputTokens === 5 &&
|
||||
usage?.totalCost === 0.25,
|
||||
JSON.stringify(usage),
|
||||
);
|
||||
|
||||
const seqs = daemon.frames.filter((f) => f.seq > 0).map((f) => f.seq);
|
||||
check(
|
||||
"2 seq strictly monotonic",
|
||||
seqs.length > 0 && seqs.every((s, i) => i === 0 || s > seqs[i - 1]),
|
||||
JSON.stringify(seqs),
|
||||
);
|
||||
check("2 handler return values are undefined (never a promise)", lastReturn === undefined);
|
||||
const seqs = daemon.frames.filter((f) => f.seq > 0).map((f) => f.seq);
|
||||
check(
|
||||
"2 seq strictly monotonic",
|
||||
seqs.length > 0 && seqs.every((s, i) => i === 0 || s > seqs[i - 1]),
|
||||
JSON.stringify(seqs),
|
||||
);
|
||||
check(
|
||||
"2 handler return values are undefined (never a promise)",
|
||||
lastReturn === undefined,
|
||||
);
|
||||
|
||||
// --- Scenario 3: prompt / abort ------------------------------------
|
||||
daemon.pushAll(
|
||||
JSON.stringify({ v: 1, type: "prompt", sessionId: "other-session", seq: 0, ts: Date.now(), promptId: "p0", message: "wrong session" }),
|
||||
);
|
||||
daemon.pushAll(
|
||||
JSON.stringify({ v: 1, type: "prompt", sessionId: SESSION_ID, seq: 0, ts: Date.now(), promptId: "p1", message: "run the tests" }),
|
||||
);
|
||||
await waitFor(() => pi.sentMessages.length > 0, 2000);
|
||||
check(
|
||||
"3 prompt delivered as steer, foreign session ignored",
|
||||
pi.sentMessages.length === 1 &&
|
||||
pi.sentMessages[0].message === "run the tests" &&
|
||||
JSON.stringify(pi.sentMessages[0].options) === '{"deliverAs":"steer"}',
|
||||
JSON.stringify(pi.sentMessages),
|
||||
);
|
||||
daemon.pushAll(JSON.stringify({ v: 1, type: "abort", sessionId: SESSION_ID, seq: 0, ts: Date.now() }));
|
||||
await waitFor(() => pi.aborted > 0, 2000);
|
||||
check("3 abort calls pi.abort", pi.aborted === 1);
|
||||
// --- Scenario 3: prompt / abort ------------------------------------
|
||||
daemon.pushAll(
|
||||
JSON.stringify({
|
||||
v: 1,
|
||||
type: "prompt",
|
||||
sessionId: "other-session",
|
||||
seq: 0,
|
||||
ts: Date.now(),
|
||||
promptId: "p0",
|
||||
message: "wrong session",
|
||||
}),
|
||||
);
|
||||
daemon.pushAll(
|
||||
JSON.stringify({
|
||||
v: 1,
|
||||
type: "prompt",
|
||||
sessionId: SESSION_ID,
|
||||
seq: 0,
|
||||
ts: Date.now(),
|
||||
promptId: "p1",
|
||||
message: "run the tests",
|
||||
}),
|
||||
);
|
||||
await waitFor(() => pi.sentMessages.length > 0, 2000);
|
||||
check(
|
||||
"3 prompt delivered as steer, foreign session ignored",
|
||||
pi.sentMessages.length === 1 &&
|
||||
pi.sentMessages[0].message === "run the tests" &&
|
||||
JSON.stringify(pi.sentMessages[0].options) === '{"deliverAs":"steer"}',
|
||||
JSON.stringify(pi.sentMessages),
|
||||
);
|
||||
daemon.pushAll(
|
||||
JSON.stringify({
|
||||
v: 1,
|
||||
type: "abort",
|
||||
sessionId: SESSION_ID,
|
||||
seq: 0,
|
||||
ts: Date.now(),
|
||||
}),
|
||||
);
|
||||
await waitFor(() => pi.aborted > 0, 2000);
|
||||
check("3 abort calls pi.abort", pi.aborted === 1);
|
||||
|
||||
// --- Scenario 4: drop + reconnect, replay, overflow ----------------
|
||||
const seqBeforeDrop = Math.max(...daemon.frames.map((f) => f.seq));
|
||||
welcomeLastSeq = seqBeforeDrop; // daemon has everything up to the drop
|
||||
daemon.dropConnections();
|
||||
await sleep(150); // let the client notice
|
||||
fire("message_end", { message: { role: "user", content: "offline-1", timestamp: 2000 } });
|
||||
fire("message_end", { message: { role: "user", content: "offline-2", timestamp: 2001 } });
|
||||
const reconnected = await waitFor(
|
||||
() => daemon.frames.filter((f) => f.type === "hello").length >= 2,
|
||||
10000,
|
||||
);
|
||||
check("4 reconnects after drop", reconnected);
|
||||
const replayedTexts = daemon.frames
|
||||
.filter((f) => f.type === "message_end" && f.seq > seqBeforeDrop)
|
||||
.map((f) => (f.message as { text?: string }).text);
|
||||
check(
|
||||
"4 buffered events replayed after welcome.lastSeq",
|
||||
replayedTexts.includes("offline-1") && replayedTexts.includes("offline-2"),
|
||||
JSON.stringify(replayedTexts),
|
||||
);
|
||||
const allSeqs = daemon.frames.filter((f) => f.seq > 0).map((f) => f.seq);
|
||||
const dupCount = allSeqs.length - new Set(allSeqs).size;
|
||||
check("4 no duplicate seq delivery", dupCount === 0, `duplicates: ${dupCount}`);
|
||||
const seqsAfter = daemon.frames.filter((f) => f.seq > seqBeforeDrop).map((f) => f.seq);
|
||||
check(
|
||||
"4 seq continues monotonically across reconnect",
|
||||
seqsAfter.length >= 2 && seqsAfter.every((s, i) => i === 0 || s > seqsAfter[i - 1]),
|
||||
JSON.stringify(seqsAfter),
|
||||
);
|
||||
// --- Scenario 4: drop + reconnect, replay, overflow ----------------
|
||||
const seqBeforeDrop = Math.max(...daemon.frames.map((f) => f.seq));
|
||||
welcomeLastSeq = seqBeforeDrop; // daemon has everything up to the drop
|
||||
daemon.dropConnections();
|
||||
await sleep(150); // let the client notice
|
||||
fire("message_end", {
|
||||
message: { role: "user", content: "offline-1", timestamp: 2000 },
|
||||
});
|
||||
fire("message_end", {
|
||||
message: { role: "user", content: "offline-2", timestamp: 2001 },
|
||||
});
|
||||
const reconnected = await waitFor(
|
||||
() => daemon.frames.filter((f) => f.type === "hello").length >= 2,
|
||||
10000,
|
||||
);
|
||||
check("4 reconnects after drop", reconnected);
|
||||
const replayedTexts = daemon.frames
|
||||
.filter((f) => f.type === "message_end" && f.seq > seqBeforeDrop)
|
||||
.map((f) => (f.message as { text?: string }).text);
|
||||
check(
|
||||
"4 buffered events replayed after welcome.lastSeq",
|
||||
replayedTexts.includes("offline-1") && replayedTexts.includes("offline-2"),
|
||||
JSON.stringify(replayedTexts),
|
||||
);
|
||||
const allSeqs = daemon.frames.filter((f) => f.seq > 0).map((f) => f.seq);
|
||||
const dupCount = allSeqs.length - new Set(allSeqs).size;
|
||||
check(
|
||||
"4 no duplicate seq delivery",
|
||||
dupCount === 0,
|
||||
`duplicates: ${dupCount}`,
|
||||
);
|
||||
const seqsAfter = daemon.frames
|
||||
.filter((f) => f.seq > seqBeforeDrop)
|
||||
.map((f) => f.seq);
|
||||
check(
|
||||
"4 seq continues monotonically across reconnect",
|
||||
seqsAfter.length >= 2 &&
|
||||
seqsAfter.every((s, i) => i === 0 || s > seqsAfter[i - 1]),
|
||||
JSON.stringify(seqsAfter),
|
||||
);
|
||||
|
||||
// overflow both bounded buffers, then reconnect once more
|
||||
const seqBeforeOverflow = Math.max(...daemon.frames.map((f) => f.seq));
|
||||
welcomeLastSeq = seqBeforeOverflow;
|
||||
daemon.dropConnections();
|
||||
await sleep(150);
|
||||
for (let i = 0; i < 10050; i++) {
|
||||
fire("agent_settled", {}); // persisted kind: fills replay buffer + send queue
|
||||
}
|
||||
const reconnected2 = await waitFor(
|
||||
() => daemon.frames.filter((f) => f.type === "hello").length >= 3,
|
||||
15000,
|
||||
);
|
||||
check("4 reconnects after overflow window", reconnected2);
|
||||
const overflowNotice = await waitFor(
|
||||
() => daemon.frames.some((f) => f.type === "buffer_overflow"),
|
||||
5000,
|
||||
);
|
||||
const overflow = daemon.frames.find((f) => f.type === "buffer_overflow");
|
||||
check(
|
||||
"4 buffer_overflow notice after drops",
|
||||
overflowNotice && typeof (overflow?.dropped as number) === "number" && (overflow?.dropped as number) > 0,
|
||||
JSON.stringify(overflow ?? null),
|
||||
);
|
||||
// overflow both bounded buffers, then reconnect once more
|
||||
const seqBeforeOverflow = Math.max(...daemon.frames.map((f) => f.seq));
|
||||
welcomeLastSeq = seqBeforeOverflow;
|
||||
daemon.dropConnections();
|
||||
await sleep(150);
|
||||
for (let i = 0; i < 10050; i++) {
|
||||
fire("agent_settled", {}); // persisted kind: fills replay buffer + send queue
|
||||
}
|
||||
const reconnected2 = await waitFor(
|
||||
() => daemon.frames.filter((f) => f.type === "hello").length >= 3,
|
||||
15000,
|
||||
);
|
||||
check("4 reconnects after overflow window", reconnected2);
|
||||
const overflowNotice = await waitFor(
|
||||
() => daemon.frames.some((f) => f.type === "buffer_overflow"),
|
||||
5000,
|
||||
);
|
||||
const overflow = daemon.frames.find((f) => f.type === "buffer_overflow");
|
||||
check(
|
||||
"4 buffer_overflow notice after drops",
|
||||
overflowNotice &&
|
||||
typeof (overflow?.dropped as number) === "number" &&
|
||||
(overflow?.dropped as number) > 0,
|
||||
JSON.stringify(overflow ?? null),
|
||||
);
|
||||
|
||||
// --- Scenario 6: session_shutdown ----------------------------------
|
||||
const helloCountAtShutdown = daemon.frames.filter((f) => f.type === "hello").length;
|
||||
fire("session_shutdown", { reason: "quit" });
|
||||
const closed = await waitFor(() => daemon.connections() === 0, 3000);
|
||||
check("6 shutdown closes connection", closed);
|
||||
await sleep(2600); // longer than one backoff attempt (~1-2s)
|
||||
const helloCountAfter = daemon.frames.filter((f) => f.type === "hello").length;
|
||||
check("6 no reconnect after shutdown", helloCountAfter === helloCountAtShutdown);
|
||||
// --- Scenario 9: daemon hangs mid-handshake (no welcome) ----------
|
||||
// Fast watchdog knobs; read lazily by the plugin so late env wins.
|
||||
process.env.LVMH_WELCOME_TIMEOUT_MS = "400";
|
||||
process.env.LVMH_STALL_CHECK_MS = "120";
|
||||
process.env.LVMH_STALL_BYTES = "1000000";
|
||||
const hungDaemon = await startMiniDaemon(() => 0, { suppressWelcome: true });
|
||||
const prevUrl = process.env.LVMH_URL;
|
||||
process.env.LVMH_URL = hungDaemon.url;
|
||||
const pi3 = makeFakePi();
|
||||
factory(pi3);
|
||||
pi3.handlers.get("session_start")?.({ reason: "startup" }, makeFakeCtx());
|
||||
const hungHellos = await waitFor(
|
||||
() => hungDaemon.frames.filter((f) => f.type === "hello").length >= 2,
|
||||
12000,
|
||||
);
|
||||
check(
|
||||
"9 welcome timeout: abandons hung socket and retries",
|
||||
hungHellos,
|
||||
String(hungDaemon.frames.length),
|
||||
);
|
||||
pi3.handlers.get("session_shutdown")?.({ reason: "quit" }, makeFakeCtx());
|
||||
hungDaemon.close();
|
||||
process.env.LVMH_URL = prevUrl;
|
||||
|
||||
daemon.close();
|
||||
// --- Scenario 10: daemon dies without TCP close (wedged socket) ----
|
||||
// wedge() pauses daemon-side reads: TCP stays "open", undici keeps
|
||||
// buffering sends. The plugin must detect bufferedAmount not draining
|
||||
// and reconnect, or mirroring would be dead forever + memory unbounded.
|
||||
// Fresh plugin instance: its stall interval inherits the fast knobs set
|
||||
// in scenario 9 (interval period is fixed at first connect).
|
||||
const stallDaemon = await startMiniDaemon(() => 0);
|
||||
const urlBeforeStall: string | undefined = process.env.LVMH_URL;
|
||||
process.env.LVMH_URL = stallDaemon.url;
|
||||
const pi4 = makeFakePi();
|
||||
factory(pi4);
|
||||
const fire4 = (name: string, event: unknown): void => {
|
||||
const h = pi4.handlers.get(name);
|
||||
if (h === undefined) throw new Error(`missing handler ${name}`);
|
||||
const r = h(event, makeFakeCtx());
|
||||
if (r instanceof Promise) r.catch(() => undefined);
|
||||
};
|
||||
fire4("session_start", { reason: "startup" });
|
||||
await waitFor(() => stallDaemon.frames.some((f) => f.type === "hello"), 5000);
|
||||
stallDaemon.wedge();
|
||||
const bigText: string = "w".repeat(1024 * 1024);
|
||||
for (let i = 0; i < 60; i++) {
|
||||
fire4("message_end", {
|
||||
message: { role: "user", content: bigText, timestamp: 3000 + i },
|
||||
});
|
||||
}
|
||||
const wedgedRecovered = await waitFor(
|
||||
() => stallDaemon.frames.filter((f) => f.type === "hello").length >= 2,
|
||||
20000,
|
||||
);
|
||||
check(
|
||||
"10 stall watchdog: wedged socket detected, reconnected",
|
||||
wedgedRecovered,
|
||||
String(stallDaemon.frames.length),
|
||||
);
|
||||
stallDaemon.unwedge();
|
||||
fire4("message_end", {
|
||||
message: { role: "user", content: "after-stall", timestamp: 4000 },
|
||||
});
|
||||
const afterStallDelivered = await waitFor(
|
||||
() =>
|
||||
stallDaemon.frames.some(
|
||||
(f) =>
|
||||
f.type === "message_end" &&
|
||||
(f.message as { text?: string })?.text === "after-stall",
|
||||
),
|
||||
15000,
|
||||
);
|
||||
check("10 mirroring works after stall recovery", afterStallDelivered);
|
||||
fire4("session_shutdown", { reason: "quit" });
|
||||
stallDaemon.close();
|
||||
process.env.LVMH_URL = urlBeforeStall;
|
||||
delete process.env.LVMH_WELCOME_TIMEOUT_MS;
|
||||
delete process.env.LVMH_STALL_CHECK_MS;
|
||||
delete process.env.LVMH_STALL_BYTES;
|
||||
|
||||
// --- Scenario 5: dead host -----------------------------------------
|
||||
const deadDaemon = await startMiniDaemon(() => 0);
|
||||
const deadUrl = deadDaemon.url;
|
||||
deadDaemon.close(); // port now closed
|
||||
await sleep(100);
|
||||
process.env.LVMH_URL = deadUrl;
|
||||
const pi2 = makeFakePi();
|
||||
factory(pi2);
|
||||
pi2.handlers.get("session_start")?.({ reason: "startup" }, makeFakeCtx());
|
||||
await sleep(400);
|
||||
const t0 = Date.now();
|
||||
await sleep(100);
|
||||
const loopResponsive = Date.now() - t0 < 500;
|
||||
check("5 dead host: pi loads, event loop responsive", loopResponsive);
|
||||
let threw = false;
|
||||
try {
|
||||
pi2.handlers.get("message_end")?.({ message: { role: "user", content: "hi", timestamp: 1 } }, makeFakeCtx());
|
||||
} catch {
|
||||
threw = true;
|
||||
}
|
||||
check("5 dead host: handlers never throw", !threw);
|
||||
// --- Scenario 6: session_shutdown ----------------------------------
|
||||
const helloCountAtShutdown = daemon.frames.filter(
|
||||
(f) => f.type === "hello",
|
||||
).length;
|
||||
fire("session_shutdown", { reason: "quit" });
|
||||
const closed = await waitFor(() => daemon.connections() === 0, 3000);
|
||||
check("6 shutdown closes connection", closed);
|
||||
await sleep(2600); // longer than one backoff attempt (~1-2s)
|
||||
const helloCountAfter = daemon.frames.filter(
|
||||
(f) => f.type === "hello",
|
||||
).length;
|
||||
check(
|
||||
"6 no reconnect after shutdown",
|
||||
helloCountAfter === helloCountAtShutdown,
|
||||
);
|
||||
|
||||
delete process.env.LVMH_URL;
|
||||
delete process.env.LVMH_TOKEN;
|
||||
console.log(failures === 0 ? "\nALL CHECKS PASSED" : `\n${failures} CHECK(S) FAILED`);
|
||||
process.exit(failures === 0 ? 0 : 1);
|
||||
daemon.close();
|
||||
|
||||
// --- Scenario 5: dead host -----------------------------------------
|
||||
const deadDaemon = await startMiniDaemon(() => 0);
|
||||
const deadUrl = deadDaemon.url;
|
||||
deadDaemon.close(); // port now closed
|
||||
await sleep(100);
|
||||
process.env.LVMH_URL = deadUrl;
|
||||
const pi2 = makeFakePi();
|
||||
factory(pi2);
|
||||
pi2.handlers.get("session_start")?.({ reason: "startup" }, makeFakeCtx());
|
||||
await sleep(400);
|
||||
const t0 = Date.now();
|
||||
await sleep(100);
|
||||
const loopResponsive = Date.now() - t0 < 500;
|
||||
check("5 dead host: pi loads, event loop responsive", loopResponsive);
|
||||
let threw = false;
|
||||
try {
|
||||
pi2.handlers.get("message_end")?.(
|
||||
{ message: { role: "user", content: "hi", timestamp: 1 } },
|
||||
makeFakeCtx(),
|
||||
);
|
||||
} catch {
|
||||
threw = true;
|
||||
}
|
||||
check("5 dead host: handlers never throw", !threw);
|
||||
|
||||
delete process.env.LVMH_URL;
|
||||
delete process.env.LVMH_TOKEN;
|
||||
console.log(
|
||||
failures === 0 ? "\nALL CHECKS PASSED" : `\n${failures} CHECK(S) FAILED`,
|
||||
);
|
||||
process.exit(failures === 0 ? 0 : 1);
|
||||
}
|
||||
|
||||
main().catch((err: unknown) => {
|
||||
console.error("smoke harness crashed:", err);
|
||||
process.exit(1);
|
||||
console.error("smoke harness crashed:", err);
|
||||
process.exit(1);
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user