diff --git a/src/cli/commands/web.ts b/src/cli/commands/web.ts new file mode 100644 index 0000000..7f0f6f4 --- /dev/null +++ b/src/cli/commands/web.ts @@ -0,0 +1,252 @@ +import http from "node:http"; +import { spawn } from "node:child_process"; +import { InvalidArgumentError, type Command } from "commander"; +import { IpcClient } from "../../daemon/ipc.js"; +import { CANVAS_PAGE } from "../../web/canvas-page.js"; + +export const DEFAULT_WEB_PORT = 3100; + +export function parseWebPort(raw: string | undefined): number { + if (raw === undefined) return DEFAULT_WEB_PORT; + if (!/^\d+$/.test(raw)) { + throw new InvalidArgumentError("--port must be a positive integer"); + } + const port = Number(raw); + if (!Number.isInteger(port) || port <= 0 || port > 65535) { + throw new InvalidArgumentError("--port must be a positive integer"); + } + return port; +} + +export type SessionSubpathAction = "stream" | "logs" | "get"; + +export interface SessionSubpath { + action: SessionSubpathAction; + id: string; +} + +/** + * Pure route split for /api/sessions/:id[/stream|/logs]. More specific routes + * must be tried before this one so "stream" never lands in :id handling. + */ +export function parseSessionSubpath(parts: string[]): SessionSubpath | null { + if (parts.length < 3 || parts[0] !== "api" || parts[1] !== "sessions" || !parts[2]) { + return null; + } + const id = decodeURIComponent(parts[2]); + if (parts.length === 3) return { action: "get", id }; + if (parts.length === 4 && parts[3] === "stream") return { action: "stream", id }; + if (parts.length === 4 && parts[3] === "logs") return { action: "logs", id }; + return null; +} + +export function sseData(data: unknown): string { + return `data: ${JSON.stringify(data)}\n\n`; +} + +export function sseNamed(name: string, data: unknown): string { + return `event: ${name}\n${sseData(data)}`; +} + +export function sseComment(text = "conectado"): string { + return `: ${text}\n\n`; +} + +export function openBrowser(url: string): boolean { + const opener = + process.platform === "darwin" + ? "open" + : process.platform === "win32" + ? "cmd" + : "xdg-open"; + const args = process.platform === "win32" ? ["/c", "start", "", url] : [url]; + try { + const child = spawn(opener, args, { detached: true, stdio: "ignore" }); + child.unref(); + return true; + } catch { + return false; + } +} + +export interface WebBridge { + request(method: string, params: unknown): Promise; + subscribe( + sessionId: string, + onEvent: (event: { type?: string } & Record) => void, + onDone?: () => void, + onError?: (error: Error) => void, + ): () => void; +} + +function sendJson(res: http.ServerResponse, code: number, value: unknown): void { + res.writeHead(code, { "content-type": "application/json; charset=utf-8" }); + res.end(JSON.stringify(value)); +} + +export function createWebHandler(bridge: WebBridge): http.RequestListener { + return (req, res) => { + void (async () => { + try { + // Loopback base: only the path and query of the incoming request are + // read, nothing is ever fetched. (A non-loopback dummy base trips + // the S5332 cleartext-protocol rule.) + const url = new URL(req.url || "/", "http://127.0.0.1"); + const parts = url.pathname.split("/").filter(Boolean); + + if (req.method === "GET" && url.pathname === "/") { + res.writeHead(200, { "content-type": "text/html; charset=utf-8" }); + res.end(CANVAS_PAGE); + return; + } + if (req.method === "GET" && url.pathname === "/api/health") { + sendJson(res, 200, { ok: true }); + return; + } + if (req.method === "GET" && url.pathname === "/api/status") { + sendJson(res, 200, await bridge.request("daemon.status", {})); + return; + } + if (req.method === "GET" && url.pathname === "/api/sessions") { + const all = url.searchParams.get("all") === "1"; + sendJson(res, 200, await bridge.request("session.list", { all })); + return; + } + if (req.method === "GET" && url.pathname === "/api/usage") { + const period = url.searchParams.get("period") || "today"; + sendJson(res, 200, await bridge.request("usage.query", { period })); + return; + } + + const sub = req.method === "GET" ? parseSessionSubpath(parts) : null; + if (sub?.action === "stream") { + // Pre-flight before the 200: otherwise the browser holds an + // EventSource open on pings with no visible error when the daemon + // is down or the session does not exist. + try { + await bridge.request("session.get", { id: sub.id }); + } catch { + sendJson(res, 404, { error: "session not found or daemon unavailable" }); + return; + } + res.writeHead(200, { + "content-type": "text/event-stream; charset=utf-8", + "cache-control": "no-cache", + connection: "keep-alive", + }); + res.write(sseComment()); + const keepalive = setInterval(() => { + try { + res.write(sseComment("ping")); + } catch { + // Client is gone; req close below owns the cleanup. + } + }, 25000); + const off = bridge.subscribe( + sub.id, + (event) => { + try { + res.write(sseData(event)); + } catch { + // Client is gone; req close below owns the cleanup. + } + }, + () => { + try { + res.write(sseNamed("done", {})); + } catch { + // Client is gone; req close below owns the cleanup. + } + }, + () => { + try { + res.write(sseNamed("error", { error: "daemon disconnected" })); + } catch { + // Client is gone; req close below owns the cleanup. + } + }, + ); + req.on("close", () => { + clearInterval(keepalive); + off(); + }); + return; + } + if (sub?.action === "logs") { + sendJson(res, 200, await bridge.request("session.logs", { id: sub.id })); + return; + } + if (sub?.action === "get") { + sendJson(res, 200, await bridge.request("session.get", { id: sub.id })); + return; + } + sendJson(res, 404, { error: "not found" }); + } catch (error) { + // The pre-flight and stream branches answer above, so reaching here + // means no bytes were written yet and a JSON error is still safe. + try { + sendJson(res, 500, { error: error instanceof Error ? error.message : String(error) }); + } catch { + try { + res.end(); + } catch { + // Last resort: the socket is already unusable. + } + } + } + })(); + }; +} + +export interface WebCommandOptions { + port?: string; + open?: boolean; +} + +export function registerWebCommand(program: Command): void { + program + .command("web") + .description("Serve the realtime runs canvas in the browser (localhost only, no auth)") + .option("--port ", "port to listen on (default: 3100)", String(DEFAULT_WEB_PORT)) + .option("--open", "open the canvas in the default browser") + .action(async (opts: WebCommandOptions) => { + let port: number; + try { + port = parseWebPort(opts.port); + } catch (error) { + console.error(error instanceof Error ? error.message : String(error)); + process.exit(1); + throw error; + } + + const client = new IpcClient(); + try { + await client.ensureDaemonStarted(); + } catch { + // The bridge answers with JSON errors when the daemon is down, so a + // cold start failing here must not take the page down with it. + } + + const server = http.createServer(createWebHandler(client)); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(port, "127.0.0.1", () => resolve()); + }).catch((error: unknown) => { + console.error(`Failed to listen on 127.0.0.1:${port}: ${error instanceof Error ? error.message : String(error)}`); + process.exit(1); + }); + + const url = `http://127.0.0.1:${port}/`; + console.log(`CodeDeck web on ${url}`); + if (opts.open && !openBrowser(url)) { + console.log(`Could not open a browser, visit ${url} manually.`); + } + + const shutdown = () => { + server.close(() => process.exit(0)); + setTimeout(() => process.exit(0), 1000).unref?.(); + }; + process.on("SIGINT", shutdown); + process.on("SIGTERM", shutdown); + }); +} diff --git a/src/cli/index.ts b/src/cli/index.ts index 7de7726..63bbd2d 100644 --- a/src/cli/index.ts +++ b/src/cli/index.ts @@ -19,6 +19,7 @@ import { registerModelsCommand } from "./commands/models.js"; import { registerOpenCommand } from "./commands/open.js"; import { registerSetupCommand } from "./commands/setup.js"; import { registerUsageCommand } from "./commands/usage.js"; +import { registerWebCommand } from "./commands/web.js"; import { getCliInvocation, getCliName } from "./cli-name.js"; function getVersion(): string { @@ -69,6 +70,7 @@ Recommended flow: $ ${cli} run "task" --bg --json # starts in background $ ${cli} wait # waits without ps/show loop $ ${cli} logs --follow # inspect progress + $ ${cli} web --open # realtime runs canvas in the browser Run '${cli} --help' for command-specific options. Docs: https://github.com/4ndreello/run-agent @@ -89,6 +91,7 @@ registerModelsCommand(program); registerOpenCommand(program); registerSetupCommand(program); registerUsageCommand(program); +registerWebCommand(program); // Make `codedeck help` behave like `codedeck --help` program.command("help", { hidden: true }).action(() => program.outputHelp()); diff --git a/src/core/session.ts b/src/core/session.ts index 575b968..cc809d9 100644 --- a/src/core/session.ts +++ b/src/core/session.ts @@ -92,6 +92,20 @@ export function isActiveStatus(status: SessionStatus): boolean { return status === "starting" || status === "working" || status === "needs_input" || status === "idle"; } +// Display-level liveness shared by `ps` and the daemon read boundary: an +// active session whose recorded process is gone is a corpse ("dead"), never +// "working". Pure (liveness injected) so it is unit-testable; the stored +// row is never mutated, only the served view. +export type LiveSessionStatus = SessionStatus | "dead"; +export function liveStatus( + status: SessionStatus, + pid: number | null | undefined, + alive: boolean, +): LiveSessionStatus { + if (isActiveStatus(status) && pid != null && !alive) return "dead"; + return status; +} + export function generateSessionId(): string { // 4-char hex like spec (a83f) but ensure uniqueness with 8 chars if needed // Use 8 hex chars, display first 4 but store full diff --git a/src/daemon/daemon.ts b/src/daemon/daemon.ts index 70e6470..e7a6b23 100644 --- a/src/daemon/daemon.ts +++ b/src/daemon/daemon.ts @@ -11,7 +11,7 @@ import { getPaths, ensureDirs } from "../config/paths.js"; import { createIpcServer } from "./ipc.js"; import type { IpcRequest, IpcResponse, UsageQueryParams } from "./protocol.js"; import { getRegistry } from "../drivers/registry.js"; -import { isTerminalStatus, normalizeAgentId, type AgentId, type Session, type SessionStatus } from "../core/session.js"; +import { isTerminalStatus, liveStatus, normalizeAgentId, type AgentId, type Session, type SessionStatus } from "../core/session.js"; import { parseSandbox, type AgentDriver, type CodexSandbox, type DriverSession } from "../core/driver.js"; import { generateSessionId, generateBranchName } from "../core/session.js"; import { getGitInfo, getBaseCommit } from "../git/repository.js"; @@ -44,6 +44,15 @@ function resolveRequestSandbox(value: unknown): CodexSandbox | undefined { } } +// Read-boundary liveness: serve a corpse as "dead" (same rule `ps` shows) +// instead of the stale stored "working". Returns the same object when no +// correction applies, so callers can skip copies; never mutates the store. +function withLiveStatus(s: Session): Session { + if (s.pid == null) return s; + const status = liveStatus(s.status, s.pid, processAlive(s.pid)); + return status === s.status ? s : { ...s, status: status as Session["status"] }; +} + class Daemon { private db: Database; private sessions: SessionStore; @@ -520,7 +529,7 @@ class Daemon { case "session.list": { const p = params as any; const all = p?.all; - const list = this.sessions.list(100, all); + const list = this.sessions.list(100, all).map(withLiveStatus); const hidden = all ? 0 : this.sessions.countHiddenByWindow(); // Enrich with last event? send({ result: { sessions: list, hidden } }); @@ -529,8 +538,9 @@ class Daemon { case "session.get": { const p = params as any; - const s = this.sessions.get(p.id); - if (!s) { send({ error: { code: "SESSION_NOT_FOUND", message: `Session ${p.id} not found` } }); return; } + const stored = this.sessions.get(p.id); + if (!stored) { send({ error: { code: "SESSION_NOT_FOUND", message: `Session ${p.id} not found` } }); return; } + const s = withLiveStatus(stored); const evCount = this.events.count(s.id); const recent = this.events.list(s.id, 10); send({ result: { session: s, events: recent, eventCount: evCount } }); diff --git a/src/web/canvas-page.ts b/src/web/canvas-page.ts new file mode 100644 index 0000000..24e63bd --- /dev/null +++ b/src/web/canvas-page.ts @@ -0,0 +1,920 @@ +/** Realtime runs canvas served by `codedeck web`. + * + * Single self-contained page (no external requests): the script polls + * `api/sessions` and opens one `EventSource` per active session. + * Promoted from spikes/web-top.mjs; keep the spike behavior in sync + * only by re-running the Playwright check, never by importing it. + */ +export const CANVAS_PAGE: string = ` + + + + +CodeDeck canvas realtime + + + + + +
+

Canvas das runs

conectando...

+
+ + + +
+
TrabalhandoEsperando vocêEm pausaConcluída
+

AGORA MESMO

    +
    arraste para organizar · scroll dá zoom · clique num agente para o detalhe
    + + + +`; diff --git a/tests/daemon-live-status.test.ts b/tests/daemon-live-status.test.ts new file mode 100644 index 0000000..16fd344 --- /dev/null +++ b/tests/daemon-live-status.test.ts @@ -0,0 +1,58 @@ +import { describe, expect, it, beforeEach, afterEach } from "vitest"; +import { Daemon } from "../src/daemon/daemon.js"; +import { makeTempDir, removeTempDir, seam, seed, fakeSocket } from "./helpers/daemon-seam.js"; + +let dir: string; + +beforeEach(() => { + dir = makeTempDir("live-status-"); + process.env.RUN_AGENT_DIR = dir; +}); + +afterEach(() => { + delete process.env.RUN_AGENT_DIR; + removeTempDir(dir); +}); + +async function listStatuses(daemon: Daemon): Promise { + const { writes, socket } = fakeSocket(); + await seam(daemon).handleRequest({ id: "r1", method: "session.list", params: { all: true } }, socket); + const body = JSON.parse(writes[0]) as { result?: { sessions?: { id: string; status: string }[] } }; + return (body.result?.sessions ?? []).map((s) => `${s.id}:${s.status}`); +} + +describe("session.list liveness guard", () => { + it("reports dead for a working session whose process is gone", async () => { + const daemon = new Daemon(); + seed(daemon, "s-corpse", "working", { pid: 999999999 }); + expect(await listStatuses(daemon)).toContain("s-corpse:dead"); + }); + + it("keeps working for a working session whose process is alive", async () => { + const daemon = new Daemon(); + seed(daemon, "s-live", "working", { pid: process.pid }); + expect(await listStatuses(daemon)).toContain("s-live:working"); + }); + + it("keeps a terminal status even when its pid is gone", async () => { + const daemon = new Daemon(); + seed(daemon, "s-done", "completed", { pid: 999999999 }); + expect(await listStatuses(daemon)).toContain("s-done:completed"); + }); + + it("does not mutate the stored row", async () => { + const daemon = new Daemon(); + seed(daemon, "s-corpse", "working", { pid: 999999999 }); + await listStatuses(daemon); + expect(seam(daemon).sessions.get("s-corpse")?.status).toBe("working"); + }); + + it("session.get reports dead for the same corpse", async () => { + const daemon = new Daemon(); + seed(daemon, "s-corpse", "working", { pid: 999999999 }); + const { writes, socket } = fakeSocket(); + await seam(daemon).handleRequest({ id: "r2", method: "session.get", params: { id: "s-corpse" } }, socket); + const body = JSON.parse(writes[0]) as { result?: { session?: { status: string } } }; + expect(body.result?.session?.status).toBe("dead"); + }); +}); diff --git a/tests/web-command.test.ts b/tests/web-command.test.ts new file mode 100644 index 0000000..7a9b01d --- /dev/null +++ b/tests/web-command.test.ts @@ -0,0 +1,225 @@ +import http from "node:http"; +import { Command } from "commander"; +import { describe, expect, it, afterEach, vi } from "vitest"; +import { CANVAS_PAGE } from "../src/web/canvas-page.js"; +import { + createWebHandler, + parseSessionSubpath, + parseWebPort, + registerWebCommand, + sseComment, + sseData, + sseNamed, + type WebBridge, +} from "../src/cli/commands/web.js"; + +function bridge(overrides: Partial = {}): WebBridge & { calls: string[] } { + const calls: string[] = []; + const base: WebBridge & { calls: string[] } = { + calls, + request: async (method: string) => { + calls.push(method); + if (method === "session.list") return { sessions: [] }; + if (method === "session.get") return { session: { id: "a1" } }; + if (method === "session.logs") return { events: [] }; + if (method === "daemon.status") return { ok: true }; + if (method === "usage.query") return { totals: {} }; + throw new Error(`unexpected method ${method}`); + }, + subscribe: (_id, _onEvent, _onDone) => () => {}, + ...overrides, + }; + return base; +} + +let servers: http.Server[] = []; +afterEach(async () => { + await Promise.all( + servers.map((s) => new Promise((resolve) => s.close(() => resolve()))), + ); + servers = []; +}); + +async function listen(handler: http.RequestListener): Promise { + const server = http.createServer(handler); + servers.push(server); + await new Promise((resolve) => server.listen(0, "127.0.0.1", () => resolve())); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("no address"); + return `http://127.0.0.1:${address.port}`; +} + +describe("parseWebPort", () => { + it("defaults to 3100", () => { + expect(parseWebPort(undefined)).toBe(3100); + }); + + it("accepts a valid port", () => { + expect(parseWebPort("8080")).toBe(8080); + }); + + it("rejects non-numeric, zero, and out-of-range ports", () => { + for (const raw of ["abc", "0", "-1", "65536", "3.5", ""]) { + expect(() => parseWebPort(raw)).toThrow("--port must be a positive integer"); + } + }); +}); + +describe("parseSessionSubpath", () => { + it("matches get, stream, and logs", () => { + expect(parseSessionSubpath(["api", "sessions", "a1"])).toEqual({ action: "get", id: "a1" }); + expect(parseSessionSubpath(["api", "sessions", "a1", "stream"])).toEqual({ + action: "stream", + id: "a1", + }); + expect(parseSessionSubpath(["api", "sessions", "a1", "logs"])).toEqual({ + action: "logs", + id: "a1", + }); + }); + + it("decodes the session id", () => { + expect(parseSessionSubpath(["api", "sessions", "a%201", "logs"])).toEqual({ + action: "logs", + id: "a 1", + }); + }); + + it("rejects anything else", () => { + expect(parseSessionSubpath(["api", "sessions"])).toBeNull(); + expect(parseSessionSubpath(["api", "sessions", "", "stream"])).toBeNull(); + expect(parseSessionSubpath(["api", "sessions", "a1", "send"])).toBeNull(); + expect(parseSessionSubpath(["api", "sessions", "a1", "stream", "x"])).toBeNull(); + expect(parseSessionSubpath(["api", "other", "a1"])).toBeNull(); + }); +}); + +describe("sse framing", () => { + it("frames data, named events, and comments", () => { + expect(sseData({ a: 1 })).toBe('data: {"a":1}\n\n'); + expect(sseNamed("done", {})).toBe("event: done\ndata: {}\n\n"); + expect(sseComment()).toBe(": conectado\n\n"); + }); +}); + +describe("web handler", () => { + it("serves the canvas page at /", async () => { + const base = await listen(createWebHandler(bridge())); + const res = await fetch(`${base}/`); + expect(res.status).toBe(200); + expect(res.headers.get("content-type")).toContain("text/html"); + const html = await res.text(); + expect(html).toContain("Canvas das runs"); + }); + + it("answers health and unknown routes", async () => { + const base = await listen(createWebHandler(bridge())); + const health = await fetch(`${base}/api/health`); + expect(health.status).toBe(200); + expect(await health.json()).toEqual({ ok: true }); + + const missing = await fetch(`${base}/nope`); + expect(missing.status).toBe(404); + }); + + it("proxies session list and get", async () => { + const base = await listen(createWebHandler(bridge())); + const list = await fetch(`${base}/api/sessions`); + expect(list.status).toBe(200); + expect(await list.json()).toEqual({ sessions: [] }); + + const get = await fetch(`${base}/api/sessions/a1`); + expect(get.status).toBe(200); + }); + + it("returns 404 on the stream pre-flight when the session is unknown", async () => { + const b = bridge(); + const inner = b.request; + b.request = async (method: string, params: unknown) => { + b.calls.push(method); + if (method === "session.get") throw new Error("gone"); + return inner(method, params); + }; + const base = await listen(createWebHandler(b)); + const res = await fetch(`${base}/api/sessions/zz/stream`); + expect(res.status).toBe(404); + expect(b.calls).toContain("session.get"); + expect(b.calls).not.toContain("session.subscribe"); + }); + + it("streams one event then done, and unsubscribes on disconnect", async () => { + let offCalled = false; + const b = bridge({ + subscribe: (id, onEvent, onDone) => { + expect(id).toBe("a1"); + const timer = setTimeout(() => { + onEvent({ type: "text.delta", delta: "hi" }); + onDone?.(); + }, 10); + return () => { + offCalled = true; + clearTimeout(timer); + }; + }, + }); + const base = await listen(createWebHandler(b)); + const res = await fetch(`${base}/api/sessions/a1/stream`); + expect(res.status).toBe(200); + expect(res.headers.get("content-type")).toContain("text/event-stream"); + + const reader = res.body!.getReader(); + const decoder = new TextDecoder(); + let text = ""; + for (let i = 0; i < 4; i++) { + const { done, value } = await reader.read(); + if (done) break; + text += decoder.decode(value, { stream: true }); + if (text.includes("event: done")) break; + } + await reader.cancel(); + expect(text).toContain(": conectado"); + expect(text).toContain('"text.delta"'); + expect(text).toContain("event: done"); + // Give the server close handler a tick to run the unsubscribe. + await vi.waitFor(() => expect(offCalled).toBe(true)); + }); +}); + +describe("canvas page", () => { + it("wires the realtime endpoints", () => { + expect(CANVAS_PAGE).toContain("api/sessions"); + expect(CANVAS_PAGE).toContain("new EventSource"); + expect(CANVAS_PAGE).toContain("text.delta"); + }); + + it("has the canvas controls", () => { + expect(CANVAS_PAGE).toContain("btnMotion"); + expect(CANVAS_PAGE).toContain("btnReset"); + expect(CANVAS_PAGE).toContain("btnHide"); + expect(CANVAS_PAGE).toContain("fitCamera"); + expect(CANVAS_PAGE).toContain("currentTarget"); + expect(CANVAS_PAGE).toContain("drawMarkers"); + expect(CANVAS_PAGE).toContain("syncUrl"); + expect(CANVAS_PAGE).toContain("selectSession"); + expect(CANVAS_PAGE).toContain("inputSnippet"); + expect(CANVAS_PAGE).toContain('id="detail"'); + }); + + it("makes no external requests", () => { + const externals = CANVAS_PAGE.match(/https?:\/\/[^"'\s>]+/g) ?? []; + expect(externals).toEqual([]); + }); + + it("keeps the template literal intact (no backticks or interpolation)", () => { + expect(CANVAS_PAGE).not.toContain("`"); + expect(CANVAS_PAGE).not.toContain("${"); + }); +}); + +describe("registerWebCommand", () => { + it("registers the web command", () => { + const program = new Command(); + registerWebCommand(program); + expect(program.commands.map((c) => c.name())).toContain("web"); + }); +});