Merge pull request #3133 from ruvnet/fix/meta-proxy-stale-owner

fix(proxy): transactionally activate the effective daemon

v3/@claude-flow/cli/__tests__/proxy-activation.test.ts

@@ -0,0 +1,48 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import * as fs from 'node:fs';
+import * as os from 'node:os';
+import * as path from 'node:path';
+
+let stateDir: string;
+let previousState: string | undefined;
+
+beforeEach(() => {
+  stateDir = fs.mkdtempSync(path.join(os.tmpdir(), 'ruflo-proxy-activation-'));
+  previousState = process.env.RUFLO_STATE_DIR;
+  process.env.RUFLO_STATE_DIR = stateDir;
+  vi.resetModules();
+  vi.doMock('../src/funnel/index.js', () => ({ funnelStateDir: () => stateDir }));
+});
+
+afterEach(() => {
+  if (previousState === undefined) delete process.env.RUFLO_STATE_DIR;
+  else process.env.RUFLO_STATE_DIR = previousState;
+  fs.rmSync(stateDir, { recursive: true, force: true });
+  vi.restoreAllMocks();
+});
+
+describe('effective Meta-Proxy activation', () => {
+  it('rejects a non-loopback bind before any version request', async () => {
+    const { probeEffectiveProxy } = await import('../src/proxy/activation.js');
+    fs.writeFileSync(path.join(stateDir, 'proxy-config.toml'), 'bind = "0.0.0.0:11435"\n');
+    let fetched = false;
+    await expect(probeEffectiveProxy(async () => {
+      fetched = true;
+      return new Response('{}');
+    })).rejects.toThrow(/non-loopback/i);
+    expect(fetched).toBe(false);
+  });
+
+  it('serializes two installer callers with the shared lease', async () => {
+    const { acquireProxyInstallLease } = await import('../src/proxy/activation.js');
+    const releaseFirst = await acquireProxyInstallLease();
+    let waited = false;
+    const releaseSecond = await acquireProxyInstallLease(async () => {
+      waited = true;
+      releaseFirst();
+    });
+    expect(waited).toBe(true);
+    expect(fs.existsSync(path.join(stateDir, 'meta-proxy-install.lock'))).toBe(true);
+    releaseSecond();
+  });
+});

v3/@claude-flow/cli/__tests__/proxy-install-command.test.ts

@@ -53,30 +53,32 @@ describe('proxy install - pinned release default', () => {
   });
 
   it('uses the pinned release when confirmed without a release override', async () => {
-    const installProxy = vi.fn();
+    const installAndActivateProxy = vi.fn();
     // Register the mock BEFORE importing proxy-lifecycle.js — that module
     // imports the installer at load time, so importing it first binds the
     // real one and the test performs an actual download.
-    vi.doMock('../src/proxy/install.js', () => ({ installProxy, uninstallProxy: vi.fn() }));
+    vi.doMock('../src/proxy/install.js', () => ({ uninstallProxy: vi.fn() }));
+    vi.doMock('../src/proxy/activation.js', () => ({ installAndActivateProxy }));
 
     const { DEFAULT_PROXY_RELEASE } = await import('../src/commands/proxy-lifecycle.js');
-    installProxy.mockResolvedValue({ version: DEFAULT_PROXY_RELEASE, binaryPath: '/tmp/meta-proxy', sha256: 'abc' });
+    installAndActivateProxy.mockResolvedValue({ version: DEFAULT_PROXY_RELEASE, binaryPath: '/tmp/meta-proxy', sha256: 'abc', pid: 123 });
 
     const installSub = await getInstallSub();
     const result = await installSub.action!(ctxWithFlags({ yes: true }));
 
     expect(result?.success).toBe(true);
-    expect(installProxy).toHaveBeenCalledWith(expect.objectContaining({ version: DEFAULT_PROXY_RELEASE }));
+    expect(installAndActivateProxy).toHaveBeenCalledWith(DEFAULT_PROXY_RELEASE, expect.any(Function));
   });
 
   it('honors an explicit release override', async () => {
-    const installProxy = vi.fn().mockResolvedValue({ version: '9.9.9', binaryPath: '/tmp/meta-proxy', sha256: 'abc' });
-    vi.doMock('../src/proxy/install.js', () => ({ installProxy, uninstallProxy: vi.fn() }));
+    const installAndActivateProxy = vi.fn().mockResolvedValue({ version: '9.9.9', binaryPath: '/tmp/meta-proxy', sha256: 'abc', pid: 123 });
+    vi.doMock('../src/proxy/install.js', () => ({ uninstallProxy: vi.fn() }));
+    vi.doMock('../src/proxy/activation.js', () => ({ installAndActivateProxy }));
 
     const installSub = await getInstallSub();
     const result = await installSub.action!(ctxWithFlags({ release: '9.9.9', yes: true }));
 
     expect(result?.success).toBe(true);
-    expect(installProxy).toHaveBeenCalledWith(expect.objectContaining({ version: '9.9.9' }));
+    expect(installAndActivateProxy).toHaveBeenCalledWith('9.9.9', expect.any(Function));
   });
 });

v3/@claude-flow/cli/src/commands/proxy-lifecycle.ts

@@ -9,7 +9,8 @@
 import type { Command, CommandResult } from '../types.js';
 import { output } from '../output.js';
 import { hasConsent, recordConsent, revokeConsent } from '../funnel/index.js';
-import { installProxy, uninstallProxy } from '../proxy/install.js';
+import { uninstallProxy } from '../proxy/install.js';
+import { installAndActivateProxy } from '../proxy/activation.js';
 import {
   startForeground,
   startBackground,
@@ -159,8 +160,8 @@ const installSub: Command = {
     try {
       const spinner = output.createSpinner({ text: `Installing meta-proxy ${version}...`, spinner: 'dots' });
       spinner.start();
-      const result = await installProxy({ version, log: (line) => spinner.setText(line) });
-      spinner.succeed(`meta-proxy ${version} installed`);
+      const result = await installAndActivateProxy(version, (line) => spinner.setText(line));
+      spinner.succeed(`meta-proxy ${version} installed and verified effective`);
       output.writeln(`  binary: ${result.binaryPath}`);
       output.writeln(`  sha256: ${result.sha256}`);
       return { success: true, data: result };
@@ -196,8 +197,8 @@ const updateSub: Command = {
     try {
       const spinner = output.createSpinner({ text: `Updating meta-proxy to ${version}...`, spinner: 'dots' });
       spinner.start();
-      const result = await installProxy({ version, log: (line) => spinner.setText(line) });
-      spinner.succeed(`meta-proxy updated to ${version}`);
+      const result = await installAndActivateProxy(version, (line) => spinner.setText(line));
+      spinner.succeed(`meta-proxy updated to ${version} and verified effective`);
       output.writeln(`  binary: ${result.binaryPath}`);
       return { success: true, data: result };
     } catch (e) {

v3/@claude-flow/cli/src/proxy/activation.ts

@@ -0,0 +1,190 @@
+/** Transactional activation of the daemon that actually owns the proxy port. */
+import { spawn, spawnSync } from 'node:child_process';
+import * as fs from 'node:fs';
+import { homedir } from 'node:os';
+import * as path from 'node:path';
+import { installProxy, type InstallResult } from './install.js';
+import {
+  isLoopbackBind,
+  proxyBinaryPath,
+  proxyConfigPath,
+  proxyInstallLockPath,
+  proxyInstallManifestPath,
+  proxyLogFilePath,
+  proxyPidFilePath,
+} from './paths.js';
+
+export interface EffectiveProxy { version: string; pid: number; executable: string; }
+type Wait = (milliseconds: number) => Promise<void>;
+const waitNormally: Wait = (milliseconds) => new Promise((resolve) => setTimeout(resolve, milliseconds));
+
+function processExists(pid: number): boolean {
+  try { process.kill(pid, 0); return true; } catch (error) {
+    return (error as NodeJS.ErrnoException).code === 'EPERM';
+  }
+}
+
+function effectiveEndpoint(): string {
+  let bind = '127.0.0.1:11435';
+  try {
+    const match = fs.readFileSync(proxyConfigPath(), 'utf8').match(/^bind\s*=\s*"([^"]+)"\s*$/m);
+    if (match?.[1]) bind = match[1];
+  } catch { /* documented default */ }
+  if (!isLoopbackBind(bind)) throw new Error(`Refusing to probe non-loopback Meta-Proxy bind "${bind}".`);
+  return `http://${bind}`;
+}
+
+function executableFor(pid: number, platform: NodeJS.Platform): string | null {
+  try {
+    if (platform === 'linux') {
+      const result = spawnSync('readlink', ['-f', `/proc/${pid}/exe`], { encoding: 'utf8', timeout: 2_000 });
+      return result.status === 0 ? result.stdout.trim() || null : null;
+    }
+    if (platform === 'win32') {
+      const command = `$p = Get-CimInstance Win32_Process -Filter 'ProcessId = ${pid}'; if ($null -ne $p) { [Console]::Out.Write($p.ExecutablePath) }`;
+      const result = spawnSync('powershell.exe', ['-NoProfile', '-NonInteractive', '-Command', command], { encoding: 'utf8', timeout: 2_000, windowsHide: true });
+      return result.status === 0 ? result.stdout.trim() || null : null;
+    }
+    const result = spawnSync('ps', ['-p', String(pid), '-o', 'comm='], { encoding: 'utf8', timeout: 2_000 });
+    return result.status === 0 ? result.stdout.trim() || null : null;
+  } catch { return null; }
+}
+
+function isSupportedOwner(executable: string, platform: NodeJS.Platform): boolean {
+  const name = platform === 'win32' ? 'meta-proxy.exe' : 'meta-proxy';
+  const normalize = (value: string) => platform === 'win32' ? path.resolve(value).toLowerCase() : path.resolve(value);
+  const allowed = [
+    proxyBinaryPath(),
+    path.join(homedir(), '.metaharness', 'meta-proxy', 'bin', name),
+    path.join(homedir(), '.metaharness', 'bin', name),
+    path.join(homedir(), '.cargo', 'bin', name),
+  ].map(normalize);
+  return allowed.includes(normalize(executable));
+}
+
+export async function probeEffectiveProxy(
+  fetcher: typeof fetch = globalThis.fetch,
+  platform: NodeJS.Platform = process.platform,
+): Promise<EffectiveProxy | null> {
+  const endpoint = effectiveEndpoint();
+  try {
+    const controller = new AbortController();
+    const timer = setTimeout(() => controller.abort(), 1_000);
+    let response: Response;
+    try { response = await fetcher(`${endpoint}/version`, { signal: controller.signal }); }
+    finally { clearTimeout(timer); }
+    if (!response.ok) return null;
+    const body = await response.json() as { version?: unknown; pid?: unknown };
+    if (typeof body.version !== 'string' || !/^\d+\.\d+\.\d+(?:[-+][A-Za-z0-9.-]+)?$/.test(body.version)) return null;
+    if (!Number.isSafeInteger(body.pid) || Number(body.pid) <= 0) return null;
+    const executable = executableFor(Number(body.pid), platform);
+    return executable ? { version: body.version, pid: Number(body.pid), executable } : null;
+  } catch { return null; }
+}
+
+export async function acquireProxyInstallLease(wait: Wait = waitNormally): Promise<() => void> {
+  const lock = proxyInstallLockPath();
+  fs.mkdirSync(path.dirname(lock), { recursive: true, mode: 0o700 });
+  for (let attempt = 0; attempt < 200; attempt++) {
+    try {
+      fs.mkdirSync(lock, { mode: 0o700 });
+      fs.writeFileSync(path.join(lock, 'owner'), `${process.pid}\n`, { mode: 0o600 });
+      return () => fs.rmSync(lock, { recursive: true, force: true });
+    } catch (error) {
+      if ((error as NodeJS.ErrnoException).code !== 'EEXIST') throw error;
+      try {
+        const age = Date.now() - fs.statSync(lock).mtimeMs;
+        const owner = Number.parseInt(fs.readFileSync(path.join(lock, 'owner'), 'utf8').trim(), 10);
+        if (age > 120_000 && (!Number.isSafeInteger(owner) || owner <= 0 || !processExists(owner))) {
+          fs.rmSync(lock, { recursive: true, force: true });
+          continue;
+        }
+      } catch { /* another installer may still be writing its owner */ }
+      await wait(50);
+    }
+  }
+  throw new Error('Another Ruflo/MetaHarness installer still owns the Meta-Proxy install lease.');
+}
+
+async function stopEffective(wait: Wait): Promise<EffectiveProxy | null> {
+  const owner = await probeEffectiveProxy();
+  if (!owner) return null;
+  if (!isSupportedOwner(owner.executable, process.platform)) {
+    throw new Error(`Meta-Proxy port owner pid ${owner.pid} is not a recognized Ruflo/MetaHarness binary; refusing to signal it.`);
+  }
+  process.kill(owner.pid, 'SIGTERM');
+  for (let attempt = 0; attempt < 40; attempt++) {
+    await wait(50);
+    const current = await probeEffectiveProxy();
+    if (!current || current.pid !== owner.pid) return owner;
+  }
+  throw new Error(`Stale Meta-Proxy pid ${owner.pid} did not stop.`);
+}
+
+function launch(binary: string): number {
+  fs.mkdirSync(path.dirname(proxyLogFilePath()), { recursive: true, mode: 0o700 });
+  const log = fs.openSync(proxyLogFilePath(), 'a', 0o600);
+  try {
+    const child = spawn(binary, [], { detached: true, stdio: ['ignore', log, log], windowsHide: true });
+    if (!child.pid) throw new Error('Meta-Proxy did not return a process id.');
+    child.unref();
+    fs.writeFileSync(proxyPidFilePath(), `${child.pid}\n`, { mode: 0o600 });
+    return child.pid;
+  } finally { fs.closeSync(log); }
+}
+
+async function launchAndVerify(binary: string, version: string, wait: Wait): Promise<EffectiveProxy> {
+  const pid = launch(binary);
+  for (let attempt = 0; attempt < 100; attempt++) {
+    await wait(50);
+    const current = await probeEffectiveProxy();
+    if (current?.pid === pid && current.version === version && path.resolve(current.executable) === path.resolve(binary)) return current;
+    if (current && current.pid !== pid) {
+      try { process.kill(pid, 'SIGTERM'); } catch { /* already exited */ }
+      throw new Error(`Competing Meta-Proxy pid ${current.pid} won the port with version ${current.version}.`);
+    }
+  }
+  try { process.kill(pid, 'SIGTERM'); } catch { /* already exited */ }
+  throw new Error(`Meta-Proxy v${version} did not become the effective daemon.`);
+}
+
+export async function installAndActivateProxy(version: string, log?: (line: string) => void): Promise<InstallResult & { pid: number }> {
+  const wait = waitNormally;
+  const binary = proxyBinaryPath();
+  const manifest = proxyInstallManifestPath();
+  const binaryBackup = `${binary}.rollback`;
+  const manifestBackup = `${manifest}.rollback`;
+  let release: (() => void) | null = null;
+  let prior: EffectiveProxy | null = null;
+  try {
+    release = await acquireProxyInstallLease(wait);
+    prior = await stopEffective(wait);
+    fs.rmSync(binaryBackup, { force: true });
+    fs.rmSync(manifestBackup, { force: true });
+    if (fs.existsSync(binary)) fs.copyFileSync(binary, binaryBackup);
+    if (fs.existsSync(manifest)) fs.copyFileSync(manifest, manifestBackup);
+    const installed = await installProxy({ version, log });
+    const effective = await launchAndVerify(installed.binaryPath, installed.version, wait);
+    return { ...installed, pid: effective.pid };
+  } catch (error) {
+    const failure = error instanceof Error ? error.message : String(error);
+    if (fs.existsSync(binaryBackup)) {
+      try {
+        await stopEffective(wait);
+        fs.rmSync(binary, { force: true });
+        fs.renameSync(binaryBackup, binary);
+        fs.rmSync(manifest, { force: true });
+        if (fs.existsSync(manifestBackup)) fs.renameSync(manifestBackup, manifest);
+        if (prior) await launchAndVerify(prior.executable === binary ? binary : prior.executable, prior.version, wait);
+      } catch (rollbackError) {
+        throw new Error(`${failure} Rollback failed: ${rollbackError instanceof Error ? rollbackError.message : String(rollbackError)}`);
+      }
+      throw new Error(`${failure} Previous Meta-Proxy ${prior ? 'was restored and verified' : 'binary was restored; it was not running before the upgrade'}.`);
+    }
+    throw error;
+  } finally {
+    fs.rmSync(binaryBackup, { force: true });
+    fs.rmSync(manifestBackup, { force: true });
+    release?.();
+  }
+}

v3/@claude-flow/cli/src/proxy/install.ts

@@ -170,6 +170,9 @@ export async function installProxy(opts: InstallOptions): Promise<InstallResult>
     const tmp = `${finalPath}.tmp`;
     fs.copyFileSync(extractedBinaryPath, tmp);
     fs.chmodSync(tmp, 0o755);
+    // The activation transaction has already stopped and backed up the old
+    // daemon. Windows rename does not replace an existing executable.
+    fs.rmSync(finalPath, { force: true });
     fs.renameSync(tmp, finalPath);
 
     const liveSha = sha256Hex(fs.readFileSync(finalPath));

v3/@claude-flow/cli/src/proxy/paths.ts

@@ -31,6 +31,11 @@ export function proxyInstallManifestPath(): string {
   return join(funnelStateDir(), 'proxy', 'install-manifest.json');
 }
 
+/** Shared by Ruflo and MetaHarness so two IDE windows cannot race an upgrade. */
+export function proxyInstallLockPath(): string {
+  return join(funnelStateDir(), 'meta-proxy-install.lock');
+}
+
 export function proxyConfigPath(): string {
   return join(funnelStateDir(), 'proxy-config.toml');
 }

v3/pnpm-lock.yaml

@@ -121,7 +121,7 @@ importers:
         specifier: 3.0.3
         version: link:../codex
       '@claude-flow/mcp':
-        specifier: ^3.0.0-alpha.9
+        specifier: 3.0.0-alpha.10
         version: link:../mcp
       '@claude-flow/neural':
         specifier: 3.0.0-alpha.9