fix(tui): report headless --print/--json failures instead of leaking tracebacks (#5963)

* fix(tui): report headless --print/--json failures instead of leaking tracebacks

The headless --print and --json runners dispatched straight into the
embedded client with no error boundary, so session-creation, thread-
resolution, chat, and stream failures escaped main() and surfaced as
unhandled Python tracebacks from the console script - for example the
documented CheckpointModeMismatchError case, which the checkpoint
operations/troubleshooting docs recorded as a raw traceback.

Both runners now catch Exception (KeyboardInterrupt/SystemExit still
propagate), mirroring the TUI's stream_actions() boundary:

- --print writes one concise 'Error: ...' line to stderr and returns 1
- --json appends a terminal {"type": "error"} record to the NDJSON
  stream (diagnostics stay on stderr) and returns 1; every stdout line
  remains independently valid JSON, including failures mid-stream

The checkpoint operations/troubleshooting docs (en + zh) no longer
describe the headless runners as printing an uncaught traceback.

Fixes #5962

* fix(tui): close the broken-pipe edge and document the --json error contract

Review follow-ups on #5963 (both non-blocking):

- print(answer) sat outside the headless boundary, so a consumer closing
  the pipe early (deerflow --print ... | head) still got a
  BrokenPipeError traceback plus the interpreter's shutdown-flush noise.
  The final --print write and the --json terminal error-record write now
  catch BrokenPipeError, point the stdout descriptor at devnull (the
  documented SIGPIPE-adjacent recipe) and exit 1 without a traceback.
- The terminal {"type": "error"} record is a real contract for --json
  scripters but was only documented in the checkpoint pages; the
  backend/docs/TUI.md command table and the README cheat line now state
  the error record and the exit-1 semantics where CLI consumers look.

README.md

@@ -2251,7 +2251,7 @@ deerflow --tui-transparent                    # use the terminal's default backg
 deerflow --continue                           # resume the most recent thread
 deerflow --resume THREAD                      # resume a thread by id
 deerflow --print "summarize this repo"        # headless one-shot answer to stdout
-deerflow --json  "hello"                       # headless newline-delimited StreamEvents
+deerflow --json  "hello"                       # headless newline-delimited StreamEvents; failure -> one {"type": "error"} record, exit 1
 deerflow --recursion-limit 250 --print "task" # override the headless agent-loop limit
 ```
 

backend/docs/TUI.md

@@ -26,8 +26,8 @@ Launch modes:
 | `deerflow chat` | Same TUI conversation surface |
 | `deerflow --continue` | Resume the most recent thread |
 | `deerflow --resume THREAD` | Resume a thread by id |
-| `deerflow --print "question"` | Headless one-shot answer to stdout |
-| `deerflow --json "question"` | Headless newline-delimited `StreamEvent`s |
+| `deerflow --print "question"` | Headless one-shot answer to stdout; on failure a concise stderr line and exit 1 |
+| `deerflow --json "question"` | Headless newline-delimited `StreamEvent`s; on failure one terminal `{"type": "error"}` record and exit 1 |
 | `deerflow --recursion-limit 250 --print "question"` | Set the headless agent-loop super-step limit |
 | `echo "q" \| deerflow --print` | Read the message from stdin |
 | `DEER_FLOW_TUI=1 deerflow` | Force the TUI via environment |

backend/packages/harness/deerflow/tui/cli.py

@@ -259,15 +259,43 @@ def _make_session():
     return open_session(persistence=False)
 
 
+def _error_text(exc: Exception) -> str:
+    """One-line user-facing error, matching the TUI's AssistantError convention."""
+    return str(exc) or type(exc).__name__
+
+
+def _silence_closed_stdout() -> None:
+    """Prevent the interpreter's shutdown flush from raising on a closed pipe.
+
+    A consumer that closes the pipe early (``deerflow --print ... | head``)
+    makes the final stdout write raise BrokenPipeError; Python then re-raises
+    the same error from the interpreter's shutdown flush of sys.stdout unless
+    the descriptor is pointed somewhere harmless first.
+    """
+    try:
+        devnull = os.open(os.devnull, os.O_WRONLY)
+        os.dup2(devnull, sys.stdout.fileno())
+    except (OSError, ValueError):
+        pass
+
+
 def _run_print(plan: LaunchPlan) -> int:
     message = _resolve_message(plan)
     if not message:
         print("No message provided.", file=sys.stderr)
         return 2
-    session = _make_session()
-    thread_id = session.resolve_thread(plan)
-    answer = session.client.chat(message, thread_id=thread_id, **_run_overrides(plan))
-    print(answer)
+    try:
+        session = _make_session()
+        thread_id = session.resolve_thread(plan)
+        answer = session.client.chat(message, thread_id=thread_id, **_run_overrides(plan))
+    except Exception as exc:  # noqa: BLE001 - headless boundary: report, never traceback
+        print(f"Error: {_error_text(exc)}", file=sys.stderr)
+        return 1
+    try:
+        print(answer)
+    except BrokenPipeError:
+        _silence_closed_stdout()
+        return 1
     return 0
 
 
@@ -276,12 +304,22 @@ def _run_json(plan: LaunchPlan) -> int:
     if not message:
         print("No message provided.", file=sys.stderr)
         return 2
-    session = _make_session()
-    thread_id = session.resolve_thread(plan)
-    for event in session.client.stream(message, thread_id=thread_id, **_run_overrides(plan)):
-        payload = {"type": event.type, "data": event.data}
-        sys.stdout.write(json.dumps(payload, ensure_ascii=False, default=str) + "\n")
-        sys.stdout.flush()
+    try:
+        session = _make_session()
+        thread_id = session.resolve_thread(plan)
+        for event in session.client.stream(message, thread_id=thread_id, **_run_overrides(plan)):
+            payload = {"type": event.type, "data": event.data}
+            sys.stdout.write(json.dumps(payload, ensure_ascii=False, default=str) + "\n")
+            sys.stdout.flush()
+    except Exception as exc:  # noqa: BLE001 - headless boundary: report, never traceback
+        error_text = _error_text(exc)
+        print(f"Error: {error_text}", file=sys.stderr)
+        try:
+            sys.stdout.write(json.dumps({"type": "error", "data": {"message": error_text}}, ensure_ascii=False, default=str) + "\n")
+            sys.stdout.flush()
+        except BrokenPipeError:
+            _silence_closed_stdout()
+        return 1
     return 0
 
 

backend/tests/test_tui_cli_main.py

@@ -1,6 +1,7 @@
 """Integration tests for ``main`` dispatch (headless paths), with a fake session."""
 
 import json
+from types import SimpleNamespace
 
 import pytest
 
@@ -87,3 +88,119 @@ def test_main_headless_help_returns_2_and_prints_usage(monkeypatch, capsys):
     rc = cli.main(["--cli"])
     assert rc == 2
     assert "deerflow" in capsys.readouterr().err
+
+
+# --------------------------------------------------------------------------- #
+# Headless error boundary: client/setup failures must not escape main() as
+# tracebacks. --print reports on stderr; --json appends one terminal error
+# record while keeping every stdout line valid NDJSON. KeyboardInterrupt and
+# SystemExit are not part of the boundary.
+# --------------------------------------------------------------------------- #
+
+
+def _raising_session(exc, *, stream_events_before_failure=0):
+    def fail(*args, **kwargs):
+        raise exc
+
+    def stream(message, *, thread_id=None, **kwargs):
+        for _ in range(stream_events_before_failure):
+            yield StreamEvent(type="messages-tuple", data={"type": "ai", "content": "hi", "id": "m1"})
+        raise exc
+
+    return SimpleNamespace(resolve_thread=lambda plan: "thread-1", client=SimpleNamespace(chat=fail, stream=stream))
+
+
+def test_main_print_client_failure_reports_error_without_traceback(monkeypatch, capsys):
+    session = _raising_session(RuntimeError("Thread requires delta mode"))
+    monkeypatch.setattr(cli, "_make_session", lambda: session)
+    rc = cli.main(["--print", "hello"])
+    captured = capsys.readouterr()
+    assert rc == 1
+    assert "Thread requires delta mode" in captured.err
+    assert captured.out == ""
+    assert "Traceback" not in captured.err
+
+
+def test_main_json_failure_before_first_event_emits_single_error_record(monkeypatch, capsys):
+    session = _raising_session(RuntimeError("Thread requires delta mode"))
+    monkeypatch.setattr(cli, "_make_session", lambda: session)
+    rc = cli.main(["--json", "hello"])
+    captured = capsys.readouterr()
+    assert rc == 1
+    payloads = [json.loads(ln) for ln in captured.out.splitlines() if ln.strip()]
+    assert payloads == [{"type": "error", "data": {"message": "Thread requires delta mode"}}]
+    assert "Thread requires delta mode" in captured.err
+
+
+def test_main_json_failure_after_event_keeps_every_line_valid_json(monkeypatch, capsys):
+    session = _raising_session(RuntimeError("checkpoint mode mismatch"), stream_events_before_failure=1)
+    monkeypatch.setattr(cli, "_make_session", lambda: session)
+    rc = cli.main(["--json", "hello"])
+    captured = capsys.readouterr()
+    assert rc == 1
+    payloads = [json.loads(ln) for ln in captured.out.splitlines() if ln.strip()]
+    assert payloads[0]["type"] == "messages-tuple"
+    assert payloads[-1] == {"type": "error", "data": {"message": "checkpoint mode mismatch"}}
+
+
+def test_main_session_creation_failure_reports_error_without_traceback(monkeypatch, capsys):
+    def fail():
+        raise RuntimeError("no checkpointer configured")
+
+    monkeypatch.setattr(cli, "_make_session", fail)
+    rc = cli.main(["--print", "hello"])
+    captured = capsys.readouterr()
+    assert rc == 1
+    assert "no checkpointer configured" in captured.err
+    assert "Traceback" not in captured.err
+
+
+def test_main_thread_resolution_failure_emits_json_error_record(monkeypatch, capsys):
+    def fail():
+        raise RuntimeError("thread lookup failed")
+
+    session = SimpleNamespace(resolve_thread=lambda plan: fail(), client=SimpleNamespace())
+    monkeypatch.setattr(cli, "_make_session", lambda: session)
+    rc = cli.main(["--json", "hello"])
+    captured = capsys.readouterr()
+    assert rc == 1
+    payloads = [json.loads(ln) for ln in captured.out.splitlines() if ln.strip()]
+    assert payloads[-1]["type"] == "error"
+
+
+def test_main_headless_boundary_preserves_keyboard_interrupt(monkeypatch):
+    session = _raising_session(KeyboardInterrupt())
+    monkeypatch.setattr(cli, "_make_session", lambda: session)
+    with pytest.raises(KeyboardInterrupt):
+        cli.main(["--print", "hello"])
+
+
+class _BrokenStdout:
+    """A stdout whose consumer closed the pipe early (e.g. ``| head``)."""
+
+    def isatty(self):
+        return False
+
+    def write(self, *_args, **_kwargs):
+        raise BrokenPipeError
+
+    def flush(self):
+        pass
+
+    def fileno(self):
+        raise OSError  # forces the devnull dup2 to take its best-effort path
+
+
+def test_main_print_broken_pipe_exits_without_traceback(monkeypatch):
+    monkeypatch.setattr(cli, "_make_session", _FakeSession)
+    monkeypatch.setattr(cli.sys, "stdout", _BrokenStdout())
+    rc = cli.main(["--print", "hello"])
+    assert rc == 1
+
+
+def test_main_json_broken_pipe_error_record_fails_silently(monkeypatch):
+    session = _raising_session(RuntimeError("boom"))
+    monkeypatch.setattr(cli, "_make_session", lambda: session)
+    monkeypatch.setattr(cli.sys, "stdout", _BrokenStdout())
+    rc = cli.main(["--json", "hello"])
+    assert rc == 1

frontend/src/content/en/harness/checkpoints/operations.mdx

@@ -206,7 +206,7 @@ Three consequences are worth stating plainly:
 
 - **Redis is rejected on the sync path.** Selecting `database.checkpoint_cache.type: redis` raises `ValueError("database.checkpoint_cache.type 'redis' is not supported on the sync checkpointer path (TUI/embedded); use 'memory'.")` at checkpointer construction, so a configuration that is valid for the Gateway can break the TUI. The rejection is honest rather than a fallback: the sync path is process-local anyway.
 - **There is no degraded read path.** Delta materialization needs the graph's channel table, and the TUI has no equivalent to the Gateway's raw-checkpoint fallback.
-- **An uncaught mismatch is a traceback on the headless runners.** The embedded client raises `CheckpointModeMismatchError` in-process — no HTTP status, no run record, no SSE frame — and the CLI dispatches straight to its print or JSON runner without a handler, so `--print` and `--json` fail with a Python traceback on stderr. The Textual TUI is the exception: its `stream_actions()` helper catches every exception from `client.stream()` and emits `AssistantError` plus `RunEnded`, so an interactive user sees the failure in the UI instead.
+- **A mismatch is a reported error on every surface.** The embedded client raises `CheckpointModeMismatchError` in-process — no HTTP status, no run record, no SSE frame. The CLI's headless runners catch it: `--print` prints a concise `Error: ...` line on stderr and exits non-zero, and `--json` appends a terminal `{"type": "error"}` record to the NDJSON stream (diagnostics stay on stderr) and exits non-zero. The Textual TUI's `stream_actions()` helper catches the same exception and emits `AssistantError` plus `RunEnded`, so an interactive user sees the failure in the UI.
 
 Third-party embedders hit a related guard: `create_deerflow_agent` with `checkpoint_channel_mode="delta"` **and** a checkpointer is rejected with `ValueError`, because graphs built there bypass checkpoint-mode marker injection and the fail-closed compatibility gate, so a mixed-mode store would silently corrupt thread state. Delta without a checkpointer is ephemeral and allowed; persisted delta graphs must go through the guarded application paths (`make_lead_agent` or `DeerFlowClient`).
 

frontend/src/content/en/harness/checkpoints/troubleshooting.mdx

@@ -42,7 +42,7 @@ This chapter is organized by **what you see**. Each entry gives the cause, the f
 
 The threads router does not pre-gate run creation, so a mismatch there fails asynchronously: `POST .../runs` returns a `200` run record that later turns `status: error`, `POST .../runs/stream` returns `200 text/event-stream` and then an `event: error` frame, and `POST .../runs/wait` returns `200` JSON with `{"status": "error", "error": "<the mismatch text>"}`. One run path reads a checkpoint before admission, though: when a thread's run-event feed is empty but a checkpoint head exists, `ensure_checkpoint_history_seeded` materializes that head inside `start_run`, where only `ConflictError` and `UnsupportedStrategyError` are mapped, so the mismatch escapes as an unhandled `500` on `POST /api/threads/{thread_id}/runs`, `/runs/stream`, and `/runs/wait`. With the default `run_events.backend: memory` a Gateway restart empties every existing thread's feed, so that branch is reachable after a restart.
 
-In the embedded TUI the client raises in-process before the graph streams: the Textual runner turns it into an `AssistantError` plus `RunEnded` pair, while the headless `--print` and `--json` runners have no handler and print an uncaught traceback instead of a status code.
+In the embedded TUI the client raises in-process before the graph streams: the Textual runner turns it into an `AssistantError` plus `RunEnded` pair, and the headless runners handle it the same way — `--print` prints a concise `Error: ...` line on stderr and exits non-zero, while `--json` appends a terminal `{"type": "error"}` record to the NDJSON stream (diagnostics stay on stderr) and exits non-zero.
 
 **Fix**: pick one of two exits.
 

frontend/src/content/zh/harness/checkpoints/operations.mdx

@@ -200,7 +200,7 @@ PYTHONPATH=. uv run python scripts/benchmark/checkpoint/summarize_production.py
 
 - **同步路径拒绝 redis。** 选择 `database.checkpoint_cache.type: redis` 会在构建 checkpointer 时抛出 `ValueError("database.checkpoint_cache.type 'redis' is not supported on the sync checkpointer path (TUI/embedded); use 'memory'.")`，因此对网关合法的配置可能让 TUI 起不来。这是诚实的拒绝而非降级：同步路径本来就是进程内的。
 - **没有降级读取路径。** 增量物化需要图的通道表，TUI 没有网关那种退回原始 checkpointer 读取的等价物。
-- **未捕获的模式不一致在无头入口上会变成 traceback。** 内嵌客户端在进程内抛出 `CheckpointModeMismatchError`——没有 HTTP 状态码、没有运行记录、没有 SSE 帧——而 CLI 直接分派到它的 print 或 JSON 入口且没有处理器，于是 `--print`、`--json` 在 stderr 上打印一段 Python traceback。Textual TUI 是例外：它的 `stream_actions()` 会捕获来自 `client.stream()` 的每个异常并发出 `AssistantError` 与 `RunEnded`，所以交互用户是在界面里看到这次失败。
+- **模式不一致在所有入口上都会作为错误上报。** 内嵌客户端在进程内抛出 `CheckpointModeMismatchError`——没有 HTTP 状态码、没有运行记录、没有 SSE 帧。CLI 的无头入口会捕获它：`--print` 在 stderr 上打印一行简洁的 `Error: ...` 并以非零码退出，`--json` 向 NDJSON 流追加一条终止性的 `{"type": "error"}` 记录（诊断文字留在 stderr）并以非零码退出。Textual TUI 的 `stream_actions()` 会捕获同一个异常并发出 `AssistantError` 与 `RunEnded`，交互用户在界面里看到这次失败。
 
 第三方嵌入者还会撞上相关守卫：`create_deerflow_agent` 在 `checkpoint_channel_mode="delta"` **且**传入 checkpointer 时抛出 `ValueError`，因为在那里构建的图绕过了检查点模式标记注入与失败关闭的兼容门，混合模式的存储会静默损坏线程状态。不带 checkpointer 的 delta 是临时的，允许使用；需要持久化的增量图必须走受守卫的应用路径（`make_lead_agent` 或 `DeerFlowClient`）。
 

frontend/src/content/zh/harness/checkpoints/troubleshooting.mdx

@@ -42,7 +42,7 @@ import { Callout } from "nextra/components";
 
 threads 路由不会对创建 run 提前门禁，因此那里的不匹配改为异步失败：`POST .../runs` 返回 `200` run 记录，随后变为 `status: error`；`POST .../runs/stream` 返回 `200 text/event-stream` 后发出 `event: error` 帧；`POST .../runs/wait` 返回 `200` JSON，内容为 `{"status": "error", "error": "<不匹配文本>"}`。不过有一条 run 路径会在准入前读取检查点：当线程的 run event 流为空但存在检查点头部时，`ensure_checkpoint_history_seeded` 会在 `start_run` 内物化该头部，而那里只映射 `ConflictError` 与 `UnsupportedStrategyError`，因此不匹配会以未处理的 `500` 逃出，作用于 `POST /api/threads/{thread_id}/runs`、`/runs/stream` 与 `/runs/wait`。默认的 `run_events.backend: memory` 会让 Gateway 重启后每个既有线程的流都变空，所以这条分支在重启后可达。
 
-在内嵌 TUI 里，client 在图开始流式输出前就在进程内抛出：Textual runner 会把它变成 `AssistantError` 与 `RunEnded` 的组合，而无头的 `--print`、`--json` runner 没有对应处理器，会打印未捕获的 traceback，而不是状态码。
+在内嵌 TUI 里，client 在图开始流式输出前就在进程内抛出：Textual runner 会把它变成 `AssistantError` 与 `RunEnded` 的组合；无头入口的处理方式一致——`--print` 在 stderr 上打印一行简洁的 `Error: ...` 并以非零码退出，`--json` 向 NDJSON 流追加一条终止性的 `{"type": "error"}` 记录（诊断文字留在 stderr）并以非零码退出。
 
 **处理**：二选一。