ga: diagnose blank LLM responses; tui keeps [Warn]/[Error] labels

frontends/tests/test_blank_response_diagnostics.py

@@ -0,0 +1,132 @@
+"""Tests for blank-response diagnostics + severity-preserving TUI rendering (#815).
+
+ga.py and tui_v3.py have heavy import side effects, so the helpers under test
+are extracted from source via ast/exec (same pattern as test_bridge_utils.py).
+Run: pytest frontends/tests/test_blank_response_diagnostics.py -v
+"""
+import ast
+import re
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+from urllib.parse import urlparse
+
+ROOT = Path(__file__).resolve().parent.parent.parent
+_GA_SRC = (ROOT / "ga.py").read_text(encoding="utf-8")
+_TUI_SRC = (ROOT / "frontends" / "tui_v3.py").read_text(encoding="utf-8")
+
+
+def _load_ga_helpers(names):
+    tree = ast.parse(_GA_SRC)
+    nodes = [
+        node for node in tree.body
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name in names
+    ]
+    namespace = {"re": re, "sys": sys, "urlparse": urlparse}
+    exec(compile(ast.Module(body=nodes, type_ignores=[]), "ga.py", "exec"), namespace)
+    return namespace
+
+
+def _load_tui_helpers(names):
+    tree = ast.parse(_TUI_SRC)
+    nodes = [
+        node for node in tree.body
+        if (isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name in names)
+        or (isinstance(node, ast.Assign)
+            and {t.id for t in node.targets if isinstance(t, ast.Name)} & names)
+    ]
+    namespace = {"re": re}
+    exec(compile(ast.Module(body=nodes, type_ignores=[]), "tui_v3.py", "exec"), namespace)
+    return namespace
+
+
+_GA = _load_ga_helpers({"describe_blank_response", "_url_host"})
+_TUI = _load_tui_helpers({"_ACTION_RE", "_SEVERITY_RE", "_severity_sub"})
+describe_blank_response = _GA["describe_blank_response"]
+url_host = _GA["_url_host"]
+
+
+class TestDescribeBlankResponse:
+    def test_whitespace_only_payload_with_text_block(self):
+        resp = SimpleNamespace(raw="[{'type': 'text', 'text': ' '}]", content=" ", thinking="")
+        assert describe_blank_response(resp, " ", "") == (
+            "kind=whitespace_only len(content)=1 len(thinking)=0 blocks=text"
+        )
+
+    def test_empty_payload_has_no_blocks_part(self):
+        resp = SimpleNamespace(raw="[]", content="", thinking="")
+        assert describe_blank_response(resp, "", "") == (
+            "kind=empty_payload len(content)=0 len(thinking)=0"
+        )
+
+    def test_blocks_are_deduped_and_sorted(self):
+        raw = "[{'type': 'thinking', 'thinking': 'x'}, {'type': 'text', 'text': ' '}]"
+        resp = SimpleNamespace(raw=raw, content="\n ", thinking="")
+        diag = describe_blank_response(resp, "\n ", "")
+        assert "kind=whitespace_only" in diag
+        assert "blocks=text+thinking" in diag
+
+    def test_no_raw_means_no_blocks_part(self):
+        resp = SimpleNamespace(raw="", content=" ", thinking="")
+        assert "blocks" not in describe_blank_response(resp, " ", "")
+
+    def test_both_blank_but_whitespace_reports_both_lengths(self):
+        resp = SimpleNamespace(raw="", content="   ", thinking="  ")
+        assert describe_blank_response(resp, "   ", "  ") == (
+            "kind=whitespace_only len(content)=3 len(thinking)=2"
+        )
+
+    def test_backend_appends_model_and_host(self):
+        backend = SimpleNamespace(model="gpt-4o", api_base="https://api.openai.com/v1")
+        resp = SimpleNamespace(raw="", content=" ", thinking="")
+        assert describe_blank_response(resp, " ", "", backend).endswith(
+            "model=gpt-4o host=api.openai.com"
+        )
+
+    def test_host_never_leaks_credentials(self):
+        backend = SimpleNamespace(model="m", api_base="https://user:s3cret@relay.example.com:8443/v1")
+        resp = SimpleNamespace(raw="", content=" ", thinking="")
+        diag = describe_blank_response(resp, " ", "", backend)
+        assert "host=relay.example.com" in diag
+        assert "s3cret" not in diag
+
+    def test_missing_model_falls_back_to_question_mark(self):
+        backend = SimpleNamespace(model=None, api_base="")
+        resp = SimpleNamespace(raw="", content=" ", thinking="")
+        assert describe_blank_response(resp, " ", "", backend).endswith("model=? host=unknown")
+
+
+class TestUrlHost:
+    def test_extracts_hostname(self):
+        assert url_host("https://api.openai.com/v1") == "api.openai.com"
+
+    def test_drops_port_and_userinfo(self):
+        assert url_host("http://user:pw@relay.example.com:8443/v1") == "relay.example.com"
+
+    def test_garbage_input_returns_unknown(self):
+        assert url_host("") == "unknown"
+        assert url_host("not a url") == "unknown"
+
+
+class TestSeverityRendering:
+    def test_action_info_debug_collapse_to_neutral_bullet(self):
+        out = _TUI["_ACTION_RE"].sub('· ', "[Action] run\n[Info] hi\n[Debug] x")
+        assert out == "· run\n· hi\n· x"
+
+    def test_severity_tags_keep_their_label(self):
+        out = _TUI["_SEVERITY_RE"].sub(_TUI["_severity_sub"], "[Warn] a\n[Error] b\n[Warning] c")
+        assert out == "· [Warn] a\n· [Error] b\n· [Warning] c"
+
+    def test_regexes_are_disjoint(self):
+        for tag in ("Warn", "Warning", "Error"):
+            assert _TUI["_ACTION_RE"].search(f"[{tag}] x") is None
+            assert _TUI["_SEVERITY_RE"].search(f"[{tag}] x") is not None
+        for tag in ("Action", "Status", "Info", "Debug"):
+            assert _TUI["_ACTION_RE"].search(f"[{tag}] x") is not None
+            assert _TUI["_SEVERITY_RE"].search(f"[{tag}] x") is None
+
+    def test_full_pipeline_matches_issue_815_example(self):
+        raw = "[Info] working\n[Warn] Empty LLM response (1/3); retrying\n[Action] run"
+        out = _TUI["_SEVERITY_RE"].sub(
+            _TUI["_severity_sub"], _TUI["_ACTION_RE"].sub('· ', raw))
+        assert out == "· working\n· [Warn] Empty LLM response (1/3); retrying\n· run"

frontends/tui_v3.py

@@ -1695,7 +1695,12 @@ def repl(m: re.Match) -> str:
 
 
 _ACTION_RE = re.compile(
-    r'^[ \t]*\[(?:Action|Status|Info|Debug|Warn|Warning|Error)\][ \t]*', re.M)
+    r'^[ \t]*\[(?:Action|Status|Info|Debug)\][ \t]*', re.M)
+# severity tags stay visible as `· [Warn]` so failures don't read as routine output (#815)
+_SEVERITY_RE = re.compile(r'^[ \t]*\[(Warn|Warning|Error)\][ \t]*', re.M)
+
+def _severity_sub(m):
+    return '· [' + m.group(1) + '] '
 
 
 @dataclass
@@ -5331,7 +5336,7 @@ def xrepl(m: re.Match) -> str:           # prompted-XML form: body IS the result
 
         out = _XML_TOOL_RE.sub(xrepl, out)
         self._last_tool_n = idx
-        out = _ACTION_RE.sub('· ', _TURN_MK_RE.sub('', out))
+        out = _SEVERITY_RE.sub(_severity_sub, _ACTION_RE.sub('· ', _TURN_MK_RE.sub('', out)))
         return strip_meta_tags(out)        # empty when fully meta — render nothing,
                                             # else early '...' placeholders pollute _sent
 

ga.py

@@ -1,5 +1,6 @@
 import sys, os, re, json, time, threading, importlib, webbrowser
 from datetime import datetime
+from urllib.parse import urlparse
 from pathlib import Path
 import tempfile, traceback, subprocess, itertools, collections, difflib, shutil
 if sys.stdout is None: sys.stdout = open(os.devnull, "w")
@@ -155,6 +156,24 @@ def format_error(e):
         return f"{exc_type.__name__}: {str(e)} @ {fname}:{f.lineno}, {f.name} -> `{f.line}`"
     return f"{exc_type.__name__}: {str(e)}"
 
+def _url_host(url):
+    # urlparse rejects malformed IPv6 brackets; a diagnostic must never crash the retry path
+    try: return urlparse(url).hostname or 'unknown'
+    except ValueError: return 'unknown'
+
+def describe_blank_response(response, content, thinking, backend=None):
+    '''Redacted one-liner for do_no_tool's blank-response guard: kind, sizes,
+    block types, model and gateway host — never credentials.'''
+    cs, ts = content or '', thinking or ''
+    kind = 'empty_payload' if not cs and not ts else 'whitespace_only'
+    parts = [f'kind={kind}', f'len(content)={len(cs)}', f'len(thinking)={len(ts)}']
+    blocks = re.findall(r"'type':\s*'([a-z_]+)'", getattr(response, 'raw', '') or '')
+    if blocks: parts.append('blocks=' + '+'.join(sorted(set(blocks))))
+    if backend is not None:
+        parts.append(f"model={getattr(backend, 'model', '') or '?'}")
+        parts.append(f"host={_url_host(getattr(backend, 'api_base', '') or '')}")
+    return ' '.join(parts)
+
 def log_memory_access(path):
     if 'memory' not in path: return
     stats_file = os.path.join(script_dir, 'memory/file_access_stats.json')
@@ -481,7 +500,9 @@ def do_no_tool(self, args, response):
         content = getattr(response, 'content', '') or ""
         thinking = getattr(response, 'thinking', '') or ""
         if not response or (not content.strip() and not thinking.strip()):
-            yield "[Warn] LLM returned an empty response. Retrying...\n"
+            backend = getattr(getattr(self.parent, 'llmclient', None), 'backend', None)
+            yield (f"[Warn] Empty LLM response ({getattr(self, '_empty_ct', 0) + 1}/3); retrying "
+                   f"[{describe_blank_response(response, content, thinking, backend)}]\n")
             return self._retry_or_exit("[ERROR] Blank response, regenerate and tooluse")
         if '[!!! 流异常中断' in content[-100:] or '!!!Error:' in content[50:][-100:] or (content.endswith('</summary>') and len(content) < 100):
             return self._retry_or_exit("[ERROR] Incomplete response. Regenerate and tooluse.")