* 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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> Co-authored-by: whoisasx <adil.business4064@gmail.com>
This commit is contained in:
Dhruv Sharma
GitHub
parent
ff0c3b741d
commit
73bed33c2e
|
|
@ -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.
|
||||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>;
|
||||
return {
|
||||
...actual,
|
||||
recordActivityEvent: (event: unknown) => recordActivityEvent(event),
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("node:child_process", async (importOriginal) => {
|
||||
const actual = (await importOriginal()) as Record<string, unknown>;
|
||||
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<typeof vi.fn>;
|
||||
onExit: ReturnType<typeof vi.fn>;
|
||||
write: ReturnType<typeof vi.fn>;
|
||||
resize: ReturnType<typeof vi.fn>;
|
||||
kill: ReturnType<typeof vi.fn>;
|
||||
emitData: (data: string) => void;
|
||||
emitExit: (exitCode: number) => Promise<void>;
|
||||
};
|
||||
|
||||
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<string, unknown> };
|
||||
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<string, unknown> };
|
||||
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<typeof makeFakeRequest>[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<string, unknown> } | undefined {
|
||||
const found = recordActivityEvent.mock.calls.find(
|
||||
([e]) => (e as { kind: string }).kind === kind,
|
||||
);
|
||||
return found?.[0] as { data: Record<string, unknown> } | 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<string, Socket>();
|
||||
const winPipeBuffers = new Map<string, Buffer>();
|
||||
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<string, unknown>; sessionId?: string }> {
|
||||
return recordActivityEvent.mock.calls
|
||||
.map(([e]) => e as { kind: string; data: Record<string, unknown>; 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<string, unknown>; sessionId?: string }> {
|
||||
return recordActivityEvent.mock.calls
|
||||
.map(([e]) => e as { kind: string; data: Record<string, unknown>; 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 () => {
|
||||
|
|
|
|||
|
|
@ -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<typeof setInterval> | 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<string, unknown>): 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<Socket>();
|
||||
|
||||
/** 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<string, unknown>,
|
||||
): 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<string, () => void>();
|
||||
// Windows: named pipe sockets keyed by session ID
|
||||
const winPipes = new Map<string, ReturnType<typeof netConnect>>();
|
||||
|
|
@ -716,6 +865,7 @@ export function createMuxWebSocket(tmuxPath?: string | null): WebSocketServer |
|
|||
const winPipeBuffers = new Map<string, Buffer>();
|
||||
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;
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>;
|
||||
return {
|
||||
...actual,
|
||||
recordActivityEvent: (event: unknown) => recordActivityEvent(event),
|
||||
};
|
||||
});
|
||||
|
||||
// ── Mock services + plugin registry ───────────────────────────────────
|
||||
|
||||
function makeSession(id: string, overrides: Partial<Session> = {}): 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<string, string>;
|
||||
}): Request {
|
||||
const body = opts?.body ?? JSON.stringify({ action: "synchronize", number: 42 });
|
||||
const headers: Record<string, string> = {
|
||||
"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<string, unknown>;
|
||||
};
|
||||
// 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<string, unknown>;
|
||||
};
|
||||
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<string, unknown> };
|
||||
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<string, unknown> };
|
||||
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<string, unknown> }).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<typeof vi.fn>).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<string, unknown> };
|
||||
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([]);
|
||||
});
|
||||
});
|
||||
|
|
@ -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<Response> {
|
||||
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<Response> {
|
|||
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<Response> {
|
|||
const sessionIds = new Set<string>();
|
||||
const projectIds = new Set<string>();
|
||||
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<Response> {
|
|||
}
|
||||
|
||||
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<Response> {
|
|||
{ 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 },
|
||||
|
|
|
|||
|
|
@ -40,7 +40,7 @@ export function findWebhookProjects(
|
|||
const webhookPath = getProjectWebhookPath(project);
|
||||
if (!webhookPath || webhookPath !== pathname) return [];
|
||||
const scm = registry.get<SCM>("scm", project.scm.plugin);
|
||||
if (!scm?.parseWebhook || !scm.verifyWebhook) return [];
|
||||
if (!scm?.parseWebhook) return [];
|
||||
return [{ projectId, project, scm }];
|
||||
});
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue