diff --git a/ga.py b/ga.py
index e1ee909f2..17d58b106 100644
--- a/ga.py
+++ b/ga.py
@@ -483,7 +483,7 @@ def do_no_tool(self, args, response):
if not response or (not content.strip() and not thinking.strip()):
yield "[Warn] LLM returned an empty response. Retrying...\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('') and len(content) < 100):
+ if getattr(response, 'error', None) or (content.endswith('') and len(content) < 100):
return self._retry_or_exit("[ERROR] Incomplete response. Regenerate and tooluse.")
if 'max_tokens !!!]' in content[-100:]:
return self._retry_or_exit("[ERROR] max_tokens limit reached. Use multi small steps to do it.")
diff --git a/llmcore.py b/llmcore.py
index 423503248..3cbc15280 100644
--- a/llmcore.py
+++ b/llmcore.py
@@ -131,11 +131,14 @@ def auto_make_url(base, path):
if b.endswith(p): return b
return f"{b}/{p}" if re.search(r'/v\d+(/|$)', b) else f"{b}/v1/{p}"
+class LLMError(str):
+ """Internal diagnostic chunk, never inferred from model-generated text."""
+
def _parse_claude_json(data):
if data.get("stop_reason") == "refusal":
- err = "[Error: Claude refusal]"
+ err = LLMError("[Error: Claude refusal]")
yield err
- return [{"type": "text", "text": err}]
+ return [{"type": "text", "text": err, "error": str(err)}]
content_blocks = data.get("content", [])
_record_usage(data.get("usage", {}), "messages")
for b in content_blocks:
@@ -205,11 +208,11 @@ def _parse_claude_sse(resp_lines):
err = evt.get("error", {})
emsg = err.get("message", str(err)) if isinstance(err, dict) else str(err)
_raise_if_retryable_overload(emsg) # 走 _stream_with_retry，避免落到 ga 应用层
- warn = f"\n\n!!!Error: SSE {emsg}"; break
+ warn = LLMError(f"\n\n!!!Error: SSE {emsg}"); break
if not warn:
- if not got_message_stop and not stop_reason: warn = "\n\n[!!! 流异常中断，未收到完整响应 !!!]"
- elif stop_reason == "max_tokens": warn = "\n\n[!!! Response truncated: max_tokens !!!]"
- elif stop_reason == "refusal": warn = "\n\n[Error: Claude refusal]"
+ if not got_message_stop and not stop_reason: warn = LLMError("\n\n[!!! 流异常中断，未收到完整响应 !!!]")
+ elif stop_reason == "max_tokens": warn = LLMError("\n\n[!!! Response truncated: max_tokens !!!]")
+ elif stop_reason == "refusal": warn = LLMError("\n\n[Error: Claude refusal]")
if current_block:
if current_block["type"] == "tool_use":
try: current_block["input"] = json.loads(tool_json_buf) if tool_json_buf else {}
@@ -281,7 +284,7 @@ def _parse_openai_sse(resp_lines, api_mode="chat_completions"):
err = evt.get("error", {})
emsg = err.get("message", str(err)) if isinstance(err, dict) else str(err)
_raise_if_retryable_overload(emsg)
- if emsg: content_text += f"!!!Error: {emsg}"; yield f"!!!Error: {emsg}"
+ if emsg: content_text += f"!!!Error: {emsg}"; yield LLMError(f"!!!Error: {emsg}")
break
elif etype == "response.completed":
usage = evt.get("response", {}).get("usage", {})
@@ -303,7 +306,7 @@ def _parse_openai_sse(resp_lines, api_mode="chat_completions"):
err = ((evt.get("response") or {}).get("error") or {})
emsg = err.get("message", str(err)) if isinstance(err, dict) else str(err)
_raise_if_retryable_overload(emsg)
- if emsg: content_text += f"!!!Error: {emsg}"; yield f"!!!Error: {emsg}"
+ if emsg: content_text += f"!!!Error: {emsg}"; yield LLMError(f"!!!Error: {emsg}")
break
blocks = []
if reasoning_text: blocks.append({"type": "thinking", "thinking": reasoning_text})
@@ -404,7 +407,7 @@ def _parse_openai_json(data, api_mode="chat_completions"):
err = data.get("error") or {}
emsg = err.get("message", str(err)) if isinstance(err, dict) else str(err)
_raise_if_retryable_overload(emsg)
- if emsg: blocks.append({"type": "text", "text": f"!!!Error: {emsg}"}); yield f"!!!Error: {emsg}"
+ if emsg: blocks.append({"type": "text", "text": f"!!!Error: {emsg}"}); yield LLMError(f"!!!Error: {emsg}")
elif status == "incomplete" and not any(b.get("type") == "text" for b in blocks):
reason = ((data.get("incomplete_details") or {}).get("reason", "")) or "unknown"
marker = f"[!!! output truncated: {reason}]"
@@ -455,7 +458,7 @@ def _sleep(d): # interruptible sleep; True if aborted
return _stopped()
for attempt in range(sess.max_retries + 1):
if _stopped(): return []
- streamed = False
+ streamed = False; error = None
STATS.update(t_start=time.time(), t_ttft=None)
if not sess.stream: STATS['t_ttft'] = STATS['t_start']
try:
@@ -471,8 +474,8 @@ def _sleep(d): # interruptible sleep; True if aborted
continue
try: body = r.text.strip()[:500]
except: body = ""
- err = f"!!!Error: HTTP {r.status_code}" + (f" (retry-after > {cap:.0f}s)" if d is None and r.status_code in _RETRYABLE and attempt < sess.max_retries else "") + (f": {body}" if body else "")
- yield err; return [{"type": "text", "text": err}]
+ err = LLMError(f"!!!Error: HTTP {r.status_code}" + (f" (retry-after > {cap:.0f}s)" if d is None and r.status_code in _RETRYABLE and attempt < sess.max_retries else "") + (f": {body}" if body else ""))
+ yield err; return [{"type": "text", "text": err, "error": str(err)}]
gen = parse_fn(r)
try:
while True:
@@ -480,24 +483,27 @@ def _sleep(d): # interruptible sleep; True if aborted
STATS['t_end'] = time.time(); return []
chunk = next(gen)
if chunk and STATS.get('t_ttft') is None: STATS['t_ttft'] = time.time()
+ if isinstance(chunk, LLMError): error = str(chunk)
streamed = True; yield chunk
except StopIteration as e:
if not e.value and not streamed: raise requests.ConnectionError("empty response")
STATS['t_end'] = time.time()
STATS['tps'] = STATS.get('out', 0) / max(1e-9, STATS['t_end'] - max(STATS['t_ttft'] or 0, STATS['t_start']))
- return e.value or []
+ blocks = e.value or []
+ if error: blocks.append({"type": "text", "text": "", "error": error})
+ return blocks
except (requests.Timeout, requests.ConnectionError, requests.exceptions.ChunkedEncodingError) as e:
- err = f"!!!Error: {type(e).__name__}: {e}" if str(e) else f"!!!Error: {type(e).__name__}"
+ err = LLMError(f"!!!Error: {type(e).__name__}: {e}" if str(e) else f"!!!Error: {type(e).__name__}")
if getattr(sess, 'should_stop', None) and sess.should_stop(): return []
if attempt < sess.max_retries:
d = _delay(None, attempt)
print(f"[LLM Retry] {type(e).__name__}, retry in {d:.1f}s ({attempt+1}/{sess.max_retries+1})")
if _sleep(d): return []
continue
- yield err; return [{"type": "text", "text": err}]
+ yield err; return [{"type": "text", "text": err, "error": str(err)}]
except Exception as e:
- err = f"\n\n[!!! 流异常中断 {type(e).__name__}: {e} !!!]" if streamed else f"!!!Error: {type(e).__name__}: {e}"
- yield err; return [{"type": "text", "text": err}]
+ err = LLMError(f"\n\n[!!! 流异常中断 {type(e).__name__}: {e} !!!]" if streamed else f"!!!Error: {type(e).__name__}: {e}")
+ yield err; return [{"type": "text", "text": err, "error": str(err)}]
def _openai_stream(sess, messages):
model, api_mode = sess.model, sess.api_mode
@@ -681,17 +687,20 @@ def _ask_gen():
self.history.append({"role": "user", "content": [{"type": "text", "text": prompt}]})
trim_messages_history(self.history, self)
messages = self.make_messages(self.history)
- content_blocks = None; content = ''
+ content_blocks = None; content = ''; error = None
gen = self.raw_ask(messages)
try:
- while True: chunk = next(gen); content += chunk; yield chunk
+ while True:
+ chunk = next(gen)
+ if isinstance(chunk, LLMError): error = str(chunk)
+ content += chunk; yield chunk
except StopIteration as e: content_blocks = e.value or []
if len(content_blocks) > 1: print(f"[DEBUG BaseSession.ask] content_blocks: {content_blocks}")
for block in (content_blocks or []):
if block.get('type', '') == 'tool_use':
tu = {'name': block.get('name', ''), 'arguments': block.get('input', {})}
yield f'{json.dumps(tu, ensure_ascii=False)}'
- if content.strip() and not content.startswith("!!!Error:"): self.history.append({"role": "assistant", "content": [{"type": "text", "text": content}]})
+ if content.strip() and not error: self.history.append({"role": "assistant", "content": [{"type": "text", "text": content}]})
return _ask_gen()
def _keep_claude_block(b): return not isinstance(b, dict) or b.get("type") != "thinking" or b.get("signature")
@@ -839,12 +848,16 @@ def ask(self, msg):
while True: yield next(gen)
except StopIteration as e: content_blocks = e.value or []
if content_blocks and (_injected := _ensure_text_block(content_blocks)): yield _injected
- if content_blocks and not (len(content_blocks) == 1 and content_blocks[0].get("text", "").startswith("!!!Error:")):
+ error = next((b["error"] for b in content_blocks if b.get("error")), None)
+ if content_blocks and not error:
history_blocks = content_blocks
if self.omit_thinking: history_blocks = [b for b in content_blocks if b.get("type") != "thinking"]
self.history.append({"role": "assistant", "content": history_blocks})
text_parts = [b["text"] for b in content_blocks if b.get("type") == "text"]
content = "\n".join(text_parts).strip()
+ if error:
+ raw = "[" + ",\n".join(repr(b) for b in content_blocks) + "]"
+ return MockResponse("", content, [], raw, error=error)
tool_calls = [MockToolCall(b["name"], b.get("input", {}), id=b.get("id", "")) for b in content_blocks if b.get("type") == "tool_use"]
if not tool_calls: tool_calls, content = _parse_text_tool_calls(content)
thinking_parts = [b["thinking"] for b in content_blocks if b.get("type") == "thinking"]
@@ -856,7 +869,7 @@ def ask(self, msg):
thinking = think_match.group(1).strip()
content = re.sub(think_pattern, "", content, flags=re.DOTALL)
raw = "[" + ",\n".join(repr(b) for b in content_blocks) + "]"
- return MockResponse(thinking, content, tool_calls, raw)
+ return MockResponse(thinking, content, [] if error else tool_calls, raw, error=error)
class NativeOAISession(NativeClaudeSession):
native_ua = "codex_exec/0.139.0 (Windows 10.0.26200; x86_64) unknown (codex_exec; 0.139.0)"
@@ -884,7 +897,8 @@ def __init__(self, name, args, id=''):
self.function = MockFunction(name, arg_str); self.id = id
class MockResponse:
- def __init__(self, thinking, content, tool_calls, raw, stop_reason='end_turn'):
+ def __init__(self, thinking, content, tool_calls, raw, stop_reason='end_turn', error=None):
+ self.error = error
self.thinking = thinking; self.content = content
self.tool_calls = tool_calls; self.raw = raw
self.stop_reason = 'tool_use' if tool_calls else stop_reason
@@ -914,10 +928,12 @@ def chat(self, messages, tools=None):
print("Full prompt length:", len(full_prompt), 'chars')
gen = self.backend.ask(full_prompt)
_write_llm_log('Prompt', full_prompt, self.log_path)
- raw_text = ''
+ raw_text = ''; error = None
for chunk in gen:
+ if isinstance(chunk, LLMError): error = str(chunk)
raw_text += chunk; yield chunk
_write_llm_log('Response', raw_text, self.log_path, model=self.backend.model)
+ if error: return MockResponse("", raw_text, [], raw_text, error=error)
return self._parse_mixed_response(raw_text)
def _prepare_tool_instruction(self, tools):
@@ -1119,7 +1135,7 @@ def _prepare(self, idx, messages):
return messages if self._native else session.make_messages(messages)
def raw_ask(self, messages):
base, n = self._pick(), len(self._sessions)
- test_error = lambda x: isinstance(x, str) and x.lstrip().startswith(('!!!Error:', '[Error:'))
+ test_error = lambda x: isinstance(x, LLMError)
for attempt in range(self._retries + 1):
idx = (base + attempt) % n
session = self._sessions[idx]
@@ -1132,10 +1148,10 @@ def raw_ask(self, messages):
if not yielded and test_error(chunk): continue
yield chunk; yielded = True
except StopIteration as e: return_val = e.value or []
- is_err = test_error(last_chunk)
+ is_err = test_error(last_chunk) and not yielded
if not is_err:
if attempt > 0: self._cur_idx = idx; self._switched_at = time.time()
- elif isinstance(last_chunk, str) and '[!!! 流异常中断' in last_chunk and n > 1:
+ elif test_error(last_chunk) and n > 1:
self._cur_idx = (idx + 1) % n; self._switched_at = time.time()
print(f'[MixinSession] Partial failure, next call → s{self._cur_idx} ({self.current.name})')
return return_val
diff --git a/tests/test_error_provenance.py b/tests/test_error_provenance.py
new file mode 100644
index 000000000..e1e2efdeb
--- /dev/null
+++ b/tests/test_error_provenance.py
@@ -0,0 +1,126 @@
+import json
+import threading
+import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import llmcore as core
+from agent_loop import exhaust
+
+
+TEXT = "!!!Error: ConnectionError: empty response"
+
+
+def stream(chunks, blocks):
+ yield from chunks
+ return blocks
+
+
+class ErrorProvenanceTests(unittest.TestCase):
+ def transport(self, parser, response=None):
+ response = response or SimpleNamespace(status_code=200)
+ response.__enter__ = lambda: response
+ sess = SimpleNamespace(name="test", max_retries=0, stream=True,
+ connect_timeout=1, read_timeout=1,
+ proxies=None, verify=True)
+ with patch.object(core.requests, "post") as post:
+ post.return_value.__enter__.return_value = response
+ chunks = []
+ gen = core._stream_with_retry(sess, "https://unused", {}, {}, parser)
+ try:
+ while True:
+ chunks.append(next(gen))
+ except StopIteration as exc:
+ return chunks, exc.value
+
+ def native(self, chunks, blocks, cls=core.NativeClaudeSession):
+ sess = cls.__new__(cls)
+ sess.lock = threading.Lock()
+ sess.history = []
+ sess.omit_thinking = False
+ sess.raw_ask = lambda messages: stream(chunks, blocks)
+ with patch.object(core, "trim_messages_history"):
+ result = exhaust(sess.ask({"role": "user", "content": []}))
+ return result, sess.history
+
+ def test_empty_transport_is_typed(self):
+ chunks, blocks = self.transport(lambda r: stream([], []))
+ self.assertIsInstance(chunks[-1], core.LLMError)
+ self.assertEqual(str(chunks[-1]), TEXT)
+ self.assertEqual(blocks[-1]["error"], TEXT)
+
+ def test_same_text_native_response_and_history(self):
+ for cls in (core.NativeClaudeSession, core.NativeOAISession):
+ for failed in (False, True):
+ with self.subTest(cls=cls, failed=failed):
+ chunk = core.LLMError(TEXT) if failed else TEXT
+ chunks, blocks = self.transport(lambda r: stream([chunk], [{"type": "text", "text": TEXT}]))
+ response, history = self.native(chunks, blocks, cls)
+ self.assertEqual(response.content, TEXT)
+ self.assertEqual(bool(response.error), failed)
+ self.assertEqual(len(history), 1 if failed else 2)
+
+ def test_same_text_protocol_client_and_history(self):
+ for failed in (False, True):
+ with self.subTest(failed=failed):
+ sess = core.BaseSession.__new__(core.BaseSession)
+ sess.lock = threading.Lock()
+ sess.history = []
+ sess.name = sess.model = "test"
+ sess.make_messages = lambda messages: messages
+ chunk = core.LLMError(TEXT) if failed else TEXT
+ sess.raw_ask = lambda messages: stream([chunk], [{"type": "text", "text": TEXT}])
+ client = core.ToolClient(sess)
+ client._build_protocol_prompt = lambda messages, tools: "test"
+ with patch.object(core, "trim_messages_history"):
+ response = exhaust(client.chat([]))
+ self.assertEqual(response.content, TEXT)
+ self.assertEqual(bool(response.error), failed)
+ self.assertEqual(len(sess.history), 1 if failed else 2)
+
+ def test_native_error_does_not_execute_partial_tool(self):
+ chunks, blocks = self.transport(lambda r: stream([core.LLMError(TEXT)], [
+ {"type": "tool_use", "id": "t", "name": "danger", "input": {}},
+ {"type": "text", "text": TEXT}]))
+ response, history = self.native(chunks, blocks)
+ self.assertTrue(response.error)
+ self.assertEqual(response.tool_calls, [])
+ self.assertEqual(len(history), 1)
+
+ def test_mixin_only_falls_back_for_typed_errors(self):
+ for failed in (False, True):
+ with self.subTest(failed=failed):
+ first = core.LLMError(TEXT) if failed else TEXT
+ calls = []
+ def raw(index, chunk):
+ calls.append(index)
+ yield chunk
+ return [{"type": "text", "text": str(chunk)}]
+ mix = core.MixinSession.__new__(core.MixinSession)
+ mix.__dict__.update(_sessions=[
+ SimpleNamespace(name="a", raw_ask=lambda m: raw(0, first)),
+ SimpleNamespace(name="b", raw_ask=lambda m: raw(1, "OK"))],
+ _retries=1, _base_delay=0, _native=True, _cur_idx=0, _switched_at=0)
+ mix._pick = lambda: 0
+ mix._prepare = lambda idx, messages: messages
+ with patch.object(core.time, "sleep"):
+ chunks = list(mix.raw_ask([]))
+ self.assertEqual(calls, [0, 1] if failed else [0])
+ self.assertEqual(chunks, ["OK"] if failed else [TEXT])
+
+ def test_provider_error_events_are_typed(self):
+ cases = [
+ lambda: core._parse_claude_sse([b'data: {"type":"error","error":{"message":"bad request"}}']),
+ lambda: core._parse_claude_sse([]),
+ lambda: core._parse_openai_sse([b'data: {"type":"error","error":{"message":"bad request"}}'], "responses"),
+ lambda: core._parse_openai_json({"status": "failed", "error": {"message": "bad request"}}, "responses"),
+ ]
+ for parser in cases:
+ with self.subTest(parser=parser):
+ chunks, blocks = self.transport(lambda r: parser())
+ self.assertTrue(any(isinstance(c, core.LLMError) for c in chunks))
+ self.assertTrue(any(b.get("error") for b in blocks))
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_no_tool_retry.py b/tests/test_no_tool_retry.py
new file mode 100644
index 000000000..0d3c74986
--- /dev/null
+++ b/tests/test_no_tool_retry.py
@@ -0,0 +1,46 @@
+import unittest
+from types import SimpleNamespace
+
+from agent_loop import exhaust
+from ga import GenericAgentHandler
+
+
+class NoToolRetryTests(unittest.TestCase):
+ def setUp(self):
+ self.handler = GenericAgentHandler.__new__(GenericAgentHandler)
+ self.handler._in_plan_mode = lambda: False
+
+ def respond(self, content, thinking="", error=None):
+ return exhaust(self.handler.do_no_tool({}, SimpleNamespace(content=content, thinking=thinking, error=error)))
+
+ def test_short_error_retries(self):
+ outcome = self.respond("!!!Error: ConnectionError: empty response", error="transport")
+ self.assertIsNotNone(outcome.next_prompt)
+ self.assertFalse(outcome.should_exit)
+
+ def test_error_prefix_retries_with_long_details(self):
+ self.assertIsNotNone(self.respond("!!!Error: ConnectionError: " + "x" * 150, error="transport").next_prompt)
+
+ def test_error_retries_are_bounded(self):
+ for attempt in range(3):
+ outcome = self.respond("!!!Error: ConnectionError: empty response", error="transport")
+ self.assertEqual(outcome.should_exit, attempt == 2)
+ self.assertEqual(outcome.next_prompt is None, attempt == 2)
+
+ def test_existing_retry_cases(self):
+ for content in ("",
+ "Working", "max_tokens !!!]"):
+ with self.subTest(content=content):
+ self.setUp()
+ self.assertIsNotNone(self.respond(content).next_prompt)
+
+ def test_normal_and_quoted_responses_finish(self):
+ for content, thinking in (("!!!Error: ConnectionError: empty response", ""), ("x" * 60 + "!!!Error: disconnected", ""), ("Done.", ""), ("", "Reasoning only"),
+ ("The marker is '!!!Error:'.", "")):
+ with self.subTest(content=content):
+ self.assertIsNone(self.respond(content, thinking).next_prompt)
+ self.assertFalse(hasattr(self.handler, "_empty_ct"))
+
+
+if __name__ == "__main__":
+ unittest.main()