fix: stream bash output and harden OpenCode connectivity (#2522)

* fix(ui): stream bash tool output while running * perf(ui): render streaming bash output incrementally * fix(ui): keep tool duration timer running * fix(web): recover stalled OpenCode SSE streams * fix(web): prevent OpenCode restart storms * fix: address streaming recovery review
2026-07-29 17:26:11 +03:00
parent c84305bf7e
commit 0f830f8804
12 changed files with 372 additions and 41 deletions
@@ -120,6 +120,8 @@ runtime. PATH and `OPENCODE_SERVER_PASSWORD` remain lifecycle-owned and cannot
 be replaced by injected values. External OpenCode processes receive no
 OpenChamber tool injection.

+Transport-triggered health checks share the periodic monitor's failure accounting interval. Rapid WS reconnect callbacks therefore cannot exhaust the managed-process restart threshold using one cached unhealthy result; an exited managed process still restarts immediately.
+
 ## Public exports (env-runtime.js)
 - `createOpenCodeEnvRuntime(dependencies)`: creates runtime that owns OpenCode CLI environment and binary discovery state.
 - OpenCode CLI resolution order is persisted settings, environment overrides, bundled Desktop CLI when available, PATH, known install locations, then platform shell discovery.
@@ -350,6 +352,7 @@ an authoritative loopback callback URL even when OpenChamber binds port `0`.
 - `registerOpenCodeProxy(app, dependencies)`: registers OpenCode proxy routes and middleware.
 - Owns:
  - SSE forwarders: `GET /api/global/event`, `GET /api/event`
+    - Downstream heartbeats keep clients and intermediaries alive, while a separate upstream-only stall watchdog closes the downstream response when OpenCode stops producing bytes so clients reconnect instead of trusting synthetic heartbeats indefinitely. Each watchdog reset uses the current load-aware timeout, matching the shared event transport.
  - Session message forwarder: `POST /api/session/:sessionId/message`
  - Generic `/api/*` forwarding with hop-by-hop header filtering
  - Windows `/session` merge fallback path behavior
@@ -40,6 +40,7 @@ export const createOpenCodeLifecycleRuntime = (deps) => {
    getManagedOpenCodeShellEnvSnapshot,
    getManagedOpenCodeEnv = async () => ({}),
    getActiveSessionCount = () => 0,
+    now = Date.now,
  } = deps;

  const killProcessOnPort = (port) => {
@@ -871,18 +872,21 @@ export const createOpenCodeLifecycleRuntime = (deps) => {
  const STALE_BUSY_GRACE_MS = 2 * 60 * 1000;
  let lastUnhealthyWithBusySessionsAt = 0;
  let consecutiveHealthFailures = 0;
+  let lastCountedHealthFailureAt = 0;
  let healthProbePromise = null;
  let healthCheckCyclePromise = null;
  let lastHealthProbeResult = null;
+  let healthFailureCountIntervalMs = 15_000;

  const resetHealthFailureState = () => {
    consecutiveHealthFailures = 0;
    lastUnhealthyWithBusySessionsAt = 0;
+    lastCountedHealthFailureAt = 0;
  };

  const probeOpenCodeHealth = async () => {
-    const now = Date.now();
-    if (lastHealthProbeResult && now - lastHealthProbeResult.at < HEALTH_CHECK_RESULT_CACHE_MS) {
+    const checkedAt = now();
+    if (lastHealthProbeResult && checkedAt - lastHealthProbeResult.at < HEALTH_CHECK_RESULT_CACHE_MS) {
      return lastHealthProbeResult.healthy;
    }

@@ -892,7 +896,7 @@ export const createOpenCodeLifecycleRuntime = (deps) => {

    healthProbePromise = isOpenCodeProcessHealthy()
      .then((healthy) => {
-        lastHealthProbeResult = { at: Date.now(), healthy };
+        lastHealthProbeResult = { at: now(), healthy };
        return healthy;
      })
      .finally(() => {
@@ -909,13 +913,13 @@ export const createOpenCodeLifecycleRuntime = (deps) => {
      return false;
    }

-    const now = Date.now();
+    const checkedAt = now();
    if (!lastUnhealthyWithBusySessionsAt) {
-      lastUnhealthyWithBusySessionsAt = now;
+      lastUnhealthyWithBusySessionsAt = checkedAt;
      return true;
    }

-    if (now - lastUnhealthyWithBusySessionsAt >= STALE_BUSY_GRACE_MS) {
+    if (checkedAt - lastUnhealthyWithBusySessionsAt >= STALE_BUSY_GRACE_MS) {
      console.warn(
        `[lifecycle] OpenCode unhealthy with ${activeCount} busy session(s) for > 2 min — forcing restart`
      );
@@ -940,6 +944,11 @@ export const createOpenCodeLifecycleRuntime = (deps) => {
          await restartOpenCode();
          return;
        }
+        const checkedAt = now();
+        if (lastCountedHealthFailureAt && checkedAt - lastCountedHealthFailureAt < healthFailureCountIntervalMs) {
+          return;
+        }
+        lastCountedHealthFailureAt = checkedAt;
        consecutiveHealthFailures += 1;
        console.warn(
          `[lifecycle] ${source} health check failed (${consecutiveHealthFailures}/${HEALTH_CHECK_MAX_CONSECUTIVE_FAILURES})`
@@ -974,6 +983,7 @@ export const createOpenCodeLifecycleRuntime = (deps) => {
    }

    const effectiveIntervalMs = HEALTH_CHECK_INTERVAL_OVERRIDE_MS || healthCheckIntervalMs;
+    healthFailureCountIntervalMs = effectiveIntervalMs;

    state.healthCheckInterval = setInterval(async () => {
      try {
@@ -12,9 +12,11 @@ const { createOpenCodeLifecycleRuntime } = await import('./lifecycle.js');

 const originalOpencodeBinary = process.env.OPENCODE_BINARY;
 const originalPath = process.env.PATH;
+const originalFetch = globalThis.fetch;

 afterEach(() => {
  spawnMock.mockReset();
+  globalThis.fetch = originalFetch;
  if (typeof originalOpencodeBinary === 'string') {
    process.env.OPENCODE_BINARY = originalOpencodeBinary;
  } else {
@@ -43,7 +45,7 @@ const createMockChild = () => {
  return child;
 };

-const createRuntime = (overrides = {}) => {
+const createRuntime = (overrides = {}, stateOverrides = {}) => {
  const state = {
    openCodeWorkingDirectory: '/tmp/project',
    openCodeProcess: null,
@@ -65,6 +67,7 @@ const createRuntime = (overrides = {}) => {
    resolvedWslBinary: null,
    resolvedWslOpencodePath: null,
    resolvedWslDistro: null,
+    ...stateOverrides,
  };

  return createOpenCodeLifecycleRuntime({
@@ -105,6 +108,69 @@ const createRuntime = (overrides = {}) => {
 };

 describe('OpenCode lifecycle', () => {
+  it('does not count rapid transport-triggered checks as independent health failures', async () => {
+    const close = vi.fn(async () => {});
+    const warn = vi.spyOn(console, 'warn').mockImplementation(() => {});
+    let now = 1;
+    globalThis.fetch = vi.fn(async () => ({
+      ok: false,
+      json: async () => null,
+    }));
+    const runtime = createRuntime({ now: () => now }, {
+      openCodePort: 45678,
+      openCodeProcess: {
+        pid: null,
+        exitCode: null,
+        signalCode: null,
+        close,
+      },
+      isOpenCodeReady: true,
+    });
+
+    for (let attempt = 0; attempt < 25; attempt += 1) {
+      await runtime.triggerHealthCheck();
+    }
+
+    expect(close).not.toHaveBeenCalled();
+    expect(warn).toHaveBeenCalledTimes(1);
+
+    now += 15_000;
+    await runtime.triggerHealthCheck();
+
+    expect(warn).toHaveBeenCalledTimes(2);
+    expect(warn).toHaveBeenLastCalledWith(expect.stringContaining('(2/20)'));
+    warn.mockRestore();
+  });
+
+  it('restarts an exited managed process without waiting for the failure interval', async () => {
+    const close = vi.fn(async () => {});
+    const replacement = createMockChild();
+    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({}, {
+      openCodePort: 45678,
+      openCodeProcess: {
+        pid: null,
+        exitCode: 1,
+        signalCode: null,
+        close,
+      },
+    });
+
+    await runtime.triggerHealthCheck();
+
+    expect(close).toHaveBeenCalledTimes(1);
+    expect(spawnMock).toHaveBeenCalledTimes(1);
+  });
+
  it('launches managed OpenCode with the managed PATH', async () => {
    delete process.env.OPENCODE_BINARY;
    const child = createMockChild();
@@ -6,6 +6,9 @@ import {
  shouldForwardProxyResponseHeader,
 } from '../../proxy-headers.js';
 import { createRealpathCache } from '../path-realpath-cache.js';
+import { DEFAULT_UPSTREAM_STALL_TIMEOUT_MS } from '../event-stream/upstream-reader.js';
+
+const DEFAULT_SSE_HEARTBEAT_INTERVAL_MS = 20_000;

 export const createDirectoryQueryCanonicalizer = ({ realpath, ...cacheOptions } = {}) => {
  const realpathCache = createRealpathCache({ fallbackOnError: true, realpath, ...cacheOptions });
@@ -186,6 +189,9 @@ export const registerOpenCodeProxy = (app, deps) => {
    getOpenCodeAuthHeaders,
    buildOpenCodeUrl,
    ensureOpenCodeApiPrefix,
+    SSE_HEARTBEAT_INTERVAL_MS = DEFAULT_SSE_HEARTBEAT_INTERVAL_MS,
+    SSE_UPSTREAM_STALL_TIMEOUT_MS = DEFAULT_UPSTREAM_STALL_TIMEOUT_MS,
+    getSseUpstreamStallTimeoutMs = () => SSE_UPSTREAM_STALL_TIMEOUT_MS,
  } = deps;

  if (app.get('opencodeProxyConfigured')) {
@@ -340,6 +346,8 @@ export const registerOpenCodeProxy = (app, deps) => {
    let upstream = null;
    let reader = null;
    let heartbeatTimer = null;
+    let upstreamStallTimer = null;
+    let didUpstreamStall = false;
    let writeQueue = Promise.resolve(true);
    const sseBoundary = createSseBoundaryTracker();

@@ -392,8 +400,6 @@ export const registerOpenCodeProxy = (app, deps) => {
        res.socket.setNoDelay(true);
      }

-      const SSE_HEARTBEAT_INTERVAL_MS = 20_000;
-
      const scheduleHeartbeat = () => {
        heartbeatTimer = setTimeout(async () => {
          if (abortController.signal.aborted || res.writableEnded || res.destroyed) {
@@ -410,6 +416,20 @@ export const registerOpenCodeProxy = (app, deps) => {
        }, SSE_HEARTBEAT_INTERVAL_MS);
      };

+      const clearUpstreamStallTimer = () => {
+        clearTimeout(upstreamStallTimer);
+        upstreamStallTimer = null;
+      };
+
+      const resetUpstreamStallTimer = () => {
+        clearUpstreamStallTimer();
+        upstreamStallTimer = setTimeout(() => {
+          didUpstreamStall = true;
+          abortController.abort();
+        }, getSseUpstreamStallTimeoutMs());
+        upstreamStallTimer.unref?.();
+      };
+
      const enqueueSseWrite = (value) => {
        writeQueue = writeQueue
          .catch(() => false)
@@ -423,6 +443,7 @@ export const registerOpenCodeProxy = (app, deps) => {
      };

      scheduleHeartbeat();
+      resetUpstreamStallTimer();

      reader = upstream.body.getReader();
      while (!abortController.signal.aborted) {
@@ -431,6 +452,7 @@ export const registerOpenCodeProxy = (app, deps) => {
          break;
        }
        if (value && value.length > 0) {
+          resetUpstreamStallTimer();
          sseBoundary.observe(value);
          const canContinue = await enqueueSseWrite(value);
          if (!canContinue) {
@@ -442,6 +464,10 @@ export const registerOpenCodeProxy = (app, deps) => {
      res.end();
    } catch (error) {
      if (isAbortError(error)) {
+        if (didUpstreamStall && !res.writableEnded && !res.destroyed) {
+          await writeQueue.catch(() => false);
+          res.end();
+        }
        return;
      }
      console.error('[proxy] OpenCode SSE proxy error:', error?.message ?? error);
@@ -455,6 +481,10 @@ export const registerOpenCodeProxy = (app, deps) => {
        clearTimeout(heartbeatTimer);
        heartbeatTimer = null;
      }
+      if (upstreamStallTimer) {
+        clearTimeout(upstreamStallTimer);
+        upstreamStallTimer = null;
+      }
      req.off('close', closeUpstream);
      try {
        if (reader) {
@@ -13,6 +13,7 @@ export const createServerUtilsRuntime = (dependencies) => {
    getOpenCodeAuthHeaders,
    buildOpenCodeUrl,
    ensureOpenCodeApiPrefix,
+    getUpstreamStallTimeoutMs,
    getUiNotificationClients,
    getOpenCodePort,
    setOpenCodePortState,
@@ -212,6 +213,7 @@ export const createServerUtilsRuntime = (dependencies) => {
      getOpenCodeAuthHeaders,
      buildOpenCodeUrl,
      ensureOpenCodeApiPrefix,
+      getSseUpstreamStallTimeoutMs: getUpstreamStallTimeoutMs,
      getUiNotificationClients,
    });
  };