feat: service proxy, health monitor, hover card, and homepage section

Built-in service proxy with branch-based URLs, service health
monitoring, workspace hover card with service status, and
"Forget about ports" homepage section.
This commit is contained in:
Mohamed Boudra
2026-04-01 10:31:15 +07:00
parent 806079c7ad
commit 5b3ec572cb
39 changed files with 3776 additions and 498 deletions
@@ -251,6 +251,7 @@ describe("DaemonClient", () => {
index: 1,
command: "npm install",
cwd: "/tmp/project/.paseo/worktrees/feature-a",
log: "phase-one\n",
status: "running",
exitCode: null,
},
@@ -277,6 +278,7 @@ describe("DaemonClient", () => {
index: 1,
command: "npm install",
cwd: "/tmp/project/.paseo/worktrees/feature-a",
log: "phase-one\n",
status: "running",
exitCode: null,
},
@@ -36,6 +36,7 @@ import type {
ProjectIconResponse,
OpenProjectResponseMessage,
ArchiveWorkspaceResponseMessage,
WorkspaceSetupStatusResponseMessage,
ListCommandsResponse,
ListProviderModelsResponseMessage,
ListAvailableProvidersResponse,
@@ -462,6 +463,7 @@ export type InspectScheduleOptions = {
};
type OpenProjectPayload = OpenProjectResponseMessage["payload"];
type ArchiveWorkspacePayload = ArchiveWorkspaceResponseMessage["payload"];
type WorkspaceSetupStatusPayload = WorkspaceSetupStatusResponseMessage["payload"];
export type FetchAgentResult = {
agent: AgentSnapshotPayload;
@@ -1318,6 +1320,21 @@ export class DaemonClient {
});
}
async fetchWorkspaceSetupStatus(
workspaceId: string,
requestId?: string,
): Promise<WorkspaceSetupStatusPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "workspace_setup_status_request",
workspaceId,
},
responseType: "workspace_setup_status_response",
timeout: 10000,
});
}
async fetchAgent(agentId: string, requestId?: string): Promise<FetchAgentResult | null> {
const resolvedRequestId = this.createRequestId(requestId);
const message = SessionInboundMessageSchema.parse({
@@ -178,6 +178,7 @@ export type ToolCallDetail =
index: number;
command: string;
cwd: string;
log: string;
status: "running" | "completed" | "failed";
exitCode: number | null;
durationMs?: number;
+17 -1
View File
@@ -118,6 +118,8 @@ import {
createServiceProxyMiddleware,
createServiceProxyUpgradeHandler,
} from "./service-proxy.js";
import { ServiceHealthMonitor } from "./service-health-monitor.js";
import { createServiceStatusEmitter } from "./service-status-projection.js";
import {
createVoiceMcpSocketBridgeManager,
type VoiceMcpSocketBridgeManager,
@@ -225,6 +227,18 @@ export async function createPaseoDaemon(
let boundListenTarget: ListenTarget | null = null;
const serviceRouteStore = new ServiceRouteStore();
let wsServer: VoiceAssistantWebSocketServer | null = null;
const serviceHealthMonitor = new ServiceHealthMonitor({
routeStore: serviceRouteStore,
onChange: createServiceStatusEmitter({
sessions: () =>
wsServer?.listActiveSessions().map((session) => ({
emit: (message) => session.emitServerMessage(message),
})) ?? [],
routeStore: serviceRouteStore,
daemonPort: () => (boundListenTarget?.type === "tcp" ? boundListenTarget.port : null),
}),
});
// Host allowlist / DNS rebinding protection (vite-like semantics).
// For non-TCP (unix sockets), skip host validation.
@@ -450,7 +464,6 @@ export async function createPaseoDaemon(
"Voice mode configured for agent-scoped resume flow (no dedicated voice assistant provider)",
);
logger.info({ elapsed: elapsed() }, "Preparing voice and MCP runtime");
let wsServer: VoiceAssistantWebSocketServer | null = null;
let voiceMcpBridgeManager: VoiceMcpSocketBridgeManager | null = null;
// Create in-memory transport for Session's Agent MCP client (voice assistant tools)
@@ -676,6 +689,7 @@ export async function createPaseoDaemon(
checkoutDiffManager,
serviceRouteStore,
() => (boundListenTarget?.type === "tcp" ? boundListenTarget.port : null),
(hostname) => serviceHealthMonitor.getStatusForHostname(hostname),
);
logger.info({ elapsed: elapsed() }, "Bootstrap complete, ready to start listening");
@@ -766,9 +780,11 @@ export async function createPaseoDaemon(
// Start speech service after listening so synchronous Sherpa native
// model loading doesn't block the server from accepting connections.
speechService.start();
serviceHealthMonitor.start();
};
const stop = async () => {
serviceHealthMonitor.stop();
await closeAllAgents(logger, agentManager);
await agentManager.flush().catch(() => undefined);
detachAgentStoragePersistence();
@@ -0,0 +1,455 @@
import net from "node:net";
import { scheduler } from "node:timers/promises";
import { afterEach, describe, expect, it, vi } from "vitest";
import { findFreePort, ServiceRouteStore } from "./service-proxy.js";
import {
ServiceHealthMonitor,
type ServiceStatusEntry,
} from "./service-health-monitor.js";
type TcpServerHandle = {
port: number;
server: net.Server;
};
async function startTcpServer(): Promise<TcpServerHandle> {
const server = net.createServer((socket) => {
socket.end();
});
await new Promise<void>((resolve, reject) => {
server.once("error", reject);
server.listen(0, "127.0.0.1", () => {
server.off("error", reject);
resolve();
});
});
const address = server.address();
if (!address || typeof address === "string") {
throw new Error("Failed to resolve TCP server address");
}
return { port: address.port, server };
}
async function closeServer(server: net.Server): Promise<void> {
if (!server.listening) {
return;
}
await new Promise<void>((resolve, reject) => {
server.close((error) => {
if (error) {
reject(error);
return;
}
resolve();
});
});
}
async function advancePoll(ms: number): Promise<void> {
await vi.advanceTimersByTimeAsync(ms);
for (let i = 0; i < 5; i += 1) {
await scheduler.yield();
}
}
describe("ServiceHealthMonitor", () => {
const servers = new Set<net.Server>();
afterEach(async () => {
vi.useRealTimers();
for (const server of servers) {
await closeServer(server);
}
servers.clear();
});
it("marks a healthy port as running after successful TCP connect", async () => {
vi.useFakeTimers();
const healthy = await startTcpServer();
servers.add(healthy.server);
const routeStore = new ServiceRouteStore();
routeStore.registerRoute({
hostname: "api.localhost",
port: healthy.port,
workspaceId: "workspace-a",
serviceName: "api",
});
const onChange = vi.fn<(workspaceId: string, services: ServiceStatusEntry[]) => void>();
const monitor = new ServiceHealthMonitor({
routeStore,
onChange,
pollIntervalMs: 1_000,
probeTimeoutMs: 100,
graceMs: 0,
});
monitor.start();
await advancePoll(1_000);
monitor.stop();
expect(onChange).toHaveBeenCalledTimes(1);
expect(onChange).toHaveBeenCalledWith("workspace-a", [
{
serviceName: "api",
hostname: "api.localhost",
port: healthy.port,
status: "running",
},
]);
});
it("marks an unreachable port as stopped after consecutive failures", async () => {
vi.useFakeTimers();
const deadPort = await findFreePort();
const routeStore = new ServiceRouteStore();
routeStore.registerRoute({
hostname: "api.localhost",
port: deadPort,
workspaceId: "workspace-a",
serviceName: "api",
});
const onChange = vi.fn<(workspaceId: string, services: ServiceStatusEntry[]) => void>();
const monitor = new ServiceHealthMonitor({
routeStore,
onChange,
pollIntervalMs: 1_000,
probeTimeoutMs: 100,
graceMs: 0,
failuresBeforeStopped: 2,
});
monitor.start();
await advancePoll(1_000);
expect(onChange).not.toHaveBeenCalled();
await advancePoll(1_000);
monitor.stop();
expect(onChange).toHaveBeenCalledTimes(1);
expect(onChange).toHaveBeenCalledWith("workspace-a", [
{
serviceName: "api",
hostname: "api.localhost",
port: deadPort,
status: "stopped",
},
]);
});
it("does not emit when status has not changed", async () => {
vi.useFakeTimers();
const healthy = await startTcpServer();
servers.add(healthy.server);
const routeStore = new ServiceRouteStore();
routeStore.registerRoute({
hostname: "api.localhost",
port: healthy.port,
workspaceId: "workspace-a",
serviceName: "api",
});
const onChange = vi.fn<(workspaceId: string, services: ServiceStatusEntry[]) => void>();
const monitor = new ServiceHealthMonitor({
routeStore,
onChange,
pollIntervalMs: 1_000,
probeTimeoutMs: 100,
graceMs: 0,
});
monitor.start();
await advancePoll(3_000);
monitor.stop();
expect(onChange).toHaveBeenCalledTimes(1);
});
it("respects startup grace period — does not probe newly registered routes for 5 seconds", async () => {
vi.useFakeTimers();
const healthy = await startTcpServer();
servers.add(healthy.server);
const routeStore = new ServiceRouteStore();
routeStore.registerRoute({
hostname: "api.localhost",
port: healthy.port,
workspaceId: "workspace-a",
serviceName: "api",
});
const onChange = vi.fn<(workspaceId: string, services: ServiceStatusEntry[]) => void>();
const monitor = new ServiceHealthMonitor({
routeStore,
onChange,
pollIntervalMs: 1_000,
probeTimeoutMs: 100,
graceMs: 5_000,
});
monitor.start();
await advancePoll(4_000);
expect(onChange).not.toHaveBeenCalled();
await advancePoll(1_000);
monitor.stop();
expect(onChange).toHaveBeenCalledTimes(1);
expect(onChange).toHaveBeenCalledWith("workspace-a", [
{
serviceName: "api",
hostname: "api.localhost",
port: healthy.port,
status: "running",
},
]);
});
it("requires 2 consecutive failures before marking stopped (debounce)", async () => {
vi.useFakeTimers();
const healthy = await startTcpServer();
servers.add(healthy.server);
const routeStore = new ServiceRouteStore();
routeStore.registerRoute({
hostname: "api.localhost",
port: healthy.port,
workspaceId: "workspace-a",
serviceName: "api",
});
const onChange = vi.fn<(workspaceId: string, services: ServiceStatusEntry[]) => void>();
const monitor = new ServiceHealthMonitor({
routeStore,
onChange,
pollIntervalMs: 1_000,
probeTimeoutMs: 100,
graceMs: 0,
failuresBeforeStopped: 2,
});
monitor.start();
await advancePoll(1_000);
expect(onChange).toHaveBeenCalledTimes(1);
await closeServer(healthy.server);
servers.delete(healthy.server);
await advancePoll(1_000);
expect(onChange).toHaveBeenCalledTimes(1);
await advancePoll(1_000);
monitor.stop();
expect(onChange).toHaveBeenCalledTimes(2);
expect(onChange).toHaveBeenLastCalledWith("workspace-a", [
{
serviceName: "api",
hostname: "api.localhost",
port: healthy.port,
status: "stopped",
},
]);
});
it("stops probing routes that are removed from the store", async () => {
vi.useFakeTimers();
const healthy = await startTcpServer();
servers.add(healthy.server);
const routeStore = new ServiceRouteStore();
routeStore.registerRoute({
hostname: "api.localhost",
port: healthy.port,
workspaceId: "workspace-a",
serviceName: "api",
});
const onChange = vi.fn<(workspaceId: string, services: ServiceStatusEntry[]) => void>();
const monitor = new ServiceHealthMonitor({
routeStore,
onChange,
pollIntervalMs: 1_000,
probeTimeoutMs: 100,
graceMs: 0,
failuresBeforeStopped: 2,
});
monitor.start();
await advancePoll(1_000);
expect(onChange).toHaveBeenCalledTimes(1);
routeStore.removeRoute("api.localhost");
await closeServer(healthy.server);
servers.delete(healthy.server);
await advancePoll(3_000);
monitor.stop();
expect(onChange).toHaveBeenCalledTimes(1);
});
it("calls onChange with workspaceId and full service list when status transitions", async () => {
vi.useFakeTimers();
const api = await startTcpServer();
const web = await startTcpServer();
servers.add(api.server);
servers.add(web.server);
const routeStore = new ServiceRouteStore();
routeStore.registerRoute({
hostname: "api.localhost",
port: api.port,
workspaceId: "workspace-a",
serviceName: "api",
});
routeStore.registerRoute({
hostname: "web.localhost",
port: web.port,
workspaceId: "workspace-a",
serviceName: "web",
});
const onChange = vi.fn<(workspaceId: string, services: ServiceStatusEntry[]) => void>();
const monitor = new ServiceHealthMonitor({
routeStore,
onChange,
pollIntervalMs: 1_000,
probeTimeoutMs: 100,
graceMs: 0,
});
monitor.start();
await advancePoll(1_000);
monitor.stop();
expect(onChange).toHaveBeenCalledTimes(1);
expect(onChange).toHaveBeenCalledWith("workspace-a", [
{
serviceName: "api",
hostname: "api.localhost",
port: api.port,
status: "running",
},
{
serviceName: "web",
hostname: "web.localhost",
port: web.port,
status: "running",
},
]);
});
it("getStatusForHostname returns current status after probe", async () => {
vi.useFakeTimers();
const healthy = await startTcpServer();
servers.add(healthy.server);
const routeStore = new ServiceRouteStore();
routeStore.registerRoute({
hostname: "api.localhost",
port: healthy.port,
workspaceId: "workspace-a",
serviceName: "api",
});
const onChange = vi.fn<(workspaceId: string, services: ServiceStatusEntry[]) => void>();
const monitor = new ServiceHealthMonitor({
routeStore,
onChange,
pollIntervalMs: 1_000,
probeTimeoutMs: 100,
graceMs: 0,
});
expect(monitor.getStatusForHostname("api.localhost")).toBeNull();
monitor.start();
await advancePoll(1_000);
monitor.stop();
expect(monitor.getStatusForHostname("api.localhost")).toBe("running");
expect(monitor.getStatusForHostname("unknown.localhost")).toBeNull();
});
it("coalesces multiple service changes in same workspace into one onChange call per poll cycle", async () => {
vi.useFakeTimers();
const api = await startTcpServer();
const web = await startTcpServer();
servers.add(api.server);
servers.add(web.server);
const routeStore = new ServiceRouteStore();
routeStore.registerRoute({
hostname: "api.localhost",
port: api.port,
workspaceId: "workspace-a",
serviceName: "api",
});
routeStore.registerRoute({
hostname: "web.localhost",
port: web.port,
workspaceId: "workspace-a",
serviceName: "web",
});
const onChange = vi.fn<(workspaceId: string, services: ServiceStatusEntry[]) => void>();
const monitor = new ServiceHealthMonitor({
routeStore,
onChange,
pollIntervalMs: 1_000,
probeTimeoutMs: 100,
graceMs: 0,
failuresBeforeStopped: 2,
});
monitor.start();
await advancePoll(1_000);
expect(onChange).toHaveBeenCalledTimes(1);
onChange.mockClear();
await closeServer(api.server);
await closeServer(web.server);
servers.delete(api.server);
servers.delete(web.server);
await advancePoll(1_000);
expect(onChange).not.toHaveBeenCalled();
await advancePoll(1_000);
monitor.stop();
expect(onChange).toHaveBeenCalledTimes(1);
expect(onChange).toHaveBeenCalledWith("workspace-a", [
{
serviceName: "api",
hostname: "api.localhost",
port: api.port,
status: "stopped",
},
{
serviceName: "web",
hostname: "web.localhost",
port: web.port,
status: "stopped",
},
]);
});
});
@@ -0,0 +1,203 @@
import net from "node:net";
import type { ServiceRouteEntry, ServiceRouteStore } from "./service-proxy.js";
export interface ServiceStatusEntry {
serviceName: string;
hostname: string;
port: number;
status: "running" | "stopped";
}
type RouteHealthState = {
status: ServiceStatusEntry["status"] | null;
consecutiveFailures: number;
registeredAt: number;
};
export class ServiceHealthMonitor {
private readonly routeStore: ServiceRouteStore;
private readonly onChange: (
workspaceId: string,
services: ServiceStatusEntry[],
) => void;
private readonly pollIntervalMs: number;
private readonly probeTimeoutMs: number;
private readonly graceMs: number;
private readonly failuresBeforeStopped: number;
private readonly routeStates = new Map<string, RouteHealthState>();
private readonly lastEmittedSnapshots = new Map<string, string>();
private intervalHandle: NodeJS.Timeout | null = null;
private pollInFlight = false;
constructor({
routeStore,
onChange,
pollIntervalMs = 3_000,
probeTimeoutMs = 500,
graceMs = 5_000,
failuresBeforeStopped = 2,
}: {
routeStore: ServiceRouteStore;
onChange: (workspaceId: string, services: ServiceStatusEntry[]) => void;
pollIntervalMs?: number;
probeTimeoutMs?: number;
graceMs?: number;
failuresBeforeStopped?: number;
}) {
this.routeStore = routeStore;
this.onChange = onChange;
this.pollIntervalMs = pollIntervalMs;
this.probeTimeoutMs = probeTimeoutMs;
this.graceMs = graceMs;
this.failuresBeforeStopped = failuresBeforeStopped;
}
start(): void {
if (this.intervalHandle) {
return;
}
const now = Date.now();
for (const route of this.routeStore.listRoutes()) {
this.getOrCreateState(route.hostname, now);
}
this.intervalHandle = setInterval(() => {
void this.poll();
}, this.pollIntervalMs);
}
stop(): void {
if (this.intervalHandle) {
clearInterval(this.intervalHandle);
this.intervalHandle = null;
}
}
private async poll(): Promise<void> {
if (this.pollInFlight) {
return;
}
this.pollInFlight = true;
try {
const routes = this.routeStore.listRoutes();
const activeHostnames = new Set(routes.map((route) => route.hostname));
const changedWorkspaceIds = new Set<string>();
const now = Date.now();
for (const route of routes) {
const state = this.getOrCreateState(route.hostname, now);
if (now - state.registeredAt < this.graceMs) {
continue;
}
const isHealthy = await this.probeRoute(route.port);
const previousStatus = state.status;
if (isHealthy) {
state.consecutiveFailures = 0;
state.status = "running";
} else {
state.consecutiveFailures += 1;
if (state.consecutiveFailures >= this.failuresBeforeStopped) {
state.status = "stopped";
}
}
if (state.status !== null && state.status !== previousStatus) {
changedWorkspaceIds.add(route.workspaceId);
}
}
this.pruneRemovedRoutes(activeHostnames);
for (const workspaceId of changedWorkspaceIds) {
const services = this.buildWorkspaceServiceList(workspaceId);
const snapshot = JSON.stringify(services);
if (snapshot === this.lastEmittedSnapshots.get(workspaceId)) {
continue;
}
this.lastEmittedSnapshots.set(workspaceId, snapshot);
this.onChange(workspaceId, services);
}
} finally {
this.pollInFlight = false;
}
}
private getOrCreateState(hostname: string, registeredAt: number): RouteHealthState {
const existing = this.routeStates.get(hostname);
if (existing) {
return existing;
}
const state: RouteHealthState = {
status: null,
consecutiveFailures: 0,
registeredAt,
};
this.routeStates.set(hostname, state);
return state;
}
private pruneRemovedRoutes(activeHostnames: Set<string>): void {
for (const hostname of this.routeStates.keys()) {
if (activeHostnames.has(hostname)) {
continue;
}
this.routeStates.delete(hostname);
}
}
private buildWorkspaceServiceList(workspaceId: string): ServiceStatusEntry[] {
return this.routeStore
.listRoutesForWorkspace(workspaceId)
.flatMap((route) => {
const state = this.routeStates.get(route.hostname);
if (!state?.status) {
return [];
}
return [this.toServiceStatusEntry(route, state.status)];
});
}
getStatusForHostname(hostname: string): ServiceStatusEntry["status"] | null {
return this.routeStates.get(hostname)?.status ?? null;
}
private toServiceStatusEntry(
route: ServiceRouteEntry,
status: ServiceStatusEntry["status"],
): ServiceStatusEntry {
return {
serviceName: route.serviceName,
hostname: route.hostname,
port: route.port,
status,
};
}
private probeRoute(port: number): Promise<boolean> {
return new Promise((resolve) => {
const socket = net.connect({ host: "127.0.0.1", port });
let settled = false;
const finish = (healthy: boolean) => {
if (settled) {
return;
}
settled = true;
socket.destroy();
resolve(healthy);
};
socket.setTimeout(this.probeTimeoutMs);
socket.once("connect", () => finish(true));
socket.once("timeout", () => finish(false));
socket.once("error", () => finish(false));
});
}
}
+149 -18
View File
@@ -28,9 +28,14 @@ function closeServer(server: http.Server): Promise<void> {
// ---------------------------------------------------------------------------
describe("ServiceRouteStore", () => {
it("addRoute and findRoute with exact match", () => {
it("registerRoute and findRoute with exact match", () => {
const store = new ServiceRouteStore();
store.addRoute("editor.localhost", 3000);
store.registerRoute({
hostname: "editor.localhost",
port: 3000,
workspaceId: "/repo/.paseo/worktrees/feature-a",
serviceName: "editor",
});
const route = store.findRoute("editor.localhost");
expect(route).toEqual({ hostname: "editor.localhost", port: 3000 });
@@ -38,7 +43,12 @@ describe("ServiceRouteStore", () => {
it("findRoute strips port from host header", () => {
const store = new ServiceRouteStore();
store.addRoute("editor.localhost", 3000);
store.registerRoute({
hostname: "editor.localhost",
port: 3000,
workspaceId: "/repo/.paseo/worktrees/feature-a",
serviceName: "editor",
});
const route = store.findRoute("editor.localhost:6767");
expect(route).toEqual({ hostname: "editor.localhost", port: 3000 });
@@ -46,25 +56,132 @@ describe("ServiceRouteStore", () => {
it("findRoute subdomain match", () => {
const store = new ServiceRouteStore();
store.addRoute("editor.localhost", 3000);
store.registerRoute({
hostname: "editor.localhost",
port: 3000,
workspaceId: "/repo/.paseo/worktrees/feature-a",
serviceName: "editor",
});
const route = store.findRoute("fix-auth.editor.localhost");
expect(route).toEqual({ hostname: "editor.localhost", port: 3000 });
});
it("listRoutes returns enriched entries", () => {
const store = new ServiceRouteStore();
store.registerRoute({
hostname: "a.localhost",
port: 3000,
workspaceId: "/repo/.paseo/worktrees/feature-a",
serviceName: "web",
});
store.registerRoute({
hostname: "b.localhost",
port: 4000,
workspaceId: "/repo/.paseo/worktrees/feature-b",
serviceName: "docs",
});
const routes = store.listRoutes();
expect(routes).toHaveLength(2);
expect(routes).toContainEqual({
hostname: "a.localhost",
port: 3000,
workspaceId: "/repo/.paseo/worktrees/feature-a",
serviceName: "web",
});
expect(routes).toContainEqual({
hostname: "b.localhost",
port: 4000,
workspaceId: "/repo/.paseo/worktrees/feature-b",
serviceName: "docs",
});
});
it("listRoutesForWorkspace returns only routes for that workspace", () => {
const store = new ServiceRouteStore();
store.registerRoute({
hostname: "a.localhost",
port: 3000,
workspaceId: "/repo/.paseo/worktrees/feature-a",
serviceName: "web",
});
store.registerRoute({
hostname: "b.localhost",
port: 4000,
workspaceId: "/repo/.paseo/worktrees/feature-b",
serviceName: "docs",
});
store.registerRoute({
hostname: "c.localhost",
port: 5000,
workspaceId: "/repo/.paseo/worktrees/feature-a",
serviceName: "api",
});
expect(store.listRoutesForWorkspace("/repo/.paseo/worktrees/feature-a")).toEqual([
{
hostname: "a.localhost",
port: 3000,
workspaceId: "/repo/.paseo/worktrees/feature-a",
serviceName: "web",
},
{
hostname: "c.localhost",
port: 5000,
workspaceId: "/repo/.paseo/worktrees/feature-a",
serviceName: "api",
},
]);
});
it("removeRoute works", () => {
const store = new ServiceRouteStore();
store.addRoute("editor.localhost", 3000);
store.registerRoute({
hostname: "editor.localhost",
port: 3000,
workspaceId: "/repo/.paseo/worktrees/feature-a",
serviceName: "editor",
});
store.removeRoute("editor.localhost");
expect(store.findRoute("editor.localhost")).toBeNull();
});
it("removeRoute cleans up workspace index", () => {
const store = new ServiceRouteStore();
store.registerRoute({
hostname: "editor.localhost",
port: 3000,
workspaceId: "/repo/.paseo/worktrees/feature-a",
serviceName: "editor",
});
store.removeRoute("editor.localhost");
expect(store.listRoutesForWorkspace("/repo/.paseo/worktrees/feature-a")).toEqual([]);
});
it("removeRoutesForPort works", () => {
const store = new ServiceRouteStore();
store.addRoute("a.localhost", 3000);
store.addRoute("b.localhost", 3000);
store.addRoute("c.localhost", 4000);
store.registerRoute({
hostname: "a.localhost",
port: 3000,
workspaceId: "/repo/.paseo/worktrees/feature-a",
serviceName: "web",
});
store.registerRoute({
hostname: "b.localhost",
port: 3000,
workspaceId: "/repo/.paseo/worktrees/feature-a",
serviceName: "api",
});
store.registerRoute({
hostname: "c.localhost",
port: 4000,
workspaceId: "/repo/.paseo/worktrees/feature-b",
serviceName: "docs",
});
store.removeRoutesForPort(3000);
@@ -76,22 +193,36 @@ describe("ServiceRouteStore", () => {
});
});
it("findRoute returns null for unknown hosts", () => {
it("removeRoutesForPort cleans up workspace index", () => {
const store = new ServiceRouteStore();
store.addRoute("editor.localhost", 3000);
store.registerRoute({
hostname: "a.localhost",
port: 3000,
workspaceId: "/repo/.paseo/worktrees/feature-a",
serviceName: "web",
});
store.registerRoute({
hostname: "b.localhost",
port: 3000,
workspaceId: "/repo/.paseo/worktrees/feature-a",
serviceName: "api",
});
expect(store.findRoute("unknown.example.com")).toBeNull();
store.removeRoutesForPort(3000);
expect(store.listRoutesForWorkspace("/repo/.paseo/worktrees/feature-a")).toEqual([]);
});
it("listRoutes returns all routes", () => {
it("findRoute returns null for unknown hosts", () => {
const store = new ServiceRouteStore();
store.addRoute("a.localhost", 3000);
store.addRoute("b.localhost", 4000);
store.registerRoute({
hostname: "editor.localhost",
port: 3000,
workspaceId: "/repo/.paseo/worktrees/feature-a",
serviceName: "editor",
});
const routes = store.listRoutes();
expect(routes).toHaveLength(2);
expect(routes).toContainEqual({ hostname: "a.localhost", port: 3000 });
expect(routes).toContainEqual({ hostname: "b.localhost", port: 4000 });
expect(store.findRoute("unknown.example.com")).toBeNull();
});
});
+74 -15
View File
@@ -29,21 +29,49 @@ export interface ServiceRoute {
port: number;
}
export interface ServiceRouteEntry extends ServiceRoute {
workspaceId: string;
serviceName: string;
}
export class ServiceRouteStore {
private routes = new Map<string, number>();
private routes = new Map<string, ServiceRouteEntry>();
private workspaceHostnames = new Map<string, Set<string>>();
addRoute(hostname: string, port: number): void {
this.routes.set(hostname, port);
this.registerRoute({
hostname,
port,
workspaceId: "",
serviceName: hostname,
});
}
registerRoute(entry: ServiceRouteEntry): void {
const previous = this.routes.get(entry.hostname);
if (previous) {
this.removeHostnameFromWorkspaceIndex(previous.workspaceId, previous.hostname);
}
const storedEntry = { ...entry };
this.routes.set(storedEntry.hostname, storedEntry);
this.addHostnameToWorkspaceIndex(storedEntry.workspaceId, storedEntry.hostname);
}
removeRoute(hostname: string): void {
const entry = this.routes.get(hostname);
if (!entry) {
return;
}
this.routes.delete(hostname);
this.removeHostnameFromWorkspaceIndex(entry.workspaceId, hostname);
}
removeRoutesForPort(port: number): void {
for (const [hostname, p] of this.routes) {
if (p === port) {
for (const [hostname, entry] of this.routes) {
if (entry.port === port) {
this.routes.delete(hostname);
this.removeHostnameFromWorkspaceIndex(entry.workspaceId, hostname);
}
}
}
@@ -53,29 +81,60 @@ export class ServiceRouteStore {
const hostname = host.replace(/:\d+$/, "");
// 1. Exact match
const exactPort = this.routes.get(hostname);
if (exactPort !== undefined) {
return { hostname, port: exactPort };
const exactRoute = this.routes.get(hostname);
if (exactRoute !== undefined) {
return { hostname: exactRoute.hostname, port: exactRoute.port };
}
// 2. Subdomain match — walk up the labels looking for a registered parent
const parts = hostname.split(".");
for (let i = 1; i < parts.length; i++) {
const candidate = parts.slice(i).join(".");
const candidatePort = this.routes.get(candidate);
if (candidatePort !== undefined) {
return { hostname: candidate, port: candidatePort };
const candidateRoute = this.routes.get(candidate);
if (candidateRoute !== undefined) {
return { hostname: candidateRoute.hostname, port: candidateRoute.port };
}
}
return null;
}
listRoutes(): ServiceRoute[] {
return Array.from(this.routes.entries()).map(([hostname, port]) => ({
hostname,
port,
}));
listRoutes(): ServiceRouteEntry[] {
return Array.from(this.routes.values()).map((entry) => ({ ...entry }));
}
listRoutesForWorkspace(workspaceId: string): ServiceRouteEntry[] {
const hostnames = this.workspaceHostnames.get(workspaceId);
if (!hostnames) {
return [];
}
const routes: ServiceRouteEntry[] = [];
for (const hostname of hostnames) {
const entry = this.routes.get(hostname);
if (entry) {
routes.push({ ...entry });
}
}
return routes;
}
private addHostnameToWorkspaceIndex(workspaceId: string, hostname: string): void {
const hostnames = this.workspaceHostnames.get(workspaceId) ?? new Set<string>();
hostnames.add(hostname);
this.workspaceHostnames.set(workspaceId, hostnames);
}
private removeHostnameFromWorkspaceIndex(workspaceId: string, hostname: string): void {
const hostnames = this.workspaceHostnames.get(workspaceId);
if (!hostnames) {
return;
}
hostnames.delete(hostname);
if (hostnames.size === 0) {
this.workspaceHostnames.delete(workspaceId);
}
}
}
@@ -0,0 +1,269 @@
import { describe, expect, it, vi } from "vitest";
import { ServiceRouteStore } from "./service-proxy.js";
import {
buildWorkspaceServicePayloads,
createServiceStatusEmitter,
} from "./service-status-projection.js";
describe("service-status-projection", () => {
it("buildWorkspaceServicePayloads returns service payloads from workspace routes", () => {
const routeStore = new ServiceRouteStore();
routeStore.registerRoute({
hostname: "api.localhost",
port: 3001,
workspaceId: "workspace-a",
serviceName: "api",
});
routeStore.registerRoute({
hostname: "docs.localhost",
port: 3002,
workspaceId: "workspace-b",
serviceName: "docs",
});
routeStore.registerRoute({
hostname: "web.localhost",
port: 3003,
workspaceId: "workspace-a",
serviceName: "web",
});
expect(buildWorkspaceServicePayloads(routeStore, "workspace-a", 6767)).toEqual([
{
serviceName: "api",
hostname: "api.localhost",
port: 3001,
url: "http://api.localhost:6767",
status: "stopped",
},
{
serviceName: "web",
hostname: "web.localhost",
port: 3003,
url: "http://web.localhost:6767",
status: "stopped",
},
]);
});
it("computes URLs with and without a daemon port", () => {
const routeStore = new ServiceRouteStore();
routeStore.registerRoute({
hostname: "api.localhost",
port: 3001,
workspaceId: "workspace-a",
serviceName: "api",
});
expect(buildWorkspaceServicePayloads(routeStore, "workspace-a", 6767)).toEqual([
{
serviceName: "api",
hostname: "api.localhost",
port: 3001,
url: "http://api.localhost:6767",
status: "stopped",
},
]);
expect(buildWorkspaceServicePayloads(routeStore, "workspace-a", null)).toEqual([
{
serviceName: "api",
hostname: "api.localhost",
port: 3001,
url: null,
status: "stopped",
},
]);
});
it("createServiceStatusEmitter emits updates to all active sessions", () => {
const routeStore = new ServiceRouteStore();
routeStore.registerRoute({
hostname: "api.localhost",
port: 3001,
workspaceId: "workspace-a",
serviceName: "api",
});
const sessionA = { emit: vi.fn() };
const sessionB = { emit: vi.fn() };
const emitUpdate = createServiceStatusEmitter({
sessions: () => [sessionA, sessionB],
routeStore,
daemonPort: 6767,
});
emitUpdate("workspace-a", [
{
serviceName: "api",
hostname: "api.localhost",
port: 3001,
status: "running",
},
]);
expect(sessionA.emit).toHaveBeenCalledWith({
type: "service_status_update",
payload: {
workspaceId: "workspace-a",
services: [
{
serviceName: "api",
hostname: "api.localhost",
port: 3001,
url: "http://api.localhost:6767",
status: "running",
},
],
},
});
expect(sessionB.emit).toHaveBeenCalledWith({
type: "service_status_update",
payload: {
workspaceId: "workspace-a",
services: [
{
serviceName: "api",
hostname: "api.localhost",
port: 3001,
url: "http://api.localhost:6767",
status: "running",
},
],
},
});
});
it("uses resolveStatus to set initial service status when provided", () => {
const routeStore = new ServiceRouteStore();
routeStore.registerRoute({
hostname: "api.localhost",
port: 3001,
workspaceId: "workspace-a",
serviceName: "api",
});
routeStore.registerRoute({
hostname: "web.localhost",
port: 3003,
workspaceId: "workspace-a",
serviceName: "web",
});
const statuses = new Map<string, "running" | "stopped">([
["api.localhost", "running"],
]);
expect(
buildWorkspaceServicePayloads(routeStore, "workspace-a", 6767, (hostname) =>
statuses.get(hostname) ?? null,
),
).toEqual([
{
serviceName: "api",
hostname: "api.localhost",
port: 3001,
url: "http://api.localhost:6767",
status: "running",
},
{
serviceName: "web",
hostname: "web.localhost",
port: 3003,
url: "http://web.localhost:6767",
status: "stopped",
},
]);
});
it("emits workspace-specific batches", () => {
const routeStore = new ServiceRouteStore();
routeStore.registerRoute({
hostname: "api.localhost",
port: 3001,
workspaceId: "workspace-a",
serviceName: "api",
});
routeStore.registerRoute({
hostname: "web.localhost",
port: 3002,
workspaceId: "workspace-a",
serviceName: "web",
});
routeStore.registerRoute({
hostname: "docs.localhost",
port: 3003,
workspaceId: "workspace-b",
serviceName: "docs",
});
const session = { emit: vi.fn() };
const emitUpdate = createServiceStatusEmitter({
sessions: () => [session],
routeStore,
daemonPort: null,
});
emitUpdate("workspace-a", [
{
serviceName: "api",
hostname: "api.localhost",
port: 3001,
status: "running",
},
{
serviceName: "web",
hostname: "web.localhost",
port: 3002,
status: "stopped",
},
]);
emitUpdate("workspace-b", [
{
serviceName: "docs",
hostname: "docs.localhost",
port: 3003,
status: "running",
},
]);
expect(session.emit).toHaveBeenNthCalledWith(1, {
type: "service_status_update",
payload: {
workspaceId: "workspace-a",
services: [
{
serviceName: "api",
hostname: "api.localhost",
port: 3001,
url: null,
status: "running",
},
{
serviceName: "web",
hostname: "web.localhost",
port: 3002,
url: null,
status: "stopped",
},
],
},
});
expect(session.emit).toHaveBeenNthCalledWith(2, {
type: "service_status_update",
payload: {
workspaceId: "workspace-b",
services: [
{
serviceName: "docs",
hostname: "docs.localhost",
port: 3003,
url: null,
status: "running",
},
],
},
});
});
});
@@ -0,0 +1,86 @@
import type {
ServiceStatusUpdateMessage,
SessionOutboundMessage,
WorkspaceServicePayload,
} from "../shared/messages.js";
import type { ServiceStatusEntry } from "./service-health-monitor.js";
import type { ServiceRouteStore } from "./service-proxy.js";
type SessionEmitter = {
emit(message: SessionOutboundMessage): void;
};
function resolveDaemonPort(daemonPort: number | null | (() => number | null)): number | null {
if (typeof daemonPort === "function") {
return daemonPort();
}
return daemonPort;
}
function toServiceUrl(hostname: string, daemonPort: number | null): string | null {
if (daemonPort === null) {
return null;
}
return `http://${hostname}:${daemonPort}`;
}
export function buildWorkspaceServicePayloads(
routeStore: ServiceRouteStore,
workspaceId: string,
daemonPort: number | null,
resolveStatus?: (hostname: string) => "running" | "stopped" | null,
): WorkspaceServicePayload[] {
return routeStore.listRoutesForWorkspace(workspaceId).map((route) => ({
serviceName: route.serviceName,
hostname: route.hostname,
port: route.port,
url: toServiceUrl(route.hostname, daemonPort),
status: resolveStatus?.(route.hostname) ?? "stopped",
}));
}
function buildServiceStatusUpdateMessage(params: {
workspaceId: string;
services: WorkspaceServicePayload[];
}): ServiceStatusUpdateMessage {
return {
type: "service_status_update",
payload: {
workspaceId: params.workspaceId,
services: params.services,
},
};
}
export function createServiceStatusEmitter({
sessions,
routeStore,
daemonPort,
}: {
sessions: () => SessionEmitter[];
routeStore: ServiceRouteStore;
daemonPort: number | null | (() => number | null);
}): (workspaceId: string, services: ServiceStatusEntry[]) => void {
return (workspaceId, services) => {
const resolvedDaemonPort = resolveDaemonPort(daemonPort);
const serviceStatusByHostname = new Map(
services.map((service) => [service.hostname, service.status] as const),
);
const projected = buildWorkspaceServicePayloads(routeStore, workspaceId, resolvedDaemonPort).map(
(service) => ({
...service,
status: serviceStatusByHostname.get(service.hostname) ?? service.status,
}),
);
const message = buildServiceStatusUpdateMessage({
workspaceId,
services: projected,
});
for (const session of sessions()) {
session.emit(message);
}
};
}
+39
View File
@@ -28,6 +28,7 @@ import {
type UnsubscribeCheckoutDiffRequest,
type DirectorySuggestionsRequest,
type ProjectPlacementPayload,
type WorkspaceSetupSnapshot,
type WorkspaceDescriptorPayload,
type WorkspaceStateBucket,
} from "./messages.js";
@@ -62,6 +63,7 @@ import {
import { experimental_createMCPClient } from "ai";
import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js";
import type { VoiceCallerContext, VoiceMcpStdioConfig, VoiceSpeakHandler } from "./voice-types.js";
import { buildWorkspaceServicePayloads } from "./service-status-projection.js";
export type AgentMcpTransportFactory = () => Promise<Transport>;
import { buildProviderRegistry } from "./agent/provider-registry.js";
@@ -180,6 +182,7 @@ import {
handleCreatePaseoWorktreeRequest as handleCreateWorktreeRequest,
handlePaseoWorktreeArchiveRequest as handleWorktreeArchiveRequest,
handlePaseoWorktreeListRequest as handleWorktreeListRequest,
handleWorkspaceSetupStatusRequest as handleWorkspaceSetupStatusRequestMessage,
killTerminalsUnderPath as killWorktreeTerminalsUnderPath,
registerPendingWorktreeWorkspace as registerPendingWorktreeWorkspaceSession,
} from "./worktree-session.js";
@@ -375,6 +378,7 @@ export type SessionOptions = {
terminalManager: TerminalManager | null;
serviceRouteStore?: ServiceRouteStore;
getDaemonTcpPort?: () => number | null;
resolveServiceStatus?: (hostname: string) => "running" | "stopped" | null;
voice?: {
voiceAgentMcpStdio?: VoiceMcpStdioConfig | null;
turnDetection?: Resolvable<TurnDetectionProvider | null>;
@@ -572,6 +576,7 @@ export class Session {
private readonly terminalManager: TerminalManager | null;
private readonly serviceRouteStore: ServiceRouteStore | null;
private readonly getDaemonTcpPort: (() => number | null) | null;
private readonly resolveServiceStatus: ((hostname: string) => "running" | "stopped" | null) | null;
private readonly subscribedTerminalDirectories = new Set<string>();
private unsubscribeTerminalsChanged: (() => void) | null = null;
private terminalExitSubscriptions: Map<string, () => void> = new Map();
@@ -582,6 +587,7 @@ export class Session {
private peakInflightRequests = 0;
private readonly checkoutDiffSubscriptions = new Map<string, () => void>();
private readonly workspaceGitWatchTargets = new Map<string, WorkspaceGitWatchTarget>();
private readonly workspaceSetupSnapshots = new Map<string, WorkspaceSetupSnapshot>();
private readonly voiceAgentMcpStdio: VoiceMcpStdioConfig | null;
private readonly registerVoiceSpeakHandler?: (
agentId: string,
@@ -625,6 +631,7 @@ export class Session {
terminalManager,
serviceRouteStore,
getDaemonTcpPort,
resolveServiceStatus,
voice,
voiceBridge,
dictation,
@@ -651,6 +658,7 @@ export class Session {
this.terminalManager = terminalManager;
this.serviceRouteStore = serviceRouteStore ?? null;
this.getDaemonTcpPort = getDaemonTcpPort ?? null;
this.resolveServiceStatus = resolveServiceStatus ?? null;
if (this.terminalManager) {
this.unsubscribeTerminalsChanged = this.terminalManager.subscribeTerminalsChanged((event) =>
this.handleTerminalsChanged(event),
@@ -716,6 +724,10 @@ export class Session {
};
}
public emitServerMessage(message: SessionOutboundMessage): void {
this.emit(message);
}
/**
* Send initial state to client after connection
*/
@@ -1617,6 +1629,10 @@ export class Session {
await this.handleCreatePaseoWorktreeRequest(msg);
break;
case "workspace_setup_status_request":
await this.handleWorkspaceSetupStatusRequest(msg);
break;
case "open_project_request":
await this.handleOpenProjectRequest(msg);
break;
@@ -4918,6 +4934,14 @@ export class Session {
status: "done",
activityAt: null,
diffStat,
services: this.serviceRouteStore
? buildWorkspaceServicePayloads(
this.serviceRouteStore,
workspace.workspaceId,
this.getDaemonTcpPort?.() ?? null,
this.resolveServiceStatus ?? undefined,
)
: [],
};
}
@@ -5565,6 +5589,9 @@ export class Session {
paseoHome: this.paseoHome,
emitWorkspaceUpdateForCwd: (cwd, emitOptions) =>
this.emitWorkspaceUpdateForCwd(cwd, emitOptions),
cacheWorkspaceSetupSnapshot: (workspaceId, snapshot) => {
this.workspaceSetupSnapshots.set(workspaceId, snapshot);
},
emit: (message) => this.emit(message),
sessionLogger: this.sessionLogger,
terminalManager: this.terminalManager,
@@ -5576,6 +5603,18 @@ export class Session {
);
}
private async handleWorkspaceSetupStatusRequest(
request: Extract<SessionInboundMessage, { type: "workspace_setup_status_request" }>,
): Promise<void> {
return handleWorkspaceSetupStatusRequestMessage(
{
emit: (message) => this.emit(message),
workspaceSetupSnapshots: this.workspaceSetupSnapshots,
},
request,
);
}
private async handleArchiveWorkspaceRequest(
request: Extract<SessionInboundMessage, { type: "archive_workspace_request" }>,
): Promise<void> {
@@ -244,6 +244,7 @@ export class VoiceAssistantWebSocketServer {
private readonly terminalManager: TerminalManager | null;
private readonly serviceRouteStore: ServiceRouteStore | null;
private readonly getDaemonTcpPort: (() => number | null) | null;
private readonly resolveServiceStatus: ((hostname: string) => "running" | "stopped" | null) | null;
private readonly dictation: {
finalTimeoutMs?: number;
} | null;
@@ -312,6 +313,7 @@ export class VoiceAssistantWebSocketServer {
checkoutDiffManager?: CheckoutDiffManager,
serviceRouteStore?: ServiceRouteStore | null,
getDaemonTcpPort?: () => number | null,
resolveServiceStatus?: (hostname: string) => "running" | "stopped" | null,
) {
this.logger = logger.child({ module: "websocket-server" });
this.serverId = serverId;
@@ -350,6 +352,7 @@ export class VoiceAssistantWebSocketServer {
this.onLifecycleIntent = onLifecycleIntent ?? null;
this.serviceRouteStore = serviceRouteStore ?? null;
this.getDaemonTcpPort = getDaemonTcpPort ?? null;
this.resolveServiceStatus = resolveServiceStatus ?? null;
this.serverCapabilities = buildServerCapabilities({
readiness: this.speech?.getReadiness() ?? null,
});
@@ -420,6 +423,16 @@ export class VoiceAssistantWebSocketServer {
}
}
public listActiveSessions(): Session[] {
return Array.from(
new Set(
[...this.sessions.values(), ...this.externalSessionsByKey.values()].map(
(connection) => connection.session,
),
),
);
}
public publishSpeechReadiness(readiness: SpeechReadinessSnapshot | null): void {
this.updateServerCapabilities(buildServerCapabilities({ readiness }));
}
@@ -652,6 +665,7 @@ export class VoiceAssistantWebSocketServer {
terminalManager: this.terminalManager,
serviceRouteStore: this.serviceRouteStore ?? undefined,
getDaemonTcpPort: this.getDaemonTcpPort ?? undefined,
resolveServiceStatus: this.resolveServiceStatus ?? undefined,
voice: {
...(this.voice ?? {}),
turnDetection: () => this.speech?.resolveTurnDetection() ?? null,
@@ -5,7 +5,12 @@ import { join } from "path";
import { tmpdir } from "os";
import type { AgentTimelineItem } from "./agent/agent-sdk-types.js";
import { createAgentWorktree, runAsyncWorktreeBootstrap } from "./worktree-bootstrap.js";
import {
createAgentWorktree,
runAsyncWorktreeBootstrap,
spawnWorktreeServices,
} from "./worktree-bootstrap.js";
import { ServiceRouteStore } from "./service-proxy.js";
describe("runAsyncWorktreeBootstrap", () => {
let tempDir: string;
@@ -114,12 +119,15 @@ describe("runAsyncWorktreeBootstrap", () => {
expect(persistedSetupItems[0].detail.commands[0]).toMatchObject({
index: 1,
command: 'echo "line-one"; echo "line-two" 1>&2',
log: expect.stringContaining("line-one"),
status: "completed",
exitCode: 0,
});
expect(persistedSetupItems[0].detail.commands[0]?.log).toContain("line-two");
expect(persistedSetupItems[0].detail.commands[1]).toMatchObject({
index: 2,
command: 'echo "line-three"',
log: "line-three\n",
status: "completed",
exitCode: 0,
});
@@ -247,6 +255,64 @@ describe("runAsyncWorktreeBootstrap", () => {
expect(persistedSetupItem.detail.log).toContain("prefix-");
expect(persistedSetupItem.detail.log).toContain("-suffix");
expect(persistedSetupItem.detail.log).toContain("...<output truncated in the middle>...");
expect(persistedSetupItem.detail.commands[0]?.log).toContain("prefix-");
expect(persistedSetupItem.detail.commands[0]?.log).toContain("-suffix");
expect(persistedSetupItem.detail.commands[0]?.log).toContain(
"...<output truncated in the middle>...",
);
});
it("keeps only the final carriage-return-updated content in command logs", async () => {
writeFileSync(
join(repoDir, "paseo.json"),
JSON.stringify({
worktree: {
setup: [
`node -e "process.stdout.write('fetch 1/3\\\\rfetch 2/3\\\\rfetch 3/3\\\\nready\\\\n')"`,
],
},
}),
);
execSync("git add paseo.json", { cwd: repoDir, stdio: "pipe" });
execSync("git -c commit.gpgsign=false commit -m 'add carriage return setup'", {
cwd: repoDir,
stdio: "pipe",
});
const worktreeBootstrap = await createAgentWorktree({
cwd: repoDir,
branchName: "feature-carriage-return",
baseBranch: "main",
worktreeSlug: "feature-carriage-return",
paseoHome,
});
const persisted: AgentTimelineItem[] = [];
await runAsyncWorktreeBootstrap({
agentId: "agent-carriage-return",
worktree: worktreeBootstrap.worktree,
shouldBootstrap: worktreeBootstrap.shouldBootstrap,
terminalManager: null,
appendTimelineItem: async (item) => {
persisted.push(item);
return true;
},
emitLiveTimelineItem: async () => true,
});
const persistedSetupItem = persisted.find(
(item): item is Extract<AgentTimelineItem, { type: "tool_call" }> =>
item.type === "tool_call" && item.name === "paseo_worktree_setup",
);
expect(persistedSetupItem?.detail.type).toBe("worktree_setup");
if (!persistedSetupItem || persistedSetupItem.detail.type !== "worktree_setup") {
throw new Error("Expected worktree_setup tool detail");
}
expect(persistedSetupItem.detail.log).toContain("\nfetch 3/3\nready\n");
expect(persistedSetupItem.detail.log).not.toContain("\nfetch 1/3\n");
expect(persistedSetupItem.detail.log).not.toContain("\nfetch 2/3\n");
expect(persistedSetupItem.detail.commands[0]?.log).toBe("fetch 3/3\nready\n");
});
it("waits for terminal output before sending bootstrap commands", async () => {
@@ -440,4 +506,85 @@ describe("runAsyncWorktreeBootstrap", () => {
);
expect(terminalToolCall?.status).toBe("completed");
});
it("spawns services without PASEO_SERVICE_URL when the daemon has no TCP port", async () => {
writeFileSync(
join(repoDir, "paseo.json"),
JSON.stringify({
services: {
web: {
command: "npm run dev",
},
},
}),
);
execSync("git add paseo.json", { cwd: repoDir, stdio: "pipe" });
execSync("git -c commit.gpgsign=false commit -m 'add service config'", {
cwd: repoDir,
stdio: "pipe",
});
const routeStore = new ServiceRouteStore();
const createTerminalCalls: Array<{ cwd: string; name?: string; env?: Record<string, string> }> = [];
const results = await spawnWorktreeServices({
repoRoot: repoDir,
workspaceId: repoDir,
branchName: "feature-socket-service",
daemonPort: null,
routeStore,
terminalManager: {
async getTerminals() {
return [];
},
async createTerminal(options) {
createTerminalCalls.push(options);
return {
id: "term-service",
name: options.name ?? "Terminal",
cwd: options.cwd,
send: () => {},
subscribe: () => () => {},
onExit: () => () => {},
getState: () => ({
rows: 1,
cols: 1,
grid: [[{ char: "$" }]],
scrollback: [],
cursor: { row: 0, col: 0 },
}),
kill: () => {},
};
},
registerCwdEnv() {},
getTerminal() {
return undefined;
},
killTerminal() {},
listDirectories() {
return [];
},
killAll() {},
subscribeTerminalsChanged() {
return () => {};
},
},
});
expect(results).toHaveLength(1);
expect(routeStore.listRoutes()).toEqual([
{
hostname: "feature-socket-service.web.localhost",
port: expect.any(Number),
workspaceId: repoDir,
serviceName: "web",
},
]);
expect(createTerminalCalls).toHaveLength(1);
expect(createTerminalCalls[0]?.cwd).toBe(repoDir);
expect(createTerminalCalls[0]?.name).toBe("web");
expect(createTerminalCalls[0]?.env?.PORT).toEqual(expect.any(String));
expect(createTerminalCalls[0]?.env?.HOST).toBe("127.0.0.1");
expect(createTerminalCalls[0]?.env?.PASEO_SERVICE_URL).toBeUndefined();
});
});
+114 -56
View File
@@ -10,6 +10,7 @@ import {
getServiceConfigs,
getWorktreeTerminalSpecs,
listPaseoWorktrees,
processCarriageReturns,
resolveWorktreeRuntimeEnv,
runWorktreeSetupCommands,
slugify,
@@ -254,15 +255,13 @@ function buildWorktreeSetupLog(input: {
const total = results.length;
for (const [index, result] of results.entries()) {
lines.push(`==> [${index + 1}/${total}] Running: ${result.command}`);
const accumulator = outputAccumulatorsByIndex?.get(index + 1);
const output = accumulator
? renderMiddleTruncationAccumulator(accumulator)
: truncateTextInMiddle(
`${result.stdout ?? ""}${result.stderr ?? ""}`,
MAX_WORKTREE_SETUP_COMMAND_OUTPUT_BYTES,
);
if (output.text.length > 0) {
lines.push(output.text.replace(/\n$/, ""));
const output = buildWorktreeSetupCommandLog({
index: index + 1,
result,
outputAccumulatorsByIndex,
});
if (output.log.length > 0) {
lines.push(output.log.replace(/\n$/, ""));
}
if (output.truncated) {
anyTruncated = true;
@@ -279,6 +278,26 @@ function buildWorktreeSetupLog(input: {
};
}
function buildWorktreeSetupCommandLog(input: {
index: number;
result: WorktreeSetupCommandResult;
outputAccumulatorsByIndex?: Map<number, WorktreeSetupOutputAccumulator>;
}): { log: string; truncated: boolean } {
const { index, result, outputAccumulatorsByIndex } = input;
const accumulator = outputAccumulatorsByIndex?.get(index);
const rendered = accumulator
? renderMiddleTruncationAccumulator(accumulator)
: truncateTextInMiddle(
`${result.stdout ?? ""}${result.stderr ?? ""}`,
MAX_WORKTREE_SETUP_COMMAND_OUTPUT_BYTES,
);
return {
log: processCarriageReturns(rendered.text),
truncated: rendered.truncated,
};
}
export function createWorktreeSetupProgressAccumulator(): WorktreeSetupProgressAccumulator {
return {
resultsByIndex: new Map(),
@@ -341,14 +360,26 @@ export function buildWorktreeSetupDetail(input: {
results: WorktreeSetupCommandResult[];
outputAccumulatorsByIndex?: Map<number, WorktreeSetupOutputAccumulator>;
}): Extract<ToolCallDetail, { type: "worktree_setup" }> {
const commands = input.results.map((result, index) => ({
index: index + 1,
command: result.command,
cwd: result.cwd,
status: commandStatusFromResult(result),
exitCode: result.exitCode,
...(result.durationMs > 0 ? { durationMs: result.durationMs } : {}),
}));
let anyCommandTruncated = false;
const commands = input.results.map((result, index) => {
const renderedLog = buildWorktreeSetupCommandLog({
index: index + 1,
result,
outputAccumulatorsByIndex: input.outputAccumulatorsByIndex,
});
if (renderedLog.truncated) {
anyCommandTruncated = true;
}
return {
index: index + 1,
command: result.command,
cwd: result.cwd,
log: renderedLog.log,
status: commandStatusFromResult(result),
exitCode: result.exitCode,
...(result.durationMs > 0 ? { durationMs: result.durationMs } : {}),
};
});
const renderedLog = buildWorktreeSetupLog({
results: input.results,
outputAccumulatorsByIndex: input.outputAccumulatorsByIndex,
@@ -360,7 +391,7 @@ export function buildWorktreeSetupDetail(input: {
branchName: input.worktree.branchName,
log: renderedLog.log,
commands,
...(renderedLog.truncated ? { truncated: true } : {}),
...(renderedLog.truncated || anyCommandTruncated ? { truncated: true } : {}),
};
}
@@ -704,18 +735,14 @@ export async function runAsyncWorktreeBootstrap(
await runWorktreeTerminalBootstrap(options, runtimeEnv);
if (
!options.terminalManager ||
!options.serviceRouteStore ||
options.daemonPort === null ||
options.daemonPort === undefined
) {
if (!options.terminalManager || !options.serviceRouteStore) {
return;
}
try {
await spawnWorktreeServices({
repoRoot: options.worktree.worktreePath,
workspaceId: options.worktree.worktreePath,
branchName: options.worktree.branchName,
daemonPort: options.daemonPort,
routeStore: options.serviceRouteStore,
@@ -743,13 +770,15 @@ export interface WorktreeServiceResult {
export async function spawnWorktreeServices(options: {
repoRoot: string;
workspaceId: string;
branchName: string | null;
daemonPort: number;
daemonPort?: number | null;
routeStore: ServiceRouteStore;
terminalManager: TerminalManager;
logger?: Logger;
}): Promise<WorktreeServiceResult[]> {
const { repoRoot, branchName, daemonPort, routeStore, terminalManager, logger } = options;
const { repoRoot, workspaceId, branchName, daemonPort, routeStore, terminalManager, logger } =
options;
const serviceConfigs = getServiceConfigs(repoRoot);
if (serviceConfigs.size === 0) {
return [];
@@ -758,43 +787,72 @@ export async function spawnWorktreeServices(options: {
const results: WorktreeServiceResult[] = [];
for (const [serviceName, config] of serviceConfigs) {
const port = config.port ?? (await findFreePort());
const branchHostnameLabel = branchName ? slugify(branchName) : null;
let hostname: string | null = null;
let port: number | null = null;
const isDefaultBranch =
branchName === null || branchName === "main" || branchName === "master";
const hostname = isDefaultBranch
? `${serviceName}.localhost`
: `${branchHostnameLabel}.${serviceName}.localhost`;
try {
port = config.port ?? (await findFreePort());
const branchHostnameLabel = branchName ? slugify(branchName) : null;
routeStore.addRoute(hostname, port);
const isDefaultBranch =
branchName === null || branchName === "main" || branchName === "master";
hostname = isDefaultBranch
? `${serviceName}.localhost`
: `${branchHostnameLabel}.${serviceName}.localhost`;
const env: Record<string, string> = {
PORT: String(port),
HOST: "127.0.0.1",
PASEO_SERVICE_URL: `http://${hostname}:${daemonPort}`,
};
routeStore.registerRoute({
hostname,
port,
workspaceId,
serviceName,
});
const terminal = await terminalManager.createTerminal({
cwd: repoRoot,
name: serviceName,
env,
});
const env: Record<string, string> = {
PORT: String(port),
HOST: "127.0.0.1",
};
if (daemonPort !== null && daemonPort !== undefined) {
env.PASEO_SERVICE_URL = `http://${hostname}:${daemonPort}`;
}
await waitForTerminalBootstrapReadiness(terminal);
terminal.send({ type: "input", data: `${config.command}\r` });
const terminal = await terminalManager.createTerminal({
cwd: repoRoot,
name: serviceName,
env,
});
logger?.info(
{ serviceName, hostname, port, terminalId: terminal.id },
`Registered service proxy: ${hostname} -> 127.0.0.1:${port}`,
);
await waitForTerminalBootstrapReadiness(terminal);
terminal.send({ type: "input", data: `${config.command}\r` });
results.push({
serviceName,
hostname,
port,
terminalId: terminal.id,
});
logger?.info(
{ serviceName, hostname, port, terminalId: terminal.id },
`Registered service proxy: ${hostname} -> 127.0.0.1:${port}`,
);
results.push({
serviceName,
hostname,
port,
terminalId: terminal.id,
});
} catch (error) {
if (hostname && port !== null) {
routeStore.removeRoute(hostname);
}
logger?.error(
{
err: error,
serviceName,
repoRoot,
branchName,
hostname,
port,
command: config.command,
},
"Failed to spawn worktree service",
);
throw error;
}
}
return results;
@@ -1,13 +1,17 @@
import { execSync } from "node:child_process";
import { mkdtempSync, readFileSync, realpathSync, rmSync, writeFileSync } from "node:fs";
import { existsSync, mkdtempSync, readFileSync, realpathSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import { afterEach, describe, expect, test, vi } from "vitest";
import type { SessionOutboundMessage } from "./messages.js";
import { ServiceRouteStore } from "./service-proxy.js";
import { createPaseoWorktreeInBackground } from "./worktree-session.js";
import { computeWorktreePath, createWorktree } from "../utils/worktree.js";
import {
createPaseoWorktreeInBackground,
handleCreatePaseoWorktreeRequest,
handleWorkspaceSetupStatusRequest,
} from "./worktree-session.js";
import { createWorktree } from "../utils/worktree.js";
function createLogger() {
return {
@@ -109,8 +113,17 @@ describe("createPaseoWorktreeInBackground", () => {
cleanupPaths.push(tempDir);
const paseoHome = path.join(tempDir, ".paseo");
const worktreePath = await computeWorktreePath(repoDir, "feature-no-setup", paseoHome);
const createdWorktree = await createWorktree({
branchName: "feature-no-setup",
cwd: repoDir,
baseBranch: "main",
worktreeSlug: "feature-no-setup",
runSetup: false,
paseoHome,
});
const worktreePath = createdWorktree.worktreePath;
const emitted: SessionOutboundMessage[] = [];
const snapshots = new Map<string, unknown>();
const routeStore = new ServiceRouteStore();
const logger = createLogger();
const terminalManager = createTerminalManagerStub();
@@ -121,6 +134,7 @@ describe("createPaseoWorktreeInBackground", () => {
{
paseoHome,
emitWorkspaceUpdateForCwd,
cacheWorkspaceSetupSnapshot: (workspaceId, snapshot) => snapshots.set(workspaceId, snapshot),
emit: (message) => emitted.push(message),
sessionLogger: logger,
terminalManager: terminalManager.manager,
@@ -131,9 +145,11 @@ describe("createPaseoWorktreeInBackground", () => {
{
requestCwd: repoDir,
repoRoot: repoDir,
baseBranch: "main",
slug: "feature-no-setup",
worktreePath,
worktree: {
branchName: "feature-no-setup",
worktreePath,
},
shouldBootstrap: true,
},
);
@@ -166,12 +182,30 @@ describe("createPaseoWorktreeInBackground", () => {
commands: [],
},
});
expect(snapshots.get(worktreePath)).toMatchObject({
status: "completed",
error: null,
detail: {
type: "worktree_setup",
worktreePath,
branchName: "feature-no-setup",
log: "",
commands: [],
},
});
expect(routeStore.listRoutes()).toEqual([
{ hostname: "feature-no-setup.web.localhost", port: expect.any(Number) },
{
hostname: "feature-no-setup.web.localhost",
port: expect.any(Number),
workspaceId: worktreePath,
serviceName: "web",
},
]);
expect(terminalManager.terminals).toHaveLength(1);
expect(terminalManager.terminals[0]?.cwd).toBe(worktreePath);
expect(terminalManager.terminals[0]?.env?.PORT).toEqual(expect.any(String));
expect(terminalManager.terminals[0]?.env?.PASEO_SERVICE_URL).toBeDefined();
expect(terminalManager.terminals[0]?.sent).toEqual(["npm run dev\r"]);
expect(archiveWorkspaceRecord).not.toHaveBeenCalled();
expect(emitWorkspaceUpdateForCwd).toHaveBeenCalledWith(worktreePath);
@@ -181,9 +215,25 @@ describe("createPaseoWorktreeInBackground", () => {
const { tempDir, repoDir } = createGitRepo();
cleanupPaths.push(tempDir);
writeFileSync(path.join(repoDir, "paseo.json"), "{ invalid json\n");
execSync("git add paseo.json", { cwd: repoDir, stdio: "pipe" });
execSync("git -c commit.gpgsign=false commit -m 'broken config'", {
cwd: repoDir,
stdio: "pipe",
});
const paseoHome = path.join(tempDir, ".paseo");
const worktreePath = await computeWorktreePath(repoDir, "broken-feature", paseoHome);
const createdWorktree = await createWorktree({
branchName: "broken-feature",
cwd: repoDir,
baseBranch: "main",
worktreeSlug: "broken-feature",
runSetup: false,
paseoHome,
});
const worktreePath = createdWorktree.worktreePath;
const emitted: SessionOutboundMessage[] = [];
const snapshots = new Map<string, unknown>();
const logger = createLogger();
const emitWorkspaceUpdateForCwd = vi.fn(async () => {});
const archiveWorkspaceRecord = vi.fn(async () => {});
@@ -192,6 +242,7 @@ describe("createPaseoWorktreeInBackground", () => {
{
paseoHome,
emitWorkspaceUpdateForCwd,
cacheWorkspaceSetupSnapshot: (workspaceId, snapshot) => snapshots.set(workspaceId, snapshot),
emit: (message) => emitted.push(message),
sessionLogger: logger,
terminalManager: null,
@@ -202,9 +253,11 @@ describe("createPaseoWorktreeInBackground", () => {
{
requestCwd: repoDir,
repoRoot: repoDir,
baseBranch: "does-not-exist",
slug: "broken-feature",
worktreePath,
worktree: {
branchName: "broken-feature",
worktreePath,
},
shouldBootstrap: true,
},
);
@@ -216,8 +269,12 @@ describe("createPaseoWorktreeInBackground", () => {
expect(progressMessages[0]?.payload.status).toBe("running");
expect(progressMessages[0]?.payload.error).toBeNull();
expect(progressMessages[1]?.payload.status).toBe("failed");
expect(progressMessages[1]?.payload.error).toContain("does-not-exist");
expect(progressMessages[1]?.payload.error).toContain("Failed to parse paseo.json");
expect(progressMessages[1]?.payload.detail.commands).toEqual([]);
expect(snapshots.get(worktreePath)).toMatchObject({
status: "failed",
error: expect.stringContaining("Failed to parse paseo.json"),
});
expect(archiveWorkspaceRecord).toHaveBeenCalledWith(worktreePath);
expect(emitWorkspaceUpdateForCwd).toHaveBeenCalledWith(worktreePath);
});
@@ -233,8 +290,17 @@ describe("createPaseoWorktreeInBackground", () => {
cleanupPaths.push(tempDir);
const paseoHome = path.join(tempDir, ".paseo");
const worktreePath = await computeWorktreePath(repoDir, "feature-running-setup", paseoHome);
const createdWorktree = await createWorktree({
branchName: "feature-running-setup",
cwd: repoDir,
baseBranch: "main",
worktreeSlug: "feature-running-setup",
runSetup: false,
paseoHome,
});
const worktreePath = createdWorktree.worktreePath;
const emitted: SessionOutboundMessage[] = [];
const snapshots = new Map<string, unknown>();
const logger = createLogger();
const emitWorkspaceUpdateForCwd = vi.fn(async () => {});
const archiveWorkspaceRecord = vi.fn(async () => {});
@@ -243,6 +309,7 @@ describe("createPaseoWorktreeInBackground", () => {
{
paseoHome,
emitWorkspaceUpdateForCwd,
cacheWorkspaceSetupSnapshot: (workspaceId, snapshot) => snapshots.set(workspaceId, snapshot),
emit: (message) => emitted.push(message),
sessionLogger: logger,
terminalManager: null,
@@ -253,9 +320,11 @@ describe("createPaseoWorktreeInBackground", () => {
{
requestCwd: repoDir,
repoRoot: repoDir,
baseBranch: "main",
slug: "feature-running-setup",
worktreePath,
worktree: {
branchName: "feature-running-setup",
worktreePath,
},
shouldBootstrap: true,
},
);
@@ -285,12 +354,13 @@ describe("createPaseoWorktreeInBackground", () => {
);
const setupOutputMessage = runningMessages.find((message) =>
message.payload.detail.log.includes("phase-one"),
message.payload.detail.commands[0]?.log.includes("phase-one"),
);
expect(setupOutputMessage?.payload.detail.log).toContain("phase-one");
expect(setupOutputMessage?.payload.detail.commands[0]).toMatchObject({
index: 1,
command: 'sh -c "printf \'phase-one\\\\n\'; sleep 0.1; printf \'phase-two\\\\n\'"',
log: expect.stringContaining("phase-one"),
status: "running",
});
@@ -308,9 +378,14 @@ describe("createPaseoWorktreeInBackground", () => {
expect(progressMessages.at(-1)?.payload.detail.commands[0]).toMatchObject({
index: 1,
command: 'sh -c "printf \'phase-one\\\\n\'; sleep 0.1; printf \'phase-two\\\\n\'"',
log: expect.stringContaining("phase-two"),
status: "completed",
exitCode: 0,
});
expect(snapshots.get(worktreePath)).toMatchObject({
status: "completed",
error: null,
});
});
test("emits completed when reusing an existing worktree without bootstrapping", async () => {
@@ -339,6 +414,7 @@ describe("createPaseoWorktreeInBackground", () => {
});
const emitted: SessionOutboundMessage[] = [];
const snapshots = new Map<string, unknown>();
const routeStore = new ServiceRouteStore();
const logger = createLogger();
const terminalManager = createTerminalManagerStub();
@@ -349,6 +425,7 @@ describe("createPaseoWorktreeInBackground", () => {
{
paseoHome,
emitWorkspaceUpdateForCwd,
cacheWorkspaceSetupSnapshot: (workspaceId, snapshot) => snapshots.set(workspaceId, snapshot),
emit: (message) => emitted.push(message),
sessionLogger: logger,
terminalManager: terminalManager.manager,
@@ -359,9 +436,11 @@ describe("createPaseoWorktreeInBackground", () => {
{
requestCwd: repoDir,
repoRoot: repoDir,
baseBranch: "main",
slug: "reused-worktree",
worktreePath: existingWorktree.worktreePath,
worktree: {
branchName: "reused-worktree",
worktreePath: existingWorktree.worktreePath,
},
shouldBootstrap: false,
},
);
@@ -387,12 +466,28 @@ describe("createPaseoWorktreeInBackground", () => {
commands: [],
},
});
expect(routeStore.listRoutes()).toEqual([]);
expect(terminalManager.terminals).toHaveLength(0);
expect(routeStore.listRoutes()).toEqual([
{
hostname: "reused-worktree.web.localhost",
port: expect.any(Number),
workspaceId: existingWorktree.worktreePath,
serviceName: "web",
},
]);
expect(terminalManager.terminals).toHaveLength(1);
expect(terminalManager.terminals[0]?.cwd).toBe(existingWorktree.worktreePath);
expect(terminalManager.terminals[0]?.name).toBe("web");
expect(terminalManager.terminals[0]?.env?.PORT).toEqual(expect.any(String));
expect(terminalManager.terminals[0]?.env?.PASEO_SERVICE_URL).toBeDefined();
expect(terminalManager.terminals[0]?.sent).toEqual(["npm run dev\r"]);
expect(
readFileSync(path.join(existingWorktree.worktreePath, "README.md"), "utf8"),
).toContain("hello");
expect(() => readFileSync(path.join(existingWorktree.worktreePath, "setup-ran.txt"), "utf8")).toThrow();
expect(snapshots.get(existingWorktree.worktreePath)).toMatchObject({
status: "completed",
error: null,
});
expect(archiveWorkspaceRecord).not.toHaveBeenCalled();
expect(emitWorkspaceUpdateForCwd).toHaveBeenCalledWith(existingWorktree.worktreePath);
});
@@ -410,8 +505,17 @@ describe("createPaseoWorktreeInBackground", () => {
cleanupPaths.push(tempDir);
const paseoHome = path.join(tempDir, ".paseo");
const worktreePath = await computeWorktreePath(repoDir, "feature-service-failure", paseoHome);
const createdWorktree = await createWorktree({
branchName: "feature-service-failure",
cwd: repoDir,
baseBranch: "main",
worktreeSlug: "feature-service-failure",
runSetup: false,
paseoHome,
});
const worktreePath = createdWorktree.worktreePath;
const emitted: SessionOutboundMessage[] = [];
const snapshots = new Map<string, unknown>();
const routeStore = new ServiceRouteStore();
const logger = createLogger();
const terminalManager = createTerminalManagerStub({
@@ -426,6 +530,7 @@ describe("createPaseoWorktreeInBackground", () => {
{
paseoHome,
emitWorkspaceUpdateForCwd,
cacheWorkspaceSetupSnapshot: (workspaceId, snapshot) => snapshots.set(workspaceId, snapshot),
emit: (message) => emitted.push(message),
sessionLogger: logger,
terminalManager: terminalManager.manager,
@@ -436,9 +541,11 @@ describe("createPaseoWorktreeInBackground", () => {
{
requestCwd: repoDir,
repoRoot: repoDir,
baseBranch: "main",
slug: "feature-service-failure",
worktreePath,
worktree: {
branchName: "feature-service-failure",
worktreePath,
},
shouldBootstrap: true,
},
);
@@ -452,15 +559,237 @@ describe("createPaseoWorktreeInBackground", () => {
expect(progressMessages[1]?.payload.status).toBe("completed");
expect(progressMessages[1]?.payload.error).toBeNull();
expect(emitted.some((message) => message.type === "workspace_setup_progress" && message.payload.status === "failed")).toBe(false);
expect(logger.warn).toHaveBeenCalledWith(
expect(logger.error).toHaveBeenCalledWith(
expect.objectContaining({
err: expect.any(Error),
cwd: repoDir,
repoRoot: repoDir,
worktreeSlug: "feature-service-failure",
worktreePath,
}),
"Failed to spawn worktree services after workspace setup completed",
);
expect(snapshots.get(worktreePath)).toMatchObject({
status: "completed",
error: null,
});
expect(archiveWorkspaceRecord).not.toHaveBeenCalled();
expect(emitWorkspaceUpdateForCwd).toHaveBeenCalledWith(worktreePath);
});
test("launches services in socket mode without requiring a daemon TCP port", async () => {
const { tempDir, repoDir } = createGitRepo({
paseoConfig: {
services: {
web: {
command: "npm run dev",
},
},
},
});
cleanupPaths.push(tempDir);
const paseoHome = path.join(tempDir, ".paseo");
const createdWorktree = await createWorktree({
branchName: "feature-socket-mode",
cwd: repoDir,
baseBranch: "main",
worktreeSlug: "feature-socket-mode",
runSetup: false,
paseoHome,
});
const worktreePath = createdWorktree.worktreePath;
const emitted: SessionOutboundMessage[] = [];
const snapshots = new Map<string, unknown>();
const routeStore = new ServiceRouteStore();
const logger = createLogger();
const terminalManager = createTerminalManagerStub();
const emitWorkspaceUpdateForCwd = vi.fn(async () => {});
const archiveWorkspaceRecord = vi.fn(async () => {});
await createPaseoWorktreeInBackground(
{
paseoHome,
emitWorkspaceUpdateForCwd,
cacheWorkspaceSetupSnapshot: (workspaceId, snapshot) => snapshots.set(workspaceId, snapshot),
emit: (message) => emitted.push(message),
sessionLogger: logger,
terminalManager: terminalManager.manager,
archiveWorkspaceRecord,
serviceRouteStore: routeStore,
daemonPort: null,
},
{
requestCwd: repoDir,
repoRoot: repoDir,
worktree: {
branchName: "feature-socket-mode",
worktreePath,
},
shouldBootstrap: true,
},
);
expect(routeStore.listRoutes()).toEqual([
{
hostname: "feature-socket-mode.web.localhost",
port: expect.any(Number),
workspaceId: worktreePath,
serviceName: "web",
},
]);
expect(terminalManager.terminals).toHaveLength(1);
expect(terminalManager.terminals[0]?.cwd).toBe(worktreePath);
expect(terminalManager.terminals[0]?.env?.PORT).toEqual(expect.any(String));
expect(terminalManager.terminals[0]?.env?.PASEO_SERVICE_URL).toBeUndefined();
expect(terminalManager.terminals[0]?.sent).toEqual(["npm run dev\r"]);
expect(snapshots.get(worktreePath)).toMatchObject({
status: "completed",
error: null,
});
expect(archiveWorkspaceRecord).not.toHaveBeenCalled();
expect(emitWorkspaceUpdateForCwd).toHaveBeenCalledWith(worktreePath);
});
test("returns the cached workspace setup snapshot for status requests", async () => {
const emitted: SessionOutboundMessage[] = [];
const snapshots = new Map([
[
"/repo/.paseo/worktrees/feature-a",
{
status: "completed",
detail: {
type: "worktree_setup",
worktreePath: "/repo/.paseo/worktrees/feature-a",
branchName: "feature-a",
log: "done",
commands: [],
},
error: null,
},
],
]);
await handleWorkspaceSetupStatusRequest(
{
emit: (message) => emitted.push(message),
workspaceSetupSnapshots: snapshots,
},
{
type: "workspace_setup_status_request",
workspaceId: "/repo/.paseo/worktrees/feature-a",
requestId: "req-status",
},
);
expect(emitted).toContainEqual({
type: "workspace_setup_status_response",
payload: {
requestId: "req-status",
workspaceId: "/repo/.paseo/worktrees/feature-a",
snapshot: {
status: "completed",
detail: {
type: "worktree_setup",
worktreePath: "/repo/.paseo/worktrees/feature-a",
branchName: "feature-a",
log: "done",
commands: [],
},
error: null,
},
},
});
});
test("returns null when no cached workspace setup snapshot exists", async () => {
const emitted: SessionOutboundMessage[] = [];
await handleWorkspaceSetupStatusRequest(
{
emit: (message) => emitted.push(message),
workspaceSetupSnapshots: new Map(),
},
{
type: "workspace_setup_status_request",
workspaceId: "/repo/.paseo/worktrees/missing",
requestId: "req-missing",
},
);
expect(emitted).toContainEqual({
type: "workspace_setup_status_response",
payload: {
requestId: "req-missing",
workspaceId: "/repo/.paseo/worktrees/missing",
snapshot: null,
},
});
});
});
describe("handleCreatePaseoWorktreeRequest", () => {
test("creates the worktree before emitting the response", async () => {
const { tempDir, repoDir } = createGitRepo();
const paseoHome = path.join(tempDir, ".paseo");
const emitted: SessionOutboundMessage[] = [];
const backgroundWork = vi.fn(async () => {});
try {
await handleCreatePaseoWorktreeRequest(
{
paseoHome,
sessionLogger: createLogger(),
emit: (message) => emitted.push(message),
registerPendingWorktreeWorkspace: vi.fn(async (options) => {
expect(existsSync(options.worktreePath)).toBe(true);
return {
workspaceId: options.worktreePath,
projectId: options.repoRoot,
} as any;
}),
describeWorkspaceRecord: vi.fn(async (workspace) => ({
id: workspace.workspaceId,
projectId: workspace.projectId,
projectDisplayName: path.basename(repoDir),
projectRootPath: repoDir,
projectKind: "git",
workspaceKind: "worktree",
name: path.basename(workspace.workspaceId),
status: "done",
activityAt: null,
})),
createPaseoWorktreeInBackground: backgroundWork,
},
{
type: "create_paseo_worktree_request",
cwd: repoDir,
worktreeSlug: "response-after-create",
requestId: "req-1",
},
);
const response = emitted.find(
(message): message is Extract<SessionOutboundMessage, { type: "create_paseo_worktree_response" }> =>
message.type === "create_paseo_worktree_response",
);
expect(response?.payload.error).toBeNull();
expect(response?.payload.workspace?.id).toBeTruthy();
expect(existsSync(response!.payload.workspace!.id)).toBe(true);
expect(backgroundWork).toHaveBeenCalledWith(
expect.objectContaining({
requestCwd: repoDir,
repoRoot: repoDir,
worktree: {
branchName: "response-after-create",
worktreePath: response!.payload.workspace!.id,
},
shouldBootstrap: true,
}),
);
} finally {
rmSync(tempDir, { recursive: true, force: true });
}
});
});
+76 -49
View File
@@ -11,6 +11,7 @@ import {
type ProjectPlacementPayload,
type SessionInboundMessage,
type SessionOutboundMessage,
type WorkspaceSetupSnapshot,
type WorkspaceDescriptorPayload,
} from "./messages.js";
import type {
@@ -27,7 +28,6 @@ import {
createWorktreeSetupProgressAccumulator,
getWorktreeSetupProgressResults,
spawnWorktreeServices,
type CreateAgentWorktreeResult,
} from "./worktree-bootstrap.js";
import type { TerminalManager } from "../terminal/terminal-manager.js";
import type { ServiceRouteStore } from "./service-proxy.js";
@@ -117,12 +117,18 @@ type CreatePaseoWorktreeInBackgroundDependencies = {
cwd: string,
options?: { dedupeGitState?: boolean },
) => Promise<void>;
cacheWorkspaceSetupSnapshot: (workspaceId: string, snapshot: WorkspaceSetupSnapshot) => void;
emit: EmitSessionMessage;
sessionLogger: Logger;
terminalManager: TerminalManager | null;
archiveWorkspaceRecord: (workspaceId: string) => Promise<void>;
serviceRouteStore: ServiceRouteStore | null;
daemonPort: number | null;
daemonPort?: number | null;
};
type HandleWorkspaceSetupStatusRequestDependencies = {
emit: EmitSessionMessage;
workspaceSetupSnapshots: ReadonlyMap<string, WorkspaceSetupSnapshot>;
};
type HandleCreatePaseoWorktreeRequestDependencies = {
@@ -597,7 +603,7 @@ export async function handleCreatePaseoWorktreeRequest(
throw new Error(`Invalid worktree name: ${validation.error}`);
}
const worktreePath = await computeWorktreePath(repoRoot, normalizedSlug, dependencies.paseoHome);
await computeWorktreePath(repoRoot, normalizedSlug, dependencies.paseoHome);
const createdWorktree = await createAgentWorktree({
cwd: repoRoot,
branchName: normalizedSlug,
@@ -645,6 +651,21 @@ export async function handleCreatePaseoWorktreeRequest(
}
}
export async function handleWorkspaceSetupStatusRequest(
dependencies: HandleWorkspaceSetupStatusRequestDependencies,
request: Extract<SessionInboundMessage, { type: "workspace_setup_status_request" }>,
): Promise<void> {
const workspaceId = normalizePersistedWorkspaceId(request.workspaceId);
dependencies.emit({
type: "workspace_setup_status_response",
payload: {
requestId: request.requestId,
workspaceId,
snapshot: dependencies.workspaceSetupSnapshots.get(workspaceId) ?? null,
},
});
}
export async function createPaseoWorktreeInBackground(
dependencies: CreatePaseoWorktreeInBackgroundDependencies,
options: {
@@ -658,20 +679,25 @@ export async function createPaseoWorktreeInBackground(
let setupResults: WorktreeSetupCommandResult[] = [];
let setupStarted = false;
const progressAccumulator = createWorktreeSetupProgressAccumulator();
const workspaceId = normalizePersistedWorkspaceId(worktree.worktreePath);
const emitSetupProgress = (status: "running" | "completed" | "failed", error: string | null) => {
const snapshot: WorkspaceSetupSnapshot = {
status,
detail: buildWorktreeSetupDetail({
worktree,
results:
status === "running" ? getWorktreeSetupProgressResults(progressAccumulator) : setupResults,
outputAccumulatorsByIndex: progressAccumulator.outputAccumulatorsByIndex,
}),
error,
};
dependencies.cacheWorkspaceSetupSnapshot(workspaceId, snapshot);
dependencies.emit({
type: "workspace_setup_progress",
payload: {
workspaceId: normalizePersistedWorkspaceId(worktree.worktreePath),
status,
detail: buildWorktreeSetupDetail({
worktree,
results:
status === "running" ? getWorktreeSetupProgressResults(progressAccumulator) : setupResults,
outputAccumulatorsByIndex: progressAccumulator.outputAccumulatorsByIndex,
}),
error,
workspaceId,
...snapshot,
},
});
};
@@ -682,36 +708,35 @@ export async function createPaseoWorktreeInBackground(
if (!options.shouldBootstrap) {
emitSetupProgress("completed", null);
return;
}
const setupCommands = getWorktreeSetupCommands(worktree.worktreePath);
if (setupCommands.length === 0) {
setupStarted = true;
emitSetupProgress("completed", null);
} else {
const runtimeEnv = await resolveWorktreeRuntimeEnv({
worktreePath: worktree.worktreePath,
branchName: worktree.branchName,
repoRootPath: options.repoRoot,
});
dependencies.terminalManager?.registerCwdEnv({
cwd: worktree.worktreePath,
env: runtimeEnv,
});
setupStarted = true;
setupResults = await runWorktreeSetupCommands({
worktreePath: worktree.worktreePath,
branchName: worktree.branchName,
cleanupOnFailure: false,
repoRootPath: options.repoRoot,
runtimeEnv,
onEvent: (event) => {
applyWorktreeSetupProgressEvent(progressAccumulator, event);
emitSetupProgress("running", null);
},
});
emitSetupProgress("completed", null);
const setupCommands = getWorktreeSetupCommands(worktree.worktreePath);
if (setupCommands.length === 0) {
setupStarted = true;
emitSetupProgress("completed", null);
} else {
const runtimeEnv = await resolveWorktreeRuntimeEnv({
worktreePath: worktree.worktreePath,
branchName: worktree.branchName,
repoRootPath: options.repoRoot,
});
dependencies.terminalManager?.registerCwdEnv({
cwd: worktree.worktreePath,
env: runtimeEnv,
});
setupStarted = true;
setupResults = await runWorktreeSetupCommands({
worktreePath: worktree.worktreePath,
branchName: worktree.branchName,
cleanupOnFailure: false,
repoRootPath: options.repoRoot,
runtimeEnv,
onEvent: (event) => {
applyWorktreeSetupProgressEvent(progressAccumulator, event);
emitSetupProgress("running", null);
},
});
emitSetupProgress("completed", null);
}
}
} catch (error) {
if (error instanceof WorktreeSetupError) {
@@ -738,18 +763,14 @@ export async function createPaseoWorktreeInBackground(
return;
}
if (
!dependencies.terminalManager ||
!dependencies.serviceRouteStore ||
dependencies.daemonPort === null ||
dependencies.daemonPort === undefined
) {
if (!dependencies.terminalManager || !dependencies.serviceRouteStore) {
return;
}
try {
await spawnWorktreeServices({
repoRoot: worktree.worktreePath,
workspaceId: worktree.worktreePath,
branchName: worktree.branchName,
daemonPort: dependencies.daemonPort,
routeStore: dependencies.serviceRouteStore,
@@ -757,8 +778,14 @@ export async function createPaseoWorktreeInBackground(
logger: dependencies.sessionLogger,
});
} catch (error) {
dependencies.sessionLogger.warn(
{ err: error, worktreePath: worktree.worktreePath },
dependencies.sessionLogger.error(
{
err: error,
cwd: options.requestCwd,
repoRoot: options.repoRoot,
worktreeSlug: worktree.branchName,
worktreePath: worktree.worktreePath,
},
"Failed to spawn worktree services after workspace setup completed",
);
}
+49
View File
@@ -199,6 +199,7 @@ const WorktreeSetupCommandSnapshotSchema = z.object({
index: z.number().int().positive(),
command: z.string(),
cwd: z.string(),
log: z.string(),
status: z.enum(["running", "completed", "failed"]),
exitCode: z.number().nullable(),
durationMs: z.number().nonnegative().optional(),
@@ -985,6 +986,12 @@ export const CreatePaseoWorktreeRequestSchema = z.object({
requestId: z.string(),
});
export const WorkspaceSetupStatusRequestSchema = z.object({
type: z.literal("workspace_setup_status_request"),
workspaceId: z.string(),
requestId: z.string(),
});
export const OpenProjectRequestSchema = z.object({
type: z.literal("open_project_request"),
cwd: z.string(),
@@ -1228,6 +1235,7 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [
PaseoWorktreeListRequestSchema,
PaseoWorktreeArchiveRequestSchema,
CreatePaseoWorktreeRequestSchema,
WorkspaceSetupStatusRequestSchema,
OpenProjectRequestSchema,
ArchiveWorkspaceRequestSchema,
FileExplorerRequestSchema,
@@ -1573,6 +1581,14 @@ export const ProjectPlacementPayloadSchema = z.object({
checkout: ProjectCheckoutLitePayloadSchema,
});
export const WorkspaceServicePayloadSchema = z.object({
serviceName: z.string(),
hostname: z.string(),
port: z.number().int().positive(),
url: z.string().nullable(),
status: z.enum(["running", "stopped"]),
});
export const WorkspaceDescriptorPayloadSchema = z.object({
id: z.string(),
projectId: z.string(),
@@ -1590,6 +1606,7 @@ export const WorkspaceDescriptorPayloadSchema = z.object({
})
.nullable()
.optional(),
services: z.array(WorkspaceServicePayloadSchema).default([]),
});
export const AgentUpdateMessageSchema = z.object({
@@ -1682,6 +1699,14 @@ export const WorkspaceUpdateMessageSchema = z.object({
]),
});
export const ServiceStatusUpdateMessageSchema = z.object({
type: z.literal("service_status_update"),
payload: z.object({
workspaceId: z.string(),
services: z.array(WorkspaceServicePayloadSchema),
}),
});
export const WorkspaceSetupProgressMessageSchema = z.object({
type: z.literal("workspace_setup_progress"),
payload: z.object({
@@ -1692,6 +1717,21 @@ export const WorkspaceSetupProgressMessageSchema = z.object({
}),
});
export const WorkspaceSetupSnapshotSchema = z.object({
status: z.enum(["running", "completed", "failed"]),
detail: WorktreeSetupDetailPayloadSchema,
error: z.string().nullable(),
});
export const WorkspaceSetupStatusResponseMessageSchema = z.object({
type: z.literal("workspace_setup_status_response"),
payload: z.object({
requestId: z.string(),
workspaceId: z.string(),
snapshot: WorkspaceSetupSnapshotSchema.nullable(),
}),
});
export const OpenProjectResponseMessageSchema = z.object({
type: z.literal("open_project_response"),
payload: z.object({
@@ -2263,7 +2303,9 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [
ArtifactMessageSchema,
AgentUpdateMessageSchema,
WorkspaceUpdateMessageSchema,
ServiceStatusUpdateMessageSchema,
WorkspaceSetupProgressMessageSchema,
WorkspaceSetupStatusResponseMessageSchema,
AgentStreamMessageSchema,
AgentStatusMessageSchema,
FetchAgentsResponseMessageSchema,
@@ -2348,14 +2390,20 @@ export type RpcErrorMessage = z.infer<typeof RpcErrorMessageSchema>;
export type ArtifactMessage = z.infer<typeof ArtifactMessageSchema>;
export type AgentUpdateMessage = z.infer<typeof AgentUpdateMessageSchema>;
export type WorkspaceSetupProgressMessage = z.infer<typeof WorkspaceSetupProgressMessageSchema>;
export type WorkspaceSetupSnapshot = z.infer<typeof WorkspaceSetupSnapshotSchema>;
export type WorkspaceSetupStatusResponseMessage = z.infer<
typeof WorkspaceSetupStatusResponseMessageSchema
>;
export type AgentStreamMessage = z.infer<typeof AgentStreamMessageSchema>;
export type AgentStatusMessage = z.infer<typeof AgentStatusMessageSchema>;
export type ProjectCheckoutLitePayload = z.infer<typeof ProjectCheckoutLitePayloadSchema>;
export type ProjectPlacementPayload = z.infer<typeof ProjectPlacementPayloadSchema>;
export type WorkspaceStateBucket = z.infer<typeof WorkspaceStateBucketSchema>;
export type WorkspaceDescriptorPayload = z.infer<typeof WorkspaceDescriptorPayloadSchema>;
export type WorkspaceServicePayload = z.infer<typeof WorkspaceServicePayloadSchema>;
export type FetchAgentsResponseMessage = z.infer<typeof FetchAgentsResponseMessageSchema>;
export type FetchWorkspacesResponseMessage = z.infer<typeof FetchWorkspacesResponseMessageSchema>;
export type ServiceStatusUpdateMessage = z.infer<typeof ServiceStatusUpdateMessageSchema>;
export type OpenProjectResponseMessage = z.infer<typeof OpenProjectResponseMessageSchema>;
export type ArchiveWorkspaceResponseMessage = z.infer<typeof ArchiveWorkspaceResponseMessageSchema>;
export type FetchAgentResponseMessage = z.infer<typeof FetchAgentResponseMessageSchema>;
@@ -2468,6 +2516,7 @@ export type PaseoWorktreeListRequest = z.infer<typeof PaseoWorktreeListRequestSc
export type PaseoWorktreeListResponse = z.infer<typeof PaseoWorktreeListResponseSchema>;
export type PaseoWorktreeArchiveRequest = z.infer<typeof PaseoWorktreeArchiveRequestSchema>;
export type PaseoWorktreeArchiveResponse = z.infer<typeof PaseoWorktreeArchiveResponseSchema>;
export type WorkspaceSetupStatusRequest = z.infer<typeof WorkspaceSetupStatusRequestSchema>;
export type OpenProjectRequest = z.infer<typeof OpenProjectRequestSchema>;
export type ArchiveWorkspaceRequest = z.infer<typeof ArchiveWorkspaceRequestSchema>;
export type FileExplorerRequest = z.infer<typeof FileExplorerRequestSchema>;
@@ -44,6 +44,7 @@ describe("workspace message schemas", () => {
name: "",
status: "not-a-bucket",
activityAt: null,
services: [],
},
},
});
@@ -51,6 +52,70 @@ describe("workspace message schemas", () => {
expect(result.success).toBe(false);
});
test("parses workspace descriptors with services", () => {
const parsed = SessionOutboundMessageSchema.parse({
type: "workspace_update",
payload: {
kind: "upsert",
workspace: {
id: "/repo",
projectId: "/repo",
projectDisplayName: "repo",
projectRootPath: "/repo",
projectKind: "non_git",
workspaceKind: "directory",
name: "repo",
status: "done",
activityAt: null,
services: [
{
serviceName: "web",
hostname: "web.localhost",
port: 3000,
url: "http://web.localhost:6767",
status: "running",
},
],
},
},
});
expect(parsed.type).toBe("workspace_update");
if (parsed.type !== "workspace_update" || parsed.payload.kind !== "upsert") {
throw new Error("Expected workspace_update upsert payload");
}
expect(parsed.payload.workspace.services).toEqual([
{
serviceName: "web",
hostname: "web.localhost",
port: 3000,
url: "http://web.localhost:6767",
status: "running",
},
]);
});
test("parses service_status_update payload", () => {
const parsed = SessionOutboundMessageSchema.parse({
type: "service_status_update",
payload: {
workspaceId: "/repo",
services: [
{
serviceName: "web",
hostname: "web.localhost",
port: 3000,
url: null,
status: "stopped",
},
],
},
});
expect(parsed.type).toBe("service_status_update");
expect(parsed.payload.workspaceId).toBe("/repo");
});
test("parses workspace_setup_progress payload", () => {
const parsed = SessionOutboundMessageSchema.parse({
type: "workspace_setup_progress",
@@ -67,6 +132,7 @@ describe("workspace message schemas", () => {
index: 1,
command: "npm install",
cwd: "/repo/.paseo/worktrees/feature-a",
log: "done",
status: "completed",
exitCode: 0,
durationMs: 100,
@@ -79,4 +145,37 @@ describe("workspace message schemas", () => {
expect(parsed.type).toBe("workspace_setup_progress");
});
test("parses workspace_setup_status_request", () => {
const parsed = SessionInboundMessageSchema.parse({
type: "workspace_setup_status_request",
workspaceId: "/repo/.paseo/worktrees/feature-a",
requestId: "req-status",
});
expect(parsed.type).toBe("workspace_setup_status_request");
});
test("parses workspace_setup_status_response payload", () => {
const parsed = SessionOutboundMessageSchema.parse({
type: "workspace_setup_status_response",
payload: {
requestId: "req-status",
workspaceId: "/repo/.paseo/worktrees/feature-a",
snapshot: {
status: "completed",
detail: {
type: "worktree_setup",
worktreePath: "/repo/.paseo/worktrees/feature-a",
branchName: "feature-a",
log: "done",
commands: [],
},
error: null,
},
},
});
expect(parsed.type).toBe("workspace_setup_status_response");
});
});
@@ -79,6 +79,7 @@ describe("shared tool-call display mapping", () => {
index: 1,
command: "npm install",
cwd: "/tmp/repo/.paseo/worktrees/repo/branch",
log: "==> [1/1] Running: npm install\n",
status: "running",
exitCode: null,
},
+113 -35
View File
@@ -4,7 +4,9 @@ import { existsSync, mkdirSync, readFileSync, realpathSync, rmSync, statSync } f
import { join, basename, dirname, resolve, sep } from "path";
import net from "node:net";
import { createHash } from "node:crypto";
import * as pty from "node-pty";
import { createNameId } from "mnemonic-id";
import stripAnsi from "strip-ansi";
import {
normalizeBaseRefName,
readPaseoWorktreeMetadata,
@@ -13,6 +15,7 @@ import {
writePaseoWorktreeRuntimeMetadata,
} from "./worktree-metadata.js";
import { resolvePaseoHome } from "../server/paseo-home.js";
import { ensureNodePtySpawnHelperExecutableForCurrentPlatform } from "../terminal/terminal.js";
interface PaseoConfig {
worktree?: {
@@ -234,6 +237,56 @@ export function getServiceConfigs(repoRoot: string): Map<string, ServiceConfig>
return result;
}
export function processCarriageReturns(text: string): string {
if (!text.includes("\r")) {
return text;
}
const output: string[] = [];
let line: string[] = [];
let cursor = 0;
const flushLine = () => {
output.push(line.join(""));
line = [];
cursor = 0;
};
for (let index = 0; index < text.length; index += 1) {
const char = text[index];
if (char === "\r") {
if (text[index + 1] === "\n") {
flushLine();
output.push("\n");
index += 1;
continue;
}
cursor = 0;
continue;
}
if (char === "\n") {
flushLine();
output.push("\n");
continue;
}
if (cursor < line.length) {
line[cursor] = char;
} else {
line.push(char);
}
cursor += 1;
}
if (line.length > 0) {
output.push(line.join(""));
}
return output.join("");
}
async function execSetupCommand(
command: string,
options: { cwd: string; env: NodeJS.ProcessEnv },
@@ -279,6 +332,27 @@ async function execSetupCommandStreamed(options: {
const stderrChunks: string[] = [];
let settled = false;
const emitOutput = (stream: "stdout" | "stderr", chunk: string) => {
const text = stripAnsi(chunk);
if (!text) {
return;
}
if (stream === "stdout") {
stdoutChunks.push(text);
} else {
stderrChunks.push(text);
}
options.onEvent?.({
type: "output",
index: options.index,
total: options.total,
command: options.command,
cwd: options.cwd,
stream,
chunk: text,
});
};
const finish = (exitCode: number | null) => {
if (settled) {
return;
@@ -314,48 +388,52 @@ async function execSetupCommandStreamed(options: {
cwd: options.cwd,
});
const child = spawn("/bin/bash", ["-lc", options.command], {
cwd: options.cwd,
env: options.env,
stdio: ["ignore", "pipe", "pipe"],
});
child.stdout?.on("data", (chunk: Buffer | string) => {
const text = chunk.toString();
stdoutChunks.push(text);
options.onEvent?.({
type: "output",
index: options.index,
total: options.total,
command: options.command,
const spawnWithPipes = () => {
const child = spawn("/bin/bash", ["-lc", options.command], {
cwd: options.cwd,
stream: "stdout",
chunk: text,
env: options.env,
stdio: ["ignore", "pipe", "pipe"],
});
});
child.stderr?.on("data", (chunk: Buffer | string) => {
const text = chunk.toString();
stderrChunks.push(text);
options.onEvent?.({
type: "output",
index: options.index,
total: options.total,
command: options.command,
child.stdout?.on("data", (chunk: Buffer | string) => {
emitOutput("stdout", chunk.toString());
});
child.stderr?.on("data", (chunk: Buffer | string) => {
emitOutput("stderr", chunk.toString());
});
child.on("error", (error) => {
emitOutput("stderr", error instanceof Error ? error.message : String(error));
finish(null);
});
child.on("close", (code) => {
finish(typeof code === "number" ? code : null);
});
};
try {
ensureNodePtySpawnHelperExecutableForCurrentPlatform();
const terminal = pty.spawn("/bin/bash", ["-lc", options.command], {
cwd: options.cwd,
stream: "stderr",
chunk: text,
env: options.env,
name: "xterm-color",
cols: 120,
rows: 30,
});
});
child.on("error", (error) => {
stderrChunks.push(error instanceof Error ? error.message : String(error));
finish(null);
});
terminal.onData((data) => {
emitOutput("stdout", data);
});
child.on("close", (code) => {
finish(typeof code === "number" ? code : null);
});
terminal.onExit(({ exitCode }) => {
finish(typeof exitCode === "number" ? exitCode : null);
});
} catch (error) {
emitOutput("stderr", error instanceof Error ? error.message : String(error));
spawnWithPipes();
}
});
}