diff --git a/README.md b/README.md index fec4f6e355..8b7bf0cc37 100644 --- a/README.md +++ b/README.md @@ -2266,6 +2266,8 @@ deerflow --json "hello" # headless newline-delimited Stre deerflow --recursion-limit 250 --print "task" # override the headless agent-loop limit ``` +Headless `--print` and `--json` exit with status `1` when the run fails, including provider errors returned as fallback messages. `--print` still writes the fallback text to stdout; `--json` appends a terminal error record. + A keyboard-driven chat surface with a streaming transcript (Markdown-rendered answers), compact tool-activity cards, a `/` slash-command palette, display-only `/clear`, `/goal` goal management, `/model` and `/threads` pickers, input history, PageUp/PageDown transcript navigation, and `Esc` / `Ctrl+C` interrupt. Transcript refreshes preserve your reading position after you scroll upward and resume following new output when you return to the bottom. `/clear` removes rows from the current terminal display without deleting the thread or its persisted conversation; `/new` and `/clear` ask you to wait during an active run instead of resetting in-flight display state. Sessions opened in the TUI also appear in the Web UI sidebar — it writes the shared thread store under the local default user, so terminal and web stay in sync **without running the Gateway**. See [backend/docs/TUI.md](backend/docs/TUI.md) for the full guide. diff --git a/backend/docs/TUI.md b/backend/docs/TUI.md index e5d8ce6027..8e6c5edf58 100644 --- a/backend/docs/TUI.md +++ b/backend/docs/TUI.md @@ -36,6 +36,13 @@ Launch modes: If no TTY is available and no headless flag is given, `deerflow` prints guidance instead of hanging. +Provider failures (for example an expired credential) usually do not raise: the +LLM error middleware turns them into a final AI message flagged +`deerflow_error_fallback`. Headless runs treat that as a failure too — `--print` +still writes the fallback text to stdout, then prints +`Error: LLM request failed (error_type=…, error_reason=…)` to stderr; `--json` +appends the same terminal `{"type": "error"}` record. Both exit `1`. + Transparent rendering is opt-in; the solid DeerFlow palette remains the default. The transparent mode uses Textual's `ansi_default` background for the main screen, header, transcript, status, palette, composer, and modal surfaces while diff --git a/backend/packages/harness/deerflow/AGENTS.md b/backend/packages/harness/deerflow/AGENTS.md index 737721d834..73412a6187 100644 --- a/backend/packages/harness/deerflow/AGENTS.md +++ b/backend/packages/harness/deerflow/AGENTS.md @@ -52,10 +52,10 @@ drift. ### Embedded Client (`packages/harness/deerflow/client.py`) -`DeerFlowClient` provides in-process access without HTTP/FastAPI, sharing Gateway's `deerflow` modules, config, data directories, and response schemas. +`DeerFlowClient` embeds the harness without HTTP/FastAPI, sharing Gateway modules, config, data directories, and response schemas. **Agent Conversation**: -- `chat(message, thread_id)` — synchronous, accumulates streaming deltas per message-id and returns the final AI text +- `chat(message, thread_id)` — synchronous; returns final AI text. - `stream(message, thread_id)` — subscribes to LangGraph `stream_mode=["values", "messages", "custom"]` and yields `StreamEvent`: - `"values"` — state snapshot (title, messages, artifacts, summary_text). Always forward `summary_text` (current summary or `None`), including unchanged values/resets. Never re-emit AI text delivered via `messages`; serialized `ToolMessage` entries retain non-`None` native `artifact` - `"messages-tuple"` — current-turn AI text **deltas** by `id` and each tool call/result once; excludes resumed history and preserves result `artifact` diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py index 64b936130f..ece067c452 100644 --- a/backend/packages/harness/deerflow/client.py +++ b/backend/packages/harness/deerflow/client.py @@ -146,6 +146,30 @@ class StreamEvent: data: dict[str, Any] = field(default_factory=dict) +class _AIMessageAccumulator: + """Accumulate text for the last content-bearing AI message id. + + Metadata-only updates do not select a new id. Both ``chat()`` and the + headless CLI use this rule to choose the final answer. + """ + + def __init__(self) -> None: + # Join once at the end to avoid quadratic copying of long responses. + self._chunks: dict[str, list[str]] = {} + self.last_id = "" + + def observe(self, event: StreamEvent) -> None: + if event.type != "messages-tuple" or event.data.get("type") != "ai": + return + if delta := event.data.get("content", ""): + msg_id = event.data.get("id") or "" + self._chunks.setdefault(msg_id, []).append(delta) + self.last_id = msg_id + + def answer(self) -> str: + return "".join(self._chunks.get(self.last_id, ())) + + class DeerFlowClient: """Embedded Python client for DeerFlow agent system. @@ -1289,18 +1313,10 @@ class DeerFlowClient: The accumulated text of the last AI message, or empty string if no AI text was produced. """ - # Per-id delta lists joined once at the end — avoids the O(n²) cost - # of repeated ``str + str`` on a growing buffer for long responses. - chunks: dict[str, list[str]] = {} - last_id: str = "" + answer = _AIMessageAccumulator() for event in self.stream(message, thread_id=thread_id, **kwargs): - if event.type == "messages-tuple" and event.data.get("type") == "ai": - msg_id = event.data.get("id") or "" - delta = event.data.get("content", "") - if delta: - chunks.setdefault(msg_id, []).append(delta) - last_id = msg_id - return "".join(chunks.get(last_id, ())) + answer.observe(event) + return answer.answer() # ------------------------------------------------------------------ # Public API — configuration queries diff --git a/backend/packages/harness/deerflow/tui/AGENTS.md b/backend/packages/harness/deerflow/tui/AGENTS.md index 730d9abb55..00f53fa5f3 100644 --- a/backend/packages/harness/deerflow/tui/AGENTS.md +++ b/backend/packages/harness/deerflow/tui/AGENTS.md @@ -3,7 +3,7 @@ A terminal-native UI over the embedded harness, exposed as the `deerflow` console script (`[project.scripts]` in `packages/harness/pyproject.toml`). It is a UI shell over `DeerFlowClient` and does **not** fork agent behavior. `textual` is an optional dependency (`deerflow-harness[tui]`; also in the backend dev group); the console script degrades to headless help when it is absent. Full guide: [docs/TUI.md](../../../../docs/TUI.md). **Module layout** (all layers except `app.py` are pure / Textual-free and unit-tested directly): -- `cli.py` — `plan_launch()` (pure launch-mode decision) + headless `--print` / `--json` + `main()` entry point. TTY → TUI, else headless help. `--tui-transparent` / `DEER_FLOW_TUI_TRANSPARENT` opt into terminal-default backgrounds without changing the solid-theme default. Uses an **absolute** `from deerflow.tui.app import run_tui` so the `app.py` module name doesn't trip `test_harness_boundary.py` (which records relative import module names verbatim). +- `cli.py` — `plan_launch()` (pure launch-mode decision) + headless `--print` / `--json` + `main()` entry point. Headless modes exit `1` when the final AI message carries `deerflow_error_fallback` (provider failures arrive as flagged messages, not exceptions). `_RunOutcome` and `chat()` share `_AIMessageAccumulator`: deltas accumulate per message id and the last content-bearing AI id supplies the text; metadata-only events preserve that id. Fallback writers must supply text or flag an id whose text was already emitted. Keep the client import lazy so help and launch planning do not load the agent graph. TTY → TUI, else headless help. `--tui-transparent` / `DEER_FLOW_TUI_TRANSPARENT` opt into terminal-default backgrounds without changing the solid-theme default. Uses an **absolute** `from deerflow.tui.app import run_tui` so the `app.py` module name doesn't trip `test_harness_boundary.py` (which records relative import module names verbatim). - `view_state.py` — `ViewState` + `reduce(state, action)`, the testable heart. Rows: user / assistant / tool / system. Title captured from `values` events. - `runtime.py` — `translate(StreamEvent) -> [Action]` (pure) + `stream_actions()` which brackets a run with `RunStarted`/`RunEnded` and turns model errors into an `AssistantError` row. - `message_format.py` / `command_registry.py` / `input_history.py` / `render.py` / `theme.py` — pure helpers (tool summaries, slash registry + `resolve()`, ↑/↓ history, Rich renderers). The command registry must exclude every shared `RESERVED_SLASH_SKILL_NAMES` entry that the agent runtime rejects from both its picker and resolver; the `context` skill is the exception for ordinary task text, while the exact composer-only `/context compact` alias remains unavailable as a skill activation. diff --git a/backend/packages/harness/deerflow/tui/cli.py b/backend/packages/harness/deerflow/tui/cli.py index fb71bc1bfe..208500c445 100644 --- a/backend/packages/harness/deerflow/tui/cli.py +++ b/backend/packages/harness/deerflow/tui/cli.py @@ -15,7 +15,10 @@ import os import sys from collections.abc import Sequence from dataclasses import dataclass -from typing import Literal +from typing import TYPE_CHECKING, Literal + +if TYPE_CHECKING: + from deerflow.client import StreamEvent _UNSET = object() @@ -207,6 +210,8 @@ deerflow — DeerFlow terminal workbench set the headless agent-loop super-step limit deerflow extensions --help install and manage trusted Python extensions echo "question" | deerflow --print + + --print and --json exit 1 when the run fails, including an LLM error. """ @@ -279,6 +284,53 @@ def _silence_closed_stdout() -> None: pass +class _RunOutcome: + """Track the final AI message of a headless run and whether it failed. + + Provider failures do not raise out of the stream: the LLM error middleware + turns them into an AI message flagged ``deerflow_error_fallback``. That text + is still the run's output, but a script calling ``deerflow --print`` needs a + non-zero exit status to tell it apart from a real answer. + """ + + def __init__(self) -> None: + from deerflow.client import _AIMessageAccumulator + + self._messages = _AIMessageAccumulator() + self._errors: dict[str, dict] = {} + + def observe(self, event: StreamEvent) -> None: + self._messages.observe(event) + if event.type != "messages-tuple" or event.data.get("type") != "ai": + return + msg_id = event.data.get("id") or "" + additional_kwargs = event.data.get("additional_kwargs") or {} + if additional_kwargs.get("deerflow_error_fallback"): + self._errors[msg_id] = additional_kwargs + + def answer(self) -> str: + return self._messages.answer() + + def error_text(self) -> str | None: + """One-line description when the final AI message is an error fallback.""" + # Error fallbacks carry text, or follow up on an id whose text was already sent. + error = self._errors.get(self._messages.last_id) + if error is None: + return None + details = ", ".join(f"{key}={error[key]}" for key in ("error_type", "error_reason") if error.get(key)) + return f"LLM request failed ({details})" if details else "LLM request failed" + + +def _report_json_error(error_text: str) -> int: + 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 + + def _run_print(plan: LaunchPlan) -> int: message = _resolve_message(plan) if not message: @@ -287,15 +339,20 @@ def _run_print(plan: LaunchPlan) -> int: try: session = _make_session() thread_id = session.resolve_thread(plan) - answer = session.client.chat(message, thread_id=thread_id, **_run_overrides(plan)) + outcome = _RunOutcome() + for event in session.client.stream(message, thread_id=thread_id, **_run_overrides(plan)): + outcome.observe(event) 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) + print(outcome.answer()) except BrokenPipeError: _silence_closed_stdout() return 1 + if (error_text := outcome.error_text()) is not None: + print(f"Error: {error_text}", file=sys.stderr) + return 1 return 0 @@ -307,19 +364,16 @@ def _run_json(plan: LaunchPlan) -> int: try: session = _make_session() thread_id = session.resolve_thread(plan) + outcome = _RunOutcome() for event in session.client.stream(message, thread_id=thread_id, **_run_overrides(plan)): + outcome.observe(event) 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 _report_json_error(_error_text(exc)) + if (error_text := outcome.error_text()) is not None: + return _report_json_error(error_text) return 0 diff --git a/backend/tests/test_client.py b/backend/tests/test_client.py index 302f14252b..b49167c9d6 100644 --- a/backend/tests/test_client.py +++ b/backend/tests/test_client.py @@ -25,7 +25,7 @@ from app.gateway.routers.threads import ThreadGoalResponse from app.gateway.routers.uploads import UploadResponse from deerflow.agents.middlewares.view_image_middleware import ViewImageMiddleware from deerflow.agents.thread_state import DeltaThreadState, ThreadState -from deerflow.client import DeerFlowClient +from deerflow.client import DeerFlowClient, StreamEvent from deerflow.config.agents_config import AgentConfig from deerflow.config.authorization_config import AuthorizationConfig, AuthorizationProviderConfig from deerflow.config.extensions_config import ExtensionsConfig, McpServerConfig @@ -752,6 +752,32 @@ class TestStream: assert any(event.data.get("content") == "Hello!" for event in ai_events) assert any(event.data.get("additional_kwargs", {}).get("token_usage_attribution", {}).get("kind") == "final_answer" for event in ai_events) + @pytest.mark.parametrize("streamed", [True, False]) + def test_stream_carries_llm_error_fallback_flag_to_headless_cli(self, client, streamed): + """``deerflow --print`` / ``--json`` read the fallback flag from stream events to exit non-zero.""" + from deerflow.tui.cli import _RunOutcome + + fallback = AIMessage( + content="The configured LLM provider rejected the request because authentication or access is invalid.", + id="ai-1", + additional_kwargs={"deerflow_error_fallback": True, "error_type": "AuthenticationError", "error_reason": "auth"}, + ) + chunks = [("values", {"messages": [HumanMessage(content="hi", id="h-1"), fallback]})] + if streamed: + chunks.insert(0, ("messages", (AIMessageChunk(content=fallback.content, id="ai-1"), {}))) + agent = _make_agent_mock(chunks) + + with ( + patch.object(client, "_ensure_agent"), + patch.object(client, "_agent", agent), + ): + outcome = _RunOutcome() + for event in client.stream("hi", thread_id="t-stream-fallback"): + outcome.observe(event) + + assert outcome.answer() == fallback.content + assert outcome.error_text() == "LLM request failed (error_type=AuthenticationError, error_reason=auth)" + @pytest.mark.parametrize("streamed", [True, False]) def test_stream_emits_text_a_later_node_appends_to_a_sent_ai_message(self, client, streamed): """A guard's ``after_model`` replaces the message under the same id after it was sent.""" @@ -1224,6 +1250,54 @@ class TestStream: class TestChat: + @pytest.mark.parametrize( + ("events", "expected"), + [ + pytest.param( + [ + StreamEvent(type="messages-tuple", data={"type": "ai", "id": "draft", "content": "draft"}), + StreamEvent(type="messages-tuple", data={"type": "ai", "id": "answer", "content": "fi"}), + StreamEvent(type="messages-tuple", data={"type": "ai", "id": "draft", "content": " revised"}), + StreamEvent(type="messages-tuple", data={"type": "ai", "id": "answer", "content": "nal"}), + StreamEvent(type="messages-tuple", data={"type": "tool", "id": "tool", "content": "ignored"}), + StreamEvent(type="values", data={"messages": [{"type": "ai", "content": "ignored"}]}), + ], + "final", + id="interleaved-message-ids", + ), + pytest.param( + [ + StreamEvent(type="messages-tuple", data={"type": "ai", "id": "answer", "content": "answer"}), + StreamEvent(type="messages-tuple", data={"type": "ai", "id": "empty", "content": "", "additional_kwargs": {"reasoning_content": "thinking"}}), + StreamEvent(type="end"), + ], + "answer", + id="metadata-only-message", + ), + pytest.param( + [ + StreamEvent(type="messages-tuple", data={"type": "ai", "content": "no"}), + StreamEvent(type="messages-tuple", data={"type": "ai", "id": None, "content": " id"}), + StreamEvent(type="messages-tuple", data={"type": "ai", "id": "", "content": " answer"}), + ], + "no id answer", + id="missing-message-id", + ), + pytest.param([StreamEvent(type="messages-tuple", data={"type": "ai", "id": "empty", "content": ""})], "", id="no-ai-text"), + ], + ) + def test_headless_and_chat_share_final_answer_selection(self, client, events, expected): + """The CLI and chat must agree on interleaved deltas and metadata-only events.""" + from deerflow.tui.cli import _RunOutcome + + with patch.object(client, "stream", return_value=iter(events)): + assert client.chat("q", thread_id="t-shared-answer") == expected + + outcome = _RunOutcome() + for event in events: + outcome.observe(event) + assert outcome.answer() == expected + def test_returns_last_message(self, client): """chat() returns the last AI message text.""" ai1 = AIMessage(content="thinking...", id="ai-1") diff --git a/backend/tests/test_tui_cli_main.py b/backend/tests/test_tui_cli_main.py index 70832d9541..ff7336f08f 100644 --- a/backend/tests/test_tui_cli_main.py +++ b/backend/tests/test_tui_cli_main.py @@ -11,13 +11,8 @@ from deerflow.tui import cli class _FakeClient: def __init__(self): - self.chat_kwargs = None self.stream_kwargs = None - def chat(self, message, *, thread_id=None, **kwargs): - self.chat_kwargs = kwargs - return f"answer:{message}" - def stream(self, message, *, thread_id=None, **kwargs): self.stream_kwargs = kwargs yield StreamEvent(type="messages-tuple", data={"type": "ai", "content": "hi", "id": "m1"}) @@ -35,25 +30,25 @@ class _FakeSession: return None -def test_main_print_outputs_chat_answer(monkeypatch, capsys): +def test_main_print_outputs_final_ai_answer(monkeypatch, capsys): monkeypatch.setattr(cli, "_make_session", _FakeSession) rc = cli.main(["--print", "hello"]) assert rc == 0 - assert "answer:hello" in capsys.readouterr().out + assert capsys.readouterr().out == "hi\n" def test_main_print_passes_explicit_recursion_limit(monkeypatch, capsys): monkeypatch.setattr(cli, "_make_session", _FakeSession) rc = cli.main(["--recursion-limit", "250", "--print", "hello"]) assert rc == 0 - assert _FakeSession.latest.client.chat_kwargs == {"recursion_limit": 250} + assert _FakeSession.latest.client.stream_kwargs == {"recursion_limit": 250} def test_main_print_omits_default_recursion_limit(monkeypatch, capsys): monkeypatch.setattr(cli, "_make_session", _FakeSession) rc = cli.main(["--print", "hello"]) assert rc == 0 - assert _FakeSession.latest.client.chat_kwargs == {} + assert _FakeSession.latest.client.stream_kwargs == {} def test_main_json_emits_ndjson_stream_events(monkeypatch, capsys): @@ -107,7 +102,7 @@ def _raising_session(exc, *, stream_events_before_failure=0): 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)) + return SimpleNamespace(resolve_thread=lambda plan: "thread-1", client=SimpleNamespace(stream=stream)) def test_main_print_client_failure_reports_error_without_traceback(monkeypatch, capsys): @@ -204,3 +199,95 @@ def test_main_json_broken_pipe_error_record_fails_silently(monkeypatch): monkeypatch.setattr(cli.sys, "stdout", _BrokenStdout()) rc = cli.main(["--json", "hello"]) assert rc == 1 + + +# --------------------------------------------------------------------------- # +# Headless exit status: an LLM error fallback is a failed run, not an answer. +# --------------------------------------------------------------------------- # + + +def _ai(msg_id, content, **additional_kwargs): + data = {"type": "ai", "content": content, "id": msg_id} + if additional_kwargs: + data["additional_kwargs"] = additional_kwargs + return StreamEvent(type="messages-tuple", data=data) + + +class _ScriptedClient: + def __init__(self, events): + self._events = events + + def stream(self, message, *, thread_id=None, **kwargs): + yield from self._events + + +class _ScriptedSession: + def __init__(self, events): + self.client = _ScriptedClient(events) + + def resolve_thread(self, plan): + return "thread-1" + + +def _fallback_events(): + return [ + _ai("m1", "Let me look."), + _ai( + "m2", + "The configured LLM provider rejected the request because authentication or access is invalid.", + deerflow_error_fallback=True, + error_type="AuthenticationError", + error_reason="auth", + ), + ] + + +@pytest.fixture +def run_headless(monkeypatch): + def _run(argv, events): + monkeypatch.setattr(cli, "_make_session", lambda: _ScriptedSession(events)) + return cli.main(argv) + + return _run + + +@pytest.mark.parametrize("mode", ["--print", "--json"]) +def test_headless_exits_nonzero_when_final_message_is_error_fallback(mode, run_headless, capsys): + assert run_headless([mode, "hello"], _fallback_events()) == 1 + assert capsys.readouterr().err == "Error: LLM request failed (error_type=AuthenticationError, error_reason=auth)\n" + + +def test_json_appends_error_record_after_error_fallback(run_headless, capsys): + assert run_headless(["--json", "hello"], _fallback_events()) == 1 + payloads = [json.loads(ln) for ln in capsys.readouterr().out.splitlines() if ln.strip()] + assert [p["type"] for p in payloads] == ["messages-tuple", "messages-tuple", "error"] + assert payloads[-1]["data"] == {"message": "LLM request failed (error_type=AuthenticationError, error_reason=auth)"} + + +def test_print_still_writes_fallback_text_to_stdout(run_headless, capsys): + run_headless(["--print", "hello"], _fallback_events()) + assert "authentication or access is invalid" in capsys.readouterr().out + + +@pytest.mark.parametrize("mode", ["--print", "--json"]) +def test_headless_detects_fallback_flag_sent_as_metadata_only_followup(mode, run_headless): + events = [_ai("m1", "partial answer"), _ai("m1", "", deerflow_error_fallback=True, error_reason="Model returned an empty terminal response")] + assert run_headless([mode, "hello"], events) == 1 + + +@pytest.mark.parametrize("mode", ["--print", "--json"]) +def test_headless_exits_zero_on_normal_answer(mode, run_headless, capsys): + assert run_headless([mode, "hello"], [_ai("m1", "4")]) == 0 + assert capsys.readouterr().err == "" + + +def test_print_outputs_only_last_ai_message(run_headless, capsys): + assert run_headless(["--print", "hello"], [_ai("m1", "draft"), _ai("m2", "fi"), _ai("m2", "nal")]) == 0 + assert capsys.readouterr().out == "final\n" + + +@pytest.mark.parametrize("mode", ["--print", "--json"]) +def test_headless_recovered_run_is_not_a_failure(mode, run_headless): + # A fallback earlier in the thread does not fail a run whose final answer succeeded. + events = [_ai("m1", "oops", deerflow_error_fallback=True, error_reason="auth"), _ai("m2", "real answer")] + assert run_headless([mode, "hello"], events) == 0