Merge remote-tracking branch 'origin/main' into feat/windows-platform-adapter
This commit is contained in:
Priyanshu Choudhary
commit
daa8edf69b
|
|
@ -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` |
|
||||
|
|
@ -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<typeof vi.spyOn>;
|
||||
|
|
@ -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) => {
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
||||
|
|
|
|||
|
|
@ -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", () => {
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<typeof setTimeout> | null = null;
|
||||
private errorSubscribers = new Set<(error: string) => void>();
|
||||
private intervalId: ReturnType<typeof setInterval> | 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<SessionPatch[] | null> {
|
||||
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<void> {
|
||||
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) {
|
||||
|
|
|
|||
|
|
@ -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" });
|
||||
|
|
|
|||
|
|
@ -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<Response> {
|
||||
const encoder = new TextEncoder();
|
||||
const correlationId = createCorrelationId("sse");
|
||||
const { searchParams } = new URL(request.url);
|
||||
const projectFilter = searchParams.get("project");
|
||||
type ServicesConfig = Awaited<ReturnType<typeof getServices>>["config"];
|
||||
let heartbeat: ReturnType<typeof setInterval> | undefined;
|
||||
let updates: ReturnType<typeof setInterval> | 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",
|
||||
},
|
||||
});
|
||||
}
|
||||
|
|
@ -31,6 +31,10 @@ vi.mock("@/components/SessionDetail", () => ({
|
|||
},
|
||||
}));
|
||||
|
||||
vi.mock("@/hooks/useMuxSessionActivity", () => ({
|
||||
useMuxSessionActivity: () => null,
|
||||
}));
|
||||
|
||||
function makeWorkerSession(): DashboardSession {
|
||||
return {
|
||||
id: "worker-1",
|
||||
|
|
|
|||
|
|
@ -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(() => {
|
||||
|
|
|
|||
|
|
@ -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 ? (
|
||||
<div
|
||||
className="dashboard-alert mb-6 flex flex-col gap-1.5 border border-[color-mix(in_srgb,var(--color-status-error)_28%,transparent)] bg-[color-mix(in_srgb,var(--color-status-error)_10%,transparent)] px-3.5 py-2.5 text-[11px] md:mb-4"
|
||||
role="alert"
|
||||
|
|
@ -454,7 +459,7 @@ function DashboardInner({
|
|||
Orchestrator failed to load
|
||||
</span>
|
||||
<span className="break-words text-[var(--color-text-secondary)]">
|
||||
{visibleDashboardLoadError}
|
||||
{visibleLoadError}
|
||||
</span>
|
||||
<span className="text-[var(--color-text-secondary)]">
|
||||
Confirm <span className="font-mono text-[10px]">agent-orchestrator.yaml</span> exists and is
|
||||
|
|
@ -612,8 +617,8 @@ function DashboardInner({
|
|||
<div className="sidebar-mobile-backdrop" onClick={() => setMobileMenuOpen(false)} />
|
||||
)}
|
||||
|
||||
<main className="dashboard-main dashboard-main--desktop overflow-y-auto">
|
||||
<DynamicFavicon sseAttentionLevels={sseAttentionLevels} projectName={projectName} />
|
||||
<main className="dashboard-main dashboard-main--desktop">
|
||||
<DynamicFavicon attentionLevels={attentionLevels} projectName={projectName} />
|
||||
<div className="dashboard-main__subhead">
|
||||
<h1 className="dashboard-main__title">Dashboard</h1>
|
||||
<p className="dashboard-main__subtitle">
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<string, ReturnType<typeof getAttentionLevel>> = {};
|
||||
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({
|
|||
<div className="sidebar-mobile-backdrop" onClick={() => setMobileMenuOpen(false)} />
|
||||
)}
|
||||
<div className="dashboard-main flex-1 overflow-y-auto px-4 py-4 md:px-7 md:py-6">
|
||||
<DynamicFavicon sseAttentionLevels={sseAttentionLevels} projectName={projectName ? `${projectName} PRs` : "Pull Requests"} />
|
||||
<DynamicFavicon attentionLevels={attentionLevels} projectName={projectName ? `${projectName} PRs` : "Pull Requests"} />
|
||||
{isMobile ? (
|
||||
<section className="mobile-pr-page-header">
|
||||
<div className="mobile-pr-page-header__top">
|
||||
|
|
|
|||
|
|
@ -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<typeof vi.fn>;
|
||||
};
|
||||
|
||||
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(<Dashboard initialSessions={[]} />);
|
||||
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 <Dashboard initialSessions={[]} />;
|
||||
}
|
||||
|
||||
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(<Wrapper />);
|
||||
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(<Dashboard initialSessions={[]} />);
|
||||
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 <Dashboard initialSessions={[]} />;
|
||||
}
|
||||
|
||||
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(<Wrapper />);
|
||||
|
||||
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", () => {
|
||||
|
|
|
|||
|
|
@ -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<string, number>();
|
||||
|
|
@ -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<typeof makeSession>[];
|
||||
}) {
|
||||
const [, tick] = useState(0);
|
||||
forceUpdate = () => tick((n) => n + 1);
|
||||
return <Dashboard initialSessions={initialSessions} />;
|
||||
}
|
||||
|
||||
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(<Dashboard initialSessions={initialSessions} />);
|
||||
render(<ControlledDashboard initialSessions={initialSessions} />);
|
||||
|
||||
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(<ControlledDashboard initialSessions={initialSessions} />);
|
||||
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);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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(<DynamicFavicon sseAttentionLevels={levels} projectName="Test" />);
|
||||
const levels: AttentionMap = { "s-1": "working", "s-2": "done" };
|
||||
render(<DynamicFavicon attentionLevels={levels} projectName="Test" />);
|
||||
|
||||
const link = document.querySelector<HTMLLinkElement>('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(<DynamicFavicon sseAttentionLevels={levels} projectName="Test" />);
|
||||
const levels: AttentionMap = { "s-1": "review", "s-2": "working" };
|
||||
render(<DynamicFavicon attentionLevels={levels} projectName="Test" />);
|
||||
|
||||
const link = document.querySelector<HTMLLinkElement>('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(<DynamicFavicon sseAttentionLevels={levels} projectName="Test" />);
|
||||
const levels: AttentionMap = { "s-1": "respond" };
|
||||
render(<DynamicFavicon attentionLevels={levels} projectName="Test" />);
|
||||
|
||||
const link = document.querySelector<HTMLLinkElement>('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(<DynamicFavicon sseAttentionLevels={levels} projectName="Test" />);
|
||||
const levels: AttentionMap = { "s-1": "action", "s-2": "working" };
|
||||
render(<DynamicFavicon attentionLevels={levels} projectName="Test" />);
|
||||
|
||||
const link = document.querySelector<HTMLLinkElement>('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(<DynamicFavicon sseAttentionLevels={levels} projectName="Test" />);
|
||||
const levels: AttentionMap = { "s-1": "action", "s-2": "respond" };
|
||||
render(<DynamicFavicon attentionLevels={levels} projectName="Test" />);
|
||||
|
||||
const link = document.querySelector<HTMLLinkElement>('link[rel="icon"]');
|
||||
expect(link!.href).toContain("%23ef4444"); // red
|
||||
});
|
||||
|
||||
it("uses first letter of projectName as initial", () => {
|
||||
const levels: SSEAttentionMap = {};
|
||||
render(<DynamicFavicon sseAttentionLevels={levels} projectName="MyApp" />);
|
||||
const levels: AttentionMap = {};
|
||||
render(<DynamicFavicon attentionLevels={levels} projectName="MyApp" />);
|
||||
|
||||
const link = document.querySelector<HTMLLinkElement>('link[rel="icon"]');
|
||||
expect(link!.href).toContain("M"); // initial letter
|
||||
});
|
||||
|
||||
it("defaults to A when no projectName given", () => {
|
||||
const levels: SSEAttentionMap = {};
|
||||
render(<DynamicFavicon sseAttentionLevels={levels} />);
|
||||
const levels: AttentionMap = {};
|
||||
render(<DynamicFavicon attentionLevels={levels} />);
|
||||
|
||||
const link = document.querySelector<HTMLLinkElement>('link[rel="icon"]');
|
||||
expect(link!.href).toContain("A");
|
||||
|
|
@ -105,13 +105,13 @@ describe("DynamicFavicon", () => {
|
|||
|
||||
it("updates favicon when attention levels change", () => {
|
||||
const { rerender } = render(
|
||||
<DynamicFavicon sseAttentionLevels={{ "s-1": "working" }} projectName="Test" />,
|
||||
<DynamicFavicon attentionLevels={{ "s-1": "working" }} projectName="Test" />,
|
||||
);
|
||||
|
||||
let link = document.querySelector<HTMLLinkElement>('link[rel="icon"]');
|
||||
expect(link!.href).toContain("%2322c55e"); // green
|
||||
|
||||
rerender(<DynamicFavicon sseAttentionLevels={{ "s-1": "respond" }} projectName="Test" />);
|
||||
rerender(<DynamicFavicon attentionLevels={{ "s-1": "respond" }} projectName="Test" />);
|
||||
|
||||
link = document.querySelector<HTMLLinkElement>('link[rel="icon"]');
|
||||
expect(link!.href).toContain("%23ef4444"); // red
|
||||
|
|
|
|||
|
|
@ -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<typeof vi.fn>;
|
||||
};
|
||||
|
||||
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();
|
||||
});
|
||||
});
|
||||
|
|
@ -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],
|
||||
);
|
||||
}
|
||||
|
|
@ -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<SessionActivity | null>(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;
|
||||
}
|
||||
Binary file not shown.
|
|
@ -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 };
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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<MuxContextValue | undefined>(undefined);
|
||||
|
|
@ -76,6 +78,7 @@ export function MuxProvider({ children }: { children: ReactNode }) {
|
|||
"connecting",
|
||||
);
|
||||
const [sessions, setSessions] = useState<SessionPatch[]>([]);
|
||||
const [lastError, setLastError] = useState<string | null>(null);
|
||||
const reconnectAttempt = useRef(0);
|
||||
const reconnectTimer = useRef<ReturnType<typeof setTimeout> | 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 <MuxContext.Provider value={contextValue}>{children}</MuxContext.Provider>;
|
||||
|
|
|
|||
Loading…
Reference in New Issue