From ddc1d2b5940b2a4c5a9c8a78a19bc01734e255ca Mon Sep 17 00:00:00 2001 From: Ben Vinegar Date: Sun, 22 Feb 2026 20:21:06 -0500 Subject: [PATCH 1/2] bridge: add in-memory /logs endpoint for broker bridge --- slack-bridge/broker-bridge.mjs | 117 +++++++++++++++++++++--- test/broker-bridge.integration.test.mjs | 102 +++++++++++++++++++++ 2 files changed, 205 insertions(+), 14 deletions(-) diff --git a/slack-bridge/broker-bridge.mjs b/slack-bridge/broker-bridge.mjs index 9c15a09..742df1e 100755 --- a/slack-bridge/broker-bridge.mjs +++ b/slack-bridge/broker-bridge.mjs @@ -51,21 +51,64 @@ const DEDUPE_TTL_MS = clampInt( const MAX_BACKOFF_MS = 30_000; const INBOX_PROTOCOL_VERSION = "2026-02-1"; const BROKER_HEALTH_PATH = path.join(homedir(), ".pi", "agent", "broker-health.json"); +const LOG_BUFFER_MAX_LINES = 1000; + +const logLineBuffer = []; function ts() { return new Date().toISOString(); } +function formatLogArg(arg) { + if (typeof arg === "string") return arg; + if (arg instanceof Error) return arg.stack || arg.message; + try { + return JSON.stringify(arg); + } catch { + return String(arg); + } +} + +function pushLogLine(line) { + const lines = String(line).split(/\r?\n/); + for (const rawLine of lines) { + const normalizedLine = rawLine.trimEnd(); + if (!normalizedLine) continue; + logLineBuffer.push(normalizedLine); + } + + const overflow = logLineBuffer.length - LOG_BUFFER_MAX_LINES; + if (overflow > 0) { + logLineBuffer.splice(0, overflow); + } +} + +function logWithLevel(level, ...args) { + const timestampPrefix = `[${ts()}]`; + const line = [timestampPrefix, ...args.map(formatLogArg)].join(" "); + pushLogLine(line); + + if (level === "error") { + console.error(timestampPrefix, ...args); + return; + } + if (level === "warn") { + console.warn(timestampPrefix, ...args); + return; + } + console.log(timestampPrefix, ...args); +} + function logInfo(...args) { - console.log(`[${ts()}]`, ...args); + logWithLevel("info", ...args); } function logError(...args) { - console.error(`[${ts()}]`, ...args); + logWithLevel("error", ...args); } function logWarn(...args) { - console.warn(`[${ts()}]`, ...args); + logWithLevel("warn", ...args); } for (const key of [ @@ -669,13 +712,37 @@ async function processPulledMessage(message) { return true; } +function getLogLinesForResponse(url) { + const nParam = url.searchParams.get("n"); + const filterParam = url.searchParams.get("filter"); + + let requestedLineCount = null; + if (nParam !== null) { + const parsedN = Number.parseInt(nParam, 10); + if (!Number.isFinite(parsedN) || parsedN < 1) { + throw new Error("n must be a positive integer"); + } + requestedLineCount = Math.min(parsedN, LOG_BUFFER_MAX_LINES); + } + + let lines = logLineBuffer; + + const normalizedFilter = filterParam?.trim().toLowerCase(); + if (normalizedFilter) { + lines = lines.filter((line) => line.toLowerCase().includes(normalizedFilter)); + } + + if (requestedLineCount !== null) { + lines = lines.slice(-requestedLineCount); + } + + return lines; +} + function startApiServer() { const server = createServer(async (req, res) => { - if (req.method !== "POST") { - res.writeHead(405, { "Content-Type": "application/json" }); - res.end(JSON.stringify({ error: "Method not allowed" })); - return; - } + const url = new URL(req.url, `http://localhost:${API_PORT}`); + const pathname = url.pathname; const remoteAddr = req.socket.remoteAddress; if (remoteAddr !== "127.0.0.1" && remoteAddr !== "::1" && remoteAddr !== "::ffff:127.0.0.1") { @@ -690,6 +757,31 @@ function startApiServer() { return; } + if (pathname === "/logs") { + if (req.method !== "GET") { + res.writeHead(405, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ error: "Method not allowed" })); + return; + } + + try { + const lines = getLogLinesForResponse(url); + res.writeHead(200, { "Content-Type": "text/plain; charset=utf-8" }); + res.end(lines.length > 0 ? `${lines.join("\n")}\n` : ""); + return; + } catch (err) { + res.writeHead(400, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ error: err instanceof Error ? err.message : "invalid query params" })); + return; + } + } + + if (req.method !== "POST") { + res.writeHead(405, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ error: "Method not allowed" })); + return; + } + let rawApiRequestBody = ""; for await (const chunk of req) rawApiRequestBody += chunk; @@ -703,9 +795,6 @@ function startApiServer() { } try { - const url = new URL(req.url, `http://localhost:${API_PORT}`); - const pathname = url.pathname; - if (pathname === "/send") { const validationError = validateSendParams(apiRequestBody); if (validationError) { @@ -715,7 +804,7 @@ function startApiServer() { } const { channel, text, thread_ts } = apiRequestBody; - + const result = await sendViaBroker({ action: "chat.postMessage", routing: { channel, ...(thread_ts ? { thread_ts } : {}) }, @@ -767,7 +856,7 @@ function startApiServer() { } const { channel, timestamp, emoji } = apiRequestBody; - + await sendViaBroker({ action: "reactions.add", routing: { channel, timestamp, emoji }, @@ -780,7 +869,7 @@ function startApiServer() { } res.writeHead(404, { "Content-Type": "application/json" }); - res.end(JSON.stringify({ error: "Not found. Endpoints: POST /send, POST /reply, POST /react" })); + res.end(JSON.stringify({ error: "Not found. Endpoints: POST /send, POST /reply, POST /react, GET /logs" })); } catch (err) { res.writeHead(500, { "Content-Type": "application/json" }); res.end(JSON.stringify({ error: err instanceof Error ? err.message : "unknown error" })); diff --git a/test/broker-bridge.integration.test.mjs b/test/broker-bridge.integration.test.mjs index 4135f22..750e935 100644 --- a/test/broker-bridge.integration.test.mjs +++ b/test/broker-bridge.integration.test.mjs @@ -70,6 +70,108 @@ describe("broker pull bridge semi-integration", () => { tempDirs.length = 0; }); + it("serves in-memory recent logs via GET /logs", async () => { + await sodium.ready; + + const apiPort = await reserveFreePort(); + + const broker = createServer(async (req, res) => { + if (req.method === "POST" && req.url === "/api/inbox/pull") { + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: true, messages: [] })); + return; + } + + if (req.method === "POST" && req.url === "/api/inbox/ack") { + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: true, acked: 0 })); + return; + } + + if (req.method === "POST" && req.url === "/api/send") { + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: true, ts: "1234.5678" })); + return; + } + + res.writeHead(404, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: false, error: "not found" })); + }); + + await new Promise((resolve) => broker.listen(0, "127.0.0.1", resolve)); + servers.push(broker); + + const brokerAddress = broker.address(); + if (!brokerAddress || typeof brokerAddress === "string") { + throw new Error("failed to get broker test server address"); + } + + const testFileDir = path.dirname(fileURLToPath(import.meta.url)); + const repoRoot = path.dirname(testFileDir); + const bridgePath = path.join(repoRoot, "slack-bridge", "broker-bridge.mjs"); + const bridgeCwd = path.join(repoRoot, "slack-bridge"); + + let bridgeStdout = ""; + let bridgeStderr = ""; + + const bridge = spawn("node", [bridgePath], { + cwd: bridgeCwd, + env: { + ...process.env, + SLACK_BROKER_URL: `http://127.0.0.1:${brokerAddress.port}`, + SLACK_BROKER_WORKSPACE_ID: "T123BROKER", + SLACK_BROKER_SERVER_PRIVATE_KEY: b64(32, 11), + SLACK_BROKER_SERVER_PUBLIC_KEY: b64(32, 12), + SLACK_BROKER_SERVER_SIGNING_PRIVATE_KEY: b64(32, 13), + SLACK_BROKER_PUBLIC_KEY: b64(32, 14), + SLACK_BROKER_SIGNING_PUBLIC_KEY: b64(32, 15), + SLACK_BROKER_ACCESS_TOKEN: "test-broker-token", + SLACK_ALLOWED_USERS: "U_ALLOWED", + SLACK_BROKER_POLL_INTERVAL_MS: "100", + BRIDGE_API_PORT: String(apiPort), + }, + stdio: ["ignore", "pipe", "pipe"], + }); + + bridge.stdout.on("data", (chunk) => { + bridgeStdout += chunk.toString(); + }); + bridge.stderr.on("data", (chunk) => { + bridgeStderr += chunk.toString(); + }); + + children.push(bridge); + + await waitFor( + () => bridgeStdout.includes("Outbound API listening"), + 10_000, + 50, + `timeout waiting for startup log; stdout=${bridgeStdout}; stderr=${bridgeStderr}`, + ); + + const allLogsResponse = await fetch(`http://127.0.0.1:${apiPort}/logs`); + expect(allLogsResponse.status).toBe(200); + expect(allLogsResponse.headers.get("content-type")).toContain("text/plain"); + const allLogsText = await allLogsResponse.text(); + expect(allLogsText).toContain("Outbound API listening"); + + const filteredLogsResponse = await fetch(`http://127.0.0.1:${apiPort}/logs?filter=outbound`); + expect(filteredLogsResponse.status).toBe(200); + const filteredLogsText = await filteredLogsResponse.text(); + expect(filteredLogsText.toLowerCase()).toContain("outbound api listening"); + + const limitedLogsResponse = await fetch(`http://127.0.0.1:${apiPort}/logs?n=1`); + expect(limitedLogsResponse.status).toBe(200); + const limitedLogsText = await limitedLogsResponse.text(); + const limitedLines = limitedLogsText.trim() ? limitedLogsText.trim().split("\n") : []; + expect(limitedLines.length).toBeLessThanOrEqual(1); + + const invalidNResponse = await fetch(`http://127.0.0.1:${apiPort}/logs?n=0`); + expect(invalidNResponse.status).toBe(400); + const invalidNBody = await invalidNResponse.json(); + expect(invalidNBody.error).toContain("positive integer"); + }); + it("acks poison messages from broker to avoid infinite retry loops", async () => { await sodium.ready; From 59ae225ea117b394810f2586914bf72f49dcee22 Mon Sep 17 00:00:00 2001 From: Ben Vinegar Date: Sun, 22 Feb 2026 20:26:06 -0500 Subject: [PATCH 2/2] bridge: avoid exposing mutable log buffer reference --- slack-bridge/broker-bridge.mjs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/slack-bridge/broker-bridge.mjs b/slack-bridge/broker-bridge.mjs index 742df1e..e9658d3 100755 --- a/slack-bridge/broker-bridge.mjs +++ b/slack-bridge/broker-bridge.mjs @@ -725,7 +725,7 @@ function getLogLinesForResponse(url) { requestedLineCount = Math.min(parsedN, LOG_BUFFER_MAX_LINES); } - let lines = logLineBuffer; + let lines = logLineBuffer.slice(); const normalizedFilter = filterParam?.trim().toLowerCase(); if (normalizedFilter) {