diff --git a/packages/web/server/index.js b/packages/web/server/index.js index 1c386b3c..4eac3336 100644 --- a/packages/web/server/index.js +++ b/packages/web/server/index.js @@ -1079,6 +1079,19 @@ const openCodeLifecycleRuntime = createOpenCodeLifecycleRuntime({ } return [...new Set(directories)]; }, + // A managed restart can move OpenCode to a NEW port (the old one may stay + // occupied by an orphaned process, e.g. killProcessOnPort is a no-op on + // Windows). Rebind the message-stream upstream readers to the current port + // so the UI keeps receiving events instead of staying pinned to the old + // process (#2638). The runtime is created later by the startup pipeline; + // by the time any restart runs, it is assigned. + onOpenCodeRestarted: () => { + try { + messageStreamRuntime?.rebindUpstream(); + } catch (error) { + console.warn('Failed to rebind message stream after OpenCode restart:', error?.message ?? error); + } + }, getManagedOpenCodeEnv: async () => { const settings = await readSettingsFromDiskMigrated().catch(() => null); const managedEnv = settings?.agentControlToolEnabled === false diff --git a/packages/web/server/lib/event-stream/rebind.test.js b/packages/web/server/lib/event-stream/rebind.test.js new file mode 100644 index 00000000..ebf821f2 --- /dev/null +++ b/packages/web/server/lib/event-stream/rebind.test.js @@ -0,0 +1,186 @@ +import { EventEmitter } from 'node:events'; +import { describe, expect, it, vi } from 'vitest'; + +import { createGlobalMessageStreamHub } from './global-hub.js'; +import { createMessageStreamWsRuntime } from './runtime.js'; + +class FakeSocket extends EventEmitter { + constructor() { + super(); + this.readyState = 1; + this.sent = []; + this.closeCalls = []; + } + + send(payload) { + this.sent.push(JSON.parse(payload)); + } + + ping() { + void 0; + } + + close(code, reason) { + if (this.readyState === 3) { + return; + } + this.readyState = 3; + this.closeCalls.push({ code, reason }); + this.emit('close'); + } +} + +function createSseResponse({ blocks = [], signal, holdOpen = false }) { + const encoder = new TextEncoder(); + let index = 0; + + return { + ok: true, + body: { + getReader() { + return { + async read() { + if (index < blocks.length) { + const next = blocks[index++]; + return { value: encoder.encode(next), done: false }; + } + + if (!holdOpen) { + return { value: undefined, done: true }; + } + + return new Promise((resolve, reject) => { + const onAbort = () => { + signal.removeEventListener('abort', onAbort); + const error = new Error('Aborted'); + error.name = 'AbortError'; + reject(error); + }; + signal.addEventListener('abort', onAbort, { once: true }); + }); + }, + }; + }, + }, + }; +} + +describe('rebindUpstream (#2638)', () => { + it('restarts the shared hub upstream so a connected client resumes receiving events on the new port', async () => { + const server = new EventEmitter(); + const wsClients = new Set(); + let port = 4096; + let fetchCalls = 0; + + // Port changes after a managed restart: buildOpenCodeUrl resolves the + // CURRENT port on every attempt, exactly like production network-runtime. + const buildOpenCodeUrl = vi.fn(() => `http://127.0.0.1:${port}/global/event`); + const fetchImpl = vi.fn(async (_url, options) => { + fetchCalls += 1; + if (fetchCalls === 1) { + return createSseResponse({ + signal: options.signal, + holdOpen: true, + blocks: ['id: evt-1\ndata: {"type":"server.connected","properties":{}}\n\n'], + }); + } + return createSseResponse({ + signal: options.signal, + holdOpen: true, + blocks: ['id: evt-2\ndata: {"type":"session.updated","properties":{"sessionID":"ses_1"}}\n\n'], + }); + }); + + const globalHub = createGlobalMessageStreamHub({ + buildOpenCodeUrl, + getOpenCodeAuthHeaders: () => ({}), + fetchImpl, + upstreamReconnectDelayMs: 0, + }); + + const runtime = createMessageStreamWsRuntime({ + server, + uiAuthController: null, + isRequestOriginAllowed: async () => true, + rejectWebSocketUpgrade() { + throw new Error('upgrade should not be used in this test'); + }, + globalEventHub: globalHub, + buildOpenCodeUrl, + getOpenCodeAuthHeaders: () => ({}), + processForwardedEventPayload() {}, + wsClients, + heartbeatIntervalMs: 5000, + upstreamReconnectDelayMs: 0, + fetchImpl, + }); + + const socket = new FakeSocket(); + runtime.wsServer.emit('connection', socket, { url: '/api/global/event/ws' }); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(fetchCalls).toBe(1); + expect(socket.sent.some((frame) => frame.type === 'event' && frame.eventId === 'evt-1')).toBe(true); + + // The managed process was restarted onto a new port while the old + // process's SSE stream stays open (orphaned survivor). + port = 5000; + runtime.rebindUpstream(); + await new Promise((resolve) => setTimeout(resolve, 20)); + + // The hub dialed the new port and the connected client received events + // from the new upstream without reconnecting its own socket. + expect(fetchCalls).toBe(2); + expect(fetchImpl.mock.calls[1][0]).toContain(':5000/global/event'); + expect(socket.sent.some((frame) => frame.type === 'event' && frame.eventId === 'evt-2')).toBe(true); + + socket.close(); + await runtime.close(); + }); + + it('closes directory-scoped sockets so their pinned readers reconnect to the new port', async () => { + const server = new EventEmitter(); + const wsClients = new Set(); + let port = 4096; + let fetchCalls = 0; + + const buildOpenCodeUrl = vi.fn(() => `http://127.0.0.1:${port}/event`); + const fetchImpl = vi.fn(async (_url, options) => { + fetchCalls += 1; + return createSseResponse({ + signal: options.signal, + holdOpen: true, + blocks: ['id: evt-1\ndata: {"type":"server.connected","properties":{}}\n\n'], + }); + }); + + const runtime = createMessageStreamWsRuntime({ + server, + uiAuthController: null, + isRequestOriginAllowed: async () => true, + rejectWebSocketUpgrade() { + throw new Error('upgrade should not be used in this test'); + }, + buildOpenCodeUrl, + getOpenCodeAuthHeaders: () => ({}), + processForwardedEventPayload() {}, + wsClients, + heartbeatIntervalMs: 5000, + upstreamReconnectDelayMs: 0, + fetchImpl, + }); + + const directorySocket = new FakeSocket(); + runtime.wsServer.emit('connection', directorySocket, { url: '/api/event/ws?directory=%2Fproj' }); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(fetchCalls).toBe(1); + + port = 5000; + runtime.rebindUpstream(); + + expect(directorySocket.readyState).toBe(3); + expect(directorySocket.closeCalls.length).toBeGreaterThan(0); + + directorySocket.close(); + await runtime.close(); + }); +}); diff --git a/packages/web/server/lib/event-stream/runtime.js b/packages/web/server/lib/event-stream/runtime.js index 72a28893..07b025dc 100644 --- a/packages/web/server/lib/event-stream/runtime.js +++ b/packages/web/server/lib/event-stream/runtime.js @@ -70,6 +70,12 @@ export function createMessageStreamWsRuntime({ noServer: true, }); + // Directory-scoped streams create one upstream reader per client + // connection. Track those sockets so a managed OpenCode restart can close + // them: each reader is pinned to the port it connected at and would + // otherwise keep streaming from an orphaned process on the old port (#2638). + const directorySockets = new Set(); + const ownsGlobalHub = !globalEventHub; const globalHub = globalEventHub ?? createGlobalMessageStreamHub({ buildOpenCodeUrl, @@ -103,6 +109,11 @@ export function createMessageStreamWsRuntime({ return; } + directorySockets.add(socket); + socket.on('close', () => { + directorySockets.delete(socket); + }); + acceptDirectoryMessageStreamWsConnection({ socket, requestedLastEventId, @@ -156,6 +167,27 @@ export function createMessageStreamWsRuntime({ return { wsServer, + /** + * Rebind all upstream readers to the current OpenCode port. Called after + * a managed process restart: the restart can land on a NEW port while + * the old process (or an orphaned survivor of it) still holds the + * previous one, and a healthy-but-pinned SSE connection never notices — + * so the UI would stop receiving events until the app restarts (#2638). + * Restarting the shared hub re-dials `buildOpenCodeUrl` (which reads the + * current port) on its next attempt; directory-scoped readers are + * rebuilt by closing their client sockets, which reconnect with + * `Last-Event-ID` and re-establish the stream against the new port. + */ + rebindUpstream() { + globalHub.stop(); + globalHub.start(); + for (const socket of Array.from(directorySockets)) { + try { + socket.close(1012, 'OpenCode upstream restarted'); + } catch { + } + } + }, async close() { server.off('upgrade', upgradeHandler); globalBridge.close(); diff --git a/packages/web/server/lib/opencode/lifecycle.js b/packages/web/server/lib/opencode/lifecycle.js index 172d88b2..bb13aab0 100644 --- a/packages/web/server/lib/opencode/lifecycle.js +++ b/packages/web/server/lib/opencode/lifecycle.js @@ -48,6 +48,7 @@ export const createOpenCodeLifecycleRuntime = (deps) => { getActiveSessionCount = () => 0, reapManagedOrphanedProcesses = reapOrphanedProcesses, getWarmupDirectories = async () => [], + onOpenCodeRestarted = null, now = Date.now, } = deps; @@ -694,6 +695,17 @@ export const createOpenCodeLifecycleRuntime = (deps) => { setupProxy(state.expressApp); ensureOpenCodeApiPrefix(); } + + // The restart may have landed on a NEW port (the old one can remain + // occupied by an orphaned process, e.g. Windows killProcessOnPort is a + // no-op). Upstream event readers pinned to the old process would keep + // the UI silent forever, so rebind them to the current port. Best + // effort: a failure here must not fail the restart itself. + try { + onOpenCodeRestarted?.(); + } catch (error) { + console.warn('Failed to rebind event stream after OpenCode restart:', error?.message ?? error); + } })(); try { diff --git a/packages/web/server/lib/opencode/lifecycle.test.js b/packages/web/server/lib/opencode/lifecycle.test.js index fb57c3f2..04bf4c33 100644 --- a/packages/web/server/lib/opencode/lifecycle.test.js +++ b/packages/web/server/lib/opencode/lifecycle.test.js @@ -286,6 +286,70 @@ describe('OpenCode lifecycle', () => { expect(spawnMock).toHaveBeenCalledTimes(1); }); + it('calls onOpenCodeRestarted after a successful managed restart', async () => { + const close = vi.fn(async () => {}); + const replacement = createMockChild(); + const onOpenCodeRestarted = vi.fn(); + globalThis.fetch = vi.fn(async () => ({ + ok: false, + json: async () => null, + })); + spawnMock.mockImplementationOnce(() => { + queueMicrotask(() => { + replacement.stdout.emit('data', 'opencode server listening on http://127.0.0.1:45678\n'); + }); + return replacement; + }); + const runtime = createRuntime({ onOpenCodeRestarted }, { + openCodePort: 45678, + openCodeProcess: { + pid: null, + exitCode: 1, + signalCode: null, + close, + }, + }); + + await runtime.triggerHealthCheck(); + + expect(close).toHaveBeenCalledTimes(1); + expect(spawnMock).toHaveBeenCalledTimes(1); + // The restart completed on a (possibly new) port — the event-stream + // upstreams must rebind so the UI keeps receiving events (#2638). + expect(onOpenCodeRestarted).toHaveBeenCalledTimes(1); + }); + + it('does not call onOpenCodeRestarted when a managed restart fails', async () => { + const close = vi.fn(async () => {}); + const onOpenCodeRestarted = vi.fn(); + globalThis.fetch = vi.fn(async () => ({ + ok: false, + json: async () => null, + })); + spawnMock.mockImplementation(() => { + const child = createMockChild(); + queueMicrotask(() => { + child.emit('error', new Error('spawn failed')); + }); + return child; + }); + const runtime = createRuntime({ onOpenCodeRestarted }, { + openCodePort: 45678, + openCodeProcess: { + pid: null, + exitCode: 1, + signalCode: null, + close, + }, + }); + + // triggerHealthCheck logs instead of rethrowing; call restartOpenCode + // directly to observe the failure result. + await expect(runtime.restartOpenCode()).rejects.toThrow(); + + expect(onOpenCodeRestarted).not.toHaveBeenCalled(); + }); + it('launches managed OpenCode with the managed PATH', async () => { delete process.env.OPENCODE_BINARY; const child = createMockChild(); diff --git a/scripts/repro/issue-2638/README.md b/scripts/repro/issue-2638/README.md new file mode 100644 index 00000000..357ee8af --- /dev/null +++ b/scripts/repro/issue-2638/README.md @@ -0,0 +1,47 @@ +# Reproduction: Chat UI stops updating until the desktop app is restarted (#2638) + +Reproduces https://github.com/openchamber/openchamber/issues/2638 using the +real server modules (`lifecycle.js`, `global-hub.js`, `network-runtime.js`), +real processes, and real ports — no mocks. + +## Run + +```sh +# Windows-orphan scenario (the reported bug): +node scripts/repro/issue-2638/reproduce-2638.mjs + +# Control: healthy restart on Linux (hub reconnects, UI keeps updating): +node scripts/repro/issue-2638/reproduce-2638.mjs --baseline +``` + +Requires `lsof` (used only for cleanup). The default run simulates Windows +(`process.platform` is temporarily overridden to `win32`) because the bug is +specific to the Windows process-lifecycle path. + +## What it demonstrates + +1. A managed OpenCode process starts; the global message-stream hub connects to + its `/global/event` SSE stream and chat events flow to the UI. +2. The managed process "exits" but the actual server process survives on the + old port (on Windows `killProcessOnPort` is a no-op and `taskkill` cannot + reach the orphaned tree — the report shows leftover `opencode.exe serve` + processes on historical ports). +3. `restartOpenCode()` gives up after 5 s — logs + `Timed out waiting for OpenCode port to be released` — and spawns a + fresh server on a NEW port, leaving the orphaned process running. +4. HTTP/proxy traffic follows `state.openCodePort` to the new server, but the + hub's upstream SSE reader stays pinned to the OLD server's `/global/event` + stream (that connection never closed), so events emitted by the new server + never reach the UI: the chat UI goes stale while the new server keeps + persisting session data — visible only after restarting the app. + +The `--baseline` control proves the reconnect logic itself is fine: when the +old process dies and the port is properly released, the hub reconnects to the +new port and events are delivered. + +## Files + +- `reproduce-2638.mjs` — the reproduction driver (assertions + summary). +- `fake-opencode-serve.mjs` — a fake `opencode serve` binary whose launcher + spawns a detached server core that survives the launcher's death + (Windows-style orphan), plus an in-process mode for the baseline control. diff --git a/scripts/repro/issue-2638/fake-opencode-serve.mjs b/scripts/repro/issue-2638/fake-opencode-serve.mjs new file mode 100755 index 00000000..0fc36dc0 --- /dev/null +++ b/scripts/repro/issue-2638/fake-opencode-serve.mjs @@ -0,0 +1,151 @@ +#!/usr/bin/env node +// Fake `opencode serve` used to reproduce https://github.com/openchamber/openchamber/issues/2638 +// +// Modes (controlled by env): +// FAKE_OPENCODE_CORE=1 – server-core mode: binds the port, serves +// /global/health + SSE /global/event, ignores +// SIGTERM so it survives its launcher's death +// (Windows-style orphaned server process). +// FAKE_OPENCODE_BASELINE=1 – in-process mode: the server runs inside the +// managed process and dies with it (normal +// Linux behavior used as a control). +// default (launcher) – spawns a detached core grandchild, waits for +// it to bind, prints the `opencode server +// listening on ...` line the lifecycle greps +// for, stays alive, and on SIGTERM exits WITHOUT +// killing the core (mimics opencode.exe dying +// while its server child survives). +import http from 'node:http'; +import net from 'node:net'; +import fs from 'node:fs'; +import path from 'node:path'; +import { spawn } from 'node:child_process'; + +const args = process.argv.slice(2); +const portIndex = args.indexOf('--port'); +const hostnameIndex = args.indexOf('--hostname'); +const port = portIndex >= 0 ? Number(args[portIndex + 1]) : 0; +const hostname = hostnameIndex >= 0 ? args[hostnameIndex + 1] : '127.0.0.1'; +const pidDir = process.env.FAKE_OPENCODE_PID_DIR || null; + +function writePidFile(label) { + if (!pidDir) return; + try { + fs.mkdirSync(pidDir, { recursive: true }); + fs.writeFileSync(path.join(pidDir, `${label}-${port}.pid`), String(process.pid)); + } catch { + // best effort + } +} + +function createServer() { + const clients = new Set(); + const emitted = []; + const server = http.createServer((req, res) => { + const url = new URL(req.url, `http://${hostname}:${port}`); + if (url.pathname === '/global/health') { + res.writeHead(200, { 'content-type': 'application/json' }); + res.end(JSON.stringify({ healthy: true })); + return; + } + if (url.pathname === '/global/event') { + res.writeHead(200, { + 'content-type': 'text/event-stream', + 'cache-control': 'no-cache', + connection: 'keep-alive', + }); + clients.add(res); + req.on('close', () => clients.delete(res)); + // SSE keep-alive comments — exactly what a real OpenCode server sends, + // which prevents the upstream reader's 20s stall timer from firing. + const keepalive = setInterval(() => { + res.write(': keepalive\n\n'); + }, 1000); + req.on('close', () => clearInterval(keepalive)); + return; + } + if (url.pathname === '/emit') { + const type = url.searchParams.get('type') || 'session.updated'; + const id = url.searchParams.get('id') || `evt-${Date.now()}`; + emitted.push({ id, type }); + const block = `id: ${id}\ndata: ${JSON.stringify({ type, id })}\n\n`; + for (const client of clients) client.write(block); + res.writeHead(200, { 'content-type': 'application/json' }); + res.end(JSON.stringify({ ok: true, id })); + return; + } + if (url.pathname === '/events') { + res.writeHead(200, { 'content-type': 'application/json' }); + res.end(JSON.stringify(emitted)); + return; + } + res.writeHead(404); + res.end('not found'); + }); + return { server }; +} + +const runServer = (label) => { + createServer().server.listen(port, hostname, () => { + console.log(`opencode server listening on http://${hostname}:${port}`); + }); + writePidFile(label); + setInterval(() => {}, 1 << 30); +}; + +const main = async () => { + // Server-core mode: the orphaned server. Survives its launcher's death. + if (process.env.FAKE_OPENCODE_CORE === '1') { + runServer('core'); + process.on('SIGTERM', () => {}); + process.on('SIGINT', () => {}); + return; + } + + // Baseline in-process mode (control): dies with the managed process, so the + // port is properly released on restart. + if (process.env.FAKE_OPENCODE_BASELINE === '1') { + runServer('baseline'); + return; + } + + // Launcher mode: spawn a detached core grandchild, wait for it to bind, + // print the listening line, and on SIGTERM exit leaving the core running. + const core = spawn(process.execPath, [process.argv[1], ...args], { + detached: true, + stdio: ['ignore', 'ignore', 'ignore'], + env: { ...process.env, FAKE_OPENCODE_CORE: '1' }, + }); + core.unref(); + + for (let i = 0; i < 200; i += 1) { + if (core.exitCode !== null) throw new Error('core exited early'); + const ok = await new Promise((resolve) => { + const socket = net.connect({ port, host: hostname }); + const timer = setTimeout(() => { + socket.destroy(); + resolve(false); + }, 200); + socket.once('connect', () => { + clearTimeout(timer); + socket.destroy(); + resolve(true); + }); + socket.once('error', () => { + clearTimeout(timer); + resolve(false); + }); + }); + if (ok) break; + await new Promise((r) => setTimeout(r, 50)); + } + + writePidFile('launcher'); + console.log(`opencode server listening on http://${hostname}:${port}`); + // On SIGTERM exit ourselves, leaving the detached core running. + process.on('SIGTERM', () => process.exit(0)); + process.on('SIGINT', () => process.exit(0)); + setInterval(() => {}, 1 << 30); +}; + +await main(); diff --git a/scripts/repro/issue-2638/reproduce-2638.mjs b/scripts/repro/issue-2638/reproduce-2638.mjs new file mode 100644 index 00000000..d4f99af4 --- /dev/null +++ b/scripts/repro/issue-2638/reproduce-2638.mjs @@ -0,0 +1,376 @@ +// Reproduction for https://github.com/openchamber/openchamber/issues/2638 +// "[Bug] Chat UI stops updating until the desktop app is restarted" +// +// Run with: node reproduce-2638.mjs (Windows-orphan scenario) +// node reproduce-2638.mjs --baseline (control: healthy restart) +// +// What it wires up (real repo modules, real processes, real ports): +// - createOpenCodeLifecycleRuntime (packages/web/server/lib/opencode/lifecycle.js) +// - createGlobalMessageStreamHub (packages/web/server/lib/event-stream/global-hub.js) +// - createOpenCodeNetworkRuntime (packages/web/server/lib/opencode/network-runtime.js) +// - a fake `opencode serve` binary (fake-opencode-serve.mjs) +// +// Scenario (issue #2638): +// 1. OpenCode starts; the global message-stream hub connects to its +// /global/event SSE stream. Chat UI updates flow (baseline event e1 +// reaches the hub). +// 2. The managed OpenCode process "exits" but the actual server process +// survives on the old port (on Windows killProcessOnPort is a no-op and +// taskkill cannot reach the orphaned tree — the report shows leftover +// `opencode.exe serve` processes on historical ports). +// 3. restartOpenCode() gives up after 5 s +// ("Timed out waiting for OpenCode port to be released") and +// spawns a fresh server on a NEW port. +// 4. HTTP/proxy traffic follows state.openCodePort to the NEW server, but +// the hub's upstream SSE reader is still pinned to the OLD server's +// /global/event stream (that connection never closed), so events from +// the new server never reach the UI. Chat UI goes stale while the new +// server keeps persisting session data — visible only after restarting +// the app, exactly as reported. + +import { spawnSync } from 'node:child_process'; +import net from 'node:net'; +import fs from 'node:fs'; +import path from 'node:path'; +import os from 'node:os'; +import { fileURLToPath } from 'node:url'; + +const __dirname = path.dirname(fileURLToPath(import.meta.url)); + +// Repository root (override with REPO=/path/to/openchamber if needed). Default: +// walk up from this script until we find the repo root (AGENTS.md + package.json). +const findRepoRoot = () => { + let dir = __dirname; + for (let i = 0; i < 8; i += 1) { + if (fs.existsSync(path.join(dir, 'AGENTS.md')) && fs.existsSync(path.join(dir, 'package.json'))) { + return dir; + } + const parent = path.dirname(dir); + if (parent === dir) break; + dir = parent; + } + return '/home/runner/work/openchamber/openchamber'; +}; +const REPO = process.env.REPO || findRepoRoot(); + +const BASELINE = process.argv.includes('--baseline'); + +// Simulate the Windows behavior reported in #2638 (process.platform is read +// at call time inside lifecycle.js: killProcessOnPort no-ops on win32 and +// terminateChildProcess takes the taskkill path, which cannot exist here). +if (!BASELINE) { + Object.defineProperty(process, 'platform', { value: 'win32' }); +} + +const { createOpenCodeLifecycleRuntime } = await import( + path.join(REPO, 'packages/web/server/lib/opencode/lifecycle.js') +); + +// --- shared state + real network runtime ----------------------------------- +const state = { + openCodeWorkingDirectory: '/tmp', + openCodeProcess: null, + openCodePort: null, + openCodeBaseUrl: null, + currentRestartPromise: null, + isRestartingOpenCode: false, + openCodeApiPrefix: '', + openCodeApiPrefixDetected: false, + openCodeApiDetectionTimer: null, + lastOpenCodeError: null, + isOpenCodeReady: false, + openCodeNotReadySince: 0, + isExternalOpenCode: false, + isShuttingDown: false, + healthCheckInterval: null, + expressApp: null, + useWslForOpencode: false, + resolvedWslBinary: null, + resolvedWslOpencodePath: null, + resolvedWslDistro: null, + lastOpenCodeLaunchDiagnostics: null, +}; + +const { createOpenCodeNetworkRuntime } = await import( + path.join(REPO, 'packages/web/server/lib/opencode/network-runtime.js') +); +const networkRuntime = createOpenCodeNetworkRuntime({ + state, + getOpenCodeAuthHeaders: () => ({}), + configuredOpenCodeHostname: '127.0.0.1', +}); + +const fakeBinary = path.join(__dirname, 'fake-opencode-serve.mjs'); +const pidDir = fs.mkdtempSync(path.join(os.tmpdir(), 'oc-repro-2638-')); +process.env.OPENCODE_BINARY = fakeBinary; +process.env.FAKE_OPENCODE_PID_DIR = pidDir; +if (BASELINE) process.env.FAKE_OPENCODE_BASELINE = '1'; + +const lifecycle = createOpenCodeLifecycleRuntime({ + state, + env: { + ENV_CONFIGURED_OPENCODE_PORT: 0, + ENV_CONFIGURED_OPENCODE_HOST: null, + ENV_EFFECTIVE_PORT: 0, + ENV_CONFIGURED_OPENCODE_HOSTNAME: '127.0.0.1', + ENV_SKIP_OPENCODE_START: false, + }, + syncToHmrState: () => {}, + syncFromHmrState: () => {}, + getOpenCodeAuthHeaders: () => ({}), + buildOpenCodeUrl: (...args) => networkRuntime.buildOpenCodeUrl(...args), + waitForReady: (...args) => networkRuntime.waitForReady(...args), + normalizeApiPrefix: (...args) => networkRuntime.normalizeApiPrefix(...args), + applyOpencodeBinaryFromSettings: async () => {}, + ensureOpencodeCliEnv: () => {}, + ensureLocalOpenCodeServerPassword: async () => 'password', + resolveManagedOpenCodeLaunchSpec: (binary) => ({ binary, args: [], wrapperType: null }), + setOpenCodePort: (port) => { state.openCodePort = port; }, + setDetectedOpenCodeApiPrefix: () => {}, + setupProxy: () => {}, + ensureOpenCodeApiPrefix: () => {}, + clearResolvedOpenCodeBinary: () => {}, + buildAugmentedPath: () => process.env.PATH, + buildManagedOpenCodePath: () => process.env.PATH, + getManagedOpenCodeShellEnvSnapshot: async () => ({}), + getManagedOpenCodeEnv: async () => ({}), + reapManagedOrphanedProcesses: async () => ({ reaped: 0 }), + getWarmupDirectories: async () => [], + // Production index.js wires this to the message-stream runtime's + // rebindUpstream(); mirror it here so the harness exercises the fix. + onOpenCodeRestarted: () => { + try { + rebindHub?.(); + } catch { + } + }, +}); + +const { createGlobalMessageStreamHub } = await import( + path.join(REPO, 'packages/web/server/lib/event-stream/global-hub.js') +); + +// The hub represents the server→OpenCode SSE push pipeline that feeds the +// renderer (both the server-side PushWatcher and the browser WS bridge). +const received = []; +const statuses = []; +let rebindHub = null; +const hub = createGlobalMessageStreamHub({ + buildOpenCodeUrl: (p) => networkRuntime.buildOpenCodeUrl(p, ''), + getOpenCodeAuthHeaders: () => ({}), + upstreamStallTimeoutMs: 20000, + upstreamReconnectDelayMs: 250, +}); +hub.subscribeEvent(({ eventId, payload }) => { + received.push({ eventId, type: payload?.type, id: payload?.id }); +}); +hub.subscribeStatus((status) => statuses.push(status)); +rebindHub = () => { + hub.stop(); + hub.start(); +}; + +// --- helpers ---------------------------------------------------------------- +const warnLog = []; +const origWarn = console.warn; +console.warn = (...args) => { + warnLog.push(args.map(String).join(' ')); + origWarn(...args); +}; + +const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); + +async function waitFor(fn, timeoutMs, what) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await fn()) return true; + await sleep(50); + } + throw new Error(`timeout waiting for ${what}`); +} + +const emitEvent = async (port, id, type = 'session.updated') => { + const res = await fetch(`http://127.0.0.1:${port}/emit?type=${type}&id=${id}`); + if (!res.ok) throw new Error(`emit ${id} failed on port ${port}`); + return res.json(); +}; + +const persistedEvents = async (port) => { + const res = await fetch(`http://127.0.0.1:${port}/events`); + return res.ok ? res.json() : []; +}; + +const portOpen = (port) => new Promise((resolve) => { + const socket = net.connect({ port, host: '127.0.0.1' }); + const timer = setTimeout(() => { socket.destroy(); resolve(false); }, 300); + socket.once('connect', () => { clearTimeout(timer); socket.destroy(); resolve(true); }); + socket.once('error', () => { clearTimeout(timer); resolve(false); }); +}); + +const pidFilePids = () => { + const out = []; + for (const file of fs.readdirSync(pidDir)) { + if (!file.endsWith('.pid')) continue; + try { + out.push({ label: file.replace(/\.pid$/, ''), pid: Number(fs.readFileSync(path.join(pidDir, file), 'utf8')) }); + } catch { + // ignore + } + } + return out; +}; + +const killPortPids = (port) => { + try { + const result = spawnSync('lsof', ['-ti', `:${port}`], { encoding: 'utf8' }); + const pids = String(result.stdout || '').trim().split(/\s+/).map(Number).filter(Boolean); + for (const pid of pids) { + if (pid === process.pid) continue; // never kill ourselves (TIME_WAIT client sockets) + try { process.kill(pid, 'SIGKILL'); } catch { /* already gone */ } + } + } catch { /* lsof unavailable */ } +}; + +const cleanup = async () => { + for (const { label, pid } of pidFilePids()) { + if (label.startsWith('launcher') || label.startsWith('baseline')) { + try { process.kill(pid, 'SIGTERM'); } catch { /* gone */ } + } else { + try { process.kill(pid, 'SIGKILL'); } catch { /* gone */ } + } + } + if (state.openCodePort) killPortPids(state.openCodePort); + // Belt and braces: kill any surviving fake-opencode processes from this run. + try { + const result = spawnSync('pgrep', ['-f', 'fake-opencode-serve.mjs'], { encoding: 'utf8' }); + const pids = String(result.stdout || '').trim().split(/\s+/).map(Number).filter(Boolean); + for (const pid of pids) { + if (pid === process.pid) continue; + try { process.kill(pid, 'SIGKILL'); } catch { /* gone */ } + } + } catch { /* pgrep unavailable */ } + await sleep(300); + try { fs.rmSync(pidDir, { recursive: true, force: true }); } catch { /* ignore */ } + console.warn = origWarn; +}; + +// --- run --------------------------------------------------------------------- +console.log(`\n=== reproduce-2638 (${BASELINE ? 'BASELINE control' : 'Windows-orphan scenario'}) ===\n`); +let failures = 0; +const check = (label, ok, detail = '') => { + console.log(` ${ok ? 'PASS' : 'FAIL'} ${label}${detail ? ` — ${detail}` : ''}`); + if (!ok) failures += 1; +}; + +try { + // 1. Bootstrapping starts the managed OpenCode (launcher + server core) on P1. + await lifecycle.bootstrapOpenCodeAtStartup(); + const p1 = state.openCodePort; + console.log(`[1] bootstrap OK — managed OpenCode listening on port ${p1} (pid ${state.openCodeProcess?.pid})`); + + // 2. Connect the message-stream hub (server→OpenCode SSE push pipeline). + hub.start(); + await waitFor(() => statuses.some((s) => s.type === 'connect'), 10000, 'hub connect to P1'); + console.log('[2] message-stream hub connected to /global/event'); + + // 3. Baseline delivery: an event emitted by P1 reaches the UI pipeline. + await emitEvent(p1, 'evt-before-restart'); + await waitFor(() => received.some((r) => r.eventId === 'evt-before-restart'), 5000, 'event delivery'); + check('events flow to the UI before the restart (baseline)', received.some((r) => r.eventId === 'evt-before-restart')); + + // 4. The managed process "exits" while the actual server survives on P1 + // (simulates the Windows orphan: launcher dies, server core keeps the + // port and the SSE stream). In baseline mode the server runs in-process + // and dies with the managed process instead. + const launcherPid = state.openCodeProcess.pid; + process.kill(launcherPid, 'SIGTERM'); + await waitFor(async () => { + try { process.kill(launcherPid, 0); return false; } catch { return true; } + }, 5000, 'launcher exit'); + console.log(`[4] managed process (pid ${launcherPid}) exited; ${BASELINE ? 'server process died with it' : `orphaned server core still listening on ${p1}`}`); + + // 5. Trigger the reported restart path ("Refreshing OpenCode after manual + // configuration reload" / periodic health check). + console.log('[5] triggering restart (refreshOpenCodeAfterConfigChange)...'); + await lifecycle.refreshOpenCodeAfterConfigChange('manual configuration reload'); + const p2 = state.openCodePort; + console.log(`[5] restarted — new managed OpenCode listening on port ${p2}`); + + // 6. Assert the reported log line: the old port was never released. In the + // baseline control the port IS released, so the warning must be absent. + const timeoutWarn = warnLog.find((line) => line.includes('Timed out waiting for OpenCode port') && line.includes(String(p1))); + if (BASELINE) { + check(`no "Timed out waiting for OpenCode port ${p1}" warning in baseline control`, !timeoutWarn); + } else { + check(`"Timed out waiting for OpenCode port ${p1} to be released" is logged`, Boolean(timeoutWarn), timeoutWarn || ''); + } + check('new port differs from old port (leaked process pinned the old one)', p2 !== p1, `p1=${p1} p2=${p2}`); + + // 7. Assert the orphaned old server is still running (process pile-up from + // the report: "six opencode.exe serve processes were still running"). + // In baseline mode we instead expect the port to be properly released. + const oldCoreStillUp = await portOpen(p1); + const pids = pidFilePids(); + const orphanPids = pids.filter(({ label }) => label.startsWith('core')).map(({ pid }) => pid); + const orphanAlive = orphanPids.length > 0 && orphanPids.every((pid) => { + try { process.kill(pid, 0); return true; } catch { return false; } + }); + if (BASELINE) { + check('old port properly released (no orphan in baseline control)', !oldCoreStillUp, + `old port ${p1} ${oldCoreStillUp ? 'still open' : 'released'}`); + } else { + check('orphaned server process still running on the old port', oldCoreStillUp && orphanAlive, + `old port ${p1} still open; orphan core pids ${orphanPids.join(', ')}`); + } + + // 8. The stale-UI reproduction: the new server persists events, but in the + // orphan scenario they never reach the UI because the hub is still pinned + // to the old SSE stream. In baseline mode the hub must reconnect to the + // new port and deliver them (wait for the reconnect before emitting — + // the upstream reader only learns about the new port on its next attempt). + const connectsBefore = statuses.filter((s) => s.type === 'connect').length; + if (!BASELINE) { + await sleep(1000); + } else { + await waitFor(() => statuses.filter((s) => s.type === 'connect').length >= connectsBefore + 1, 15000, 'hub reconnect to new port'); + } + await emitEvent(p2, 'evt-after-restart'); + console.log(`[8] emitted evt-after-restart on new port ${p2} — waiting to see if the UI receives it...`); + await sleep(2500); + const deliveredAfter = received.filter((r) => r.eventId === 'evt-after-restart').length; + if (BASELINE) { + check('NEW server event IS delivered to the UI (hub reconnected in baseline)', deliveredAfter === 1, + `delivered=${deliveredAfter}, total hub events=${received.length}`); + } else { + // Fixed: the lifecycle hook rebinds the hub after the managed restart, + // so the UI receives events from the new port even when the old port is + // orphaned. The wait mirrors the baseline branch (the reader re-dials on + // its next attempt after the rebind). + await waitFor(() => statuses.filter((s) => s.type === 'connect').length >= connectsBefore + 1, 15000, 'hub reconnect to new port after rebind'); + check('NEW server event IS delivered to the UI (rebound after restart)', deliveredAfter === 1, + `delivered=${deliveredAfter}, total hub events=${received.length}`); + } + + const persistedOnNew = await persistedEvents(p2); + check('NEW server persisted the event (data survives, UI does not show it)', persistedOnNew.some((e) => e.id === 'evt-after-restart'), + `persisted on port ${p2}: ${JSON.stringify(persistedOnNew)}`); + + // 9. Orphan scenario: with the rebind the hub is no longer pinned to the + // old server — events emitted by the old (zombie) server must NOT reach + // the UI anymore (it left the previous upstream behind). + if (!BASELINE) { + await emitEvent(p1, 'evt-zombie-old-server'); + await sleep(2000); + check('OLD zombie server events no longer reach the UI (hub rebound to new upstream)', !received.some((r) => r.eventId === 'evt-zombie-old-server')); + } + + console.log(`\n=== ${failures === 0 ? 'REPRODUCED (all checks passed)' : `${failures} check(s) FAILED`} ===\n`); + console.log(`hub statuses observed: ${JSON.stringify(statuses)}`); +} catch (error) { + console.error('\nReproduction script error:', error); + failures += 1; +} finally { + await cleanup(); +} + +process.exit(failures === 0 ? 0 : 1);