diff --git a/packages/opencode/src/altimate/native/connections/credential-store.ts b/packages/opencode/src/altimate/native/connections/credential-store.ts index c70466cc23..2cf82ef872 100644 --- a/packages/opencode/src/altimate/native/connections/credential-store.ts +++ b/packages/opencode/src/altimate/native/connections/credential-store.ts @@ -167,7 +167,9 @@ export async function deleteCredential( * Entries are keyed by connection name only. Only `saveConnection` writes them, for the saved (global) connection * of that name; a project's own connections file never stores a secret here. A project file that defines a * connection of the same name without its secret was reading the saved connection's secret, and loses it when that - * saved connection is removed or re-saved without it. + * saved connection is removed or re-saved without it. The registry keeps the entries on removal while an env var or + * the local config of a project this process currently serves still names the connection; projects it does not + * serve (closed, never opened, or in another process) are not consulted. */ export async function forgetCredentials(name: string, keep?: ConnectionConfig): Promise { const store = await getBackend() diff --git a/packages/opencode/src/altimate/native/connections/registry.ts b/packages/opencode/src/altimate/native/connections/registry.ts index 1f6195cc9f..5154de671b 100644 --- a/packages/opencode/src/altimate/native/connections/registry.ts +++ b/packages/opencode/src/altimate/native/connections/registry.ts @@ -26,7 +26,10 @@ import { forgetCredentials, notRemovedWarning, resolveConfig, saveConnection } f // altimate_change start — the shared home, which honours the test-isolation override import { Global } from "@/global" // altimate_change end -import { startTunnel, extractSshConfig, closeTunnel } from "./ssh-tunnel" +// altimate_change start — tunnels are owned by the connect attempt that opened them; instance disposal hook +import { openTunnel, extractSshConfig, type TunnelHandle } from "./ssh-tunnel" +import { registerDisposer } from "@/effect/instance-registry" +// altimate_change end import type { WarehouseInfo } from "../types" import { Telemetry } from "../../../telemetry" import * as SignInNotice from "./sign-in-notice" @@ -36,17 +39,111 @@ import { fileLog } from "@/altimate/util/file-log" import { SCRATCH_CONFIG, SCRATCH_DESCRIPTION, SCRATCH_NAME, closeScratch, getScratch, isScratch } from "./scratch" // altimate_change end -/** In-memory config store. */ -let configs = new Map() +// altimate_change start — registry state per project directory +/** + * One project's registry. A server serves several project directories from one process (each request carries its + * own `Instance.directory`), so configs, connectors, in-flight connects and the load latch are kept per project: + * project A never sees project B's project-local entries, and switching projects never closes another project's + * connectors. The global file and `ALTIMATE_CODE_CONN_*` env are shared sources merged into each project's state. + */ +interface State { + /** The project directory this state was loaded for. */ + dir: string + /** In-memory config store. */ + configs: Map + /** Cached connector instances. */ + connectors: Map + /** SSH tunnels of cached connectors, owned by the attempt that created each connector. */ + tunnels: Map + /** In-flight connects. An attempt is current only while it occupies its name's slot. */ + pending: Map + /** Whether this state has been loaded from its sources. */ + loaded: boolean +} + +/** + * One connect attempt. Its SSH tunnel is its own handle, never looked up by name, so a superseded attempt can only + * close the tunnel it opened. Dropping the attempt from `pending` (reload, remove, add, dispose, reset) closes that + * tunnel and makes the attempt discard whatever it finishes creating. + */ +interface Attempt { + promise?: Promise + tunnel?: TunnelHandle +} + +function createState(dir: string): State { + return { dir, configs: new Map(), connectors: new Map(), tunnels: new Map(), pending: new Map(), loaded: false } +} + +/** Registry state per resolved project directory. */ +const states = new Map() + +/** Invalidate the in-flight connect for `name`, closing the tunnel it opened. Synchronous. */ +function dropAttempt(s: State, name: string): void { + const attempt = s.pending.get(name) + s.pending.delete(name) + attempt?.tunnel?.close() +} + +/** Synchronously detach the cached connector and tunnel for `name`; returns a task that closes them. */ +function detachCached(s: State, name: string): () => Promise { + const connector = s.connectors.get(name) + const tunnel = s.tunnels.get(name) + s.connectors.delete(name) + s.tunnels.delete(name) + return async () => { + // The tunnel is closed first so a hanging connector close cannot keep it open. + tunnel?.close() + await connector?.close().catch(() => {}) + } +} + +let disposerRegistered = false +/** + * Close a project's connectors, in-flight connects and tunnels when its instance is disposed (`/instance/dispose`, + * config updates, reload). Its configs stay: they may hold session-only credentials from `warehouse_add` (no OS + * credential store), which must last for the process, and a populated registry is never re-read anyway. + */ +function ensureDisposer(): void { + if (disposerRegistered) return + disposerRegistered = true + registerDisposer(async (directory) => { + const s = states.get(path.resolve(directory)) + if (s) await closeState(s) + }) +} + +/** + * Set by setConfigs() (tests, benchmarks): one injected registry that every project uses until reset() or load(), + * so a test that injects outside an instance still sees its configs from inside one. load() only deactivates it, + * so connectors it cached stay reachable by closeAll(). + */ +let injectedState: State | undefined +let injectionActive = false -/** Cached connector instances. */ -const connectors = new Map() +/** The registry state the current request uses: the injected one while active, else its project's. */ +function state(): State { + if (injectionActive && injectedState) return injectedState + return projectState() +} -/** In-flight connector creation promises to prevent race conditions. */ -const pending = new Map>() +/** The registry state of the project the current request is working in. */ +function projectState(): State { + const dir = path.resolve(projectRoot()) + let s = states.get(dir) + if (!s) { + ensureDisposer() + s = createState(dir) + states.set(dir, s) + } + return s +} -/** Whether the registry has been loaded. */ -let loaded = false +/** Every state that may hold connectors, for process-wide teardown. */ +function allStates(): State[] { + return injectedState ? [...states.values(), injectedState] : [...states.values()] +} +// altimate_change end const SCRATCH_NAME_CONFLICT = `Connection "${SCRATCH_NAME}" conflicts with the reserved built-in scratch warehouse name. ` + @@ -77,12 +174,7 @@ export function globalConfigPath(): string { * context — early CLI paths and unit tests — and the working directory is the * right answer in exactly those cases, since `run --dir` has already chdir'd. * - * The registry itself is still process-global: `loaded` latches on first - * access, so one server process serving two projects keeps the first project's - * configs for both. That predates this change — the base was simply the launch - * directory for everyone — and making the registry per-instance is a larger - * change than this fix. What this does guarantee is that whichever project - * loads resolves against itself, not against wherever the server was started. + * It is also the key of the per-project registry state (see `state()`). */ function projectRoot(): string { try { @@ -95,10 +187,16 @@ function projectRoot(): string { export function localConfigPath(): string { // altimate_change start — resolve against the project, not the launch cwd - return path.join(projectRoot(), ".altimate-code", "connections.json") + return localConfigPathFor(projectRoot()) // altimate_change end } +// altimate_change start — one definition of a project's connections file +function localConfigPathFor(projectDir: string): string { + return path.join(projectDir, ".altimate-code", "connections.json") +} +// altimate_change end + // --------------------------------------------------------------------------- // Store path resolution // --------------------------------------------------------------------------- @@ -196,20 +294,18 @@ function loadFromEnv(): Record { return result } -/** Load all connection configs. Local overrides global; env overrides both. */ -export function load(includeProjectConfig = true): void { +// altimate_change start — load into one project's state +/** Read the sources into `s`. Local overrides global; env overrides both. */ +function loadInto(s: State): void { + const configs = s.configs configs.clear() - // altimate_change start — absolutize store paths against the directory that - // declared them, so a later process.chdir() (--dir) cannot move the store. - const base = projectRoot() + // absolutize store paths against the directory that declared them, so a later process.chdir() (--dir) cannot + // move the store. `s.dir` is the project this state belongs to. + const base = s.dir const global = resolveStorePaths(loadFromFile(globalConfigPath()), path.dirname(globalConfigPath())) - // altimate_change start — see ensureLoaded(): a later reload must not pull in another project's file - const local = includeProjectConfig ? resolveStorePaths(loadFromFile(localConfigPath()), base) : {} - if (includeProjectConfig) loadedProjectDir = base - // altimate_change end + const local = resolveStorePaths(loadFromFile(localConfigPathFor(base)), base) const env = resolveStorePaths(loadFromEnv(), base) - // altimate_change end // Merge: global < local < env for (const [name, config] of Object.entries(global)) { @@ -223,39 +319,27 @@ export function load(includeProjectConfig = true): void { } if (configs.has(SCRATCH_NAME)) Log.Default.warn(SCRATCH_NAME_CONFLICT) - loaded = true - // altimate_change start — real sources were read, so the registry is no longer test-injected - injected = false - // altimate_change end + s.loaded = true } -/** Ensure configs are loaded. */ -function ensureLoaded(): void { - if (!loaded) load() - // altimate_change start — the registry latches on first load, so a connections file created after an empty - // answer (the `warehouse_list` hint) was never seen. Re-read only while nothing is loaded: there is no state to - // replace or close, and a populated registry never changes under a running session. - else if (configs.size === 0 && !injected) load(projectConfigApplies()) - // altimate_change end +/** Load the current project's connection configs from their sources, ending any setConfigs() injection. */ +export function load(): void { + injectionActive = false + loadInto(projectState()) } -// altimate_change start — the registry is process-global; this is the project whose local file it was loaded with -let loadedProjectDir: string | undefined - -/** - * Whether the current project's local connections file is the one this registry reads. A server serving several - * directories shares one registry (pre-existing: the first load's project-local file applies process-wide), so a - * later empty-registry reload must not load another project's file into it. - */ -export function projectConfigApplies(): boolean { - return loadedProjectDir === undefined || loadedProjectDir === projectRoot() +/** Ensure the current project's configs are loaded; returns its state. */ +function ensureLoaded(): State { + const s = state() + if (!s.loaded) loadInto(s) + // A connections file created after an empty answer (the `warehouse_list` hint) must be seen. Re-read only while + // this project's registry is empty: there is nothing to replace or close, and a populated registry never changes + // under a running session. An injected registry (setConfigs) is never re-read. + else if (s.configs.size === 0 && s !== injectedState) loadInto(s) + return s } // altimate_change end -// altimate_change start — set by setConfigs() (tests): an injected empty registry is not a stale one -let injected = false -// altimate_change end - // --------------------------------------------------------------------------- // Driver factory // --------------------------------------------------------------------------- @@ -304,7 +388,17 @@ export function canonicalType(type: string | undefined | null): string | null { } // altimate_change end -async function createConnector(name: string, config: ConnectionConfig): Promise { +// altimate_change start — the attempt owns the SSH tunnel it opens +const SUPERSEDED = (name: string) => + new Error(`Connection "${name}" was reloaded or removed while it was connecting. Retry the request.`) + +async function createConnector( + name: string, + config: ConnectionConfig, + attempt: Attempt, + isCurrent: () => boolean, +): Promise { + // altimate_change end const driverPath = DRIVER_MAP[config.type.toLowerCase()] if (!driverPath) { // altimate_change start — friendlier error for known-but-unsupported databases @@ -360,12 +454,21 @@ async function createConnector(name: string, config: ConnectionConfig): Promise< // Handle SSH tunnel const sshConfig = extractSshConfig(resolvedConfig) if (sshConfig) { - const tunnel = await startTunnel(name, sshConfig) + // altimate_change start — a superseded attempt opens nothing; a tunnel that finishes opening after its attempt + // was dropped is closed by its own handle (dropAttempt closed it already) and never published + if (!isCurrent()) throw SUPERSEDED(name) + attempt.tunnel = openTunnel(sshConfig) + const tunnelInfo = await attempt.tunnel.ready + if (!isCurrent()) { + attempt.tunnel.close() + throw SUPERSEDED(name) + } + // altimate_change end // Rewrite host/port to use the local tunnel resolvedConfig = { ...resolvedConfig, host: "127.0.0.1", - port: tunnel.localPort, + port: tunnelInfo.localPort, } } @@ -419,9 +522,9 @@ async function createConnector(name: string, config: ConnectionConfig): Promise< connector = await mod.connect(resolvedConfig) } catch (e) { // Clean up SSH tunnel if driver creation fails - if (sshConfig) { - closeTunnel(name) - } + // altimate_change start — this attempt's own tunnel, never another attempt's + attempt.tunnel?.close() + // altimate_change end throw e } return connector @@ -528,14 +631,16 @@ function accountOf(config: ConnectionConfig | undefined): string { // altimate_change start — connection names without the census `list()` sends (used by `debug bundle`) export function names(): string[] { - ensureLoaded() - return [...configs.keys()] + return [...ensureLoaded().configs.keys()] } // altimate_change end /** Get a connector instance (creates lazily). */ export async function get(name: string): Promise { - ensureLoaded() + // altimate_change start — captured once, so a connect that finishes later lands in the project that started it + const s = ensureLoaded() + const { configs, connectors, tunnels, pending } = s + // altimate_change end if (isScratch(name)) { if (configs.has(name)) throw new Error(SCRATCH_NAME_CONFLICT) return getScratch() @@ -550,8 +655,10 @@ export async function get(name: string): Promise { } // If a connector is already being created, await the same Promise - const inflight = pending.get(name) + // altimate_change start — the slot holds the attempt + const inflight = pending.get(name)?.promise if (inflight) return inflight + // altimate_change end const config = configs.get(name) if (!config) { @@ -578,17 +685,33 @@ export async function get(name: string): Promise { }) } catch {} // altimate_change end + // altimate_change start — this attempt occupies the slot before any await; reload()/remove()/add()/dispose drop it + const attempt: Attempt = {} + pending.set(name, attempt) + const current = () => pending.get(name) === attempt + // altimate_change end const promise = (async () => { try { - const connector = await createConnector(name, config) + // altimate_change start — the attempt closes only the tunnel it opened + const connector = await createConnector(name, config, attempt, current) try { await connector.connect() } catch (connectErr) { - // If connect() fails after tunnel was started, clean up the tunnel - closeTunnel(name) + // If connect() fails after tunnel was started, clean up the tunnel (first, so a hanging close cannot hold it) + attempt.tunnel?.close() + await connector.close().catch(() => {}) throw connectErr } + // reload()/remove()/add() dropped this attempt while it was connecting: its config is gone or replaced, so + // it must not be cached over the current one. + if (!current()) { + attempt.tunnel?.close() + await connector.close().catch(() => {}) + throw SUPERSEDED(name) + } connectors.set(name, connector) + if (attempt.tunnel) tunnels.set(name, attempt.tunnel) + // altimate_change end fileLog("INFO", "warehouse-connect", "connected", { name, type: config.type, auth: authMethod, duration_ms: Date.now() - startTime }) try { Telemetry.track({ @@ -602,7 +725,11 @@ export async function get(name: string): Promise { }) } catch {} return connector - } catch (e) { + } catch (caught) { + // altimate_change start — an attempt dropped by reload/remove/add/dispose fails with whatever its closed tunnel + // or socket raised; report it as superseded (with the retry hint), not as a broken connection + const e = current() ? caught : SUPERSEDED(name) + // altimate_change end fileLog("WARN", "warehouse-connect", "connect failed", { name, type: config?.type, @@ -626,11 +753,15 @@ export async function get(name: string): Promise { } catch {} throw e } finally { - pending.delete(name) + // altimate_change start — a newer attempt for this name may already own the slot + if (current()) pending.delete(name) + // altimate_change end } })() - pending.set(name, promise) + // altimate_change start + attempt.promise = promise + // altimate_change end return promise } @@ -639,7 +770,9 @@ let censusSent = false /** List all configured connections. */ export function list(): { warehouses: (WarehouseInfo & { conflict?: string })[] } { - ensureLoaded() + // altimate_change start — the current project's registry + const { configs } = ensureLoaded() + // altimate_change end const warehouses: (WarehouseInfo & { conflict?: string })[] = [] for (const [name, config] of configs) { const type = typeof config.type === "string" ? config.type.toLowerCase() : "" @@ -700,8 +833,11 @@ export async function test( recoverable?: boolean }> { try { + // altimate_change start — the same project's state get() used + const s = state() const connector = await get(name) - const config = configs.get(name) + const config = s.configs.get(name) + // altimate_change end const dbType = config?.type?.toLowerCase() if (dbType === "mongodb" || dbType === "mongo") { // MongoDB doesn't support SQL — use the standard ping command @@ -742,7 +878,10 @@ export async function add( } // altimate_change end try { - ensureLoaded() + // altimate_change start — act on the current project's state, captured before any await + const s = ensureLoaded() + const { configs } = s + // altimate_change end // Normalize field names before saving so sensitive fields under alias // names (e.g., keyfileJson → credentials_json) are properly detected @@ -776,17 +915,12 @@ export async function add( // session can connect even when keytar is unavailable. Only the disk // file uses the sanitized version (credentials stripped). configs.set(name, normalized) - - // Clear cached connector - const cached = connectors.get(name) - if (cached) { - try { - await cached.close() - } catch { - // ignore - } - connectors.delete(name) - } + // altimate_change start — in the same synchronous step: a connect still in flight for the old config must not be + // cached over the new one, and the old connector's tunnel closes with it + dropAttempt(s, name) + const release = detachCached(s, name) + await release() + // altimate_change end const result: { success: boolean; name: string; type: string; warnings?: string[] } = { success: true, @@ -808,38 +942,38 @@ export async function remove(name: string): Promise<{ success: boolean; error?: if (isScratch(name)) return { success: false, error: `"${SCRATCH_NAME}" is built in and cannot be removed.` } // altimate_change end try { - ensureLoaded() - - // Close connector if cached - const cached = connectors.get(name) - if (cached) { - try { - await cached.close() - } catch { - // ignore - } - connectors.delete(name) - } - - // Close SSH tunnel if active - closeTunnel(name) + // altimate_change start — act on the current project's state, captured before any await + const s = ensureLoaded() // A connect still in flight for this name will finish and keep its name in use; forgetting it now would let a // later reconnect on that account be put down to another connection that shares it. - if (!pending.has(name)) ReconnectLog.forget(name) - - // Remove from global config file + const inflight = s.pending.has(name) + const projectConfigPath = localConfigPath() + // One synchronous step before any await. The global file is written first, so a failed write leaves this + // project's state untouched. Then the in-flight connect is invalidated (and its tunnel closed), the config is + // gone so no connect or empty-registry re-read can bring it back, and the cached connector and tunnel are + // detached. Only a file that holds the entry is rewritten (a missing global directory is not an error). const globalPath = globalConfigPath() const existing = loadFromFile(globalPath) - delete existing[name] - fs.writeFileSync(globalPath, JSON.stringify(existing, null, 2), "utf-8") - - // Remove from in-memory - configs.delete(name) + if (Object.hasOwn(existing, name)) { + delete existing[name] + fs.writeFileSync(globalPath, JSON.stringify(existing, null, 2), "utf-8") + } + dropAttempt(s, name) + s.configs.delete(name) + const release = detachCached(s, name) + await release() + if (!inflight) ReconnectLog.forget(name) + // altimate_change end // altimate_change start — remove its secrets from the OS credential store too. Entries are keyed by name - // only, so they are kept while this project's config or an env var still defines a connection by that name. - // (Own properties only: a connection named "constructor" must not count as defined.) - const stillDefined = Object.hasOwn(loadFromFile(localConfigPath()), name) || Object.hasOwn(loadFromEnv(), name) + // only, so they are kept while an env var or the local config of a project this process serves still defines a + // connection by that name (see forgetCredentials); the served set is read now, after the awaits above. (Own + // properties only: a connection named "constructor" must not count as defined.) + // Open instances may not have performed a warehouse operation yet, so states alone is incomplete. + const projectDirs = new Set([...states.keys(), ...Instance.directories()]) + const projectConfigPaths = [projectConfigPath, ...[...projectDirs].map(localConfigPathFor)] + const stillDefined = + projectConfigPaths.some((file) => Object.hasOwn(loadFromFile(file), name)) || Object.hasOwn(loadFromEnv(), name) const notRemoved = stillDefined ? [] : await forgetCredentials(name) if (notRemoved.length > 0) return { success: true, warnings: [notRemovedWarning(name, notRemoved)] } // altimate_change end @@ -851,64 +985,78 @@ export async function remove(name: string): Promise<{ success: boolean; error?: } // altimate_change start — a way to release native handles without reloading +/** Close one project's cached connectors and their SSH tunnels, and drop its in-flight connects. */ +async function closeState(s: State): Promise { + // Detach everything synchronously, so nothing started during the awaits below is dropped or reuses a closing + // connector. In-flight attempts are invalidated and their tunnels closed by their own handles. + for (const name of [...s.pending.keys()]) dropAttempt(s, name) + const releases = [...s.connectors.keys()].map((name) => detachCached(s, name)) + // Started together: each release closes its tunnel synchronously, so one hanging connector close holds no tunnel. + await Promise.all(releases.map((release) => release())) +} + /** - * Close every cached connector and drop it from the cache. + * Close every project's cached connectors (and the scratch warehouse) and drop them from the cache — process-wide + * teardown. * * `reset()` cannot do this: it is synchronous, and `close()` is not. Callers - * that create real connectors (tests, and `reload()` below) must await this, + * that create real connectors (tests) must await this, * otherwise the native handle behind each connector stays open — on Windows * that also blocks deleting the file underneath it. */ export async function closeAll(): Promise { await closeScratch() - for (const [, connector] of connectors) { - try { - await connector.close() - } catch { - // ignore - } - } - connectors.clear() + for (const s of allStates()) await closeState(s) } // altimate_change end -/** Reload all configs and clear cached connectors. */ +/** Reload the current project's configs and clear its cached connectors; other projects are untouched. */ export async function reload(): Promise { - await closeAll() - loaded = false - load() + // altimate_change start — only this project's connectors. The new configs are in place before the first await, + // so no connect can start against the old ones while the old connectors close. + const s = state() + const target = projectState() + const closing = closeState(s) + injectionActive = false + loadInto(target) + await closing + // altimate_change end } /** Get the raw config for a connection (for testing). */ export function getConfig(name: string): ConnectionConfig | undefined { - ensureLoaded() + // altimate_change start — the current project's registry + const { configs } = ensureLoaded() + // altimate_change end if (isScratch(name) && !configs.has(name)) return { ...SCRATCH_CONFIG } return configs.get(name) } -/** Reset the registry state (for testing). */ +/** Reset the registry state of every project (for testing). */ export function reset(): void { - configs.clear() - connectors.clear() - pending.clear() - loaded = false - censusSent = false - // altimate_change start - injected = false - loadedProjectDir = undefined + // altimate_change start — invalidate in-flight connects first, so none is cached into a dropped state + for (const s of allStates()) for (const name of [...s.pending.keys()]) dropAttempt(s, name) + states.clear() + injectedState = undefined + injectionActive = false // altimate_change end + censusSent = false } /** - * Set configs directly (for testing without file system). + * Set configs directly (for testing without file system). Every project uses them until reset() or load(). */ export function setConfigs(newConfigs: Record): void { + // altimate_change start — one injected state shared by every project; its cached connectors are kept, as before + injectedState ??= createState("") + injectionActive = true + const configs = injectedState.configs + // altimate_change end configs.clear() for (const [name, config] of Object.entries(newConfigs)) { configs.set(name, config) } - loaded = true // altimate_change start - injected = true + injectedState.loaded = true // altimate_change end } diff --git a/packages/opencode/src/altimate/native/connections/ssh-tunnel.ts b/packages/opencode/src/altimate/native/connections/ssh-tunnel.ts index 4a837fb01c..27a4ffecd5 100644 --- a/packages/opencode/src/altimate/native/connections/ssh-tunnel.ts +++ b/packages/opencode/src/altimate/native/connections/ssh-tunnel.ts @@ -3,15 +3,29 @@ * * Uses the `ssh2` package (dynamic import). If ssh2 is not installed, * throws a clear error message. + * + * Tunnels are not keyed by connection name. Each `openTunnel()` call returns a + * handle owned by the caller (the connection registry's connect attempt), and + * only that handle closes it: an attempt that was superseded can never register + * over, or close, a newer attempt's tunnel. */ import type { ConnectionConfig } from "@altimateai/drivers" +import type { Server, Socket } from "net" +import type { Duplex } from "stream" export interface TunnelInfo { localPort: number close(): void } +/** A tunnel being opened. `close()` works before or after `ready` settles and is idempotent. */ +export interface TunnelHandle { + /** Resolves once the local forwarding port listens; rejects on failure or when closed first. */ + readonly ready: Promise + close(): void +} + export interface SshConfig { ssh_host: string ssh_port?: number @@ -22,161 +36,165 @@ export interface SshConfig { port: number } -/** Active tunnels keyed by connection name. */ -const activeTunnels = new Map() +/** Every tunnel opened and not yet closed, for process-exit cleanup and `closeAllTunnels()`. */ +const openTunnels = new Set() /** Clean up all tunnels on process exit. */ let cleanupRegistered = false function ensureCleanup(): void { if (cleanupRegistered) return cleanupRegistered = true - const cleanup = () => { - for (const [, tunnel] of activeTunnels) { - try { - tunnel.close() - } catch { - // best-effort cleanup - } - } - activeTunnels.clear() - } - process.once("exit", cleanup) + process.once("exit", closeAllTunnels) process.once("SIGINT", () => { - cleanup() + closeAllTunnels() process.exit(0) }) process.once("SIGTERM", () => { - cleanup() + closeAllTunnels() process.exit(0) }) } /** - * Start an SSH tunnel for a connection. - * Returns a TunnelInfo with the local port to connect to. + * Open an SSH tunnel that forwards a local port to `config.host:config.port`. + * + * One terminal state covers every path: an error, a close before `ready`, or a + * late `ready`/listen callback after either closes the client, the local server + * and forwarded sockets, and never publishes a tunnel. */ -export async function startTunnel( - name: string, - config: SshConfig, -): Promise { - // Close existing tunnel for this name - const existing = activeTunnels.get(name) - if (existing) { - existing.close() - activeTunnels.delete(name) - } - - let ssh2: any - try { - // @ts-expect-error — optional dependency - ssh2 = await import("ssh2") - } catch { - throw new Error( - "SSH tunnel requires the ssh2 package. Run: bun add ssh2 @types/ssh2", - ) +export function openTunnel(config: SshConfig): TunnelHandle { + let closed = false + let client: any + let server: Server | undefined + const sockets = new Set() + let rejectReady: ((e: Error) => void) | undefined + + const close = () => { + if (closed) return + closed = true + openTunnels.delete(handle) + for (const socket of sockets) socket.destroy() + try { + server?.close() + } catch { + // ignore + } + try { + client?.end() + } catch { + // ignore + } + rejectReady?.(new Error("SSH tunnel was closed before it was ready")) } - ensureCleanup() - - const net = await import("net") - - return new Promise((resolve, reject) => { - const client = new ssh2.Client() - - client.on("ready", () => { - // Create a local TCP server that forwards to remote host:port via SSH - const server = net.createServer((localSocket) => { - client.forwardOut( - "127.0.0.1", - 0, - config.host, - config.port, - (err: Error | undefined, stream: any) => { - if (err) { + const ready = (async (): Promise => { + let ssh2: any + try { + // @ts-expect-error — optional dependency + ssh2 = await import("ssh2") + } catch { + throw new Error("SSH tunnel requires the ssh2 package. Run: bun add ssh2 @types/ssh2") + } + const net = await import("net") + if (closed) throw new Error("SSH tunnel was closed before it was ready") + ensureCleanup() + + return new Promise((resolve, reject) => { + rejectReady = reject + client = new ssh2.Client() + + client.on("ready", () => { + if (closed) return + // Create a local TCP server that forwards to remote host:port via SSH + server = net.createServer((localSocket) => { + let remoteStream: Duplex | undefined + localSocket.on("error", () => { + remoteStream?.destroy() + localSocket.destroy() + }) + if (closed) { + localSocket.destroy() + return + } + sockets.add(localSocket) + localSocket.on("close", () => sockets.delete(localSocket)) + client.forwardOut("127.0.0.1", 0, config.host, config.port, (err: Error | undefined, stream: any) => { + remoteStream = stream + stream?.on("error", () => { + localSocket.destroy() + stream.destroy() + }) + if (err || closed || localSocket.destroyed) { + stream?.destroy() localSocket.destroy() return } localSocket.pipe(stream).pipe(localSocket) - }, - ) + }) + }) + + server.on("error", (err: Error) => { + reject(new Error(`SSH tunnel local server error: ${err.message}`)) + close() + }) + + server.listen(0, "127.0.0.1", () => { + if (closed) return + const addr = server?.address() + const localPort = typeof addr === "object" && addr ? addr.port : 0 + resolve({ localPort, close }) + }) }) - server.listen(0, "127.0.0.1", () => { - const addr = server.address() - const localPort = - typeof addr === "object" && addr ? addr.port : 0 - - const tunnelInfo: TunnelInfo = { - localPort, - close() { - try { - server.close() - } catch { - // ignore - } - try { - client.end() - } catch { - // ignore - } - activeTunnels.delete(name) - }, - } - - activeTunnels.set(name, tunnelInfo) - resolve(tunnelInfo) + client.on("error", (err: Error) => { + reject(new Error(`SSH connection error: ${err.message}`)) + close() }) - server.on("error", (err: Error) => { - client.end() - reject(new Error(`SSH tunnel local server error: ${err.message}`)) + // A clean remote disconnect ends the tunnel too: before ready it rejects, after it closes the listener. + client.on("close", () => { + reject(new Error("SSH connection closed")) + close() }) - }) - - client.on("error", (err: Error) => { - reject(new Error(`SSH connection error: ${err.message}`)) - }) - - const connectOptions: Record = { - host: config.ssh_host, - port: config.ssh_port ?? 22, - username: config.ssh_user ?? "root", - } - if (config.ssh_private_key) { - connectOptions.privateKey = config.ssh_private_key - } else if (config.ssh_password) { - connectOptions.password = config.ssh_password - } + const connectOptions: Record = { + host: config.ssh_host, + port: config.ssh_port ?? 22, + username: config.ssh_user ?? "root", + } - client.connect(connectOptions) - }) -} + if (config.ssh_private_key) { + connectOptions.privateKey = config.ssh_private_key + } else if (config.ssh_password) { + connectOptions.password = config.ssh_password + } -/** Get an active tunnel by connection name. */ -export function getActiveTunnel(name: string): TunnelInfo | undefined { - return activeTunnels.get(name) -} + client.connect(connectOptions) + }) + })() -/** Close a specific tunnel by name. */ -export function closeTunnel(name: string): void { - const tunnel = activeTunnels.get(name) - if (tunnel) { - tunnel.close() - activeTunnels.delete(name) - } + const handle: TunnelHandle = { ready, close } + openTunnels.add(handle) + // Any failure ends in the terminal state; the caller still observes the rejection through `ready`. + ready.catch(() => close()) + return handle } -/** Close all active tunnels. */ +/** Close every open tunnel (process teardown and tests). */ export function closeAllTunnels(): void { - for (const [, tunnel] of activeTunnels) { + for (const tunnel of [...openTunnels]) { try { tunnel.close() } catch { // best-effort } } - activeTunnels.clear() + openTunnels.clear() +} + +/** Number of tunnels currently open or opening (diagnostics and tests). */ +export function openTunnelCount(): number { + return openTunnels.size } /** diff --git a/packages/opencode/src/altimate/tools/warehouse-list.ts b/packages/opencode/src/altimate/tools/warehouse-list.ts index eb6c2ba66a..7c6f75f405 100644 --- a/packages/opencode/src/altimate/tools/warehouse-list.ts +++ b/packages/opencode/src/altimate/tools/warehouse-list.ts @@ -6,7 +6,7 @@ import * as Precedence from "../workspace/precedence" // altimate_change end // altimate_change start — point at the locations the registry actually reads -import { CONN_ENV_PREFIX, globalConfigPath, localConfigPath, projectConfigApplies } from "../native/connections/registry" +import { CONN_ENV_PREFIX, globalConfigPath, localConfigPath } from "../native/connections/registry" /** The no-connections hint. Paths come from the registry's own helpers so it cannot drift from what is loaded. */ export function noConnectionsMessage(): string { @@ -17,8 +17,8 @@ export function noConnectionsMessage(): string { "", "Alternatively, connections are read from these sources (later ones override earlier ones):", ` 1. ${globalConfigPath()} (global)`, - ...(projectConfigApplies() ? [` 2. ${localConfigPath()} (this project)`] : []), - ` ${projectConfigApplies() ? 3 : 2}. ${CONN_ENV_PREFIX} environment variables, each holding one connection's config as JSON (set them in the shell before starting altimate-code; a running process cannot see variables exported later)`, + ` 2. ${localConfigPath()} (this project)`, + ` 3. ${CONN_ENV_PREFIX} environment variables, each holding one connection's config as JSON (set them in the shell before starting altimate-code; a running process cannot see variables exported later)`, "", "After adding the first connection file, call warehouse_list again. Edits to an existing connections.json take effect after a restart. Example connections.json:", "{", diff --git a/packages/opencode/src/project/instance.ts b/packages/opencode/src/project/instance.ts index 5ed8c18373..22d3410862 100644 --- a/packages/opencode/src/project/instance.ts +++ b/packages/opencode/src/project/instance.ts @@ -81,6 +81,11 @@ export const Instance = { return input.fn() }) }, + // altimate_change start — enumerate open projects even before their warehouse registry is loaded + directories(): string[] { + return [...cache.keys()] + }, + // altimate_change end get current() { return context.use() }, diff --git a/packages/opencode/test/altimate/registry-per-project.test.ts b/packages/opencode/test/altimate/registry-per-project.test.ts new file mode 100644 index 0000000000..f192d63a71 --- /dev/null +++ b/packages/opencode/test/altimate/registry-per-project.test.ts @@ -0,0 +1,536 @@ +/** + * The connection registry keeps one state per project directory. + * + * One server process serves several project directories (each request carries its own `Instance.directory`). The + * registry used to be process-global, so the first project to load decided every project's connections, and closing + * or reloading in one project closed another project's connectors. These tests drive two project directories in one + * process, through `process.chdir` (the CLI path) and through `Instance.provide` (the server path). + */ + +import { describe, test, expect, beforeEach, afterEach, afterAll, mock, spyOn } from "bun:test" +import * as fs from "fs" +import * as os from "os" +import * as path from "path" +import * as net from "net" +import { EventEmitter } from "events" +import { PassThrough } from "stream" + +// ssh2 is an optional dependency, normally not installed for tests; a fake client lets the real tunnel module run. +// Bun's mock.module leaks across test files, so the genuine module (or its absence) is restored in afterAll. +let genuineSsh2: unknown +try { + // @ts-expect-error — optional dependency + genuineSsh2 = await import("ssh2") +} catch { + genuineSsh2 = undefined +} +afterAll(() => { + // Absent: a module cannot be un-mocked, so later files get a client that fails loudly instead of this fake. + mock.module("ssh2", () => + genuineSsh2 + ? (genuineSsh2 as Record) + : { + Client: class { + constructor() { + throw new Error("Cannot find package 'ssh2'") + } + }, + }, + ) +}) + +const sshClients: { host: string; ended: boolean }[] = [] +/** Hosts whose SSH `ready` is held until the test releases it. */ +const heldReady = new Map void>() +/** Simulates the SSH server closing the connection cleanly (no error event). */ +const remoteClose = new Map void>() +let forwardedStream: PassThrough | undefined +mock.module("ssh2", () => ({ + Client: class { + private handlers = new Map void>() + private record = { host: "", ended: false } + on(event: string, cb: (...args: unknown[]) => void) { + this.handlers.set(event, cb) + return this + } + connect(options: { host: string }) { + this.record.host = options.host + sshClients.push(this.record) + const ready = () => this.handlers.get("ready")?.() + remoteClose.set(options.host, () => this.handlers.get("close")?.()) + if (options.host.startsWith("held-")) heldReady.set(options.host, ready) + else queueMicrotask(ready) + } + forwardOut(_a: string, _b: number, _c: string, _d: number, cb: (err?: Error, stream?: PassThrough) => void) { + if (forwardedStream) cb(undefined, forwardedStream) + else cb(new Error("no forwarding in tests")) + } + end() { + this.record.ended = true + } + }, +})) + +import * as Registry from "../../src/altimate/native/connections/registry" +import { openTunnel, closeAllTunnels, openTunnelCount } from "../../src/altimate/native/connections/ssh-tunnel" +import { Instance } from "../../src/project/instance" +import { disposeInstance } from "../../src/effect/instance-registry" + +/** Wait (bounded) until the fake SSH client for `host` has connected. */ +async function sshConnected(host: string) { + for (let i = 0; i < 500 && !sshClients.some((c) => c.host === host); i++) await new Promise((r) => setTimeout(r, 2)) + expect(sshClients.some((c) => c.host === host)).toBe(true) +} + +const ended = (host: string) => sshClients.filter((c) => c.host === host).map((c) => c.ended) + +function sshStore(sshHost: string, extra: Record = {}) { + return { type: "sqlite", path: "store.sqlite", create: true, ssh_host: sshHost, host: "db", port: 5432, ...extra } +} + +let projectA: string +let projectB: string +let home: string +let prevCwd: string +let prevHome: string | undefined +let ambientConnEnv: Record + +function writeLocal(dir: string, entries: Record) { + fs.mkdirSync(path.join(dir, ".altimate-code"), { recursive: true }) + fs.writeFileSync(path.join(dir, ".altimate-code", "connections.json"), JSON.stringify(entries)) +} + +function writeGlobal(entries: Record) { + fs.mkdirSync(path.join(home, ".altimate-code"), { recursive: true }) + fs.writeFileSync(path.join(home, ".altimate-code", "connections.json"), JSON.stringify(entries)) +} + +function inProject(dir: string, fn: () => T): T { + process.chdir(dir) + return fn() +} + +beforeEach(() => { + // Ambient connection env vars would leak the caller's shell into every assertion. + ambientConnEnv = {} + for (const [key, value] of Object.entries(process.env)) { + if (!key.startsWith(Registry.CONN_ENV_PREFIX) || value === undefined) continue + ambientConnEnv[key] = value + delete process.env[key] + } + prevCwd = process.cwd() + prevHome = process.env.OPENCODE_TEST_HOME + home = fs.realpathSync(fs.mkdtempSync(path.join(os.tmpdir(), "rpp-home-"))) + process.env.OPENCODE_TEST_HOME = home + projectA = fs.realpathSync(fs.mkdtempSync(path.join(os.tmpdir(), "rpp-a-"))) + projectB = fs.realpathSync(fs.mkdtempSync(path.join(os.tmpdir(), "rpp-b-"))) + forwardedStream = undefined + sshClients.length = 0 + heldReady.clear() + remoteClose.clear() + Registry.reset() +}) + +afterEach(async () => { + await Registry.closeAll() + closeAllTunnels() + Registry.reset() + process.chdir(prevCwd) + if (prevHome === undefined) delete process.env.OPENCODE_TEST_HOME + else process.env.OPENCODE_TEST_HOME = prevHome + for (const dir of [projectA, projectB, home]) fs.rmSync(dir, { recursive: true, force: true }) + for (const key of Object.keys(process.env)) { + if (key.startsWith(Registry.CONN_ENV_PREFIX)) delete process.env[key] + } + Object.assign(process.env, ambientConnEnv) +}) + +describe("project-local connections are per project", () => { + test("A's local entry is invisible to B, and B's to A (A loads first)", () => { + writeLocal(projectA, { only_a: { type: "postgres", host: "a" } }) + writeLocal(projectB, { only_b: { type: "postgres", host: "b" } }) + expect(inProject(projectA, () => Registry.names())).toEqual(["only_a"]) + expect(inProject(projectB, () => Registry.names())).toEqual(["only_b"]) + expect(inProject(projectA, () => Registry.getConfig("only_b"))).toBeUndefined() + expect(inProject(projectB, () => Registry.getConfig("only_a"))).toBeUndefined() + }) + + test("the same holds when B loads first", () => { + writeLocal(projectA, { only_a: { type: "postgres", host: "a" } }) + writeLocal(projectB, { only_b: { type: "postgres", host: "b" } }) + expect(inProject(projectB, () => Registry.names())).toEqual(["only_b"]) + expect(inProject(projectA, () => Registry.names())).toEqual(["only_a"]) + expect(inProject(projectB, () => Registry.list().warehouses.map((w) => w.name))).not.toContain("only_a") + }) + + test("a file added late to one project is seen there only, in either order", () => { + expect(inProject(projectA, () => Registry.names())).toEqual([]) + expect(inProject(projectB, () => Registry.names())).toEqual([]) + + writeLocal(projectB, { late_b: { type: "postgres" } }) + expect(inProject(projectA, () => Registry.names())).toEqual([]) + expect(inProject(projectB, () => Registry.names())).toEqual(["late_b"]) + + writeLocal(projectA, { late_a: { type: "postgres" } }) + expect(inProject(projectA, () => Registry.names())).toEqual(["late_a"]) + expect(inProject(projectB, () => Registry.names())).toEqual(["late_b"]) + }) + + test("each project re-reads only while its own registry is empty", () => { + writeLocal(projectA, { a1: { type: "postgres" } }) + expect(inProject(projectA, () => Registry.names())).toEqual(["a1"]) + expect(inProject(projectB, () => Registry.names())).toEqual([]) + + // A is populated: an edit is not picked up. B is empty: its new file is. + writeLocal(projectA, { a1: { type: "postgres" }, a2: { type: "postgres" } }) + writeLocal(projectB, { b1: { type: "postgres" } }) + expect(inProject(projectA, () => Registry.names())).toEqual(["a1"]) + expect(inProject(projectB, () => Registry.names())).toEqual(["b1"]) + }) + + test("global and env entries are visible in both projects", () => { + writeGlobal({ shared_global: { type: "postgres", host: "g" } }) + process.env.ALTIMATE_CODE_CONN_SHARED_ENV = JSON.stringify({ type: "postgres", host: "e" }) + writeLocal(projectA, { only_a: { type: "postgres" } }) + expect(inProject(projectA, () => Registry.names().sort())).toEqual(["only_a", "shared_env", "shared_global"]) + expect(inProject(projectB, () => Registry.names().sort())).toEqual(["shared_env", "shared_global"]) + }) + + test("the server path: Instance.provide with two directories", async () => { + writeLocal(projectA, { srv: { type: "postgres", host: "host-a" } }) + writeLocal(projectB, { srv: { type: "postgres", host: "host-b" } }) + const hostIn = (dir: string) => + Instance.provide({ directory: dir, fn: async () => Registry.getConfig("srv")?.host }) + expect(await hostIn(projectA)).toBe("host-a") + expect(await hostIn(projectB)).toBe("host-b") + expect(await hostIn(projectA)).toBe("host-a") + }) +}) + +describe("connectors are per project", () => { + test("the same name resolves to each project's own config and connector", async () => { + writeLocal(projectA, { store: { type: "sqlite", path: "store.sqlite", create: true } }) + writeLocal(projectB, { store: { type: "sqlite", path: "store.sqlite", create: true } }) + expect(inProject(projectA, () => Registry.getConfig("store")?.path)).toBe(path.join(projectA, "store.sqlite")) + expect(inProject(projectB, () => Registry.getConfig("store")?.path)).toBe(path.join(projectB, "store.sqlite")) + + const a = await inProject(projectA, () => Registry.get("store")) + await a.execute("CREATE TABLE marker_a (x INTEGER)") + const b = await inProject(projectB, () => Registry.get("store")) + expect(b).not.toBe(a) + const tables = await b.execute("SELECT name FROM sqlite_master WHERE type = 'table'") + expect(tables.rows.flat()).not.toContain("marker_a") + // the cached connector is per project too + expect(await inProject(projectA, () => Registry.get("store"))).toBe(a) + }) + + test("removing or reloading in B does not close A's connector", async () => { + writeLocal(projectA, { store: { type: "sqlite", path: "store.sqlite", create: true } }) + writeLocal(projectB, { store: { type: "sqlite", path: "store.sqlite", create: true } }) + const a = await inProject(projectA, () => Registry.get("store")) + await inProject(projectB, () => Registry.get("store")) + + await inProject(projectB, () => Registry.reload()) + expect((await a.execute("SELECT 1 AS one")).rows).toEqual([[1]]) + await inProject(projectB, () => Registry.remove("store")) + expect((await a.execute("SELECT 1 AS one")).rows).toEqual([[1]]) + expect(await inProject(projectA, () => Registry.get("store"))).toBe(a) + // B's local file still defines it, so B can connect again with a fresh connector + const b2 = await inProject(projectB, () => Registry.get("store")) + expect(b2).not.toBe(a) + }) + + test("same-named SSH tunnels are per project; removing in B leaves A's open", async () => { + const tunnelled = (sshHost: string) => ({ + store: { type: "sqlite", path: "store.sqlite", create: true, ssh_host: sshHost, host: "db", port: 5432 }, + }) + writeLocal(projectA, tunnelled("bastion-a")) + writeLocal(projectB, tunnelled("bastion-b")) + await inProject(projectA, () => Registry.get("store")) + await inProject(projectB, () => Registry.get("store")) + // B starting its tunnel must not have closed A's + expect(sshClients.map((c) => [c.host, c.ended])).toEqual([ + ["bastion-a", false], + ["bastion-b", false], + ]) + + await inProject(projectB, () => Registry.remove("store")) + expect(sshClients.map((c) => [c.host, c.ended])).toEqual([ + ["bastion-a", false], + ["bastion-b", true], + ]) + }) + + test("a connect in flight when its project reloads is not cached over the reloaded state", async () => { + writeLocal(projectA, { store: { type: "sqlite", path: "store.sqlite", create: true } }) + const inflight = inProject(projectA, () => Registry.get("store")) + await inProject(projectA, () => Registry.reload()) + await expect(inflight).rejects.toThrow("reloaded or removed while it was connecting") + const fresh = await inProject(projectA, () => Registry.get("store")) + expect((await fresh.execute("SELECT 1 AS one")).rows).toEqual([[1]]) + }) +}) + +describe("SSH forwarding errors", () => { + for (const side of ["local", "remote"] as const) { + test(`${side} socket errors destroy both forwarding endpoints without throwing`, async () => { + const stream = new PassThrough() + forwardedStream = stream + let accept: ((socket: net.Socket) => void) | undefined + const server = Object.assign(new EventEmitter(), { + listen: (_port: number, _host: string, ready: () => void) => ready(), + address: () => ({ port: 12345 }), + close: () => {}, + }) + const createServer = spyOn(net, "createServer").mockImplementation(((listener: (socket: net.Socket) => void) => { + accept = listener + return server + }) as typeof net.createServer) + const tunnel = openTunnel({ ssh_host: "forwarding", host: "db", port: 5432 }) + const local = new net.Socket() + try { + await tunnel.ready + expect(accept).toBeDefined() + accept!(local) + const endpoint = side === "local" ? local : stream + expect(() => endpoint.emit("error", new Error("ECONNRESET"))).not.toThrow() + expect(local.destroyed).toBe(true) + expect(stream.destroyed).toBe(true) + } finally { + local.destroy() + tunnel.close() + createServer.mockRestore() + } + }) + } +}) + +describe("each connect attempt owns its SSH tunnel", () => { + test("a stale tunnel start that finishes after a newer start never publishes or closes the newer tunnel", async () => { + writeLocal(projectA, { store: sshStore("held-old") }) + const stale = inProject(projectA, () => Registry.get("store")) + await sshConnected("held-old") + + // replace the config while the old tunnel is still opening; the new attempt connects through its own tunnel + const added = await inProject(projectA, () => Registry.add("store", sshStore("new") as any)) + expect(added.success).toBe(true) + await expect(stale).rejects.toThrow() + const fresh = await inProject(projectA, () => Registry.get("store")) + + // the old SSH handshake completes late: nothing is registered, the newer tunnel stays open + heldReady.get("held-old")?.() + await new Promise((r) => setTimeout(r, 5)) + expect(ended("held-old")).toEqual([true]) + expect(ended("new")).toEqual([false]) + expect(openTunnelCount()).toBe(1) + expect((await fresh.execute("SELECT 1 AS one")).rows).toEqual([[1]]) + }) + + test("a failing attempt closes only the tunnel it opened", async () => { + writeLocal(projectB, { store: sshStore("b-host") }) + await inProject(projectB, () => Registry.get("store")) + // same name in A, store missing and not creatable: the driver fails after A's tunnel opened + writeLocal(projectA, { store: sshStore("a-host", { create: false, path: "missing.sqlite" }) }) + await expect(inProject(projectA, () => Registry.get("store"))).rejects.toThrow("not found") + expect(ended("a-host")).toEqual([true]) + expect(ended("b-host")).toEqual([false]) + expect(openTunnelCount()).toBe(1) + }) + + test("replacing an SSH connection with a non-SSH one closes the old tunnel (cached and in flight)", async () => { + writeLocal(projectA, { store: sshStore("cached-host"), other: sshStore("held-pending", { path: "other.sqlite" }) }) + await inProject(projectA, () => Registry.get("store")) + const pending = inProject(projectA, () => Registry.get("other")) + await sshConnected("held-pending") + + const plain = { type: "sqlite", path: "plain.sqlite", create: true } as any + await inProject(projectA, () => Registry.add("store", plain)) + await inProject(projectA, () => Registry.add("other", plain)) + await expect(pending).rejects.toThrow() + expect(ended("cached-host")).toEqual([true]) + expect(ended("held-pending")).toEqual([true]) + expect(openTunnelCount()).toBe(0) + const store = await inProject(projectA, () => Registry.get("store")) + expect((await store.execute("SELECT 1 AS one")).rows).toEqual([[1]]) + expect(openTunnelCount()).toBe(0) + }) + + test("a clean SSH disconnect before ready fails the connect and leaves no tunnel", async () => { + writeLocal(projectA, { store: sshStore("held-closing") }) + const pending = inProject(projectA, () => Registry.get("store")) + await sshConnected("held-closing") + remoteClose.get("held-closing")?.() + await expect(pending).rejects.toThrow("SSH connection closed") + expect(openTunnelCount()).toBe(0) + }) + + test("reload with a pending SSH connect closes that attempt's tunnel", async () => { + writeLocal(projectA, { store: sshStore("held-reload") }) + const pending = inProject(projectA, () => Registry.get("store")) + await sshConnected("held-reload") + await inProject(projectA, () => Registry.reload()) + await expect(pending).rejects.toThrow() + expect(ended("held-reload")).toEqual([true]) + expect(openTunnelCount()).toBe(0) + }) +}) + +describe("instance disposal", () => { + test("disposing a project's instance closes its connectors and tunnels only", async () => { + writeLocal(projectA, { store: sshStore("a-host") }) + writeLocal(projectB, { store: sshStore("b-host") }) + const a = await inProject(projectA, () => Registry.get("store")) + const b = await inProject(projectB, () => Registry.get("store")) + + await disposeInstance(projectA) + expect(ended("a-host")).toEqual([true]) + expect(ended("b-host")).toEqual([false]) + await expect(a.execute("SELECT 1")).rejects.toThrow() + expect((await b.execute("SELECT 1 AS one")).rows).toEqual([[1]]) + + // reopening A connects again with a new connector + const a2 = await inProject(projectA, () => Registry.get("store")) + expect(a2).not.toBe(a) + expect((await a2.execute("SELECT 1 AS one")).rows).toEqual([[1]]) + }) + + test("session-only credentials from warehouse_add survive disposal", async () => { + // The test preload disables the OS credential store, so the password exists only in memory. + await inProject(projectA, () => Registry.add("pg", { type: "postgres", host: "h", password: "s3cret-test" } as any)) + await disposeInstance(projectA) + expect(inProject(projectA, () => Registry.getConfig("pg")?.password)).toBe("s3cret-test") + }) + + test("a connect dropped while its tunnel is opening fails as superseded, with the retry hint", async () => { + writeLocal(projectA, { store: sshStore("held-dispose") }) + const pending = inProject(projectA, () => Registry.get("store")) + await sshConnected("held-dispose") + await disposeInstance(projectA) + await expect(pending).rejects.toThrow("reloaded or removed while it was connecting. Retry the request.") + }) +}) + +describe("remove()", () => { + test("a failed global file write leaves the project's connection usable", async () => { + writeGlobal({ g: { type: "sqlite", path: path.join(projectA, "g.sqlite"), create: true } }) + const connector = await inProject(projectA, () => Registry.get("g")) + const globalDir = path.join(home, ".altimate-code") + fs.chmodSync(globalDir, 0o500) + fs.chmodSync(path.join(globalDir, "connections.json"), 0o400) + try { + const result = await inProject(projectA, () => Registry.remove("g")) + expect(result.success).toBe(false) + expect(inProject(projectA, () => Registry.names())).toContain("g") + expect(await inProject(projectA, () => Registry.get("g"))).toBe(connector) + } finally { + fs.chmodSync(globalDir, 0o700) + fs.chmodSync(path.join(globalDir, "connections.json"), 0o600) + } + }) +}) + +describe("credentials of a removed connection", () => { + test("are kept for an open project that has never used the warehouse registry", async () => { + const { setSecretBackendForTests } = await import("../../src/altimate/native/connections/credential-store") + const store = new Map() + setSecretBackendForTests({ + name: "test", + get: async (k: string) => store.get(k) ?? null, + set: async (k: string, v: string) => void store.set(k, v), + delete: async (k: string) => store.delete(k), + } as any) + try { + writeLocal(projectB, { shared: { type: "postgres", host: "b" } }) + await Instance.provide({ directory: projectB, fn: async () => {} }) + await inProject(projectA, () => Registry.add("shared", { type: "postgres", host: "g", password: "pw-test" } as any)) + expect(store.get("shared/password")).toBe("pw-test") + + expect((await inProject(projectA, () => Registry.remove("shared"))).success).toBe(true) + expect(store.get("shared/password")).toBe("pw-test") + } finally { + await Instance.provide({ directory: projectB, fn: () => Instance.dispose() }) + setSecretBackendForTests(undefined) + } + }) + + test("are kept while another served project's local config still names the connection", async () => { + const { setSecretBackendForTests } = await import("../../src/altimate/native/connections/credential-store") + const store = new Map() + setSecretBackendForTests({ + name: "test", + get: async (k: string) => store.get(k) ?? null, + set: async (k: string, v: string) => void store.set(k, v), + delete: async (k: string) => store.delete(k), + } as any) + try { + writeLocal(projectB, { shared: { type: "postgres", host: "b" } }) + expect(inProject(projectB, () => Registry.names())).toEqual(["shared"]) + await inProject(projectA, () => Registry.add("shared", { type: "postgres", host: "g", password: "pw-test" } as any)) + expect(store.get("shared/password")).toBe("pw-test") + + await inProject(projectA, () => Registry.remove("shared")) + expect(store.get("shared/password")).toBe("pw-test") + + // with no served project naming it, removal forgets the secret + fs.rmSync(path.join(projectB, ".altimate-code"), { recursive: true }) + await inProject(projectA, () => Registry.add("shared", { type: "postgres", host: "g", password: "pw-test" } as any)) + await inProject(projectA, () => Registry.remove("shared")) + expect(store.has("shared/password")).toBe(false) + } finally { + setSecretBackendForTests(undefined) + } + }) +}) + +describe("warehouse_add keeps in-memory credentials in its project", () => { + test("credentials added in A stay in A's state; B sees only the saved, sanitized entry", async () => { + // The test preload disables the OS credential store, so the password is held in memory only. + const added = await inProject(projectA, () => + Registry.add("added_pg", { type: "postgres", host: "h", user: "u", password: "s3cret-test" } as any), + ) + expect(added.success).toBe(true) + expect(inProject(projectA, () => Registry.getConfig("added_pg")?.password)).toBe("s3cret-test") + + const inB = inProject(projectB, () => Registry.getConfig("added_pg")) + expect(inB?.host).toBe("h") + expect(inB?.password).toBeUndefined() + + // switching back and forth does not lose A's in-memory credential + expect(inProject(projectA, () => Registry.getConfig("added_pg")?.password)).toBe("s3cret-test") + }) +}) + +describe("test helpers", () => { + test("setConfigs applies to every project until reset", () => { + Registry.setConfigs({ injected: { type: "postgres" } }) + expect(inProject(projectA, () => Registry.names())).toEqual(["injected"]) + expect(inProject(projectB, () => Registry.names())).toEqual(["injected"]) + Registry.reset() + writeLocal(projectA, { only_a: { type: "postgres" } }) + expect(inProject(projectA, () => Registry.names())).toEqual(["only_a"]) + expect(inProject(projectB, () => Registry.names())).toEqual([]) + }) + + test("reset clears every project", () => { + writeLocal(projectA, { only_a: { type: "postgres" } }) + expect(inProject(projectA, () => Registry.names())).toEqual(["only_a"]) + fs.rmSync(path.join(projectA, ".altimate-code"), { recursive: true }) + Registry.reset() + expect(inProject(projectA, () => Registry.names())).toEqual([]) + }) + + test("a connect in flight across reset() is not cached into the dropped state", async () => { + writeLocal(projectA, { store: { type: "sqlite", path: "store.sqlite", create: true } }) + const inflight = inProject(projectA, () => Registry.get("store")) + Registry.reset() + await expect(inflight).rejects.toThrow("reloaded or removed while it was connecting") + }) + + test("closeAll releases connectors cached under setConfigs after load()", async () => { + Registry.setConfigs({ mem: { type: "sqlite", path: ":memory:" } }) + const injected = await Registry.get("mem") + inProject(projectA, () => Registry.load()) + await Registry.closeAll() + await expect(injected.execute("SELECT 1")).rejects.toThrow() + }) +}) diff --git a/packages/opencode/test/altimate/ssh-tunnel.test.ts b/packages/opencode/test/altimate/ssh-tunnel.test.ts index b6ab3e9b6d..8038cf2b93 100644 --- a/packages/opencode/test/altimate/ssh-tunnel.test.ts +++ b/packages/opencode/test/altimate/ssh-tunnel.test.ts @@ -1,5 +1,5 @@ import { describe, test, expect, afterEach } from "bun:test" -import { extractSshConfig, closeTunnel, closeAllTunnels, getActiveTunnel } from "../../src/altimate/native/connections/ssh-tunnel" +import { extractSshConfig, closeAllTunnels, openTunnelCount } from "../../src/altimate/native/connections/ssh-tunnel" afterEach(() => { closeAllTunnels() @@ -71,16 +71,13 @@ describe("extractSshConfig", () => { }) // --------------------------------------------------------------------------- -// closeTunnel / getActiveTunnel — idempotent operations on empty state +// closeAllTunnels — idempotent on empty state // --------------------------------------------------------------------------- describe("SSH tunnel state management", () => { - test("closeTunnel is a no-op for non-existent tunnel", () => { - closeTunnel("nonexistent-tunnel-name") - expect(getActiveTunnel("nonexistent-tunnel-name")).toBeUndefined() - }) - - test("getActiveTunnel returns undefined for non-existent tunnel", () => { - expect(getActiveTunnel("nonexistent")).toBeUndefined() + test("closeAllTunnels is a no-op when nothing is open", () => { + closeAllTunnels() + closeAllTunnels() + expect(openTunnelCount()).toBe(0) }) }) diff --git a/packages/opencode/test/altimate/warehouse-list-hint.test.ts b/packages/opencode/test/altimate/warehouse-list-hint.test.ts index 2128e69a24..9e80598c78 100644 --- a/packages/opencode/test/altimate/warehouse-list-hint.test.ts +++ b/packages/opencode/test/altimate/warehouse-list-hint.test.ts @@ -24,8 +24,13 @@ function toolCtx() { let project: string let prevCwd: string let ambientConnEnv: Record +let ambientGlobalConfig: string | undefined beforeEach(() => { + // Registry.add() in other files persists to the shared test-home global file; set it aside for the empty-registry cases. + const globalFile = Registry.globalConfigPath() + ambientGlobalConfig = fs.existsSync(globalFile) ? fs.readFileSync(globalFile, "utf-8") : undefined + fs.rmSync(globalFile, { force: true }) // Ambient connection env vars would make the empty-registry assertions depend on the caller's shell. ambientConnEnv = {} for (const [key, value] of Object.entries(process.env)) { @@ -45,6 +50,12 @@ afterEach(() => { Registry.reset() fs.rmSync(project, { recursive: true, force: true }) Object.assign(process.env, ambientConnEnv) + const globalFile = Registry.globalConfigPath() + if (ambientGlobalConfig === undefined) fs.rmSync(globalFile, { force: true }) + else { + fs.mkdirSync(path.dirname(globalFile), { recursive: true }) + fs.writeFileSync(globalFile, ambientGlobalConfig) + } }) describe("warehouse_list no-connections hint", () => { @@ -132,6 +143,9 @@ describe("warehouse_list no-connections hint", () => { JSON.stringify({ "proj-b": { type: "postgres" } }), ) process.chdir(other) + // each project has its own registry + expect(Registry.names()).toEqual(["proj-b"]) + process.chdir(project) expect(Registry.names()).toEqual(["proj-a"]) } finally { process.chdir(project) @@ -150,12 +164,11 @@ describe("warehouse_list no-connections hint", () => { JSON.stringify({ "proj-b": { type: "postgres" } }), ) process.chdir(other) - expect(Registry.names()).toEqual([]) - expect(Registry.projectConfigApplies()).toBe(false) - expect(noConnectionsMessage()).not.toContain(Registry.localConfigPath()) - expect(noConnectionsMessage()).toContain(Registry.globalConfigPath()) + expect(Registry.names()).toEqual(["proj-b"]) + expect(noConnectionsMessage()).toContain(Registry.localConfigPath()) process.chdir(dirA) - expect(Registry.projectConfigApplies()).toBe(true) + expect(Registry.names()).toEqual([]) + expect(noConnectionsMessage()).toContain(Registry.localConfigPath()) // the same project still picks up its own file after an empty answer fs.mkdirSync(path.join(dirA, ".altimate-code")) fs.writeFileSync(