From 0f830f880477be08cb24dc3eb4144c894ec1f5a2 Mon Sep 17 00:00:00 2001 From: Bohdan Triapitsyn Date: Wed, 29 Jul 2026 17:26:11 +0300 Subject: [PATCH] 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 --- .../chat/hooks/useStreamingTextThrottle.ts | 20 ++- .../chat/message/parts/DOCUMENTATION.md | 1 + .../chat/message/parts/ToolPart.test.ts | 40 +++++ .../chat/message/parts/ToolPart.tsx | 154 +++++++++++++++--- .../chat/message/parts/toolOutput.ts | 19 +++ packages/web/server/index.js | 1 + .../web/server/lib/opencode/DOCUMENTATION.md | 3 + packages/web/server/lib/opencode/lifecycle.js | 22 ++- .../web/server/lib/opencode/lifecycle.test.js | 68 +++++++- packages/web/server/lib/opencode/proxy.js | 34 +++- .../lib/opencode/server-utils-runtime.js | 2 + packages/web/server/opencode-proxy.test.js | 49 ++++++ 12 files changed, 372 insertions(+), 41 deletions(-) create mode 100644 packages/ui/src/components/chat/message/parts/toolOutput.ts diff --git a/packages/ui/src/components/chat/hooks/useStreamingTextThrottle.ts b/packages/ui/src/components/chat/hooks/useStreamingTextThrottle.ts index 8d4175ac..8ccce97f 100644 --- a/packages/ui/src/components/chat/hooks/useStreamingTextThrottle.ts +++ b/packages/ui/src/components/chat/hooks/useStreamingTextThrottle.ts @@ -5,10 +5,20 @@ interface UseStreamingTextThrottleInput { isStreaming: boolean; throttleMs?: number; identityKey?: string; + allowTextReplacement?: boolean; } const DEFAULT_STREAMING_TEXT_THROTTLE_MS = 100; +export const getStreamingThrottleText = ( + current: string, + next: string, + isStreaming: boolean, + allowTextReplacement: boolean, +): string => { + return isStreaming && !allowTextReplacement && current.length > next.length ? current : next; +}; + const computeStreamingThrottleDelay = (lastEmitAt: number, now: number, throttleMs: number): number => { const elapsed = now - lastEmitAt; return Math.max(0, throttleMs - elapsed); @@ -33,6 +43,7 @@ export const useStreamingTextThrottle = ({ isStreaming, throttleMs = DEFAULT_STREAMING_TEXT_THROTTLE_MS, identityKey, + allowTextReplacement = false, }: UseStreamingTextThrottleInput): string => { const [throttledText, setThrottledText] = React.useState(text); const latestTextRef = React.useRef(text); @@ -64,7 +75,7 @@ export const useStreamingTextThrottle = ({ const state = stateRef.current; state.pendingText = text; const currentThrottled = throttledTextRef.current; - const stableText = isStreaming && currentThrottled.length > text.length ? currentThrottled : text; + const stableText = getStreamingThrottleText(currentThrottled, text, isStreaming, allowTextReplacement); if (!isStreaming) { clearTimer(state); @@ -88,17 +99,14 @@ export const useStreamingTextThrottle = ({ state.timer = null; state.lastEmitAt = Date.now(); setThrottledText((prev) => { - if (isStreaming && prev.length > state.pendingText.length) { - return prev; - } - return state.pendingText; + return getStreamingThrottleText(prev, state.pendingText, isStreaming, allowTextReplacement); }); }, remaining); return () => { clearTimer(state); }; - }, [isStreaming, text, throttleMs]); + }, [allowTextReplacement, isStreaming, text, throttleMs]); React.useEffect(() => { const state = stateRef.current; diff --git a/packages/ui/src/components/chat/message/parts/DOCUMENTATION.md b/packages/ui/src/components/chat/message/parts/DOCUMENTATION.md index 2a0e86e4..02fbd092 100644 --- a/packages/ui/src/components/chat/message/parts/DOCUMENTATION.md +++ b/packages/ui/src/components/chat/message/parts/DOCUMENTATION.md @@ -55,6 +55,7 @@ Use this doc when you ask an agent to change tool/header/description behavior. - Every other tool, including search/fetch, OpenCode built-ins, custom tools, plugins, and MCP tools, is **expandable** and renders through `ToolPart`. - The managed `openchamber` plugin tool uses the expandable path and hides its broad protocol input. The plugin supplies the selected action's human description as the native tool title; the UI renders that metadata without owning an action map. The full versioned result envelope renders through the same neutral JSON summary/tree/raw views as other tools, without a tool-specific output card. - `ToolPart` defers expanded content after a user toggle, preventing large tool input/output payloads from mounting during the initial chat render. +- Running bash output falls back to `state.metadata.output` until canonical `state.output` arrives. Its fixed-height output viewport follows new output until the user scrolls up, then resumes following when the user returns to the bottom. Live output appends or replaces rewritten snapshots as plain text without worker highlighting; finalized output bypasses the throttle and receives the normal one-time highlighted rendering. - Thinking/Justification duration is hidden in `sorted` mode (handled in `ReasoningPart.tsx` + `JustificationBlock.tsx`). ## "I want to change description for Perplexity" (example recipe) diff --git a/packages/ui/src/components/chat/message/parts/ToolPart.test.ts b/packages/ui/src/components/chat/message/parts/ToolPart.test.ts index a56eac78..b263ea52 100644 --- a/packages/ui/src/components/chat/message/parts/ToolPart.test.ts +++ b/packages/ui/src/components/chat/message/parts/ToolPart.test.ts @@ -2,6 +2,46 @@ import { describe, expect, test } from 'bun:test'; import { readTaskTagSessionIdFromOutput } from './taskSessionIdParser'; import { tryParseJsonOutput } from '../toolRenderers'; +import { getStreamingThrottleText } from '../../hooks/useStreamingTextThrottle'; +import { getStreamingOutputAppend, getToolOutput } from './toolOutput'; + +describe('getToolOutput', () => { + test('prefers authoritative state output', () => { + expect(getToolOutput('bash', 'final output', 'streamed output')).toBe('final output'); + expect(getToolOutput('bash', '', 'streamed output')).toBe(''); + }); + + test('falls back to streamed metadata output for bash', () => { + expect(getToolOutput('bash', undefined, 'streamed output')).toBe('streamed output'); + expect(getToolOutput('bash', undefined, '')).toBe(undefined); + }); + + test('does not expose metadata output for other tools', () => { + expect(getToolOutput('read', undefined, 'metadata output')).toBe(undefined); + }); +}); + +describe('getStreamingOutputAppend', () => { + test('returns only newly appended output', () => { + expect(getStreamingOutputAppend('first\n', 'first\nsecond\n')).toBe('second\n'); + }); + + test('requires replacement when output is rewritten or shortened', () => { + expect(getStreamingOutputAppend('progress 10%', 'progress 20%')).toBe(undefined); + expect(getStreamingOutputAppend('long output', 'short')).toBe(undefined); + }); +}); + +describe('streaming output transitions', () => { + test('allows bash snapshots to be rewritten or shortened while running', () => { + expect(getStreamingThrottleText('progress 10%', 'progress 20%', true, true)).toBe('progress 20%'); + expect(getStreamingThrottleText('long output', 'short', true, true)).toBe('short'); + }); + + test('preserves monotonic streaming text by default', () => { + expect(getStreamingThrottleText('long output', 'short', true, false)).toBe('long output'); + }); +}); describe('readTaskTagSessionIdFromOutput', () => { test('parses task tags without state attributes', () => { diff --git a/packages/ui/src/components/chat/message/parts/ToolPart.tsx b/packages/ui/src/components/chat/message/parts/ToolPart.tsx index 9ed08f5d..0a3cfc61 100644 --- a/packages/ui/src/components/chat/message/parts/ToolPart.tsx +++ b/packages/ui/src/components/chat/message/parts/ToolPart.tsx @@ -55,6 +55,8 @@ import { areRenderRelevantPartsEqual } from '../renderCompare'; import { useI18n } from '@/lib/i18n'; import { getDiffPatchEntries, getPatchText, type DiffPatchEntry } from './toolDiffUtils'; import { isEmbeddedSessionChat } from '@/components/layout/contextPanelEmbeddedChat'; +import { useStreamingTextThrottle } from '../../hooks/useStreamingTextThrottle'; +import { getStreamingOutputAppend, getToolOutput } from './toolOutput'; const TOOL_ROW_TEXT_CLASS = '!text-[length:var(--text-meta)] !leading-5 sm:!leading-6 tracking-normal'; const TOOL_ROW_TITLE_CLASS = cn('typography-meta font-medium', TOOL_ROW_TEXT_CLASS); @@ -170,7 +172,6 @@ const normalizeToolName = (toolName: string | undefined | null): string => { return trimmed; }; -const MAX_DURATION_MS = 5 * 60 * 1000; // 5 minutes cap const GIT_REFRESH_MUTATING_TOOLS = new Set([ 'bash', 'edit', @@ -181,7 +182,7 @@ const GIT_REFRESH_MUTATING_TOOLS = new Set([ ]); const formatDuration = (start: number, end?: number, now: number = Date.now()) => { - const duration = Math.min(Math.max(0, (end ?? now) - start), MAX_DURATION_MS); + const duration = Math.max(0, (end ?? now) - start); const seconds = duration / 1000; const displaySeconds = seconds < 0.05 && end !== undefined ? 0.1 : seconds; @@ -798,6 +799,7 @@ interface ToolScrollableSectionProps { className?: string; outerClassName?: string; disableHorizontal?: boolean; + followKey?: string; } const ToolScrollableSection: React.FC = ({ @@ -806,22 +808,54 @@ const ToolScrollableSection: React.FC = ({ className, outerClassName, disableHorizontal = false, -}) => ( -
- -
- {children} -
- -
-); + followKey, +}) => { + const scrollRef = React.useRef(null); + const isFollowingRef = React.useRef(true); + + React.useLayoutEffect(() => { + const element = scrollRef.current; + if (followKey === undefined) { + isFollowingRef.current = true; + return; + } + if (!element || !isFollowingRef.current) { + return; + } + element.scrollTop = element.scrollHeight; + }, [followKey]); + + return ( +
+ { + if (followKey !== undefined && event.deltaY < 0) { + isFollowingRef.current = false; + } + }} + onScroll={(event) => { + if (followKey === undefined) { + return; + } + const element = event.currentTarget; + isFollowingRef.current = element.scrollHeight - element.scrollTop - element.clientHeight <= 2; + }} + className={cn( + 'tool-output-surface p-2 rounded-xl w-full min-w-0', + maxHeightClass, + disableHorizontal ? 'overflow-y-auto overflow-x-hidden' : 'overflow-auto', + className, + )} + > +
+ {children} +
+ +
+ ); +}; const getToolOutputLanguage = ( output: string, @@ -848,12 +882,53 @@ const getToolOutputText = ( return formatEditOutput(output, part.tool, metadata); }; +const StreamingPlainTextOutput: React.FC<{ output: string }> = ({ output }) => { + const preRef = React.useRef(null); + const previousOutputRef = React.useRef(''); + + React.useLayoutEffect(() => { + const element = preRef.current; + if (!element) { + return; + } + + const firstChild = element.firstChild; + const textNode = firstChild instanceof globalThis.Text + ? firstChild + : document.createTextNode(''); + if (textNode !== firstChild) { + element.replaceChildren(textNode); + } + + const append = getStreamingOutputAppend(previousOutputRef.current, output); + if (append === undefined) { + textNode.data = output; + } else if (append.length > 0) { + textNode.appendData(append); + } + previousOutputRef.current = output; + }, [output]); + + return ( + 
+    );
+};
+
 const ToolScrollableTextOutput: React.FC<{
     output: string;
     part: ToolPartType;
     metadata: Record | undefined;
     input: Record | undefined;
-}> = ({ output, part, metadata, input }) => {
+    isStreaming?: boolean;
+}> = ({ output, part, metadata, input, isStreaming = false }) => {
     const { t } = useI18n();
     const renderedOutput = getToolOutputText(output, part, metadata);
     const outputLanguage = getToolOutputLanguage(output, part, metadata, input);
@@ -884,6 +959,14 @@ const ToolScrollableTextOutput: React.FC<{
         }
     }, [renderedOutput, t]);
 
+    if (part.tool === 'bash' && isStreaming) {
+        return (
+            

+                
+            

+        );
+    }
+
     if (jsonResult.isJson) {
         return (
             

@@ -1509,9 +1592,17 @@ const ToolExpandedContent: React.FC = React.memo(({
     const stateWithData = state as ToolStateWithMetadata;
     const metadata = stateWithData.metadata;
     const input = stateWithData.input;
-    const rawOutput = stateWithData.output;
+    const rawOutput = getToolOutput(part.tool, stateWithData.output, metadata?.output);
     const hasStringOutput = typeof rawOutput === 'string' && rawOutput.length > 0;
-    const outputString = typeof rawOutput === 'string' ? rawOutput : '';
+    const rawOutputString = typeof rawOutput === 'string' ? rawOutput : '';
+    const isStreamingBash = part.tool === 'bash' && state.status === 'running';
+    const throttledOutputString = useStreamingTextThrottle({
+        text: rawOutputString,
+        isStreaming: isStreamingBash,
+        identityKey: part.id,
+        allowTextReplacement: isStreamingBash,
+    });
+    const outputString = isStreamingBash ? throttledOutputString : rawOutputString;
     const attachments = stateWithData.attachments;
     const fileDiff = isRecord(metadata?.filediff) ? metadata.filediff : undefined;
     const diffContent = getPatchText((metadata as { patch?: unknown } | undefined)?.patch)
@@ -1577,13 +1668,14 @@ const ToolExpandedContent: React.FC = React.memo(({
 
     const renderScrollableBlock = (
         content: React.ReactNode,
-        options?: { maxHeightClass?: string; className?: string; disableHorizontal?: boolean; outerClassName?: string }
+        options?: { maxHeightClass?: string; className?: string; disableHorizontal?: boolean; outerClassName?: string; followKey?: string }
     ) => (
         
             {content}
         
@@ -1799,16 +1891,22 @@ const ToolExpandedContent: React.FC = React.memo(({
         }
 
         if (hasStringOutput && outputString.trim()) {
-            return renderScrollableBlock(
+            const output = (
                 ,
+                    isStreaming={isStreamingBash}
+                />
+            );
+
+            return renderScrollableBlock(
+                output,
                 {
                     className: part.tool === 'bash' ? 'p-1 rounded-none' : 'p-1',
-                    maxHeightClass: part.tool === 'bash' ? 'max-h-[46vh]' : undefined,
+                    maxHeightClass: isStreamingBash ? 'h-[46vh]' : part.tool === 'bash' ? 'max-h-[46vh]' : undefined,
+                    followKey: isStreamingBash ? outputString : undefined,
                 }
             );
         }
@@ -1819,6 +1917,10 @@ const ToolExpandedContent: React.FC = React.memo(({
         );
     };
 
+    const hasVisibleOutput = outputString.trim().length > 0;
+    const shouldRenderResult = (state.status === 'completed' && 'output' in state)
+        || (part.tool === 'bash' && hasVisibleOutput);
+
     if (isTodoTool) {
         if (state.status === 'error' && 'error' in state) {
             return (
@@ -1897,7 +1999,7 @@ const ToolExpandedContent: React.FC = React.memo(({
                         

                     ) : null}
 
-                    {state.status === 'completed' && 'output' in state && (
+                    {shouldRenderResult && (
                         

                             {(part.tool === 'edit' || part.tool === 'multiedit' || part.tool === 'apply_patch' || part.tool === 'write') && hasVisualDiffEntry ? (
                                 

diff --git a/packages/ui/src/components/chat/message/parts/toolOutput.ts b/packages/ui/src/components/chat/message/parts/toolOutput.ts
new file mode 100644
index 00000000..88d1d0ab
--- /dev/null
+++ b/packages/ui/src/components/chat/message/parts/toolOutput.ts
@@ -0,0 +1,19 @@
+export const getToolOutput = (
+    tool: string,
+    stateOutput: unknown,
+    metadataOutput: unknown,
+): string | undefined => {
+    if (typeof stateOutput === 'string') {
+        return stateOutput;
+    }
+
+    if (tool === 'bash' && typeof metadataOutput === 'string' && metadataOutput.length > 0) {
+        return metadataOutput;
+    }
+
+    return undefined;
+};
+
+export const getStreamingOutputAppend = (previous: string, next: string): string | undefined => {
+    return next.startsWith(previous) ? next.slice(previous.length) : undefined;
+};
diff --git a/packages/web/server/index.js b/packages/web/server/index.js
index 7a6dce53..c0b7f8eb 100644
--- a/packages/web/server/index.js
+++ b/packages/web/server/index.js
@@ -893,6 +893,7 @@ const serverUtilsRuntime = createServerUtilsRuntime({
   getOpenCodeAuthHeaders,
   buildOpenCodeUrl,
   ensureOpenCodeApiPrefix,
+  getUpstreamStallTimeoutMs,
   getUiNotificationClients: () => uiNotificationClients,
   getOpenCodePort: () => openCodePort,
   setOpenCodePortState: (value) => {
diff --git a/packages/web/server/lib/opencode/DOCUMENTATION.md b/packages/web/server/lib/opencode/DOCUMENTATION.md
index 0f862970..e5041171 100644
--- a/packages/web/server/lib/opencode/DOCUMENTATION.md
+++ b/packages/web/server/lib/opencode/DOCUMENTATION.md
@@ -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
diff --git a/packages/web/server/lib/opencode/lifecycle.js b/packages/web/server/lib/opencode/lifecycle.js
index 2938bb8c..7a4f5cba 100644
--- a/packages/web/server/lib/opencode/lifecycle.js
+++ b/packages/web/server/lib/opencode/lifecycle.js
@@ -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 {
diff --git a/packages/web/server/lib/opencode/lifecycle.test.js b/packages/web/server/lib/opencode/lifecycle.test.js
index ee752525..6436f6e9 100644
--- a/packages/web/server/lib/opencode/lifecycle.test.js
+++ b/packages/web/server/lib/opencode/lifecycle.test.js
@@ -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();
diff --git a/packages/web/server/lib/opencode/proxy.js b/packages/web/server/lib/opencode/proxy.js
index 822a4bf8..6662f7bd 100644
--- a/packages/web/server/lib/opencode/proxy.js
+++ b/packages/web/server/lib/opencode/proxy.js
@@ -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) {
diff --git a/packages/web/server/lib/opencode/server-utils-runtime.js b/packages/web/server/lib/opencode/server-utils-runtime.js
index bf835849..c52c55c5 100644
--- a/packages/web/server/lib/opencode/server-utils-runtime.js
+++ b/packages/web/server/lib/opencode/server-utils-runtime.js
@@ -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,
     });
   };
diff --git a/packages/web/server/opencode-proxy.test.js b/packages/web/server/opencode-proxy.test.js
index 1abd4334..da8c7f72 100644
--- a/packages/web/server/opencode-proxy.test.js
+++ b/packages/web/server/opencode-proxy.test.js
@@ -82,6 +82,55 @@ describe('OpenCode proxy SSE forwarding', () => {
     expect(seenAuthorization).toBe('Bearer test-token');
   });
 
+  it('closes downstream SSE when the OpenCode upstream stalls despite proxy heartbeats', async () => {
+    let stallTimeoutReads = 0;
+    const upstream = express();
+    upstream.get('/global/event', (_req, res) => {
+      res.setHeader('Content-Type', 'text/event-stream');
+      res.flushHeaders();
+      setTimeout(() => res.write(':upstream-alive\n\n'), 40);
+      setTimeout(() => res.write('data: still-alive\n\n'), 80);
+    });
+    upstreamServer = await listen(upstream);
+    const upstreamPort = upstreamServer.address().port;
+
+    const app = express();
+    registerOpenCodeProxy(app, {
+      fs: {},
+      os: {},
+      path,
+      OPEN_CODE_READY_GRACE_MS: 0,
+      SSE_HEARTBEAT_INTERVAL_MS: 10,
+      getSseUpstreamStallTimeoutMs: () => {
+        stallTimeoutReads += 1;
+        return stallTimeoutReads === 1 ? 50 : 100;
+      },
+      getRuntime: () => ({
+        openCodePort: upstreamPort,
+        isOpenCodeReady: true,
+        openCodeNotReadySince: 0,
+        isRestartingOpenCode: false,
+      }),
+      getOpenCodeAuthHeaders: () => ({}),
+      buildOpenCodeUrl: (requestPath) => `http://127.0.0.1:${upstreamPort}${requestPath}`,
+      ensureOpenCodeApiPrefix: () => {},
+    });
+    proxyServer = await listen(app);
+    const proxyPort = proxyServer.address().port;
+
+    const response = await fetch(`http://127.0.0.1:${proxyPort}/api/global/event`, {
+      headers: { Accept: 'text/event-stream' },
+      signal: AbortSignal.timeout(2000),
+    });
+
+    expect(response.status).toBe(200);
+    const body = await response.text();
+    expect(body).toContain(':heartbeat\n\n');
+    expect(body).toContain(':upstream-alive\n\n');
+    expect(body).toContain('data: still-alive\n\n');
+    expect(stallTimeoutReads).toBeGreaterThanOrEqual(3);
+  });
+
   it('holds a request through OpenCode warmup and succeeds once ready (no 503/backoff)', async () => {
     const upstream = express();
     upstream.get('/config/providers', (_req, res) => {