fix(tests): harden env isolation and replace remaining sleep-sync races

The full 42k-test run and complete npm check surfaced three more classes:

- Environment isolation: local ~/.honcho defaultHost and SSH_* variables
  leaked into Python/TUI tests. Pin the default Honcho host in the
  hermetic fixture, isolate the one fallback test from ~/.honcho, and
  blank SSH_* around terminalSetup tests. This flipped 20 false failures
  back to deterministic behavior on developer machines.
- Background-thread sleep-sync: Honcho async writer tests patched
  time.sleep globally, then busy-polled with that same mocked sleep. Under
  full-suite load the poller could starve the writer. Each test now waits
  on an Event emitted by the exact flush/retry transition; 30/30 passed
  under 15-way contention.
- Desktop streaming: the test slept 80ms and assumed a 500ms timer could
  not fire before its assertion. A loaded runner descheduled the test for
  >500ms and both chunks arrived. Producer controls now gate second-chunk
  and completion transitions explicitly.

Also make file-retry observability complete: a self-healed flaky file now
prints BOTH attempts' full output in the FLAKY summary. Two behavioral
runner tests prove pass-on-retry is green+loud+traceback-preserving, while
a deterministic failure remains red.

apps/desktop/src/components/assistant-ui/thread/streaming.test.tsx

@@ -243,7 +243,12 @@ function assistantImageMessage(running = false): ThreadMessage {
   } as ThreadMessage
 }
 
-function StreamingHarness() {
+interface StreamingControls {
+  emitSecond: () => void
+  complete: () => void
+}
+
+function StreamingHarness({ onControls }: { onControls?: (controls: StreamingControls) => void } = {}) {
   const [messages, setMessages] = useState<ThreadMessage[]>([userMessage()])
   const [isRunning, setIsRunning] = useState(true)
 
@@ -252,6 +257,19 @@ function StreamingHarness() {
       setMessages([userMessage(), assistantMessage('first chunk')])
     }, 50)
 
+    if (onControls) {
+      onControls({
+        emitSecond: () => {
+          setMessages([userMessage(), assistantMessage('first chunk second chunk')])
+        },
+        complete: () => {
+          setMessages([userMessage(), assistantMessage('first chunk second chunk', false)])
+          setIsRunning(false)
+        }
+      })
+      return () => window.clearTimeout(first)
+    }
+
     const second = window.setTimeout(() => {
       setMessages([userMessage(), assistantMessage('first chunk second chunk')])
     }, 500)
@@ -266,7 +284,7 @@ function StreamingHarness() {
       window.clearTimeout(second)
       window.clearTimeout(complete)
     }
-  }, [])
+  }, [onControls])
 
   const runtime = useExternalStoreRuntime<ThreadMessage>({
     messages,
@@ -399,26 +417,30 @@ describe('assistant-ui streaming renderer', () => {
   })
 
   it('renders assistant text incrementally before completion', async () => {
-    const { container } = render(<StreamingHarness />)
+    let controls: StreamingControls | undefined
+    const registerControls = (next: StreamingControls) => {
+      controls = next
+    }
+    const { container } = render(<StreamingHarness onControls={registerControls} />)
 
     expect(screen.getByRole('status', { name: 'Hermes is loading a response' })).toBeTruthy()
 
-    await wait(80)
-
     await waitFor(() => {
       expect(container.textContent).toContain('first chunk')
     })
     expect(container.textContent).not.toContain('second chunk')
     expect(screen.queryByRole('status', { name: 'Hermes is loading a response' })).toBeNull()
 
-    await wait(500)
-
+    // Producer-gated, not wall-clock-gated: the old test slept 80ms and
+    // assumed a 500ms timer could not fire before the assertion. On a loaded
+    // runner the test thread could be descheduled for >500ms, so both chunks
+    // arrived and this clean behavior test flaked.
+    act(() => controls?.emitSecond())
     await waitFor(() => {
       expect(container.textContent).toContain('first chunk second chunk')
     })
 
-    await wait(250)
-
+    act(() => controls?.complete())
     await waitFor(() => {
       expect(container.textContent).toContain('first chunk second chunk')
     })

scripts/run_tests_parallel.py

@@ -241,8 +241,8 @@ def _run_one_file(
 
     ``retries`` > 0 enables the one-shot flake retry: a non-zero exit is
     re-run in a fresh subprocess; if the re-run passes, the file counts as
-    passed but the output is prefixed with a FLAKY banner and the file is
-    recorded in ``_FLAKY_FILES`` so the summary can call it out. A
+    passed but the output is prefixed with a FLAKY banner and the file/output
+    are recorded in ``_FLAKY_RESULTS`` so the summary can call it out. A
     deterministic failure fails every attempt, so real regressions cannot
     be laundered green.
 
@@ -279,19 +279,22 @@ def _run_one_file(
         )
         subproc_wall += subproc_wall2
         if rc == 0:
-            with _flaky_lock:
-                _FLAKY_FILES.append(file)
             output = (
                 f"⚠ FLAKY: failed on attempt 1, passed on retry "
                 f"(attempt {attempt + 1}). Fix the flake — do not ignore this.\n"
                 f"--- first-attempt output ---\n{first_output}\n"
                 f"--- retry output ---\n{output}"
             )
+            with _flaky_lock:
+                _FLAKY_RESULTS.append((file, output))
     return file, rc, output, summary, subproc_wall
 
 
-# Files that failed once and passed on retry — reported in the summary.
-_FLAKY_FILES: List[Path] = []
+# Files that failed once and passed on retry, with both attempts' output.
+# Keeping the traceback is load-bearing: a self-healed flake without its
+# failing assertion is only a filename, which forces another expensive full
+# run to rediscover the race.
+_FLAKY_RESULTS: List[Tuple[Path, str]] = []
 _flaky_lock = threading.Lock()
 
 
@@ -958,11 +961,12 @@ def main() -> int:
 
     # Flaky files: failed once, passed on the automatic retry. Green, but
     # loudly reported so they get fixed instead of silently re-flaking.
-    if _FLAKY_FILES:
+    if _FLAKY_RESULTS:
         print()
-        print(f"=== ⚠ {len(_FLAKY_FILES)} FLAKY file{'s' if len(_FLAKY_FILES) != 1 else ''} (failed once, passed on retry — fix these) ===")
-        for f in _FLAKY_FILES:
+        print(f"=== ⚠ {len(_FLAKY_RESULTS)} FLAKY file{'s' if len(_FLAKY_RESULTS) != 1 else ''} (failed once, passed on retry — fix these) ===")
+        for f, output in _FLAKY_RESULTS:
             print(f"  {_format_file(f, repo_root)}")
+            print(output.rstrip())
 
     # Save durations for future --slice runs. Each slice writes its own
     # partial test_durations.json; a CI merge step joins them later.

tests/conftest.py

@@ -214,6 +214,10 @@ _HERMES_BEHAVIORAL_VARS = frozenset({
     "HERMES_KANBAN_CLAIM_LOCK",
     "HERMES_KANBAN_DISPATCH_IN_GATEWAY",
     "HERMES_TENANT",
+    # Honcho host selection changes which nested config block wins. A local
+    # shell override leaked "myhost" into the full suite and flipped 20
+    # otherwise-unrelated config tests away from the default "hermes" host.
+    "HERMES_HONCHO_HOST",
     # Dashboard OAuth auth gate (PR #30156). When set, the bundled
     # dashboard-auth `nous` plugin auto-registers itself on plugin discovery,
     # which is triggered by any `/api/status` call. That leaks a provider
@@ -342,6 +346,13 @@ def _hermetic_environment(tmp_path, monkeypatch):
     for name in _HERMES_BEHAVIORAL_VARS:
         monkeypatch.delenv(name, raising=False)
 
+    # Honcho's fallback host/config resolution legitimately reads the user's
+    # global ~/.honcho/config.json. Keep HOME stable (subprocess tests depend
+    # on it), but pin the host so ordinary tests cannot inherit a developer's
+    # defaultHost and silently select the wrong nested config block. Tests of
+    # custom host resolution override/delete this explicitly.
+    monkeypatch.setenv("HERMES_HONCHO_HOST", "hermes")
+
     # 3. Redirect HERMES_HOME to a per-test tempdir. Code that reads
     #    ``~/.hermes/*`` via ``get_hermes_home()`` now gets the tempdir.
     #

tests/honcho_plugin/test_async_memory.py

@@ -10,7 +10,7 @@ Covers:
 """
 
 import json
-import time
+import threading
 from unittest.mock import MagicMock, patch
 
 
@@ -312,17 +312,16 @@ class TestAsyncWriterThread:
         sess.add_message("user", "async msg")
 
         flushed = []
+        flushed_event = threading.Event()
 
-        def capture(s):
-            flushed.append(s)
+        def capture(session):
+            flushed.append(session)
+            flushed_event.set()
             return True
 
         mgr._flush_session = capture
         mgr._async_queue.put(sess)
-        # Give the daemon thread time to process
-        deadline = time.time() + 2.0
-        while not flushed and time.time() < deadline:
-            time.sleep(0.05)
+        assert flushed_event.wait(timeout=10), "async writer never flushed"
 
         mgr.shutdown()
         assert len(flushed) == 1
@@ -332,7 +331,7 @@ class TestAsyncWriterThread:
         mgr = _make_manager(write_frequency="async")
         thread = mgr._async_thread
         mgr.shutdown()
-        thread.join(timeout=3)
+        thread.join(timeout=10)
         assert not thread.is_alive()
 
 
@@ -347,20 +346,20 @@ class TestAsyncWriterRetry:
         sess.add_message("user", "msg")
 
         call_count = [0]
+        retry_done = threading.Event()
 
-        def flaky_flush(s):
+        def flaky_flush(session):
             call_count[0] += 1
             if call_count[0] == 1:
                 raise ConnectionError("network blip")
-            # second call succeeds silently
+            retry_done.set()
+            return True
 
         mgr._flush_session = flaky_flush
 
         with patch("time.sleep"):  # skip the 2s sleep in retry
             mgr._async_queue.put(sess)
-            deadline = time.time() + 3.0
-            while call_count[0] < 2 and time.time() < deadline:
-                time.sleep(0.05)
+            assert retry_done.wait(timeout=10), "async writer never retried"
 
         mgr.shutdown()
         assert call_count[0] == 2
@@ -371,18 +370,19 @@ class TestAsyncWriterRetry:
         sess.add_message("user", "msg")
 
         call_count = [0]
+        retry_done = threading.Event()
 
-        def always_fail(s):
+        def always_fail(session):
             call_count[0] += 1
+            if call_count[0] >= 2:
+                retry_done.set()
             raise RuntimeError("always broken")
 
         mgr._flush_session = always_fail
 
         with patch("time.sleep"):
             mgr._async_queue.put(sess)
-            deadline = time.time() + 3.0
-            while call_count[0] < 2 and time.time() < deadline:
-                time.sleep(0.05)
+            assert retry_done.wait(timeout=10), "async writer never retried"
 
         mgr.shutdown()
         # Should have tried exactly twice (initial + one retry) and not crashed
@@ -395,18 +395,19 @@ class TestAsyncWriterRetry:
         sess.add_message("user", "msg")
 
         call_count = [0]
+        retry_done = threading.Event()
 
-        def fail_then_succeed(_session):
+        def fail_then_succeed(session):
             call_count[0] += 1
+            if call_count[0] >= 2:
+                retry_done.set()
             return call_count[0] > 1
 
         mgr._flush_session = fail_then_succeed
 
         with patch("time.sleep"):
             mgr._async_queue.put(sess)
-            deadline = time.time() + 3.0
-            while call_count[0] < 2 and time.time() < deadline:
-                time.sleep(0.05)
+            assert retry_done.wait(timeout=10), "async writer never retried"
 
         mgr.shutdown()
         assert call_count[0] == 2

tests/honcho_plugin/test_client.py

@@ -507,7 +507,10 @@ class TestResolveActiveHost:
 
     def test_profiles_import_failure_falls_back(self):
         import sys
-        with patch.dict(os.environ, {}, clear=False):
+        with patch.dict(os.environ, {}, clear=False), patch(
+            "plugins.memory.honcho.client.resolve_config_path",
+            return_value=Path("/nonexistent/test-honcho-config.json"),
+        ):
             os.environ.pop("HERMES_HONCHO_HOST", None)
             # Temporarily remove hermes_cli.profiles to simulate import failure
             saved = sys.modules.get("hermes_cli.profiles")

tests/test_run_tests_parallel.py

@@ -277,3 +277,85 @@ def test_positional_path_not_treated_as_flag(tmp_path: Path) -> None:
     # Discovery found the probe file (2 tests), proving the positional path
     # was consumed as a root, not forwarded to pytest as a bad flag.
     assert "test_flagprobe.py" in proc.stdout, proc.stdout
+
+
+def test_file_retry_self_heals_and_prints_both_attempts(tmp_path: Path) -> None:
+    """A pass-on-retry is green, loud, and retains the failing traceback."""
+    repo_root = Path(__file__).resolve().parent.parent
+    runner = repo_root / "scripts" / "run_tests_parallel.py"
+    marker = tmp_path / "ran-once"
+    probe = tmp_path / "test_flaky_probe.py"
+    probe.write_text(
+        textwrap.dedent(
+            f"""
+            from pathlib import Path
+
+            def test_flaky_once():
+                marker = Path({str(marker)!r})
+                if not marker.exists():
+                    marker.write_text("failed once")
+                    assert False, "simulated first-attempt flake"
+                assert True
+            """
+        ),
+        encoding="utf-8",
+    )
+
+    proc = subprocess.run(
+        [
+            sys.executable,
+            str(runner),
+            "--files",
+            str(probe),
+            "--file-retries",
+            "1",
+            "-j",
+            "1",
+            "-q",
+        ],
+        cwd=repo_root,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        timeout=60,
+    )
+
+    assert proc.returncode == 0, proc.stdout
+    assert "FLAKY file" in proc.stdout
+    assert "simulated first-attempt flake" in proc.stdout
+    assert "first-attempt output" in proc.stdout
+    assert "retry output" in proc.stdout
+
+
+def test_file_retry_does_not_launder_deterministic_failure(tmp_path: Path) -> None:
+    """A real regression fails both attempts and the runner remains red."""
+    repo_root = Path(__file__).resolve().parent.parent
+    runner = repo_root / "scripts" / "run_tests_parallel.py"
+    probe = tmp_path / "test_red_probe.py"
+    probe.write_text(
+        "def test_always_red():\n    assert False, 'deterministic regression'\n",
+        encoding="utf-8",
+    )
+
+    proc = subprocess.run(
+        [
+            sys.executable,
+            str(runner),
+            "--files",
+            str(probe),
+            "--file-retries",
+            "1",
+            "-j",
+            "1",
+            "-q",
+        ],
+        cwd=repo_root,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        timeout=60,
+    )
+
+    assert proc.returncode == 1, proc.stdout
+    assert "deterministic regression" in proc.stdout
+    assert "FLAKY file" not in proc.stdout

ui-tui/src/__tests__/terminalSetup.test.ts

@@ -1,4 +1,4 @@
-import { describe, expect, it, vi } from 'vitest'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
 import {
   configureDetectedTerminalKeybindings,
@@ -9,6 +9,19 @@ import {
   stripJsonComments
 } from '../lib/terminalSetup.js'
 
+// Tests run from developer shells as well as CI. An inherited SSH_* variable
+// must not silently force every configure call down the remote-session reject
+// path; remote behavior is tested explicitly with per-call env objects below.
+beforeEach(() => {
+  vi.stubEnv('SSH_CONNECTION', '')
+  vi.stubEnv('SSH_TTY', '')
+  vi.stubEnv('SSH_CLIENT', '')
+})
+
+afterEach(() => {
+  vi.unstubAllEnvs()
+})
+
 describe('terminalSetup helpers', () => {
   it('detects VS Code family terminals from environment', () => {
     expect(detectVSCodeLikeTerminal({ CURSOR_TRACE_ID: 'x' } as NodeJS.ProcessEnv)).toBe('cursor')