fix(cli): graceful fallback when no Win32 console buffer (headless services)

cli.py

@@ -71,6 +71,50 @@
 except (ImportError, AttributeError):
     _STEADY_CURSOR = None
 
+# --------------------------------------------------------------------------
+# Console-less environment safety net.
+#
+# When hermes runs as a child of a Windows service (NSSM / LocalSystem), the
+# child process has no Win32 console buffer. prompt_toolkit's
+# print_formatted_text() lazily calls create_output() which then raises
+# NoConsoleScreenBufferError, killing the process before AIAgent even starts.
+# We wrap _pt_print so any console-related failure falls back to plain
+# sys.stdout.write — the dispatcher captures stdout regardless, so output is
+# preserved and the agent loop can actually run.
+# --------------------------------------------------------------------------
+try:
+    from prompt_toolkit.output.win32 import NoConsoleScreenBufferError as _PT_NoConsole
+except Exception:
+    class _PT_NoConsole(Exception):  # type: ignore[no-redef]
+        pass
+
+import sys as _sys_for_pt_fallback
+
+
+def _safe_pt_print(text: str) -> None:
+    """print_formatted_text with a plain-stdout fallback for headless services.
+
+    prompt_toolkit fails fatally on Windows when there's no console buffer
+    (e.g. running under a Windows service). We catch that specific error
+    plus any unexpected runtime explosion and degrade gracefully so the
+    process can keep going.
+    """
+    try:
+        _pt_print(_PT_ANSI(text))
+    except _PT_NoConsole:
+        try:
+            _sys_for_pt_fallback.stdout.write(str(text) + "\n")
+            _sys_for_pt_fallback.stdout.flush()
+        except Exception:
+            pass
+    except Exception:
+        # Any other prompt_toolkit/IO failure: don't let a print kill the agent.
+        try:
+            _sys_for_pt_fallback.stdout.write(str(text) + "\n")
+            _sys_for_pt_fallback.stdout.flush()
+        except Exception:
+            pass
+
 try:
     from hermes_cli.pt_input_extras import install_shift_enter_alias, install_ctrl_enter_alias
     install_shift_enter_alias()
@@ -1844,7 +1888,7 @@ def _cprint(text: str):
     try:
         from prompt_toolkit.application import get_app_or_none, run_in_terminal
     except Exception:
-        _pt_print(_PT_ANSI(text))
+        _safe_pt_print(text)
         return
 
     app = None
@@ -1857,24 +1901,19 @@ def _cprint(text: str):
     # direct prompt_toolkit print is safe and matches existing behavior
     # (spinner frames, streamed tokens, tool activity prefixes, …).
     if app is None or not getattr(app, "_is_running", False):
-        try:
-            _pt_print(_PT_ANSI(text))
-        except Exception:
-            # Fallback when stdout is not a real console (e.g. subprocess
-            # worker logging to a file). prompt_toolkit raises
-            # NoConsoleScreenBufferError (Windows) or OSError (other).
-            try:
-                print(text)
-            except Exception:
-                pass
+        # _safe_pt_print already handles NoConsoleScreenBufferError (Windows
+        # headless services) / OSError (no real stdout console) by falling
+        # back to a bare print(). Keeps headless-service path working while
+        # preserving main's fallback intent.
+        _safe_pt_print(text)
         return
 
     try:
         loop = app.loop  # type: ignore[attr-defined]
     except Exception:
         loop = None
     if loop is None:
-        _pt_print(_PT_ANSI(text))
+        _safe_pt_print(text)
         return
 
     import asyncio as _asyncio
@@ -1890,7 +1929,7 @@ def _cprint(text: str):
         current_loop = None
     # Same thread as the app's loop → safe to print directly.
     if current_loop is loop and loop.is_running():
-        _pt_print(_PT_ANSI(text))
+        _safe_pt_print(text)
         return
 
     # Cross-thread emission: ask the app's event loop to schedule a
@@ -1909,9 +1948,14 @@ def _schedule():
         # because run_in_terminal already invoked the lambda in that case
         # (the mock path), which would double-print the line.
         try:
+            # Use ensure_future to actually await run_in_terminal's coroutine
+            # (prompt_toolkit ≥ 3.0). Leaving it unawaited silently drops the
+            # output (fixes #23185 Bug A). Use _safe_pt_print inside the
+            # lambda so headless-service callers (no Win32 console buffer)
+            # still degrade to bare print() instead of crashing.
             import asyncio as _aio
             import inspect as _inspect
-            coro = run_in_terminal(lambda: _pt_print(_PT_ANSI(text)))
+            coro = run_in_terminal(lambda: _safe_pt_print(text))
             if coro is not None and (_inspect.isawaitable(coro) or _inspect.iscoroutine(coro)):
                 _aio.ensure_future(coro)
             # else: run_in_terminal ran the lambda synchronously; nothing more
@@ -1923,7 +1967,7 @@ def _schedule():
         loop.call_soon_threadsafe(_schedule)
     except Exception:
         try:
-            _pt_print(_PT_ANSI(text))
+            _safe_pt_print(text)
         except Exception:
             pass