Merge origin/main into deferred OpenCode restart branch
This commit is contained in:
@@ -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
|
||||
@@ -1474,6 +1487,10 @@ async function main(options = {}) {
|
||||
// relay candidate lazily at request time, so a late-bound holder is enough.
|
||||
let relayServiceInstance = null;
|
||||
|
||||
// Same pattern for the tunnel runtime: created after the base routes so
|
||||
// /api/system/info resolves port + tunnel URL lazily at request time.
|
||||
let tunnelRuntimeContextHolder = null;
|
||||
|
||||
const bootstrapResult = bootstrapRuntime.setupBaseRoutes(app, {
|
||||
process,
|
||||
openchamberVersion: OPENCHAMBER_VERSION,
|
||||
@@ -1510,6 +1527,15 @@ async function main(options = {}) {
|
||||
apiOnly,
|
||||
};
|
||||
},
|
||||
// Port this instance serves on and the active tunnel's public URL (if
|
||||
// any), for /api/system/info. Resolved lazily because the tunnel runtime
|
||||
// is created after these base routes are registered.
|
||||
getServerPort: () => {
|
||||
const activePort = tunnelRuntimeContextHolder?.getActivePort?.();
|
||||
if (Number.isFinite(activePort) && activePort > 0) return activePort;
|
||||
return Number.isFinite(port) && port > 0 ? port : null;
|
||||
},
|
||||
getTunnelUrl: () => tunnelRuntimeContextHolder?.tunnelService?.getPublicUrl?.() ?? null,
|
||||
verboseRequestLogs: OPENCHAMBER_VERBOSE_REQUEST_LOGS,
|
||||
uiPassword,
|
||||
tunnelAuthController,
|
||||
@@ -1585,6 +1611,7 @@ async function main(options = {}) {
|
||||
|
||||
const tunnelRuntimeContext = tunnelWiringRuntime.initialize(app, port);
|
||||
const { tunnelService, startTunnelWithNormalizedRequest } = tunnelRuntimeContext;
|
||||
tunnelRuntimeContextHolder = tunnelRuntimeContext;
|
||||
|
||||
// Private relay host service: config + management routes + host client
|
||||
// lifecycle. Loopback port comes from the same source the tunnel uses so
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
@@ -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();
|
||||
|
||||
@@ -34,4 +34,6 @@ Own filesystem API behavior for the web server runtime, including workspace-boun
|
||||
|
||||
## Notes for contributors
|
||||
- Keep filesystem policy (workspace root checks, error mapping, exec timeout behavior) inside this module, not in the composition root.
|
||||
- Filesystem `EPERM`/`EACCES` failures use the stable `reason: "os-permission"` response marker. Policy denials such as workspace-boundary or missing-grant failures must not use that marker because a native folder picker cannot remediate them.
|
||||
- If adding new `/api/fs/*` endpoints, add them in `routes.js` and extend this document.
|
||||
- `GET /api/fs/list` may resolve symlinks with `realpath` to read directory contents, but the response `path` and each entry `path` must stay in the caller's requested path space (`path.join(requestedPath, name)`). Returning real paths breaks file-tree expansion for directories reached through workspace symlinks.
|
||||
|
||||
@@ -16,6 +16,16 @@ const pruneOutsideFileGrants = () => {
|
||||
}
|
||||
};
|
||||
|
||||
const isOsPermissionError = (error) => (
|
||||
error
|
||||
&& typeof error === 'object'
|
||||
&& (error.code === 'EACCES' || error.code === 'EPERM')
|
||||
);
|
||||
|
||||
const sendOsPermissionDenied = (res, message) => (
|
||||
res.status(403).json({ error: message, reason: 'os-permission' })
|
||||
);
|
||||
|
||||
export const mintOutsideFileGrant = async (targetPath, {
|
||||
scopes = ['stat', 'read', 'raw'],
|
||||
fsPromises = nodeFsPromises,
|
||||
@@ -382,6 +392,7 @@ export const registerFsRoutes = (app, dependencies) => {
|
||||
path,
|
||||
fsPromises,
|
||||
spawn,
|
||||
platform = process.platform,
|
||||
crypto,
|
||||
normalizeDirectoryPath,
|
||||
resolveProjectDirectory,
|
||||
@@ -393,6 +404,27 @@ export const registerFsRoutes = (app, dependencies) => {
|
||||
realpath: fsPromises.realpath.bind(fsPromises),
|
||||
});
|
||||
|
||||
const spawnDetached = (command, args) => new Promise((resolve, reject) => {
|
||||
let child;
|
||||
try {
|
||||
child = spawn(command, args, { windowsHide: true, stdio: 'ignore', detached: true });
|
||||
} catch (error) {
|
||||
reject(new Error('Failed to launch file browser', { cause: error }));
|
||||
return;
|
||||
}
|
||||
const onError = (error) => {
|
||||
child.removeListener('spawn', onSpawn);
|
||||
reject(new Error('Failed to launch file browser', { cause: error }));
|
||||
};
|
||||
const onSpawn = () => {
|
||||
child.removeListener('error', onError);
|
||||
child.unref();
|
||||
resolve();
|
||||
};
|
||||
child.once('error', onError);
|
||||
child.once('spawn', onSpawn);
|
||||
});
|
||||
|
||||
const execJobs = new Map();
|
||||
const commandTimeoutMs = createCommandTimeoutMs();
|
||||
const gitReadCacheTtlMs = createGitReadCacheTtlMs();
|
||||
@@ -454,7 +486,7 @@ export const registerFsRoutes = (app, dependencies) => {
|
||||
// Non-cacheable commands always execute and are never stored.
|
||||
const runCommandWithGitReadCache = async ({ shell, shellFlag, command, resolvedCwd }) => {
|
||||
const cacheable = gitReadCacheTtlMs > 0 && isCacheableGitReadCommand(command);
|
||||
const cacheKey = cacheable ? `${resolvedCwd} | ||||