From edd1acf0690e541ba80d341e6ca409dbecb9bab3 Mon Sep 17 00:00:00 2001 From: anandgupta42 Date: Mon, 19 Oct 2026 11:55:17 -0700 Subject: [PATCH 1/7] fix: keep warehouse connection registry state per project directory - `registry.ts`: configs, connectors, in-flight connects and the load latch live in a `Map`; global file and env stay shared sources - SSH tunnel keys are project-scoped; stale in-flight connects are not cached after reload/remove/add - `setConfigs()` injects for all projects until `reset()`/`load()` - drop `projectConfigApplies()` Closes #1566 Co-Authored-By: Claude Opus 5.5 --- .../altimate/native/connections/registry.ts | 285 ++++++++++++------ .../altimate/native/connections/ssh-tunnel.ts | 8 +- .../src/altimate/tools/warehouse-list.ts | 6 +- .../altimate/registry-per-project.test.ts | 267 ++++++++++++++++ .../test/altimate/warehouse-list-hint.test.ts | 12 +- 5 files changed, 473 insertions(+), 105 deletions(-) create mode 100644 packages/opencode/test/altimate/registry-per-project.test.ts diff --git a/packages/opencode/src/altimate/native/connections/registry.ts b/packages/opencode/src/altimate/native/connections/registry.ts index 1f6195cc9f..2a481c5b8a 100644 --- a/packages/opencode/src/altimate/native/connections/registry.ts +++ b/packages/opencode/src/altimate/native/connections/registry.ts @@ -36,17 +36,68 @@ 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; scopes SSH tunnel keys. */ + dir: string + /** In-memory config store. */ + configs: Map + /** Cached connector instances. */ + connectors: Map + /** In-flight connector creation promises to prevent race conditions. */ + pending: Map> + /** Whether this state has been loaded from its sources. */ + loaded: boolean +} -/** Cached connector instances. */ -const connectors = new Map() +function createState(dir: string): State { + return { dir, configs: new Map(), connectors: new Map(), pending: new Map(), loaded: false } +} -/** In-flight connector creation promises to prevent race conditions. */ -const pending = new Map>() +/** Registry state per resolved project directory. */ +const states = new Map() -/** Whether the registry has been loaded. */ -let loaded = false +/** + * 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 + +/** 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() +} + +/** 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) { + s = createState(dir) + states.set(dir, s) + } + return s +} + +/** Every state that may hold connectors, for process-wide teardown. */ +function allStates(): State[] { + return injectedState ? [...states.values(), injectedState] : [...states.values()] +} + +/** SSH tunnels are keyed by connection name process-wide; scope the key to the project that owns the connection. */ +function tunnelKey(s: State, name: string): string { + return `${s.dir}\u0000${name}` +} +// altimate_change end const SCRATCH_NAME_CONFLICT = `Connection "${SCRATCH_NAME}" conflicts with the reserved built-in scratch warehouse name. ` + @@ -77,12 +128,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 { @@ -196,20 +242,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(path.join(base, ".altimate-code", "connections.json")), base) const env = resolveStorePaths(loadFromEnv(), base) - // altimate_change end // Merge: global < local < env for (const [name, config] of Object.entries(global)) { @@ -223,39 +267,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 +336,9 @@ export function canonicalType(type: string | undefined | null): string | null { } // altimate_change end -async function createConnector(name: string, config: ConnectionConfig): Promise { +// altimate_change start — `tunnel` is the project-scoped SSH tunnel key +async function createConnector(name: string, config: ConnectionConfig, tunnel: string): 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 +394,14 @@ 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 — project-scoped tunnel key + const tunnelInfo = await startTunnel(tunnel, sshConfig) + // altimate_change end // Rewrite host/port to use the local tunnel resolvedConfig = { ...resolvedConfig, host: "127.0.0.1", - port: tunnel.localPort, + port: tunnelInfo.localPort, } } @@ -420,7 +456,9 @@ async function createConnector(name: string, config: ConnectionConfig): Promise< } catch (e) { // Clean up SSH tunnel if driver creation fails if (sshConfig) { - closeTunnel(name) + // altimate_change start — project-scoped tunnel key + closeTunnel(tunnel) + // altimate_change end } throw e } @@ -528,14 +566,17 @@ 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, pending } = s + const tunnel = tunnelKey(s, name) + // altimate_change end if (isScratch(name)) { if (configs.has(name)) throw new Error(SCRATCH_NAME_CONFLICT) return getScratch() @@ -578,16 +619,30 @@ export async function get(name: string): Promise { }) } catch {} // altimate_change end + // altimate_change start — this attempt's pending slot; reload()/remove()/add() drop it to invalidate the attempt + const attempt: { promise?: Promise } = {} + const current = () => attempt.promise !== undefined && pending.get(name) === attempt.promise + // altimate_change end const promise = (async () => { try { - const connector = await createConnector(name, config) + // altimate_change start — project-scoped tunnel key + const connector = await createConnector(name, config, tunnel) try { await connector.connect() } catch (connectErr) { // If connect() fails after tunnel was started, clean up the tunnel - closeTunnel(name) + closeTunnel(tunnel) 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()) { + await connector.close().catch(() => {}) + // a newer attempt under this name owns the tunnel key now + if (!pending.has(name) && !connectors.has(name)) closeTunnel(tunnel) + throw new Error(`Connection "${name}" was reloaded or removed while it was connecting. Retry the request.`) + } + // altimate_change end connectors.set(name, connector) fileLog("INFO", "warehouse-connect", "connected", { name, type: config.type, auth: authMethod, duration_ms: Date.now() - startTime }) try { @@ -626,10 +681,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 } })() + // altimate_change start + attempt.promise = promise + // altimate_change end pending.set(name, promise) return promise } @@ -639,7 +699,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 +762,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 +807,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, connectors, pending } = s + // altimate_change end // Normalize field names before saving so sensitive fields under alias // names (e.g., keyfileJson → credentials_json) are properly detected @@ -776,6 +844,9 @@ 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) + // altimate_change start — a connect still in flight for the old config must not be cached over the new one + pending.delete(name) + // altimate_change end // Clear cached connector const cached = connectors.get(name) @@ -808,7 +879,16 @@ 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() + // altimate_change start — act on the current project's state, captured before any await + const s = ensureLoaded() + const { configs, connectors, pending } = s + const projectConfigPath = localConfigPath() + // 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. + const inflight = pending.has(name) + // that connect must not be cached once the connection is gone + pending.delete(name) + // altimate_change end // Close connector if cached const cached = connectors.get(name) @@ -822,10 +902,10 @@ export async function remove(name: string): Promise<{ success: boolean; error?: } // Close SSH tunnel if active - closeTunnel(name) - // 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) + // altimate_change start — project-scoped tunnel key + closeTunnel(tunnelKey(s, name)) + if (!inflight) ReconnectLog.forget(name) + // altimate_change end // Remove from global config file const globalPath = globalConfigPath() @@ -839,7 +919,7 @@ export async function remove(name: string): Promise<{ success: boolean; error?: // 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) + const stillDefined = Object.hasOwn(loadFromFile(projectConfigPath), 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 +931,79 @@ 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 { + s.pending.clear() + for (const [name, connector] of s.connectors) { + try { + await connector.close() + } catch { + // ignore + } + closeTunnel(tunnelKey(s, name)) + } + s.connectors.clear() +} + /** - * 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; states captured before the await + const s = state() + const target = projectState() + await closeState(s) + injectionActive = false + loadInto(target) + // 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 + 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..b4bee902b7 100644 --- a/packages/opencode/src/altimate/native/connections/ssh-tunnel.ts +++ b/packages/opencode/src/altimate/native/connections/ssh-tunnel.ts @@ -22,7 +22,10 @@ export interface SshConfig { port: number } -/** Active tunnels keyed by connection name. */ +/** + * Active tunnels keyed by an opaque tunnel key. The connection registry passes a key scoped to the project that + * owns the connection, so two projects with the same connection name never share or close each other's tunnel. + */ const activeTunnels = new Map() /** Clean up all tunnels on process exit. */ @@ -119,7 +122,8 @@ export async function startTunnel( } catch { // ignore } - activeTunnels.delete(name) + // a newer tunnel may already be registered under this key + if (activeTunnels.get(name) === tunnelInfo) activeTunnels.delete(name) }, } 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/test/altimate/registry-per-project.test.ts b/packages/opencode/test/altimate/registry-per-project.test.ts new file mode 100644 index 0000000000..3e11a978d2 --- /dev/null +++ b/packages/opencode/test/altimate/registry-per-project.test.ts @@ -0,0 +1,267 @@ +/** + * 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, mock } from "bun:test" +import * as fs from "fs" +import * as os from "os" +import * as path from "path" + +// ssh2 is an optional dependency that is not installed for tests; a fake client lets the real tunnel module run. +const sshClients: { host: string; ended: boolean }[] = [] +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) + queueMicrotask(() => this.handlers.get("ready")?.()) + } + forwardOut(_a: string, _b: number, _c: string, _d: number, cb: (err?: Error) => void) { + cb(new Error("no forwarding in tests")) + } + end() { + this.record.ended = true + } + }, +})) + +import * as Registry from "../../src/altimate/native/connections/registry" +import { closeAllTunnels } from "../../src/altimate/native/connections/ssh-tunnel" +import { Instance } from "../../src/project/instance" + +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-"))) + sshClients.length = 0 + 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("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("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/warehouse-list-hint.test.ts b/packages/opencode/test/altimate/warehouse-list-hint.test.ts index 2128e69a24..bbfc2d840e 100644 --- a/packages/opencode/test/altimate/warehouse-list-hint.test.ts +++ b/packages/opencode/test/altimate/warehouse-list-hint.test.ts @@ -132,6 +132,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 +153,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( From 2153fc35c8856ff4df1ad664d67429aaddd0e69d Mon Sep 17 00:00:00 2001 From: anandgupta42 Date: Mon, 19 Oct 2026 12:01:52 -0700 Subject: [PATCH 2/7] fix: close the reload and reset windows for in-flight warehouse connects - `reload()` detaches and reloads before awaiting connector closes - `reset()` invalidates pending connects before dropping states Refs #1566 Co-Authored-By: Claude Opus 5.5 --- .../src/altimate/native/connections/registry.ts | 15 ++++++++++----- .../test/altimate/registry-per-project.test.ts | 7 +++++++ 2 files changed, 17 insertions(+), 5 deletions(-) diff --git a/packages/opencode/src/altimate/native/connections/registry.ts b/packages/opencode/src/altimate/native/connections/registry.ts index 2a481c5b8a..7167d8bf9e 100644 --- a/packages/opencode/src/altimate/native/connections/registry.ts +++ b/packages/opencode/src/altimate/native/connections/registry.ts @@ -933,8 +933,11 @@ 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 synchronously, so nothing started during the awaits below is dropped or reuses a closing connector. + const cached = [...s.connectors] + s.connectors.clear() s.pending.clear() - for (const [name, connector] of s.connectors) { + for (const [name, connector] of cached) { try { await connector.close() } catch { @@ -942,7 +945,6 @@ async function closeState(s: State): Promise { } closeTunnel(tunnelKey(s, name)) } - s.connectors.clear() } /** @@ -962,12 +964,14 @@ export async function closeAll(): Promise { /** Reload the current project's configs and clear its cached connectors; other projects are untouched. */ export async function reload(): Promise { - // altimate_change start — only this project's connectors; states captured before the await + // 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() - await closeState(s) + const closing = closeState(s) injectionActive = false loadInto(target) + await closing // altimate_change end } @@ -982,7 +986,8 @@ export function getConfig(name: string): ConnectionConfig | undefined { /** Reset the registry state of every project (for testing). */ export function reset(): void { - // altimate_change start + // altimate_change start — invalidate in-flight connects first, so none is cached into a dropped state + for (const s of allStates()) s.pending.clear() states.clear() injectedState = undefined injectionActive = false diff --git a/packages/opencode/test/altimate/registry-per-project.test.ts b/packages/opencode/test/altimate/registry-per-project.test.ts index 3e11a978d2..e2798aa42c 100644 --- a/packages/opencode/test/altimate/registry-per-project.test.ts +++ b/packages/opencode/test/altimate/registry-per-project.test.ts @@ -257,6 +257,13 @@ describe("test helpers", () => { 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") From aa91c26ae69c1e0278be89f988b97ad8a0156913 Mon Sep 17 00:00:00 2001 From: anandgupta42 Date: Mon, 19 Oct 2026 13:43:00 -0700 Subject: [PATCH 3/7] fix: give each warehouse connect attempt its own SSH tunnel and dispose per project - `ssh-tunnel.ts`: unkeyed `openTunnel()` handles, closable while opening - registry: attempts close only their own tunnel; add/remove/reload/reset detach synchronously - instance disposal closes that project's connectors and tunnels - keep keychain secrets while a served project's local config names the connection Refs #1566 Co-Authored-By: Claude Opus 5.5 --- .../native/connections/credential-store.ts | 4 +- .../altimate/native/connections/registry.ts | 218 +++++++++------- .../altimate/native/connections/ssh-tunnel.ts | 238 +++++++++--------- .../altimate/registry-per-project.test.ts | 160 +++++++++++- .../opencode/test/altimate/ssh-tunnel.test.ts | 15 +- 5 files changed, 410 insertions(+), 225 deletions(-) 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 7167d8bf9e..20c0a86b07 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" @@ -44,25 +47,71 @@ import { SCRATCH_CONFIG, SCRATCH_DESCRIPTION, SCRATCH_NAME, closeScratch, getScr * 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; scopes SSH tunnel keys. */ + /** The project directory this state was loaded for. */ dir: string /** In-memory config store. */ configs: Map /** Cached connector instances. */ connectors: Map - /** In-flight connector creation promises to prevent race conditions. */ - pending: 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(), pending: new Map(), loaded: false } + 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 and tunnels when its instance is disposed (`/instance/dispose`, reload). */ +function ensureDisposer(): void { + if (disposerRegistered) return + disposerRegistered = true + registerDisposer(async (directory) => { + const key = path.resolve(directory) + const s = states.get(key) + if (!s) return + states.delete(key) + 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, @@ -82,6 +131,7 @@ function projectState(): State { const dir = path.resolve(projectRoot()) let s = states.get(dir) if (!s) { + ensureDisposer() s = createState(dir) states.set(dir, s) } @@ -92,11 +142,6 @@ function projectState(): State { function allStates(): State[] { return injectedState ? [...states.values(), injectedState] : [...states.values()] } - -/** SSH tunnels are keyed by connection name process-wide; scope the key to the project that owns the connection. */ -function tunnelKey(s: State, name: string): string { - return `${s.dir}\u0000${name}` -} // altimate_change end const SCRATCH_NAME_CONFLICT = @@ -141,10 +186,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 // --------------------------------------------------------------------------- @@ -252,7 +303,7 @@ function loadInto(s: State): void { // move the store. `s.dir` is the project this state belongs to. const base = s.dir const global = resolveStorePaths(loadFromFile(globalConfigPath()), path.dirname(globalConfigPath())) - const local = resolveStorePaths(loadFromFile(path.join(base, ".altimate-code", "connections.json")), base) + const local = resolveStorePaths(loadFromFile(localConfigPathFor(base)), base) const env = resolveStorePaths(loadFromEnv(), base) // Merge: global < local < env @@ -336,8 +387,16 @@ export function canonicalType(type: string | undefined | null): string | null { } // altimate_change end -// altimate_change start — `tunnel` is the project-scoped SSH tunnel key -async function createConnector(name: string, config: ConnectionConfig, tunnel: string): 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) { @@ -394,8 +453,15 @@ async function createConnector(name: string, config: ConnectionConfig, tunnel: s // Handle SSH tunnel const sshConfig = extractSshConfig(resolvedConfig) if (sshConfig) { - // altimate_change start — project-scoped tunnel key - const tunnelInfo = await startTunnel(tunnel, 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 = { @@ -455,11 +521,9 @@ async function createConnector(name: string, config: ConnectionConfig, tunnel: s connector = await mod.connect(resolvedConfig) } catch (e) { // Clean up SSH tunnel if driver creation fails - if (sshConfig) { - // altimate_change start — project-scoped tunnel key - closeTunnel(tunnel) - // altimate_change end - } + // altimate_change start — this attempt's own tunnel, never another attempt's + attempt.tunnel?.close() + // altimate_change end throw e } return connector @@ -574,8 +638,7 @@ export function names(): string[] { export async function get(name: string): Promise { // altimate_change start — captured once, so a connect that finishes later lands in the project that started it const s = ensureLoaded() - const { configs, connectors, pending } = s - const tunnel = tunnelKey(s, name) + const { configs, connectors, tunnels, pending } = s // altimate_change end if (isScratch(name)) { if (configs.has(name)) throw new Error(SCRATCH_NAME_CONFLICT) @@ -591,8 +654,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) { @@ -619,31 +684,33 @@ export async function get(name: string): Promise { }) } catch {} // altimate_change end - // altimate_change start — this attempt's pending slot; reload()/remove()/add() drop it to invalidate the attempt - const attempt: { promise?: Promise } = {} - const current = () => attempt.promise !== undefined && pending.get(name) === attempt.promise + // 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 { - // altimate_change start — project-scoped tunnel key - const connector = await createConnector(name, config, tunnel) + // 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(tunnel) + // 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(() => {}) - // a newer attempt under this name owns the tunnel key now - if (!pending.has(name) && !connectors.has(name)) closeTunnel(tunnel) - throw new Error(`Connection "${name}" was reloaded or removed while it was connecting. Retry the request.`) + throw SUPERSEDED(name) } - // altimate_change end 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({ @@ -690,7 +757,6 @@ export async function get(name: string): Promise { // altimate_change start attempt.promise = promise // altimate_change end - pending.set(name, promise) return promise } @@ -809,7 +875,7 @@ export async function add( try { // altimate_change start — act on the current project's state, captured before any await const s = ensureLoaded() - const { configs, connectors, pending } = s + const { configs } = s // altimate_change end // Normalize field names before saving so sensitive fields under alias @@ -844,21 +910,13 @@ 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) - // altimate_change start — a connect still in flight for the old config must not be cached over the new one - pending.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 - // Clear cached connector - const cached = connectors.get(name) - if (cached) { - try { - await cached.close() - } catch { - // ignore - } - connectors.delete(name) - } - const result: { success: boolean; name: string; type: string; warnings?: string[] } = { success: true, name, @@ -881,29 +939,17 @@ export async function remove(name: string): Promise<{ success: boolean; error?: try { // altimate_change start — act on the current project's state, captured before any await const s = ensureLoaded() - const { configs, connectors, pending } = s - const projectConfigPath = localConfigPath() // 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. - const inflight = pending.has(name) - // that connect must not be cached once the connection is gone - pending.delete(name) - // altimate_change end - - // 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 - // altimate_change start — project-scoped tunnel key - closeTunnel(tunnelKey(s, name)) + const inflight = s.pending.has(name) + // One synchronous step before any await: the in-flight connect is invalidated (and its tunnel closed), the + // config is gone so no new connect can start against it, and the cached connector and tunnel are detached. + dropAttempt(s, name) + s.configs.delete(name) + const release = detachCached(s, name) + // Every served project's local file, read now: the state map can change during the awaits below. + const projectConfigPaths = [localConfigPath(), ...[...states.keys()].map(localConfigPathFor)] + await release() if (!inflight) ReconnectLog.forget(name) // altimate_change end @@ -913,13 +959,12 @@ export async function remove(name: string): Promise<{ success: boolean; error?: delete existing[name] fs.writeFileSync(globalPath, JSON.stringify(existing, null, 2), "utf-8") - // Remove from in-memory - configs.delete(name) - // 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(projectConfigPath), 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). (Own properties only: a connection named "constructor" must + // not count as defined.) + 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 @@ -933,18 +978,11 @@ 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 synchronously, so nothing started during the awaits below is dropped or reuses a closing connector. - const cached = [...s.connectors] - s.connectors.clear() - s.pending.clear() - for (const [name, connector] of cached) { - try { - await connector.close() - } catch { - // ignore - } - closeTunnel(tunnelKey(s, name)) - } + // 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)) + for (const release of releases) await release() } /** @@ -987,7 +1025,7 @@ export function getConfig(name: string): ConnectionConfig | undefined { /** Reset the registry state of every project (for testing). */ export function reset(): void { // altimate_change start — invalidate in-flight connects first, so none is cached into a dropped state - for (const s of allStates()) s.pending.clear() + for (const s of allStates()) for (const name of [...s.pending.keys()]) dropAttempt(s, name) states.clear() injectedState = undefined injectionActive = false diff --git a/packages/opencode/src/altimate/native/connections/ssh-tunnel.ts b/packages/opencode/src/altimate/native/connections/ssh-tunnel.ts index b4bee902b7..62e5a4b1a6 100644 --- a/packages/opencode/src/altimate/native/connections/ssh-tunnel.ts +++ b/packages/opencode/src/altimate/native/connections/ssh-tunnel.ts @@ -3,15 +3,28 @@ * * 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" 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,165 +35,148 @@ export interface SshConfig { port: number } -/** - * Active tunnels keyed by an opaque tunnel key. The connection registry passes a key scoped to the project that - * owns the connection, so two projects with the same connection name never share or close each other's tunnel. - */ -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) => { + 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) => { + if (err || closed) { localSocket.destroy() return } localSocket.pipe(stream).pipe(localSocket) - }, - ) - }) - - 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 - } - // a newer tunnel may already be registered under this key - if (activeTunnels.get(name) === tunnelInfo) activeTunnels.delete(name) - }, - } - - activeTunnels.set(name, tunnelInfo) - resolve(tunnelInfo) + }) + }) + + 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.on("error", (err: Error) => { - client.end() - reject(new Error(`SSH tunnel local server error: ${err.message}`)) + client.on("error", (err: Error) => { + reject(new Error(`SSH connection error: ${err.message}`)) + 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/test/altimate/registry-per-project.test.ts b/packages/opencode/test/altimate/registry-per-project.test.ts index e2798aa42c..9e9289f1cd 100644 --- a/packages/opencode/test/altimate/registry-per-project.test.ts +++ b/packages/opencode/test/altimate/registry-per-project.test.ts @@ -7,13 +7,38 @@ * process, through `process.chdir` (the CLI path) and through `Instance.provide` (the server path). */ -import { describe, test, expect, beforeEach, afterEach, mock } from "bun:test" +import { describe, test, expect, beforeEach, afterEach, afterAll, mock } from "bun:test" import * as fs from "fs" import * as os from "os" import * as path from "path" -// ssh2 is an optional dependency that is not installed for tests; a fake client lets the real tunnel module run. +// 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>() mock.module("ssh2", () => ({ Client: class { private handlers = new Map void>() @@ -25,7 +50,9 @@ mock.module("ssh2", () => ({ connect(options: { host: string }) { this.record.host = options.host sshClients.push(this.record) - queueMicrotask(() => this.handlers.get("ready")?.()) + const ready = () => this.handlers.get("ready")?.() + 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) => void) { cb(new Error("no forwarding in tests")) @@ -37,8 +64,21 @@ mock.module("ssh2", () => ({ })) import * as Registry from "../../src/altimate/native/connections/registry" -import { closeAllTunnels } from "../../src/altimate/native/connections/ssh-tunnel" +import { 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 @@ -77,6 +117,7 @@ beforeEach(() => { projectA = fs.realpathSync(fs.mkdtempSync(path.join(os.tmpdir(), "rpp-a-"))) projectB = fs.realpathSync(fs.mkdtempSync(path.join(os.tmpdir(), "rpp-b-"))) sshClients.length = 0 + heldReady.clear() Registry.reset() }) @@ -220,6 +261,117 @@ describe("connectors are per project", () => { }) }) +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("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 loads a fresh state + const a2 = await inProject(projectA, () => Registry.get("store")) + expect(a2).not.toBe(a) + expect((await a2.execute("SELECT 1 AS one")).rows).toEqual([[1]]) + }) +}) + +describe("credentials of a removed connection", () => { + 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. 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) }) }) From feb06f3285b15a11390ef76be1ec4f1b4bdc7319 Mon Sep 17 00:00:00 2001 From: anandgupta42 Date: Mon, 19 Oct 2026 13:52:01 -0700 Subject: [PATCH 4/7] fix: close remaining warehouse teardown gaps from review - `remove()` drops the global entry before awaiting cleanup; rewrites only a file holding it - `closeState()` closes all tunnels before awaiting connector closes - a clean SSH disconnect ends the tunnel - credential retention reads served projects after the awaits Refs #1566 Co-Authored-By: Claude Opus 5.5 --- .../altimate/native/connections/registry.ts | 27 ++++++++++--------- .../altimate/native/connections/ssh-tunnel.ts | 6 +++++ .../altimate/registry-per-project.test.ts | 13 +++++++++ 3 files changed, 34 insertions(+), 12 deletions(-) diff --git a/packages/opencode/src/altimate/native/connections/registry.ts b/packages/opencode/src/altimate/native/connections/registry.ts index 20c0a86b07..a55313df57 100644 --- a/packages/opencode/src/altimate/native/connections/registry.ts +++ b/packages/opencode/src/altimate/native/connections/registry.ts @@ -942,27 +942,29 @@ export async function remove(name: string): Promise<{ success: boolean; error?: // 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. const inflight = s.pending.has(name) + const projectConfigPath = localConfigPath() // One synchronous step before any await: the in-flight connect is invalidated (and its tunnel closed), the - // config is gone so no new connect can start against it, and the cached connector and tunnel are detached. + // config is gone from memory and from the global file, so no connect or empty-registry re-read can bring it back, + // and the cached connector and tunnel are detached. dropAttempt(s, name) s.configs.delete(name) const release = detachCached(s, name) - // Every served project's local file, read now: the state map can change during the awaits below. - const projectConfigPaths = [localConfigPath(), ...[...states.keys()].map(localConfigPathFor)] + const globalPath = globalConfigPath() + const existing = loadFromFile(globalPath) + // Only a file that holds the entry is rewritten (a missing global directory is not an error). + if (Object.hasOwn(existing, name)) { + delete existing[name] + fs.writeFileSync(globalPath, JSON.stringify(existing, null, 2), "utf-8") + } await release() if (!inflight) ReconnectLog.forget(name) // altimate_change end - // Remove from global config file - const globalPath = globalConfigPath() - const existing = loadFromFile(globalPath) - delete existing[name] - fs.writeFileSync(globalPath, JSON.stringify(existing, null, 2), "utf-8") - // altimate_change start — remove its secrets from the OS credential store too. Entries are keyed by 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). (Own properties only: a connection named "constructor" must - // not count as defined.) + // 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.) + const projectConfigPaths = [projectConfigPath, ...[...states.keys()].map(localConfigPathFor)] const stillDefined = projectConfigPaths.some((file) => Object.hasOwn(loadFromFile(file), name)) || Object.hasOwn(loadFromEnv(), name) const notRemoved = stillDefined ? [] : await forgetCredentials(name) @@ -982,7 +984,8 @@ async function closeState(s: State): Promise { // 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)) - for (const release of releases) await release() + // Started together: each release closes its tunnel synchronously, so one hanging connector close holds no tunnel. + await Promise.all(releases.map((release) => release())) } /** diff --git a/packages/opencode/src/altimate/native/connections/ssh-tunnel.ts b/packages/opencode/src/altimate/native/connections/ssh-tunnel.ts index 62e5a4b1a6..9d0c9ec72b 100644 --- a/packages/opencode/src/altimate/native/connections/ssh-tunnel.ts +++ b/packages/opencode/src/altimate/native/connections/ssh-tunnel.ts @@ -139,6 +139,12 @@ export function openTunnel(config: SshConfig): TunnelHandle { close() }) + // 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() + }) + const connectOptions: Record = { host: config.ssh_host, port: config.ssh_port ?? 22, diff --git a/packages/opencode/test/altimate/registry-per-project.test.ts b/packages/opencode/test/altimate/registry-per-project.test.ts index 9e9289f1cd..040a3b32f9 100644 --- a/packages/opencode/test/altimate/registry-per-project.test.ts +++ b/packages/opencode/test/altimate/registry-per-project.test.ts @@ -39,6 +39,8 @@ afterAll(() => { 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>() mock.module("ssh2", () => ({ Client: class { private handlers = new Map void>() @@ -51,6 +53,7 @@ mock.module("ssh2", () => ({ 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) } @@ -118,6 +121,7 @@ beforeEach(() => { projectB = fs.realpathSync(fs.mkdtempSync(path.join(os.tmpdir(), "rpp-b-"))) sshClients.length = 0 heldReady.clear() + remoteClose.clear() Registry.reset() }) @@ -311,6 +315,15 @@ describe("each connect attempt owns its SSH tunnel", () => { 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")) From 1c37bc772314d4b5244e9a756ca6dcc4005e49e7 Mon Sep 17 00:00:00 2001 From: anandgupta42 Date: Mon, 19 Oct 2026 14:00:44 -0700 Subject: [PATCH 5/7] fix: keep session warehouse credentials across instance disposal - disposal closes a project's connectors and tunnels but keeps its configs - `remove()` writes the global file before touching state, so a failed write changes nothing - a dropped connect reports the superseded retry error, not its closed tunnel's error Refs #1566 Co-Authored-By: Claude Opus 5.5 --- .../altimate/native/connections/registry.ts | 33 +++++++++-------- .../altimate/registry-per-project.test.ts | 36 ++++++++++++++++++- 2 files changed, 54 insertions(+), 15 deletions(-) diff --git a/packages/opencode/src/altimate/native/connections/registry.ts b/packages/opencode/src/altimate/native/connections/registry.ts index a55313df57..610b5929ba 100644 --- a/packages/opencode/src/altimate/native/connections/registry.ts +++ b/packages/opencode/src/altimate/native/connections/registry.ts @@ -99,16 +99,17 @@ function detachCached(s: State, name: string): () => Promise { } let disposerRegistered = false -/** Close a project's connectors and tunnels when its instance is disposed (`/instance/dispose`, reload). */ +/** + * 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 key = path.resolve(directory) - const s = states.get(key) - if (!s) return - states.delete(key) - await closeState(s) + const s = states.get(path.resolve(directory)) + if (s) await closeState(s) }) } @@ -724,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, @@ -943,19 +948,19 @@ export async function remove(name: string): Promise<{ success: boolean; error?: // later reconnect on that account be put down to another connection that shares it. const inflight = s.pending.has(name) const projectConfigPath = localConfigPath() - // One synchronous step before any await: the in-flight connect is invalidated (and its tunnel closed), the - // config is gone from memory and from the global file, so no connect or empty-registry re-read can bring it back, - // and the cached connector and tunnel are detached. - dropAttempt(s, name) - s.configs.delete(name) - const release = detachCached(s, name) + // 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) - // Only a file that holds the entry is rewritten (a missing global directory is not an error). 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 diff --git a/packages/opencode/test/altimate/registry-per-project.test.ts b/packages/opencode/test/altimate/registry-per-project.test.ts index 040a3b32f9..580ddcee41 100644 --- a/packages/opencode/test/altimate/registry-per-project.test.ts +++ b/packages/opencode/test/altimate/registry-per-project.test.ts @@ -348,11 +348,45 @@ describe("instance disposal", () => { await expect(a.execute("SELECT 1")).rejects.toThrow() expect((await b.execute("SELECT 1 AS one")).rows).toEqual([[1]]) - // reopening A loads a fresh state + // 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", () => { From 62a855dd54905febd99fe2e386051bc9ff8ccd3c Mon Sep 17 00:00:00 2001 From: anandgupta42 Date: Mon, 19 Oct 2026 19:37:58 -0700 Subject: [PATCH 6/7] fix: keep shared credentials for untouched open projects and handle SSH forwarding errors - `remove()` also checks the local connections file of every open instance (`Instance.directories()`), not only projects that already loaded the registry. - SSH forwarding attaches `error` handlers to the local socket and remote stream and destroys both, so a reset no longer throws an uncaught exception. - Tests: `registry-per-project.test.ts` (untouched open project, socket error). Co-Authored-By: Claude Opus 5.5 --- .../altimate/native/connections/registry.ts | 4 +- .../altimate/native/connections/ssh-tunnel.ts | 14 +++- packages/opencode/src/project/instance.ts | 5 ++ .../altimate/registry-per-project.test.ts | 71 +++++++++++++++++-- 4 files changed, 88 insertions(+), 6 deletions(-) diff --git a/packages/opencode/src/altimate/native/connections/registry.ts b/packages/opencode/src/altimate/native/connections/registry.ts index 610b5929ba..5154de671b 100644 --- a/packages/opencode/src/altimate/native/connections/registry.ts +++ b/packages/opencode/src/altimate/native/connections/registry.ts @@ -969,7 +969,9 @@ export async function remove(name: string): Promise<{ success: boolean; error?: // 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.) - const projectConfigPaths = [projectConfigPath, ...[...states.keys()].map(localConfigPathFor)] + // 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) diff --git a/packages/opencode/src/altimate/native/connections/ssh-tunnel.ts b/packages/opencode/src/altimate/native/connections/ssh-tunnel.ts index 9d0c9ec72b..27a4ffecd5 100644 --- a/packages/opencode/src/altimate/native/connections/ssh-tunnel.ts +++ b/packages/opencode/src/altimate/native/connections/ssh-tunnel.ts @@ -12,6 +12,7 @@ import type { ConnectionConfig } from "@altimateai/drivers" import type { Server, Socket } from "net" +import type { Duplex } from "stream" export interface TunnelInfo { localPort: number @@ -106,6 +107,11 @@ export function openTunnel(config: SshConfig): TunnelHandle { 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 @@ -113,7 +119,13 @@ export function openTunnel(config: SshConfig): TunnelHandle { 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) => { - if (err || closed) { + remoteStream = stream + stream?.on("error", () => { + localSocket.destroy() + stream.destroy() + }) + if (err || closed || localSocket.destroyed) { + stream?.destroy() localSocket.destroy() return } 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 index 580ddcee41..f192d63a71 100644 --- a/packages/opencode/test/altimate/registry-per-project.test.ts +++ b/packages/opencode/test/altimate/registry-per-project.test.ts @@ -7,10 +7,13 @@ * process, through `process.chdir` (the CLI path) and through `Instance.provide` (the server path). */ -import { describe, test, expect, beforeEach, afterEach, afterAll, mock } from "bun:test" +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. @@ -41,6 +44,7 @@ const sshClients: { host: string; ended: boolean }[] = [] 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>() @@ -57,8 +61,9 @@ mock.module("ssh2", () => ({ 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) => void) { - cb(new Error("no forwarding in tests")) + 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 @@ -67,7 +72,7 @@ mock.module("ssh2", () => ({ })) import * as Registry from "../../src/altimate/native/connections/registry" -import { closeAllTunnels, openTunnelCount } from "../../src/altimate/native/connections/ssh-tunnel" +import { openTunnel, closeAllTunnels, openTunnelCount } from "../../src/altimate/native/connections/ssh-tunnel" import { Instance } from "../../src/project/instance" import { disposeInstance } from "../../src/effect/instance-registry" @@ -119,6 +124,7 @@ beforeEach(() => { 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() @@ -265,6 +271,40 @@ describe("connectors are per project", () => { }) }) +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") }) @@ -390,6 +430,29 @@ describe("remove()", () => { }) 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() From 9dee23a4c417a2c6a3978162b960a0321d04c751 Mon Sep 17 00:00:00 2001 From: anandgupta42 Date: Mon, 19 Oct 2026 22:11:30 -0700 Subject: [PATCH 7/7] test: isolate `warehouse_list` hint tests from the shared global connections file `Registry.add()` in `telemetry-safety.test.ts` and `adversarial.test.ts` persists `test_add` and `evil` to the test-home global `connections.json`. When CI ran those files first in the same process, the empty-registry cases in `warehouse-list-hint.test.ts` listed them and failed. The hint tests now set that file aside in `beforeEach` and restore it in `afterEach`. Verified by seeding a global file before the suite: 5 failures before, 9/9 pass after. Co-Authored-By: Claude Opus 5.5 --- .../test/altimate/warehouse-list-hint.test.ts | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/packages/opencode/test/altimate/warehouse-list-hint.test.ts b/packages/opencode/test/altimate/warehouse-list-hint.test.ts index bbfc2d840e..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", () => {