From 73bed33c2ebe368d51c5b18a36307dab430f7ab4 Mon Sep 17 00:00:00 2001 From: Dhruv Sharma Date: Mon, 18 May 2026 18:54:00 +0530 Subject: [PATCH] feat(web): activity events for webhooks and mux WebSocket (#1656) (#1693) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(web): activity events for webhooks and mux WebSocket (#1656) Closes #1656 — adds the 10 activity events called out in the issue, covering webhook ingress (4) and the mux WebSocket terminal server (6). Builds on the ActivityEvent infrastructure landed in #1620. Webhook events (api source): - api.webhook_unverified (warn) — 401 signature verification failure - api.webhook_rejected (warn) — 413 payload exceeds maxBodyBytes - api.webhook_received (info|warn) — 202 success, with parse/lifecycle error counts - api.webhook_failed (error) — 500 outer catch / pipeline crash Mux WS events (ui source — Node-side server only): - ui.terminal_connected — one per mux WS connection - ui.terminal_disconnected — one per close - ui.terminal_heartbeat_lost (warn) — once on 3 missed pongs (was console-only) - ui.terminal_pty_lost (warn) — fires only when subscribers are still attached, distinguishing "PTY actually died" from "user closed browser" - ui.terminal_protocol_error (warn) — invalid mux client message - ui.session_broadcast_failed (warn) — emitted on the healthy→failing transition only; re-arms after a successful poll so a long outage yields one event, not 20/min Invariants honored: no raw payloads or signatures in `data`, no client IPs in `summary` (kept in `data` only), no per-keystroke / per-pong fan-out — only on state transitions. `api.webhook_unverified` is the security-audit event; data captures `slug` and `remoteAddr` but never the failed signature. Co-Authored-By: Claude Opus 4.7 * fix(web): harden webhook and PTY activity events * chore(ci): retrigger checks * fix(web): record PTY loss across mux exit paths --------- Co-authored-by: Claude Opus 4.7 Co-authored-by: whoisasx --- .changeset/activity-events-webhooks-mux.md | 18 + packages/core/src/activity-events.ts | 12 + .../server/__tests__/mux-websocket.test.ts | 526 +++++++++++++++++- packages/web/server/mux-websocket.ts | 237 +++++++- .../web/src/__tests__/webhook-route.test.ts | 328 +++++++++++ .../src/app/api/webhooks/[...slug]/route.ts | 108 +++- packages/web/src/lib/scm-webhooks.ts | 2 +- 7 files changed, 1210 insertions(+), 21 deletions(-) create mode 100644 .changeset/activity-events-webhooks-mux.md create mode 100644 packages/web/src/__tests__/webhook-route.test.ts diff --git a/.changeset/activity-events-webhooks-mux.md b/.changeset/activity-events-webhooks-mux.md new file mode 100644 index 000000000..6f4dcc73f --- /dev/null +++ b/.changeset/activity-events-webhooks-mux.md @@ -0,0 +1,18 @@ +--- +"@aoagents/ao-core": minor +"@aoagents/ao-web": minor +--- + +Wire activity events into webhook ingress and the mux WebSocket terminal server (sub-issue of #1511, follows #1620). + +- `api.webhook_unverified` (warn) — signature verification failed; data includes `slug`, `remoteAddr`, `candidateCount` (never the failed signature) +- `api.webhook_rejected` (warn) — payload exceeded `maxBodyBytes`; data includes counts and `maxBodyBytes` (never the body) +- `api.webhook_received` (info|warn) — accepted webhook; data includes `projectIds`, `matchedSessions`, `parseErrorCount`, `lifecycleErrorCount` (never the body) +- `api.webhook_failed` (error) — outer pipeline crash with `errorMessage` +- `ui.terminal_connected` / `ui.terminal_disconnected` — one event per mux WS connection lifecycle +- `ui.terminal_heartbeat_lost` (warn) — fires once on 3 missed pongs (was console-only) +- `ui.terminal_pty_lost` (warn) — fires when PTY exits with subscribers attached (distinguishes "PTY died" from "user closed browser") +- `ui.terminal_protocol_error` (warn) — invalid mux client message +- `ui.session_broadcast_failed` (warn) — emitted on the healthy→failing transition only (re-arms after a successful poll), so a long outage produces one event, not 20/min + +`api.webhook_unverified` is the security-audit event; treat 401s on webhooks as a signal worth retaining for the full 7-day window. diff --git a/packages/core/src/activity-events.ts b/packages/core/src/activity-events.ts index 789e2616d..aafcc4f42 100644 --- a/packages/core/src/activity-events.ts +++ b/packages/core/src/activity-events.ts @@ -71,6 +71,18 @@ export type ActivityEventKind = | "detecting.escalated" // Report watcher | "report_watcher.triggered" + // Webhook ingress (api source) + | "api.webhook_unverified" + | "api.webhook_rejected" + | "api.webhook_received" + | "api.webhook_failed" + // WebSocket terminal mux (ui source — Node-side server only) + | "ui.terminal_connected" + | "ui.terminal_disconnected" + | "ui.terminal_heartbeat_lost" + | "ui.terminal_pty_lost" + | "ui.terminal_protocol_error" + | "ui.session_broadcast_failed" // Recovery/forensic instrumentation | "recovery.session_failed" | "recovery.action_failed" diff --git a/packages/web/server/__tests__/mux-websocket.test.ts b/packages/web/server/__tests__/mux-websocket.test.ts index 48f38e96f..3e9c41377 100644 --- a/packages/web/server/__tests__/mux-websocket.test.ts +++ b/packages/web/server/__tests__/mux-websocket.test.ts @@ -1,15 +1,26 @@ import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; import { EventEmitter } from "node:events"; +import type { Socket } from "node:net"; +import { WebSocket } from "ws"; import type { SessionBroadcaster as SessionBroadcasterType } from "../mux-websocket"; // vi.mock factories run before module-level statements. Hoist the mock // fns so the factories close over the same instances the tests use. -const { mockSpawn, mockPtySpawn, mockTmuxHasSession } = vi.hoisted(() => ({ +const { mockSpawn, mockPtySpawn, mockTmuxHasSession, recordActivityEvent } = vi.hoisted(() => ({ mockSpawn: vi.fn(), mockPtySpawn: vi.fn(), mockTmuxHasSession: vi.fn(), + recordActivityEvent: vi.fn(), })); +vi.mock("@aoagents/ao-core", async (importOriginal) => { + const actual = (await importOriginal()) as Record; + return { + ...actual, + recordActivityEvent: (event: unknown) => recordActivityEvent(event), + }; +}); + vi.mock("node:child_process", async (importOriginal) => { const actual = (await importOriginal()) as Record; const spawnFn = (...args: unknown[]) => mockSpawn(...args); @@ -34,21 +45,72 @@ vi.mock("../tmux-utils.js", () => ({ findTmux: () => "/usr/bin/tmux", validateSessionId: () => true, resolveTmuxSession: () => "ao-177", + resolvePipePath: () => null, tmuxHasSession: (...args: unknown[]) => mockTmuxHasSession(...args), })); -const { SessionBroadcaster, TerminalManager } = await import("../mux-websocket"); +const { SessionBroadcaster, TerminalManager, createMuxWebSocket, handleWindowsPipeMessage } = + await import("../mux-websocket"); // Mock global fetch const mockFetch = vi.fn(); global.fetch = mockFetch; +type MockPty = { + dataHandlers: Array<(data: string) => void>; + exitHandlers: Array<(event: { exitCode: number }) => void>; + onData: ReturnType; + onExit: ReturnType; + write: ReturnType; + resize: ReturnType; + kill: ReturnType; + emitData: (data: string) => void; + emitExit: (exitCode: number) => Promise; +}; + +const ptyInstances: MockPty[] = []; + +function createMockPty(): MockPty { + const pty = {} as MockPty; + pty.dataHandlers = []; + pty.exitHandlers = []; + pty.onData = vi.fn((handler: (data: string) => void) => { + pty.dataHandlers.push(handler); + }); + pty.onExit = vi.fn((handler: (event: { exitCode: number }) => void) => { + pty.exitHandlers.push(handler); + }); + pty.write = vi.fn(); + pty.resize = vi.fn(); + pty.kill = vi.fn(); + pty.emitData = (data: string) => { + for (const handler of pty.dataHandlers) handler(data); + }; + pty.emitExit = async (exitCode: number) => { + await Promise.all([...pty.exitHandlers].map((handler) => handler({ exitCode }))); + }; + ptyInstances.push(pty); + return pty; +} + +function resetPtyMock(): void { + ptyInstances.length = 0; + mockSpawn.mockReset(); + mockPtySpawn.mockReset(); + mockTmuxHasSession.mockReset(); + mockTmuxHasSession.mockResolvedValue(true); + mockSpawn.mockImplementation(() => new EventEmitter()); + mockPtySpawn.mockImplementation(createMockPty); +} + describe("SessionBroadcaster", () => { let broadcaster: SessionBroadcasterType; beforeEach(() => { vi.useFakeTimers(); mockFetch.mockReset(); + recordActivityEvent.mockClear(); + resetPtyMock(); broadcaster = new SessionBroadcaster("3000"); }); @@ -262,6 +324,451 @@ describe("SessionBroadcaster", () => { expect(mockFetch).toHaveBeenCalledTimes(1); }); }); + + describe("ui.session_broadcast_failed activity events", () => { + function failedKinds(): string[] { + return recordActivityEvent.mock.calls + .map(([e]) => (e as { kind: string }).kind) + .filter((k) => k === "ui.session_broadcast_failed"); + } + + it("emits exactly once on the healthy→failing transition", async () => { + // First fetch fails — triggers emission + mockFetch.mockRejectedValueOnce(new Error("ECONNREFUSED")); + // Second fetch (3s later) also fails — should NOT emit again + mockFetch.mockRejectedValueOnce(new Error("ECONNREFUSED")); + + broadcaster.subscribe(vi.fn()); + await vi.advanceTimersByTimeAsync(10); + await vi.advanceTimersByTimeAsync(3010); + + expect(failedKinds()).toEqual(["ui.session_broadcast_failed"]); + }); + + it("re-arms after recovery (success → failure emits again)", async () => { + // fail → succeed → fail + mockFetch.mockRejectedValueOnce(new Error("net down")); + mockFetch.mockResolvedValueOnce({ ok: true, json: async () => ({ sessions: [] }) }); + mockFetch.mockRejectedValueOnce(new Error("net down again")); + + broadcaster.subscribe(vi.fn()); + await vi.advanceTimersByTimeAsync(10); + await vi.advanceTimersByTimeAsync(3010); // poll #1 → success + await vi.advanceTimersByTimeAsync(3010); // poll #2 → failure + + expect(failedKinds().length).toBe(2); + }); + + it("emits with source=ui, level=warn, and the failure URL in data", async () => { + mockFetch.mockRejectedValueOnce(new Error("ETIMEDOUT")); + + broadcaster.subscribe(vi.fn()); + await vi.advanceTimersByTimeAsync(10); + + expect(recordActivityEvent).toHaveBeenCalledWith( + expect.objectContaining({ + source: "ui", + kind: "ui.session_broadcast_failed", + level: "warn", + }), + ); + const call = recordActivityEvent.mock.calls.find( + ([e]) => (e as { kind: string }).kind === "ui.session_broadcast_failed", + )![0] as { data: Record }; + expect(call.data["url"]).toContain("/api/sessions/patches"); + expect(call.data["errorMessage"]).toContain("ETIMEDOUT"); + }); + + it("includes httpStatus when fetch returns non-OK response", async () => { + mockFetch.mockResolvedValueOnce({ ok: false, status: 503 }); + + broadcaster.subscribe(vi.fn()); + await vi.advanceTimersByTimeAsync(10); + + const call = recordActivityEvent.mock.calls.find( + ([e]) => (e as { kind: string }).kind === "ui.session_broadcast_failed", + )![0] as { data: Record }; + expect(call.data["httpStatus"]).toBe(503); + }); + }); +}); + +// ── Connection-level activity events ────────────────────────────────── +// These verify ui.terminal_* events fire at the right WS lifecycle points. +// We exercise the connection handler directly by emitting "connection" on +// the WebSocketServer and feeding a fake ws + IncomingMessage stand-in. + +class FakeWS extends EventEmitter { + readyState: 0 | 1 | 2 | 3 = WebSocket.OPEN; + bufferedAmount = 0; + ping = vi.fn(); + terminate = vi.fn(() => { + this.readyState = WebSocket.CLOSED; + }); + send = vi.fn(); +} + +class FakePipeSocket extends EventEmitter { + write = vi.fn(); + end = vi.fn(() => { + this.emit("close"); + }); + destroy = vi.fn(() => { + this.emit("close"); + }); +} + +function makeFakeRequest(opts?: { remoteAddress?: string; xff?: string }) { + return { + headers: opts?.xff ? { "x-forwarded-for": opts.xff } : {}, + socket: { remoteAddress: opts?.remoteAddress ?? "127.0.0.1" }, + }; +} + +describe("mux WebSocket connection events", () => { + beforeEach(() => { + vi.useFakeTimers(); + recordActivityEvent.mockClear(); + resetPtyMock(); + }); + + afterEach(() => { + vi.clearAllTimers(); + vi.useRealTimers(); + }); + + function emitConnection(opts?: Parameters[0]) { + const wss = createMuxWebSocket(); + if (!wss) { + throw new Error("mux WS server not created — node-pty unavailable"); + } + const ws = new FakeWS(); + wss.emit("connection", ws as unknown as WebSocket, makeFakeRequest(opts)); + return { wss, ws }; + } + + function findEvent(kind: string): { data: Record } | undefined { + const found = recordActivityEvent.mock.calls.find( + ([e]) => (e as { kind: string }).kind === kind, + ); + return found?.[0] as { data: Record } | undefined; + } + + it("emits ui.terminal_connected on a new mux connection (with remoteAddr)", () => { + emitConnection({ xff: "198.51.100.5, 10.0.0.1" }); + + expect(recordActivityEvent).toHaveBeenCalledWith( + expect.objectContaining({ source: "ui", kind: "ui.terminal_connected" }), + ); + const evt = findEvent("ui.terminal_connected")!; + expect(evt.data["remoteAddr"]).toBe("198.51.100.5"); + }); + + it("emits ui.terminal_disconnected exactly once on close", () => { + const { ws } = emitConnection(); + recordActivityEvent.mockClear(); + + ws.emit("close", 1000, Buffer.from("normal")); + + const calls = recordActivityEvent.mock.calls.filter( + ([e]) => (e as { kind: string }).kind === "ui.terminal_disconnected", + ); + expect(calls.length).toBe(1); + const evt = findEvent("ui.terminal_disconnected")!; + expect(evt.data["code"]).toBe(1000); + expect(evt.data["reason"]).toBe("normal"); + }); + + it("emits ui.terminal_heartbeat_lost once on 3 missed pongs and terminates", () => { + const { ws } = emitConnection(); + recordActivityEvent.mockClear(); + + // Each 15s interval sends a ping and increments missedPongs by 1. + // After 3 ticks (45s) it should hit MAX_MISSED_PONGS=3 and terminate. + vi.advanceTimersByTime(15_000); + vi.advanceTimersByTime(15_000); + vi.advanceTimersByTime(15_000); + + const calls = recordActivityEvent.mock.calls.filter( + ([e]) => (e as { kind: string }).kind === "ui.terminal_heartbeat_lost", + ); + expect(calls.length).toBe(1); + expect(ws.terminate).toHaveBeenCalled(); + + // Issue invariant: at most one emit per state change — extra ticks must not + // produce another event. + vi.advanceTimersByTime(15_000); + expect( + recordActivityEvent.mock.calls.filter( + ([e]) => (e as { kind: string }).kind === "ui.terminal_heartbeat_lost", + ).length, + ).toBe(1); + }); + + it("does NOT emit heartbeat_lost when pong arrives before 3 missed pings", () => { + const { ws } = emitConnection(); + recordActivityEvent.mockClear(); + + vi.advanceTimersByTime(15_000); // missedPongs=1 + ws.emit("pong"); // resets to 0 + vi.advanceTimersByTime(15_000); // missedPongs=1 + vi.advanceTimersByTime(15_000); // missedPongs=2 + + expect( + recordActivityEvent.mock.calls.filter( + ([e]) => (e as { kind: string }).kind === "ui.terminal_heartbeat_lost", + ).length, + ).toBe(0); + expect(ws.terminate).not.toHaveBeenCalled(); + }); + + it("emits ui.terminal_protocol_error on malformed client message", () => { + const { ws } = emitConnection(); + recordActivityEvent.mockClear(); + + ws.emit("message", Buffer.from("not-json{{{")); + + const calls = recordActivityEvent.mock.calls.filter( + ([e]) => (e as { kind: string }).kind === "ui.terminal_protocol_error", + ); + expect(calls.length).toBe(1); + const evt = findEvent("ui.terminal_protocol_error")!; + expect(evt.data["errorMessage"]).toBeTruthy(); + }); +}); + +describe("Windows pipe ui.terminal_pty_lost activity events", () => { + beforeEach(() => { + recordActivityEvent.mockClear(); + }); + + function framedMessage(type: number, payload: unknown): Buffer { + const body = Buffer.from(JSON.stringify(payload), "utf-8"); + const header = Buffer.alloc(5); + header.writeUInt8(type, 0); + header.writeUInt32BE(body.length, 1); + return Buffer.concat([header, body]); + } + + function openPipe() { + const ws = new FakeWS(); + const pipe = new FakePipeSocket(); + const winPipes = new Map(); + const winPipeBuffers = new Map(); + const deps = { + connect: vi.fn(() => pipe as unknown as Socket), + resolvePipePath: vi.fn(() => "\\\\.\\pipe\\ao-pty-app-1"), + }; + + handleWindowsPipeMessage( + { id: "app-1", type: "open", projectId: "proj-1" }, + ws, + winPipes, + winPipeBuffers, + deps, + ); + pipe.emit("connect"); + recordActivityEvent.mockClear(); + ws.send.mockClear(); + + return { ws, pipe, winPipes, winPipeBuffers, deps }; + } + + function ptyLostEvents(): Array<{ data: Record; sessionId?: string }> { + return recordActivityEvent.mock.calls + .map(([e]) => e as { kind: string; data: Record; sessionId?: string }) + .filter((event) => event.kind === "ui.terminal_pty_lost"); + } + + it("emits ui.terminal_pty_lost when the PTY host pipe closes while the socket is open", () => { + const { ws, pipe } = openPipe(); + + pipe.emit("close"); + + expect(ptyLostEvents()).toEqual([ + expect.objectContaining({ + sessionId: "app-1", + data: expect.objectContaining({ + sessionId: "app-1", + transport: "windows_pipe", + reason: "pipe_closed", + }), + }), + ]); + expect(ws.send).toHaveBeenCalledWith( + JSON.stringify({ ch: "terminal", id: "app-1", type: "exited", code: 0, projectId: "proj-1" }), + ); + }); + + it("emits ui.terminal_pty_lost when the PTY host reports not alive", () => { + const { ws, pipe } = openPipe(); + + pipe.emit("data", framedMessage(0x07, { alive: false })); + pipe.emit("close"); + + const events = ptyLostEvents(); + expect(events).toHaveLength(1); + expect(events[0]).toEqual( + expect.objectContaining({ + sessionId: "app-1", + data: expect.objectContaining({ + sessionId: "app-1", + transport: "windows_pipe", + reason: "host_not_alive", + }), + }), + ); + expect(ws.send).toHaveBeenCalledWith( + JSON.stringify({ ch: "terminal", id: "app-1", type: "exited", code: 0, projectId: "proj-1" }), + ); + }); + + it("does not emit ui.terminal_pty_lost for an intentional client close", () => { + const { ws, winPipes, winPipeBuffers, deps } = openPipe(); + + handleWindowsPipeMessage( + { id: "app-1", type: "close", projectId: "proj-1" }, + ws, + winPipes, + winPipeBuffers, + deps, + ); + + expect(ptyLostEvents()).toEqual([]); + }); +}); + +describe("TerminalManager ui.terminal_pty_lost activity events", () => { + beforeEach(() => { + recordActivityEvent.mockClear(); + resetPtyMock(); + }); + + function ptyLostEvents(): Array<{ data: Record; sessionId?: string }> { + return recordActivityEvent.mock.calls + .map(([e]) => e as { kind: string; data: Record; sessionId?: string }) + .filter((event) => event.kind === "ui.terminal_pty_lost"); + } + + it("emits ui.terminal_pty_lost when a subscribed PTY exits and reattach fails", async () => { + const manager = new TerminalManager("/usr/bin/tmux"); + manager.open("app-1", "proj-1", "tmux-app-1"); + const pty = ptyInstances[0]; + expect(pty).toBeDefined(); + + manager.subscribe("app-1", "proj-1", vi.fn(), vi.fn()); + mockPtySpawn.mockImplementationOnce(() => { + throw new Error("reattach unavailable"); + }); + + await pty!.emitExit(9); + + const events = ptyLostEvents(); + expect(events).toHaveLength(1); + expect(events[0]).toEqual( + expect.objectContaining({ + sessionId: "app-1", + data: expect.objectContaining({ + sessionId: "app-1", + exitCode: 9, + subscriberCount: 1, + reattachError: "reattach unavailable", + }), + }), + ); + }); + + it("emits ui.terminal_pty_lost when a subscribed PTY exits and reattach succeeds", async () => { + const manager = new TerminalManager("/usr/bin/tmux"); + manager.open("app-1", "proj-1", "tmux-app-1"); + const pty = ptyInstances[0]; + expect(pty).toBeDefined(); + + manager.subscribe("app-1", "proj-1", vi.fn(), vi.fn()); + + await pty!.emitExit(9); + + const events = ptyLostEvents(); + expect(mockPtySpawn).toHaveBeenCalledTimes(2); + expect(events).toHaveLength(1); + expect(events[0]).toEqual( + expect.objectContaining({ + sessionId: "app-1", + data: expect.objectContaining({ + sessionId: "app-1", + exitCode: 9, + subscriberCount: 1, + reattachRecovered: true, + reattachExhausted: false, + }), + }), + ); + }); + + it("does not emit ui.terminal_pty_lost when the last subscriber already left", async () => { + const manager = new TerminalManager("/usr/bin/tmux"); + manager.open("app-1", "proj-1", "tmux-app-1"); + const pty = ptyInstances[0]; + expect(pty).toBeDefined(); + + const unsubscribe = manager.subscribe("app-1", "proj-1", vi.fn(), vi.fn()); + unsubscribe(); + + await pty!.emitExit(0); + + expect(ptyLostEvents()).toEqual([]); + }); + + it("emits ui.terminal_pty_lost at most once across reattach cycles", async () => { + const manager = new TerminalManager("/usr/bin/tmux"); + manager.open("app-1", "proj-1", "tmux-app-1"); + const firstPty = ptyInstances[0]; + expect(firstPty).toBeDefined(); + + manager.subscribe("app-1", "proj-1", vi.fn(), vi.fn()); + mockPtySpawn.mockImplementationOnce(() => { + throw new Error("first reattach unavailable"); + }); + await firstPty!.emitExit(7); + expect(ptyLostEvents()).toHaveLength(1); + + // A client may try to re-open the terminal after the first PTY loss. + // A second failed reattach should not produce another activity event for + // the same terminal entry. + manager.open("app-1", "proj-1", "tmux-app-1"); + const secondPty = ptyInstances[1]; + expect(secondPty).toBeDefined(); + mockPtySpawn.mockImplementationOnce(() => { + throw new Error("second reattach unavailable"); + }); + await secondPty!.emitExit(8); + + expect(ptyLostEvents()).toHaveLength(1); + }); + + it("re-arms ui.terminal_pty_lost after a successful reattach survives the grace period", async () => { + vi.useFakeTimers(); + try { + const manager = new TerminalManager("/usr/bin/tmux"); + manager.open("app-1", "proj-1", "tmux-app-1"); + const firstPty = ptyInstances[0]; + expect(firstPty).toBeDefined(); + + manager.subscribe("app-1", "proj-1", vi.fn(), vi.fn()); + await firstPty!.emitExit(7); + expect(ptyLostEvents()).toHaveLength(1); + + await vi.advanceTimersByTimeAsync(5_000); + + const secondPty = ptyInstances[1]; + expect(secondPty).toBeDefined(); + await secondPty!.emitExit(8); + + expect(ptyLostEvents()).toHaveLength(2); + } finally { + vi.useRealTimers(); + } + }); }); describe("TerminalManager.open — tmux target args (regression for #1714)", () => { @@ -325,6 +832,7 @@ describe("TerminalManager.open — re-attach skipped when tmux session is gone ( mockSpawn.mockReset(); mockPtySpawn.mockReset(); mockTmuxHasSession.mockReset(); + recordActivityEvent.mockClear(); capturedOnExit = undefined; mockSpawn.mockImplementation(() => new EventEmitter()); @@ -355,6 +863,20 @@ describe("TerminalManager.open — re-attach skipped when tmux session is gone ( // Subscribers were notified with the original exit code. expect(exitCb).toHaveBeenCalledTimes(1); expect(exitCb).toHaveBeenCalledWith(0); + expect(recordActivityEvent).toHaveBeenCalledWith( + expect.objectContaining({ + source: "ui", + kind: "ui.terminal_pty_lost", + level: "warn", + sessionId: "ao-177", + data: expect.objectContaining({ + sessionId: "ao-177", + exitCode: 0, + reattachSkipped: true, + tmuxSessionPresent: false, + }), + }), + ); }); it("still re-attaches when has-session reports the tmux session is alive", async () => { diff --git a/packages/web/server/mux-websocket.ts b/packages/web/server/mux-websocket.ts index 5515574df..5db3d4630 100644 --- a/packages/web/server/mux-websocket.ts +++ b/packages/web/server/mux-websocket.ts @@ -16,7 +16,7 @@ import { tmuxHasSession, validateSessionId, } from "./tmux-utils.js"; -import { getEnvDefaults, isWindows } from "@aoagents/ao-core"; +import { getEnvDefaults, isWindows, recordActivityEvent } from "@aoagents/ao-core"; // These types mirror src/lib/mux-protocol.ts exactly. // tsconfig.server.json constrains rootDir to "server/", so we cannot import @@ -63,6 +63,9 @@ export class SessionBroadcaster { private errorSubscribers = new Set<(error: string) => void>(); private intervalId: ReturnType | null = null; private polling = false; + // Tracks the last fetch outcome so we only emit ui.session_broadcast_failed on + // the healthy → failing transition (not every 3s during an outage). + private lastFetchOk = true; private readonly baseUrl: string; constructor(nextPort: string) { @@ -158,18 +161,42 @@ export class SessionBroadcaster { if (!res.ok) { const msg = `Session fetch failed: HTTP ${res.status}`; console.warn(`[SessionBroadcaster] ${msg}`); + this.recordFetchFailure(msg, { httpStatus: res.status }); return { sessions: null, error: msg }; } const data = (await res.json()) as { sessions?: SessionPatch[] }; + this.lastFetchOk = true; return { sessions: data.sessions ?? null, error: null }; } catch (err) { clearTimeout(timeoutId); const msg = err instanceof Error ? err.message : String(err); console.warn("[SessionBroadcaster] fetchSnapshot error:", msg); + this.recordFetchFailure(msg); return { sessions: null, error: msg }; } } + /** + * Emit ui.session_broadcast_failed once per healthy→failing transition. + * The broadcaster polls every 3s; emitting on every failure during a long + * outage would flood the events table (~20/min). Recovery resets the flag. + */ + private recordFetchFailure(message: string, extra?: Record): void { + if (!this.lastFetchOk) return; + this.lastFetchOk = false; + recordActivityEvent({ + source: "ui", + kind: "ui.session_broadcast_failed", + level: "warn", + summary: `session broadcaster fetch failed: ${message}`, + data: { + url: `${this.baseUrl}/api/sessions/patches`, + errorMessage: message, + ...extra, + }, + }); + } + private disconnect(): void { if (this.intervalId !== null) { clearInterval(this.intervalId); @@ -199,6 +226,7 @@ interface ManagedTerminal { buffer: string[]; bufferBytes: number; reattachAttempts: number; + ptyLostEmitted: boolean; /** * Pending grace-period timer that resets reattachAttempts when the * currently-attached PTY survives REATTACH_RESET_GRACE_MS. Tracked so @@ -276,6 +304,7 @@ export class TerminalManager { buffer: [], bufferBytes: 0, reattachAttempts: 0, + ptyLostEmitted: false, }; this.terminals.set(key, terminal); } @@ -346,6 +375,7 @@ export class TerminalManager { terminal.resetTimer = undefined; if (terminal.pty === pty) { terminal.reattachAttempts = 0; + terminal.ptyLostEmitted = false; } }, REATTACH_RESET_GRACE_MS); terminal.resetTimer.unref(); @@ -383,6 +413,7 @@ export class TerminalManager { pty.onExit(async ({ exitCode }) => { console.log(`[MuxServer] PTY exited for ${id} with code ${exitCode}`); terminal.pty = null; + let reattachError: string | undefined; // Skip the re-attach loop entirely when the underlying tmux session is // gone (e.g. user pressed Ctrl-C in the pane and the launch command @@ -392,15 +423,33 @@ export class TerminalManager { // clean user-initiated termination — see issue #1756. The // MAX_REATTACH_ATTEMPTS bound from #1640 still covers tmux server // hiccups where the session does still exist. - if ( - terminal.subscribers.size > 0 && - !(await tmuxHasSession(this.TMUX, tmuxSessionId)) - ) { + if (terminal.subscribers.size > 0 && !(await tmuxHasSession(this.TMUX, tmuxSessionId))) { console.log(`[MuxServer] tmux session ${tmuxSessionId} is gone, not re-attaching`); if (terminal.resetTimer) { clearTimeout(terminal.resetTimer); terminal.resetTimer = undefined; } + if (!terminal.ptyLostEmitted) { + terminal.ptyLostEmitted = true; + recordActivityEvent({ + projectId, + sessionId: id, + source: "ui", + kind: "ui.terminal_pty_lost", + level: "warn", + summary: `terminal PTY exited (code ${exitCode}) — tmux session gone`, + data: { + sessionId: id, + exitCode, + reattachAttempts: terminal.reattachAttempts, + maxReattachAttempts: MAX_REATTACH_ATTEMPTS, + reattachExhausted: false, + reattachSkipped: true, + tmuxSessionPresent: false, + subscriberCount: terminal.subscribers.size, + }, + }); + } for (const cb of terminal.exitCallbacks) { cb(exitCode); } @@ -423,14 +472,63 @@ export class TerminalManager { ); try { this.open(id, projectId, tmuxSessionId); + if (!terminal.ptyLostEmitted) { + terminal.ptyLostEmitted = true; + recordActivityEvent({ + projectId, + sessionId: id, + source: "ui", + kind: "ui.terminal_pty_lost", + level: "warn", + summary: `terminal PTY exited (code ${exitCode}) — reattached`, + data: { + sessionId: id, + exitCode, + reattachAttempts: terminal.reattachAttempts, + maxReattachAttempts: MAX_REATTACH_ATTEMPTS, + reattachExhausted: false, + reattachRecovered: true, + subscriberCount: terminal.subscribers.size, + }, + }); + } return; // re-attached — don't notify exit } catch (err) { + reattachError = err instanceof Error ? err.message : String(err); console.error(`[MuxServer] Failed to re-attach ${id}:`, err); } } else if (terminal.reattachAttempts >= MAX_REATTACH_ATTEMPTS) { console.error(`[MuxServer] Max re-attach attempts reached for ${id}, giving up`); } + // PTY actually died (vs user closed browser): only emit when subscribers + // are still attached — otherwise the exit is just normal cleanup. + // Keep this event one-shot for the terminal entry. Clients may re-open + // the same terminal after a failed reattach; repeated PTY exits should + // not flood the activity log for the same loss condition. + if (terminal.subscribers.size > 0 && !terminal.ptyLostEmitted) { + terminal.ptyLostEmitted = true; + recordActivityEvent({ + projectId, + sessionId: id, + source: "ui", + kind: "ui.terminal_pty_lost", + level: "warn", + summary: `terminal PTY exited (code ${exitCode})${ + terminal.reattachAttempts >= MAX_REATTACH_ATTEMPTS ? " — reattach exhausted" : "" + }`, + data: { + sessionId: id, + exitCode, + reattachAttempts: terminal.reattachAttempts, + maxReattachAttempts: MAX_REATTACH_ATTEMPTS, + reattachExhausted: terminal.reattachAttempts >= MAX_REATTACH_ATTEMPTS, + subscriberCount: terminal.subscribers.size, + ...(reattachError ? { reattachError } : {}), + }, + }); + } + // Notify subscribers that the terminal has exited (re-attach failed or no subscribers) for (const cb of terminal.exitCallbacks) { cb(exitCode); @@ -515,6 +613,8 @@ export class TerminalManager { // ── Windows Pipe Relay (extracted for testability) ── +const intentionalWinPipeCloses = new WeakSet(); + /** Minimal WebSocket-like interface for the pipe relay handler */ export interface WsSink { send(data: string): void; @@ -586,8 +686,36 @@ export function handleWindowsPipeMessage( const pipeSocket = deps.connect(pipePath); winPipes.set(pipeKey, pipeSocket); winPipeBuffers.set(pipeKey, Buffer.alloc(0)); + let ptyLostEmitted = false; + const recordWindowsPtyLost = ( + reason: "pipe_closed" | "host_not_alive" | "pipe_error", + extra?: Record, + ): void => { + if (ptyLostEmitted || ws.readyState !== WS_OPEN) return; + ptyLostEmitted = true; + recordActivityEvent({ + projectId, + sessionId: id, + source: "ui", + kind: "ui.terminal_pty_lost", + level: "warn", + summary: + reason === "host_not_alive" + ? `terminal PTY host reported not alive for ${id}` + : reason === "pipe_error" + ? `terminal PTY host pipe errored for ${id}` + : `terminal PTY host pipe closed for ${id}`, + data: { + sessionId: id, + transport: "windows_pipe", + reason, + ...extra, + }, + }); + }; pipeSocket.on("error", (err) => { + recordWindowsPtyLost("pipe_error", { errorMessage: err.message }); winPipes.delete(pipeKey); winPipeBuffers.delete(pipeKey); pipeSocket.destroy(); @@ -637,9 +765,8 @@ export function handleWindowsPipeMessage( try { const status = JSON.parse(payload.toString("utf-8")) as { alive: boolean }; if (!status.alive && ws.readyState === WS_OPEN) { - ws.send( - JSON.stringify({ ch: "terminal", id, type: "exited", code: 0, ...echo }), - ); + recordWindowsPtyLost("host_not_alive"); + ws.send(JSON.stringify({ ch: "terminal", id, type: "exited", code: 0, ...echo })); } } catch { /* ignore parse errors */ @@ -651,7 +778,11 @@ export function handleWindowsPipeMessage( pipeSocket.on("close", () => { winPipes.delete(pipeKey); winPipeBuffers.delete(pipeKey); + const intentionalClose = intentionalWinPipeCloses.delete(pipeSocket); if (ws.readyState === WS_OPEN) { + if (!intentionalClose) { + recordWindowsPtyLost("pipe_closed"); + } ws.send(JSON.stringify({ ch: "terminal", id, type: "exited", code: 0, ...echo })); } }); @@ -678,6 +809,7 @@ export function handleWindowsPipeMessage( } else if (type === "close") { const pipeSocket = winPipes.get(pipeKey); if (pipeSocket) { + intentionalWinPipeCloses.add(pipeSocket); pipeSocket.end(); winPipes.delete(pipeKey); winPipeBuffers.delete(pipeKey); @@ -706,9 +838,26 @@ export function createMuxWebSocket(tmuxPath?: string | null): WebSocketServer | const wss = new WebSocketServer({ noServer: true }); - wss.on("connection", (ws) => { + wss.on("connection", (ws, request) => { console.log("[MuxServer] New mux connection"); + const connectedAt = Date.now(); + // Best-effort remote addr — proxy headers if present, else socket peer. + const xff = request?.headers["x-forwarded-for"]; + const xffStr = Array.isArray(xff) ? xff[0] : xff; + const remoteAddr = + (typeof xffStr === "string" ? xffStr.split(",")[0]?.trim() : undefined) ?? + request?.socket?.remoteAddress ?? + undefined; + + recordActivityEvent({ + source: "ui", + kind: "ui.terminal_connected", + level: "info", + summary: "mux WebSocket connection opened", + data: { remoteAddr }, + }); + const subscriptions = new Map void>(); // Windows: named pipe sockets keyed by session ID const winPipes = new Map>(); @@ -716,6 +865,7 @@ export function createMuxWebSocket(tmuxPath?: string | null): WebSocketServer | const winPipeBuffers = new Map(); let sessionUnsubscribe: (() => void) | null = null; let missedPongs = 0; + let heartbeatLostEmitted = false; const MAX_MISSED_PONGS = 3; // Heartbeat: send native WebSocket ping every 15s. @@ -728,6 +878,22 @@ export function createMuxWebSocket(tmuxPath?: string | null): WebSocketServer | missedPongs += 1; if (missedPongs >= MAX_MISSED_PONGS) { console.log("[MuxServer] Too many missed pongs, terminating connection"); + if (!heartbeatLostEmitted) { + heartbeatLostEmitted = true; + recordActivityEvent({ + source: "ui", + kind: "ui.terminal_heartbeat_lost", + level: "warn", + summary: `mux WebSocket heartbeat lost (${missedPongs} missed pongs)`, + data: { + missedPongs, + maxMissedPongs: MAX_MISSED_PONGS, + connectionAgeMs: Date.now() - connectedAt, + remoteAddr, + subscriberCount: subscriptions.size, + }, + }); + } ws.terminate(); } } @@ -759,7 +925,14 @@ export function createMuxWebSocket(tmuxPath?: string | null): WebSocketServer | if (type === "open") { if (isWindows()) { handleWindowsPipeMessage( - msg as { id: string; type: string; projectId?: string; data?: string; cols?: number; rows?: number }, + msg as { + id: string; + type: string; + projectId?: string; + data?: string; + cols?: number; + rows?: number; + }, ws, winPipes, winPipeBuffers, @@ -841,7 +1014,13 @@ export function createMuxWebSocket(tmuxPath?: string | null): WebSocketServer | } else if (type === "resize" && "cols" in msg && "rows" in msg) { if (isWindows()) { handleWindowsPipeMessage( - msg as { id: string; type: string; projectId?: string; cols: number; rows: number }, + msg as { + id: string; + type: string; + projectId?: string; + cols: number; + rows: number; + }, ws, winPipes, winPipeBuffers, @@ -853,7 +1032,14 @@ export function createMuxWebSocket(tmuxPath?: string | null): WebSocketServer | } else if (type === "close") { if (isWindows()) { handleWindowsPipeMessage( - msg as { id: string; type: string; projectId?: string; data?: string; cols?: number; rows?: number }, + msg as { + id: string; + type: string; + projectId?: string; + data?: string; + cols?: number; + rows?: number; + }, ws, winPipes, winPipeBuffers, @@ -903,6 +1089,17 @@ export function createMuxWebSocket(tmuxPath?: string | null): WebSocketServer | } } catch (err) { console.error("[MuxServer] Failed to parse message:", err); + recordActivityEvent({ + source: "ui", + kind: "ui.terminal_protocol_error", + level: "warn", + summary: "invalid mux client message — parse failed", + data: { + errorMessage: err instanceof Error ? err.message : String(err), + remoteAddr, + subscriberCount: subscriptions.size, + }, + }); const errorMsg: ServerMessage = { ch: "system", type: "error", @@ -917,8 +1114,22 @@ export function createMuxWebSocket(tmuxPath?: string | null): WebSocketServer | /** * Handle connection close */ - ws.on("close", () => { + ws.on("close", (code, reason) => { console.log("[MuxServer] Mux connection closed"); + recordActivityEvent({ + source: "ui", + kind: "ui.terminal_disconnected", + level: "info", + summary: "mux WebSocket connection closed", + data: { + code, + reason: reason?.toString("utf8") || undefined, + connectionAgeMs: Date.now() - connectedAt, + subscriberCount: subscriptions.size, + heartbeatLost: heartbeatLostEmitted, + remoteAddr, + }, + }); clearInterval(heartbeatInterval); sessionUnsubscribe?.(); sessionUnsubscribe = null; diff --git a/packages/web/src/__tests__/webhook-route.test.ts b/packages/web/src/__tests__/webhook-route.test.ts new file mode 100644 index 000000000..0bea074e5 --- /dev/null +++ b/packages/web/src/__tests__/webhook-route.test.ts @@ -0,0 +1,328 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { + createInitialCanonicalLifecycle, + createActivitySignal, + type Session, + type SessionManager, + type OrchestratorConfig, + type PluginRegistry, + type SCM, + type LifecycleManager, +} from "@aoagents/ao-core"; + +// Activity event recording is mocked so we can assert what fires without +// touching the real SQLite layer. +const recordActivityEvent = vi.fn(); +vi.mock("@aoagents/ao-core", async (importOriginal) => { + const actual = (await importOriginal()) as Record; + return { + ...actual, + recordActivityEvent: (event: unknown) => recordActivityEvent(event), + }; +}); + +// ── Mock services + plugin registry ─────────────────────────────────── + +function makeSession(id: string, overrides: Partial = {}): Session { + const lifecycle = createInitialCanonicalLifecycle("worker", new Date()); + lifecycle.session.state = "working"; + lifecycle.session.reason = "task_in_progress"; + lifecycle.session.startedAt = lifecycle.session.lastTransitionAt; + lifecycle.runtime.state = "alive"; + lifecycle.runtime.reason = "process_running"; + return { + id, + projectId: "my-app", + status: "working", + activity: "active", + activitySignal: createActivitySignal("valid", { + activity: "active", + timestamp: new Date(), + source: "native", + }), + lifecycle, + branch: "feat/x", + issueId: null, + pr: { + number: 42, + url: "u", + title: "t", + owner: "acme", + repo: "my-app", + branch: "feat/x", + baseBranch: "main", + isDraft: false, + }, + workspacePath: null, + runtimeHandle: null, + agentInfo: null, + createdAt: new Date(), + lastActivityAt: new Date(), + metadata: {}, + ...overrides, + }; +} + +const verifyWebhook = vi.fn(); +const parseWebhook = vi.fn(); +const mockSCM: SCM = { + name: "github", + detectPR: vi.fn(), + getPRState: vi.fn(), + mergePR: vi.fn(), + closePR: vi.fn(), + getCIChecks: vi.fn(), + getCISummary: vi.fn(), + getReviews: vi.fn(), + getReviewDecision: vi.fn(), + getPendingComments: vi.fn(), + getAutomatedComments: vi.fn(), + getMergeability: vi.fn(), + verifyWebhook, + parseWebhook, +} as unknown as SCM; + +const mockRegistry: PluginRegistry = { + register: vi.fn(), + get: vi.fn(() => mockSCM) as PluginRegistry["get"], + list: vi.fn(() => []), + loadBuiltins: vi.fn(), + loadFromConfig: vi.fn(), +}; + +const mockConfig: OrchestratorConfig = { + configPath: "/tmp/agent-orchestrator.yaml", + port: 3000, + readyThresholdMs: 300_000, + defaults: { runtime: "tmux", agent: "claude-code", workspace: "worktree", notifiers: [] }, + projects: { + "my-app": { + name: "My App", + repo: "acme/my-app", + path: "/tmp/my-app", + defaultBranch: "main", + sessionPrefix: "my-app", + scm: { + plugin: "github", + webhook: { enabled: true, path: "/api/webhooks/github", maxBodyBytes: 1024 }, + }, + }, + }, + notifiers: {}, + notificationRouting: { urgent: [], action: [], warning: [], info: [] }, + reactions: {}, +}; + +const mockSessionManager = { + list: vi.fn(async () => [makeSession("s1")]), +} as unknown as SessionManager; + +const mockLifecycle = { + check: vi.fn(async () => {}), +} as unknown as LifecycleManager; + +vi.mock("@/lib/services", () => ({ + getServices: vi.fn(async () => ({ + config: mockConfig, + registry: mockRegistry, + sessionManager: mockSessionManager, + lifecycleManager: mockLifecycle, + })), +})); + +import { POST as webhookPOST } from "@/app/api/webhooks/[...slug]/route"; + +function makeWebhookRequest(opts?: { + body?: string; + contentLength?: number; + headers?: Record; +}): Request { + const body = opts?.body ?? JSON.stringify({ action: "synchronize", number: 42 }); + const headers: Record = { + "content-type": "application/json", + ...opts?.headers, + }; + if (opts?.contentLength !== undefined) { + headers["content-length"] = String(opts.contentLength); + } + return new Request("http://localhost:3000/api/webhooks/github", { + method: "POST", + headers, + body, + }); +} + +beforeEach(() => { + vi.clearAllMocks(); + recordActivityEvent.mockClear(); + verifyWebhook.mockResolvedValue({ ok: true }); + parseWebhook.mockResolvedValue({ + provider: "github", + kind: "pull_request", + action: "synchronize", + rawEventType: "pull_request", + repository: { owner: "acme", name: "my-app" }, + prNumber: 42, + branch: "feat/x", + data: {}, + }); +}); + +// ── Tests ───────────────────────────────────────────────────────────── + +describe("POST /api/webhooks/[...slug] — activity events", () => { + it("rejects unverified webhook with 401 and emits api.webhook_unverified", async () => { + verifyWebhook.mockResolvedValueOnce({ ok: false, reason: "bad signature" }); + const req = makeWebhookRequest({ + headers: { "x-hub-signature-256": "sha256=bogus", "x-forwarded-for": "203.0.113.7" }, + }); + + const res = await webhookPOST(req); + expect(res.status).toBe(401); + + expect(recordActivityEvent).toHaveBeenCalledWith( + expect.objectContaining({ + source: "api", + kind: "api.webhook_unverified", + level: "warn", + }), + ); + const call = recordActivityEvent.mock.calls[0]![0] as { + summary: string; + data: Record; + }; + // Critical: signature value must NOT be in data (or anywhere) + expect(JSON.stringify(call)).not.toContain("bogus"); + expect(call.data["slug"]).toBe("github"); + expect(call.data["remoteAddr"]).toBe("203.0.113.7"); + expect(call.data["verificationSupported"]).toBe(true); + expect(call.data["reason"]).toBe("bad signature"); + }); + + it("keeps unsupported webhook verification as 404 while recording audit context", async () => { + vi.mocked(mockRegistry.get).mockReturnValueOnce({ + ...mockSCM, + verifyWebhook: undefined, + } as unknown as SCM); + + const res = await webhookPOST(makeWebhookRequest()); + expect(res.status).toBe(404); + + expect(verifyWebhook).not.toHaveBeenCalled(); + expect(parseWebhook).not.toHaveBeenCalled(); + expect(recordActivityEvent).toHaveBeenCalledWith( + expect.objectContaining({ + source: "api", + kind: "api.webhook_unverified", + level: "warn", + summary: expect.stringContaining("verification unsupported"), + }), + ); + const call = recordActivityEvent.mock.calls[0]![0] as { + data: Record; + }; + expect(call.data["slug"]).toBe("github"); + expect(call.data["verificationSupported"]).toBe(false); + expect(call.data["unsupportedVerificationCount"]).toBe(1); + expect(call.data["reason"]).toBe("verification_unsupported"); + }); + + it("emits api.webhook_rejected when content-length exceeds maxBodyBytes (413)", async () => { + const req = makeWebhookRequest({ contentLength: 2048 }); // > 1024 max + + const res = await webhookPOST(req); + expect(res.status).toBe(413); + + expect(recordActivityEvent).toHaveBeenCalledWith( + expect.objectContaining({ + source: "api", + kind: "api.webhook_rejected", + level: "warn", + }), + ); + const call = recordActivityEvent.mock.calls[0]![0] as { data: Record }; + expect(call.data["slug"]).toBe("github"); + expect(call.data["contentLength"]).toBe(2048); + expect(call.data["maxBodyBytes"]).toBe(1024); + // No body content captured + expect(JSON.stringify(call)).not.toContain("synchronize"); + }); + + it("emits api.webhook_received with counts (not body) on 202 success", async () => { + const req = makeWebhookRequest({ + headers: { "x-forwarded-for": "192.0.2.1" }, + }); + + const res = await webhookPOST(req); + expect(res.status).toBe(202); + + expect(recordActivityEvent).toHaveBeenCalledWith( + expect.objectContaining({ + source: "api", + kind: "api.webhook_received", + }), + ); + const call = recordActivityEvent.mock.calls[0]![0] as { data: Record }; + expect(call.data["slug"]).toBe("github"); + expect(call.data["remoteAddr"]).toBe("192.0.2.1"); + expect(call.data["matchedSessions"]).toBe(1); + expect(call.data["parseErrorCount"]).toBe(0); + expect(call.data["lifecycleErrorCount"]).toBe(0); + expect(call.data["projectIds"]).toEqual(["my-app"]); + // Critical: payload body is NOT included + expect(JSON.stringify(call)).not.toContain("synchronize"); + }); + + it("emits api.webhook_received with elevated level when parse/lifecycle errors occurred", async () => { + parseWebhook.mockRejectedValueOnce(new Error("boom")); + // Verification passes so we get into the parse path + verifyWebhook.mockResolvedValueOnce({ ok: true }); + + const res = await webhookPOST(makeWebhookRequest()); + expect(res.status).toBe(202); + + const received = recordActivityEvent.mock.calls.find( + ([e]) => (e as { kind: string }).kind === "api.webhook_received", + ); + expect(received).toBeDefined(); + expect((received![0] as { level: string }).level).toBe("warn"); + expect((received![0] as { data: Record }).data["parseErrorCount"]).toBe(1); + }); + + it("emits api.webhook_failed on 500 outer crash", async () => { + // Force the list() call inside POST to throw, hitting the outer catch. + (mockSessionManager.list as ReturnType).mockRejectedValueOnce( + new Error("session manager exploded"), + ); + + const res = await webhookPOST(makeWebhookRequest()); + expect(res.status).toBe(500); + + expect(recordActivityEvent).toHaveBeenCalledWith( + expect.objectContaining({ + source: "api", + kind: "api.webhook_failed", + level: "error", + }), + ); + const call = recordActivityEvent.mock.calls[0]![0] as { data: Record }; + expect(call.data["slug"]).toBe("github"); + expect(call.data["errorMessage"]).toContain("session manager exploded"); + }); + + it("does not emit any event for 404 (unknown path)", async () => { + // Empty config so no candidates match + vi.mocked(mockRegistry.get).mockReturnValueOnce(undefined as unknown as SCM); + const req = new Request("http://localhost:3000/api/webhooks/unknown", { + method: "POST", + headers: { "content-type": "application/json" }, + body: "{}", + }); + + const res = await webhookPOST(req); + expect(res.status).toBe(404); + // No webhook events for 404 — it's a config issue, not an external signal + const kinds = recordActivityEvent.mock.calls.map(([e]) => (e as { kind: string }).kind); + expect(kinds.filter((k) => k.startsWith("api.webhook_"))).toEqual([]); + }); +}); diff --git a/packages/web/src/app/api/webhooks/[...slug]/route.ts b/packages/web/src/app/api/webhooks/[...slug]/route.ts index 9ccf878ad..08660dac8 100644 --- a/packages/web/src/app/api/webhooks/[...slug]/route.ts +++ b/packages/web/src/app/api/webhooks/[...slug]/route.ts @@ -1,4 +1,5 @@ import { NextResponse } from "next/server"; +import { recordActivityEvent } from "@aoagents/ao-core"; import { getServices } from "@/lib/services"; import { buildWebhookRequest, @@ -9,11 +10,35 @@ import { export const dynamic = "force-dynamic"; +const WEBHOOK_PATH_PREFIX = "/api/webhooks/"; + +function deriveSlug(pathname: string): string { + return pathname.startsWith(WEBHOOK_PATH_PREFIX) + ? pathname.slice(WEBHOOK_PATH_PREFIX.length) + : pathname; +} + +function deriveRemoteAddr(request: Request): string | undefined { + // Next.js does not expose the socket peer address on Request. The standard + // proxy headers are the only signal — first hop in x-forwarded-for is the + // original client. Sanitizer in recordActivityEvent does not redact IPs; + // they are intentionally retained for security audit (per issue #1656). + const xff = request.headers.get("x-forwarded-for"); + if (xff) { + const first = xff.split(",")[0]?.trim(); + if (first) return first; + } + return request.headers.get("x-real-ip") ?? undefined; +} + export async function POST(request: Request): Promise { + const pathname = new URL(request.url).pathname; + const slug = deriveSlug(pathname); + const remoteAddr = deriveRemoteAddr(request); + try { const services = await getServices(); - const path = new URL(request.url).pathname; - const candidates = findWebhookProjects(services.config, services.registry, path); + const candidates = findWebhookProjects(services.config, services.registry, pathname); if (candidates.length === 0) { return NextResponse.json( @@ -36,6 +61,19 @@ export async function POST(request: Request): Promise { Number.isFinite(contentLength) && contentLength > maxBodyBytes ) { + recordActivityEvent({ + source: "api", + kind: "api.webhook_rejected", + level: "warn", + summary: `webhook payload exceeded ${maxBodyBytes} bytes for ${slug}`, + data: { + slug, + remoteAddr, + contentLength, + maxBodyBytes, + reason: "payload_too_large", + }, + }); return NextResponse.json( { error: "Webhook payload exceeds configured maxBodyBytes" }, { status: 413 }, @@ -50,12 +88,21 @@ export async function POST(request: Request): Promise { const sessionIds = new Set(); const projectIds = new Set(); let verified = false; + let verificationSupported = false; + let unsupportedVerificationCount = 0; const errors: string[] = []; const parseErrors: string[] = []; const lifecycleErrors: string[] = []; for (const candidate of candidates) { - const verification = await candidate.scm.verifyWebhook?.(webhookRequest, candidate.project); + if (!candidate.scm.verifyWebhook) { + unsupportedVerificationCount += 1; + errors.push("Webhook verification not supported by SCM plugin"); + continue; + } + + verificationSupported = true; + const verification = await candidate.scm.verifyWebhook(webhookRequest, candidate.project); if (!verification?.ok) { if (verification?.reason) errors.push(verification.reason); continue; @@ -93,12 +140,52 @@ export async function POST(request: Request): Promise { } if (!verified) { + const unsupportedOnly = !verificationSupported && unsupportedVerificationCount > 0; + recordActivityEvent({ + source: "api", + kind: "api.webhook_unverified", + level: "warn", + summary: unsupportedOnly + ? `webhook verification unsupported for ${slug}` + : `webhook signature verification failed for ${slug}`, + data: { + slug, + remoteAddr, + candidateCount: candidates.length, + verificationSupported, + unsupportedVerificationCount, + reason: unsupportedOnly + ? "verification_unsupported" + : (errors[0] ?? "verification_failed"), + }, + }); return NextResponse.json( - { error: errors[0] ?? "Webhook verification failed", ok: false }, - { status: 401 }, + { + error: unsupportedOnly + ? "No SCM webhook configured for this path" + : (errors[0] ?? "Webhook verification failed"), + ok: false, + verificationSupported, + }, + { status: unsupportedOnly ? 404 : 401 }, ); } + recordActivityEvent({ + source: "api", + kind: "api.webhook_received", + level: parseErrors.length > 0 || lifecycleErrors.length > 0 ? "warn" : "info", + summary: `webhook accepted for ${slug}: ${sessionIds.size} session(s) matched`, + data: { + slug, + remoteAddr, + projectIds: [...projectIds], + matchedSessions: sessionIds.size, + parseErrorCount: parseErrors.length, + lifecycleErrorCount: lifecycleErrors.length, + }, + }); + return NextResponse.json( { ok: true, @@ -111,6 +198,17 @@ export async function POST(request: Request): Promise { { status: 202 }, ); } catch (err) { + recordActivityEvent({ + source: "api", + kind: "api.webhook_failed", + level: "error", + summary: `webhook pipeline crashed for ${slug}`, + data: { + slug, + remoteAddr, + errorMessage: err instanceof Error ? err.message : String(err), + }, + }); return NextResponse.json( { error: err instanceof Error ? err.message : "Failed to process SCM webhook" }, { status: 500 }, diff --git a/packages/web/src/lib/scm-webhooks.ts b/packages/web/src/lib/scm-webhooks.ts index 63f8d301a..360a8ff6c 100644 --- a/packages/web/src/lib/scm-webhooks.ts +++ b/packages/web/src/lib/scm-webhooks.ts @@ -40,7 +40,7 @@ export function findWebhookProjects( const webhookPath = getProjectWebhookPath(project); if (!webhookPath || webhookPath !== pathname) return []; const scm = registry.get("scm", project.scm.plugin); - if (!scm?.parseWebhook || !scm.verifyWebhook) return []; + if (!scm?.parseWebhook) return []; return [{ projectId, project, scm }]; }); }