github/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-05-08 21:08:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(channels): dedupe transport and gateway test scaffolds

Browse Source
This commit is contained in:
Peter Steinberger
2026-02-16 14:52:15 +00:00
parent f717a13039
commit 93ca0ed54f
95 changed files with 4068 additions and 5221 deletions
Download Patch File Download Diff File Expand all files Collapse all files

266
src/gateway/server.agent.gateway-server-agent-b.e2e.test.ts
View File

@@ -72,6 +72,73 @@ function expectChannels(call: Record<string, unknown>, channel: string) {
expect(call.messageChannel).toBe(channel);
}
function readAgentCommandCall(fromEnd = 1) {
return vi.mocked(agentCommand).mock.calls.at(-fromEnd)?.[0] as Record<string, unknown>;
}
function expectAgentRoutingCall(params: {
channel: string;
deliver: boolean;
to?: string;
fromEnd?: number;
}) {
const call = readAgentCommandCall(params.fromEnd);
expectChannels(call, params.channel);
if ("to" in params) {
expect(call.to).toBe(params.to);
} else {
expect(call.to).toBeUndefined();
}
expect(call.deliver).toBe(params.deliver);
expect(call.bestEffortDeliver).toBe(true);
expect(typeof call.sessionId).toBe("string");
}
async function writeMainSessionEntry(params: {
sessionId: string;
lastChannel?: string;
lastTo?: string;
}) {
await useTempSessionStorePath();
await writeSessionStore({
entries: {
main: {
sessionId: params.sessionId,
updatedAt: Date.now(),
lastChannel: params.lastChannel,
lastTo: params.lastTo,
},
},
});
}
function sendAgentWsRequest(
socket: WebSocket,
params: { reqId: string; message: string; idempotencyKey: string },
) {
socket.send(
JSON.stringify({
type: "req",
id: params.reqId,
method: "agent",
params: { message: params.message, idempotencyKey: params.idempotencyKey },
}),
);
}
async function sendAgentWsRequestAndWaitFinal(
socket: WebSocket,
params: { reqId: string; message: string; idempotencyKey: string; timeoutMs?: number },
) {
const finalP = onceMessage(
socket,
(o) => o.type === "res" && o.id === params.reqId && o.payload?.status !== "accepted",
params.timeoutMs,
);
sendAgentWsRequest(socket, params);
return await finalP;
}
async function useTempSessionStorePath() {
const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-gw-"));
testState.sessionStorePath = path.join(dir, "sessions.json");
@@ -95,16 +162,10 @@ describe("gateway server agent", () => {
},
]);
setRegistry(registry);
await useTempSessionStorePath();
await writeSessionStore({
entries: {
main: {
sessionId: "sess-teams",
updatedAt: Date.now(),
lastChannel: "msteams",
lastTo: "conversation:teams-123",
},
},
await writeMainSessionEntry({
sessionId: "sess-teams",
lastChannel: "msteams",
lastTo: "conversation:teams-123",
});
const res = await rpcReq(ws, "agent", {
message: "hi",
@@ -115,13 +176,7 @@ describe("gateway server agent", () => {
});
expect(res.ok).toBe(true);
const spy = vi.mocked(agentCommand);
const call = spy.mock.calls.at(-1)?.[0] as Record<string, unknown>;
expectChannels(call, "whatsapp");
expect(call.to).toBeUndefined();
expect(call.deliver).toBe(true);
expect(call.bestEffortDeliver).toBe(true);
expect(typeof call.sessionId).toBe("string");
expectAgentRoutingCall({ channel: "whatsapp", deliver: true });
});
test("agent accepts channel aliases (imsg/teams)", async () => {
@@ -133,16 +188,10 @@ describe("gateway server agent", () => {
},
]);
setRegistry(registry);
await useTempSessionStorePath();
await writeSessionStore({
entries: {
main: {
sessionId: "sess-alias",
updatedAt: Date.now(),
lastChannel: "imessage",
lastTo: "chat_id:123",
},
},
await writeMainSessionEntry({
sessionId: "sess-alias",
lastChannel: "imessage",
lastTo: "chat_id:123",
});
const resIMessage = await rpcReq(ws, "agent", {
message: "hi",
@@ -163,14 +212,13 @@ describe("gateway server agent", () => {
});
expect(resTeams.ok).toBe(true);
const spy = vi.mocked(agentCommand);
const lastIMessageCall = spy.mock.calls.at(-2)?.[0] as Record<string, unknown>;
expectChannels(lastIMessageCall, "imessage");
expect(lastIMessageCall.to).toBeUndefined();
const lastTeamsCall = spy.mock.calls.at(-1)?.[0] as Record<string, unknown>;
expectChannels(lastTeamsCall, "msteams");
expect(lastTeamsCall.to).toBe("conversation:teams-abc");
expectAgentRoutingCall({ channel: "imessage", deliver: true, fromEnd: 2 });
expectAgentRoutingCall({
channel: "msteams",
deliver: false,
to: "conversation:teams-abc",
fromEnd: 1,
});
});
test("agent rejects unknown channel", async () => {
@@ -186,16 +234,10 @@ describe("gateway server agent", () => {
test("agent ignores webchat last-channel for routing", async () => {
testState.allowFrom = ["+1555"];
await useTempSessionStorePath();
await writeSessionStore({
entries: {
main: {
sessionId: "sess-main-webchat",
updatedAt: Date.now(),
lastChannel: "webchat",
lastTo: "+1555",
},
},
await writeMainSessionEntry({
sessionId: "sess-main-webchat",
lastChannel: "webchat",
lastTo: "+1555",
});
const res = await rpcReq(ws, "agent", {
message: "hi",
@@ -206,26 +248,14 @@ describe("gateway server agent", () => {
});
expect(res.ok).toBe(true);
const spy = vi.mocked(agentCommand);
const call = spy.mock.calls.at(-1)?.[0] as Record<string, unknown>;
expectChannels(call, "whatsapp");
expect(call.to).toBeUndefined();
expect(call.deliver).toBe(true);
expect(call.bestEffortDeliver).toBe(true);
expect(typeof call.sessionId).toBe("string");
expectAgentRoutingCall({ channel: "whatsapp", deliver: true });
});
test("agent uses webchat for internal runs when last provider is webchat", async () => {
await useTempSessionStorePath();
await writeSessionStore({
entries: {
main: {
sessionId: "sess-main-webchat-internal",
updatedAt: Date.now(),
lastChannel: "webchat",
lastTo: "+1555",
},
},
await writeMainSessionEntry({
sessionId: "sess-main-webchat-internal",
lastChannel: "webchat",
lastTo: "+1555",
});
const res = await rpcReq(ws, "agent", {
message: "hi",
@@ -236,25 +266,11 @@ describe("gateway server agent", () => {
});
expect(res.ok).toBe(true);
const spy = vi.mocked(agentCommand);
const call = spy.mock.calls.at(-1)?.[0] as Record<string, unknown>;
expectChannels(call, "webchat");
expect(call.to).toBeUndefined();
expect(call.deliver).toBe(false);
expect(call.bestEffortDeliver).toBe(true);
expect(typeof call.sessionId).toBe("string");
expectAgentRoutingCall({ channel: "webchat", deliver: false });
});
test("agent routes bare /new through session reset before running greeting prompt", async () => {
await useTempSessionStorePath();
await writeSessionStore({
entries: {
main: {
sessionId: "sess-main-before-reset",
updatedAt: Date.now(),
},
},
});
await writeMainSessionEntry({ sessionId: "sess-main-before-reset" });
const spy = vi.mocked(agentCommand);
const callsBefore = spy.mock.calls.length;
const res = await rpcReq(ws, "agent", {
@@ -280,14 +296,11 @@ describe("gateway server agent", () => {
ws,
(o) => o.type === "res" && o.id === "ag1" && o.payload?.status !== "accepted",
);
ws.send(
JSON.stringify({
type: "req",
id: "ag1",
method: "agent",
params: { message: "hi", idempotencyKey: "idem-ag" },
}),
);
sendAgentWsRequest(ws, {
reqId: "ag1",
message: "hi",
idempotencyKey: "idem-ag",
});
const ack = await ackP;
const final = await finalP;
@@ -297,29 +310,18 @@ describe("gateway server agent", () => {
});
test("agent dedupes by idempotencyKey after completion", async () => {
const firstFinalP = onceMessage(
ws,
(o) => o.type === "res" && o.id === "ag1" && o.payload?.status !== "accepted",
);
ws.send(
JSON.stringify({
type: "req",
id: "ag1",
method: "agent",
params: { message: "hi", idempotencyKey: "same-agent" },
}),
);
const firstFinal = await firstFinalP;
const firstFinal = await sendAgentWsRequestAndWaitFinal(ws, {
reqId: "ag1",
message: "hi",
idempotencyKey: "same-agent",
});
const secondP = onceMessage(ws, (o) => o.type === "res" && o.id === "ag2");
ws.send(
JSON.stringify({
type: "req",
id: "ag2",
method: "agent",
params: { message: "hi again", idempotencyKey: "same-agent" },
}),
);
sendAgentWsRequest(ws, {
reqId: "ag2",
message: "hi again",
idempotencyKey: "same-agent",
});
const second = await secondP;
expect(second.payload).toEqual(firstFinal.payload);
});
@@ -335,52 +337,28 @@ describe("gateway server agent", () => {
const idem = "reconnect-agent";
const ws1 = await dial();
const final1P = onceMessage(
ws1,
(o) => o.type === "res" && o.id === "ag1" && o.payload?.status !== "accepted",
6000,
);
ws1.send(
JSON.stringify({
type: "req",
id: "ag1",
method: "agent",
params: { message: "hi", idempotencyKey: idem },
}),
);
const final1 = await final1P;
const final1 = await sendAgentWsRequestAndWaitFinal(ws1, {
reqId: "ag1",
message: "hi",
idempotencyKey: idem,
timeoutMs: 6000,
});
ws1.close();
const ws2 = await dial();
const final2P = onceMessage(
ws2,
(o) => o.type === "res" && o.id === "ag2" && o.payload?.status !== "accepted",
6000,
);
ws2.send(
JSON.stringify({
type: "req",
id: "ag2",
method: "agent",
params: { message: "hi again", idempotencyKey: idem },
}),
);
const res = await final2P;
const res = await sendAgentWsRequestAndWaitFinal(ws2, {
reqId: "ag2",
message: "hi again",
idempotencyKey: idem,
timeoutMs: 6000,
});
expect(res.payload).toEqual(final1.payload);
ws2.close();
});
});
test("agent events stream to webchat clients when run context is registered", async () => {
await useTempSessionStorePath();
await writeSessionStore({
entries: {
main: {
sessionId: "sess-main",
updatedAt: Date.now(),
},
},
});
await writeMainSessionEntry({ sessionId: "sess-main" });
const webchatWs = new WebSocket(`ws://127.0.0.1:${port}`, {
headers: { origin: `http://127.0.0.1:${port}` },