diff --git a/docs/API.md b/docs/API.md index fcae271..d9e4fb5 100644 --- a/docs/API.md +++ b/docs/API.md @@ -1,6 +1,6 @@ # HTTP API Reference -The Conductor HTTP API runs on port **4000** (configurable via `CONDUCTOR_PORT` env var when started with `bun run server`). It provides endpoints for configuration management, process control, env var maintenance, and log streaming. All mutating operations write to SQLite and record an audit log entry. CORS is scoped to localhost any port by default. +The Conductor HTTP API listens on `127.0.0.1` (loopback only), port **4000** (configurable via `CONDUCTOR_PORT` env var when started with `bun run server`). It provides endpoints for configuration management, process control, env var maintenance, and log streaming. All mutating operations write to SQLite and record an audit log entry. CORS is scoped to localhost any port by default. ## Base URL / Port diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 1bf647c..a9d8df9 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -40,11 +40,12 @@ conductor/ │ │ ├── manager.ts WorkspaceManager: open/close/switch, installs a session into the API deps │ │ └── recent.ts workspaces.json recent list (newest first, max 20) │ ├── mcp/ MCP endpoint at /mcp (see "MCP" below) -│ │ ├── server.ts checkMcpRequest loopback guard + registerMcp (stateless Streamable HTTP) +│ │ ├── server.ts registerMcp (stateless Streamable HTTP) │ │ ├── tools.ts TOOLS registry: 47 tools, defined per area in tools/*.ts │ │ ├── route.ts defineTool, R/D/M annotations, callRoute (dispatch via app.inject) │ │ └── endpoint-file.ts ~/.conductor/endpoint.json: written on listen, removed on shutdown │ ├── api.ts Fastify 5 HTTP server (~876 lines) +│ ├── local-guard.ts checkLocalRequest: loopback peer/Host/Origin guard for every route │ └── index.ts 15-line barrel export ├── packages/cli/src/ CLI commands (Commander v15) │ ├── config-context.ts CLI → config store bridge @@ -130,7 +131,7 @@ The broadcaster at `packages/core/src/logs/broadcaster.ts` uses a pub/sub patter `packages/core/src/mcp/` exposes the API to AI agents as MCP tools. Each tool is a `defineTool` entry that maps its arguments onto an existing route, and `callRoute` runs that route through `app.inject()` on the same Fastify instance. Validation, the workspace guard and the audit log therefore stay in one place, and a non-2xx response becomes an `isError` tool result. Two kinds of tool have their own logic. `process_wait` polls the process list, and `env_list`, `env_set` and `env_import` wrap `callRoute` to redact secret values (the HTTP routes return them as stored). Its contract is in [API.md](./API.md#mcp-endpoint). -Security: the server binds `0.0.0.0`, so `checkMcpRequest` runs before the transport sees any request. It rejects non-loopback peers and any `Host` or `Origin` that is not a loopback name, with 403. +Security: the server binds `127.0.0.1` only, and `checkLocalRequest` (`packages/core/src/local-guard.ts`) runs as an `onRequest` hook on every route, `/api/*` and `/mcp` alike. It rejects non-loopback peers and any `Host` or `Origin` that is not a loopback name, with 403, which stops DNS rebinding and cross-site POSTs that skip the CORS preflight. The transport is stateless. Each POST builds a new `McpServer` and `StreamableHTTPServerTransport` (no session ID, JSON responses) and closes them when the response ends. Nothing is shared between requests, so there is no session table to leak or to expire. @@ -171,7 +172,7 @@ profiles: ## Desktop App Architecture (packages/desktop-tauri) -`packages/desktop` (Electron) is retired — no source left, just stale build output; `packages/desktop-tauri` (Tauri 2) is the desktop shell. The core engine still compiles to the same platform-specific sidecar binary (`bun build --compile → packages/core/dist-bin/conductor-server`); `scripts/stage-sidecar.mjs` copies it into `src-tauri/binaries/` with the target-triple suffix Tauri's `externalBin` mechanism expects. The Rust host (`src-tauri/src/main.rs`) spawns that sidecar and sets `CONDUCTOR_UI_DIST` on it, pointing at the Vite-built `packages/ui/dist/` — same same-origin trick as before, just launched by Rust instead of Electron's main process. Auto-update goes through Tauri's updater plugin against GitHub Releases (`tauri.conf.json`'s `plugins.updater`, signed with a minisign key) instead of electron-builder. +`packages/desktop` (Electron) is retired — no source left, just stale build output; `packages/desktop-tauri` (Tauri 2) is the desktop shell. The core engine still compiles to the same platform-specific sidecar binary (`bun build --compile → packages/core/dist-bin/conductor-server`); `scripts/stage-sidecar.mjs` copies it into `src-tauri/binaries/` with the target-triple suffix Tauri's `externalBin` mechanism expects. The Rust host (`src-tauri/src/main.rs`) spawns that sidecar and sets `CONDUCTOR_UI_DIST` on it, pointing at the Vite-built `packages/ui/dist/` — same same-origin trick as before, just launched by Rust instead of Electron's main process. Updates go through Tauri's updater plugin against GitHub Releases (`tauri.conf.json`'s `plugins.updater`, signed with a minisign key) instead of electron-builder. Nothing installs on its own: the UI checks on load, shows a notice when a release is newer, and the user installs it (which restarts the app) from Environment → About. ## Monitoring diff --git a/docs/CONFIG.md b/docs/CONFIG.md index e3640db..d0331a4 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -73,7 +73,7 @@ Every field except `id`, `name`, and `run` has a default. **Commands live at the | `watch` | string[] | `[]` | Globs relative to `cwd` (e.g. `src/**`, `*.csproj`). A matching change restarts the command after 500ms of quiet, then restarts every **running** command that transitively depends on it. Changes during a restart collapse into one follow-up restart. Build/tool dirs (`node_modules`, `.git`, `dist`, `obj`, `target`, `.next`, `.turbo`, `coverage`, `__pycache__`, `.venv`) and lockfiles (`package-lock.json`, `bun.lock`, `yarn.lock`, `pnpm-lock.yaml`, `Cargo.lock`, …) are always ignored. A command you stopped stays stopped. | | `config_files` | (string \| ConfigFile)[] | `[]` | `.env` files kept in sync with the env before every start. See [Declared config files](#declared-config-files). | | `readonly` | boolean | `false` | Informational flag; not enforced by the engine. | -| `stop_signal` | string | `"SIGTERM"` | Signal sent during graceful shutdown. Also accepts `SIGINT`, `SIGHUP`, etc. | +| `stop_signal` | string | `"SIGTERM"` | Signal sent during graceful shutdown. Also accepts `SIGINT`, `SIGHUP`, etc. An unknown name (e.g. `TERM`) is a config error. | | `stop_timeout_ms` | number | `5000` | Time before force-kill (SIGKILL or Windows taskkill). | | `stop_command` | string | _none_ | A command to run **before** stop_signal; useful for Docker Compose cleanup. | | `restart` | `manual` \| `on_failure` \| `always` | `"manual"` | Auto-restart policy, keyed on process **exit code**. `on_failure` respawns only on a non-zero exit; `always` respawns any exit. A stop or restart you asked for never counts as a crash. Capped at 5 consecutive restarts with exponential backoff (1s → 30s); the budget resets after 60s of uptime or on a manual restart. | @@ -108,13 +108,13 @@ Any environment variable whose **name** appears in the top-level `env_secrets` a Choose one of five types per command: -| Type | Description | Required Fields | -| ------------------ | --------------------------------------------------------------------------- | ---------------------------- | -| `"none"` (default) | Consider the process healthy immediately after spawning. No waiting. | _none_ | -| `"port"` | Wait for a TCP port to accept connections. Socket timeout: 2 s. | `port` (number) | -| `"http"` | Wait for an HTTP endpoint to respond with status < 500. Fetch timeout: 2 s. | `url` (string, complete URL) | -| `"command"` | Execute a shell command and wait for exit code 0. | `command` (string) | -| `"log_line"` | Wait for a stdout/stderr line containing a substring. | `pattern` (string) | +| Type | Description | Required Fields | +| ------------------ | ------------------------------------------------------------------------------------------------------------ | ---------------------------- | +| `"none"` (default) | Consider the process healthy immediately after spawning. No waiting. | _none_ | +| `"port"` | Wait for a TCP port to accept connections. Socket timeout: 2 s. | `port` (number) | +| `"http"` | Wait for an HTTP endpoint to respond with status < 500. Fetch timeout: 2 s. | `url` (string, complete URL) | +| `"command"` | Execute a shell command and wait for exit code 0. Each run is killed after 10 s (or `timeout_ms`, if lower). | `command` (string) | +| `"log_line"` | Wait for a stdout/stderr line containing a substring. | `pattern` (string) | `log_line` polls the process's own output (already collected for logging) rather than the network or a subprocess: on each of the `retries` attempts, spaced `interval_ms` apart, it checks whether any line since startup has contained `pattern`. This suits a service that only announces readiness on stdout (e.g. a dev server printing `compiled successfully`) with no port or endpoint to probe yet. It's a one-shot signal — a line either appeared or it didn't, and it can't un-appear — so unlike the other types, it does not run in the continuous post-startup health monitor. diff --git a/packages/core/bin/server.ts b/packages/core/bin/server.ts index 02fd83b..884ce73 100644 --- a/packages/core/bin/server.ts +++ b/packages/core/bin/server.ts @@ -69,7 +69,9 @@ async function main() { const app = await buildApi(deps); - await app.listen({ port: PORT, host: "0.0.0.0" }); + // Loopback only: the API can create and run arbitrary shell commands and + // has no authentication, so it must never be reachable from the network. + await app.listen({ port: PORT, host: "127.0.0.1" }); // PORT 0 asks the OS for a free port; advertise the one actually bound. const address = app.server.address(); const boundPort = typeof address === "object" && address ? address.port : PORT; diff --git a/packages/core/src/api.ts b/packages/core/src/api.ts index 47592be..0330a73 100644 --- a/packages/core/src/api.ts +++ b/packages/core/src/api.ts @@ -16,6 +16,7 @@ import { listAvailableShells } from "./executor/shell"; import { parseDockerCompose } from "./docker-compose/parser"; import { WorkspaceBusyError, type WorkspaceManager } from "./workspace/manager"; import { registerMcp } from "./mcp/server"; +import { checkLocalRequest } from "./local-guard"; import { CommandInputSchema, CommandPatchSchema, @@ -104,6 +105,20 @@ export async function buildApi(deps: ApiDependencies): Promise origin: /^https?:\/\/localhost(:\d+)?$/, }); + // Callback style for the same Bun reason as the workspace guard below. + app.addHook("onRequest", (request, reply, done) => { + const reason = checkLocalRequest({ + remoteAddress: request.socket.remoteAddress, + host: request.headers.host, + origin: request.headers.origin, + }); + if (reason) { + reply.status(403).send({ error: "forbidden" }); + return; + } + done(); + }); + const workspaces = deps.workspaces; if (workspaces) { // Registered right after cors so preflights are answered and 409s still @@ -535,21 +550,15 @@ export async function buildApi(deps: ApiDependencies): Promise } try { const profileName = request.params.profile; - - // Add commands - if (parsed.data.add?.length) { - for (const cmdId of parsed.data.add) { - deps.store.addCommandToProfile(profileName, cmdId); - deps.queries.insertAuditEntry("sync-command-to-profile", `${profileName}/${cmdId}`); - } + const add = parsed.data.add ?? []; + const remove = parsed.data.remove ?? []; + // One write: a bad id rejects the whole batch instead of half of it. + deps.store.syncProfileCommands(profileName, add, remove); + for (const cmdId of add) { + deps.queries.insertAuditEntry("sync-command-to-profile", `${profileName}/${cmdId}`); } - - // Remove commands - if (parsed.data.remove?.length) { - for (const cmdId of parsed.data.remove) { - deps.store.removeCommandFromProfile(profileName, cmdId); - deps.queries.insertAuditEntry("sync-remove-command", `${profileName}/${cmdId}`); - } + for (const cmdId of remove) { + deps.queries.insertAuditEntry("sync-remove-command", `${profileName}/${cmdId}`); } return { success: true }; @@ -608,13 +617,11 @@ export async function buildApi(deps: ApiDependencies): Promise .send({ error: body.error.issues[0]?.message ?? "Invalid request" }); } try { - // Remove from source profile and add to target profile - deps.store.removeCommandFromProfile(request.params.profile, request.params.id); - deps.store.addCommandToProfile(body.data.targetProfile, request.params.id); const command = deps.store.getCommand(request.params.id); if (!command) { return reply.status(404).send({ error: `Command "${request.params.id}" not found` }); } + deps.store.moveCommand(request.params.profile, body.data.targetProfile, command.id); deps.queries.insertAuditEntry( "move-command", `${request.params.profile}/${request.params.id} -> ${body.data.targetProfile}`, @@ -1058,7 +1065,8 @@ export async function buildApi(deps: ApiDependencies): Promise }); // MCP endpoint for AI agents. Not under /api/, so the workspace guard does - // not apply to the endpoint itself (tools dispatch back through /api/*, where it does). + // not apply to the endpoint itself (tools dispatch back through /api/*, where + // it does); the loopback guard above covers it like every other route. // Registered before the static/SPA block so its not-found handler can't swallow it. await registerMcp(app); diff --git a/packages/core/src/config/schema.ts b/packages/core/src/config/schema.ts index 033554b..e24d0c8 100644 --- a/packages/core/src/config/schema.ts +++ b/packages/core/src/config/schema.ts @@ -1,3 +1,4 @@ +import { constants } from "node:os"; import { z } from "zod"; /** @@ -55,7 +56,15 @@ export const CommandSchema = z.object({ watch: z.array(z.string()).default([]), config_files: z.array(ConfigFileSchema).default([]), readonly: z.boolean().default(false), - stop_signal: z.string().default("SIGTERM"), + // Checked here: process.kill throws on an unknown name, and the stop path + // can't tell that apart from "already gone", so a typo silently skipped + // the graceful stop and went straight to SIGKILL after stop_timeout_ms. + stop_signal: z + .string() + .refine((name) => Object.hasOwn(constants.signals, name), { + message: "Unknown signal; use a name like SIGTERM or SIGINT", + }) + .default("SIGTERM"), stop_timeout_ms: z.number().int().positive().default(5_000), stop_command: z.string().min(1).optional(), restart: z.enum(["manual", "on_failure", "always"]).default("manual"), diff --git a/packages/core/src/config/store.ts b/packages/core/src/config/store.ts index ffe16ef..7c5b686 100644 --- a/packages/core/src/config/store.ts +++ b/packages/core/src/config/store.ts @@ -64,9 +64,9 @@ export class ConfigStore { return new SpawnQueue( "__global__", this.config.commands, - (cmd) => buildCommandEnv(this.commandEnvParams(cmd)), - (cmd) => ({ - declaredKeys: declaredEnvKeys(this.commandEnvParams(cmd)), + (cmd, profile) => buildCommandEnv(this.commandEnvParams(cmd, profile)), + (cmd, profile) => ({ + declaredKeys: declaredEnvKeys(this.commandEnvParams(cmd, profile)), secretKeys: this.config.env_secrets, }), ); @@ -83,9 +83,18 @@ export class ConfigStore { }; } - /** Env params for a command, in the context of the first profile that references it. */ - private commandEnvParams(cmd: CommandConfig): BuildCommandEnvParams { - return { ...this.envParams(firstProfileOf(this.config, cmd.id)), cmd }; + /** + * Env params for a command, in the context of the profile that launched it + * (so a command shared by dev and staging gets the env of the run it is + * part of), or of the first profile that references it when it was started + * outside a known profile. + */ + private commandEnvParams(cmd: CommandConfig, launchedBy?: string): BuildCommandEnvParams { + const scope = + launchedBy && this.config.profiles[launchedBy] + ? launchedBy + : firstProfileOf(this.config, cmd.id); + return { ...this.envParams(scope), cmd }; } /** Absolute directory that relative `cwd`s resolve against (see `base_path`). */ @@ -307,17 +316,17 @@ export class ConfigStore { } const nextProfiles: Record = { ...this.config.profiles }; + // Omitted keeps the current description; "" clears it. + const nextProfile = { + ...profile, + description: description === undefined ? profile.description : description || undefined, + }; if (newName && newName !== name) { - // Rename key delete nextProfiles[name]; - nextProfiles[newName] = { - ...profile, - description: description ?? profile.description, - }; + nextProfiles[newName] = nextProfile; } else { - // Description-only update - nextProfiles[name] = { ...profile, description }; + nextProfiles[name] = nextProfile; } const nextConfig: ConductorConfig = { @@ -486,6 +495,64 @@ export class ConfigStore { return this.config.profiles[profileName]; } + /** + * Applies membership changes to one profile in a single validated write: + * every `add` is checked (known command, not already referenced) before + * anything changes, so a bad id leaves the profile and the file untouched. + */ + syncProfileCommands(profileName: string, add: string[], remove: string[]): ProfileConfig { + const profile = this.config.profiles[profileName]; + if (!profile) { + throw new ConfigError(`Unknown profile "${profileName}"`); + } + for (const commandId of add) { + if (!this.config.commands.some((c) => c.id === commandId)) { + throw new ConfigError(`Unknown command "${commandId}"`); + } + if (profile.command_ids.includes(commandId)) { + throw new ConfigError(`Profile "${profileName}" already references command "${commandId}"`); + } + } + return this.setProfileCommands({ + [profileName]: [...profile.command_ids.filter((id) => !remove.includes(id)), ...add], + })[profileName]; + } + + /** + * Moves a command reference from one profile to another in a single + * write, so a failing target (unknown, or already referencing it) can't + * leave the command dropped from the source. + */ + moveCommand(sourceName: string, targetName: string, commandId: string): void { + const source = this.config.profiles[sourceName]; + const target = this.config.profiles[targetName]; + if (!source) throw new ConfigError(`Unknown profile "${sourceName}"`); + if (!target) throw new ConfigError(`Unknown profile "${targetName}"`); + if (!this.config.commands.some((c) => c.id === commandId)) { + throw new ConfigError(`Unknown command "${commandId}"`); + } + if (sourceName === targetName) return; + if (target.command_ids.includes(commandId)) { + throw new ConfigError(`Profile "${targetName}" already references command "${commandId}"`); + } + this.setProfileCommands({ + [sourceName]: source.command_ids.filter((id) => id !== commandId), + [targetName]: [...target.command_ids, commandId], + }); + } + + /** Replaces `command_ids` of the given profiles, validated and persisted once. */ + private setProfileCommands(changes: Record): Record { + const profiles = { ...this.config.profiles }; + for (const [name, commandIds] of Object.entries(changes)) { + profiles[name] = { ...profiles[name], command_ids: commandIds }; + } + this.config = validateConfig({ ...this.config, profiles }); + this.mutableQueue.setCommands(this.config.commands); + this.persist(); + return this.config.profiles; + } + /** * Removes a command reference from a profile. */ diff --git a/packages/core/src/db/init.ts b/packages/core/src/db/init.ts index 2b90c1e..05d45e2 100644 --- a/packages/core/src/db/init.ts +++ b/packages/core/src/db/init.ts @@ -36,6 +36,17 @@ export function openDatabase(filePath: string): Database { db.exec("ALTER TABLE logs ADD COLUMN session_id INTEGER;"); } + // DBs written before idx_env_vars_key hold one row per edit of a global + // var; keep the newest (the last value written) so the index can be built. + const hasEnvVars = db + .query("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'env_vars'") + .get(); + if (hasEnvVars) { + db.exec( + "DELETE FROM env_vars WHERE id NOT IN (SELECT MAX(id) FROM env_vars GROUP BY scope, profile, key);", + ); + } + const ftsExisted = (db.query("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'logs_fts'").get() as { 1: number; diff --git a/packages/core/src/db/queries.ts b/packages/core/src/db/queries.ts index 68f40ba..a96500d 100644 --- a/packages/core/src/db/queries.ts +++ b/packages/core/src/db/queries.ts @@ -329,7 +329,7 @@ export class ConductorQueries { .prepare( `INSERT INTO env_vars (scope, profile, key, value, is_secret, updated_at) VALUES ($scope, $profile, $key, $value, $isSecret, $updatedAt) - ON CONFLICT (scope, profile, key) + ON CONFLICT (scope, COALESCE(profile, ''), key) DO UPDATE SET value = $value, is_secret = $isSecret, updated_at = $updatedAt`, ) .run({ diff --git a/packages/core/src/db/schema.sql b/packages/core/src/db/schema.sql index 819d193..ce81cb8 100644 --- a/packages/core/src/db/schema.sql +++ b/packages/core/src/db/schema.sql @@ -101,3 +101,8 @@ CREATE TABLE IF NOT EXISTS env_vars ( UNIQUE (scope, profile, key) ); +-- The table's UNIQUE never fires for global vars: SQLite treats NULLs as +-- distinct, so `profile IS NULL` rows duplicated on every edit. COALESCE makes +-- NULL compare equal; upsertEnvVar conflicts on this index. +CREATE UNIQUE INDEX IF NOT EXISTS idx_env_vars_key ON env_vars(scope, COALESCE(profile, ''), key); + diff --git a/packages/core/src/env/masker.ts b/packages/core/src/env/masker.ts index ff6316f..f8e2c9b 100644 --- a/packages/core/src/env/masker.ts +++ b/packages/core/src/env/masker.ts @@ -45,22 +45,34 @@ export function mergeEnv( * reference env vars, e.g. a command's `cwd` or a healthcheck's * `url`/`command` (see `resolvePath`). */ +const VAR_RE = /\$\{([A-Z0-9_]+)\}/gi; + export function interpolateString(value: string, env: Record): string { - return value.replace(/\$\{([A-Z0-9_]+)\}/gi, (_match, name: string) => { + return value.replace(VAR_RE, (_match, name: string) => { return env[name] ?? process.env[name] ?? ""; }); } /** * Resolves `${VAR}` style references within env values using values from - * the same env map (and falling back to process.env). + * the same env map (and falling back to process.env). References are + * followed to their final value (`A=${B}`, `B=${C}`), and a reference back + * into a value being resolved - `PATH=${PATH}:/x` - reads process.env, so it + * extends the inherited value instead of looping. */ export function interpolateEnv(env: Record): Record { const resolved: Record = {}; + const resolving = new Set(); - for (const [key, value] of Object.entries(env)) { - resolved[key] = interpolateString(value, env); - } + const resolveKey = (key: string): string => { + if (Object.hasOwn(resolved, key)) return resolved[key]; + if (!Object.hasOwn(env, key) || resolving.has(key)) return process.env[key] ?? ""; + resolving.add(key); + resolved[key] = env[key].replace(VAR_RE, (_match, name: string) => resolveKey(name)); + resolving.delete(key); + return resolved[key]; + }; + for (const key of Object.keys(env)) resolveKey(key); return resolved; } diff --git a/packages/core/src/executor/healthcheck.ts b/packages/core/src/executor/healthcheck.ts index b173761..d97edf9 100644 --- a/packages/core/src/executor/healthcheck.ts +++ b/packages/core/src/executor/healthcheck.ts @@ -3,6 +3,10 @@ import { resolve as resolvePath, isAbsolute } from "node:path"; import type { HealthcheckConfig } from "../config/schema"; import { interpolateString } from "../env/masker"; import { resolveShell } from "./shell"; +import { killTree } from "./wrapper"; + +/** Ceiling for one `command` probe; the healthcheck's own timeout_ms caps it further. */ +const COMMAND_PROBE_TIMEOUT_MS = 10_000; export class HealthcheckError extends Error {} @@ -52,14 +56,17 @@ async function checkHttp(url: string): Promise { } /** - * Runs a shell command and considers exit code 0 as healthy. + * Runs a shell command and considers exit code 0 as healthy. A probe still + * running after `timeoutMs` is killed (its whole group: the shell's child is + * what hangs) and reported as "timeout". */ async function checkCommand( command: string, cwd: string | undefined, configuredShell: string | undefined, env: Record, -): Promise { + timeoutMs: number, +): Promise { try { const { bin, flag, verbatim } = resolveShell(configuredShell); const proc = Bun.spawn({ @@ -73,9 +80,20 @@ async function checkCommand( env, stdout: "ignore", stderr: "ignore", + // Own process group, so a timeout can kill the shell and its children. + detached: process.platform !== "win32", + }); + let timer: ReturnType | undefined; + const timedOut = new Promise<"timeout">((resolve) => { + timer = setTimeout(() => resolve("timeout"), timeoutMs); }); - const exitCode = await proc.exited; - return exitCode === 0; + const outcome = await Promise.race([proc.exited, timedOut]); + clearTimeout(timer); + if (outcome === "timeout") { + killTree(proc.pid, "SIGKILL"); + return "timeout"; + } + return outcome === 0; } catch { return false; } @@ -120,8 +138,16 @@ export async function probeOnce( const resolvedCmd = interpolateString(healthcheck.command, env); const cwd = env.BASE_PATH && isAbsolute(env.BASE_PATH) ? resolvePath(env.BASE_PATH) : undefined; - const ok = await checkCommand(resolvedCmd, cwd, env.CONDUCTOR_SHELL, env); - return { ok, latencyMs: Date.now() - start, detail: `command "${resolvedCmd}"` }; + const timeoutMs = Math.min(COMMAND_PROBE_TIMEOUT_MS, healthcheck.timeout_ms); + const outcome = await checkCommand(resolvedCmd, cwd, env.CONDUCTOR_SHELL, env, timeoutMs); + return { + ok: outcome === true, + latencyMs: Date.now() - start, + detail: + outcome === "timeout" + ? `command "${resolvedCmd}" timed out after ${timeoutMs}ms` + : `command "${resolvedCmd}"`, + }; } case "none": default: diff --git a/packages/core/src/executor/queue.ts b/packages/core/src/executor/queue.ts index 2545b90..39aecab 100644 --- a/packages/core/src/executor/queue.ts +++ b/packages/core/src/executor/queue.ts @@ -91,13 +91,17 @@ export class SpawnQueue { constructor( private readonly profile: string, private commands: CommandConfig[], - private readonly resolveEnv: (cmd: CommandConfig) => Record, + /** `profile` is the one that launched `cmd` (see launchProfile), if any. */ + private readonly resolveEnv: (cmd: CommandConfig, profile?: string) => Record, /** * Which env keys were declared in Conductor for `cmd`, and which are * secret - drives `config_files` auto mode. Without it only `set:` keys * are written. */ - private readonly configFileScope?: (cmd: CommandConfig) => { + private readonly configFileScope?: ( + cmd: CommandConfig, + profile?: string, + ) => { declaredKeys: Set; secretKeys?: string[]; }, @@ -118,7 +122,8 @@ export class SpawnQueue { /** * Checks if a dependency is ready: either currently running, - * or already ran to completion (exit code 0). + * or already ran to completion on its own (exit code 0). A dependency + * stopped on request is not ready even if it exited 0 on the signal. */ private isDependencyReady(depId: string): boolean { const wrapper = this.wrappers.get(depId); @@ -126,9 +131,7 @@ export class SpawnQueue { const snapshot = wrapper.getSnapshot(); if (!snapshot) return false; if (snapshot.status === "running") return true; - return ( - (snapshot.status === "completed" || snapshot.status === "stopped") && snapshot.exitCode === 0 - ); + return snapshot.status === "completed" && snapshot.exitCode === 0; } /** @@ -154,7 +157,7 @@ export class SpawnQueue { if (status === "running") return; const snapshot = wrapper.getSnapshot(); - if (snapshot?.exitCode === 0) return; + if (status === "completed" && snapshot?.exitCode === 0) return; if (status === "stopped" || status === "completed" || status === "failed") { const reason = `Blocked: dependency "${depId}" failed (exit code ${snapshot?.exitCode ?? "?"})`; @@ -392,7 +395,9 @@ export class SpawnQueue { ): Array<[string, "stdout" | "stderr"]> { const lines: Array<[string, "stdout" | "stderr"]> = []; if (cmd.config_files.length === 0) return lines; - const scope = this.configFileScope?.(cmd) ?? { declaredKeys: new Set() }; + const scope = this.configFileScope?.(cmd, this.launchProfile.get(cmd.id)) ?? { + declaredKeys: new Set(), + }; const cwd = wrapper.resolvedCwd(); for (const plan of applyConfigFiles(cmd, { env, cwd, ...scope })) { const file = relative(cwd, plan.path) || plan.path; @@ -809,7 +814,7 @@ export class SpawnQueue { await new Promise((r) => setTimeout(r, 50)); } - await this.startSingleProcess(cmd, this.resolveEnv(cmd), onLog); + await this.startSingleProcess(cmd, this.resolveEnv(cmd, this.launchProfile.get(cmd.id)), onLog); } /** @@ -836,9 +841,12 @@ export class SpawnQueue { if (!cmd) throw new Error(`Unknown command "${commandId}" in profile "${this.profile}"`); // If a start is already in flight for this command, let it finish - // first rather than racing it — our stop+restart runs after. - const inFlight = this.startPromises.get(commandId); - if (inFlight) await inFlight.catch(() => {}); + // first rather than racing it — our stop+restart runs after. Loop: other + // restarts waiting on the same promise resume in the same tick, and the + // first of them to register below must be waited on too, or two + // restarts run side by side and orphan one process. + let inFlight: Promise | undefined; + while ((inFlight = this.startPromises.get(commandId))) await inFlight.catch(() => {}); const promise = this.restartOneInner(commandId, cmd, onLog); this.startPromises.set(commandId, promise); @@ -869,7 +877,7 @@ export class SpawnQueue { this.pendingRecovery.add(commandId); } - await this.startSingleProcess(cmd, this.resolveEnv(cmd), onLog); + await this.startSingleProcess(cmd, this.resolveEnv(cmd, this.launchProfile.get(cmd.id)), onLog); } /** diff --git a/packages/core/src/executor/wrapper.ts b/packages/core/src/executor/wrapper.ts index 6c6fd09..876d771 100644 --- a/packages/core/src/executor/wrapper.ts +++ b/packages/core/src/executor/wrapper.ts @@ -74,7 +74,7 @@ const sleep = (ms: number): Promise => new Promise((r) => setTimeout(r, ms * `detached: true` (setsid) on POSIX, the leader is its own process-group leader, * so a negative-pid kill reaches every member. */ -function killTree(pid: number, signal: NodeJS.Signals): void { +export function killTree(pid: number, signal: NodeJS.Signals): void { if (process.platform === "win32") { const args = ["/pid", String(pid), "/T"]; if (signal === "SIGKILL") args.push("/F"); @@ -451,8 +451,11 @@ export class ProcessWrapper { "stderr", ); }); + // The pumps' own catch handlers log through the same handlers, which can + // throw too (e.g. a closed DB): swallow that so `drained` never rejects. this.drained = Promise.all([subprocess.exited.catch(() => {}), stdoutPump, stderrPump]).then( () => {}, + () => {}, ); subprocess.exited @@ -530,6 +533,9 @@ export class ProcessWrapper { */ async forceKillAndWait(): Promise { if (!this.process || this.process.subprocess == null) return; + // Before the kill, so exit handlers (crash notification, restart policy) + // see a teardown we asked for, not a crash. + this.intentionalStop = true; const alreadyTerminal = this.process.status === "failed" || this.process.status === "stopped" || diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 0dec6d1..3c97bd1 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -14,6 +14,7 @@ export * from "./logs/broadcaster"; export * from "./db/init"; export * from "./db/queries"; export * from "./api"; +export * from "./local-guard"; export * from "./monitor"; export * from "./workspace/recent"; export * from "./workspace/session"; diff --git a/packages/core/src/local-guard.ts b/packages/core/src/local-guard.ts new file mode 100644 index 0000000..17fcf4f --- /dev/null +++ b/packages/core/src/local-guard.ts @@ -0,0 +1,48 @@ +const LOOPBACK_ADDRESSES = new Set(["127.0.0.1", "::1", "::ffff:127.0.0.1"]); +const LOOPBACK_HOSTNAMES = new Set(["localhost", "127.0.0.1", "::1"]); + +// hostname (letters, digits, dots, dashes) or bracketed IPv6, optional :port +const HOST_RE = /^(\[[0-9a-f:.]+\]|[a-z0-9.-]+)(?::\d+)?$/i; + +function stripBrackets(hostname: string): string { + return hostname.startsWith("[") && hostname.endsWith("]") ? hostname.slice(1, -1) : hostname; +} + +/** + * Pure guard for every request. The API can run arbitrary shell commands, so + * only same-machine callers with a loopback `Host` (anti DNS-rebinding) and, + * if sent, a loopback `Origin` (anti cross-site browser requests, including + * bodyless POSTs that skip the CORS preflight) are allowed. Returns a reason, + * or null if OK. + */ +export function checkLocalRequest(input: { + remoteAddress: string | undefined; + host: string | undefined; + origin: string | undefined; +}): string | null { + const { remoteAddress, host, origin } = input; + if (!remoteAddress || !LOOPBACK_ADDRESSES.has(remoteAddress)) { + return "non-loopback remote address"; + } + + const hostMatch = host ? HOST_RE.exec(host) : null; + if (!hostMatch || !LOOPBACK_HOSTNAMES.has(stripBrackets(hostMatch[1]!.toLowerCase()))) { + return "non-loopback Host header"; + } + + if (origin !== undefined) { + let url: URL; + try { + url = new URL(origin); + } catch { + return "invalid Origin header"; + } + if ( + (url.protocol !== "http:" && url.protocol !== "https:") || + !LOOPBACK_HOSTNAMES.has(stripBrackets(url.hostname.toLowerCase())) + ) { + return "non-loopback Origin header"; + } + } + return null; +} diff --git a/packages/core/src/mcp/server.ts b/packages/core/src/mcp/server.ts index f092782..30b99ec 100644 --- a/packages/core/src/mcp/server.ts +++ b/packages/core/src/mcp/server.ts @@ -3,54 +3,6 @@ import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js"; import { callRoute, TOOLS } from "./tools"; -const LOOPBACK_ADDRESSES = new Set(["127.0.0.1", "::1", "::ffff:127.0.0.1"]); -const LOOPBACK_HOSTNAMES = new Set(["localhost", "127.0.0.1", "::1"]); - -// hostname (letters, digits, dots, dashes) or bracketed IPv6, optional :port -const HOST_RE = /^(\[[0-9a-f:.]+\]|[a-z0-9.-]+)(?::\d+)?$/i; - -function stripBrackets(hostname: string): string { - return hostname.startsWith("[") && hostname.endsWith("]") ? hostname.slice(1, -1) : hostname; -} - -/** - * Pure guard for `/mcp`. The server binds 0.0.0.0 and the tool surface can run - * arbitrary shell commands, so only same-machine callers with a loopback - * `Host` (anti DNS-rebinding) and, if sent, a loopback `Origin` (anti - * cross-site browser requests) are allowed. Returns a reason, or null if OK. - */ -export function checkMcpRequest(input: { - remoteAddress: string | undefined; - host: string | undefined; - origin: string | undefined; -}): string | null { - const { remoteAddress, host, origin } = input; - if (!remoteAddress || !LOOPBACK_ADDRESSES.has(remoteAddress)) { - return "non-loopback remote address"; - } - - const hostMatch = host ? HOST_RE.exec(host) : null; - if (!hostMatch || !LOOPBACK_HOSTNAMES.has(stripBrackets(hostMatch[1]!.toLowerCase()))) { - return "non-loopback Host header"; - } - - if (origin !== undefined) { - let url: URL; - try { - url = new URL(origin); - } catch { - return "invalid Origin header"; - } - if ( - (url.protocol !== "http:" && url.protocol !== "https:") || - !LOOPBACK_HOSTNAMES.has(stripBrackets(url.hostname.toLowerCase())) - ) { - return "non-loopback Origin header"; - } - } - return null; -} - function createMcpServer(app: FastifyInstance): McpServer { // ponytail: serverInfo version is fixed, not read from package.json const server = new McpServer({ name: "conductor", version: "0.1.0" }); @@ -84,13 +36,6 @@ export async function registerMcp(app: FastifyInstance): Promise { method: ["GET", "POST", "DELETE"], url: "/mcp", handler: async (request, reply) => { - const reason = checkMcpRequest({ - remoteAddress: request.socket.remoteAddress, - host: request.headers.host, - origin: request.headers.origin, - }); - if (reason) return reply.status(403).send({ error: "forbidden" }); - if (request.method !== "POST") { return reply.status(405).header("Allow", "POST").send(METHOD_NOT_ALLOWED); } diff --git a/packages/core/src/monitor/health-monitor.ts b/packages/core/src/monitor/health-monitor.ts index 64b4860..4f7f3e6 100644 --- a/packages/core/src/monitor/health-monitor.ts +++ b/packages/core/src/monitor/health-monitor.ts @@ -15,6 +15,8 @@ export type HealthFlipHandler = (isHealthy: boolean, detail: string) => void; export class HealthMonitor { private intervalId?: ReturnType; private lastHealthy: boolean | undefined = undefined; + /** A probe can outlast the interval; ticks during one are skipped, not stacked. */ + private probing = false; constructor( /** Command healthcheck config to repeatedly probe */ @@ -35,6 +37,8 @@ export class HealthMonitor { if (intervalMs < 1000 || !this.healthcheck || this.healthcheck.type === "log_line") return; this.intervalId = setInterval(async () => { + if (this.probing) return; + this.probing = true; try { const result = await probeOnce(this.healthcheck, this.getEnv()); @@ -46,6 +50,8 @@ export class HealthMonitor { } } catch { // probeOnce should always return. If it throws, keep previous state. + } finally { + this.probing = false; } }, intervalMs); } diff --git a/packages/core/src/workspace/session.ts b/packages/core/src/workspace/session.ts index 62dcfda..e77f89f 100644 --- a/packages/core/src/workspace/session.ts +++ b/packages/core/src/workspace/session.ts @@ -119,7 +119,12 @@ export function openSession(dir: string, opts: SessionOptions): WorkspaceSession // Every log line from any managed process is persisted and broadcast // so both the CLI (via `conductor logs`) and the UI's live SSE stream // can see it, regardless of who started the process. + // Set once the DB is closed. A process's last lines can still be in its pipe + // after stop() returns (waiting for them stalled every close on Windows), + // and they must not reach insertLog on a closed DB. + let dbClosed = false; const onLog: LogHandler = (entry: LogEntry) => { + if (dbClosed) return; const row = queries.insertLog({ process_id: String(entry.pid), command_id: entry.commandId, @@ -162,6 +167,7 @@ export function openSession(dir: string, opts: SessionOptions): WorkspaceSession } finally { collector.stop(); clearInterval(logRetentionInterval); + dbClosed = true; db.close(); } }, diff --git a/packages/core/test/config.test.ts b/packages/core/test/config.test.ts index f914033..1c1c71f 100644 --- a/packages/core/test/config.test.ts +++ b/packages/core/test/config.test.ts @@ -23,6 +23,17 @@ describe("validateConfig", () => { expect(config.commands[0]?.shell).toBe(true); }); + test("rejects an unknown stop_signal instead of silently skipping the graceful stop", () => { + const withSignal = (stop_signal: string) => + validateConfig({ + commands: [{ id: "a", name: "A", run: "true", stop_signal }], + profiles: { dev: { command_ids: ["a"] } }, + }); + expect(() => withSignal("TERM")).toThrow(/stop_signal: Unknown signal/); + expect(() => withSignal("SIGFOO")).toThrow(ConfigError); + expect(withSignal("SIGINT").commands[0]?.stop_signal).toBe("SIGINT"); + }); + test("rejects config missing required fields", () => { expect(() => validateConfig({ commands: [{}], profiles: { dev: { command_ids: [] } } }), diff --git a/packages/core/test/env-vars.test.ts b/packages/core/test/env-vars.test.ts new file mode 100644 index 0000000..33da837 --- /dev/null +++ b/packages/core/test/env-vars.test.ts @@ -0,0 +1,62 @@ +import { describe, test, expect } from "bun:test"; +import { Database } from "bun:sqlite"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { ConductorQueries, openDatabase } from "../src"; +import { removeTempDir } from "./helpers/remove-temp-dir"; + +describe("env var upsert", () => { + test("editing a global var updates it in place and returns the new value", () => { + const q = new ConductorQueries(openDatabase(":memory:")); + q.upsertEnvVar({ scope: "global", profile: null, key: "FOO", value: "1", isSecret: false }); + const row = q.upsertEnvVar({ + scope: "global", + profile: null, + key: "FOO", + value: "2", + isSecret: false, + }); + expect(row.value).toBe("2"); + expect(q.listEnvVars("global").map((r) => r.value)).toEqual(["2"]); + }); + + test("profile vars still upsert per profile", () => { + const q = new ConductorQueries(openDatabase(":memory:")); + for (const value of ["1", "2"]) { + q.upsertEnvVar({ scope: "profile", profile: "dev", key: "FOO", value, isSecret: false }); + } + q.upsertEnvVar({ scope: "profile", profile: "prod", key: "FOO", value: "p", isSecret: false }); + expect(q.listEnvVars("profile", "dev").map((r) => r.value)).toEqual(["2"]); + expect(q.listEnvVars("profile", "prod").map((r) => r.value)).toEqual(["p"]); + }); + + test("opening a DB with duplicated global vars keeps only the newest", async () => { + const dir = mkdtempSync(join(tmpdir(), "conductor-env-")); + const path = join(dir, "old.sqlite"); + try { + const old = new Database(path, { create: true }); + old.exec(`CREATE TABLE env_vars ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + scope TEXT NOT NULL CHECK (scope IN ('global', 'profile')), + profile TEXT, key TEXT NOT NULL, value TEXT NOT NULL, + is_secret INTEGER NOT NULL DEFAULT 0, updated_at TEXT NOT NULL, + UNIQUE (scope, profile, key))`); + const insert = old.prepare( + "INSERT INTO env_vars (scope, profile, key, value, updated_at) VALUES ('global', NULL, 'FOO', ?, 'x')", + ); + insert.run("old"); + insert.run("new"); + old.close(); + + const db = openDatabase(path); + try { + expect(new ConductorQueries(db).listEnvVars("global").map((r) => r.value)).toEqual(["new"]); + } finally { + db.close(); // Windows can't remove an open DB file + } + } finally { + await removeTempDir(dir); + } + }); +}); diff --git a/packages/core/test/healthcheck.test.ts b/packages/core/test/healthcheck.test.ts index aa166e6..64e41a7 100644 --- a/packages/core/test/healthcheck.test.ts +++ b/packages/core/test/healthcheck.test.ts @@ -161,6 +161,20 @@ describe("probeOnce - type: command", () => { expect(result.ok).toBe(false); }); + test.skipIf(process.platform === "win32")( + "gives up on a probe that hangs instead of waiting forever", + async () => { + const started = Date.now(); + const result = await probeOnce( + healthcheck({ type: "command", command: "sleep 30", timeout_ms: 300 }), + {}, + ); + expect(result.ok).toBe(false); + expect(result.detail).toContain("timed out"); + expect(Date.now() - started).toBeLessThan(3000); + }, + ); + test("interpolates ${VAR} in the configured command from env", async () => { const result = await probeOnce(healthcheck({ type: "command", command: "exit ${CODE}" }), { CODE: "0", diff --git a/packages/core/test/helpers/mcp-harness.ts b/packages/core/test/helpers/mcp-harness.ts index 5eba68d..8095777 100644 --- a/packages/core/test/helpers/mcp-harness.ts +++ b/packages/core/test/helpers/mcp-harness.ts @@ -1,4 +1,4 @@ -import { mkdirSync, mkdtempSync, rmSync } from "node:fs"; +import { mkdirSync, mkdtempSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import type { FastifyInstance } from "fastify"; @@ -12,6 +12,7 @@ import { WorkspaceManager, type ApiDependencies, } from "../../src"; +import { removeTempDir } from "./remove-temp-dir"; export interface McpHarness { root: string; @@ -90,17 +91,7 @@ export async function startMcpHarness(): Promise { for (const client of clients) await client.close().catch(() => {}); await manager.close().catch(() => {}); await app.close(); - // Windows lets go of a just-closed workspace's files a moment later (EBUSY), so retry briefly. - for (let attempt = 0; ; attempt++) { - try { - rmSync(root, { recursive: true, force: true }); - break; - } catch (err) { - const code = (err as NodeJS.ErrnoException).code; - if (attempt >= 30 || (code !== "EBUSY" && code !== "EPERM")) throw err; - await Bun.sleep(100); - } - } + await removeTempDir(root); }, }; } diff --git a/packages/core/test/helpers/remove-temp-dir.ts b/packages/core/test/helpers/remove-temp-dir.ts new file mode 100644 index 0000000..3c33fd9 --- /dev/null +++ b/packages/core/test/helpers/remove-temp-dir.ts @@ -0,0 +1,26 @@ +import { rmSync } from "node:fs"; + +/** + * Removes a test's temp dir. Windows lets go of a just-stopped child's cwd and + * a just-closed SQLite file a moment later (EBUSY/EPERM), and since Bun 1.4.4 + * that moment is often 1.5-2s, sometimes past the retry window. On Windows the + * dir is then left for the OS temp cleanup instead of failing the test: the + * tests check themselves that processes are gone, and a temp dir that outlives + * them is not a product bug. Anywhere else a failure still throws. + */ +export async function removeTempDir(dir: string): Promise { + for (let attempt = 0; ; attempt++) { + try { + rmSync(dir, { recursive: true, force: true }); + return; + } catch (err) { + const code = (err as NodeJS.ErrnoException).code; + if (code !== "EBUSY" && code !== "EPERM") throw err; + if (attempt >= 30) { + if (process.platform === "win32") return; + throw err; + } + await Bun.sleep(100); + } + } +} diff --git a/packages/core/test/local-guard.test.ts b/packages/core/test/local-guard.test.ts new file mode 100644 index 0000000..bf2e5e3 --- /dev/null +++ b/packages/core/test/local-guard.test.ts @@ -0,0 +1,46 @@ +import { describe, test, expect, beforeEach, afterEach } from "bun:test"; +import { startMcpHarness, type McpHarness } from "./helpers/mcp-harness"; + +// The loopback guard used to cover only /mcp, leaving the same command-running +// surface open on /api to cross-site pages and DNS rebinding. +describe("loopback guard on /api", () => { + let h: McpHarness; + + beforeEach(async () => { + h = await startMcpHarness(); + }); + + afterEach(async () => { + await h.stop(); + }); + + test("a cross-site bodyless POST is refused before the handler runs", async () => { + const res = await fetch(`${h.baseUrl}/api/workspaces/close`, { + method: "POST", + headers: { origin: "https://evil.example" }, + }); + expect(res.status).toBe(403); + }); + + test("a DNS-rebound Host is refused", async () => { + const res = await fetch(`${h.baseUrl}/api/workspaces`, { headers: { host: "evil.example" } }); + expect(res.status).toBe(403); + }); + + test("a non-loopback peer is refused", async () => { + const res = await h.app.inject({ + method: "GET", + url: "/api/health", + remoteAddress: "10.0.0.5", + }); + expect(res.statusCode).toBe(403); + }); + + test("same-machine requests, with or without a loopback Origin, pass", async () => { + expect((await fetch(`${h.baseUrl}/api/health`)).status).toBe(200); + const fromDevUi = await fetch(`${h.baseUrl}/api/health`, { + headers: { origin: "http://localhost:3000" }, + }); + expect(fromDevUi.status).toBe(200); + }); +}); diff --git a/packages/core/test/masker.test.ts b/packages/core/test/masker.test.ts index dcafd23..0de2fe7 100644 --- a/packages/core/test/masker.test.ts +++ b/packages/core/test/masker.test.ts @@ -21,6 +21,30 @@ describe("interpolateEnv", () => { const result = interpolateEnv({ HOST: "localhost", URL: "http://${HOST}:3000" }); expect(result.URL).toBe("http://localhost:3000"); }); + + test("follows chained references to their final value", () => { + const result = interpolateEnv({ + CALLBACK: "${API_URL}/cb", + API_URL: "http://${HOST}:3000", + HOST: "localhost", + }); + expect(result.CALLBACK).toBe("http://localhost:3000/cb"); + }); + + test("a self-reference extends the inherited value instead of looping", () => { + process.env.__MASKER_TEST_PATH__ = "/usr/bin"; + try { + const result = interpolateEnv({ __MASKER_TEST_PATH__: "${__MASKER_TEST_PATH__}:/opt/bin" }); + expect(result.__MASKER_TEST_PATH__).toBe("/usr/bin:/opt/bin"); + } finally { + delete process.env.__MASKER_TEST_PATH__; + } + }); + + test("a reference cycle ends instead of recursing forever", () => { + const result = interpolateEnv({ A: "a${B}", B: "b${A}" }); + expect(result.A).toBe("ab"); + }); }); describe("interpolateString", () => { diff --git a/packages/core/test/mcp.test.ts b/packages/core/test/mcp.test.ts index 3941473..e1e971e 100644 --- a/packages/core/test/mcp.test.ts +++ b/packages/core/test/mcp.test.ts @@ -1,39 +1,39 @@ import { describe, test, expect, beforeEach, afterEach } from "bun:test"; import Fastify from "fastify"; -import { checkMcpRequest, callRoute } from "../src"; +import { checkLocalRequest, callRoute } from "../src"; import { startMcpHarness, textOf, type McpHarness } from "./helpers/mcp-harness"; -describe("checkMcpRequest", () => { +describe("checkLocalRequest", () => { const ok = { remoteAddress: "127.0.0.1", host: "localhost:4000", origin: undefined }; test("accepts loopback IPv4, IPv6 and v4-mapped remote addresses", () => { for (const remoteAddress of ["127.0.0.1", "::1", "::ffff:127.0.0.1"]) { - expect(checkMcpRequest({ ...ok, remoteAddress })).toBeNull(); + expect(checkLocalRequest({ ...ok, remoteAddress })).toBeNull(); } }); test("rejects a non-loopback or missing remote address", () => { - expect(checkMcpRequest({ ...ok, remoteAddress: "10.0.0.5" })).not.toBeNull(); - expect(checkMcpRequest({ ...ok, remoteAddress: undefined })).not.toBeNull(); + expect(checkLocalRequest({ ...ok, remoteAddress: "10.0.0.5" })).not.toBeNull(); + expect(checkLocalRequest({ ...ok, remoteAddress: undefined })).not.toBeNull(); }); test("checks the Host header hostname", () => { - expect(checkMcpRequest({ ...ok, host: "evil.com" })).not.toBeNull(); - expect(checkMcpRequest({ ...ok, host: "evil.com:4000" })).not.toBeNull(); - expect(checkMcpRequest({ ...ok, host: "localhost@evil.com" })).not.toBeNull(); - expect(checkMcpRequest({ ...ok, host: undefined })).not.toBeNull(); + expect(checkLocalRequest({ ...ok, host: "evil.com" })).not.toBeNull(); + expect(checkLocalRequest({ ...ok, host: "evil.com:4000" })).not.toBeNull(); + expect(checkLocalRequest({ ...ok, host: "localhost@evil.com" })).not.toBeNull(); + expect(checkLocalRequest({ ...ok, host: undefined })).not.toBeNull(); for (const host of ["localhost:4000", "localhost", "127.0.0.1:4000", "[::1]:4000", "[::1]"]) { - expect(checkMcpRequest({ ...ok, host })).toBeNull(); + expect(checkLocalRequest({ ...ok, host })).toBeNull(); } }); test("checks the Origin header only when present", () => { - expect(checkMcpRequest({ ...ok, origin: "http://evil.com" })).not.toBeNull(); - expect(checkMcpRequest({ ...ok, origin: "null" })).not.toBeNull(); - expect(checkMcpRequest({ ...ok, origin: "file://localhost" })).not.toBeNull(); - expect(checkMcpRequest({ ...ok, origin: "http://localhost:3000" })).toBeNull(); - expect(checkMcpRequest({ ...ok, origin: "http://[::1]:3000" })).toBeNull(); - expect(checkMcpRequest(ok)).toBeNull(); + expect(checkLocalRequest({ ...ok, origin: "http://evil.com" })).not.toBeNull(); + expect(checkLocalRequest({ ...ok, origin: "null" })).not.toBeNull(); + expect(checkLocalRequest({ ...ok, origin: "file://localhost" })).not.toBeNull(); + expect(checkLocalRequest({ ...ok, origin: "http://localhost:3000" })).toBeNull(); + expect(checkLocalRequest({ ...ok, origin: "http://[::1]:3000" })).toBeNull(); + expect(checkLocalRequest(ok)).toBeNull(); }); }); diff --git a/packages/core/test/queue.test.ts b/packages/core/test/queue.test.ts index 08f8f55..2118c96 100644 --- a/packages/core/test/queue.test.ts +++ b/packages/core/test/queue.test.ts @@ -595,6 +595,47 @@ describe("SpawnQueue - restartOne races a concurrent start", () => { }); }); +describe("SpawnQueue - overlapping restarts", () => { + test("restarts queued behind the same start run one after another, never side by side", async () => { + // Two restarts that both awaited the same in-flight start used to resume + // together and each spawn a process: one wrapper got overwritten in + // `wrappers` and its process kept running untracked. + const dir = mkdtempSync(join(tmpdir(), "conductor-queue-restarts-")); + const logPath = join(dir, "spawns.log"); + try { + const cmd = makeCommand({ + id: "busy", + name: "Busy", + run: `bun -e "require('fs').appendFileSync(process.env.SPAWN_LOG, process.pid + '\\n'); setInterval(() => {}, 1000)"`, + }); + const queue = new SpawnQueue("test", [cmd], () => testEnv({ SPAWN_LOG: logPath })); + + await queue.startOne("busy"); + await Promise.all([ + queue.restartOne("busy"), + queue.restartOne("busy"), + queue.restartOne("busy"), + ]); + await new Promise((r) => setTimeout(r, 100)); // let a stray spawn's write land + + const pids = readFileSync(logPath, "utf-8").trim().split("\n").filter(Boolean).map(Number); + const alive = pids.filter((pid) => { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } + }); + expect(alive).toEqual([queue.getWrapper("busy")!.pid!]); + + await queue.stopAll(); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); +}); + describe("SpawnQueue.startMany", () => { test("only starts the given command ids, not every command in the queue", async () => { // Mirrors how the store's single global queue holds every command from @@ -797,6 +838,28 @@ describe("SpawnQueue - crash and unhealthy notifications", () => { } }, 10_000); + test("starting a command that is already running does not report a crash", async () => { + // The respawn force-kills the running process first (exit 137); that + // used to look like a crash and an auto-restart, then a recovery. + const cmd = makeCommand({ + id: "again", + name: "Again", + run: `bun -e "setInterval(() => {}, 1000)"`, + restart: "on_failure", + }); + const logs: string[] = []; + const queue = new SpawnQueue("test", [cmd], () => testEnv()); + try { + await queue.startOne("again", (e) => logs.push(e.message)); + await queue.startOne("again"); + await wait(1500); // past the first auto-restart's backoff + expect(queue.listNotifications().map((n) => n.type)).toEqual([]); + expect(logs.some((line) => line.startsWith("[restart]"))).toBe(false); + } finally { + await queue.stopAll(); + } + }, 10_000); + test("a crash during startup fails fast instead of exhausting healthcheck retries", async () => { const cmd = makeCommand({ id: "early", @@ -976,6 +1039,40 @@ describe("SpawnQueue - a new run after stopping everything", () => { }); }); +describe("SpawnQueue - a dependency stopped by hand", () => { + test("is started again, even if it exited 0 on the stop signal", async () => { + // A service that handles SIGTERM exits 0 and ends up "stopped" with exit + // code 0, which used to count as a satisfied dependency. + const db = makeCommand({ + id: "db", + name: "DB", + run: `bun -e "process.on('SIGTERM', () => process.exit(0)); setInterval(() => {}, 1000)"`, + }); + const api = makeCommand({ + id: "api", + name: "API", + run: `bun -e "setInterval(() => {}, 1000)"`, + deps: ["db"], + }); + const queue = new SpawnQueue("test", [db, api], () => testEnv()); + try { + await queue.startOne("api"); + await new Promise((r) => setTimeout(r, 300)); // let db install its handler + await queue.stopOne("db"); + const stopped = queue.getWrapper("db")?.getSnapshot(); + if (process.platform !== "win32") { + expect(stopped?.status).toBe("stopped"); + expect(stopped?.exitCode).toBe(0); + } + + await queue.startOne("api"); + expect(queue.getWrapper("db")?.status).toBe("running"); + } finally { + await queue.stopAll(); + } + }, 10_000); +}); + describe("SpawnQueue - starting a command mid-run", () => { test("reuses a one-shot dependency that already completed in this run", async () => { const dir = mkdtempSync(join(tmpdir(), "conductor-queue-midrun-")); diff --git a/packages/core/test/store.test.ts b/packages/core/test/store.test.ts index 20923da..fb44bfb 100644 --- a/packages/core/test/store.test.ts +++ b/packages/core/test/store.test.ts @@ -201,3 +201,109 @@ describe("ConfigStore command categories", () => { expect(readFileSync(configPath, "utf-8")).not.toContain("category:"); }); }); + +describe("ConfigStore process env", () => { + test("a command shared by two profiles gets the env of the profile that launched it", async () => { + const store = new ConfigStore( + configPath, + validateConfig({ + version: "1", + // shell: false + bun, so it reads the env the same way under cmd.exe. + commands: [ + { + id: "show", + name: "Show", + run: 'bun -e "process.stdout.write(process.env.TARGET)"', + shell: false, + }, + ], + profiles: { + dev: { env: { TARGET: "dev-db" }, command_ids: ["show"] }, + staging: { env: { TARGET: "staging-db" }, command_ids: ["show"] }, + }, + }), + ); + const queue = store.getQueue(); + const lines: string[] = []; + try { + await queue.startMany(["show"], (e) => lines.push(e.message), "staging"); + await queue.waitForExit(); + expect(lines).toContain("staging-db"); + expect(lines).not.toContain("dev-db"); + } finally { + await queue.close(); + } + }); +}); + +describe("ConfigStore profile membership writes are all-or-nothing", () => { + function twoProfiles() { + return new ConfigStore( + configPath, + validateConfig({ + version: "1", + commands: [ + { id: "api", name: "API", run: "true" }, + { id: "web", name: "Web", run: "true" }, + ], + profiles: { + dev: { command_ids: ["api"] }, + staging: { command_ids: ["api"] }, + }, + }), + ); + } + + test("a move the target rejects leaves the command in the source", () => { + const store = twoProfiles(); + expect(() => store.moveCommand("dev", "staging", "api")).toThrow(ConfigError); + expect(() => store.moveCommand("dev", "nope", "api")).toThrow(ConfigError); + expect(store.getConfig().profiles.dev.command_ids).toEqual(["api"]); + }); + + test("a valid move updates both profiles", () => { + const store = twoProfiles(); + store.addCommandToProfile("dev", "web"); + store.moveCommand("dev", "staging", "web"); + expect(store.getConfig().profiles.dev.command_ids).toEqual(["api"]); + expect(store.getConfig().profiles.staging.command_ids).toEqual(["api", "web"]); + expect(readFileSync(configPath, "utf-8")).toContain("web"); + }); + + test("a sync with one bad id changes nothing", () => { + const store = twoProfiles(); + expect(() => store.syncProfileCommands("dev", ["web", "ghost"], ["api"])).toThrow(ConfigError); + expect(store.getConfig().profiles.dev.command_ids).toEqual(["api"]); + store.syncProfileCommands("dev", ["web"], ["api"]); + expect(store.getConfig().profiles.dev.command_ids).toEqual(["web"]); + }); +}); + +describe("ConfigStore.updateProfile description", () => { + function withDescription() { + return new ConfigStore( + configPath, + validateConfig({ version: "1", profiles: { dev: { description: "local" } } }), + ); + } + + test('"" clears it', () => { + const store = withDescription(); + store.updateProfile("dev", { description: "" }); + expect(store.getConfig().profiles.dev.description).toBeUndefined(); + }); + + test("a rename without a description keeps it, even to the same name", () => { + const store = withDescription(); + store.updateProfile("dev", { newName: "dev" }); + expect(store.getConfig().profiles.dev.description).toBe("local"); + store.updateProfile("dev", { newName: "local-dev" }); + expect(store.getConfig().profiles["local-dev"].description).toBe("local"); + }); + + test("a rename can clear it at the same time", () => { + const store = withDescription(); + store.updateProfile("dev", { newName: "qa", description: "" }); + expect(store.getConfig().profiles.qa.description).toBeUndefined(); + }); +}); diff --git a/packages/core/test/workspace-manager.test.ts b/packages/core/test/workspace-manager.test.ts index 7113495..571dd8d 100644 --- a/packages/core/test/workspace-manager.test.ts +++ b/packages/core/test/workspace-manager.test.ts @@ -1,5 +1,5 @@ import { describe, test, expect, beforeEach, afterEach } from "bun:test"; -import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { mkdirSync, mkdtempSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import type { FastifyInstance } from "fastify"; @@ -16,6 +16,7 @@ import { type ApiDependencies, type CommandConfig, } from "../src"; +import { removeTempDir } from "./helpers/remove-temp-dir"; // Portable long-runner (see queue.test.ts): `bun -e`, shell: false, short // stop timeout so Windows' SIGKILL fallback still fits bun:test's 5s budget. @@ -78,17 +79,7 @@ beforeEach(async () => { afterEach(async () => { await manager.close().catch(() => {}); await app.close(); - // Windows lets go of a just-stopped child's cwd a moment after stop resolves (EBUSY), so retry briefly. - for (let attempt = 0; ; attempt++) { - try { - rmSync(root, { recursive: true, force: true }); - break; - } catch (err) { - const code = (err as NodeJS.ErrnoException).code; - if (attempt >= 30 || (code !== "EBUSY" && code !== "EPERM")) throw err; - await Bun.sleep(100); - } - } + await removeTempDir(root); }); describe("WorkspaceManager + API", () => { diff --git a/packages/core/test/workspace-session.test.ts b/packages/core/test/workspace-session.test.ts index 2e36142..179c9d6 100644 --- a/packages/core/test/workspace-session.test.ts +++ b/packages/core/test/workspace-session.test.ts @@ -110,6 +110,23 @@ describe("openSession", () => { }, ); + test("a log line arriving after close() is dropped, not written to the closed DB", async () => { + // A process's last lines can still be in its pipe after stop() returns. + const session = openSession(dir, { broadcaster: new LogBroadcaster(), dbPath: ":memory:" }); + await session.close(); + expect(() => + session.onLog({ + commandId: "late", + commandName: "Late", + profile: "default", + pid: 1, + stream: "stdout", + message: "trailing line", + timestamp: new Date().toISOString(), + }), + ).not.toThrow(); + }); + test("close() still releases the DB when stopping processes fails", async () => { const session = openSession(dir, { broadcaster: new LogBroadcaster(), dbPath: ":memory:" }); session.store.getQueue().stopAll = async () => { diff --git a/packages/core/test/wrapper.test.ts b/packages/core/test/wrapper.test.ts index b1ffc1a..78db4c3 100644 --- a/packages/core/test/wrapper.test.ts +++ b/packages/core/test/wrapper.test.ts @@ -223,6 +223,21 @@ describe("ProcessWrapper.start", () => { }); }); +describe("ProcessWrapper.drained", () => { + test("resolves even when a log handler throws", async () => { + const wrapper = new ProcessWrapper( + makeCommand({ id: "loud", name: "Loud", run: `bun -e "console.log('hi')"` }), + "test", + testEnv(), + ); + wrapper.onLog(() => { + throw new Error("database is closed"); + }); + await runToExit(wrapper); + expect(await wrapper.drained.then(() => "resolved")).toBe("resolved"); + }); +}); + describe("ProcessWrapper.stop", () => { test("stops a long-running process within its timeout budget", async () => { const cmd = makeCommand({ diff --git a/packages/desktop-tauri/src-tauri/src/main.rs b/packages/desktop-tauri/src-tauri/src/main.rs index 6dceea6..eb25c2b 100644 --- a/packages/desktop-tauri/src-tauri/src/main.rs +++ b/packages/desktop-tauri/src-tauri/src/main.rs @@ -126,9 +126,25 @@ async fn start_sidecar(app: &AppHandle, state: &SidecarState) -> Result Vec { + std::process::Command::new("pgrep") + .args(["-P", &pid.to_string()]) + .output() + .map(|out| { + String::from_utf8_lossy(&out.stdout) + .lines() + .filter_map(|line| line.trim().parse().ok()) + .collect() + }) + .unwrap_or_default() +} + +/// Sends SIGTERM and gives the sidecar up to 10s to run its own graceful +/// shutdown (which stops every managed dev process it started, escalating to +/// SIGKILL after each one's stop_timeout_ms, 5s by default) before escalating +/// to SIGKILL. CommandChild::kill() alone is a hard kill, not this. async fn stop_sidecar(state: &SidecarState) { let Some(child) = state.child.lock().unwrap().take() else { return; @@ -151,11 +167,21 @@ async fn stop_sidecar(state: &SidecarState) { libc::kill(pid as i32, libc::SIGTERM); } - let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); while !state.exited.load(Ordering::SeqCst) && tokio::time::Instant::now() < deadline { tokio::time::sleep(Duration::from_millis(100)).await; } if !state.exited.load(Ordering::SeqCst) { + // Managed processes run in their own process groups (setsid), so + // killing the sidecar alone would orphan any it had not stopped + // yet. List its children first: once it dies they are reparented + // and `pgrep -P` no longer finds them. + for child in child_pids(pid) { + unsafe { + libc::kill(-child, libc::SIGKILL); + libc::kill(child, libc::SIGKILL); + } + } unsafe { libc::kill(pid as i32, libc::SIGKILL); } @@ -381,16 +407,8 @@ fn main() { } }); - // Silent startup check-and-install, packaged builds only. - if !cfg!(debug_assertions) { - let handle = app.handle().clone(); - tauri::async_runtime::spawn(async move { - match install_update(handle).await { - Err(err) if err != UP_TO_DATE => eprintln!("[updater] {err}"), - _ => {} - } - }); - } + // No startup install: it restarts the app, stopping every managed + // process. The UI checks on load and offers it (About card). Ok(()) }) diff --git a/packages/ui/src/App.tsx b/packages/ui/src/App.tsx index cb24720..319c828 100644 --- a/packages/ui/src/App.tsx +++ b/packages/ui/src/App.tsx @@ -15,6 +15,7 @@ import { SectionHeading } from "./components/SectionHeading"; import { useUiStore } from "./store/ui"; import { useProcesses } from "./hooks/useProcesses"; import { useNotifications } from "./hooks/useNotifications"; +import { useUpdateNotice } from "./hooks/useUpdateNotice"; const TOAST_COLOR: Record = { recovered: "green", unhealthy: "orange" }; @@ -43,6 +44,7 @@ export default function App() { const { view, selectedProcessKey, selectProcess, setView } = useUiStore(); const { data: processes } = useProcesses(); useNotificationToasts(); + useUpdateNotice(); // Re-derived from the live process list on every poll so pid/status/logs stay correct const selectedProcess = selectedProcessKey diff --git a/packages/ui/src/components/ProfileGridView.tsx b/packages/ui/src/components/ProfileGridView.tsx index 0595234..b6b3006 100644 --- a/packages/ui/src/components/ProfileGridView.tsx +++ b/packages/ui/src/components/ProfileGridView.tsx @@ -92,8 +92,9 @@ export function ProfileGridView() { if (!editName.trim()) return; const changes: { newName?: string; description?: string } = {}; if (editName !== editTargetName) changes.newName = editName; - if (editDescription !== profiles?.[editTargetName]?.description) - changes.description = editDescription || undefined; + // "" (not undefined, which JSON drops) is how a cleared description is sent. + if (editDescription !== (profiles?.[editTargetName]?.description ?? "")) + changes.description = editDescription; // If nothing changed, just close the modal if (Object.keys(changes).length === 0) { setEditModalOpen(false); diff --git a/packages/ui/src/hooks/useUpdateNotice.ts b/packages/ui/src/hooks/useUpdateNotice.ts new file mode 100644 index 0000000..ebc4d82 --- /dev/null +++ b/packages/ui/src/hooks/useUpdateNotice.ts @@ -0,0 +1,26 @@ +import { useEffect } from "react"; +import { notifications as toasts } from "@mantine/notifications"; + +/** + * In the desktop app, checks once for a newer release and says so. Installing + * restarts the app (stopping every managed process), so it is never done + * unprompted: the user installs from Environment → About when it suits them. + */ +export function useUpdateNotice() { + useEffect(() => { + const tauri = window.__TAURI__; + if (!tauri) return; + tauri.core + .invoke<{ current: string; available: string | null }>("check_update") + .then(({ available }) => { + if (!available) return; + toasts.show({ + title: "Update available", + message: `Conductor v${available} is ready. Install it from Environment → About when your processes can restart.`, + autoClose: false, + }); + }) + // Offline or no release feed: nothing to tell the user at startup. + .catch(() => {}); + }, []); +} diff --git a/packages/ui/test/api.test.ts b/packages/ui/test/api.test.ts index 05c290a..1981909 100644 --- a/packages/ui/test/api.test.ts +++ b/packages/ui/test/api.test.ts @@ -40,6 +40,13 @@ describe("profiles", () => { expect((await api.fetchProfiles()).qa).toBeUndefined(); }); + test("updateProfile can clear a description", async () => { + await api.createProfile("docs", "has one"); + await api.updateProfile("docs", { description: "" }); + expect((await api.fetchProfiles()).docs?.description).toBeUndefined(); + await api.deleteProfile("docs"); + }); + test("errors surface the server's message, not a generic fallback", async () => { const err = await api.createProfile("dev").catch((e: Error) => e); expect(err).toBeInstanceOf(Error);