diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md new file mode 100644 index 000000000..04eb6fa6a --- /dev/null +++ b/docs/ARCHITECTURE.md @@ -0,0 +1,204 @@ +# Agent Orchestrator — Technical Architecture + +This document explains how the various parts of the Agent Orchestrator communicate with each other: where HTTP is used, where WebSocket is used, and what each carries. + +--- + +## System Overview + +```mermaid +graph TB + subgraph Browser["Browser / Dashboard"] + UI["React Dashboard\n(Next.js App Router)"] + XTerm["xterm.js\nTerminal UI"] + end + + subgraph NextJS["Next.js Server — :3000 (single process)"] + subgraph HTTPAPI["① HTTP REST /api/* — request / response"] + Sessions["GET /api/sessions\nGET /api/sessions/:id\nPOST /api/sessions/:id/message\nPOST /api/sessions/:id/restore\nPOST /api/sessions/:id/kill\nPOST /api/spawn\nGET /api/projects GET /api/agents\nGET /api/issues POST /api/prs/:id/merge\nPOST /api/webhooks/**"] + Patches["GET /api/sessions/patches\n(lightweight: id, status, activity,\nattentionLevel, lastActivityAt)"] + end + SessionMgr["Session Manager\n(reads flat files in\n~/.agent-orchestrator/)"] + end + + subgraph MuxServer["② WebSocket Server — :14801 (separate Node process)"] + MuxWS["ws://host:14801/mux\nMultiplexed — two sub-channels\nover one connection"] + TermMgr["TerminalManager\n(node-pty → tmux PTY)"] + Broadcaster["SessionBroadcaster\n(setInterval every 3s →\nGET /api/sessions/patches)"] + end + + subgraph Agents["AI Agents (one tmux window each)"] + ClaudeCode["Claude Code"] + Codex["Codex"] + Aider["Aider"] + OpenCode["OpenCode"] + end + + subgraph External["External Services"] + GitHub["GitHub API"] + Linear["Linear API"] + end + + %% ① HTTP — user actions & data fetching + UI -- "① HTTP GET/POST\n(on demand: load sessions,\nsend message, spawn, merge PR…)" --> Sessions + + %% ② WebSocket terminal sub-channel + XTerm -- "② WS sub-channel 'terminal'\nkeystrokes → {ch:terminal, type:data}\noutput ← {ch:terminal, type:data}" --> MuxWS + MuxWS --> TermMgr + TermMgr -- "PTY read/write" --> ClaudeCode + TermMgr -- "PTY read/write" --> Codex + TermMgr -- "PTY read/write" --> Aider + TermMgr -- "PTY read/write" --> OpenCode + + %% ② WebSocket sessions sub-channel + Broadcaster -- "HTTP GET /api/sessions/patches\nevery 3s" --> Patches + Patches -- "reads" --> SessionMgr + Broadcaster -- "② WS sub-channel 'sessions'\n{ch:sessions, type:snapshot,\n sessions:[{id,status,activity,\n attentionLevel,lastActivityAt}]}" --> MuxWS + MuxWS -- "session patches\n→ useSessionEvents()\n→ useMuxSessionActivity()" --> UI + + %% Mux auto-recovery calls back to Next.js + TermMgr -- "① HTTP POST /api/sessions/:id/restore\n(auto-recovery when tmux dies)" --> Sessions + + %% External + Sessions -- "REST calls" --> GitHub + Sessions --> Linear + GitHub -- "POST /api/webhooks/**" --> Sessions +``` + +--- + +## Communication Channels + +### 1. HTTP / REST — `/api/*` on port 3000 + +Used for all request-response interactions. The browser calls these on demand; the CLI and the WebSocket server also use them. + +| Endpoint | Method | Purpose | +|----------|--------|---------| +| `/api/sessions` | GET | List all sessions (with PR / issue metadata) | +| `/api/sessions/light` | GET | Lightweight session list (minimal fields) | +| `/api/sessions/patches` | GET | Ultra-light patches (id, status, activity, attentionLevel) — polled by the WS server every 3s | +| `/api/sessions/:id` | GET | Full session detail | +| `/api/sessions/:id/message` | POST | Send a message/command to a live agent | +| `/api/sessions/:id/restore` | POST | Respawn a terminated session | +| `/api/sessions/:id/kill` | POST | Terminate a running session | +| `/api/sessions/:id/files` | GET | Browse workspace files | +| `/api/sessions/:id/diff/**` | GET | File diff view | +| `/api/sessions/:id/sub-sessions` | GET / POST | List / create sub-sessions (forked agents) | +| `/api/spawn` | POST | Spawn a new agent session | +| `/api/projects` | GET | List configured projects | +| `/api/agents` | GET | List registered agent plugins | +| `/api/issues` | GET | Fetch backlog issues | +| `/api/backlog` | GET | Backlog summary | +| `/api/prs/:id/merge` | POST | Merge a PR | +| `/api/observability` | GET | Health and metrics summary | +| `/api/verify` | POST | Verify environment setup | +| `/api/setup-labels` | POST | Set up GitHub labels | +| `/api/webhooks/**` | POST | Inbound webhooks from GitHub / GitLab | + +--- + +### 2. WebSocket (Multiplexed) — `ws://localhost:14801/mux` + +A **bidirectional multiplexed channel** on a separate Node.js process. A single WebSocket connection carries two independent sub-channels: + +- **`terminal` channel** — raw PTY I/O for xterm.js +- **`sessions` channel** — real-time session status patches (fed by `SessionBroadcaster` polling `/api/sessions/patches` every 3s) + +```mermaid +sequenceDiagram + participant XTerm as xterm.js + participant MuxClient as MuxProvider (browser) + participant MuxWS as WS Server :14801/mux + participant PTY as node-pty (tmux) + participant Next as Next.js :3000 + + MuxClient->>MuxWS: connect ws://localhost:14801/mux + + Note over MuxClient,MuxWS: Open a terminal + MuxClient->>MuxWS: {ch:"terminal", id:"sess-1", type:"open"} + MuxWS->>PTY: attach tmux PTY + MuxWS-->>MuxClient: {ch:"terminal", id:"sess-1", type:"opened"} + + Note over MuxClient,MuxWS: Terminal I/O + XTerm->>MuxClient: user keystrokes + MuxClient->>MuxWS: {ch:"terminal", id:"sess-1", type:"data", data:"ls\r"} + MuxWS->>PTY: write to PTY + PTY-->>MuxWS: output bytes + MuxWS-->>MuxClient: {ch:"terminal", id:"sess-1", type:"data", data:"file1 file2\r\n"} + MuxClient-->>XTerm: render output + + Note over MuxWS,Next: Session patches (every 3s) + MuxWS->>Next: GET /api/sessions/patches + Next-->>MuxWS: [{id, status, activity, attentionLevel, lastActivityAt}] + MuxWS-->>MuxClient: {ch:"sessions", type:"snapshot", sessions:[...]} + MuxClient-->>MuxClient: useSessionEvents() + useMuxSessionActivity() update React state + + Note over MuxWS,Next: Auto-recovery (session dead) + MuxWS->>Next: POST /api/sessions/sess-1/restore + Next-->>MuxWS: 200 OK + MuxWS->>PTY: reattach to new tmux session +``` + +**Message types:** + +| Direction | Channel | Type | Payload | +|-----------|---------|------|---------| +| Client→Server | `terminal` | `open` | `{ id }` | +| Client→Server | `terminal` | `data` | `{ id, data: string }` | +| Client→Server | `terminal` | `resize` | `{ id, cols, rows }` | +| Client→Server | `terminal` | `close` | `{ id }` | +| Client→Server | `subscribe` | — | `{ topics: ["sessions"] }` | +| Client→Server | `system` | `ping` | — | +| Server→Client | `terminal` | `opened` | `{ id }` | +| Server→Client | `terminal` | `data` | `{ id, data: string }` | +| Server→Client | `terminal` | `exited` | `{ id, code }` | +| Server→Client | `terminal` | `error` | `{ id, message }` | +| Server→Client | `sessions` | `snapshot` | `{ sessions: SessionPatch[] }` | +| Server→Client | `system` | `pong` | — | + +--- + +## Process Map + +```mermaid +graph LR + subgraph Host + CLI["ao CLI\n(packages/cli)"] + Next["Next.js\npackages/web — :3000"] + MuxSrv["Terminal WS Server\npackages/web/server — :14801"] + end + + subgraph Storage["Flat-file Storage"] + Sessions2["~/.agent-orchestrator/\n{hash}-{project}/\n sessions/{id} ← key-value\n worktrees/{id}/\n archive/{id}_{ts}/"] + end + + CLI -- "pnpm ao start\nspawns both servers" --> Next + CLI -- "spawns" --> MuxSrv + Next -- "reads / writes" --> Sessions2 + MuxSrv -- "GET /api/sessions/patches (every 3s)" --> Next + MuxSrv -- "POST /api/sessions/:id/restore (recovery)" --> Next +``` + +The CLI (`ao start`) forks two long-running processes: +- **Next.js** on `:3000` — serves the dashboard and all REST routes +- **Terminal WS server** on `:14801` — handles multiplexed WebSocket + PTY management + session patch polling + +Both processes share no in-memory state; coordination happens through flat files in `~/.agent-orchestrator/` and HTTP calls from the WS server to Next.js. + +--- + +## Data Flow Summary + +| Scenario | Protocol | Path | +|----------|----------|------| +| Load dashboard | HTTP GET | Browser → `:3000/` (SSR page) | +| List sessions | HTTP GET | Browser → `:3000/api/sessions` | +| Spawn new agent | HTTP POST | Browser → `:3000/api/spawn` | +| Send message to agent | HTTP POST | Browser → `:3000/api/sessions/:id/message` | +| Real-time session status | WebSocket | Browser ← `:14801/mux` `sessions` sub-channel (pushed every 3s) | +| Terminal output / input | WebSocket | Browser ↔ `:14801/mux` `terminal` sub-channel (bidirectional) | +| WS server fetches patches | HTTP GET | `:14801` → `:3000/api/sessions/patches` (every 3s) | +| WS server restores session | HTTP POST | `:14801` → `:3000/api/sessions/:id/restore` | +| GitHub notifies of CI / PR | HTTP POST | GitHub → `:3000/api/webhooks/github` | +| CLI queries sessions | HTTP GET | `ao` CLI → `:3000/api/sessions` | diff --git a/packages/cli/__tests__/commands/project.test.ts b/packages/cli/__tests__/commands/project.test.ts index ebf8c2291..264b64f03 100644 --- a/packages/cli/__tests__/commands/project.test.ts +++ b/packages/cli/__tests__/commands/project.test.ts @@ -41,7 +41,7 @@ vi.mock("../../src/lib/prompts.js", () => ({ promptConfirm: vi.fn(async () => true), })); -import { registerProject_cmd } from "../../src/commands/project.js"; +import { registerProjectCommand } from "../../src/commands/project.js"; let program: Command; let logSpy: ReturnType; @@ -52,7 +52,7 @@ beforeEach(() => { vi.clearAllMocks(); program = new Command(); program.exitOverride(); - registerProject_cmd(program); + registerProjectCommand(program); logSpy = vi.spyOn(console, "log").mockImplementation(() => {}); errorSpy = vi.spyOn(console, "error").mockImplementation(() => {}); _exitSpy = vi.spyOn(process, "exit").mockImplementation(((code?: number) => { diff --git a/packages/cli/__tests__/program.test.ts b/packages/cli/__tests__/program.test.ts index 656ac51be..e854a2b03 100644 --- a/packages/cli/__tests__/program.test.ts +++ b/packages/cli/__tests__/program.test.ts @@ -6,4 +6,8 @@ describe("createProgram", () => { it("uses the CLI package version", () => { expect(createProgram().version()).toBe(packageJson.version); }); + + it("registers the project command", () => { + expect(createProgram().commands.some((command) => command.name() === "project")).toBe(true); + }); }); diff --git a/packages/cli/src/commands/project.ts b/packages/cli/src/commands/project.ts index b052fa5ac..5dfc35c21 100644 --- a/packages/cli/src/commands/project.ts +++ b/packages/cli/src/commands/project.ts @@ -29,7 +29,7 @@ function assertPortfolioEnabled(): void { process.exit(1); } -export function registerProject_cmd(program: Command): void { +export function registerProjectCommand(program: Command): void { const project = program.command("project").description("Manage portfolio projects"); // ao project ls diff --git a/packages/cli/src/program.ts b/packages/cli/src/program.ts index 0680a8971..f5110afb9 100644 --- a/packages/cli/src/program.ts +++ b/packages/cli/src/program.ts @@ -14,6 +14,7 @@ import { registerDoctor } from "./commands/doctor.js"; import { registerUpdate } from "./commands/update.js"; import { registerSetup } from "./commands/setup.js"; import { registerPlugin } from "./commands/plugin.js"; +import { registerProjectCommand } from "./commands/project.js"; import { registerMigrateStorage } from "./commands/migrate-storage.js"; import { registerCompletion } from "./commands/completion.js"; import { getConfigInstruction } from "./lib/config-instruction.js"; @@ -45,6 +46,7 @@ export function createProgram(): Command { registerUpdate(program); registerSetup(program); registerPlugin(program); + registerProjectCommand(program); registerMigrateStorage(program); registerCompletion(program); diff --git a/packages/core/src/__tests__/lifecycle-manager.test.ts b/packages/core/src/__tests__/lifecycle-manager.test.ts index 726011c31..9f2ff06c2 100644 --- a/packages/core/src/__tests__/lifecycle-manager.test.ts +++ b/packages/core/src/__tests__/lifecycle-manager.test.ts @@ -2689,6 +2689,558 @@ describe("reactions", () => { ); expect(alertsNotifier.notify).not.toHaveBeenCalled(); }); + + it("CI failure tracker survives status oscillation and escalates after retries", async () => { + const notifier = createMockNotifier(); + + config.reactions = { + "ci-failed": { + auto: true, + action: "send-to-agent", + message: "CI is failing. Fix it.", + retries: 2, + escalateAfter: 2, + }, + }; + + const batchMock = mockBatchEnrichment({ ciStatus: "failing" }); + const mockSCM = createMockSCM({ + enrichSessionsPRBatch: batchMock, + }); + + const registry: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string, name: string) => { + if (slot === "runtime") return plugins.runtime; + if (slot === "agent") return plugins.agent; + if (slot === "scm") return mockSCM; + if (slot === "notifier" && name === "desktop") return notifier; + return null; + }), + }; + + vi.mocked(mockSessionManager.send).mockResolvedValue(undefined); + + const lm = setupCheck("app-1", { + session: makeSession({ status: "pr_open", pr: makePR() }), + registry, + }); + + // Oscillation 1: pr_open → ci_failed (attempt 1 — send to agent) + await lm.check("app-1"); + expect(lm.getStates().get("app-1")).toBe("ci_failed"); + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + vi.mocked(mockSessionManager.send).mockClear(); + + // CI starts passing → ci_failed → pr_open (tracker survives) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "passing" }), + ); + await lm.check("app-1"); + expect(lm.getStates().get("app-1")).toBe("pr_open"); + expect(mockSessionManager.send).not.toHaveBeenCalled(); + + // Oscillation 2: pr_open → ci_failed (attempt 2 — send to agent) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "failing" }), + ); + await lm.check("app-1"); + expect(lm.getStates().get("app-1")).toBe("ci_failed"); + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + vi.mocked(mockSessionManager.send).mockClear(); + + // CI passes again + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "passing" }), + ); + await lm.check("app-1"); + + // Oscillation 3: pr_open → ci_failed (attempt 3 > retries:2 — escalate) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "failing" }), + ); + vi.mocked(notifier.notify).mockClear(); + await lm.check("app-1"); + + // Should NOT send to agent — should escalate to human + expect(mockSessionManager.send).not.toHaveBeenCalled(); + expect(notifier.notify).toHaveBeenCalledWith( + expect.objectContaining({ type: "reaction.escalated" }), + ); + + // After escalation, tracker is marked escalated — needs 2 stable passing polls to clear + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "passing" }), + ); + await lm.check("app-1"); // stableCount = 1 + await lm.check("app-1"); // stableCount = 2 → clearReactionTracker + vi.mocked(mockSessionManager.send).mockClear(); + vi.mocked(notifier.notify).mockClear(); + + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "failing" }), + ); + await lm.check("app-1"); + + // Fresh budget — sends to agent (attempt 1 again), not escalate + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + expect(notifier.notify).not.toHaveBeenCalledWith( + expect.objectContaining({ type: "reaction.escalated" }), + ); + }); + + it("merge conflict tracker resets on resolve — recurrence gets fresh budget", async () => { + const notifier = createMockNotifier(); + + config.reactions = { + "merge-conflicts": { + auto: true, + action: "send-to-agent", + message: "Resolve merge conflicts.", + retries: 1, + escalateAfter: 1, + }, + }; + + const batchMock = mockBatchEnrichment({ hasConflicts: true }); + const mockSCM = createMockSCM({ + enrichSessionsPRBatch: batchMock, + }); + + const registry: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string, name: string) => { + if (slot === "runtime") return plugins.runtime; + if (slot === "agent") return plugins.agent; + if (slot === "scm") return mockSCM; + if (slot === "notifier" && name === "desktop") return notifier; + return null; + }), + }; + + vi.mocked(mockSessionManager.send).mockResolvedValue(undefined); + + const lm = setupCheck("app-1", { + session: makeSession({ status: "pr_open", pr: makePR() }), + registry, + }); + + // First conflict — dispatched to agent (attempt 1) + await lm.check("app-1"); + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + vi.mocked(mockSessionManager.send).mockClear(); + + // Conflicts resolve — tracker clears (incident boundary) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ hasConflicts: false }), + ); + await lm.check("app-1"); + const metadata = readMetadataRaw(env.sessionsDir, "app-1"); + expect(metadata?.["lastMergeConflictDispatched"]).toBeFalsy(); + + // Conflicts recur — fresh tracker (attempt 1 again, not 2) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ hasConflicts: true }), + ); + vi.mocked(notifier.notify).mockClear(); + await lm.check("app-1"); + + // Fresh budget — sends to agent (attempt 1), not escalate + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + expect(notifier.notify).not.toHaveBeenCalledWith( + expect.objectContaining({ type: "reaction.escalated" }), + ); + }); + + it("non-persistent reaction keys still clear on status exit", async () => { + config.reactions = { + "changes-requested": { + auto: true, + action: "send-to-agent", + message: "Address review comments.", + retries: 1, + escalateAfter: 1, + }, + }; + + const batchMock = mockBatchEnrichment({ reviewDecision: "changes_requested" }); + const mockSCM = createMockSCM({ + enrichSessionsPRBatch: batchMock, + }); + const registry = createMockRegistry({ + runtime: plugins.runtime, + agent: plugins.agent, + scm: mockSCM, + }); + + vi.mocked(mockSessionManager.send).mockResolvedValue(undefined); + + const lm = setupCheck("app-1", { + session: makeSession({ status: "pr_open", pr: makePR() }), + registry, + }); + + // Transition to changes_requested (attempt 1 — send to agent) + await lm.check("app-1"); + expect(lm.getStates().get("app-1")).toBe("changes_requested"); + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + vi.mocked(mockSessionManager.send).mockClear(); + + // Transition away — tracker clears (non-persistent key) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "passing", reviewDecision: "none" }), + ); + await lm.check("app-1"); + + // Transition back — fresh tracker (attempt 1 again, NOT 2) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ reviewDecision: "changes_requested" }), + ); + await lm.check("app-1"); + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + + // Transition away and back again — still attempt 1, not escalating + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "passing", reviewDecision: "none" }), + ); + await lm.check("app-1"); + vi.mocked(mockSessionManager.send).mockClear(); + + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ reviewDecision: "changes_requested" }), + ); + await lm.check("app-1"); + // With retries:1, attempt 2 would escalate. But tracker was cleared, + // so this is attempt 1 again — still sends to agent. + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + }); + + it("CI escalation silences further dispatches — clears only after stable CI pass", async () => { + // retries:1 → attempt 1 sends, attempt 2 escalates. + // After escalation: tracker.escalated=true silences subsequent oscillations. + // Tracker clears only after 2 consecutive passing polls; then next failure gets fresh budget. + const notifier = createMockNotifier(); + + config.reactions = { + "ci-failed": { + auto: true, + action: "send-to-agent", + message: "CI is failing. Fix it.", + retries: 1, + escalateAfter: 1, + }, + }; + + const batchMock = mockBatchEnrichment({ ciStatus: "failing" }); + const mockSCM = createMockSCM({ + enrichSessionsPRBatch: batchMock, + }); + + const registry: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string, name: string) => { + if (slot === "runtime") return plugins.runtime; + if (slot === "agent") return plugins.agent; + if (slot === "scm") return mockSCM; + if (slot === "notifier" && name === "desktop") return notifier; + return null; + }), + }; + + vi.mocked(mockSessionManager.send).mockResolvedValue(undefined); + + const lm = setupCheck("app-1", { + session: makeSession({ status: "pr_open", pr: makePR() }), + registry, + }); + + // Oscillation 1: pr_open → ci_failed (attempt 1 — send to agent) + await lm.check("app-1"); + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + vi.mocked(mockSessionManager.send).mockClear(); + + // CI passes briefly (ci_failed → pr_open, stableCount = 1) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "passing" }), + ); + await lm.check("app-1"); + + // Oscillation 2: pr_open → ci_failed (attempt 2 > retries:1 — escalate, tracker.escalated = true) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "failing" }), + ); + await lm.check("app-1"); + expect(mockSessionManager.send).not.toHaveBeenCalled(); + expect(notifier.notify).toHaveBeenCalledWith( + expect.objectContaining({ type: "reaction.escalated" }), + ); + vi.mocked(notifier.notify).mockClear(); + + // CI passes once (stableCount = 1 — not enough to clear yet) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "passing" }), + ); + await lm.check("app-1"); + + // Oscillation 3: pr_open → ci_failed — escalated tracker short-circuits, NO dispatch + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "failing" }), + ); + await lm.check("app-1"); + expect(mockSessionManager.send).not.toHaveBeenCalled(); + expect(notifier.notify).not.toHaveBeenCalledWith( + expect.objectContaining({ type: "reaction.escalated" }), + ); + vi.mocked(mockSessionManager.send).mockClear(); + vi.mocked(notifier.notify).mockClear(); + + // CI passes twice stably (stableCount → 1 → 2 → tracker cleared) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "passing" }), + ); + await lm.check("app-1"); // stableCount = 1 + await lm.check("app-1"); // stableCount = 2 → clearReactionTracker + + // Oscillation 4: pr_open → ci_failed — fresh budget: attempt 1, sends (not escalate) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "failing" }), + ); + await lm.check("app-1"); + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + expect(notifier.notify).not.toHaveBeenCalledWith( + expect.objectContaining({ type: "reaction.escalated" }), + ); + }); + + it("single passing poll does not reset escalated ci-failed tracker", async () => { + // Regression: one passing poll must NOT clear the tracker. Requires 2 consecutive passing polls. + const notifier = createMockNotifier(); + + config.reactions = { + "ci-failed": { + auto: true, + action: "send-to-agent", + message: "CI is failing.", + retries: 1, + escalateAfter: 1, + }, + }; + + const batchMock = mockBatchEnrichment({ ciStatus: "failing" }); + const mockSCM = createMockSCM({ enrichSessionsPRBatch: batchMock }); + + const registry: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string, name: string) => { + if (slot === "runtime") return plugins.runtime; + if (slot === "agent") return plugins.agent; + if (slot === "scm") return mockSCM; + if (slot === "notifier" && name === "desktop") return notifier; + return null; + }), + }; + + vi.mocked(mockSessionManager.send).mockResolvedValue(undefined); + + const lm = setupCheck("app-1", { + session: makeSession({ status: "pr_open", pr: makePR() }), + registry, + }); + + // Reach escalated state: attempt 1 → send, attempt 2 → escalate + await lm.check("app-1"); // pr_open → ci_failed: attempt 1, send + vi.mocked(mockSessionManager.send).mockClear(); + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "passing" })); + await lm.check("app-1"); // ci_failed → pr_open + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "failing" })); + await lm.check("app-1"); // pr_open → ci_failed: attempt 2 → escalate + expect(notifier.notify).toHaveBeenCalledWith(expect.objectContaining({ type: "reaction.escalated" })); + vi.mocked(notifier.notify).mockClear(); + + // ONE passing poll (stableCount = 1, not enough) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "passing" })); + await lm.check("app-1"); + + // Next CI failure: tracker still escalated → short-circuit + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "failing" })); + await lm.check("app-1"); + expect(mockSessionManager.send).not.toHaveBeenCalled(); + expect(notifier.notify).not.toHaveBeenCalledWith(expect.objectContaining({ type: "reaction.escalated" })); + }); + + it("pending CI does not count toward ci-failed tracker resolution", async () => { + // Regression: real CI goes failing → pending (new run started) → failing. + // "pending" must NOT count as resolution — only "passing" does. + // Without this, 2 pending polls between failures wipe the tracker and we're back at #1409. + config.reactions = { + "ci-failed": { + auto: true, + action: "send-to-agent", + message: "CI is failing.", + retries: 2, + }, + }; + + const batchMock = mockBatchEnrichment({ ciStatus: "failing" }); + const mockSCM = createMockSCM({ enrichSessionsPRBatch: batchMock }); + + const registry: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string) => { + if (slot === "runtime") return plugins.runtime; + if (slot === "agent") return plugins.agent; + if (slot === "scm") return mockSCM; + return null; + }), + }; + + vi.mocked(mockSessionManager.send).mockResolvedValue(undefined); + + const lm = setupCheck("app-1", { + session: makeSession({ status: "pr_open", pr: makePR() }), + registry, + }); + + // CI failing: pr_open → ci_failed, attempt 1 — send + await lm.check("app-1"); + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + vi.mocked(mockSessionManager.send).mockClear(); + + // CI goes pending (agent pushed a fix, new run started): ci_failed → pr_open + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "pending" })); + await lm.check("app-1"); // stableCount must NOT increment + await lm.check("app-1"); // two pending polls — must NOT clear tracker + + // CI fails again (run completed failing): pr_open → ci_failed, attempt 2 — send + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "failing" })); + await lm.check("app-1"); + // If pending had wrongly cleared the tracker, this would be attempt 1 (fresh), not attempt 2. + // Attempt 2 ≤ retries:2 → sends to agent (not escalates) + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + vi.mocked(mockSessionManager.send).mockClear(); + + // CI goes pending again, then failing — attempt 3 > retries:2 → escalate + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "pending" })); + await lm.check("app-1"); // pending: no clear + await lm.check("app-1"); // pending: no clear + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "failing" })); + await lm.check("app-1"); + expect(mockSessionManager.send).not.toHaveBeenCalled(); // escalated, not sent to agent + }); + + it("only passing CI resets ci-failed tracker — pending mid-run does not interfere", async () => { + // Complementary to previous: failing → pending(many) → passing(2) → failing SHOULD clear. + // Pending during CI run doesn't block resolution; only the final passing state matters. + const notifier = createMockNotifier(); + + config.reactions = { + "ci-failed": { + auto: true, + action: "send-to-agent", + message: "CI is failing.", + retries: 1, + escalateAfter: 1, + }, + }; + + const batchMock = mockBatchEnrichment({ ciStatus: "failing" }); + const mockSCM = createMockSCM({ enrichSessionsPRBatch: batchMock }); + + const registry: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string, name: string) => { + if (slot === "runtime") return plugins.runtime; + if (slot === "agent") return plugins.agent; + if (slot === "scm") return mockSCM; + if (slot === "notifier" && name === "desktop") return notifier; + return null; + }), + }; + + vi.mocked(mockSessionManager.send).mockResolvedValue(undefined); + + const lm = setupCheck("app-1", { + session: makeSession({ status: "pr_open", pr: makePR() }), + registry, + }); + + // Reach escalated state: attempt 1 → send, attempt 2 → escalate + await lm.check("app-1"); // attempt 1, send + vi.mocked(mockSessionManager.send).mockClear(); + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "passing" })); + await lm.check("app-1"); // ci_failed → pr_open + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "failing" })); + await lm.check("app-1"); // attempt 2 → escalate + vi.mocked(notifier.notify).mockClear(); + + // CI goes pending (new run) — stableCount stays 0, does NOT progress toward resolution + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "pending" })); + await lm.check("app-1"); + await lm.check("app-1"); + await lm.check("app-1"); // many pending polls — stableCount never reaches threshold + + // CI finally passes (2 stable polls) → tracker cleared + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "passing" })); + await lm.check("app-1"); // stableCount = 1 + await lm.check("app-1"); // stableCount = 2 → clearReactionTracker + + // Next CI failure gets fresh budget: attempt 1, send + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "failing" })); + await lm.check("app-1"); + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + expect(notifier.notify).not.toHaveBeenCalledWith( + expect.objectContaining({ type: "reaction.escalated" }), + ); + }); + + it("merge-conflict notify action preserves warning priority", async () => { + const notifier = createMockNotifier(); + + config.reactions = { + "merge-conflicts": { + auto: true, + action: "notify", + }, + }; + config.notificationRouting = { + urgent: ["desktop"], + action: ["desktop"], + warning: ["desktop"], + info: [], + }; + + const batchMock = mockBatchEnrichment({ hasConflicts: true }); + const mockSCM = createMockSCM({ + enrichSessionsPRBatch: batchMock, + }); + + const registry: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string, name: string) => { + if (slot === "runtime") return plugins.runtime; + if (slot === "agent") return plugins.agent; + if (slot === "scm") return mockSCM; + if (slot === "notifier" && name === "desktop") return notifier; + return null; + }), + }; + + const lm = setupCheck("app-1", { + session: makeSession({ status: "pr_open", pr: makePR() }), + registry, + }); + + await lm.check("app-1"); + + // With info routing empty and warning routing to desktop, + // notify should fire at "warning" priority (not "info") + expect(notifier.notify).toHaveBeenCalledWith( + expect.objectContaining({ + type: "reaction.triggered", + priority: "warning", + }), + ); + }); }); describe("pollAll terminal status accounting", () => { diff --git a/packages/core/src/lifecycle-manager.ts b/packages/core/src/lifecycle-manager.ts index 68b9e4178..344eb5052 100644 --- a/packages/core/src/lifecycle-manager.ts +++ b/packages/core/src/lifecycle-manager.ts @@ -92,6 +92,19 @@ function parseDuration(str: string): number { } } +/** Reaction keys for conditions that can oscillate (e.g. CI failing→pending→failing). + * Their trackers survive status exit so the escalation budget accumulates + * across oscillations instead of resetting to zero each time. + * Note: "merge-conflicts" is NOT here — statusToEventType never emits + * "merge.conflicts", so the transition handler at line ~1892 can't reach it. + * Merge-conflict tracker lifecycle is managed in maybeDispatchMergeConflicts. */ +const PERSISTENT_REACTION_KEYS = new Set(["ci-failed"]); + +/** Number of consecutive CI-passing polls required before the ci-failed tracker + * (including its escalated flag) is cleared, allowing a fresh budget for the + * next real CI failure incident. */ +const CI_PASSING_STABLE_THRESHOLD = 2; + type WorkspaceBranchProbe = | { kind: "branch"; branch: string } | { kind: "detached" } @@ -385,6 +398,9 @@ export interface LifecycleManagerDeps { interface ReactionTracker { attempts: number; firstTriggered: Date; + /** True after this reaction has escalated. Short-circuits further dispatches + * until the underlying condition resolves and the tracker is explicitly cleared. */ + escalated?: boolean; } /** Create a LifecycleManager instance. */ @@ -1169,6 +1185,11 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan reactionTrackers.set(trackerKey, tracker); } + // Already escalated — wait for the condition to resolve before resuming. + if (tracker.escalated) { + return { reactionType: reactionKey, success: true, action: "escalated", escalated: true }; + } + // Increment attempts before checking escalation tracker.attempts++; @@ -1201,6 +1222,11 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan data: { reactionKey, attempts: tracker.attempts }, }); await notifyHuman(event, reactionConfig.priority ?? "urgent"); + + // Mark as escalated — silences further dispatches until the underlying + // condition resolves and clearReactionTracker() is called explicitly. + tracker.escalated = true; + return { reactionType: reactionKey, success: true, @@ -1656,35 +1682,44 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan (reactionConfig.auto !== false || reactionConfig.action === "notify") ) { try { - if (reactionConfig.action === "send-to-agent") { + // Build enriched config with dynamic base branch message. + // Preserve "warning" priority from old direct-dispatch code unless + // the user explicitly set a different priority in their config. + const enrichedConfig = { + ...reactionConfig, + priority: reactionConfig.priority ?? ("warning" as const), + }; + if (reactionConfig.action === "send-to-agent" && !reactionConfig.message) { const baseBranch = session.pr.baseBranch ?? "the default branch"; const behindNote = cachedData.isBehind ? ` is behind ${baseBranch} and` : ""; - const message = - reactionConfig.message ?? - `Your PR branch${behindNote} has merge conflicts with ${baseBranch}. Rebase your branch on ${baseBranch}, resolve the conflicts, and push. You should not need to call gh for merge status unless you need additional context — this information is current.`; - await sessionManager.send(session.id, message); - } else { - const event = createEvent("merge.conflicts", { - sessionId: session.id, - projectId: session.projectId, - message: `${session.id}: PR has merge conflicts`, - }); - await notifyHuman(event, reactionConfig.priority ?? "warning"); + enrichedConfig.message = `Your PR branch${behindNote} has merge conflicts with ${baseBranch}. Rebase your branch on ${baseBranch}, resolve the conflicts, and push. You should not need to call gh for merge status unless you need additional context — this information is current.`; } - updateSessionMetadata(session, { - lastMergeConflictDispatched: "true", - }); + const result = await executeReaction( + session.id, + session.projectId, + conflictReactionKey, + enrichedConfig, + ); + // Only set dedup flag for non-escalated success — escalation hands off + // to the human, so we must NOT suppress future agent dispatches if the + // condition recurs after the tracker resets. + if (result.success && result.action !== "escalated") { + updateSessionMetadata(session, { + lastMergeConflictDispatched: "true", + }); + } } catch { - // Send failed — will retry on next poll cycle + // Dispatch failed — will retry on next poll cycle } } } else if (lastDispatched === "true") { - // Conflicts resolved — clear so we can re-dispatch if they recur - clearReactionTracker(session.id, conflictReactionKey); + // Conflicts resolved — clear dedup flag and reaction tracker so future + // conflicts start a fresh incident with a fresh escalation budget. updateSessionMetadata(session, { lastMergeConflictDispatched: "", }); + clearReactionTracker(session.id, conflictReactionKey); } } @@ -1849,6 +1884,28 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan updateSessionMetadata(session, metadataUpdates); } + // CI resolution tracking — reset the ci-failed tracker (including its escalated + // flag) once CI has been passing for CI_PASSING_STABLE_THRESHOLD consecutive polls. + // This lets the next real CI failure start with a fresh budget. + if (session.pr) { + const prKey = `${session.pr.owner}/${session.pr.repo}#${session.pr.number}`; + const cachedData = prEnrichmentCache.get(prKey); + if (cachedData) { + if (cachedData.ciStatus === "passing") { + const stableCount = Number(session.metadata["ciPassingStableCount"] ?? "0") + 1; + if (stableCount >= CI_PASSING_STABLE_THRESHOLD) { + clearReactionTracker(session.id, "ci-failed"); + updateSessionMetadata(session, { ciPassingStableCount: "" }); + } else { + updateSessionMetadata(session, { ciPassingStableCount: String(stableCount) }); + } + } else if (session.metadata["ciPassingStableCount"]) { + // pending or failing resets the stability window — only "passing" counts as resolution + updateSessionMetadata(session, { ciPassingStableCount: "" }); + } + } + } + if (newStatus !== oldStatus) { const correlationId = createCorrelationId("lifecycle-transition"); // State transition detected @@ -1879,11 +1936,14 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan allCompleteEmitted = false; } - // Clear reaction trackers for the old status so retries reset on state changes + // Clear reaction trackers for the old status so retries reset on state changes. + // Persistent keys (ci-failed) are excluded — their trackers survive oscillation + // so the escalation budget accumulates across cycles. On escalation, the tracker + // is cleared in executeReaction so future incidents get a fresh budget. const oldEventType = statusToEventType(undefined, oldStatus); if (oldEventType) { const oldReactionKey = eventToReactionKey(oldEventType); - if (oldReactionKey) { + if (oldReactionKey && !PERSISTENT_REACTION_KEYS.has(oldReactionKey)) { clearReactionTracker(session.id, oldReactionKey); } } diff --git a/packages/web/server/__tests__/mux-websocket.test.ts b/packages/web/server/__tests__/mux-websocket.test.ts index 2fe0e83ef..d882df667 100644 --- a/packages/web/server/__tests__/mux-websocket.test.ts +++ b/packages/web/server/__tests__/mux-websocket.test.ts @@ -15,6 +15,7 @@ describe("SessionBroadcaster", () => { }); afterEach(() => { + vi.clearAllTimers(); vi.useRealTimers(); }); @@ -29,13 +30,10 @@ describe("SessionBroadcaster", () => { describe("subscribe", () => { it("sends an immediate snapshot to a new subscriber", async () => { const patches = [makePatch("s1")]; - // Mock the snapshot fetch mockFetch.mockResolvedValueOnce({ ok: true, json: async () => ({ sessions: patches }), }); - // Mock the SSE connect (hangs forever) - mockFetch.mockReturnValueOnce(new Promise(() => {})); const callback = vi.fn(); broadcaster.subscribe(callback); @@ -50,94 +48,88 @@ describe("SessionBroadcaster", () => { expect(callback).toHaveBeenCalledWith(patches); }); - it("starts SSE connection on first subscriber", async () => { + it("starts polling interval on first subscriber", async () => { mockFetch.mockResolvedValueOnce({ ok: true, json: async () => ({ sessions: [] }), }); - // SSE connect - mockFetch.mockReturnValueOnce(new Promise(() => {})); broadcaster.subscribe(vi.fn()); await vi.advanceTimersByTimeAsync(0); + // Snapshot fetch is called once on subscribe + expect(mockFetch).toHaveBeenCalledTimes(1); + + // After 3 seconds, polling interval should trigger a second fetch + mockFetch.mockResolvedValueOnce({ + ok: true, + json: async () => ({ sessions: [] }), + }); + await vi.advanceTimersByTimeAsync(3000); + expect(mockFetch).toHaveBeenCalledTimes(2); - expect(mockFetch).toHaveBeenCalledWith( - "http://localhost:3000/api/events", - expect.objectContaining({ - headers: { Accept: "text/event-stream" }, - }), - ); }); - it("does not start a second SSE connection for additional subscribers", async () => { + it("does not start a second polling interval for additional subscribers", async () => { mockFetch.mockResolvedValue({ ok: true, json: async () => ({ sessions: [] }), }); - // SSE connect (first subscriber triggers it) - mockFetch.mockReturnValueOnce(new Promise(() => {})); broadcaster.subscribe(vi.fn()); broadcaster.subscribe(vi.fn()); await vi.advanceTimersByTimeAsync(0); - // 1 snapshot for sub1 + 1 SSE connect + 1 snapshot for sub2 = 3 - // (SSE connect is only called once) - const sseConnects = mockFetch.mock.calls.filter( - (call) => call[0] === "http://localhost:3000/api/events", - ); - expect(sseConnects).toHaveLength(1); + // 1 snapshot for sub1 + 1 snapshot for sub2 = 2 + expect(mockFetch).toHaveBeenCalledTimes(2); + + // After 3 seconds, only one polling fetch happens + await vi.advanceTimersByTimeAsync(3000); + expect(mockFetch).toHaveBeenCalledTimes(3); }); - it("returns an unsubscribe function that disconnects when last subscriber leaves", async () => { + it("returns an unsubscribe function that stops polling when last subscriber leaves", async () => { mockFetch.mockResolvedValueOnce({ ok: true, json: async () => ({ sessions: [] }), }); - mockFetch.mockReturnValueOnce(new Promise(() => {})); const unsub = broadcaster.subscribe(vi.fn()); await vi.advanceTimersByTimeAsync(0); + // Unsubscribe triggers disconnect unsub(); - // After unsubscribe, the abort should be triggered (disconnect) - // Further subscribe should trigger a new connection + + // Reset and advance past polling interval + mockFetch.mockResolvedValueOnce({ + ok: true, + json: async () => ({ sessions: [] }), + }); + await vi.advanceTimersByTimeAsync(3000); + + // Should not have called fetch again after unsubscribe + expect(mockFetch).toHaveBeenCalledTimes(1); }); }); describe("broadcast", () => { - it("delivers patches to all subscribers", async () => { + it("delivers patches to all subscribers on each poll", async () => { const patches = [makePatch("s1"), makePatch("s2")]; - // Create a readable stream that sends one SSE event - const encoder = new TextEncoder(); - const sseData = `data: ${JSON.stringify({ type: "snapshot", sessions: patches })}\n\n`; - let readerDone = false; - const mockReader = { - read: vi.fn().mockImplementation(async () => { - if (!readerDone) { - readerDone = true; - return { done: false, value: encoder.encode(sseData) }; - } - return { done: true, value: undefined }; - }), - }; - - // Snapshot fetch + // Initial snapshot for first subscriber mockFetch.mockResolvedValueOnce({ ok: true, - json: async () => ({ sessions: [] }), + json: async () => ({ sessions: patches }), }); - // SSE connect returns a stream + // Initial snapshot for second subscriber mockFetch.mockResolvedValueOnce({ ok: true, - body: { getReader: () => mockReader }, + json: async () => ({ sessions: patches }), }); - // Second subscriber snapshot + // Polling fetch after 3s mockFetch.mockResolvedValueOnce({ ok: true, - json: async () => ({ sessions: [] }), + json: async () => ({ sessions: patches }), }); const cb1 = vi.fn(); @@ -147,30 +139,29 @@ describe("SessionBroadcaster", () => { await vi.advanceTimersByTimeAsync(10); - // Both callbacks should have received the broadcast + // Both callbacks should have received initial snapshot expect(cb1).toHaveBeenCalledWith(patches); expect(cb2).toHaveBeenCalledWith(patches); + + // Advance past poll interval (3s) and add buffer for promise resolution + await vi.advanceTimersByTimeAsync(3010); + + // Should be called again from polling + expect(cb1).toHaveBeenCalledTimes(2); + expect(cb2).toHaveBeenCalledTimes(2); }); it("isolates subscriber errors — one throw does not skip others", async () => { const patches = [makePatch("s1")]; - const encoder = new TextEncoder(); - const sseData = `data: ${JSON.stringify({ type: "snapshot", sessions: patches })}\n\n`; - let readerDone = false; - const mockReader = { - read: vi.fn().mockImplementation(async () => { - if (!readerDone) { - readerDone = true; - return { done: false, value: encoder.encode(sseData) }; - } - return { done: true, value: undefined }; - }), - }; - // Return null for snapshots so the initial callback doesn't fire (and throw) - mockFetch.mockResolvedValueOnce({ ok: false, status: 500 }); - mockFetch.mockResolvedValueOnce({ ok: true, body: { getReader: () => mockReader } }); - mockFetch.mockResolvedValueOnce({ ok: false, status: 500 }); + mockFetch.mockResolvedValueOnce({ + ok: true, + json: async () => ({ sessions: patches }), + }); + mockFetch.mockResolvedValueOnce({ + ok: true, + json: async () => ({ sessions: patches }), + }); const throwingCb = vi.fn().mockImplementation(() => { throw new Error("ws.send failed"); @@ -181,6 +172,7 @@ describe("SessionBroadcaster", () => { await vi.advanceTimersByTimeAsync(10); + // goodCb should have received patches despite throwingCb error expect(goodCb).toHaveBeenCalledWith(patches); }); }); @@ -188,7 +180,6 @@ describe("SessionBroadcaster", () => { describe("fetchSnapshot", () => { it("returns null on fetch failure", async () => { mockFetch.mockRejectedValueOnce(new Error("network error")); - mockFetch.mockReturnValueOnce(new Promise(() => {})); const callback = vi.fn(); broadcaster.subscribe(callback); @@ -200,7 +191,6 @@ describe("SessionBroadcaster", () => { it("returns null on non-OK response", async () => { mockFetch.mockResolvedValueOnce({ ok: false, status: 500 }); - mockFetch.mockReturnValueOnce(new Promise(() => {})); const callback = vi.fn(); broadcaster.subscribe(callback); @@ -210,48 +200,28 @@ describe("SessionBroadcaster", () => { }); }); - describe("SSE reconnection", () => { - it("reconnects after SSE connection drops if subscribers remain", async () => { - mockFetch.mockResolvedValueOnce({ ok: true, json: async () => ({ sessions: [] }) }); - // SSE connect fails - mockFetch.mockResolvedValueOnce({ ok: false, status: 503 }); - // Reconnect SSE after backoff - mockFetch.mockReturnValueOnce(new Promise(() => {})); - - broadcaster.subscribe(vi.fn()); - await vi.advanceTimersByTimeAsync(0); - - // Advance past the 5s reconnect backoff - await vi.advanceTimersByTimeAsync(6000); - - const sseConnects = mockFetch.mock.calls.filter( - (call) => call[0] === "http://localhost:3000/api/events", - ); - expect(sseConnects.length).toBeGreaterThanOrEqual(2); - }); - }); - describe("disconnect", () => { - it("clears reconnect timer on disconnect", async () => { - mockFetch.mockResolvedValueOnce({ ok: true, json: async () => ({ sessions: [] }) }); - // SSE connect fails to trigger reconnect path - mockFetch.mockResolvedValueOnce({ ok: false, status: 503 }); + it("stops polling when last subscriber unsubscribes", async () => { + mockFetch.mockResolvedValueOnce({ + ok: true, + json: async () => ({ sessions: [] }), + }); const unsub = broadcaster.subscribe(vi.fn()); await vi.advanceTimersByTimeAsync(0); - // Unsubscribe triggers disconnect, which should clear reconnect timer + // Unsubscribe triggers disconnect unsub(); - // Advance past reconnect backoff — should NOT trigger a new connect - mockFetch.mockReturnValueOnce(new Promise(() => {})); - await vi.advanceTimersByTimeAsync(6000); + // Advance past polling interval + mockFetch.mockResolvedValueOnce({ + ok: true, + json: async () => ({ sessions: [] }), + }); + await vi.advanceTimersByTimeAsync(3000); - const sseConnects = mockFetch.mock.calls.filter( - (call) => call[0] === "http://localhost:3000/api/events", - ); - // Only 1 connect attempt, no reconnect after disconnect - expect(sseConnects).toHaveLength(1); + // Should only have 1 fetch (initial snapshot) + expect(mockFetch).toHaveBeenCalledTimes(1); }); }); }); diff --git a/packages/web/server/mux-websocket.ts b/packages/web/server/mux-websocket.ts index 5be08b17e..8b621cb48 100644 --- a/packages/web/server/mux-websocket.ts +++ b/packages/web/server/mux-websocket.ts @@ -2,9 +2,8 @@ * Multiplexed WebSocket server for terminal multiplexing. * Manages multiple terminal connections over a single persistent WebSocket. * - * Session updates are delivered via a single shared SSE connection from this - * process to Next.js /api/events, then broadcast to all subscribed clients. - * This replaces per-client HTTP polling and makes session updates event-driven. + * Session updates are delivered via polling of Next.js /api/sessions/patches + * every 3s, then broadcast to all subscribed clients via WebSocket. */ import { WebSocketServer, WebSocket } from "ws"; @@ -33,6 +32,7 @@ type ServerMessage = | { ch: "terminal"; id: string; type: "opened" } | { ch: "terminal"; id: string; type: "error"; message: string } | { ch: "sessions"; type: "snapshot"; sessions: SessionPatch[] } + | { ch: "sessions"; type: "error"; error: string } | { ch: "system"; type: "pong" } | { ch: "system"; type: "error"; message: string }; @@ -48,14 +48,15 @@ interface SessionPatch { } /** - * Manages a single shared SSE connection to Next.js /api/events. - * Broadcasts session patches to all subscribed callbacks. - * Lazily connects on first subscriber, disconnects when the last one leaves. + * Manages polling of session patches from Next.js /api/sessions/patches. + * Broadcasts to all subscribed callbacks. + * Lazily starts polling on first subscriber, stops when the last one leaves. */ export class SessionBroadcaster { private subscribers = new Set<(sessions: SessionPatch[]) => void>(); - private abortController: AbortController | null = null; - private reconnectTimer: ReturnType | null = null; + private errorSubscribers = new Set<(error: string) => void>(); + private intervalId: ReturnType | null = null; + private polling = false; private readonly baseUrl: string; constructor(nextPort: string) { @@ -63,27 +64,50 @@ export class SessionBroadcaster { } /** - * Subscribe to session patches. Returns an unsubscribe function. - * Sends an immediate snapshot to the new subscriber, then live SSE pushes. + * Subscribe to session patches and errors. Returns an unsubscribe function. + * Sends an immediate snapshot to the new subscriber, then polling updates. */ - subscribe(callback: (sessions: SessionPatch[]) => void): () => void { + subscribe(callback: (sessions: SessionPatch[]) => void, onError?: (error: string) => void): () => void { const wasEmpty = this.subscribers.size === 0; this.subscribers.add(callback); + if (onError) this.errorSubscribers.add(onError); // Immediately send a one-off snapshot to just this new subscriber - void this.fetchSnapshot().then((sessions) => { - if (sessions && this.subscribers.has(callback)) { - callback(sessions); + void this.fetchSnapshot().then((result) => { + if (result.sessions && this.subscribers.has(callback)) { + try { + callback(result.sessions); + } catch { + // Isolate subscriber errors so one bad subscriber doesn't break others + } + } else if (result.error && onError && this.errorSubscribers.has(onError)) { + try { + onError(result.error); + } catch { + // Isolate subscriber errors + } } }); - // Start the shared SSE connection if this is the first subscriber + // Start polling if this is the first subscriber if (wasEmpty) { - void this.connect(); + this.intervalId = setInterval(() => { + if (this.polling) return; + this.polling = true; + void this.fetchSnapshot() + .then((result) => { + if (result.sessions && this.intervalId !== null) this.broadcast(result.sessions); + else if (result.error && this.intervalId !== null) this.broadcastError(result.error); + }) + .finally(() => { + this.polling = false; + }); + }, 3000); } return () => { this.subscribers.delete(callback); + if (onError) this.errorSubscribers.delete(onError); if (this.subscribers.size === 0) { this.disconnect(); } @@ -100,101 +124,45 @@ export class SessionBroadcaster { } } - /** One-shot HTTP fetch of the current session list for immediate delivery. */ - private async fetchSnapshot(): Promise { - try { - const controller = new AbortController(); - const timeoutId = setTimeout(() => controller.abort(), 4000); + private broadcastError(error: string): void { + for (const callback of this.errorSubscribers) { try { - const res = await fetch(`${this.baseUrl}/api/sessions/patches`, { - signal: controller.signal, - }); - clearTimeout(timeoutId); - if (!res.ok) return null; - const data = (await res.json()) as { sessions?: SessionPatch[] }; - return data.sessions ?? null; - } catch { - clearTimeout(timeoutId); - return null; + callback(error); + } catch (err) { + console.error("[MuxServer] Session error subscriber threw:", err); } - } catch { - return null; } } - /** Open a persistent SSE connection and stream events to all subscribers. */ - private async connect(): Promise { - if (this.abortController) return; - + /** One-shot HTTP fetch of the current session list. */ + private async fetchSnapshot(): Promise<{ sessions: SessionPatch[] | null; error: string | null }> { const controller = new AbortController(); - this.abortController = controller; - const { signal } = controller; - + const timeoutId = setTimeout(() => controller.abort(), 4000); try { - const res = await fetch(`${this.baseUrl}/api/events`, { - signal, - headers: { Accept: "text/event-stream" }, + const res = await fetch(`${this.baseUrl}/api/sessions/patches`, { + signal: controller.signal, }); - - if (!res.ok || !res.body) { - throw new Error(`SSE connect failed: ${res.status}`); - } - - const reader = res.body.getReader(); - const decoder = new TextDecoder(); - let buffer = ""; - - while (!signal.aborted) { - const { done, value } = await reader.read(); - if (done) break; - - buffer += decoder.decode(value, { stream: true }); - const lines = buffer.split("\n"); - buffer = lines.pop() ?? ""; - - for (const line of lines) { - if (!line.startsWith("data: ")) continue; - try { - const event = JSON.parse(line.slice(6)) as { - type: string; - sessions?: SessionPatch[]; - }; - if (event.type === "snapshot" && event.sessions) { - this.broadcast(event.sessions); - } - } catch { - // ignore malformed events - } - } + clearTimeout(timeoutId); + if (!res.ok) { + const msg = `Session fetch failed: HTTP ${res.status}`; + console.warn(`[SessionBroadcaster] ${msg}`); + return { sessions: null, error: msg }; } + const data = (await res.json()) as { sessions?: SessionPatch[] }; + return { sessions: data.sessions ?? null, error: null }; } catch (err) { - if (signal.aborted) return; // intentional disconnect, not an error - console.warn("[MuxServer] SSE connection lost:", err instanceof Error ? err.message : err); - } finally { - // Only clear our own controller — a concurrent connect() may have already - // set a new one (e.g. disconnect() → subscribe() → connect() in the same tick). - if (this.abortController === controller) { - this.abortController = null; - } - } - - // Reconnect with backoff if there are still subscribers - if (this.subscribers.size > 0) { - console.log("[MuxServer] SSE reconnecting in 5s"); - this.reconnectTimer = setTimeout(() => { - this.reconnectTimer = null; - if (this.subscribers.size > 0) void this.connect(); - }, 5000); + clearTimeout(timeoutId); + const msg = err instanceof Error ? err.message : String(err); + console.warn("[SessionBroadcaster] fetchSnapshot error:", msg); + return { sessions: null, error: msg }; } } private disconnect(): void { - if (this.reconnectTimer) { - clearTimeout(this.reconnectTimer); - this.reconnectTimer = null; + if (this.intervalId !== null) { + clearInterval(this.intervalId); + this.intervalId = null; } - this.abortController?.abort(); - this.abortController = null; } } @@ -222,6 +190,7 @@ interface ManagedTerminal { } const RING_BUFFER_MAX = 50 * 1024; // 50KB max per terminal +const WS_BUFFER_HIGH_WATERMARK = 64 * 1024; // 64KB const MAX_REATTACH_ATTEMPTS = 3; /** @@ -773,12 +742,22 @@ export function createMuxWebSocket(tmuxPath?: string | null): WebSocketServer | } } else if (msg.ch === "subscribe") { if (msg.topics.includes("sessions") && !sessionUnsubscribe) { - sessionUnsubscribe = broadcaster.subscribe((sessions) => { - if (ws.readyState === WebSocket.OPEN) { + sessionUnsubscribe = broadcaster.subscribe( + (sessions) => { + if (ws.readyState !== WebSocket.OPEN) return; + if (ws.bufferedAmount > WS_BUFFER_HIGH_WATERMARK) { + console.warn("[MuxServer] Skipping session snapshot — socket backpressured"); + return; + } const snapMsg: ServerMessage = { ch: "sessions", type: "snapshot", sessions }; ws.send(JSON.stringify(snapMsg)); - } - }); + }, + (error) => { + if (ws.readyState !== WebSocket.OPEN) return; + const errMsg: ServerMessage = { ch: "sessions", type: "error", error }; + ws.send(JSON.stringify(errMsg)); + }, + ); } } } catch (err) { diff --git a/packages/web/src/__tests__/api-routes.test.ts b/packages/web/src/__tests__/api-routes.test.ts index 12ccac890..8d6474ad6 100644 --- a/packages/web/src/__tests__/api-routes.test.ts +++ b/packages/web/src/__tests__/api-routes.test.ts @@ -221,7 +221,6 @@ import { POST as killPOST } from "@/app/api/sessions/[id]/kill/route"; import { POST as restorePOST } from "@/app/api/sessions/[id]/restore/route"; import { POST as remapPOST } from "@/app/api/sessions/[id]/remap/route"; import { POST as mergePOST } from "@/app/api/prs/[id]/merge/route"; -import { GET as eventsGET } from "@/app/api/events/route"; import { GET as observabilityGET } from "@/app/api/observability/route"; import { GET as runtimeTerminalGET } from "@/app/api/runtime/terminal/route"; import { GET as verifyGET, POST as verifyPOST } from "@/app/api/verify/route"; @@ -1293,35 +1292,6 @@ describe("API Routes", () => { }); }); - // ── GET /api/events (SSE) ────────────────────────────────────────── - - describe("GET /api/events", () => { - it("returns SSE content type", async () => { - const req = makeRequest("/api/events", { method: "GET" }); - const res = await eventsGET(req); - expect(res.headers.get("Content-Type")).toBe("text/event-stream"); - expect(res.headers.get("Cache-Control")).toBe("no-cache"); - }); - - it("streams initial snapshot event", async () => { - const req = makeRequest("/api/events", { method: "GET" }); - const res = await eventsGET(req); - const reader = res.body!.getReader(); - const { value } = await reader.read(); - reader.cancel(); - const text = new TextDecoder().decode(value); - expect(text).toContain("data: "); - const jsonStr = text.replace("data: ", "").trim(); - const event = JSON.parse(jsonStr); - expect(event.type).toBe("snapshot"); - expect(event.correlationId).toBeTruthy(); - expect(Array.isArray(event.sessions)).toBe(true); - expect(event.sessions.length).toBeGreaterThan(0); - expect(event.sessions[0]).toHaveProperty("id"); - expect(event.sessions[0]).toHaveProperty("attentionLevel"); - }); - }); - describe("GET /api/observability", () => { it("returns observability summary with correlation header", async () => { const req = makeRequest("/api/observability", { method: "GET" }); diff --git a/packages/web/src/app/api/events/route.ts b/packages/web/src/app/api/events/route.ts deleted file mode 100644 index d89014601..000000000 --- a/packages/web/src/app/api/events/route.ts +++ /dev/null @@ -1,266 +0,0 @@ -import { getServices } from "@/lib/services"; -import { sessionToDashboard } from "@/lib/serialize"; -import { getAttentionLevel } from "@/lib/types"; -import { filterWorkerSessions } from "@/lib/project-utils"; -import { - createCorrelationId, - createProjectObserver, - type ProjectObserver, -} from "@aoagents/ao-core"; - -export const dynamic = "force-dynamic"; - -const SESSION_EVENTS_POLL_INTERVAL_MS = 5000; - -/** - * GET /api/events — SSE stream for real-time lifecycle events - * - * Sends session state updates to connected clients. - * Polls SessionManager.list() on an interval (no SSE push from core yet). - */ -export async function GET(request: Request): Promise { - const encoder = new TextEncoder(); - const correlationId = createCorrelationId("sse"); - const { searchParams } = new URL(request.url); - const projectFilter = searchParams.get("project"); - type ServicesConfig = Awaited>["config"]; - let heartbeat: ReturnType | undefined; - let updates: ReturnType | undefined; - let observerProjectId: string | undefined; - let observer: ProjectObserver | null = null; - let streamClosed = false; - - const ensureObserver = (config: ServicesConfig): ProjectObserver | null => { - if (!observerProjectId) { - const requestedProjectId = - projectFilter && projectFilter !== "all" && config.projects[projectFilter] - ? projectFilter - : undefined; - observerProjectId = requestedProjectId ?? Object.keys(config.projects)[0]; - } - if (!observerProjectId) return null; - if (!observer) { - observer = createProjectObserver(config, "web-events"); - } - return observer; - }; - - const stopStream = () => { - if (streamClosed) return; - streamClosed = true; - if (heartbeat) clearInterval(heartbeat); - if (updates) clearInterval(updates); - }; - - const encodeEvent = (payload: unknown) => encoder.encode(`data: ${JSON.stringify(payload)}\n\n`); - - const stream = new ReadableStream({ - start(controller) { - const safeEnqueue = (payload: unknown): boolean => { - if (streamClosed) return false; - try { - controller.enqueue(encodeEvent(payload)); - return true; - } catch { - stopStream(); - return false; - } - }; - - void (async () => { - try { - const { config } = await getServices(); - const projectObserver = ensureObserver(config); - if (projectObserver && observerProjectId) { - projectObserver.recordOperation({ - metric: "sse_connect", - operation: "sse.connect", - outcome: "success", - correlationId, - projectId: observerProjectId, - data: { path: "/api/events" }, - level: "info", - }); - projectObserver.setHealth({ - surface: "sse.events", - status: "ok", - projectId: observerProjectId, - correlationId, - details: { projectId: observerProjectId, connection: "open" }, - }); - } - } catch { - void 0; - } - - try { - const { config, sessionManager } = await getServices(); - const requestedProjectId = - projectFilter && projectFilter !== "all" && config.projects[projectFilter] - ? projectFilter - : undefined; - const sessions = await sessionManager.list(requestedProjectId); - const workerSessions = filterWorkerSessions(sessions, projectFilter, config.projects); - const dashboardSessions = workerSessions.map(sessionToDashboard); - const projectObserver = ensureObserver(config); - - const attentionZones = config.dashboard?.attentionZones ?? "simple"; - const initialEvent = { - type: "snapshot", - correlationId, - emittedAt: new Date().toISOString(), - sessions: dashboardSessions.map((s) => ({ - id: s.id, - status: s.status, - activity: s.activity, - attentionLevel: getAttentionLevel(s, attentionZones), - lastActivityAt: s.lastActivityAt, - })), - }; - safeEnqueue(initialEvent); - if (projectObserver && observerProjectId) { - projectObserver.recordOperation({ - metric: "sse_snapshot", - operation: "sse.snapshot", - outcome: "success", - correlationId, - projectId: observerProjectId, - data: { sessionCount: dashboardSessions.length, initial: true }, - level: "info", - }); - } - } catch (error) { - safeEnqueue({ - type: "error", - correlationId, - emittedAt: new Date().toISOString(), - error: error instanceof Error ? error.message : "Failed to load live dashboard data", - }); - } - })(); - - // Send periodic heartbeat - heartbeat = setInterval(() => { - if (streamClosed) return; - try { - controller.enqueue(encoder.encode(`: heartbeat\n\n`)); - } catch { - stopStream(); - } - }, 15000); - - // Poll for session state changes frequently enough that new workers - // appear in the dashboard/sidebar quickly after the orchestrator spawns them. - updates = setInterval(() => { - void (async () => { - let dashboardSessions; - try { - const { config, sessionManager } = await getServices(); - const requestedProjectId = - projectFilter && projectFilter !== "all" && config.projects[projectFilter] - ? projectFilter - : undefined; - const sessions = await sessionManager.list(requestedProjectId); - const workerSessions = filterWorkerSessions(sessions, projectFilter, config.projects); - dashboardSessions = workerSessions.map(sessionToDashboard); - const projectObserver = ensureObserver(config); - - if (projectObserver && observerProjectId) { - projectObserver.setHealth({ - surface: "sse.events", - status: "ok", - projectId: observerProjectId, - correlationId, - details: { - projectId: observerProjectId, - sessionCount: dashboardSessions.length, - lastEventAt: new Date().toISOString(), - }, - }); - } - - try { - const attentionZones = config.dashboard?.attentionZones ?? "simple"; - const event = { - type: "snapshot", - correlationId, - emittedAt: new Date().toISOString(), - sessions: dashboardSessions.map((s) => ({ - id: s.id, - status: s.status, - activity: s.activity, - attentionLevel: getAttentionLevel(s, attentionZones), - lastActivityAt: s.lastActivityAt, - })), - }; - safeEnqueue(event); - if (projectObserver && observerProjectId) { - projectObserver.recordOperation({ - metric: "sse_snapshot", - operation: "sse.snapshot", - outcome: "success", - correlationId, - projectId: observerProjectId, - data: { sessionCount: dashboardSessions.length, initial: false }, - level: "info", - }); - } - } catch (error) { - safeEnqueue({ - type: "error", - correlationId, - emittedAt: new Date().toISOString(), - error: error instanceof Error ? error.message : "Live dashboard update failed", - }); - } - } catch (error) { - safeEnqueue({ - type: "error", - correlationId, - emittedAt: new Date().toISOString(), - error: error instanceof Error ? error.message : "Live dashboard update failed", - }); - } - })(); - }, SESSION_EVENTS_POLL_INTERVAL_MS); - }, - cancel() { - stopStream(); - void (async () => { - try { - const { config } = await getServices(); - const projectObserver = ensureObserver(config); - if (!projectObserver || !observerProjectId) return; - projectObserver.recordOperation({ - metric: "sse_disconnect", - operation: "sse.disconnect", - outcome: "success", - correlationId, - projectId: observerProjectId, - data: { path: "/api/events" }, - level: "info", - }); - projectObserver.setHealth({ - surface: "sse.events", - status: "warn", - projectId: observerProjectId, - correlationId, - reason: "SSE connection closed", - details: { projectId: observerProjectId, connection: "closed" }, - }); - } catch { - void 0; - } - })(); - }, - }); - - return new Response(stream, { - headers: { - "Content-Type": "text/event-stream", - "Cache-Control": "no-cache", - Connection: "keep-alive", - "X-Accel-Buffering": "no", - }, - }); -} diff --git a/packages/web/src/app/sessions/[id]/page.test.tsx b/packages/web/src/app/sessions/[id]/page.test.tsx index 8545ee4b0..b6f405c02 100644 --- a/packages/web/src/app/sessions/[id]/page.test.tsx +++ b/packages/web/src/app/sessions/[id]/page.test.tsx @@ -31,6 +31,10 @@ vi.mock("@/components/SessionDetail", () => ({ }, })); +vi.mock("@/hooks/useMuxSessionActivity", () => ({ + useMuxSessionActivity: () => null, +})); + function makeWorkerSession(): DashboardSession { return { id: "worker-1", diff --git a/packages/web/src/app/sessions/[id]/page.tsx b/packages/web/src/app/sessions/[id]/page.tsx index 307cd7638..fb32d5a67 100644 --- a/packages/web/src/app/sessions/[id]/page.tsx +++ b/packages/web/src/app/sessions/[id]/page.tsx @@ -11,7 +11,7 @@ import { type DashboardSession, type ActivityState, getAttentionLevel } from "@/ import { activityIcon } from "@/lib/activity-icons"; import type { ProjectInfo } from "@/lib/project-name"; import { getSessionTitle } from "@/lib/format"; -import { useSSESessionActivity } from "@/hooks/useSSESessionActivity"; +import { useMuxSessionActivity } from "@/hooks/useMuxSessionActivity"; import { useMuxOptional } from "@/providers/MuxProvider"; import type { SessionPatch } from "@/lib/mux-protocol"; import { projectSessionPath } from "@/lib/routes"; @@ -439,8 +439,8 @@ export default function SessionPage() { } }, []); - // Subscribe to SSE for real-time activity updates (title emoji) - const sseActivity = useSSESessionActivity(id, sessionProjectId ?? expectedProjectId ?? undefined); + // Get real-time activity updates from WebSocket (title emoji) + const sseActivity = useMuxSessionActivity(id); // Update document title based on session data + SSE activity override useEffect(() => { diff --git a/packages/web/src/components/Dashboard.tsx b/packages/web/src/components/Dashboard.tsx index 6ac961737..a98c75b31 100644 --- a/packages/web/src/components/Dashboard.tsx +++ b/packages/web/src/components/Dashboard.tsx @@ -158,23 +158,28 @@ function DashboardInner({ } return levels; }, [initialSessions, attentionZones]); - const { sessions, connectionStatus, sseAttentionLevels, liveSessionsResolved, loadError } = - useSessionEvents({ - initialSessions, - // No project filter — sidebar needs all sessions across all projects. - // Kanban filtering is applied client-side via projectSessions below. - muxSessions: mux?.status === "connected" ? mux.sessions : undefined, - initialAttentionLevels, - attentionZones, - }); + const { sessions, attentionLevels, liveSessionsResolved, loadError } = useSessionEvents({ + initialSessions, + // No project filter — sidebar needs all sessions across all projects. + // Kanban filtering is applied client-side via projectSessions below. + muxSessions: mux?.status === "connected" ? mux.sessions : undefined, + muxLastError: mux?.lastError, + initialAttentionLevels, + attentionZones, + }); const projectSessions = useMemo(() => { if (!projectId) return sessions; return sessions.filter((s) => s.projectId === projectId); }, [sessions, projectId]); + const connectionStatus: "connected" | "reconnecting" | "disconnected" = + mux?.status === "disconnected" ? "disconnected" + : mux?.status === "connected" ? "connected" + : "reconnecting"; const recoveredFromLoadError = Boolean(dashboardLoadError) && liveSessionsResolved; - const visibleDashboardLoadError = - loadError ?? (recoveredFromLoadError ? undefined : dashboardLoadError); + const ssrLoadError = recoveredFromLoadError ? undefined : dashboardLoadError; + // Live WS error takes precedence; fall back to SSR load error when live data hasn't resolved it. + const visibleLoadError = loadError ?? ssrLoadError; const searchParams = useSearchParams(); const router = useRouter(); const routerRef = useRef(router); @@ -229,12 +234,12 @@ function DashboardInner({ setActiveOrchestrators((current) => mergeOrchestrators(current, orchestratorLinks)); }, [orchestratorLinks]); - // Update document title with live attention counts from SSE + // Update document title with live attention counts useEffect(() => { - const needsAttention = countNeedingAttention(sseAttentionLevels); + const needsAttention = countNeedingAttention(attentionLevels); const label = projectName ?? "ao"; document.title = needsAttention > 0 ? `${label} (${needsAttention} need attention)` : label; - }, [sseAttentionLevels, projectName]); + }, [attentionLevels, projectName]); useEffect(() => { setMobileMenuOpen(false); @@ -442,9 +447,9 @@ function DashboardInner({ }; const hasAnySessions = kanbanLevels.some((level) => grouped[level].length > 0); - const showEmptyState = !allProjectsView && !hasAnySessions && !visibleDashboardLoadError; + const showEmptyState = !allProjectsView && !hasAnySessions && !visibleLoadError; - const loadErrorBanner = visibleDashboardLoadError ? ( + const loadErrorBanner = visibleLoadError ? (
- {visibleDashboardLoadError} + {visibleLoadError} Confirm agent-orchestrator.yaml exists and is @@ -612,8 +617,8 @@ function DashboardInner({
setMobileMenuOpen(false)} /> )} -
- +
+

Dashboard

diff --git a/packages/web/src/components/DynamicFavicon.tsx b/packages/web/src/components/DynamicFavicon.tsx index 36603ad49..0219c20b7 100644 --- a/packages/web/src/components/DynamicFavicon.tsx +++ b/packages/web/src/components/DynamicFavicon.tsx @@ -1,15 +1,15 @@ "use client"; import { useEffect } from "react"; -import type { SSEAttentionMap } from "@/hooks/useSessionEvents"; +import type { AttentionMap } from "@/hooks/useSessionEvents"; /** - * Determine overall health from SSE attention levels. + * Determine overall health from attention levels. * - "green" — all sessions working/done/pending, nothing needs attention * - "yellow" — some sessions need review or response * - "red" — critical: sessions stuck, errored, or needing immediate action */ -function computeHealthFromLevels(levels: SSEAttentionMap): "green" | "yellow" | "red" { +function computeHealthFromLevels(levels: AttentionMap): "green" | "yellow" | "red" { const entries = Object.values(levels); if (entries.length === 0) return "green"; @@ -46,7 +46,7 @@ function generateFaviconSvg(initial: string, color: string): string { } /** Count sessions that need human attention (respond, review, action, merge). */ -export function countNeedingAttention(levels: SSEAttentionMap): number { +export function countNeedingAttention(levels: AttentionMap): number { let count = 0; for (const level of Object.values(levels)) { if ( @@ -62,23 +62,23 @@ export function countNeedingAttention(levels: SSEAttentionMap): number { } interface DynamicFaviconProps { - /** Server-computed attention levels from SSE snapshots. */ - sseAttentionLevels: SSEAttentionMap; + /** Server-computed attention levels from session snapshots. */ + attentionLevels: AttentionMap; projectName?: string; } /** * Client component that dynamically updates the browser favicon - * based on system health (session attention levels from SSE). + * based on system health (session attention levels). * - * Uses server-computed attention levels from SSE snapshots for real-time - * updates, rather than recomputing from the full sessions array. + * Uses server-computed attention levels for real-time updates, + * rather than recomputing from the full sessions array. */ -export function DynamicFavicon({ sseAttentionLevels, projectName = "A" }: DynamicFaviconProps) { +export function DynamicFavicon({ attentionLevels, projectName = "A" }: DynamicFaviconProps) { const initial = projectName.charAt(0).toUpperCase(); useEffect(() => { - const health = computeHealthFromLevels(sseAttentionLevels); + const health = computeHealthFromLevels(attentionLevels); const color = HEALTH_COLORS[health]; const href = generateFaviconSvg(initial, color); @@ -91,7 +91,7 @@ export function DynamicFavicon({ sseAttentionLevels, projectName = "A" }: Dynami } link.type = "image/svg+xml"; link.href = href; - }, [sseAttentionLevels, initial]); + }, [attentionLevels, initial]); return null; } diff --git a/packages/web/src/components/PullRequestsPage.tsx b/packages/web/src/components/PullRequestsPage.tsx index be0079b7d..17b72d84a 100644 --- a/packages/web/src/components/PullRequestsPage.tsx +++ b/packages/web/src/components/PullRequestsPage.tsx @@ -52,10 +52,10 @@ export function PullRequestsPage({ }: PullRequestsPageProps) { const orchestratorLinks = orchestrators ?? EMPTY_ORCHESTRATORS; const mux = useMuxOptional(); - // Seed initial attention levels using the same mode the server SSE will - // use when it sends snapshots (read from `config.dashboard.attentionZones` - // upstream). This prevents the sseAttentionLevels map from oscillating - // between detailed (seed/refresh) and simple (server snapshot) values. + // Seed initial attention levels using the same mode used during refresh + // (read from `config.dashboard.attentionZones` upstream). This prevents + // the attentionLevels map from oscillating between detailed (seed/refresh) + // and simple (server snapshot) values. const initialAttentionLevels = useMemo(() => { const levels: Record> = {}; for (const s of initialSessions) { @@ -63,7 +63,7 @@ export function PullRequestsPage({ } return levels; }, [initialSessions, attentionZones]); - const { sessions, sseAttentionLevels } = useSessionEvents({ + const { sessions, attentionLevels } = useSessionEvents({ initialSessions, project: projectId, muxSessions: mux?.status === "connected" ? mux.sessions : undefined, @@ -127,7 +127,7 @@ export function PullRequestsPage({

setMobileMenuOpen(false)} /> )}
- + {isMobile ? (
diff --git a/packages/web/src/components/__tests__/Dashboard.emptyState.test.tsx b/packages/web/src/components/__tests__/Dashboard.emptyState.test.tsx index bd72334d3..18aa4b742 100644 --- a/packages/web/src/components/__tests__/Dashboard.emptyState.test.tsx +++ b/packages/web/src/components/__tests__/Dashboard.emptyState.test.tsx @@ -1,13 +1,6 @@ -import { fireEvent, render, screen, waitFor } from "@testing-library/react"; +import { act, fireEvent, render, screen, waitFor } from "@testing-library/react"; +import { useState } from "react"; import { beforeEach, describe, expect, it, vi } from "vitest"; -import { Dashboard } from "../Dashboard"; - -const eventSourceConstructorMock = vi.hoisted(() => vi.fn()); -let lastEventSourceMock: { - onmessage: ((event: MessageEvent) => void) | null; - onerror: ((event?: Event) => void) | null; - close: ReturnType; -}; vi.mock("next/navigation", () => ({ useRouter: () => ({ push: vi.fn(), replace: vi.fn(), refresh: vi.fn() }), @@ -15,21 +8,29 @@ vi.mock("next/navigation", () => ({ useSearchParams: () => new URLSearchParams(), })); +let currentMuxLastError: string | null = null; + +vi.mock("@/providers/MuxProvider", () => ({ + useMuxOptional: () => ({ + subscribeTerminal: () => () => {}, + writeTerminal: () => {}, + openTerminal: () => {}, + closeTerminal: () => {}, + resizeTerminal: () => {}, + // "connecting" so muxSessions stays undefined — prevents snapshot dispatch from + // immediately flipping liveSessionsResolved and masking the SSR load error. + status: "connecting" as const, + sessions: [], + lastError: currentMuxLastError, + }), + MuxProvider: ({ children }: { children: React.ReactNode }) => <>{children}, +})); + +import { Dashboard } from "../Dashboard"; + beforeEach(() => { - const eventSourceMock = { - onmessage: null, - onerror: null, - close: vi.fn(), - }; - lastEventSourceMock = eventSourceMock; - eventSourceConstructorMock.mockReset(); - eventSourceConstructorMock.mockImplementation(() => eventSourceMock as unknown as EventSource); - global.EventSource = Object.assign(eventSourceConstructorMock, { - CONNECTING: 0, - OPEN: 1, - CLOSED: 2, - }) as unknown as typeof EventSource; global.fetch = vi.fn(); + currentMuxLastError = null; }); describe("Dashboard empty state", () => { @@ -113,36 +114,51 @@ describe("Dashboard empty state", () => { expect(screen.queryByText(/Ready to orchestrate/i)).not.toBeInTheDocument(); expect(screen.getByRole("alert")).toHaveTextContent("Orchestrator failed to load"); expect(screen.getByRole("alert")).toHaveTextContent("No agent-orchestrator.yaml found"); - expect(eventSourceConstructorMock).toHaveBeenCalledTimes(1); }); - it("shows a live load error banner when the SSE stream reports a backend failure", async () => { - render(); + it("shows live load-error banner when WS transport reports a fetch failure", async () => { + let forceUpdate: () => void = () => {}; + function Wrapper() { + const [, tick] = useState(0); + forceUpdate = () => tick((n) => n + 1); + return ; + } - await waitFor(() => { - lastEventSourceMock.onmessage?.({ - data: JSON.stringify({ type: "error", error: "session list timed out" }), - } as MessageEvent); - expect(screen.getByRole("alert")).toHaveTextContent("session list timed out"); + render(); + expect(screen.queryByRole("alert")).not.toBeInTheDocument(); + + await act(async () => { + currentMuxLastError = "Session fetch failed: HTTP 503"; + forceUpdate(); }); + + expect(screen.getByRole("alert")).toHaveTextContent("Orchestrator failed to load"); + expect(screen.getByRole("alert")).toHaveTextContent("Session fetch failed: HTTP 503"); }); - it("clears a live load error banner after a healthy snapshot with unchanged sessions", async () => { - render(); + it("clears live load-error banner when the next WS snapshot succeeds", async () => { + let forceUpdate: () => void = () => {}; + function Wrapper() { + const [, tick] = useState(0); + forceUpdate = () => tick((n) => n + 1); + return ; + } - await waitFor(() => { - lastEventSourceMock.onmessage?.({ - data: JSON.stringify({ type: "error", error: "session list timed out" }), - } as MessageEvent); - expect(screen.getByRole("alert")).toHaveTextContent("session list timed out"); + render(); + + await act(async () => { + currentMuxLastError = "Session fetch failed: HTTP 503"; + forceUpdate(); }); - await waitFor(() => { - lastEventSourceMock.onmessage?.({ - data: JSON.stringify({ type: "snapshot", sessions: [] }), - } as MessageEvent); - expect(screen.queryByRole("alert")).not.toBeInTheDocument(); + expect(screen.getByRole("alert")).toBeInTheDocument(); + + await act(async () => { + currentMuxLastError = null; + forceUpdate(); }); + + expect(screen.queryByRole("alert")).not.toBeInTheDocument(); }); it("shows empty state when only done sessions exist", () => { diff --git a/packages/web/src/components/__tests__/Dashboard.renderCadence.test.tsx b/packages/web/src/components/__tests__/Dashboard.renderCadence.test.tsx index 7ee103c91..a00ef4a5d 100644 --- a/packages/web/src/components/__tests__/Dashboard.renderCadence.test.tsx +++ b/packages/web/src/components/__tests__/Dashboard.renderCadence.test.tsx @@ -1,5 +1,5 @@ -import { act, render, waitFor } from "@testing-library/react"; -import { memo } from "react"; +import { act, render } from "@testing-library/react"; +import { memo, useState } from "react"; import { beforeEach, describe, expect, it, vi } from "vitest"; const renderCounts = new Map(); @@ -17,69 +17,108 @@ vi.mock("@/components/SessionCard", () => ({ }), })); +import type { SessionPatch } from "@/lib/mux-protocol"; + +// Module-level sessions updated by tests, read by the mock on every render. +let currentMuxSessions: SessionPatch[] = []; + +vi.mock("@/providers/MuxProvider", () => ({ + useMuxOptional: () => ({ + subscribeTerminal: () => () => {}, + writeTerminal: () => {}, + openTerminal: () => {}, + closeTerminal: () => {}, + resizeTerminal: () => {}, + status: "connected" as const, + sessions: currentMuxSessions, + }), + MuxProvider: ({ children }: { children: React.ReactNode }) => <>{children}, +})); + import { Dashboard } from "../Dashboard"; import { makeSession } from "../../__tests__/helpers"; -describe("Dashboard render cadence", () => { - let eventSourceMock: { - onmessage: ((event: MessageEvent) => void) | null; - onerror: (() => void) | null; - close: () => void; - }; +// Wrapper that exposes a forceUpdate so tests can trigger re-renders after +// updating currentMuxSessions, causing Dashboard to re-call useMuxOptional(). +let forceUpdate: () => void = () => {}; +function ControlledDashboard({ + initialSessions, +}: { + initialSessions: ReturnType[]; +}) { + const [, tick] = useState(0); + forceUpdate = () => tick((n) => n + 1); + return ; +} + +describe("Dashboard render cadence", () => { beforeEach(() => { renderCounts.clear(); - eventSourceMock = { - onmessage: null, - onerror: null, - close: vi.fn(), - }; - const eventSourceConstructor = vi.fn(() => eventSourceMock as unknown as EventSource); - global.EventSource = Object.assign(eventSourceConstructor, { - CONNECTING: 0, - OPEN: 1, - CLOSED: 2, - }) as unknown as typeof EventSource; + currentMuxSessions = []; global.fetch = vi.fn(); }); it("rerenders only the changed session card for same-membership snapshots", async () => { + const ts = new Date().toISOString(); const initialSessions = [ - makeSession({ id: "session-1", summary: "First session" }), - makeSession({ id: "session-2", summary: "Second session" }), + makeSession({ id: "session-1", status: "working", activity: "active", lastActivityAt: ts }), + makeSession({ id: "session-2", status: "working", activity: "active", lastActivityAt: ts }), ]; - render(); + render(); expect(renderCounts.get("session-1")).toBe(1); expect(renderCounts.get("session-2")).toBe(1); - await waitFor(() => expect(eventSourceMock.onmessage).not.toBeNull()); - + // Push a snapshot where only session-1 changes await act(async () => { - eventSourceMock.onmessage!({ - data: JSON.stringify({ - type: "snapshot", - sessions: [ - { - id: "session-1", - status: "working", - activity: "idle", - lastActivityAt: new Date().toISOString(), - }, - { - id: "session-2", - status: initialSessions[1].status, - activity: initialSessions[1].activity, - lastActivityAt: initialSessions[1].lastActivityAt, - }, - ], - }), - } as MessageEvent); + currentMuxSessions = [ + { + id: "session-1", + status: "working", + activity: "idle", + attentionLevel: "working", + lastActivityAt: new Date(Date.now() + 1000).toISOString(), + }, + { + id: "session-2", + status: "working", + activity: "active", + attentionLevel: "working", + lastActivityAt: ts, + }, + ]; + forceUpdate(); }); expect(renderCounts.get("session-1")).toBe(2); expect(renderCounts.get("session-2")).toBe(1); expect(fetch).not.toHaveBeenCalled(); }); + + it("does not rerender any card when snapshot data is identical", async () => { + const ts = new Date().toISOString(); + const initialSessions = [ + makeSession({ id: "session-1", status: "working", activity: "active", lastActivityAt: ts }), + ]; + + render(); + const countAfterInit = renderCounts.get("session-1") ?? 0; + + await act(async () => { + currentMuxSessions = [ + { + id: "session-1", + status: "working", + activity: "active", + attentionLevel: "working", + lastActivityAt: ts, + }, + ]; + forceUpdate(); + }); + + expect(renderCounts.get("session-1")).toBe(countAfterInit); + }); }); diff --git a/packages/web/src/components/__tests__/DynamicFavicon.test.tsx b/packages/web/src/components/__tests__/DynamicFavicon.test.tsx index 0c48802ca..2ddf5e05a 100644 --- a/packages/web/src/components/__tests__/DynamicFavicon.test.tsx +++ b/packages/web/src/components/__tests__/DynamicFavicon.test.tsx @@ -1,7 +1,7 @@ import { describe, it, expect, beforeEach } from "vitest"; import { render } from "@testing-library/react"; import { countNeedingAttention, DynamicFavicon } from "../DynamicFavicon"; -import type { SSEAttentionMap } from "@/hooks/useSessionEvents"; +import type { AttentionMap } from "@/hooks/useSessionEvents"; describe("countNeedingAttention", () => { it("returns 0 for empty map", () => { @@ -9,7 +9,7 @@ describe("countNeedingAttention", () => { }); it("returns 0 when all sessions are working/pending/done", () => { - const levels: SSEAttentionMap = { + const levels: AttentionMap = { "s-1": "working", "s-2": "pending", "s-3": "done", @@ -18,7 +18,7 @@ describe("countNeedingAttention", () => { }); it("counts respond, review, action, and merge sessions", () => { - const levels: SSEAttentionMap = { + const levels: AttentionMap = { "s-1": "respond", "s-2": "review", "s-3": "merge", @@ -30,7 +30,7 @@ describe("countNeedingAttention", () => { }); it("counts a single attention-needing session", () => { - const levels: SSEAttentionMap = { + const levels: AttentionMap = { "s-1": "respond", }; expect(countNeedingAttention(levels)).toBe(1); @@ -44,8 +44,8 @@ describe("DynamicFavicon", () => { }); it("creates a green favicon when no sessions need attention", () => { - const levels: SSEAttentionMap = { "s-1": "working", "s-2": "done" }; - render(); + const levels: AttentionMap = { "s-1": "working", "s-2": "done" }; + render(); const link = document.querySelector('link[rel="icon"]'); expect(link).not.toBeNull(); @@ -53,16 +53,16 @@ describe("DynamicFavicon", () => { }); it("creates a yellow favicon when sessions need review", () => { - const levels: SSEAttentionMap = { "s-1": "review", "s-2": "working" }; - render(); + const levels: AttentionMap = { "s-1": "review", "s-2": "working" }; + render(); const link = document.querySelector('link[rel="icon"]'); expect(link!.href).toContain("%23eab308"); // yellow }); it("creates a red favicon when sessions need response", () => { - const levels: SSEAttentionMap = { "s-1": "respond" }; - render(); + const levels: AttentionMap = { "s-1": "respond" }; + render(); const link = document.querySelector('link[rel="icon"]'); expect(link!.href).toContain("%23ef4444"); // red @@ -72,32 +72,32 @@ describe("DynamicFavicon", () => { // "action" collapses respond + review, so it contains routine review work // (ci_failed, changes_requested). Escalating to red would cry wolf on // every typical PR. - const levels: SSEAttentionMap = { "s-1": "action", "s-2": "working" }; - render(); + const levels: AttentionMap = { "s-1": "action", "s-2": "working" }; + render(); const link = document.querySelector('link[rel="icon"]'); expect(link!.href).toContain("%23eab308"); // yellow }); it("still escalates to red when detailed 'respond' is present alongside 'action'", () => { - const levels: SSEAttentionMap = { "s-1": "action", "s-2": "respond" }; - render(); + const levels: AttentionMap = { "s-1": "action", "s-2": "respond" }; + render(); const link = document.querySelector('link[rel="icon"]'); expect(link!.href).toContain("%23ef4444"); // red }); it("uses first letter of projectName as initial", () => { - const levels: SSEAttentionMap = {}; - render(); + const levels: AttentionMap = {}; + render(); const link = document.querySelector('link[rel="icon"]'); expect(link!.href).toContain("M"); // initial letter }); it("defaults to A when no projectName given", () => { - const levels: SSEAttentionMap = {}; - render(); + const levels: AttentionMap = {}; + render(); const link = document.querySelector('link[rel="icon"]'); expect(link!.href).toContain("A"); @@ -105,13 +105,13 @@ describe("DynamicFavicon", () => { it("updates favicon when attention levels change", () => { const { rerender } = render( - , + , ); let link = document.querySelector('link[rel="icon"]'); expect(link!.href).toContain("%2322c55e"); // green - rerender(); + rerender(); link = document.querySelector('link[rel="icon"]'); expect(link!.href).toContain("%23ef4444"); // red diff --git a/packages/web/src/hooks/__tests__/useSSESessionActivity.test.ts b/packages/web/src/hooks/__tests__/useSSESessionActivity.test.ts deleted file mode 100644 index 3f2102990..000000000 --- a/packages/web/src/hooks/__tests__/useSSESessionActivity.test.ts +++ /dev/null @@ -1,207 +0,0 @@ -import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; -import { renderHook, act } from "@testing-library/react"; -import { useSSESessionActivity } from "../useSSESessionActivity"; - -describe("useSSESessionActivity", () => { - let eventSourceMock: { - onmessage: ((event: MessageEvent) => void) | null; - onopen: (() => void) | null; - onerror: (() => void) | null; - readyState: number; - close: ReturnType; - }; - - beforeEach(() => { - const eventSourceConstructor = vi.fn(() => { - const instance = { - onmessage: null as ((event: MessageEvent) => void) | null, - onopen: null as (() => void) | null, - onerror: null as (() => void) | null, - readyState: 0, - close: vi.fn(), - }; - eventSourceMock = instance; - return instance as unknown as EventSource; - }); - global.EventSource = Object.assign(eventSourceConstructor, { - CONNECTING: 0, - OPEN: 1, - CLOSED: 2, - }) as unknown as typeof EventSource; - }); - - afterEach(() => { - vi.restoreAllMocks(); - }); - - it("returns null initially", () => { - const { result } = renderHook(() => useSSESessionActivity("session-1")); - expect(result.current).toBeNull(); - }); - - it("updates activity when SSE snapshot contains the target session", () => { - const { result } = renderHook(() => useSSESessionActivity("session-1")); - - act(() => { - eventSourceMock.onmessage!.call(eventSourceMock, { - data: JSON.stringify({ - type: "snapshot", - sessions: [ - { id: "session-1", status: "working", activity: "active", attentionLevel: "working", lastActivityAt: new Date().toISOString() }, - { id: "session-2", status: "working", activity: "idle", attentionLevel: "working", lastActivityAt: new Date().toISOString() }, - ], - }), - } as MessageEvent); - }); - - expect(result.current).toEqual({ activity: "active" }); - }); - - it("updates when activity changes", () => { - const { result } = renderHook(() => useSSESessionActivity("session-1")); - - act(() => { - eventSourceMock.onmessage!.call(eventSourceMock, { - data: JSON.stringify({ - type: "snapshot", - sessions: [ - { id: "session-1", status: "working", activity: "active", attentionLevel: "working", lastActivityAt: new Date().toISOString() }, - ], - }), - } as MessageEvent); - }); - - expect(result.current?.activity).toBe("active"); - - act(() => { - eventSourceMock.onmessage!.call(eventSourceMock, { - data: JSON.stringify({ - type: "snapshot", - sessions: [ - { id: "session-1", status: "working", activity: "idle", attentionLevel: "working", lastActivityAt: new Date().toISOString() }, - ], - }), - } as MessageEvent); - }); - - expect(result.current?.activity).toBe("idle"); - }); - - it("does not update when activity stays the same (referential stability)", () => { - const { result } = renderHook(() => useSSESessionActivity("session-1")); - - act(() => { - eventSourceMock.onmessage!.call(eventSourceMock, { - data: JSON.stringify({ - type: "snapshot", - sessions: [ - { id: "session-1", status: "working", activity: "active", attentionLevel: "working", lastActivityAt: new Date().toISOString() }, - ], - }), - } as MessageEvent); - }); - - const first = result.current; - - act(() => { - eventSourceMock.onmessage!.call(eventSourceMock, { - data: JSON.stringify({ - type: "snapshot", - sessions: [ - { id: "session-1", status: "working", activity: "active", attentionLevel: "working", lastActivityAt: new Date().toISOString() }, - ], - }), - } as MessageEvent); - }); - - expect(result.current).toBe(first); - }); - - it("ignores snapshots that do not contain the session", () => { - const { result } = renderHook(() => useSSESessionActivity("session-1")); - - act(() => { - eventSourceMock.onmessage!.call(eventSourceMock, { - data: JSON.stringify({ - type: "snapshot", - sessions: [ - { id: "session-other", status: "working", activity: "active", attentionLevel: "working", lastActivityAt: new Date().toISOString() }, - ], - }), - } as MessageEvent); - }); - - expect(result.current).toBeNull(); - }); - - it("passes project parameter to EventSource URL", () => { - renderHook(() => useSSESessionActivity("session-1", "my-project")); - expect(global.EventSource).toHaveBeenCalledWith("/api/events?project=my-project"); - }); - - it("uses default URL when no project is specified", () => { - renderHook(() => useSSESessionActivity("session-1")); - expect(global.EventSource).toHaveBeenCalledWith("/api/events"); - }); - - it("closes EventSource on unmount", () => { - const { unmount } = renderHook(() => useSSESessionActivity("session-1")); - unmount(); - expect(eventSourceMock.close).toHaveBeenCalled(); - }); - - it("ignores non-snapshot events", () => { - const { result } = renderHook(() => useSSESessionActivity("session-1")); - - act(() => { - eventSourceMock.onmessage!.call(eventSourceMock, { - data: JSON.stringify({ type: "heartbeat" }), - } as MessageEvent); - }); - - expect(result.current).toBeNull(); - }); - - it("handles malformed JSON gracefully", () => { - const { result } = renderHook(() => useSSESessionActivity("session-1")); - - act(() => { - eventSourceMock.onmessage!.call(eventSourceMock, { - data: "not-json", - } as MessageEvent); - }); - - expect(result.current).toBeNull(); - }); - - it("resets activity and reconnects when sessionId changes", () => { - let currentSessionId = "session-1"; - const { result, rerender } = renderHook(() => - useSSESessionActivity(currentSessionId), - ); - - const firstInstance = eventSourceMock; - - act(() => { - firstInstance.onmessage!.call(firstInstance, { - data: JSON.stringify({ - type: "snapshot", - sessions: [ - { id: "session-1", status: "working", activity: "active", attentionLevel: "working", lastActivityAt: new Date().toISOString() }, - ], - }), - } as MessageEvent); - }); - - expect(result.current?.activity).toBe("active"); - - currentSessionId = "session-2"; - rerender(); - - // State should reset to null immediately on sessionId change - expect(result.current).toBeNull(); - - // First EventSource should have been closed - expect(firstInstance.close).toHaveBeenCalled(); - }); -}); diff --git a/packages/web/src/hooks/useMuxSessionActivity.ts b/packages/web/src/hooks/useMuxSessionActivity.ts new file mode 100644 index 000000000..34d39804c --- /dev/null +++ b/packages/web/src/hooks/useMuxSessionActivity.ts @@ -0,0 +1,16 @@ +"use client"; + +import { useMemo } from "react"; +import type { ActivityState } from "@/lib/types"; +import { useMuxOptional } from "@/providers/MuxProvider"; + +export function useMuxSessionActivity( + sessionId: string, +): { activity: ActivityState | null } | null { + const mux = useMuxOptional(); + const patch = mux?.sessions.find((s) => s.id === sessionId); + return useMemo( + () => (patch ? { activity: (patch.activity as ActivityState) ?? null } : null), + [patch], + ); +} diff --git a/packages/web/src/hooks/useSSESessionActivity.ts b/packages/web/src/hooks/useSSESessionActivity.ts deleted file mode 100644 index b528b4318..000000000 --- a/packages/web/src/hooks/useSSESessionActivity.ts +++ /dev/null @@ -1,54 +0,0 @@ -"use client"; - -import { useEffect, useState } from "react"; -import type { ActivityState, SSESnapshotEvent } from "@/lib/types"; - -interface SessionActivity { - activity: ActivityState | null; -} - -/** - * Lightweight SSE subscriber that tracks a single session's activity state. - * - * Used by the session detail page to update document.title emoji in real-time - * without waiting for the full session HTTP poll cycle. - */ -export function useSSESessionActivity( - sessionId: string, - project?: string, -): SessionActivity | null { - const [activity, setActivity] = useState(null); - - useEffect(() => { - setActivity(null); - const url = project ? `/api/events?project=${encodeURIComponent(project)}` : "/api/events"; - const es = new EventSource(url); - let disposed = false; - - es.onmessage = (event: MessageEvent) => { - if (disposed) return; - try { - const data = JSON.parse(event.data as string) as { type: string }; - if (data.type !== "snapshot") return; - - const snapshot = data as SSESnapshotEvent; - const match = snapshot.sessions.find((s) => s.id === sessionId); - if (!match) return; - - setActivity((prev) => { - if (prev && prev.activity === match.activity) return prev; - return { activity: match.activity }; - }); - } catch { - return; - } - }; - - return () => { - disposed = true; - es.close(); - }; - }, [sessionId, project]); - - return activity; -} diff --git a/packages/web/src/hooks/useSessionEvents.ts b/packages/web/src/hooks/useSessionEvents.ts index d2df890d9..afe4aed7c 100644 Binary files a/packages/web/src/hooks/useSessionEvents.ts and b/packages/web/src/hooks/useSessionEvents.ts differ diff --git a/packages/web/src/lib/mux-protocol.ts b/packages/web/src/lib/mux-protocol.ts index 19e625784..c406c6af2 100644 --- a/packages/web/src/lib/mux-protocol.ts +++ b/packages/web/src/lib/mux-protocol.ts @@ -18,6 +18,7 @@ export type ServerMessage = | { ch: "terminal"; id: string; type: "opened" } | { ch: "terminal"; id: string; type: "error"; message: string } | { ch: "sessions"; type: "snapshot"; sessions: SessionPatch[] } + | { ch: "sessions"; type: "error"; error: string } | { ch: "system"; type: "pong" } | { ch: "system"; type: "error"; message: string }; diff --git a/packages/web/src/lib/types.ts b/packages/web/src/lib/types.ts index e4a530548..86c46b660 100644 --- a/packages/web/src/lib/types.ts +++ b/packages/web/src/lib/types.ts @@ -246,30 +246,6 @@ export interface DashboardOrchestratorLink { projectName: string; } -/** SSE snapshot event from /api/events */ -export interface SSESnapshotEvent { - type: "snapshot"; - correlationId?: string; - emittedAt?: string; - sessions: Array<{ - id: string; - status: SessionStatus; - activity: ActivityState | null; - attentionLevel: AttentionLevel; - lastActivityAt: string; - }>; -} - -/** SSE activity update event from /api/events */ -export interface SSEActivityEvent { - type: "session.activity"; - sessionId: string; - activity: ActivityState | null; - status: SessionStatus; - attentionLevel: AttentionLevel; - timestamp: string; -} - /** * Returns true when this PR's enrichment data couldn't be fetched due to * API rate limiting. When true, CI status / review decision / mergeability diff --git a/packages/web/src/providers/MuxProvider.tsx b/packages/web/src/providers/MuxProvider.tsx index 66bb9ed57..43c77f867 100644 --- a/packages/web/src/providers/MuxProvider.tsx +++ b/packages/web/src/providers/MuxProvider.tsx @@ -11,6 +11,8 @@ interface MuxContextValue { resizeTerminal: (id: string, cols: number, rows: number) => void; status: "connecting" | "connected" | "reconnecting" | "disconnected"; sessions: SessionPatch[]; + /** Last session-fetch error from the server, null when healthy. */ + lastError: string | null; } const MuxContext = React.createContext(undefined); @@ -76,6 +78,7 @@ export function MuxProvider({ children }: { children: ReactNode }) { "connecting", ); const [sessions, setSessions] = useState([]); + const [lastError, setLastError] = useState(null); const reconnectAttempt = useRef(0); const reconnectTimer = useRef | null>(null); const runtimeConfigRef = useRef<{ directTerminalPort?: string; proxyWsPath?: string }>({}); @@ -155,8 +158,13 @@ export function MuxProvider({ children }: { children: ReactNode }) { } else if (msg.type === "error") { console.error(`[MuxProvider] Terminal error for ${msg.id}:`, msg.message); } - } else if (msg.ch === "sessions" && msg.type === "snapshot") { - setSessions(msg.sessions); + } else if (msg.ch === "sessions") { + if (msg.type === "snapshot") { + setSessions(msg.sessions); + setLastError(null); + } else if (msg.type === "error") { + setLastError(msg.error); + } } } catch (err) { console.error("[MuxProvider] Error processing message:", err); @@ -319,8 +327,9 @@ export function MuxProvider({ children }: { children: ReactNode }) { resizeTerminal, status, sessions, + lastError, }), - [subscribeTerminal, writeTerminal, openTerminal, closeTerminal, resizeTerminal, status, sessions], + [subscribeTerminal, writeTerminal, openTerminal, closeTerminal, resizeTerminal, status, sessions, lastError], ); return {children};