diff --git a/extensions/signal/src/send.test.ts b/extensions/signal/src/send.test.ts index e2611408ac4b..9247c7bc9e53 100644 --- a/extensions/signal/src/send.test.ts +++ b/extensions/signal/src/send.test.ts @@ -710,6 +710,7 @@ describe("sendMessageSignal receipts", () => { it.each([ "Signal RPC -32602: quote rejected", 'Signal RPC -32602: Unrecognized field "quoteTimestamp"', + "Signal REST 400: quote metadata invalid", ])("falls back to an ordinary send when native quote metadata fails: %s", async (message) => { signalRpcRequestMock .mockRejectedValueOnce(new Error(message)) @@ -781,8 +782,12 @@ describe("sendMessageSignal receipts", () => { }); }); - it("does not retry ordinary send failures as quote fallback", async () => { - signalRpcRequestMock.mockRejectedValueOnce(new Error("Signal HTTP timed out after 10000ms")); + it.each([ + "Signal HTTP timed out after 10000ms", + "quote metadata not found after an unknown transport failure", + "Signal RPC -32000: quote metadata was rejected after an ambiguous send", + ])("does not retry an unconfirmed quote rejection: %s", async (message) => { + signalRpcRequestMock.mockRejectedValueOnce(new Error(message)); await expect( sendMessageSignal("+15551234567", "hello", { @@ -791,12 +796,113 @@ describe("sendMessageSignal receipts", () => { replyToAuthor: "+15550002222", replyToBody: "original", }), - ).rejects.toThrow("Signal HTTP timed out"); + ).rejects.toThrow(message); expect(signalRpcRequestMock).toHaveBeenCalledTimes(1); }); }); +describe("Signal quoted-message provider replay safety", () => { + it.each([ + { label: "HTTP 408", transportKind: "container", status: 408, fallback: false }, + { label: "HTTP 429", transportKind: "container", status: 429, fallback: false }, + { label: "HTTP 503", transportKind: "container", status: 503, fallback: false }, + { label: "HTTP 400", transportKind: "container", status: 400, fallback: true }, + { label: "RPC -32603", transportKind: "external-native", rpcCode: -32603, fallback: false }, + { label: "RPC -32602", transportKind: "external-native", rpcCode: -32602, fallback: true }, + ] as const)("replays a real $label request only after definitive rejection", async (testCase) => { + const requests: Array> = []; + const server = http.createServer((request, response) => { + let body = ""; + request.on("data", (chunk: Buffer) => { + body += chunk.toString("utf8"); + }); + request.on("end", () => { + const payload = JSON.parse(body) as Record; + requests.push(payload); + if (requests.length === 1) { + if (testCase.transportKind === "container") { + response.writeHead(testCase.status, { "content-type": "text/plain" }); + response.end("quote storage not found during upstream outage"); + } else { + response.writeHead(200, { "content-type": "application/json" }); + response.end( + JSON.stringify({ + jsonrpc: "2.0", + id: payload.id, + error: { + code: testCase.rpcCode, + message: "quote storage not found during upstream outage", + }, + }), + ); + } + return; + } + response.writeHead(200, { "content-type": "application/json" }); + response.end( + JSON.stringify( + testCase.transportKind === "container" + ? { timestamp: 1700000000002 } + : { + jsonrpc: "2.0", + id: payload.id, + result: { timestamp: 1700000000002 }, + }, + ), + ); + }); + }); + await new Promise((resolve) => { + server.listen(0, "127.0.0.1", resolve); + }); + const { port } = server.address() as { port: number }; + const cfg = { + channels: { + signal: { + accounts: { + default: { + account: "+15550001111", + transport: { kind: testCase.transportKind, url: `http://127.0.0.1:${port}` }, + }, + }, + }, + }, + } satisfies OpenClawConfig; + + vi.doUnmock("./client-adapter.js"); + vi.resetModules(); + const { sendMessageSignal: sendWithRealContainer } = await import("./send.js"); + + try { + const delivery = sendWithRealContainer("+15551234567", "deliver only once", { + cfg, + textMode: "plain", + replyToId: "1700000000001", + replyToAuthor: "+15550002222", + replyToBody: "original", + }); + if (testCase.fallback) { + await expect(delivery).resolves.toMatchObject({ messageId: "1700000000002" }); + } else { + const expectedError = + testCase.transportKind === "container" + ? `Signal REST ${testCase.status}` + : `Signal RPC ${testCase.rpcCode}`; + await expect(delivery).rejects.toThrow(expectedError); + } + expect(requests).toHaveLength(testCase.fallback ? 2 : 1); + const quoteTimestampPath = + testCase.transportKind === "container" ? "quote_timestamp" : "params.quoteTimestamp"; + expect(requests[0]).toHaveProperty(quoteTimestampPath, 1700000000001); + } finally { + await new Promise((resolve) => { + server.close(() => resolve()); + }); + } + }); +}); + describe("Signal native JSON-RPC recipient delivery", () => { it("rejects real per-recipient failures and preserves successful send receipts", async () => { const failureTypes = [ diff --git a/extensions/signal/src/send.ts b/extensions/signal/src/send.ts index 79876825e55c..c2387db19f34 100644 --- a/extensions/signal/src/send.ts +++ b/extensions/signal/src/send.ts @@ -259,6 +259,22 @@ function resolveSignalQuoteParams(opts: SignalSendOpts): function isSignalQuoteMetadataRejection(error: unknown): boolean { const message = error instanceof Error ? error.message : String(error); const normalized = normalizeLowercaseStringOrEmpty(message); + const rpcCode = /^signal rpc (-?\d+):/u.exec(normalized)?.[1]; + if (rpcCode !== undefined) { + if (rpcCode !== "-32602") { + return false; + } + } else { + const restStatusText = /^signal rest (\d{3}):/u.exec(normalized)?.[1]; + if (!restStatusText) { + return false; + } + const restStatus = Number(restStatusText); + // Only a definitive provider rejection makes replaying the send safe. + if (restStatus < 400 || restStatus >= 500 || restStatus === 408 || restStatus === 429) { + return false; + } + } if (!normalized.includes("quote")) { return false; }