bkinnightskytw
/
agent-orchestrator
mirror of https://github.com/ComposioHQ/agent-orchestrator.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: resolve merge conflict with upstream WebSocket reconnection 

Keep upstream's connectWebSocket() reconnection structure and add our
clipboard shortcuts (Cmd+C/V, Ctrl+Shift+C/V) and OSC 52 handler on
top. Clipboard paste now uses ws.current to match upstream's pattern
where the WebSocket reference can change across reconnects.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
suraj-markup 2026-03-02 00:41:13 +05:30
parent 54c0a075bc ca3286ca37
commit 941a99e0c9
11 changed files with 463 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

130
packages/cli/__tests__/commands/init.test.ts
View File

@ -5,6 +5,7 @@ import { tmpdir } from "node:os";
import { createServer } from "node:net";
import { Command } from "commander";
import { parse as yamlParse } from "yaml";
import { registerInit } from "../../src/commands/init.js";
let tmpDir: string;
@ -41,6 +42,10 @@ describe("init command", () => {
tmpDir = mkdtempSync(join(tmpdir(), "ao-init-test-"));
const outputPath = join(tmpDir, "agent-orchestrator.yaml");
// Mock isPortAvailable so test doesn't depend on actual port 3000 being free
const webDirModule = await import("../../src/lib/web-dir.js");
vi.spyOn(webDirModule, "isPortAvailable").mockResolvedValue(true);
const program = new Command();
program.exitOverride();
registerInit(program);
@ -85,4 +90,129 @@ describe("init command", () => {
await new Promise<void>((resolve) => blocker.close(() => resolve()));
}
});
it("auto mode tells user when default port is busy and which port it picked", async () => {
tmpDir = mkdtempSync(join(tmpdir(), "ao-init-test-"));
const outputPath = join(tmpDir, "agent-orchestrator.yaml");
// Occupy port 3000
const blocker = createServer();
await new Promise<void>((resolve) => {
blocker.listen(3000, "127.0.0.1", () => resolve());
});
try {
const logSpy = vi.spyOn(console, "log").mockImplementation(() => {});
const program = new Command();
program.exitOverride();
registerInit(program);
await program.parseAsync(["node", "test", "init", "--auto", "--output", outputPath]);
const logCalls = logSpy.mock.calls.map((args) => args.join(" "));
const busyMessage = logCalls.find((msg) => msg.includes("Port 3000 is busy"));
expect(busyMessage).toBeDefined();
expect(busyMessage).toContain("instead");
} finally {
await new Promise<void>((resolve) => blocker.close(() => resolve()));
}
});
it("auto mode writes name and sessionPrefix to project config", async () => {
tmpDir = mkdtempSync(join(tmpdir(), "ao-init-test-"));
const outputPath = join(tmpDir, "agent-orchestrator.yaml");
const program = new Command();
program.exitOverride();
registerInit(program);
vi.spyOn(console, "log").mockImplementation(() => {});
await program.parseAsync(["node", "test", "init", "--auto", "--output", outputPath]);
const content = readFileSync(outputPath, "utf-8");
const config = yamlParse(content) as Record<string, unknown>;
const projects = config.projects as Record<string, Record<string, unknown>>;
const projectIds = Object.keys(projects);
expect(projectIds.length).toBeGreaterThan(0);
const projectId = projectIds[0];
const project = projects[projectId];
// BUG-01: sessionPrefix must exist and not be "undefined"
expect(project.sessionPrefix).toBeDefined();
expect(project.sessionPrefix).not.toBe("undefined");
expect(typeof project.sessionPrefix).toBe("string");
expect((project.sessionPrefix as string).length).toBeGreaterThan(0);
// BUG-02: name must match the project ID
expect(project.name).toBe(projectId);
});
it("auto mode warns when no free port is found", async () => {
tmpDir = mkdtempSync(join(tmpdir(), "ao-init-test-"));
const outputPath = join(tmpDir, "agent-orchestrator.yaml");
// Mock isPortAvailable to always return false (all ports busy)
const webDirModule = await import("../../src/lib/web-dir.js");
vi.spyOn(webDirModule, "isPortAvailable").mockResolvedValue(false);
const logSpy = vi.spyOn(console, "log").mockImplementation(() => {});
const program = new Command();
program.exitOverride();
registerInit(program);
await program.parseAsync(["node", "test", "init", "--auto", "--output", outputPath]);
// Should warn about no free port
const logCalls = logSpy.mock.calls.map((args) => args.join(" "));
const hasPortWarning = logCalls.some((msg) => msg.includes("No free port found"));
expect(hasPortWarning).toBe(true);
// Should still write config with fallback port 3000
const content = readFileSync(outputPath, "utf-8");
expect(content).toContain("port: 3000");
});
it("--smart flag description includes 'coming soon'", () => {
const program = new Command();
registerInit(program);
const initCmd = program.commands.find((c) => c.name() === "init");
expect(initCmd).toBeDefined();
const smartOption = initCmd!.options.find((o) => o.long === "--smart");
expect(smartOption).toBeDefined();
expect(smartOption!.description).toContain("coming soon");
});
it("auto mode sessionPrefix uses core generateSessionPrefix heuristics", async () => {
tmpDir = mkdtempSync(join(tmpdir(), "ao-init-test-"));
const outputPath = join(tmpDir, "agent-orchestrator.yaml");
// Run from a directory whose basename is kebab-case
// The cwd() in init.ts determines the projectId via basename()
// We can't easily control cwd in tests, but we can verify the prefix
// is consistent with generateSessionPrefix for whatever projectId is used
const { generateSessionPrefix } = await import("@composio/ao-core");
const program = new Command();
program.exitOverride();
registerInit(program);
vi.spyOn(console, "log").mockImplementation(() => {});
await program.parseAsync(["node", "test", "init", "--auto", "--output", outputPath]);
const content = readFileSync(outputPath, "utf-8");
const config = yamlParse(content) as Record<string, unknown>;
const projects = config.projects as Record<string, Record<string, unknown>>;
const projectId = Object.keys(projects)[0];
const project = projects[projectId];
// sessionPrefix must match what generateSessionPrefix produces
expect(project.sessionPrefix).toBe(generateSessionPrefix(projectId));
});
});

40
packages/cli/src/commands/init.ts
View File

@ -5,6 +5,7 @@ import { cwd } from "node:process";
import { stringify as yamlStringify } from "yaml";
import chalk from "chalk";
import type { Command } from "commander";
import { generateSessionPrefix } from "@composio/ao-core";
import { git, gh, execSilent } from "../lib/shell.js";
import { isPortAvailable } from "../lib/web-dir.js";
import {
@ -16,12 +17,12 @@ import {
const DEFAULT_PORT = 3000;
const MAX_PORT_SCAN = 100;
/** Find the first available port starting from `start`, scanning upward. */
async function findFreePort(start: number): Promise<number> {
/** Find the first available port starting from `start`, scanning upward. Returns `null` if none found. */
async function findFreePort(start: number): Promise<number | null> {
for (let port = start; port < start + MAX_PORT_SCAN; port++) {
if (await isPortAvailable(port)) return port;
}
return start; // fallback — will fail at bind time with a clear error
return null;
}
async function prompt(
@ -157,7 +158,7 @@ export function registerInit(program: Command): void {
.option("--auto", "Auto-generate config with sensible defaults (no prompts)")
.option(
"--smart",
"Analyze project and generate custom rules (requires --auto, uses AI if available)",
"Analyze project and generate custom rules (coming soon — requires --auto)",
)
.action(async (opts: { output: string; auto?: boolean; smart?: boolean }) => {
const outputPath = resolve(opts.output);
@ -244,7 +245,20 @@ export function registerInit(program: Command): void {
);
const worktreeDir = await prompt(rl, "Worktree directory", "~/.worktrees");
const freePort = await findFreePort(DEFAULT_PORT);
const portStr = await prompt(rl, "Dashboard port", String(freePort));
if (freePort === null) {
console.log(
chalk.yellow(
`\n⚠ No free port found in range ${DEFAULT_PORT}–${DEFAULT_PORT + MAX_PORT_SCAN - 1}.`,
),
);
console.log(chalk.dim(" Please specify a port manually.\n"));
} else if (freePort !== DEFAULT_PORT) {
console.log(
chalk.yellow(`\n⚠ Port ${DEFAULT_PORT} is busy — suggesting ${freePort} instead.`),
);
console.log(chalk.dim(" Press Enter to accept, or type a different port.\n"));
}
const portStr = await prompt(rl, "Dashboard port", String(freePort ?? DEFAULT_PORT));
const port = parseInt(portStr, 10);
if (isNaN(port) || port < 1 || port > 65535) {
console.error(chalk.red("\nInvalid port number. Must be 1-65535."));
@ -303,6 +317,8 @@ export function registerInit(program: Command): void {
);
const projectConfig: Record<string, unknown> = {
name: projectId,
sessionPrefix: generateSessionPrefix(projectId),
repo,
path: projectPath,
defaultBranch,
@ -444,10 +460,20 @@ async function handleAutoMode(outputPath: string, smart: boolean): Promise<void>
const defaultBranch = env.defaultBranch || "main";
const port = await findFreePort(DEFAULT_PORT);
if (port === null) {
console.log(
chalk.yellow(
` ⚠ No free port found in range ${DEFAULT_PORT}–${DEFAULT_PORT + MAX_PORT_SCAN - 1}.`,
),
);
console.log(chalk.dim(" Set the port manually in the config before running ao start.\n"));
} else if (port !== DEFAULT_PORT) {
console.log(chalk.yellow(` ⚠ Port ${DEFAULT_PORT} is busy — using ${port} instead.`));
}
const config: Record<string, unknown> = {
dataDir: "~/.agent-orchestrator",
worktreeDir: "~/.worktrees",
port,
port: port ?? DEFAULT_PORT,
defaults: {
runtime: "tmux",
agent: "claude-code",
@ -456,6 +482,8 @@ async function handleAutoMode(outputPath: string, smart: boolean): Promise<void>
},
projects: {
[projectId]: {
name: projectId,
sessionPrefix: generateSessionPrefix(projectId),
repo,
path,
defaultBranch,

25
packages/cli/src/lib/web-dir.ts
View File

@ -3,7 +3,7 @@
* Shared utility to avoid duplication between dashboard.ts and start.ts.
*/
import { createServer } from "node:net";
import { Socket } from "node:net";
import { createRequire } from "node:module";
import { fileURLToPath } from "node:url";
import { resolve, dirname } from "node:path";
@ -17,19 +17,22 @@ const require = createRequire(import.meta.url);
const DEFAULT_TERMINAL_PORT = 14800;
/**
* Check if a TCP port is available by attempting to bind to it.
* Returns true if the port is free, false if in use.
* Check if a TCP port is available by attempting to connect to it on IPv4.
* A successful connect means something is already listening (port in use).
* ECONNREFUSED means nothing is listening (port free).
*
* Note: Only probes 127.0.0.1 (IPv4). Processes listening exclusively on
* IPv6 (::1 with IPV6_V6ONLY=1) will not be detected — this is acceptable
* since the dashboard binds to 0.0.0.0 by default.
*/
export function isPortAvailable(port: number): Promise<boolean> {
return new Promise((resolve) => {
const server = createServer();
server.once("error", () => {
resolve(false);
});
server.once("listening", () => {
server.close(() => resolve(true));
});
server.listen(port, "127.0.0.1");
const s = new Socket();
s.setTimeout(300);
s.once("connect", () => { s.destroy(); resolve(false); }); // something listening → in use
s.once("error", () => { s.destroy(); resolve(true); }); // ECONNREFUSED → free
s.once("timeout", () => { s.destroy(); resolve(true); }); // no response → free
s.connect(port, "127.0.0.1");
});
}

76
packages/core/src/__tests__/lifecycle-manager.test.ts
View File

@ -84,7 +84,7 @@ beforeEach(() => {
getLaunchCommand: vi.fn(),
getEnvironment: vi.fn(),
detectActivity: vi.fn().mockReturnValue("active" as ActivityState),
getActivityState: vi.fn().mockResolvedValue("active" as ActivityState),
getActivityState: vi.fn().mockResolvedValue({ state: "active" as ActivityState }),
isProcessRunning: vi.fn().mockResolvedValue(true),
getSessionInfo: vi.fn().mockResolvedValue(null),
};
@ -227,7 +227,32 @@ describe("check (single session)", () => {
expect(lm.getStates().get("app-1")).toBe("killed");
});
it("detects killed state when agent process exits (idle terminal + dead process)", async () => {
it("detects killed state when getActivityState returns exited", async () => {
vi.mocked(mockAgent.getActivityState).mockResolvedValue({ state: "exited" });
const session = makeSession({ status: "working" });
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
writeMetadata(sessionsDir, "app-1", {
worktree: "/tmp",
branch: "main",
status: "working",
project: "my-app",
});
const lm = createLifecycleManager({
config,
registry: mockRegistry,
sessionManager: mockSessionManager,
});
await lm.check("app-1");
expect(lm.getStates().get("app-1")).toBe("killed");
});
it("detects killed via terminal fallback when getActivityState returns null", async () => {
vi.mocked(mockAgent.getActivityState).mockResolvedValue(null);
vi.mocked(mockAgent.detectActivity).mockReturnValue("idle");
vi.mocked(mockAgent.isProcessRunning).mockResolvedValue(false);
@ -252,34 +277,8 @@ describe("check (single session)", () => {
expect(lm.getStates().get("app-1")).toBe("killed");
});
it("detects killed state when agent process exits (active terminal + dead process)", async () => {
// Stub agents (codex, aider, opencode) return "active" for any non-empty
// terminal output, including the shell prompt after the agent exits.
vi.mocked(mockAgent.detectActivity).mockReturnValue("active");
vi.mocked(mockAgent.isProcessRunning).mockResolvedValue(false);
const session = makeSession({ status: "working" });
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
writeMetadata(sessionsDir, "app-1", {
worktree: "/tmp",
branch: "main",
status: "working",
project: "my-app",
});
const lm = createLifecycleManager({
config,
registry: mockRegistry,
sessionManager: mockSessionManager,
});
await lm.check("app-1");
expect(lm.getStates().get("app-1")).toBe("killed");
});
it("stays working when agent is idle but process is still running", async () => {
it("stays working when agent is idle but process is still running (fallback path)", async () => {
vi.mocked(mockAgent.getActivityState).mockResolvedValue(null);
vi.mocked(mockAgent.detectActivity).mockReturnValue("idle");
vi.mocked(mockAgent.isProcessRunning).mockResolvedValue(true);
@ -305,7 +304,7 @@ describe("check (single session)", () => {
});
it("detects needs_input from agent", async () => {
vi.mocked(mockAgent.detectActivity).mockReturnValue("waiting_input");
vi.mocked(mockAgent.getActivityState).mockResolvedValue({ state: "waiting_input" });
const session = makeSession({ status: "working" });
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
@ -328,10 +327,8 @@ describe("check (single session)", () => {
expect(lm.getStates().get("app-1")).toBe("needs_input");
});
it("preserves stuck state when detectActivity throws", async () => {
vi.mocked(mockAgent.detectActivity).mockImplementation(() => {
throw new Error("probe failed");
});
it("preserves stuck state when getActivityState throws", async () => {
vi.mocked(mockAgent.getActivityState).mockRejectedValue(new Error("probe failed"));
const session = makeSession({ status: "stuck" });
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
@ -355,10 +352,8 @@ describe("check (single session)", () => {
expect(lm.getStates().get("app-1")).toBe("stuck");
});
it("preserves needs_input state when detectActivity throws", async () => {
vi.mocked(mockAgent.detectActivity).mockImplementation(() => {
throw new Error("probe failed");
});
it("preserves needs_input state when getActivityState throws", async () => {
vi.mocked(mockAgent.getActivityState).mockRejectedValue(new Error("probe failed"));
const session = makeSession({ status: "needs_input" });
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
@ -382,7 +377,8 @@ describe("check (single session)", () => {
expect(lm.getStates().get("app-1")).toBe("needs_input");
});
it("preserves stuck state when getOutput throws", async () => {
it("preserves stuck state when getActivityState returns null and getOutput throws", async () => {
vi.mocked(mockAgent.getActivityState).mockResolvedValue(null);
vi.mocked(mockRuntime.getOutput).mockRejectedValue(new Error("tmux error"));
const session = makeSession({ status: "stuck" });

40
packages/core/src/lifecycle-manager.ts
View File

@ -196,27 +196,31 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
}
}
// 2. Check agent activity via terminal output + process liveness
// 2. Check agent activity — prefer JSONL-based detection (runtime-agnostic)
if (agent && session.runtimeHandle) {
try {
const runtime = registry.get<Runtime>(
"runtime",
project.runtime ?? config.defaults.runtime,
);
const terminalOutput = runtime ? await runtime.getOutput(session.runtimeHandle, 10) : "";
// Only trust detectActivity when we actually have terminal output;
// empty output means the runtime probe failed, not that the agent exited.
if (terminalOutput) {
const activity = agent.detectActivity(terminalOutput);
if (activity === "waiting_input") return "needs_input";
// Try JSONL-based activity detection first (reads agent's session files directly)
const activityState = await agent.getActivityState(session, config.readyThresholdMs);
if (activityState) {
if (activityState.state === "waiting_input") return "needs_input";
if (activityState.state === "exited") return "killed";
// active/ready/idle/blocked — proceed to PR checks below
} else {
// getActivityState returned null — fall back to terminal output parsing
const runtime = registry.get<Runtime>(
"runtime",
project.runtime ?? config.defaults.runtime,
);
const terminalOutput = runtime
? await runtime.getOutput(session.runtimeHandle, 10)
: "";
if (terminalOutput) {
const activity = agent.detectActivity(terminalOutput);
if (activity === "waiting_input") return "needs_input";
// Check whether the agent process is still alive. Some agents
// (codex, aider, opencode) return "active" for any non-empty
// terminal output, including the shell prompt visible after exit.
// Checking isProcessRunning for both "idle" and "active" ensures
// exit detection works regardless of the agent's classifier.
const processAlive = await agent.isProcessRunning(session.runtimeHandle);
if (!processAlive) return "killed";
const processAlive = await agent.isProcessRunning(session.runtimeHandle);
if (!processAlive) return "killed";
}
}
} catch {
// On probe failure, preserve current stuck/needs_input state rather

3
packages/plugins/agent-claude-code/src/index.test.ts
View File

@ -30,7 +30,7 @@ vi.mock("node:os", () => ({
homedir: mockHomedir,
}));
import { create, manifest, default as defaultExport } from "./index.js";
import { create, manifest, default as defaultExport, resetPsCache } from "./index.js";
// ---------------------------------------------------------------------------
// Test helpers
@ -108,6 +108,7 @@ function mockJsonlFiles(
// ---------------------------------------------------------------------------
beforeEach(() => {
vi.clearAllMocks();
resetPsCache();
mockHomedir.mockReturnValue("/mock/home");
});

60
packages/plugins/agent-claude-code/src/index.ts
View File

@ -386,6 +386,55 @@ function extractCost(lines: JsonlLine[]): CostEstimate | undefined {
// Process Detection
// =============================================================================
/**
* TTL cache for `ps -eo pid,tty,args` output. Without this, listing N sessions
* would spawn N concurrent `ps` processes, each taking 30+ seconds on machines
* with many processes. The cache ensures `ps` is called at most once per TTL
* window regardless of how many sessions are being enriched.
*/
let psCache: { output: string; timestamp: number; promise?: Promise<string> } | null = null;
const PS_CACHE_TTL_MS = 5_000;
/** Reset the ps cache. Exported for testing only. */
export function resetPsCache(): void {
psCache = null;
}
async function getCachedProcessList(): Promise<string> {
const now = Date.now();
if (psCache && now - psCache.timestamp < PS_CACHE_TTL_MS) {
// Cache hit — return resolved output or wait for in-flight request
if (psCache.promise) return psCache.promise;
return psCache.output;
}
// Cache miss or expired — start a single `ps` call and share the promise.
// Guard both callbacks so they only update psCache if it still belongs to
// this request — a newer request may have replaced it while we were waiting.
const promise = execFileAsync("ps", ["-eo", "pid,tty,args"], {
timeout: 5_000,
}).then(({ stdout }) => {
if (psCache?.promise === promise) {
psCache = { output: stdout, timestamp: Date.now() };
}
return stdout;
});
// Store the in-flight promise so concurrent callers share it
psCache = { output: "", timestamp: now, promise };
try {
return await promise;
} catch {
// On failure, clear cache so the next caller retries — but only if
// psCache still points to this request (avoid clobbering a newer entry)
if (psCache?.promise === promise) {
psCache = null;
}
return "";
}
}
/**
* Check if a process named "claude" is running in the given runtime handle's context.
* Uses ps to find processes by TTY (for tmux) or by PID.
@ -397,7 +446,7 @@ async function findClaudeProcess(handle: RuntimeHandle): Promise<number | null>
const { stdout: ttyOut } = await execFileAsync(
"tmux",
["list-panes", "-t", handle.id, "-F", "#{pane_tty}"],
{ timeout: 30_000 },
{ timeout: 5_000 },
);
// Iterate all pane TTYs (multi-pane sessions) — succeed on any match
const ttys = ttyOut
@ -407,12 +456,9 @@ async function findClaudeProcess(handle: RuntimeHandle): Promise<number | null>
.filter(Boolean);
if (ttys.length === 0) return null;
// Use `args` instead of `comm` so we can match the CLI name even when
// the process runs via a wrapper (e.g. node, python). `comm` would
// report "node" instead of "claude" in those cases.
const { stdout: psOut } = await execFileAsync("ps", ["-eo", "pid,tty,args"], {
timeout: 30_000,
});
const psOut = await getCachedProcessList();
if (!psOut) return null;
const ttySet = new Set(ttys.map((t) => t.replace(/^\/dev\//, "")));
// Match "claude" as a word boundary — prevents false positives on
// names like "claude-code" or paths that merely contain the substring.

2
packages/web/src/app/page.tsx
View File

@ -101,6 +101,6 @@ export default async function Home() {
}
return (
<Dashboard sessions={sessions} stats={computeStats(sessions)} orchestratorId={orchestratorId} projectName={projectName} />
<Dashboard initialSessions={sessions} stats={computeStats(sessions)} orchestratorId={orchestratorId} projectName={projectName} />
);
}

6
packages/web/src/components/Dashboard.tsx
View File

@ -13,9 +13,10 @@ import { CI_STATUS } from "@composio/ao-core/types";
import { AttentionZone } from "./AttentionZone";
import { PRTableRow } from "./PRStatus";
import { DynamicFavicon } from "./DynamicFavicon";
import { useSessionEvents } from "@/hooks/useSessionEvents";
interface DashboardProps {
sessions: DashboardSession[];
initialSessions: DashboardSession[];
stats: DashboardStats;
orchestratorId?: string | null;
projectName?: string;
@ -23,7 +24,8 @@ interface DashboardProps {
const KANBAN_LEVELS = ["working", "pending", "review", "respond", "merge"] as const;
export function Dashboard({ sessions, stats, orchestratorId, projectName }: DashboardProps) {
export function Dashboard({ initialSessions, stats, orchestratorId, projectName }: DashboardProps) {
const sessions = useSessionEvents(initialSessions);
const [rateLimitDismissed, setRateLimitDismissed] = useState(false);
const grouped = useMemo(() => {
const zones: Record<AttentionLevel, DashboardSession[]> = {

157
packages/web/src/components/DirectTerminal.tsx
View File

@ -45,6 +45,9 @@ export function DirectTerminal({
const terminalInstance = useRef<TerminalType | null>(null);
const fitAddon = useRef<FitAddonType | null>(null);
const ws = useRef<WebSocket | null>(null);
const reconnectAttemptRef = useRef(0);
const reconnectTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
const permanentErrorRef = useRef(false);
const [fullscreen, setFullscreen] = useState(startFullscreen);
const [status, setStatus] = useState<"connecting" | "connected" | "error">("connecting");
const [error, setError] = useState<string | null>(null);
@ -65,12 +68,18 @@ export function DirectTerminal({
useEffect(() => {
if (!terminalRef.current) return;
// Prevent retry loop on persistent errors
if (error && status === "error") return;
// Reset reconnection state when sessionId changes
permanentErrorRef.current = false;
reconnectAttemptRef.current = 0;
// Dynamically import xterm.js to avoid SSR issues
let mounted = true;
let cleanup: (() => void) | null = null;
let inputDisposable: { dispose(): void } | null = null;
const PERMANENT_CLOSE_CODES = new Set([4001, 4004]); // auth failure, session not found
const MAX_RECONNECT_DELAY = 15_000;
Promise.all([
import("xterm").then((mod) => mod.Terminal),
@ -174,60 +183,12 @@ export function DirectTerminal({
// Fit terminal to container
fit.fit();
// Connect WebSocket
// WebSocket URL (stable across reconnects)
const protocol = window.location.protocol === "https:" ? "wss:" : "ws:";
const hostname = window.location.hostname;
const port = process.env.NEXT_PUBLIC_DIRECT_TERMINAL_PORT ?? "14801";
const wsUrl = `${protocol}//${hostname}:${port}/ws?session=${encodeURIComponent(sessionId)}`;
console.log("[DirectTerminal] Connecting to:", wsUrl);
const websocket = new WebSocket(wsUrl);
ws.current = websocket;
websocket.binaryType = "arraybuffer";
websocket.onopen = () => {
console.log("[DirectTerminal] WebSocket connected");
setStatus("connected");
setError(null);
// Send initial size
websocket.send(
JSON.stringify({
type: "resize",
cols: terminal.cols,
rows: terminal.rows,
}),
);
};
websocket.onmessage = (event) => {
const data =
typeof event.data === "string" ? event.data : new TextDecoder().decode(event.data);
terminal.write(data);
};
websocket.onerror = (event) => {
console.error("[DirectTerminal] WebSocket error:", event);
setStatus("error");
setError("WebSocket connection error");
};
websocket.onclose = (event) => {
console.log("[DirectTerminal] WebSocket closed:", event.code, event.reason);
if (status === "connected") {
setStatus("error");
setError("Connection closed");
}
};
// Terminal input → WebSocket
const disposable = terminal.onData((data) => {
if (websocket.readyState === WebSocket.OPEN) {
websocket.send(data);
}
});
// Intercept Cmd+C/V (Mac) and Ctrl+Shift+C/V (Linux/Win) for clipboard
terminal.attachCustomKeyEventHandler((e: KeyboardEvent) => {
if (e.type !== "keydown") return true;
@ -240,7 +201,7 @@ export function DirectTerminal({
}
if (e.code === "KeyV") {
navigator.clipboard?.readText().then((text) => {
if (text && websocket.readyState === WebSocket.OPEN) {
if (text && ws.current?.readyState === WebSocket.OPEN) {
terminal.paste(text);
}
}).catch(() => {});
@ -255,7 +216,7 @@ export function DirectTerminal({
}
if (e.code === "KeyV") {
navigator.clipboard?.readText().then((text) => {
if (text && websocket.readyState === WebSocket.OPEN) {
if (text && ws.current?.readyState === WebSocket.OPEN) {
terminal.paste(text);
}
}).catch(() => {});
@ -266,11 +227,12 @@ export function DirectTerminal({
return true;
});
// Handle window resize
// Handle window resize (works with whatever ws is current)
const handleResize = () => {
if (fit && websocket.readyState === WebSocket.OPEN) {
const currentWs = ws.current;
if (fit && currentWs?.readyState === WebSocket.OPEN) {
fit.fit();
websocket.send(
currentWs.send(
JSON.stringify({
type: "resize",
cols: terminal.cols,
@ -282,22 +244,101 @@ export function DirectTerminal({
window.addEventListener("resize", handleResize);
// Terminal input → current WebSocket
inputDisposable = terminal.onData((data) => {
if (ws.current?.readyState === WebSocket.OPEN) {
ws.current.send(data);
}
});
function connectWebSocket() {
if (!mounted) return;
console.log("[DirectTerminal] Connecting to:", wsUrl);
const websocket = new WebSocket(wsUrl);
ws.current = websocket;
websocket.binaryType = "arraybuffer";
websocket.onopen = () => {
console.log("[DirectTerminal] WebSocket connected");
reconnectAttemptRef.current = 0;
setStatus("connected");
setError(null);
// Send initial size
websocket.send(
JSON.stringify({
type: "resize",
cols: terminal.cols,
rows: terminal.rows,
}),
);
};
websocket.onmessage = (event) => {
const data =
typeof event.data === "string" ? event.data : new TextDecoder().decode(event.data);
terminal.write(data);
};
websocket.onerror = (event) => {
console.error("[DirectTerminal] WebSocket error:", event);
};
websocket.onclose = (event) => {
console.log("[DirectTerminal] WebSocket closed:", event.code, event.reason);
if (!mounted) return;
// Permanent errors — don't retry
if (PERMANENT_CLOSE_CODES.has(event.code)) {
permanentErrorRef.current = true;
setStatus("error");
setError(event.reason || `Connection refused (${event.code})`);
return;
}
// Transient failure — schedule reconnect with exponential backoff
const attempt = reconnectAttemptRef.current;
const delay = Math.min(1000 * Math.pow(2, attempt), MAX_RECONNECT_DELAY);
reconnectAttemptRef.current = attempt + 1;
console.log(`[DirectTerminal] Reconnecting in ${delay}ms (attempt ${attempt + 1})`);
setStatus("connecting");
setError(null);
reconnectTimerRef.current = setTimeout(connectWebSocket, delay);
};
}
connectWebSocket();
// Store cleanup function to be called from useEffect cleanup
cleanup = () => {
window.removeEventListener("resize", handleResize);
disposable.dispose();
websocket.close();
inputDisposable?.dispose();
inputDisposable = null;
if (reconnectTimerRef.current) {
clearTimeout(reconnectTimerRef.current);
reconnectTimerRef.current = null;
}
ws.current?.close();
terminal.dispose();
};
})
.catch((err) => {
console.error("[DirectTerminal] Failed to load xterm.js:", err);
permanentErrorRef.current = true;
setStatus("error");
setError("Failed to load terminal");
});
return () => {
mounted = false;
if (reconnectTimerRef.current) {
clearTimeout(reconnectTimerRef.current);
reconnectTimerRef.current = null;
}
cleanup?.();
};
}, [sessionId, variant]);

68
packages/web/src/hooks/useSessionEvents.ts Normal file
View File

@ -0,0 +1,68 @@
"use client";
import { useEffect, useReducer } from "react";
import type { DashboardSession, SSESnapshotEvent } from "@/lib/types";
type Action =
| { type: "reset"; sessions: DashboardSession[] }
| { type: "snapshot"; patches: SSESnapshotEvent["sessions"] };
function reducer(state: DashboardSession[], action: Action): DashboardSession[] {
switch (action.type) {
case "reset":
return action.sessions;
case "snapshot": {
const patchMap = new Map(action.patches.map((p) => [p.id, p]));
let changed = false;
const next = state.map((s) => {
const patch = patchMap.get(s.id);
if (!patch) return s;
if (
s.status === patch.status &&
s.activity === patch.activity &&
s.lastActivityAt === patch.lastActivityAt
) {
return s;
}
changed = true;
return { ...s, status: patch.status, activity: patch.activity, lastActivityAt: patch.lastActivityAt };
});
return changed ? next : state;
}
}
}
export function useSessionEvents(initialSessions: DashboardSession[]): DashboardSession[] {
const [sessions, dispatch] = useReducer(reducer, initialSessions);
// Reset state when server-rendered props change (e.g. full page refresh)
useEffect(() => {
dispatch({ type: "reset", sessions: initialSessions });
}, [initialSessions]);
useEffect(() => {
const es = new EventSource("/api/events");
es.onmessage = (event: MessageEvent) => {
try {
const data = JSON.parse(event.data as string) as { type: string };
if (data.type === "snapshot") {
const snapshot = data as SSESnapshotEvent;
dispatch({ type: "snapshot", patches: snapshot.sessions });
}
} catch {
// Ignore malformed messages
}
};
es.onerror = () => {
// EventSource auto-reconnects; nothing to do here
};
return () => {
es.close();
};
}, []);
return sessions;
}