[codex] fix dflash malformed final recovery

agent/anthropic_adapter.py

@@ -1783,11 +1783,25 @@ def _strip_orphaned_tool_blocks(result: List[Dict[str, Any]]) -> None:
                     tool_result_ids.add(block.get("tool_use_id"))
     for m in result:
         if m["role"] == "assistant" and isinstance(m["content"], list):
-            m["content"] = [
+            kept = [
                 b
                 for b in m["content"]
                 if b.get("type") != "tool_use" or b.get("id") in tool_result_ids
             ]
+            # If stripping an orphaned tool_use mutated a turn that also carries a
+            # signed thinking block, that block's Anthropic signature was computed
+            # against the ORIGINAL (un-stripped) turn content and is now invalid.
+            # Anthropic rejects the replayed turn with HTTP 400 "thinking blocks in
+            # the latest assistant message cannot be modified".  Flag the turn so
+            # _manage_thinking_signatures can demote the dead signature instead of
+            # replaying it verbatim.  See hermes-agent: extended-thinking + parallel
+            # tool batch interrupted mid-flight → non-retryable 400 crash-loop.
+            if len(kept) != len(m["content"]) and any(
+                isinstance(b, dict) and b.get("type") in {"thinking", "redacted_thinking"}
+                for b in m["content"]
+            ):
+                m["_thinking_signature_invalidated"] = True
+            m["content"] = kept
             if not m["content"]:
                 m["content"] = [{"type": "text", "text": "(tool call removed)"}]
 
@@ -1832,6 +1846,10 @@ def _merge_consecutive_roles(result: List[Dict[str, Any]]) -> List[Dict[str, Any
                     fixed[-1]["content"] = prev_content + curr_content
             else:
                 # Consecutive assistant messages — merge text content.
+                # Propagate the orphan-strip signature-invalidation flag onto the
+                # surviving (prev) dict so _manage_thinking_signatures still sees it.
+                if m.get("_thinking_signature_invalidated"):
+                    fixed[-1]["_thinking_signature_invalidated"] = True
                 # Drop thinking blocks from the *second* message: their
                 # signature was computed against a different turn boundary
                 # and becomes invalid once merged.
@@ -1920,11 +1938,26 @@ def _manage_thinking_signatures(
         else:
             # Latest assistant on direct Anthropic: keep signed, downgrade unsigned
             # to text so the reasoning isn't lost.
+            #
+            # Exception: if orphan-stripping (or another structural mutation) removed
+            # a tool_use block from THIS turn, every thinking signature on it was
+            # computed against the original turn content and is now dead.  Anthropic
+            # rejects the turn either way — replaying the signed block 400s with
+            # "thinking blocks in the latest assistant message cannot be modified",
+            # and a bare signed block with no following tool_use is also invalid.
+            # Demote ALL thinking blocks on this turn to text so the turn replays
+            # cleanly and the model can re-plan from the surviving tool results.
+            signature_dead = bool(m.get("_thinking_signature_invalidated"))
             new_content = []
             for b in m["content"]:
                 if not isinstance(b, dict) or b.get("type") not in _THINKING_TYPES:
                     new_content.append(b)
                     continue
+                if signature_dead:
+                    thinking_text = b.get("thinking", "")
+                    if thinking_text:
+                        new_content.append({"type": "text", "text": thinking_text})
+                    continue
                 if b.get("type") == "redacted_thinking":
                     # Redacted blocks use 'data' for the signature payload —
                     # drop the block when 'data' is missing (can't be validated).
@@ -1944,6 +1977,9 @@ def _manage_thinking_signatures(
             if isinstance(b, dict) and b.get("type") in _THINKING_TYPES:
                 b.pop("cache_control", None)
 
+        # Drop the internal bookkeeping flag — it must never reach the API payload.
+        m.pop("_thinking_signature_invalidated", None)
+
 
 def _evict_old_screenshots(result: List[Dict[str, Any]]) -> None:
     """Keep only the most recent ``_MAX_KEEP_IMAGES`` computer-use screenshots.

agent/background_review.py

@@ -67,6 +67,12 @@
     "from. Capture it.\n"
     "  • A skill that got loaded or consulted this session turned out "
     "to be wrong, missing a step, or outdated. Patch it NOW.\n\n"
+    "Evidence discipline: do not encode the foreground assistant's own "
+    "unverified diagnosis as a durable rule. For operational/debugging "
+    "lessons, prefer timestamp-matched tool output, logs, tests, or a user "
+    "correction. If the turn ended with an error sentinel, malformed final "
+    "response, interrupted stream, or retry/fallback exhaustion, capture the "
+    "recovery pattern only, not the failed hypothesis.\n\n"
     "Preference order — prefer the earliest action that fits, but do "
     "pick one when a signal above fired:\n"
     "  1. UPDATE A CURRENTLY-LOADED SKILL. Look back through the "
@@ -169,6 +175,12 @@
     "emerged.\n"
     "  • A skill that was loaded or consulted turned out wrong, "
     "missing, or outdated — patch it now.\n\n"
+    "Evidence discipline: do not encode the foreground assistant's own "
+    "unverified diagnosis as a durable rule. For operational/debugging "
+    "lessons, prefer timestamp-matched tool output, logs, tests, or a user "
+    "correction. If the turn ended with an error sentinel, malformed final "
+    "response, interrupted stream, or retry/fallback exhaustion, capture the "
+    "recovery pattern only, not the failed hypothesis.\n\n"
     "Preference order for skills — pick the earliest that fits:\n"
     "  1. UPDATE A CURRENTLY-LOADED SKILL. Check what skills were "
     "loaded via /skill-name or skill_view in the conversation. If one "

agent/conversation_compression.py

@@ -644,6 +644,12 @@ def try_shrink_image_parts_in_messages(api_messages: list) -> bool:
     # after a confirmed provider rejection, so the alternative is failure.
     target_bytes = 4 * 1024 * 1024
     changed_count = 0
+    # Track parts that are over the target but could NOT be shrunk under it.
+    # If any survive, retrying is pointless — the same oversized payload will
+    # be re-sent and rejected again, wasting the single retry budget.  We only
+    # report success (caller retries) when every over-threshold image was
+    # actually brought under the target.
+    unshrinkable_oversized = 0
 
     def _shrink_data_url(url: str) -> Optional[str]:
         """Return a smaller data URL, or None if shrink can't help."""
@@ -710,17 +716,34 @@ def _shrink_data_url(url: str) -> Optional[str]:
                 if resized:
                     image_value["url"] = resized
                     changed_count += 1
+                elif isinstance(url, str) and url.startswith("data:") \
+                        and len(url) > target_bytes:
+                    unshrinkable_oversized += 1
             elif isinstance(image_value, str):
                 resized = _shrink_data_url(image_value)
                 if resized:
                     part["image_url"] = resized
                     changed_count += 1
+                elif image_value.startswith("data:") \
+                        and len(image_value) > target_bytes:
+                    unshrinkable_oversized += 1
 
     if changed_count:
         logger.info(
             "image-shrink recovery: re-encoded %d image part(s) to fit under %.0f MB",
             changed_count, target_bytes / (1024 * 1024),
         )
+    if unshrinkable_oversized:
+        # At least one oversized image could not be shrunk under the target.
+        # Retrying would re-send it and fail identically, so signal "no
+        # progress" even if other parts shrank — the caller will surface the
+        # original error rather than burning its single retry on a no-op.
+        logger.warning(
+            "image-shrink recovery: %d oversized image part(s) could not be "
+            "shrunk under %.0f MB — not retrying (would re-send rejected payload)",
+            unshrinkable_oversized, target_bytes / (1024 * 1024),
+        )
+        return False
     return changed_count > 0
 
 

agent/conversation_loop.py

@@ -444,6 +444,7 @@ def run_conversation(
     agent._incomplete_scratchpad_retries = 0
     agent._codex_incomplete_retries = 0
     agent._thinking_prefill_retries = 0
+    agent._malformed_final_retries = 0
     agent._post_tool_empty_retried = False
     agent._last_content_with_tools = None
     agent._last_content_tools_all_housekeeping = False
@@ -672,6 +673,7 @@ def run_conversation(
                 # context loss.
                 agent._empty_content_retries = 0
                 agent._thinking_prefill_retries = 0
+                agent._malformed_final_retries = 0
                 agent._last_content_with_tools = None
                 agent._last_content_tools_all_housekeeping = False
                 agent._mute_post_response = False
@@ -3817,6 +3819,7 @@ def _stop_spinner():
                 if _had_prefill:
                     agent._thinking_prefill_retries = 0
                     agent._empty_content_retries = 0
+                    agent._malformed_final_retries = 0
                 # Successful tool execution — reset the post-tool nudge
                 # flag so it can fire again if the model goes empty on
                 # a LATER tool round.
@@ -4209,6 +4212,76 @@ def _stop_spinner():
                 # Reset retry counter/signature on successful content
                 agent._empty_content_retries = 0
                 agent._thinking_prefill_retries = 0
+
+                malformed_final_reason = agent._detect_malformed_tool_final_response(
+                    final_response,
+                    finish_reason,
+                    messages,
+                )
+                if malformed_final_reason:
+                    agent._malformed_final_retries += 1
+                    logger.warning(
+                        "Malformed final response after tool calls (%s) — "
+                        "recovery attempt %d (model=%s provider=%s)",
+                        malformed_final_reason,
+                        agent._malformed_final_retries,
+                        agent.model,
+                        agent.provider,
+                    )
+                    agent._buffer_status(
+                        "⚠️ Model returned a malformed final response after "
+                        f"tool calls — retrying ({agent._malformed_final_retries}/2)"
+                    )
+                    if agent._malformed_final_retries == 1:
+                        recovery_msg = agent._build_assistant_message(
+                            assistant_message,
+                            finish_reason,
+                        )
+                        recovery_msg["content"] = "[malformed final response omitted]"
+                        recovery_msg["_malformed_final_recovery_synthetic"] = True
+                        messages.append(recovery_msg)
+                        messages.append({
+                            "role": "user",
+                            "content": (
+                                "The previous final response was malformed. "
+                                "Regenerate a concise, complete final answer "
+                                "from the tool results above. Do not repeat "
+                                "punctuation or stop mid-word."
+                            ),
+                            "_malformed_final_recovery_synthetic": True,
+                        })
+                        agent._session_messages = messages
+                        continue
+
+                    if agent._fallback_chain:
+                        agent._buffer_status(
+                            "⚠️ Malformed final response repeated — "
+                            "switching to fallback provider..."
+                        )
+                        if agent._try_activate_fallback():
+                            agent._buffer_status(
+                                f"↻ Switched to fallback: {agent.model} "
+                                f"({agent.provider})"
+                            )
+                            continue
+
+                    agent._flush_status_buffer()
+                    _turn_exit_reason = "malformed_final_exhausted"
+                    assistant_msg = agent._build_assistant_message(
+                        assistant_message,
+                        finish_reason,
+                    )
+                    assistant_msg["content"] = "(malformed final response)"
+                    assistant_msg["_malformed_final_recovery_synthetic"] = True
+                    messages.append(assistant_msg)
+                    final_response = (
+                        "Model returned a malformed final response after "
+                        "tool calls and recovery was exhausted. Try again "
+                        "or switch providers."
+                    )
+                    break
+
+                agent._malformed_final_retries = 0
                 # Successful content reached — drop any buffered retry
                 # status from earlier failed attempts in this turn.
                 agent._clear_status_buffer()
@@ -4261,6 +4334,7 @@ def _stop_spinner():
                         messages[-1].get("_thinking_prefill")
                         or messages[-1].get("_empty_recovery_synthetic")
                         or messages[-1].get("_empty_terminal_sentinel")
+                        or messages[-1].get("_malformed_final_recovery_synthetic")
                     )
                 ):
                     messages.pop()

cli.py

@@ -3577,8 +3577,17 @@ def _get_status_bar_snapshot(self) -> Dict[str, Any]:
 
         compressor = getattr(agent, "context_compressor", None)
         if compressor:
+            # last_prompt_tokens is parked at the -1 sentinel right after a
+            # compression, until the next real API call reports a prompt count
+            # (awaiting_real_usage_after_compression). The status bar must not
+            # render that sentinel verbatim — it produced "-1/200K" / "-1%".
+            # Clamp it to 0 so the one transitional turn reads as empty context.
             context_tokens = getattr(compressor, "last_prompt_tokens", 0) or 0
+            if context_tokens < 0:
+                context_tokens = 0
             context_length = getattr(compressor, "context_length", 0) or 0
+            if context_length < 0:
+                context_length = 0
             snapshot["context_tokens"] = context_tokens
             snapshot["context_length"] = context_length or None
             snapshot["compressions"] = getattr(compressor, "compression_count", 0) or 0
@@ -15074,6 +15083,96 @@ def new_event_loop(self):
 # Main Entry Point
 # ============================================================================
 
+def _run_kanban_goal_loop_q(cli: "HermesCLI", first_response: str) -> None:
+    """Drive a kanban goal_mode worker through the Ralph-style goal loop.
+
+    Called from the quiet single-query path AFTER the worker's first turn,
+    only when ``HERMES_KANBAN_GOAL_MODE`` is set (dispatcher-spawned
+    goal_mode card). Wires the worker's ``run_conversation`` and the kanban
+    DB into ``goals.run_kanban_goal_loop``. All errors are swallowed by the
+    caller — a broken goal loop must never wedge a worker, the dispatcher's
+    claim TTL / crash detection is the backstop.
+    """
+    import os as _os
+
+    task_id = (_os.environ.get("HERMES_KANBAN_TASK") or "").strip()
+    if not task_id:
+        return
+
+    from hermes_cli import kanban_db as _kb
+    from hermes_cli.goals import run_kanban_goal_loop as _run_loop, DEFAULT_MAX_TURNS as _DEF_TURNS
+
+    # Resolve goal text from the card (title + body = the acceptance
+    # criteria the judge evaluates against).
+    conn = _kb.connect()
+    try:
+        task = _kb.get_task(conn, task_id)
+    finally:
+        try:
+            conn.close()
+        except Exception:
+            pass
+    if task is None:
+        return
+
+    goal_parts = [task.title or ""]
+    if task.body:
+        goal_parts.append(task.body)
+    goal_text = "\n\n".join(p for p in goal_parts if p).strip()
+    if not goal_text:
+        return
+
+    max_turns = task.goal_max_turns or _DEF_TURNS
+
+    def _run_turn(prompt: str) -> str:
+        result = cli.agent.run_conversation(
+            user_message=prompt,
+            conversation_history=cli.conversation_history,
+        )
+        # Keep session_id in sync if mid-run compression rotated it.
+        if (
+            getattr(cli.agent, "session_id", None)
+            and cli.agent.session_id != cli.session_id
+        ):
+            cli.session_id = cli.agent.session_id
+        resp = result.get("final_response", "") if isinstance(result, dict) else str(result)
+        if resp:
+            print(resp)
+        return resp or ""
+
+    def _task_status() -> "str | None":
+        c = _kb.connect()
+        try:
+            t = _kb.get_task(c, task_id)
+            return t.status if t is not None else None
+        finally:
+            try:
+                c.close()
+            except Exception:
+                pass
+
+    def _block(reason: str) -> None:
+        c = _kb.connect()
+        try:
+            _kb.block_task(c, task_id, reason=reason)
+        finally:
+            try:
+                c.close()
+            except Exception:
+                pass
+
+    _run_loop(
+        task_id=task_id,
+        goal_text=goal_text,
+        run_turn=_run_turn,
+        task_status_fn=_task_status,
+        block_fn=_block,
+        max_turns=max_turns,
+        first_response=first_response or "",
+        log=lambda m: logger.info("%s", m),
+    )
+
+
 def main(
     query: str = None,
     q: str = None,
@@ -15471,6 +15570,20 @@ def _signal_handler_q(signum, frame):
                         print(f"Error: {result['error']}", file=sys.stderr)
                     elif response:
                         print(response)
+
+                    # Kanban goal-loop mode: a worker spawned for a
+                    # goal_mode card keeps working in THIS session until an
+                    # auxiliary judge agrees the card is done, the worker
+                    # terminates the task itself, or the turn budget runs
+                    # out (→ sticky block). Gated on the env vars the
+                    # dispatcher sets in `_default_spawn`; a no-op for every
+                    # normal worker and every non-kanban `-q` run.
+                    if os.environ.get("HERMES_KANBAN_GOAL_MODE") == "1":
+                        try:
+                            _run_kanban_goal_loop_q(cli, response)
+                        except Exception as _goal_exc:
+                            logger.debug("kanban goal loop failed: %s", _goal_exc)
+
                     # Session ID goes to stderr so piped stdout is clean.
                     print(f"\nsession_id: {cli.session_id}", file=sys.stderr)