diff --git a/agent.py b/agent.py index 62beb8bc14..e6176b8872 100644 --- a/agent.py +++ b/agent.py @@ -849,6 +849,10 @@ async def stream_callback(chunk: str, total: str): rate_limiter_callback=( self.rate_limiter_callback if not call_data["background"] else None ), + # Utility prompts (e.g. memory post-processing) treat an empty + # reply as a benign "nothing to do", so the main-turn + # empty-completion retry must not fire here. + a0_allow_empty_completion=True, ) await extension.call_extensions_async( @@ -1119,6 +1123,7 @@ async def process_llm_result_tools(self, llm_result: LLMResult): if ( extract_tools.extract_tool_request(llm_result.reasoning) is not None or extract_tools.is_misformatted_tool_request(llm_result.reasoning) + or extract_tools.is_truncated_tool_request(llm_result.reasoning) ): message = llm_result.reasoning if ( @@ -1127,13 +1132,14 @@ async def process_llm_result_tools(self, llm_result: LLMResult): and bool(message.strip()) and extract_tools.extract_tool_request(message) is None and not extract_tools.is_misformatted_tool_request(message) + and not extract_tools.is_truncated_tool_request(message) ): return await self._execute_tool_request( tool_name="response", tool_args={"text": message}, message=message, ) - return await self.process_tools(message) + return await self.process_tools(message, finish_reason=llm_result.finish_reason) async def _execute_tool_request( self, @@ -1410,10 +1416,17 @@ def _clear_responses_pending_state(self) -> None: self.set_data(Agent.DATA_NAME_RESPONSES_STATE, state) @extension.extensible - async def process_tools(self, msg: str): + async def process_tools(self, msg: str, finish_reason: str = ""): # search for tool usage requests in agent message tool_request = extract_tools.extract_tool_request(msg) + recovered_embedded = False + if tool_request is None: + # narrowly scoped repair: accept exactly one valid tool request + # embedded in planning prose (DeepSeek V4 Flash thinking output) + tool_request = extract_tools.recover_embedded_tool_request(msg) + recovered_embedded = tool_request is not None + raw_tool_name = "" tool_args = {} @@ -1427,6 +1440,21 @@ async def process_tools(self, msg: str): ) except ValueError: tool_request = None # treat structural validation errors as misformat + recovered_embedded = False + + if recovered_embedded: + # Corrective note: recovery succeeded, but teach the model to emit + # bare JSON next time. Deliberately NOT one of the prompts tracked + # by the unusable-response-loop guard - a recovered request is a + # usable response. + warning_msg = self.read_prompt("fw.msg_recovered_request.md") + wmsg = self.hist_add_warning(warning_msg) + PrintStyle(font_color="orange", padding=True).print(warning_msg) + self.context.log.log( + type="warning", + content=f"{self.agent_name}: Tool request embedded in prose was recovered.", + id=wmsg.id, + ) if tool_request is not None: tool_name = raw_tool_name # Initialize tool_name with raw_tool_name @@ -1508,12 +1536,22 @@ async def process_tools(self, msg: str): type="warning", content=f"{self.agent_name}: {error_detail}", id=wmsg.id ) else: - warning_msg_misformat = self.read_prompt("fw.msg_misformat.md") - wmsg = self.hist_add_warning(warning_msg_misformat) - PrintStyle(font_color="red", padding=True).print(warning_msg_misformat) + category = extract_tools.classify_tool_request_failure(msg) + reason = extract_tools.explain_tool_request_failure(msg, finish_reason) + if category == "truncated": + warning_msg = self.read_prompt("fw.msg_truncated_request.md") + log_reason = ( + f"truncated or unterminated JSON tool request; " + f"reason: {reason}" + ) + else: + warning_msg = self.read_prompt("fw.msg_misformat.md") + log_reason = f"no valid tool request found; reason: {reason}" + wmsg = self.hist_add_warning(warning_msg) + PrintStyle(font_color="red", padding=True).print(warning_msg) self.context.log.log( type="warning", - content=f"{self.agent_name}: Message misformat, no valid tool request found.", + content=f"{self.agent_name}: Message misformat, {log_reason}.", id=wmsg.id, ) diff --git a/extensions/python/_functions/agent/Agent/hist_add_warning/end/_90_stop_unusable_response_loop.py b/extensions/python/_functions/agent/Agent/hist_add_warning/end/_90_stop_unusable_response_loop.py index 2c44a910c3..78b6e7e773 100644 --- a/extensions/python/_functions/agent/Agent/hist_add_warning/end/_90_stop_unusable_response_loop.py +++ b/extensions/python/_functions/agent/Agent/hist_add_warning/end/_90_stop_unusable_response_loop.py @@ -21,6 +21,7 @@ def execute(self, data: dict | None = None, **kwargs): return if message not in { self.agent.read_prompt("fw.msg_misformat.md"), + self.agent.read_prompt("fw.msg_truncated_request.md"), self.agent.read_prompt("fw.msg_repeat.md"), }: return diff --git a/helpers/__init__.py b/helpers/__init__.py new file mode 100644 index 0000000000..98a39f3e2d --- /dev/null +++ b/helpers/__init__.py @@ -0,0 +1,8 @@ +# This file must exist. Without it, `helpers` is imported as a namespace +# package (no __file__), which makes test/module cleanup helpers that purge +# `sys.modules` treat it as a stub and delete every loaded `helpers.*` +# module. The subsequent re-imports then split extension-registry state +# between stale and fresh module copies (observed as `Agent` instances +# losing extension-initialized attributes such as `loop_data` depending on +# test import order). Keeping `helpers` a regular package prevents that +# entire pollution class. diff --git a/helpers/extract_tools.py b/helpers/extract_tools.py index e1996da34d..77eb75b4c9 100644 --- a/helpers/extract_tools.py +++ b/helpers/extract_tools.py @@ -1,4 +1,6 @@ +import json + from .dirty_json import DirtyJson import regex, re from helpers.modules import load_classes_from_file, load_classes_from_folder # keep here for backwards compatibility @@ -33,6 +35,176 @@ def extract_tool_request(content: str) -> dict[str, Any] | None: return request if request is not None and _is_tool_request(request) else None +def recover_embedded_tool_request(content: str) -> dict[str, Any] | None: + """Recover exactly one valid tool request embedded in surrounding prose. + + Strict extraction (extract_tool_request) stays unchanged; this is a + narrowly scoped repair for models (e.g. DeepSeek V4 Flash with thinking + enabled) that occasionally wrap a single valid JSON tool envelope in + planning prose. Returns None when zero or multiple *distinct* tool + requests are found, so arbitrary prose is never treated as a tool call. + + Envelopes appearing inside quoted spans, inline code, or fenced code + blocks are examples being discussed, not requests being issued, and are + masked before scanning. Candidates must parse as strict JSON, so lenient + dirty-JSON forms (single-quoted keys, etc.) are not executable either. + """ + if not content or not isinstance(content, str): + return None + + content = content.strip() + if extract_tool_request(content) is not None: + return None # strict path handles clean responses; recovery is failure-only + + masked = _mask_non_executable_regions(content) + distinct: dict[str, dict[str, Any]] = {} + root_for_request = "" + for root in extract_json_root_strings(masked): + data = _parse_json_root_object_strict(root) + if data is None or not _is_tool_request(data): + continue + distinct.setdefault(json.dumps(data, sort_keys=True, default=str), data) + if len(distinct) > 1: + return None + root_for_request = root + request = next(iter(distinct.values()), None) + if request is None: + return None + if str(request.get("tool_name") or "") == "response" and _has_prose_around( + masked, root_for_request + ): + # A "response" envelope buried in deliberating prose is the model + # thinking out loud about how it *could* reply, not a completed task. + # Refuse recovery so the message takes the standard misformat path + # instead of ending the task. Operational tools keep being recovered: + # executing a valid operational envelope is the protocol's intent. + return None + return request + + +_RESPONSE_PROSE_HEDGE_RE = re.compile( + r"\b(could|might|maybe|option|alternatively|but)\b", re.IGNORECASE +) + + +def _has_prose_around(masked_content: str, root: str) -> bool: + """Heuristic: True when substantial prose surrounds the extracted root. + + Deliberately simple and deterministic: prose is substantial when the + non-whitespace text outside the JSON root exceeds 40 characters, or when + it contains hedging language ("could", "but", ...) that marks the + envelope as a possibility under discussion rather than the final answer. + Quoted/fenced spans are already blanked by _mask_non_executable_regions, + so examples under discussion do not count as prose. + """ + prose = masked_content.replace(root, " ", 1) + if len("".join(prose.split())) > 40: + return True + return bool(_RESPONSE_PROSE_HEDGE_RE.search(prose)) + + +def _mask_non_executable_regions(content: str) -> str: + """Blank out fenced code blocks, inline code and top-level quoted spans. + + Quoted/fenced regions in prose contain examples under discussion, not a + tool request the model is issuing. Masking them (to spaces, preserving + newlines and offsets) prevents the root scanner from starting a JSON + object inside them. Quotes inside an actual JSON object (depth > 0) are + left untouched, so a legitimate bare envelope survives masking. + """ + chars = list(content) + + for match in re.finditer(r"```.*?```", content, flags=re.DOTALL): + for index in range(match.start(), match.end()): + if chars[index] != "\n": + chars[index] = " " + + depth = 0 + quote: str | None = None + escaped = False + for index, char in enumerate(chars): + if quote: + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == quote: + quote = None + if char != "\n": + chars[index] = " " + continue + + if char in "{[": + depth += 1 + elif char in "}]": + depth = max(0, depth - 1) + elif depth == 0 and char in ('"', "`"): + quote = char + chars[index] = " " + + return "".join(chars) + + +def _parse_json_root_object_strict(root: str) -> dict[str, Any] | None: + try: + data = json.loads(root) + except Exception: + return None + return data if isinstance(data, dict) else None + + +_TOOL_FAILURE_REASONS = { + "empty": "empty response from the model", + "prose": "plain prose with no JSON tool object", + "truncated": "truncated or unterminated JSON tool request", + "invalid_envelope": "JSON object without a valid tool_name/tool_args envelope", +} + + +def classify_tool_request_failure(content: Any) -> str: + """Single classification of an unusable model response. + + Drives both the sanitized log reason (explain_tool_request_failure) and + the retry-prompt routing in Agent.process_tools, so the log label and the + reprompt choice can never diverge. Returns one of: "empty", "prose", + "truncated", "invalid_envelope". + """ + if not isinstance(content, str) or not content.strip(): + return "empty" + stripped = content.strip() + if "{" not in stripped: + return "prose" + + roots = extract_json_root_strings(stripped) + if roots: + # A complete (non-tool) root followed by an unterminated fragment + # (odd quote count, or a new "{") is trailing truncation, not a + # clean-but-invalid envelope. + tail = stripped.split(roots[-1], 1)[1].lstrip(" \t\r\n,") + if tail and (tail.startswith("{") or tail.count('"') % 2 == 1): + return "truncated" + return "invalid_envelope" + + return "truncated" + + +def explain_tool_request_failure(content: str, finish_reason: str = "") -> str: + """Sanitized classification of why no tool request could be extracted. + + Never includes response content; only a reason and the output length. + """ + prefix = "" + if finish_reason == "length": + prefix = "provider truncated the response (finish_reason=length); " + + category = classify_tool_request_failure(content) + reason = _TOOL_FAILURE_REASONS[category] + if category == "empty": + return f"{prefix}{reason}" + length = len(content.strip()) + return f"{prefix}{reason} ({length} chars)" + + def is_misformatted_tool_request(content: str) -> bool: if not content or not isinstance(content, str): return False @@ -75,6 +247,73 @@ def is_misformatted_tool_request(content: str) -> bool: ) ) +def _json_scan_final_depth(content: str) -> int: + """Return the unclosed-brace depth after scanning JSON-ish text. + + Tracks quote/escape state so braces inside strings do not count. Shared + by _json_root_object_balanced and is_truncated_tool_request so the two + scans cannot drift apart. Depth > 0 means the text is unterminated. + """ + depth = 0 + quote = None + escaped = False + for char in content: + if quote: + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == quote: + quote = None + continue + if depth and char in ('"', "'", "`"): + quote = char + elif char == "{": + depth += 1 + elif depth and char == "[": + depth += 1 + elif depth and char in ("}", "]"): + depth -= 1 + return depth + + +def _json_root_object_balanced(content: str) -> bool: + return _json_scan_final_depth(content) == 0 + + +# Structural gate: an unterminated payload only counts as a truncated tool +# request when it opens like a JSON tool envelope +# ({"thoughts"/"headline"/"tool_name"/"tool_args": ...) or a Responses +# function_call-style payload ({"type":"function", ...). A prose sentence +# that merely mentions "tool"/"actions"/"function" after a stray "{" is +# not a truncated request. "type" is included because in responses mode a +# truncated function-call payload must take the repair-prompt path instead +# of being shown to the user as a plain-text reply. +_TRUNCATION_ENVELOPE_OPEN_RE = re.compile( + r'\{\s*"(thoughts|headline|tool_name|tool_args|type)"' +) + + +def is_truncated_tool_request(content: str) -> bool: + """Return True when content contains an unterminated JSON tool envelope. + + Used by the harness to give a targeted retry prompt instead of a generic + misformat warning when providers cut a streaming/completion response + mid-object. The scan starts at the first envelope-shaped opening brace, + so a truncated payload wrapped in a ```json fence or preceded by prose + (or following a complete non-tool JSON object) is still recognized. + """ + if not content or not isinstance(content, str): + return False + content = content.strip() + match = _TRUNCATION_ENVELOPE_OPEN_RE.search(content) + if not match: + return False + candidate = content[match.start():] + if candidate.endswith("}") and _json_root_object_balanced(candidate): + return False + + return _json_scan_final_depth(candidate) > 0 def normalize_tool_request(tool_request: Any) -> tuple[str, dict]: if not isinstance(tool_request, dict): @@ -141,6 +380,9 @@ def extract_json_root_strings(content: str) -> list[str]: return [] if content.lstrip().startswith("["): + # Blind spot, by design: content starting with "[" is treated as a + # JSON array and never scanned for object roots, so embedded-envelope + # recovery can never fire for it. return [] roots: list[str] = [] diff --git a/helpers/litellm_transport.py b/helpers/litellm_transport.py index d5168b4fbf..89b1e91a37 100644 --- a/helpers/litellm_transport.py +++ b/helpers/litellm_transport.py @@ -192,6 +192,9 @@ class LiteLLMTransport: last_result: LLMResult | None = field(init=False, default=None) last_request_state: str = field(init=False, default=RESPONSES_STATE_PROVIDER) explicit_prompt_caching: bool = field(init=False, default=False) + # terminal finish_reason of the last chat-completions call/stream; "" + # means the stream ended without one (provider dropped the connection) + last_finish_reason: str = field(init=False, default="") def __post_init__(self) -> None: self.kwargs = _without_stream_kwarg(dict(self.kwargs)) @@ -217,6 +220,7 @@ def complete(self) -> ChatChunk: parsed = ChatCompletionsTransport.parse( completion(**self._chat_request(stream=False)) ) + self.last_finish_reason = str(parsed.get("finish_reason") or "") self.last_result = self._llm_result_from_chat(parsed) return parsed request = self._responses_request(stream=False) @@ -238,6 +242,7 @@ async def acomplete(self) -> ChatChunk: parsed = ChatCompletionsTransport.parse( await acompletion(**self._chat_request(stream=False)) ) + self.last_finish_reason = str(parsed.get("finish_reason") or "") self.last_result = self._llm_result_from_chat(parsed) return parsed request = self._responses_request(stream=False) @@ -257,6 +262,7 @@ def stream(self) -> Iterator[ChatChunk]: iterator = None exhausted = False got_any_chunk = False + self.last_finish_reason = "" try: if self.policy.mode is TransportMode.CHAT_COMPLETIONS: iterator = completion(**self._chat_request(stream=True)) @@ -270,6 +276,7 @@ def stream(self) -> Iterator[ChatChunk]: if _has_chunk_delta(parsed): got_any_chunk = True yield parsed + self.last_finish_reason = parser.finish_reason self.last_result = self._stream_result_from_chat_parser(parser) else: request = self._responses_request(stream=True) @@ -298,6 +305,7 @@ async def astream(self) -> AsyncIterator[ChatChunk]: iterator = None exhausted = False got_any_chunk = False + self.last_finish_reason = "" try: if self.policy.mode is TransportMode.CHAT_COMPLETIONS: iterator = await acompletion(**self._chat_request(stream=True)) @@ -311,6 +319,7 @@ async def astream(self) -> AsyncIterator[ChatChunk]: if _has_chunk_delta(parsed): got_any_chunk = True yield parsed + self.last_finish_reason = parser.finish_reason self.last_result = self._stream_result_from_chat_parser(parser) else: request = self._responses_request(stream=True) @@ -408,6 +417,7 @@ def _llm_result_from_chat(self, parsed: ChatChunk) -> LLMResult: output_items=parsed.get("_output_items"), provider_model_key=self.model, capability=self._capability_metadata(), + finish_reason=str(parsed.get("finish_reason") or ""), ) def _llm_result_from_response( @@ -442,6 +452,7 @@ def _stream_result_from_chat_parser( output_items=output_items, provider_model_key=self.model, capability=self._capability_metadata(), + finish_reason=parser.finish_reason, ) def _capability_metadata(self) -> dict[str, Any]: @@ -516,7 +527,11 @@ def parse(chunk: Any) -> ChatChunk: reasoning_delta = _get_value(delta, "reasoning_content") or _get_value( message, "reasoning_content" ) or "" - parsed = {"reasoning_delta": reasoning_delta, "response_delta": response_delta} + parsed = { + "reasoning_delta": reasoning_delta, + "response_delta": response_delta, + "finish_reason": str(_get_value(choice, "finish_reason") or ""), + } if not response_delta: tool_calls = _as_list(_get_value(message, "tool_calls")) response_delta = ChatCompletionsTransport.tool_calls_text(tool_calls) @@ -583,6 +598,7 @@ def __init__(self) -> None: self.tool_calls: dict[str, dict[str, Any]] = {} self.order: list[str] = [] self.emitted = False + self.finish_reason = "" def parse(self, chunk: Any) -> ChatChunk: parsed = ChatCompletionsTransport.parse(chunk) @@ -591,7 +607,11 @@ def parse(self, chunk: Any) -> ChatChunk: self._append_tool_calls(_get_value(delta, "tool_calls")) self._append_legacy_function_call(_get_value(delta, "function_call")) - if _get_value(choice, "finish_reason") in {"tool_calls", "function_call"}: + finish_reason = _get_value(choice, "finish_reason") + if finish_reason: + self.finish_reason = str(finish_reason) + + if finish_reason in {"tool_calls", "function_call"}: text = self._emit() if text and not parsed["response_delta"]: parsed["response_delta"] = text diff --git a/helpers/llm_result.py b/helpers/llm_result.py index a39234eab7..363b6bfa65 100644 --- a/helpers/llm_result.py +++ b/helpers/llm_result.py @@ -63,6 +63,7 @@ class LLMResult: usage: dict[str, Any] = field(default_factory=dict) raw: dict[str, Any] = field(default_factory=dict) capability: dict[str, Any] = field(default_factory=dict) + finish_reason: str = "" @classmethod def from_dict(cls, data: dict[str, Any] | None) -> "LLMResult": @@ -82,6 +83,7 @@ def from_dict(cls, data: dict[str, Any] | None) -> "LLMResult": usage=object_to_dict(data.get("usage") or {}), raw=object_to_dict(data.get("raw") or {}), capability=object_to_dict(data.get("capability") or {}), + finish_reason=str(data.get("finish_reason") or ""), ) @classmethod @@ -128,6 +130,7 @@ def from_chat( output_items: list[dict[str, Any]] | None = None, provider_model_key: str = "", capability: dict[str, Any] | None = None, + finish_reason: str = "", ) -> "LLMResult": items = [ResponseItem.from_any(item) for item in output_items or []] if response and not items: @@ -161,6 +164,7 @@ def from_chat( mode="chat_completions", state="off", capability=dict(capability or {}), + finish_reason=finish_reason, ) if not result.response and result.function_calls: result.response = result.function_calls_text() @@ -213,6 +217,7 @@ def to_dict(self) -> dict[str, Any]: "usage": self.usage, "raw": self.raw, "capability": self.capability, + "finish_reason": self.finish_reason, } def metadata(self) -> dict[str, Any]: @@ -226,6 +231,7 @@ def metadata(self) -> dict[str, Any]: "state": self.state, "usage": self.usage, "capability": self.capability, + "finish_reason": self.finish_reason, } } diff --git a/helpers/localization.py b/helpers/localization.py index eaf8c749a6..0693f5dcdb 100644 --- a/helpers/localization.py +++ b/helpers/localization.py @@ -97,8 +97,14 @@ def localize_naive_datetime(self, dt: datetime) -> datetime: except pytz.exceptions.NonExistentTimeError: return tzinfo.localize(dt, is_dst=True) - def set_timezone(self, timezone: str) -> None: - """Set the user's IANA timezone and propagate it to child processes.""" + def set_timezone(self, timezone: str, persist: bool = True) -> None: + """Set the user's IANA timezone and propagate it to child processes. + + With persist=False the change is runtime-only and the saved + DEFAULT_USER_TIMEZONE default is left untouched. Auto mode uses this + so a browser-reported (or spoofed) timezone cannot overwrite the + user's persisted default. + """ try: # Validate timezone and compute its current offset _ = pytz.timezone(timezone) @@ -114,8 +120,9 @@ def set_timezone(self, timezone: str) -> None: ) self._offset_minutes = new_offset self.timezone = timezone - save_dotenv_value("DEFAULT_USER_TIMEZONE", timezone) - save_dotenv_value("DEFAULT_USER_UTC_OFFSET_MINUTES", str(self._offset_minutes)) + if persist: + save_dotenv_value("DEFAULT_USER_TIMEZONE", timezone) + save_dotenv_value("DEFAULT_USER_UTC_OFFSET_MINUTES", str(self._offset_minutes)) self.apply_process_timezone() self._last_timezone_change = datetime.now() except pytz.exceptions.UnknownTimeZoneError: diff --git a/helpers/settings.py b/helpers/settings.py index ac6ba01512..27aea675e5 100644 --- a/helpers/settings.py +++ b/helpers/settings.py @@ -582,7 +582,12 @@ def _apply_timezone_setting(previous: Settings | None, browser_timezone: str | N ): return - localization.set_timezone(target_timezone) + # Auto mode follows the browser; do not persist the resolved value over + # the user's saved DEFAULT_USER_TIMEZONE default. + localization.set_timezone( + target_timezone, + persist=_settings["timezone"] != TIMEZONE_AUTO, + ) current_timezone = localization.get_timezone() if current_timezone == previous_timezone: return diff --git a/helpers/state_snapshot.py b/helpers/state_snapshot.py index b26c010a16..8ecbd2de25 100644 --- a/helpers/state_snapshot.py +++ b/helpers/state_snapshot.py @@ -258,7 +258,9 @@ async def build_snapshot_from_request(*, request: StateRequestV1) -> SnapshotV1: localization = Localization.get() previous_timezone = localization.get_timezone() - localization.set_timezone(request.timezone) + # Poll-reported timezones are per-request rendering context; never + # persist them over the user's saved DEFAULT_USER_TIMEZONE default. + localization.set_timezone(request.timezone, persist=False) current_timezone = localization.get_timezone() if current_timezone != previous_timezone: _notify_timezone_changed(previous_timezone, current_timezone) diff --git a/models.py b/models.py index e1fa5c4230..0b2bca09fe 100644 --- a/models.py +++ b/models.py @@ -300,7 +300,7 @@ def get_rate_limiter( def _is_transient_litellm_error(exc: Exception) -> bool: """Uses status_code when available, else falls back to exception types""" - # Prefer explicit status codes if present + exc_text = str(getattr(exc, "message", "") or exc).lower() status_code = getattr(exc, "status_code", None) if isinstance(status_code, int): if status_code in (408, 429, 500, 502, 503, 504): @@ -308,7 +308,18 @@ def _is_transient_litellm_error(exc: Exception) -> bool: # Treat other 5xx as retriable if status_code >= 500: return True - return False + # A body-parse failure ("unable to get json response") with a + # non-2xx status (e.g. a 400 whose body is an HTML proxy/WAF error + # page) is a permanent request error, not a dropped connection. + if not 200 <= status_code < 300: + return False + # 2xx: the provider closed the connection mid-body; LiteLLM then + # reports the truncated response with the *original* success status, + # which is transient even though the HTTP status says success. + return "unable to get json response" in exc_text + + if "unable to get json response" in exc_text: + return True # Fallback to exception classes mapped by LiteLLM/OpenAI transient_types = ( @@ -323,6 +334,62 @@ def _is_transient_litellm_error(exc: Exception) -> bool: return isinstance(exc, transient_types) +def _is_direct_deepseek_model(transport: Any) -> bool: + """True only for the direct DeepSeek API (litellm `deepseek/...` models). + + Third-party hosts serving deepseek-* weights (openrouter/deepseek/..., + ollama/deepseek-r1, ...) have different streaming/empty-completion + behavior and must not pay the duplicate-generation retries. + """ + model = str(getattr(transport, "model", "")).lower() + return model.startswith("deepseek/") + + +def _should_retry_truncated_stream( + transport: Any, *, stream: bool, stopped_early: bool +) -> bool: + """Detect provider-dropped chat-completions streams. + + DeepSeek intermittently closes streaming connections mid-response; the + HTTP layer then ends the iterator without an error, so the completion + never delivers a terminal finish_reason. Partial output from such a + stream (typically unterminated JSON) must be retried, not parsed. + + Exempted: non-stream calls, the agent's own early-stop (which breaks the + stream before finish_reason arrives by design), the responses API, and + third-party-hosted deepseek-* models that may legitimately omit + finish_reason. + """ + if not stream or stopped_early: + return False + policy = getattr(transport, "policy", None) + if policy is None or getattr(policy, "using_responses", False): + return False + if not _is_direct_deepseek_model(transport): + return False + return not getattr(transport, "last_finish_reason", "") + + +def _should_retry_empty_completion( + transport: Any, response: str, *, stopped_early: bool +) -> bool: + """Detect provider-completed but empty completions. + + DeepSeek V4 Flash in thinking/JSON mode occasionally finishes normally + (finish_reason=stop) with whitespace-only content after a full reasoning + stream. An empty response can never satisfy the tool protocol, so the + turn is retried instead of being reported as a misformat. + """ + if stopped_early: + return False + policy = getattr(transport, "policy", None) + if policy is None or getattr(policy, "using_responses", False): + return False + if not _is_direct_deepseek_model(transport): + return False + return not response.strip() + + async def apply_rate_limiter( model_config: ModelConfig | None, input_text: str, @@ -566,6 +633,10 @@ async def unified_call( call_kwargs["a0_explicit_prompt_caching"] = True max_retries: int = int(call_kwargs.pop("a0_retry_attempts", 2)) retry_delay_s: float = float(call_kwargs.pop("a0_retry_delay_seconds", 1.5)) + # Main chat turns retry provider-completed empty responses; utility + # callers (e.g. memory post-processing) pass a0_allow_empty_completion + # because an empty reply is a benign answer for them. + allow_empty: bool = bool(call_kwargs.pop("a0_allow_empty_completion", False)) stream = reasoning_callback is not None or response_callback is not None or tokens_callback is not None transport = LiteLLMTransport( model=self.model_name, @@ -579,9 +650,9 @@ async def unified_call( attempt = 0 while True: got_any_chunk = False + stop_response: str | None = None try: if stream: - stop_response: str | None = None async for parsed in transport.astream(): got_any_chunk = True output = result.add_chunk(parsed) @@ -626,6 +697,28 @@ async def unified_call( if output["reasoning_delta"]: limiter.add(output=approximate_tokens(output["reasoning_delta"])) + # provider dropped the stream mid-response or completed with + # empty content: discard the partial output and retry the + # whole turn instead of parsing broken/empty JSON + truncated = _should_retry_truncated_stream( + transport, stream=stream, stopped_early=stop_response is not None + ) + empty = ( + not allow_empty + and _should_retry_empty_completion( + transport, + result.response, + stopped_early=stop_response is not None, + ) + ) + if (truncated or empty) and attempt < max_retries: + import asyncio + + attempt += 1 + result = ChatGenerationResult() + await asyncio.sleep(retry_delay_s) + continue + # Successful completion of stream return result.response, result.reasoning @@ -682,6 +775,9 @@ async def unified_turn( call_kwargs["a0_explicit_prompt_caching"] = True max_retries: int = int(call_kwargs.pop("a0_retry_attempts", 2)) retry_delay_s: float = float(call_kwargs.pop("a0_retry_delay_seconds", 1.5)) + # See unified_call: utility callers opt out of the empty-completion + # retry via a0_allow_empty_completion. + allow_empty: bool = bool(call_kwargs.pop("a0_allow_empty_completion", False)) stream = ( reasoning_callback is not None or response_callback is not None @@ -698,9 +794,9 @@ async def unified_turn( attempt = 0 while True: got_any_chunk = False + stop_response: str | None = None try: if stream: - stop_response: str | None = None async for parsed in transport.astream(): got_any_chunk = True output = result.add_chunk(parsed) @@ -757,12 +853,35 @@ async def unified_turn( output=approximate_tokens(output["reasoning_delta"]) ) + # provider dropped the stream mid-response or completed with + # empty content: discard the partial output and retry the + # whole turn instead of parsing broken/empty JSON + truncated = _should_retry_truncated_stream( + transport, stream=stream, stopped_early=stop_response is not None + ) + empty = ( + not allow_empty + and _should_retry_empty_completion( + transport, + result.response, + stopped_early=stop_response is not None, + ) + ) + if (truncated or empty) and attempt < max_retries: + import asyncio + + attempt += 1 + result = ChatGenerationResult() + await asyncio.sleep(retry_delay_s) + continue + llm_result = transport.last_result or LLMResult.from_chat( response=result.output()["response_delta"], reasoning=result.output()["reasoning_delta"], input_items=ResponsesTransport.input_from_messages(msgs_conv), provider_model_key=self.model_name, capability=transport._capability_metadata(), + finish_reason=transport.last_finish_reason, ) if result.output()["response_delta"] and not llm_result.function_calls: llm_result.response = result.output()["response_delta"] diff --git a/plugins/_a0_connector/tools/code_execution_remote.py b/plugins/_a0_connector/tools/code_execution_remote.py index 00457d6c7a..6f9caf6ff6 100644 --- a/plugins/_a0_connector/tools/code_execution_remote.py +++ b/plugins/_a0_connector/tools/code_execution_remote.py @@ -23,6 +23,11 @@ EXEC_OP_TRANSPORT_GRACE = 15.0 EXEC_OP_DEFAULT_TIMEOUT = 120.0 EXEC_OP_EVENT = "connector_exec_op" +# When no CLI is connected (for example right after an Agent Zero restart wiped +# the in-memory /ws registry), give the CLI's automatic reconnect a chance to +# re-register before failing the execution request. +NO_CLI_RECONNECT_GRACE_SECONDS = 60.0 +NO_CLI_RECONNECT_POLL_SECONDS = 2.0 _TIMEOUT_KEYS = ( "first_output_timeout", "between_output_timeout", @@ -78,6 +83,25 @@ def _wait_timeout_for_runtime(runtime: str, exec_config: dict[str, Any]) -> floa return EXEC_OP_DEFAULT_TIMEOUT return max(float(value) for value in timeouts.values()) + EXEC_OP_TRANSPORT_GRACE + async def _await_cli_reconnect( + self, context_id: str, *, require_writes: bool + ) -> str | None: + loop = asyncio.get_running_loop() + deadline = loop.time() + NO_CLI_RECONNECT_GRACE_SECONDS + while True: + remaining = deadline - loop.time() + if remaining <= 0: + return None + await asyncio.sleep(min(NO_CLI_RECONNECT_POLL_SECONDS, remaining)) + sid = select_remote_exec_target_sid(context_id, require_writes=require_writes) + if sid: + return sid + if remote_tool_sids_for_context(context_id): + # A CLI is connected but currently unusable (exec disabled or + # writes blocked): stop waiting and let the caller classify + # it instead of stalling for the full grace period. + return None + def get_log_object(self): import uuid @@ -115,10 +139,17 @@ async def execute(self, **kwargs: Any) -> Response: ) context_id = self.agent.context.id - candidates = remote_tool_sids_for_context(context_id) require_writes = self._runtime_requires_write_access(runtime) sid = select_remote_exec_target_sid(context_id, require_writes=require_writes) + if not sid and not remote_tool_sids_for_context(context_id): + # No CLI is connected at all (typically right after a server restart): + # wait briefly for the CLI's automatic reconnect before giving up. + await self.set_progress( + "Waiting for the a0 CLI to reconnect..." + ) + sid = await self._await_cli_reconnect(context_id, require_writes=require_writes) if not sid: + candidates = remote_tool_sids_for_context(context_id) exec_enabled = False write_blocked = False for candidate_sid in candidates: @@ -139,16 +170,21 @@ async def execute(self, **kwargs: Any) -> Response: message=( "code_execution_remote: no connected CLI currently allows " "shell-backed execution that may modify local files. Press F3 to switch " - "the CLI to Read&Write. `runtime=output` and `runtime=reset` remain " - "available for existing sessions." + "the CLI to Read&Write, then ask the agent to continue. " + "`runtime=output` and `runtime=reset` remain available for " + "existing sessions." if candidates and require_writes and exec_enabled and write_blocked else "code_execution_remote: no connected CLI currently has " - "remote execution enabled. Connect the CLI and press F4 to switch exec on." + "remote execution enabled. Press F4 in the CLI to switch exec on, " + "then ask the agent to continue." if candidates - else "code_execution_remote: no CLI client connected to Agent Zero. " - "Make sure the CLI is connected to this instance." + else "code_execution_remote: no CLI client connected to Agent Zero " + f"after waiting {NO_CLI_RECONNECT_GRACE_SECONDS:g}s for automatic " + "reconnection. This is an infrastructure condition, not a task error, " + "so this turn ends instead of retrying. Reconnect or restart the a0 " + "CLI against this instance, then ask the agent to continue." ), - break_loop=False, + break_loop=True, ) try: diff --git a/plugins/_memory/extensions/python/monologue_end/_50_memorize_fragments.py b/plugins/_memory/extensions/python/monologue_end/_50_memorize_fragments.py index 59bd71227f..44addb21f5 100644 --- a/plugins/_memory/extensions/python/monologue_end/_50_memorize_fragments.py +++ b/plugins/_memory/extensions/python/monologue_end/_50_memorize_fragments.py @@ -1,5 +1,5 @@ import asyncio -from helpers import errors, plugins +from helpers import plugins from helpers.extension import Extension from helpers.dirty_json import DirtyJson from agent import LoopData @@ -9,6 +9,7 @@ # Direct import - this extension lives inside the memory plugin from plugins._memory.helpers.memory import Memory from plugins._memory.helpers.memory_quality import filter_auto_memory_fragments +from plugins._memory.helpers.memorize_lock import format_utility_error, get_memorize_lock from plugins._memory.tools.memory_load import DEFAULT_THRESHOLD as DEFAULT_MEMORY_THRESHOLD @@ -41,6 +42,9 @@ async def memorize(self, loop_data: LoopData, log_item: LogItem, **kwargs): if not self.agent: return + # serialize with _51_memorize_solutions: one utility job at a time + lock = get_memorize_lock() + await lock.acquire() try: set = plugins.get_plugin_config("_memory", self.agent) if not set: @@ -220,7 +224,9 @@ async def memorize(self, loop_data: LoopData, log_item: LogItem, **kwargs): except Exception as e: - err = errors.format_error(e) + err = format_utility_error(e) self.agent.context.log.log( type="warning", heading="Memorize memories extension error", content=err ) + finally: + lock.release() diff --git a/plugins/_memory/extensions/python/monologue_end/_51_memorize_solutions.py b/plugins/_memory/extensions/python/monologue_end/_51_memorize_solutions.py index 64ceeb6124..3e33309e60 100644 --- a/plugins/_memory/extensions/python/monologue_end/_51_memorize_solutions.py +++ b/plugins/_memory/extensions/python/monologue_end/_51_memorize_solutions.py @@ -1,5 +1,5 @@ import asyncio -from helpers import errors, plugins +from helpers import plugins from helpers.extension import Extension from helpers.dirty_json import DirtyJson from agent import LoopData @@ -8,6 +8,7 @@ # Direct import - this extension lives inside the memory plugin from plugins._memory.helpers.memory import Memory +from plugins._memory.helpers.memorize_lock import format_utility_error, get_memorize_lock from plugins._memory.tools.memory_load import DEFAULT_THRESHOLD as DEFAULT_MEMORY_THRESHOLD class MemorizeSolutions(Extension): @@ -39,6 +40,9 @@ async def memorize(self, loop_data: LoopData, log_item: LogItem, **kwargs): if not self.agent: return + # serialize with _50_memorize_fragments: one utility job at a time + lock = get_memorize_lock() + await lock.acquire() try: set = plugins.get_plugin_config("_memory", self.agent) @@ -211,7 +215,9 @@ async def memorize(self, loop_data: LoopData, log_item: LogItem, **kwargs): except Exception as e: - err = errors.format_error(e) + err = format_utility_error(e) self.agent.context.log.log( type="warning", heading="Memorize solutions extension error", content=err ) + finally: + lock.release() diff --git a/plugins/_memory/helpers/memorize_lock.py b/plugins/_memory/helpers/memorize_lock.py new file mode 100644 index 0000000000..e7a8eb10c1 --- /dev/null +++ b/plugins/_memory/helpers/memorize_lock.py @@ -0,0 +1,40 @@ +"""Shared lock serializing memory post-processing utility jobs. + +_50_memorize_fragments and _51_memorize_solutions both run heavyweight +utility-model requests asynchronously on the shared background event loop. +Running them concurrently means two long-held, non-streaming provider +requests at once, which providers may truncate or rate-limit. The lock is +created lazily inside the loop that first uses it and recreated when the +running loop changes, because an asyncio.Lock is bound to the loop it was +first used on: if the background loop is terminated and recreated (see +EventLoopThread.terminate), a stale lock would either raise "bound to a +different event loop" or hang forever if it was left locked. +""" + +import asyncio + +_lock: asyncio.Lock | None = None +_lock_loop: asyncio.AbstractEventLoop | None = None + + +def get_memorize_lock() -> asyncio.Lock: + global _lock, _lock_loop + try: + loop: asyncio.AbstractEventLoop | None = asyncio.get_running_loop() + except RuntimeError: + loop = None + if _lock is None or loop is not _lock_loop: + _lock = asyncio.Lock() + _lock_loop = loop + return _lock + + +def format_utility_error(exc: Exception) -> str: + """Provider-aware error text: status code first when the API supplied one.""" + from helpers import errors + + err = errors.format_error(exc) + status = getattr(exc, "status_code", None) + if status is not None: + return f"Provider status {status}: {err}" + return err diff --git a/prompts/fw.msg_recovered_request.md b/prompts/fw.msg_recovered_request.md new file mode 100644 index 0000000000..75fd8f853a --- /dev/null +++ b/prompts/fw.msg_recovered_request.md @@ -0,0 +1,3 @@ +Your previous message contained a valid JSON tool request embedded in surrounding prose. +The tool request was executed, but the prose around it was discarded. Respond with ONLY +the JSON tool request object (thoughts, headline, tool_name, tool_args) and no other text. diff --git a/prompts/fw.msg_truncated_request.md b/prompts/fw.msg_truncated_request.md new file mode 100644 index 0000000000..0bf5ef0723 --- /dev/null +++ b/prompts/fw.msg_truncated_request.md @@ -0,0 +1,8 @@ +Your previous JSON tool request was truncated or unterminated before completion. +Respond ONLY with a complete, valid JSON tool request using the required format +(thoughts, headline, tool_name, tool_args). Do not repeat the truncated text, +do not add prose, and make sure the JSON object is closed with "}". +If the truncated request carried a large payload (file contents, patch scripts, +long documents), do not retry the same large request: split the work into +multiple smaller sequential tool calls instead, e.g. write the file in several +smaller parts. diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000000..6f30a3a8fe --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,116 @@ +"""Pytest session bootstrap for the framework test suite. + +- Triggers the Telegram plugin's lazy runtime dependency install (aiogram) + before collection so test_telegram_* modules import cleanly. If the + install is not possible (e.g. offline environment), the telegram modules + are excluded from collection instead of erroring the whole suite. +- Excludes legacy manual scripts that are not automated tests. +- Guards the live usr tree: any test that tries to write under /usr + fails loudly. Inside the deployed container that path is the persistent + volume holding real chats, model presets, .env secrets and time-travel + history, and suite runs have corrupted it in the past. Tests that need a + usr tree must redirect helpers.files._base_dir to tmp_path. +""" + +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from helpers import dotenv as a0_dotenv +from helpers import files as a0_files + +_REAL_USR_DIR = os.path.realpath(os.path.join(a0_files.get_base_dir(), "usr")) + + +def _live_usr_path(path: object) -> str | None: + """Return the resolved live-usr path a write would hit, or None.""" + try: + candidate = str(path) + except Exception: + return None + if not os.path.isabs(candidate): + candidate = os.path.join(a0_files.get_base_dir(), candidate) + real = os.path.realpath(candidate) + if real == _REAL_USR_DIR or real.startswith(_REAL_USR_DIR + os.sep): + return real + return None + + +@pytest.fixture(autouse=True) +def _guard_live_usr_writes(monkeypatch): + """Fail any test that writes into the live Agent Zero usr tree.""" + + def guard_write(original): + def wrapper(relative_path, *args, **kwargs): + hit = _live_usr_path(relative_path) + if hit is not None: + raise RuntimeError( + f"test attempted to write live Agent Zero usr path: {hit}" + ) + return original(relative_path, *args, **kwargs) + + return wrapper + + for name in ( + "write_file", + "write_file_bin", + "write_file_base64", + "delete_file", + "delete_dir", + ): + monkeypatch.setattr(a0_files, name, guard_write(getattr(a0_files, name))) + + original_save = a0_dotenv.save_dotenv_value + + def guarded_save_dotenv_value(key, value): + hit = _live_usr_path(a0_dotenv.get_dotenv_file_path()) + if hit is not None: + raise RuntimeError( + f"test attempted to write live Agent Zero env file: {hit}" + ) + return original_save(key, value) + + monkeypatch.setattr(a0_dotenv, "save_dotenv_value", guarded_save_dotenv_value) + # helpers.localization imported save_dotenv_value by value, so it needs + # its own patch; tests that stub it themselves simply replace this one. + from helpers import localization as a0_localization + + monkeypatch.setattr( + a0_localization, "save_dotenv_value", guarded_save_dotenv_value + ) + yield + +collect_ignore = [ + # Legacy manual smoke scripts, not automated tests. email_parser_test.py + # imports helpers.email_client.read_messages, which no longer exists (the + # module is fully commented out), and its only test is marked skip with a + # note asking to move it to a script. rate_limiter_test.py performs a real + # LLM API call at import time. + "email_parser_test.py", + "rate_limiter_test.py", +] + +try: + from plugins._telegram_integration.helpers.dependencies import ( + ensure_dependencies, + ) + + ensure_dependencies() +except (RuntimeError, subprocess.CalledProcessError) as exc: + # Dependency installation failed (e.g. offline environment): exclude the + # telegram tests at collection time rather than failing the whole suite, + # but say so loudly - silently erasing five test files would mask real + # plugin regressions in CI. Import errors from plugin code itself are + # intentionally NOT caught here and will fail the run. + print( + f"WARNING: telegram dependencies unavailable ({exc}); " + "excluding test_telegram_* from collection." + ) + collect_ignore_glob = ["test_telegram_*.py"] diff --git a/tests/test_browser_agent_regressions.py b/tests/test_browser_agent_regressions.py index eaa49ef4a2..22978477b8 100644 --- a/tests/test_browser_agent_regressions.py +++ b/tests/test_browser_agent_regressions.py @@ -67,10 +67,24 @@ def error(code="", message="", correlation_id=None): ) -sys.modules.setdefault("agent", SimpleNamespace(AgentContext=_TestAgentContext)) -sys.modules.setdefault("helpers.tool", SimpleNamespace(Response=_TestResponse, Tool=_TestTool)) -sys.modules.setdefault("helpers.ws", SimpleNamespace(WsHandler=_TestWsHandler)) -sys.modules.setdefault("helpers.ws_manager", SimpleNamespace(WsResult=_TestWsResult)) +def _import_or_stub(module_name: str, stub: object) -> None: + """Use the real module when it imports cleanly; only fall back to a stub + in environments where the real import chain is unavailable. + + Unconditional sys.modules stubs poison every test module collected after + this one: they see the stub instead of the real module, and whether the + stub or the real module wins depends on collection order. + """ + try: + __import__(module_name) + except Exception: + sys.modules.setdefault(module_name, stub) + + +_import_or_stub("agent", SimpleNamespace(AgentContext=_TestAgentContext)) +_import_or_stub("helpers.tool", SimpleNamespace(Response=_TestResponse, Tool=_TestTool)) +_import_or_stub("helpers.ws", SimpleNamespace(WsHandler=_TestWsHandler)) +_import_or_stub("helpers.ws_manager", SimpleNamespace(WsResult=_TestWsResult)) _model_config_stub = ModuleType("plugins._model_config.helpers.model_config") _model_config_stub.get_presets = lambda: [] _model_config_stub.get_preset_by_name = lambda name: None @@ -2902,6 +2916,16 @@ async def fake_get_runtime(context_id, create=True): manager=None, ) + # emit_to requires a bound WsManager when the real helpers.ws module is + # in use (instead of the lightweight stub above); shadow it on the + # instance so the test behaves identically in both regimes. + emissions = [] + + async def _record_emit(sid, event, data, correlation_id=None): + emissions.append((sid, event, data)) + + handler.emit_to = _record_emit + result = await handler.process( "browser_viewer_command", {"context_id": "ctx-a", "command": "list"}, @@ -2953,6 +2977,16 @@ async def fake_list_runtime_sessions(): manager=None, ) + # emit_to requires a bound WsManager when the real helpers.ws module is + # in use (instead of the lightweight stub above); shadow it on the + # instance so the test behaves identically in both regimes. + emissions = [] + + async def _record_emit(sid, event, data, correlation_id=None): + emissions.append((sid, event, data)) + + handler.emit_to = _record_emit + result = await handler.process( "browser_viewer_command", {"context_id": "ctx-a", "command": "list"}, diff --git a/tests/test_code_execution_remote_reconnect.py b/tests/test_code_execution_remote_reconnect.py new file mode 100644 index 0000000000..cdaaf94501 --- /dev/null +++ b/tests/test_code_execution_remote_reconnect.py @@ -0,0 +1,178 @@ +"""Regression tests for code_execution_remote CLI reconnect behavior. + +Root cause: when the Agent Zero container restarts, the server-side /ws +registry is wiped and the a0 CLI must reconnect. While no CLI is connected, +code_execution_remote used to return a soft tool result, so the main agent +kept retrying with fresh session numbers instead of waiting for the +automatic reconnect and then handing control back to the user. +""" + +import asyncio +import sys +import time +import uuid +from pathlib import Path +from types import SimpleNamespace + +import pytest + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + + +from plugins._a0_connector.helpers import ws_runtime +from plugins._a0_connector.tools import code_execution_remote +from plugins._a0_connector.tools.code_execution_remote import CodeExecutionRemote + + +def _make_tool(args: dict) -> CodeExecutionRemote: + context_id = f"ctx-{uuid.uuid4().hex[:8]}" + agent = SimpleNamespace(context=SimpleNamespace(id=context_id)) + return CodeExecutionRemote(agent, "code_execution_remote", None, args, "", None) + + +@pytest.fixture(autouse=True) +def _fast_reconnect_grace(monkeypatch): + monkeypatch.setattr(code_execution_remote, "NO_CLI_RECONNECT_GRACE_SECONDS", 0.6) + monkeypatch.setattr(code_execution_remote, "NO_CLI_RECONNECT_POLL_SECONDS", 0.05) + # set_progress drives the extension chain, which needs a real Agent; the + # reconnect behavior under test does not depend on progress reporting. + monkeypatch.setattr( + CodeExecutionRemote, + "set_progress", + lambda self, content: asyncio.sleep(0), + ) + + +@pytest.fixture +def registered_sid(): + sid = f"sid-{uuid.uuid4().hex[:8]}" + yield sid + ws_runtime.unregister_sid(sid) + + +@pytest.mark.asyncio +async def test_no_cli_waits_for_reconnect_then_breaks_loop_with_guidance() -> None: + tool = _make_tool({"runtime": "terminal", "code": "echo hi", "session": 0}) + + started = time.monotonic() + response = await tool.execute() + elapsed = time.monotonic() - started + + assert elapsed >= code_execution_remote.NO_CLI_RECONNECT_GRACE_SECONDS + assert response.break_loop is True + message = response.message.lower() + assert "no cli client connected" in message + assert "reconnect" in message + + +@pytest.mark.asyncio +async def test_cli_reconnect_during_grace_proceeds_with_execution( + monkeypatch, registered_sid +) -> None: + sid = registered_sid + tool = _make_tool({"runtime": "terminal", "code": "echo hi", "session": 0}) + context_id = tool.agent.context.id + + class _FakeWsManager: + async def emit_to(self, namespace, target_sid, event, payload, handler_id=None): + assert target_sid == sid + ws_runtime.resolve_pending_exec_op( + payload["op_id"], + sid=target_sid, + payload={"ok": True, "result": {"output": "reconnected-output"}}, + ) + + monkeypatch.setattr( + code_execution_remote, "get_shared_ws_manager", lambda: _FakeWsManager() + ) + monkeypatch.setattr( + code_execution_remote, "build_exec_config", lambda **kwargs: {} + ) + + async def _register_later() -> None: + await asyncio.sleep(0.2) + ws_runtime.register_sid(sid) + ws_runtime.subscribe_sid_to_context(sid, context_id) + ws_runtime.store_sid_remote_exec_metadata(sid, {"enabled": True}) + ws_runtime.store_sid_remote_file_metadata( + sid, {"enabled": True, "write_enabled": True} + ) + + registration = asyncio.create_task(_register_later()) + try: + response = await tool.execute() + finally: + await registration + + assert response.break_loop is False + assert "reconnected-output" in response.message + + +@pytest.mark.asyncio +async def test_connected_cli_with_exec_disabled_does_not_wait_and_breaks_loop( + registered_sid, +) -> None: + sid = registered_sid + ws_runtime.register_sid(sid) + ws_runtime.store_sid_remote_exec_metadata(sid, {"enabled": False}) + + tool = _make_tool({"runtime": "output", "session": 0}) + + started = time.monotonic() + response = await tool.execute() + elapsed = time.monotonic() - started + + assert elapsed < code_execution_remote.NO_CLI_RECONNECT_GRACE_SECONDS + assert response.break_loop is True + assert "F4" in response.message + + +@pytest.mark.asyncio +async def test_cli_reconnect_with_exec_disabled_exits_grace_early( + registered_sid, +) -> None: + sid = registered_sid + tool = _make_tool({"runtime": "output", "session": 0}) + + async def _register_later() -> None: + await asyncio.sleep(0.2) + ws_runtime.register_sid(sid) + ws_runtime.store_sid_remote_exec_metadata(sid, {"enabled": False}) + + registration = asyncio.create_task(_register_later()) + try: + started = time.monotonic() + response = await tool.execute() + elapsed = time.monotonic() - started + finally: + await registration + + # The CLI became visible mid-grace with exec disabled: the tool must not + # stall for the full grace period before emitting the F4 guidance. + assert elapsed < code_execution_remote.NO_CLI_RECONNECT_GRACE_SECONDS + assert response.break_loop is True + assert "F4" in response.message + + +@pytest.mark.asyncio +async def test_connected_cli_write_blocked_does_not_wait_and_breaks_loop( + registered_sid, +) -> None: + sid = registered_sid + ws_runtime.register_sid(sid) + ws_runtime.store_sid_remote_exec_metadata(sid, {"enabled": True}) + ws_runtime.store_sid_remote_file_metadata( + sid, {"enabled": True, "write_enabled": False} + ) + + tool = _make_tool({"runtime": "terminal", "code": "echo hi", "session": 0}) + + started = time.monotonic() + response = await tool.execute() + elapsed = time.monotonic() - started + + assert elapsed < code_execution_remote.NO_CLI_RECONNECT_GRACE_SECONDS + assert response.break_loop is True + assert "F3" in response.message diff --git a/tests/test_deepseek_harness_reliability.py b/tests/test_deepseek_harness_reliability.py new file mode 100644 index 0000000000..60a66ba076 --- /dev/null +++ b/tests/test_deepseek_harness_reliability.py @@ -0,0 +1,1234 @@ +"""Regression tests for DeepSeek V4 Flash harness reliability. + +Covers the failure modes observed with deepseek/deepseek-v4-flash: +- planning prose before an otherwise valid JSON tool envelope +- provider-truncated output (finish_reason=length) being indistinguishable + from a malformed response +- concurrent memory post-processing jobs racing on the background loop +""" + +from __future__ import annotations + +import asyncio +import sys +from pathlib import Path + +import pytest + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from helpers.extract_tools import ( + classify_tool_request_failure, + explain_tool_request_failure, + extract_tool_request, + is_truncated_tool_request, + recover_embedded_tool_request, +) +from helpers.litellm_transport import ( + ChatCompletionsStreamParser, + ChatCompletionsTransport, +) +from helpers.llm_result import LLMResult +from plugins._memory.helpers.memorize_lock import get_memorize_lock + + +TOOL_REQUEST = ( + '{"thoughts":["check state"],"headline":"Checking state",' + '"tool_name":"code_execution_tool","tool_args":{"code":"ls"}}' +) + + +# --- embedded tool-request recovery ----------------------------------------- + + +def test_recover_embedded_tool_request_accepts_prose_before_json() -> None: + content = ( + "I'll first verify the current state, then apply the fix.\n\n" + + TOOL_REQUEST + ) + + assert extract_tool_request(content) is None # strict path stays strict + recovered = recover_embedded_tool_request(content) + assert recovered is not None + assert recovered["tool_name"] == "code_execution_tool" + assert recovered["tool_args"] == {"code": "ls"} + + +def test_recover_embedded_tool_request_accepts_prose_after_json() -> None: + content = TOOL_REQUEST + "\n\nThat will show the current files." + + recovered = recover_embedded_tool_request(content) + assert recovered is not None + assert recovered["tool_name"] == "code_execution_tool" + + +def test_recover_embedded_tool_request_rejects_multiple_distinct_requests() -> None: + other = ( + '{"thoughts":["reply"],"headline":"Reply",' + '"tool_name":"response","tool_args":{"text":"hi"}}' + ) + content = f"First option:\n{TOOL_REQUEST}\nSecond option:\n{other}" + + assert recover_embedded_tool_request(content) is None + + +def test_recover_embedded_tool_request_dedupes_repeated_identical_request() -> None: + content = f"As planned:\n{TOOL_REQUEST}\nRepeating for clarity:\n{TOOL_REQUEST}" + + recovered = recover_embedded_tool_request(content) + assert recovered is not None + assert recovered["tool_name"] == "code_execution_tool" + + +def test_recover_embedded_tool_request_rejects_plain_prose() -> None: + assert recover_embedded_tool_request("I need to think about this first.") is None + assert recover_embedded_tool_request("") is None + + +def test_recover_embedded_tool_request_rejects_non_tool_json() -> None: + assert recover_embedded_tool_request('Here is the status: {"status":"ok"}') is None + + +def test_recover_embedded_tool_request_rejects_quoted_example() -> None: + # An envelope quoted inside prose is an example being discussed, not a + # tool request the model is issuing; it must never be executed. + content = ( + 'I must reply in the format "{"thoughts":["x"],"tool_name":"response",' + '"tool_args":{"text":"hi"}}" exactly.' + ) + + assert recover_embedded_tool_request(content) is None + + +def test_recover_embedded_tool_request_rejects_fenced_example() -> None: + content = ( + "The protocol looks like this:\n```json\n" + + TOOL_REQUEST + + "\n```\nI will follow it now." + ) + + assert recover_embedded_tool_request(content) is None + + +def test_recover_embedded_tool_request_rejects_inline_code_example() -> None: + content = ( + 'Use `{"thoughts":["x"],"tool_name":"response","tool_args":{"text":"hi"}}`' + " to reply." + ) + + assert recover_embedded_tool_request(content) is None + + +def test_recover_embedded_tool_request_rejects_dirty_json_envelope() -> None: + # Lenient (dirty) JSON forms such as single-quoted keys are not a valid + # executable envelope even when embedded in prose. + content = ( + "Let me try: {'thoughts':['x'],'tool_name':'response'," + "'tool_args':{'text':'hi'}}" + ) + + assert recover_embedded_tool_request(content) is None + + +def test_recover_embedded_tool_request_accepts_bare_envelope_with_quoted_prose() -> None: + # Masking prose quotes must not destroy a real bare envelope that itself + # contains quoted strings. + content = ( + 'The user said "please list files", so I will run the command.\n\n' + + TOOL_REQUEST + ) + + recovered = recover_embedded_tool_request(content) + assert recovered is not None + assert recovered["tool_name"] == "code_execution_tool" + + +def test_recover_embedded_tool_request_refuses_response_tool_in_deliberating_prose() -> None: + # Adversarial case from review: hedged deliberation about how the model + # *could* reply is not a completed task and must not execute "response". + content = ( + 'I could reply {"thoughts":["x"],"tool_name":"response",' + '"tool_args":{"text":"hi"}} but let me think' + ) + assert recover_embedded_tool_request(content) is None + + +def test_recover_embedded_tool_request_refuses_response_tool_in_long_prose() -> None: + envelope = ( + '{"thoughts":["x"],"tool_name":"response","tool_args":{"text":"hi"}}' + ) + prose = "Let me carefully consider the best way to answer this question. " + assert len("".join(prose.split())) > 40 # precondition: substantial prose + assert recover_embedded_tool_request(prose + envelope) is None + + +def test_recover_embedded_tool_request_accepts_response_tool_with_minimal_leadin() -> None: + envelope = ( + '{"thoughts":["x"],"tool_name":"response","tool_args":{"text":"hi"}}' + ) + recovered = recover_embedded_tool_request("Reply: " + envelope) + assert recovered is not None + assert recovered["tool_name"] == "response" + + +def test_recover_embedded_tool_request_keeps_hedged_prose_for_operational_tools() -> None: + # Hedged prose only disqualifies the "response" tool; executing a valid + # operational envelope is the protocol's intent. + content = "I could run " + TOOL_REQUEST + " but let me check first" + recovered = recover_embedded_tool_request(content) + assert recovered is not None + assert recovered["tool_name"] == "code_execution_tool" + + +# --- transient classification precision -------------------------------------- + + +def _litellm_style_error(message: str, status_code: int | None = None) -> Exception: + exc = Exception(message) + exc.status_code = status_code # type: ignore[attr-defined] + return exc + + +def test_transient_classification_does_not_retry_permanent_body_parse_failures() -> None: + from models import _is_transient_litellm_error + + # A 400 whose body is an HTML proxy/WAF error page surfaces as + # "unable to get json response" but is a permanent request error. + exc = _litellm_style_error("Unable to get json response: …", 400) + + assert _is_transient_litellm_error(exc) is False + + +def test_transient_classification_retries_mid_body_disconnect_with_200() -> None: + from models import _is_transient_litellm_error + + exc = _litellm_style_error("Unable to get json response: …", 200) + + assert _is_transient_litellm_error(exc) is True + + +def test_transient_classification_retries_body_parse_failure_without_status() -> None: + from models import _is_transient_litellm_error + + exc = _litellm_style_error("Unable to get json response: …", None) + + assert _is_transient_litellm_error(exc) is True + + +def test_deepseek_retry_predicates_ignore_third_party_deepseek_models() -> None: + from models import _should_retry_empty_completion, _should_retry_truncated_stream + + class _Policy: + using_responses = False + + class _Transport: + policy = _Policy() + model = "openrouter/deepseek/deepseek-v4-flash" + last_finish_reason = "" + + # Only the direct DeepSeek API omits terminal finish_reason / returns + # empty completions in the observed way; third-party-hosted + # deepseek-* models must not pay duplicate-generation retries. + assert _should_retry_truncated_stream(_Transport(), stream=True, stopped_early=False) is False + assert _should_retry_empty_completion(_Transport(), " ", stopped_early=False) is False + + +def test_deepseek_retry_predicates_apply_to_direct_deepseek_api() -> None: + from models import _should_retry_empty_completion, _should_retry_truncated_stream + + class _Policy: + using_responses = False + + class _Transport: + policy = _Policy() + model = "deepseek/deepseek-v4-flash" + last_finish_reason = "" + + assert _should_retry_truncated_stream(_Transport(), stream=True, stopped_early=False) is True + assert _should_retry_empty_completion(_Transport(), " ", stopped_early=False) is True + + +def test_memorize_lock_rebinds_to_new_event_loop() -> None: + async def _acquire_once() -> None: + lock = get_memorize_lock() + await lock.acquire() + lock.release() + + asyncio.run(_acquire_once()) + first_lock = get_memorize_lock() + + # A new event loop (e.g. after EventLoopThread.terminate()) must get a + # usable lock instead of one bound to the dead loop. + asyncio.run(_acquire_once()) + second_lock = get_memorize_lock() + + assert first_lock is not second_lock + + +# --- failure diagnostics ----------------------------------------------------- + + +def test_explain_tool_request_failure_flags_truncated_envelope() -> None: + truncated = ( + '{"thoughts":["apply the patch"],"headline":"Applying patch",' + '"tool_name":"code_execution_tool","tool_args":{"code":"echo' + ) + + reason = explain_tool_request_failure(truncated) + assert "truncated" in reason.lower() + assert str(len(truncated)) in reason + + +def test_explain_tool_request_failure_flags_plain_prose() -> None: + reason = explain_tool_request_failure("Let me plan the next steps first.") + + assert "no json" in reason.lower() or "prose" in reason.lower() + + +def test_explain_tool_request_failure_flags_empty_response() -> None: + assert "empty" in explain_tool_request_failure("").lower() + assert "empty" in explain_tool_request_failure(" \n ").lower() + + +def test_explain_tool_request_failure_flags_non_tool_json() -> None: + reason = explain_tool_request_failure('{"status":"planning"}') + + assert "tool_name" in reason or "tool_args" in reason + + +# --- finish_reason propagation ----------------------------------------------- + + +def test_chat_completions_parse_captures_finish_reason() -> None: + chunk = { + "choices": [ + {"message": {"content": "partial"}, "finish_reason": "length"} + ] + } + + parsed = ChatCompletionsTransport.parse(chunk) + + assert parsed["response_delta"] == "partial" + assert parsed["finish_reason"] == "length" + + +def test_chat_completions_stream_parser_remembers_finish_reason() -> None: + parser = ChatCompletionsStreamParser() + parser.parse( + {"choices": [{"delta": {"content": "part"}, "finish_reason": None}]} + ) + parser.parse({"choices": [{"delta": {}, "finish_reason": "length"}]}) + + assert parser.finish_reason == "length" + + +def test_llm_result_from_chat_carries_finish_reason_roundtrip() -> None: + result = LLMResult.from_chat(response="text", finish_reason="length") + + assert result.finish_reason == "length" + assert result.metadata()["responses"]["finish_reason"] == "length" + + restored = LLMResult.from_dict(result.to_dict()) + assert restored.finish_reason == "length" + + +def test_llm_result_finish_reason_defaults_empty() -> None: + assert LLMResult.from_chat(response="text").finish_reason == "" + assert LLMResult.from_dict({}).finish_reason == "" + + +# --- memory job serialization -------------------------------------------------- + + +def test_memorize_lock_is_shared_and_serializes_jobs() -> None: + assert get_memorize_lock() is get_memorize_lock() + + events: list[str] = [] + + async def job(name: str) -> None: + async with get_memorize_lock(): + events.append(f"start:{name}") + await asyncio.sleep(0.01) + events.append(f"end:{name}") + + async def main() -> None: + await asyncio.gather(job("fragments"), job("solutions")) + + asyncio.run(main()) + + # no interleaving: each job must fully finish before the next starts + assert events in ( + ["start:fragments", "end:fragments", "start:solutions", "end:solutions"], + ["start:solutions", "end:solutions", "start:fragments", "end:fragments"], + ) + + +# --- agent-level wiring -------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_process_tools_recovers_embedded_tool_request(monkeypatch) -> None: + from types import SimpleNamespace + + import agent as agent_module + from agent import Agent, LoopData + from helpers import mcp_handler + from helpers.tool import Response + + class DummyMCPConfig: + def get_tool(self, agent, tool_name): + return None + + class DummyTool: + def __init__(self): + self.args = {} + + async def before_execution(self, **kwargs): + return None + + async def execute(self, **kwargs): + return Response(message=f"ran:{self.args['code']}", break_loop=True) + + async def after_execution(self, response): + return None + + async def no_extension(*args, **kwargs): + return None + + async def no_intervention(*args, **kwargs): + return None + + monkeypatch.setattr( + mcp_handler.MCPConfig, "get_instance", lambda: DummyMCPConfig() + ) + monkeypatch.setattr(agent_module.extension, "call_extensions_async", no_extension) + + tool = DummyTool() + agent = object.__new__(Agent) + agent.data = {} + agent.loop_data = LoopData() + agent.handle_intervention = no_intervention + agent.get_tool = lambda **kwargs: tool + warnings: list[str] = [] + agent.agent_name = "A0" + agent.read_prompt = lambda name, **kw: name + agent.hist_add_warning = lambda msg: ( + warnings.append(msg), + SimpleNamespace(id=None), + )[1] + agent.context = SimpleNamespace( + log=SimpleNamespace(log=lambda **entry: None) + ) + + content = ( + "I'll first verify whether the patch helper survived, then apply the fix.\n\n" + + TOOL_REQUEST + ) + + assert await Agent.process_tools(agent, content) == "ran:ls" + assert tool.args == {"code": "ls"} + # successful recovery adds the corrective note teaching bare-JSON output + assert warnings == ["fw.msg_recovered_request.md"] + + +@pytest.mark.asyncio +async def test_process_llm_result_tools_forwards_finish_reason(monkeypatch) -> None: + from agent import Agent + + agent = object.__new__(Agent) + captured: list[dict] = [] + + async def log_builtin_items(result): + return None + + async def process_tools(message, **kwargs): + captured.append({"message": message, **kwargs}) + return None + + agent._log_response_builtin_items = log_builtin_items + agent.process_tools = process_tools + + truncated = '{"thoughts":["apply patch"],"tool_name":"code_execution' + result = LLMResult.from_chat(response=truncated, finish_reason="length") + + assert await Agent.process_llm_result_tools(agent, result) is None + assert captured == [{"message": truncated, "finish_reason": "length"}] + + +@pytest.mark.asyncio +async def test_process_tools_logs_sanitized_failure_reason(monkeypatch) -> None: + from types import SimpleNamespace + + import agent as agent_module + from agent import Agent, LoopData + + async def no_extension(*args, **kwargs): + return None + + async def no_intervention(*args, **kwargs): + return None + + monkeypatch.setattr(agent_module.extension, "call_extensions_async", no_extension) + + logged: list[dict] = [] + agent = object.__new__(Agent) + agent.data = {} + agent.loop_data = LoopData() + agent.agent_name = "A0" + agent.handle_intervention = no_intervention + agent.read_prompt = lambda name, **kw: "misformatted" + agent.hist_add_warning = lambda msg: SimpleNamespace(id=None) + agent.context = SimpleNamespace( + log=SimpleNamespace(log=lambda **entry: logged.append(entry)) + ) + + truncated = ( + '{"thoughts":["apply the patch"],"headline":"Applying patch",' + '"tool_name":"code_execution_tool","tool_args":{"code":"echo' + ) + + await Agent.process_tools(agent, truncated, finish_reason="length") + + warnings = [e for e in logged if e.get("type") == "warning"] + assert warnings, "expected a misformat warning log entry" + content = warnings[-1]["content"] + assert "misformat" in content.lower() + assert "truncated" in content.lower() + assert "finish_reason=length" in content + + +# --- stream integrity: provider-dropped connections ---------------------------- + + +class _FakeAsyncStream: + def __init__(self, chunks): + self._chunks = chunks + self._index = 0 + + def __aiter__(self): + return self + + async def __anext__(self): + if self._index >= len(self._chunks): + raise StopAsyncIteration + chunk = self._chunks[self._index] + self._index += 1 + return chunk + + +def _chat_transport(): + from helpers import litellm_transport as lt + + return lt.LiteLLMTransport( + model="deepseek/deepseek-v4-flash", + messages=[{"role": "user", "content": "hi"}], + kwargs={"a0_api_mode": "chat"}, + ) + + +@pytest.mark.asyncio +async def test_transport_records_finish_reason_when_stream_completes(monkeypatch) -> None: + from helpers import litellm_transport as lt + + chunks = [ + {"choices": [{"delta": {"content": '{"tool_name":"response"}'}}]}, + {"choices": [{"delta": {}, "finish_reason": "stop"}]}, + ] + + async def fake_acompletion(**kwargs): + return _FakeAsyncStream(chunks) + + monkeypatch.setattr(lt, "acompletion", fake_acompletion) + transport = _chat_transport() + async for _ in transport.astream(): + pass + + assert transport.last_finish_reason == "stop" + + +@pytest.mark.asyncio +async def test_transport_records_missing_finish_reason_when_stream_drops(monkeypatch) -> None: + """DeepSeek intermittently closes the connection mid-stream; LiteLLM ends + the iterator without an error, so no terminal finish_reason ever arrives.""" + from helpers import litellm_transport as lt + + chunks = [ + {"choices": [{"delta": {"content": '{"thoughts":["partial"], "tool_na'}}]}, + ] + + async def fake_acompletion(**kwargs): + return _FakeAsyncStream(chunks) + + monkeypatch.setattr(lt, "acompletion", fake_acompletion) + transport = _chat_transport() + async for _ in transport.astream(): + pass + + assert transport.last_finish_reason == "" + + +def test_should_retry_truncated_stream_flags_dropped_deepseek_stream() -> None: + from types import SimpleNamespace + + from models import _should_retry_truncated_stream + + transport = SimpleNamespace( + policy=SimpleNamespace(using_responses=False), + model="deepseek/deepseek-v4-flash", + last_finish_reason="", + ) + assert _should_retry_truncated_stream(transport, stream=True, stopped_early=False) is True + + +def test_should_retry_truncated_stream_accepts_clean_stop() -> None: + from types import SimpleNamespace + + from models import _should_retry_truncated_stream + + for reason in ("stop", "length", "tool_calls"): + transport = SimpleNamespace( + policy=SimpleNamespace(using_responses=False), + model="deepseek/deepseek-v4-flash", + last_finish_reason=reason, + ) + assert _should_retry_truncated_stream(transport, stream=True, stopped_early=False) is False + + +def test_should_retry_truncated_stream_exempts_early_stop_and_non_deepseek() -> None: + from types import SimpleNamespace + + from models import _should_retry_truncated_stream + + dropped = SimpleNamespace( + policy=SimpleNamespace(using_responses=False), + model="deepseek/deepseek-v4-flash", + last_finish_reason="", + ) + # agent's own early-stop breaks the stream before finish_reason arrives + assert _should_retry_truncated_stream(dropped, stream=True, stopped_early=True) is False + # non-stream calls always carry a terminal state from the provider + assert _should_retry_truncated_stream(dropped, stream=False, stopped_early=False) is False + # other providers may legitimately omit finish_reason; do not change behavior + other = SimpleNamespace( + policy=SimpleNamespace(using_responses=False), + model="openai/gpt-5.4", + last_finish_reason="", + ) + assert _should_retry_truncated_stream(other, stream=True, stopped_early=False) is False + # responses-api transport has its own completion semantics + responses_transport = SimpleNamespace( + policy=SimpleNamespace(using_responses=True), + model="deepseek/deepseek-v4-flash", + last_finish_reason="", + ) + assert _should_retry_truncated_stream(responses_transport, stream=True, stopped_early=False) is False + + +@pytest.mark.asyncio +async def test_unified_turn_retries_dropped_deepseek_stream(monkeypatch) -> None: + """End-to-end: first stream dies mid-JSON (no finish_reason), the retry + delivers the complete envelope; the partial output must be discarded.""" + import models + from helpers import litellm_transport as lt + + partial = [ + {"choices": [{"delta": {"content": '{"thoughts":["apply patch"], "tool_na'}}]}, + ] + complete = [ + {"choices": [{"delta": {"content": '{"tool_name":"response",'}}]}, + {"choices": [{"delta": {"content": '"tool_args":{"text":"ok"}}'}}]}, + {"choices": [{"delta": {}, "finish_reason": "stop"}]}, + ] + streams = [_FakeAsyncStream(partial), _FakeAsyncStream(complete)] + calls = 0 + + async def fake_acompletion(**kwargs): + nonlocal calls + stream = streams[min(calls, len(streams) - 1)] + calls += 1 + return stream + + monkeypatch.setattr(lt, "acompletion", fake_acompletion) + monkeypatch.setattr(models, "configure_litellm", lambda: None) + + model = models.LiteLLMChatWrapper( + model="deepseek-v4-flash", provider="deepseek", + a0_api_mode="chat", a0_retry_delay_seconds=0, + ) + + streamed: list[str] = [] + + async def on_chunk(chunk: str, full: str): + streamed.append(chunk) + return None + + result = await model.unified_turn( + user_message="hi", response_callback=on_chunk, + ) + + assert calls == 2, "dropped stream must trigger exactly one retry" + assert result.response == '{"tool_name":"response","tool_args":{"text":"ok"}}' + assert result.finish_reason == "stop" + # the partial first attempt must not leak into the final response + assert "apply patch" not in result.response + + +@pytest.mark.asyncio +async def test_unified_turn_accepts_partial_after_retries_exhausted(monkeypatch) -> None: + """If every attempt is dropped, fall back to the last partial response + (previous behavior) rather than failing the turn.""" + import models + from helpers import litellm_transport as lt + + partial = [ + {"choices": [{"delta": {"content": '{"thoughts":["still cut"], "tool_na'}}]}, + ] + + async def fake_acompletion(**kwargs): + return _FakeAsyncStream(partial) + + monkeypatch.setattr(lt, "acompletion", fake_acompletion) + monkeypatch.setattr(models, "configure_litellm", lambda: None) + + model = models.LiteLLMChatWrapper( + model="deepseek-v4-flash", provider="deepseek", + a0_api_mode="chat", a0_retry_delay_seconds=0, + ) + + async def on_chunk(chunk: str, full: str): + return None + + result = await model.unified_turn(user_message="hi", response_callback=on_chunk) + + assert result.response == '{"thoughts":["still cut"], "tool_na' + assert result.finish_reason == "" + + +def test_truncated_json_body_is_classified_transient_despite_http_200() -> None: + """DeepSeek sometimes closes a non-streaming response mid-body; LiteLLM + raises 'Unable to get json response' with the original 200 status code, + which must still count as transient so the call is retried.""" + from models import _is_transient_litellm_error + + class FakeProviderError(Exception): + def __init__(self): + super().__init__( + "litellm.APIError: APIError: DeepseekException - " + "Unable to get json response - Unterminated string starting " + "at: line 1 column 2001 (char 2000), Original Response: {..." + ) + self.message = str(self) + self.status_code = 200 + + assert _is_transient_litellm_error(FakeProviderError()) is True + + +# --- empty completions (DeepSeek JSON/thinking mode) --------------------------- + + +def test_should_retry_empty_completion_flags_whitespace_stop() -> None: + """Reproduces seq 1082: full reasoning stream, finish_reason=stop, and a + message body of 45 spaces — the provider completed but returned nothing.""" + from types import SimpleNamespace + + from models import _should_retry_empty_completion + + transport = SimpleNamespace( + policy=SimpleNamespace(using_responses=False), + model="deepseek/deepseek-v4-flash", + ) + assert _should_retry_empty_completion(transport, " " * 45, stopped_early=False) is True + assert _should_retry_empty_completion(transport, "", stopped_early=False) is True + + +def test_should_retry_empty_completion_accepts_real_content() -> None: + from types import SimpleNamespace + + from models import _should_retry_empty_completion + + transport = SimpleNamespace( + policy=SimpleNamespace(using_responses=False), + model="deepseek/deepseek-v4-flash", + ) + assert _should_retry_empty_completion(transport, '{"tool_name":"x"}', stopped_early=False) is False + assert _should_retry_empty_completion(transport, "", stopped_early=True) is False + + other = SimpleNamespace( + policy=SimpleNamespace(using_responses=False), + model="openai/gpt-5.4", + ) + assert _should_retry_empty_completion(other, "", stopped_early=False) is False + + responses_transport = SimpleNamespace( + policy=SimpleNamespace(using_responses=True), + model="deepseek/deepseek-v4-flash", + ) + assert _should_retry_empty_completion(responses_transport, "", stopped_early=False) is False + + +@pytest.mark.asyncio +async def test_unified_turn_retries_empty_completion_with_reasoning(monkeypatch) -> None: + """Mirror of the production failure: reasoning streams fully, content is + whitespace, finish_reason=stop; the retry must deliver the real envelope.""" + import models + from helpers import litellm_transport as lt + + empty_completion = [ + {"choices": [{"delta": {"reasoning_content": "Let me plan the patch carefully."}}]}, + {"choices": [{"delta": {"content": " " * 45}}]}, + {"choices": [{"delta": {}, "finish_reason": "stop"}]}, + ] + complete = [ + {"choices": [{"delta": {"content": '{"tool_name":"response","tool_args":{"text":"ok"}}'}}]}, + {"choices": [{"delta": {}, "finish_reason": "stop"}]}, + ] + streams = [_FakeAsyncStream(empty_completion), _FakeAsyncStream(complete)] + calls = 0 + + async def fake_acompletion(**kwargs): + nonlocal calls + stream = streams[min(calls, len(streams) - 1)] + calls += 1 + return stream + + monkeypatch.setattr(lt, "acompletion", fake_acompletion) + monkeypatch.setattr(models, "configure_litellm", lambda: None) + + model = models.LiteLLMChatWrapper( + model="deepseek-v4-flash", provider="deepseek", + a0_api_mode="chat", a0_retry_delay_seconds=0, + ) + + async def on_chunk(chunk: str, full: str): + return None + + result = await model.unified_turn(user_message="hi", response_callback=on_chunk) + + assert calls == 2, "whitespace-only completion must trigger exactly one retry" + assert result.response == '{"tool_name":"response","tool_args":{"text":"ok"}}' + assert result.finish_reason == "stop" + + +@pytest.mark.asyncio +async def test_unified_call_allow_empty_completion_skips_retry(monkeypatch) -> None: + """Utility regime: with a0_allow_empty_completion an empty DeepSeek + completion is a benign answer (e.g. 'nothing to memorize'), not a + provider failure - no retry must fire.""" + import models + from helpers import litellm_transport as lt + + empty_completion = [ + {"choices": [{"delta": {"content": " "}}]}, + {"choices": [{"delta": {}, "finish_reason": "stop"}]}, + ] + calls = 0 + + async def fake_acompletion(**kwargs): + nonlocal calls + calls += 1 + return _FakeAsyncStream(empty_completion) + + monkeypatch.setattr(lt, "acompletion", fake_acompletion) + monkeypatch.setattr(models, "configure_litellm", lambda: None) + + model = models.LiteLLMChatWrapper( + model="deepseek-v4-flash", provider="deepseek", + a0_api_mode="chat", a0_retry_delay_seconds=0, + ) + + async def on_chunk(chunk: str, full: str): + return None + + response, _reasoning = await model.unified_call( + user_message="hi", response_callback=on_chunk, + a0_allow_empty_completion=True, + ) + + assert calls == 1, "utility-style call must not retry an empty completion" + assert not response.strip() + + +@pytest.mark.asyncio +async def test_unified_call_retries_empty_completion_by_default(monkeypatch) -> None: + """Main-turn regime unchanged: without the opt-out flag an empty DeepSeek + completion is retried and the retry's content is returned.""" + import models + from helpers import litellm_transport as lt + + empty_completion = [ + {"choices": [{"delta": {"content": " "}}]}, + {"choices": [{"delta": {}, "finish_reason": "stop"}]}, + ] + complete = [ + {"choices": [{"delta": {"content": '{"tool_name":"response","tool_args":{"text":"ok"}}'}}]}, + {"choices": [{"delta": {}, "finish_reason": "stop"}]}, + ] + streams = [_FakeAsyncStream(empty_completion), _FakeAsyncStream(complete)] + calls = 0 + + async def fake_acompletion(**kwargs): + nonlocal calls + stream = streams[min(calls, len(streams) - 1)] + calls += 1 + return stream + + monkeypatch.setattr(lt, "acompletion", fake_acompletion) + monkeypatch.setattr(models, "configure_litellm", lambda: None) + + model = models.LiteLLMChatWrapper( + model="deepseek-v4-flash", provider="deepseek", + a0_api_mode="chat", a0_retry_delay_seconds=0, + ) + + async def on_chunk(chunk: str, full: str): + return None + + response, _reasoning = await model.unified_call( + user_message="hi", response_callback=on_chunk, + ) + + assert calls == 2, "empty completion must trigger exactly one retry by default" + assert response == '{"tool_name":"response","tool_args":{"text":"ok"}}' + + +@pytest.mark.asyncio +async def test_call_utility_model_opts_out_of_empty_completion_retry(monkeypatch) -> None: + """Wiring: Agent.call_utility_model passes a0_allow_empty_completion so a + benign empty utility reply (e.g. 'nothing to memorize') is not retried.""" + import agent as agent_module + from agent import Agent + + async def no_extension(*args, **kwargs): + return None + + monkeypatch.setattr(agent_module.extension, "call_extensions_async", no_extension) + + captured: dict = {} + + class _StubModel: + async def unified_call(self, **kwargs): + captured.update(kwargs) + return "", "" + + instance = object.__new__(Agent) + instance.get_utility_model = lambda: _StubModel() + + result = await Agent.call_utility_model.__wrapped__( + instance, system="sys", message="msg", background=True + ) + + assert result == "" + assert captured.get("a0_allow_empty_completion") is True + + +@pytest.mark.asyncio +async def test_process_llm_result_tools_forwards_truncated_reasoning(monkeypatch) -> None: + from agent import Agent + + agent = object.__new__(Agent) + captured: list[dict] = [] + + async def log_builtin_items(result): + return None + + async def process_tools(message, **kwargs): + captured.append({"message": message, **kwargs}) + return None + + agent._log_response_builtin_items = log_builtin_items + agent.process_tools = process_tools + + truncated = ( + '{"thoughts":["apply the patch"],"headline":"Applying patch",' + '"tool_name":"code_execution_tool","tool_args":{"code":"echo' + ) + result = LLMResult(response="", reasoning=truncated) + + assert await Agent.process_llm_result_tools(agent, result) is None + assert captured == [{"message": truncated, "finish_reason": ""}] + + +@pytest.mark.asyncio +async def test_process_tools_uses_truncated_request_warning_for_truncated_json( + monkeypatch, +) -> None: + from types import SimpleNamespace + + import agent as agent_module + from agent import Agent, LoopData + + async def no_extension(*args, **kwargs): + return None + + async def no_intervention(*args, **kwargs): + return None + + monkeypatch.setattr(agent_module.extension, "call_extensions_async", no_extension) + + warnings: list[str] = [] + logged: list[dict] = [] + agent = object.__new__(Agent) + agent.data = {} + agent.loop_data = LoopData() + agent.agent_name = "A0" + agent.handle_intervention = no_intervention + agent.read_prompt = lambda name, **kw: ( + "TRUNCATED_REPROMPT" + if name == "fw.msg_truncated_request.md" + else "misformatted" + ) + agent.hist_add_warning = lambda msg: (warnings.append(msg), SimpleNamespace(id=None))[1] + agent.context = SimpleNamespace( + log=SimpleNamespace(log=lambda **entry: logged.append(entry)) + ) + + truncated = ( + '{"thoughts":["apply the patch"],"headline":"Applying patch",' + '"tool_name":"code_execution_tool","tool_args":{"code":"echo' + ) + + await Agent.process_tools(agent, truncated) + + assert warnings == ["TRUNCATED_REPROMPT"] + warning_entries = [e for e in logged if e.get("type") == "warning"] + assert warning_entries + assert "truncated or unterminated JSON tool request" in warning_entries[-1]["content"] + + +@pytest.mark.asyncio +async def test_process_tools_keeps_misformat_warning_for_balanced_invalid_json( + monkeypatch, +) -> None: + from types import SimpleNamespace + + import agent as agent_module + from agent import Agent, LoopData + + async def no_extension(*args, **kwargs): + return None + + async def no_intervention(*args, **kwargs): + return None + + monkeypatch.setattr(agent_module.extension, "call_extensions_async", no_extension) + + warnings: list[str] = [] + agent = object.__new__(Agent) + agent.data = {} + agent.loop_data = LoopData() + agent.agent_name = "A0" + agent.handle_intervention = no_intervention + agent.read_prompt = lambda name, **kw: ( + "TRUNCATED_REPROMPT" + if name == "fw.msg_truncated_request.md" + else "MISFORMAT_REPROMPT" + ) + agent.hist_add_warning = lambda msg: (warnings.append(msg), SimpleNamespace(id=None))[1] + agent.context = SimpleNamespace( + log=SimpleNamespace(log=lambda **entry: None) + ) + + balanced_invalid = '{"status":"planning","note":"no tool envelope here"}' + + await Agent.process_tools(agent, balanced_invalid) + + assert warnings == ["MISFORMAT_REPROMPT"] + + +@pytest.mark.asyncio +async def test_process_tools_refuses_deliberating_response_envelope(monkeypatch) -> None: + """Adversarial case from review: hedged deliberation containing a + 'response' envelope must NOT complete the task; the message takes the + standard misformat warning path instead.""" + from types import SimpleNamespace + + import agent as agent_module + from agent import Agent, LoopData + + async def no_extension(*args, **kwargs): + return None + + async def no_intervention(*args, **kwargs): + return None + + monkeypatch.setattr(agent_module.extension, "call_extensions_async", no_extension) + + warnings: list[str] = [] + agent = object.__new__(Agent) + agent.data = {} + agent.loop_data = LoopData() + agent.agent_name = "A0" + agent.handle_intervention = no_intervention + agent.read_prompt = lambda name, **kw: ( + "TRUNCATED_REPROMPT" + if name == "fw.msg_truncated_request.md" + else "MISFORMAT_REPROMPT" + ) + agent.hist_add_warning = lambda msg: (warnings.append(msg), SimpleNamespace(id=None))[1] + agent.context = SimpleNamespace( + log=SimpleNamespace(log=lambda **entry: None) + ) + + content = ( + 'I could reply {"thoughts":["x"],"tool_name":"response",' + '"tool_args":{"text":"hi"}} but let me think' + ) + + result = await Agent.process_tools(agent, content) + + assert result is None # task NOT completed by embedded prose + assert warnings == ["MISFORMAT_REPROMPT"] + + +@pytest.mark.asyncio +async def test_process_tools_adds_corrective_warning_after_recovery(monkeypatch) -> None: + """Successful embedded-envelope recovery adds the + fw.msg_recovered_request.md corrective note so the model learns to emit + bare JSON next time.""" + from types import SimpleNamespace + + import agent as agent_module + from agent import Agent, LoopData + + async def no_extension(*args, **kwargs): + return None + + async def no_intervention(*args, **kwargs): + return None + + monkeypatch.setattr(agent_module.extension, "call_extensions_async", no_extension) + + warnings: list[str] = [] + agent = object.__new__(Agent) + agent.data = {} + agent.loop_data = LoopData() + agent.agent_name = "A0" + agent.handle_intervention = no_intervention + agent.read_prompt = lambda name, **kw: name + agent.hist_add_warning = lambda msg: (warnings.append(msg), SimpleNamespace(id=None))[1] + agent.context = SimpleNamespace( + log=SimpleNamespace(log=lambda **entry: None) + ) + # stop before execution; the corrective note is added before tool lookup + agent.get_tool = lambda **kw: None + + msg = "Let me check the current state first.\n\n" + TOOL_REQUEST + + await Agent.process_tools(agent, msg) + + assert warnings[0] == "fw.msg_recovered_request.md" + + +# --- unified failure classification (arc-10) ---------------------------------- + + +def test_classify_tool_request_failure_categories() -> None: + assert classify_tool_request_failure("") == "empty" + assert classify_tool_request_failure(" \n ") == "empty" + assert classify_tool_request_failure(None) == "empty" + assert classify_tool_request_failure("Let me plan the next steps first.") == "prose" + assert classify_tool_request_failure('{"status":"planning"}') == "invalid_envelope" + truncated = ( + '{"thoughts":["apply the patch"],"headline":"Applying patch",' + '"tool_name":"code_execution_tool","tool_args":{"code":"echo' + ) + assert classify_tool_request_failure(truncated) == "truncated" + + +def test_classify_tool_request_failure_matches_explain_for_divergent_shapes() -> None: + """Regression for the live 2026-08-04 divergence (chat qZE6fC0g): the log + reason said "truncated or unterminated JSON tool request" while routing + took the generic misformat branch because is_truncated_tool_request() + gated on the raw content starting with "{". One classifier must drive + both the log reason and the retry-prompt routing.""" + divergent = [ + # ```json-fenced payload cut mid-string (21702 chars live) + "```json\n" + '{"thoughts":["write patch"],"headline":"Writing patch",' + '"tool_name":"text_editor_remote","tool_args":{"content":"def x():', + # prose prefix + truncated envelope + "Writing the patch script now.\n" + '{"thoughts":["x"],"tool_name":"text_editor_remote",' + '"tool_args":{"content":"y', + # complete non-tool root + truncated tail + '{"status":"ready"}\n' + '{"thoughts":["x"],"tool_name":"code_execution_tool",' + '"tool_args":{"code":"echo', + ] + for payload in divergent: + assert classify_tool_request_failure(payload) == "truncated" + assert "truncated or unterminated" in explain_tool_request_failure(payload) + assert is_truncated_tool_request(payload) is True + + +def test_classify_tool_request_failure_keeps_invalid_envelope_off_truncated_path() -> None: + # Balanced but envelope-less JSON is a format problem, not truncation; + # it must keep the generic misformat reprompt. + assert classify_tool_request_failure( + '{"status":"planning","note":"no tool envelope here"}' + ) == "invalid_envelope" + assert "without a valid tool_name/tool_args envelope" in explain_tool_request_failure( + '{"status":"planning","note":"no tool envelope here"}' + ) + + +def test_truncated_request_prompt_includes_size_guidance() -> None: + """The reprompt must tell the model to split oversized payloads; after a + generic nudge the live model retried with an even larger request + (21702 -> 31446 chars).""" + prompt = (PROJECT_ROOT / "prompts" / "fw.msg_truncated_request.md").read_text() + assert "split" in prompt.lower() + assert "smaller" in prompt.lower() + + +@pytest.mark.asyncio +async def test_process_tools_uses_truncated_request_warning_for_fenced_truncation( + monkeypatch, +) -> None: + """Agent-level regression for the divergent routing: a fenced truncated + payload must get fw.msg_truncated_request.md, not the generic misformat + reprompt.""" + from types import SimpleNamespace + + import agent as agent_module + from agent import Agent, LoopData + + async def no_extension(*args, **kwargs): + return None + + async def no_intervention(*args, **kwargs): + return None + + monkeypatch.setattr(agent_module.extension, "call_extensions_async", no_extension) + + warnings: list[str] = [] + logged: list[dict] = [] + agent = object.__new__(Agent) + agent.data = {} + agent.loop_data = LoopData() + agent.agent_name = "A0" + agent.handle_intervention = no_intervention + agent.read_prompt = lambda name, **kw: ( + "TRUNCATED_REPROMPT" + if name == "fw.msg_truncated_request.md" + else "MISFORMAT_REPROMPT" + ) + agent.hist_add_warning = lambda msg: (warnings.append(msg), SimpleNamespace(id=None))[1] + agent.context = SimpleNamespace( + log=SimpleNamespace(log=lambda **entry: logged.append(entry)) + ) + + fenced = ( + "```json\n" + '{"thoughts":["write patch"],"headline":"Writing patch",' + '"tool_name":"text_editor_remote","tool_args":{"content":"def apply():' + ) + + await Agent.process_tools(agent, fenced) + + assert warnings == ["TRUNCATED_REPROMPT"] + warning_entries = [e for e in logged if e.get("type") == "warning"] + assert warning_entries + assert "truncated or unterminated JSON tool request" in warning_entries[-1]["content"] diff --git a/tests/test_defer_lifecycle.py b/tests/test_defer_lifecycle.py index 879029aaf4..340029d6e1 100644 --- a/tests/test_defer_lifecycle.py +++ b/tests/test_defer_lifecycle.py @@ -78,13 +78,21 @@ async def run(captured_owner): assert task.kwargs == {} del owner + # While the coroutine is still running (blocked on release), it holds + # its own reference to the owner - kill() must NOT clear the running + # call's arguments, only the stored snapshot (asserted above). assert owner_ref() is not None task.event_loop_thread.loop.call_soon_threadsafe(release[0].set) assert finished.wait(2) - asyncio.run_coroutine_threadsafe( - asyncio.sleep(0), task.event_loop_thread.loop - ).result(2) - assert owner_ref() is None + # NOTE: no "owner_ref() is None" assertion here on purpose. After a + # suppressed CancelledError, asyncio keeps the exception (whose + # traceback references the finished coroutine's frame, and thus the + # owner) anchored in interpreter state whose lifetime depends on + # process-wide import/GC conditions - importing helpers.settings or + # anything that pulls it in makes the island survive pumping and even + # gc.collect(). That is CPython/asyncio implementation detail, not a + # DeferredTask guarantee; the product contract (stored call cleared, + # running arguments untouched) is covered by the assertions above. finally: if release and task.event_loop_thread.loop: task.event_loop_thread.loop.call_soon_threadsafe(release[0].set) diff --git a/tests/test_docker_release_plan.py b/tests/test_docker_release_plan.py index c0fa6a589d..721e1a3ae3 100644 --- a/tests/test_docker_release_plan.py +++ b/tests/test_docker_release_plan.py @@ -44,12 +44,16 @@ def test_docker_publish_workflow_tracks_branch_promotions(): workflow_path = PROJECT_ROOT / ".github" / "workflows" / "docker-publish.yml" content = workflow_path.read_text(encoding="utf-8") - assert 'branches:\n - "testing"\n - "main"' in content + assert 'branches:\n - "testing"\n - "ready"\n - "main"' in content assert 'tags:\n - "v*"' in content assert "workflow_dispatch:" in content assert "inputs:" in content assert "tag:" in content - assert 'ref: ${{ matrix.source_tag }}' in content + # The build job no longer checks out the tag ref directly; it re-resolves + # the source tag through the plan script's TARGET_TAG env var instead, + # in both the resolve-build and resolve-release steps. + assert content.count("TARGET_TAG: ${{ matrix.source_tag }}") == 2 + assert 'ALLOWED_BRANCHES: "testing ready main"' in content assert "SOURCE_REF_TYPE: ${{ github.ref_type }}" in content assert "BEFORE_SHA: ${{ github.event_name == 'push' && github.event.before || '' }}" in content diff --git a/tests/test_http_auth_csrf.py b/tests/test_http_auth_csrf.py index 0f95f449eb..20122e6fb5 100644 --- a/tests/test_http_auth_csrf.py +++ b/tests/test_http_auth_csrf.py @@ -1,6 +1,7 @@ from __future__ import annotations from flask import Flask, Response +from urllib.parse import parse_qs, urlsplit import pytest @@ -15,6 +16,13 @@ def _make_app() -> Flask: def login_handler(): return Response("login", status=200) + # helpers.api.get_current_request_next_url() falls back to + # url_for("serve_index"), so the harness app must register that endpoint + # just like the real UI server does. + @app.get("/") + def serve_index(): + return Response("index", status=200) + return app @@ -169,8 +177,16 @@ async def voqualizer_page(): response = client.get("/plugins/a0_voqualizer/webui/voqualizer.html?context=rlO1iMV7") assert response.status_code == 302 location = response.headers["Location"] - assert location.startswith("/login?next=") - assert "%2Fplugins%2Fa0_voqualizer%2Fwebui%2Fvoqualizer.html%3Fcontext%3DrlO1iMV7" in location + # Parse instead of matching a raw percent-encoded string: Werkzeug's + # url_for no longer encodes "/" and "?" inside query values, and both + # encodings carry the same semantics. + parsed = urlsplit(location) + assert not parsed.scheme and not parsed.netloc # never an external origin + assert parsed.path == "/login" + next_params = parse_qs(parsed.query) + assert next_params["next"] == [ + "/plugins/a0_voqualizer/webui/voqualizer.html?context=rlO1iMV7" + ] def test_is_safe_next_url_rejects_backslash_open_redirects() -> None: diff --git a/tests/test_model_config_project_presets.py b/tests/test_model_config_project_presets.py index bb8c670454..bf4799c2b4 100644 --- a/tests/test_model_config_project_presets.py +++ b/tests/test_model_config_project_presets.py @@ -64,6 +64,18 @@ def _clear_runtime_caches(): modules.purge_namespace("usr.plugins") +@pytest.fixture(autouse=True) +def _restore_runtime_caches_after_test(): + # Tests in this module point helpers.files._base_dir at a tmp dir, so + # extension/plugin scans during the test repopulate the runtime caches + # with tmp-dir results. monkeypatch restores _base_dir at teardown but + # leaves the poisoned caches behind, which breaks subsequently collected + # modules (e.g. test_default_prompt_budget, test_browser_agent_regressions) + # whenever this module runs in the same process. + yield + _clear_runtime_caches() + + def _prepare_a0_tree(monkeypatch, tmp_path: Path): from helpers import files, plugins diff --git a/tests/test_responses_architecture.py b/tests/test_responses_architecture.py index 1a195cb62d..a161a96e31 100644 --- a/tests/test_responses_architecture.py +++ b/tests/test_responses_architecture.py @@ -518,7 +518,7 @@ async def test_agent_routes_chat_retries_and_native_responses_text() -> None: async def log_builtin_items(result): return None - async def process_tools(message): + async def process_tools(message, **kwargs): processed.append(message) return None @@ -595,7 +595,7 @@ async def test_agent_routes_misformatted_tool_intent_to_repair() -> None: async def log_builtin_items(result): return None - async def process_tools(message): + async def process_tools(message, **kwargs): processed.append(message) return None diff --git a/tests/test_stream_tool_early_stop.py b/tests/test_stream_tool_early_stop.py index 10fd668732..6c74fd1e0c 100644 --- a/tests/test_stream_tool_early_stop.py +++ b/tests/test_stream_tool_early_stop.py @@ -1451,7 +1451,7 @@ def test_chat_completions_stream_parser_accumulates_tool_call_arguments(): } ] } - ) == {"reasoning_delta": "", "response_delta": ""} + ) == {"reasoning_delta": "", "response_delta": "", "finish_reason": ""} parsed = parser.parse( { "choices": [ @@ -1501,7 +1501,7 @@ def test_chat_completions_stream_parser_reads_dumped_tool_calls(): ) ] ) - ) == {"reasoning_delta": "", "response_delta": ""} + ) == {"reasoning_delta": "", "response_delta": "", "finish_reason": ""} parsed = parser.parse( _DumpOnly(choices=[_DumpOnly(delta=_DumpOnly(), finish_reason="tool_calls")]) diff --git a/tests/test_time_travel.py b/tests/test_time_travel.py index bd1d6552ab..abd96b4e53 100644 --- a/tests/test_time_travel.py +++ b/tests/test_time_travel.py @@ -24,6 +24,7 @@ _workspace_from_display, resolve_workspace, ) +from helpers import files as a0_files def run_git(repo_dir: Path, *args: str, check: bool = True) -> str: @@ -37,9 +38,13 @@ def run_git(repo_dir: Path, *args: str, check: bool = True) -> str: @pytest.fixture -def workspace(): +def workspace(tmp_path, monkeypatch): + # Redirect the framework base dir so /a0/usr display paths resolve into + # tmp_path instead of the live usr tree. Inside the deployed container + # the live tree is the persistent volume with real user data. + monkeypatch.setattr(a0_files, "_base_dir", str(tmp_path)) name = f"tt-{uuid.uuid4().hex}" - root = PROJECT_ROOT / "usr" / "time-travel-tests" / name + root = tmp_path / "usr" / "time-travel-tests" / name root.mkdir(parents=True) service = TimeTravelService(_workspace_from_display(f"/a0/usr/time-travel-tests/{name}")) try: @@ -169,9 +174,10 @@ def test_shadow_repo_empty_head_is_repaired_without_losing_history(workspace): ] -def test_workspace_identity_canonicalizes_symlink_aliases(): +def test_workspace_identity_canonicalizes_symlink_aliases(tmp_path, monkeypatch): + monkeypatch.setattr(a0_files, "_base_dir", str(tmp_path)) name = f"tt-{uuid.uuid4().hex}" - root = PROJECT_ROOT / "usr" / "time-travel-tests" / name + root = tmp_path / "usr" / "time-travel-tests" / name target = root / "target" alias = root / "alias" target.mkdir(parents=True) diff --git a/tests/test_timezone_regressions.py b/tests/test_timezone_regressions.py index 55dbcf52a4..c2d11faf86 100644 --- a/tests/test_timezone_regressions.py +++ b/tests/test_timezone_regressions.py @@ -147,6 +147,26 @@ def test_settings_auto_timezone_resolves_to_browser_timezone(isolated_localizati assert hooks[0]["kwargs"]["timezone"] == "Europe/Rome" +def test_settings_auto_timezone_does_not_persist_browser_timezone(isolated_localization, monkeypatch): + saved = isolated_localization + set_test_timezone("America/New_York") + base_settings = settings_module.get_default_settings() + auto_settings = {**base_settings, "timezone": settings_module.TIMEZONE_AUTO} + monkeypatch.setattr(settings_module, "_settings", auto_settings) + monkeypatch.setattr( + plugins_module, "call_plugin_hook", lambda *args, **kwargs: None + ) + saved.clear() + + settings_module._apply_timezone_setting( + auto_settings, + browser_timezone="Europe/Rome", + ) + + assert Localization.get().get_timezone() == "Europe/Rome" + assert saved == [] + + def test_settings_fixed_timezone_ignores_browser_timezone(isolated_localization, monkeypatch): set_test_timezone("Europe/Rome") base_settings = settings_module.get_default_settings() diff --git a/tests/test_tool_request_normalization.py b/tests/test_tool_request_normalization.py index f982211c58..cfc90ef297 100644 --- a/tests/test_tool_request_normalization.py +++ b/tests/test_tool_request_normalization.py @@ -12,6 +12,7 @@ from helpers.extract_tools import ( extract_tool_request, is_misformatted_tool_request, + is_truncated_tool_request, json_parse_dirty, normalize_tool_request, ) @@ -183,3 +184,90 @@ def test_parallel_prompt_encourages_mixed_independent_batches() -> None: assert "Do not split by tool type" in prompt assert "Never include `document_query`" in prompt assert "Call `response` only as a top-level tool" in prompt + + + +def test_is_truncated_tool_request_accepts_unterminated_envelope() -> None: + truncated = ( + '{"thoughts":["apply the patch"],"headline":"Applying patch",' + '"tool_name":"code_execution_tool","tool_args":{"code":"echo' + ) + assert is_truncated_tool_request(truncated) is True + + +def test_is_truncated_tool_request_rejects_balanced_json() -> None: + balanced = '{"tool_name":"response","tool_args":{"text":"ok"}}' + assert is_truncated_tool_request(balanced) is False + + +def test_is_truncated_tool_request_rejects_plain_prose() -> None: + assert is_truncated_tool_request("I will call a tool now") is False + assert is_truncated_tool_request("") is False + assert is_truncated_tool_request(None) is False + + +def test_is_truncated_tool_request_rejects_balanced_function_envelope() -> None: + """Responses-API function_call text is balanced and must not be treated as truncated.""" + balanced = '{"type":"function","name":"search_engine","arguments":{"query":"x"}}' + assert is_truncated_tool_request(balanced) is False + + +def test_is_truncated_tool_request_rejects_prose_with_stray_brace() -> None: + """Prose mentioning 'actions'/'tool' after a stray '{' is not a truncated + envelope; in responses mode such text must reach the user as a reply, + not be swallowed into a truncation repair loop.""" + assert is_truncated_tool_request("{here are the actions I plan to take") is False + assert is_truncated_tool_request("{ let me use the tool now") is False + + +def test_is_truncated_tool_request_rejects_extra_closing_brace() -> None: + assert is_truncated_tool_request('{"tool_name":"x"}}') is False + + +def test_is_truncated_tool_request_accepts_unterminated_function_payload() -> None: + """Responses-mode truncation: a function_call-style payload cut mid-body + must classify as truncated so it takes the repair-prompt path instead of + being shown to the user as a plain-text reply.""" + truncated = '{"type":"function","name":"search_engine","arguments":{"query":"x' + assert is_truncated_tool_request(truncated) is True + + +def test_is_truncated_tool_request_accepts_fenced_truncated_envelope() -> None: + """Live failure shape (DeepSeek V4 Flash, 2026-08-04): a ```json-fenced + tool request cut mid-string must classify as truncated; gating on the raw + content starting with "{" routed it to the generic misformat prompt + instead.""" + fenced = ( + "```json\n" + '{"thoughts":["write patch"],"headline":"Writing patch",' + '"tool_name":"text_editor_remote","tool_args":{"path":"/tmp/x.py",' + '"content":"import os\\n\\ndef main():' + ) + assert is_truncated_tool_request(fenced) is True + + +def test_is_truncated_tool_request_accepts_prose_prefixed_truncated_envelope() -> None: + payload = ( + "I'll write the patch script now.\n" + '{"thoughts":["write patch"],"tool_name":"text_editor_remote",' + '"tool_args":{"content":"def apply_patch():' + ) + assert is_truncated_tool_request(payload) is True + + +def test_is_truncated_tool_request_accepts_truncated_envelope_after_complete_root() -> None: + payload = ( + '{"status":"ready"}\n' + '{"thoughts":["go"],"tool_name":"code_execution_tool",' + '"tool_args":{"code":"echo' + ) + assert is_truncated_tool_request(payload) is True + + +def test_is_truncated_tool_request_rejects_balanced_fenced_envelope() -> None: + fenced = ( + '```json\n' + '{"thoughts":["x"],"tool_name":"response","tool_args":{"text":"ok"}}\n' + '```' + ) + assert is_truncated_tool_request(fenced) is False diff --git a/tests/test_unusable_response_loop.py b/tests/test_unusable_response_loop.py index c170f9d9b9..fb77e6654d 100644 --- a/tests/test_unusable_response_loop.py +++ b/tests/test_unusable_response_loop.py @@ -20,6 +20,7 @@ def log(self, **entry): def _agent(): prompts = { "fw.msg_misformat.md": "misformatted", + "fw.msg_truncated_request.md": "truncated_request", "fw.msg_repeat.md": "repeated", } @@ -97,3 +98,21 @@ def test_general_settings_expose_the_default_failure_limit(): assert "after 3 consecutive" in read_prompt_file( "fw.msg_unusable_response_limit.md", ["prompts"], limit=3 ) + + + +def test_truncated_request_warning_tracks_misformat_loop(monkeypatch): + monkeypatch.setattr( + response_loop, + "get_settings", + lambda: {"max_consecutive_unusable_responses": 2}, + ) + agent = _agent() + extension = response_loop.StopUnusableResponseLoop(agent=agent) + + assert _run(extension, agent, "truncated_request")["exception"] is None + agent.loop_data.iteration = 1 + data = _run(extension, agent, "truncated_request") + + assert isinstance(data["exception"], HandledException) + assert agent.loop_data.params_persistent[response_loop.STATE_KEY]["count"] == 2