From 629e6eb1b5f9cbabb7407fb9c1758d0a492fb008 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 16:46:33 -0700 Subject: [PATCH] refactor(turn): split finalize_turn into phase helpers (576 -> 264 LOC) --- agent/turn_finalizer.py | 813 +++++++++++++-------------- tests/agent/test_micro_compaction.py | 6 +- 2 files changed, 406 insertions(+), 413 deletions(-) diff --git a/agent/turn_finalizer.py b/agent/turn_finalizer.py index ca52cff981..741fe09e5f 100644 --- a/agent/turn_finalizer.py +++ b/agent/turn_finalizer.py @@ -1,14 +1,14 @@ """Post-loop turn finalization for ``run_conversation``. -Lifted verbatim: budget summary, trajectory save, persist, diagnostics, response -transforms, result assembly, steer drain, memory/skill review. Synchronous, single -return. ``logger`` is imported lazily from ``agent.conversation_loop`` (no cycle, -same logger name).""" +Budget summary, trajectory save, persist, diagnostics, response transforms, result +assembly, steer drain, memory/skill review. Synchronous, single return. ``logger`` is +imported lazily from ``agent.conversation_loop`` (no cycle, same logger name).""" from __future__ import annotations import logging import os +from typing import Any, Callable, List, Optional, Tuple from agent.codex_responses_adapter import _summarize_user_message_for_log from agent.context_compressor import _DB_PERSISTED_MARKER @@ -16,6 +16,16 @@ from agent.message_content import flatten_message_text from agent.message_metadata import append_message, stamp_message_timestamp from agent.message_sanitization import _sanitize_surrogates +# Verification-continuation nudges (verify-on-stop / pre_verify) must be stripped from +# returned/live history to avoid role-alternation breaks; the assistant response is +# real content and is not flagged. (#65919) +_VERIFICATION_CONTINUATION_FLAGS = ( + "_verification_stop_synthetic", + "_pre_verify_synthetic", +) + +_SENTENCE_END = {".", "!", "?", "。", "！", "？", "`", ")"} + def _assistant_row_missing_visible_text(msg: dict) -> bool: """True when an assistant row has no visible text (blank final or tool-only).""" @@ -39,15 +49,6 @@ def _fill_assistant_tail_content(agent, tail: dict, final_response) -> None: agent._db_flush_scan_prefix = None -# Verification-continuation nudges (verify-on-stop / pre_verify) must be stripped from -# returned/live history to avoid role-alternation breaks; the assistant response is -# real content and is not flagged. (#65919) -_VERIFICATION_CONTINUATION_FLAGS = ( - "_verification_stop_synthetic", - "_pre_verify_synthetic", -) - - def _record_kanban_budget_exhausted( kanban_task: str, api_call_count: int, @@ -56,8 +57,10 @@ def _record_kanban_budget_exhausted( ) -> None: """Record a terminal ``timed_out`` outcome for a kanban worker out of budget. - Idempotent via the ``_end_run`` CAS (``WHERE ended_at IS NULL``): a no-op if - another path already closed the run, so safe from multiple exit paths (#87096).""" + Routed via ``_record_task_failure`` (not ``kanban_block``) so it counts toward the + consecutive-failure circuit breaker (#29747). Idempotent via the ``_end_run`` CAS + (``WHERE ended_at IS NULL``): a no-op if another path already closed the run, so + safe from multiple exit paths (#87096).""" try: from hermes_cli import kanban_db as _kb _conn = _kb.connect() @@ -112,6 +115,306 @@ def _clone_background_review_messages(messages): return [_clone_message_for_send(message) for message in messages] +def _invoke_hook_safely(name: str, logger: logging.Logger, **kwargs) -> list: + """Fire a lifecycle plugin hook; a failing hook is logged, never fatal.""" + try: + from hermes_cli.lifecycle import invoke_hook + return invoke_hook(name, **kwargs) + except Exception as exc: + logger.warning("%s hook failed: %s", name, exc) + return [] + + +def _guarded_cleanup(label: str, fn: Callable[[], Any], errors: List[str], logger) -> None: + """Post-loop cleanup must never lose the response: each step is guarded + independently and errors surface via ``cleanup_errors`` (#8049).""" + try: + fn() + except Exception as err: + errors.append(f"{label}: {err}") + logger.error("finalize_turn: _%s failed: %s", label, err, exc_info=True) + + +def _resolve_budget_fallback( + agent, + *, + final_response, + api_call_count, + interrupted, + failed, + messages, + _turn_exit_reason, + _pending_verification_response, + _pending_verification_response_previewed, + logger, +) -> Tuple[Any, Any, bool]: + """Iteration-budget exhaustion. Returns ``(final_response, _turn_exit_reason, + preserved_verification_fallback)``.""" + budget_exhausted = ( + api_call_count >= agent.max_iterations + or agent.iteration_budget.remaining <= 0 + ) + budget_fallback_eligible = ( + budget_exhausted + and not interrupted + and not failed + and str(_turn_exit_reason) in {"unknown", "budget_exhausted"} + ) + preserved_verification_fallback = False + if final_response is None and budget_fallback_eligible: + _turn_exit_reason = f"max_iterations_reached({api_call_count}/{agent.max_iterations})" + if _pending_verification_response: + # A verification gate withheld a composed answer, then the budget ran out: + # preserve it rather than make another fallible call. The explicit pending + # value is the provenance guard; unrelated error exits never enter here. + final_response = _pending_verification_response + # Previewed only if the reused candidate was actually streamed as interim. + if _pending_verification_response_previewed: + agent._response_was_previewed = True + preserved_verification_fallback = True + else: + # _handle_max_iterations makes one extra toolless request for a summary. + agent._emit_status( + f"⚠️ Iteration budget exhausted ({api_call_count}/{agent.max_iterations}) " + "— asking model to summarise" + ) + if not agent.quiet_mode: + agent._safe_print( + f"\n⚠️ Iteration budget exhausted ({api_call_count}/{agent.max_iterations}) " + "— requesting summary..." + ) + final_response = agent._handle_max_iterations(messages, api_call_count) + + if budget_exhausted: + # A kanban worker must record a terminal outcome whether or not a fallback + # path was eligible, so the dispatcher learns the worker could not complete. + _kanban_task = os.environ.get("HERMES_KANBAN_TASK") + if _kanban_task: + _record_kanban_budget_exhausted( + _kanban_task, api_call_count, agent.max_iterations, logger, + ) + return final_response, _turn_exit_reason, preserved_verification_fallback + + +def _rollback_interrupted_preflight_display(agent, interrupted) -> None: + """Roll back the preflight-seeded display count only when an interrupt wins before + any provider response; compaction state (incl. ``-1``) stays with the real-usage + path. Type-pinned guards keep MagicMock/SimpleNamespace doubles inert.""" + _preflight_snapshot = getattr(agent, "_turn_preflight_display_snapshot", None) + if ( + interrupted is True + and isinstance(_preflight_snapshot, int) + and not isinstance(_preflight_snapshot, bool) + and getattr(agent, "_turn_received_provider_response", False) is not True + and getattr(agent, "context_compressor", None) is not None + ): + _rollback_fn = getattr( + agent.context_compressor, + "rollback_interrupted_preflight_display_tokens", + None, + ) + if callable(_rollback_fn): + _rollback_fn(_preflight_snapshot) + + +def _close_transcript_tail(agent, messages, final_response, interrupted, failed): + """Shape the transcript tail before the durable snapshot; returns the (possibly + stream-recovered) ``final_response``.""" + # Strip private retry scaffolding first, or a later "continue" replays + # assistant("(empty)") / recovery nudges into the same empty-response loop. Only + # the synthetic verification nudges go; the assistant candidate persists (#65919). + agent._drop_trailing_empty_response_scaffolding(messages) + _drop_verification_continuation_scaffolding(messages) + + # An empty terminal completion is not authoritative when the stream already + # delivered text; recover before persist so a blank tail isn't frozen (#95514). + _recovered_from_stream = False + if not interrupted and not failed: + _streamed = getattr(agent, "_current_streamed_assistant_text", "") or "" + _streamed = _streamed.strip() if isinstance(_streamed, str) else "" + _final_visible = flatten_message_text(final_response).strip() if final_response else "" + if not _final_visible and _streamed: + final_response = _streamed + _recovered_from_stream = True + + # An interrupt can leave a tool result as the tail (no scaffolding flag rewinds + # it); close the sequence so strict providers don't see ``tool → user``. An + # explicit placeholder is used since final_response is usually empty (#48879). + if interrupted: + from agent.message_sanitization import close_interrupted_tool_sequence + close_interrupted_tool_sequence(messages, final_response) + + # Recovery ``break`` sites can return a final_response with no closing assistant + # row; enforce "delivered final_response ⇒ assistant row" here. Compare content, + # not role, so a matching verification candidate isn't dup'd. + if final_response and not interrupted: + try: + _tail = messages[-1] if messages else None + except Exception: + _tail = None + _tail_role = _tail.get("role") if isinstance(_tail, dict) else None + if _tail_role != "assistant": + # Append so the durable turn closes with the answer (#43849/#44100). + append_message(messages, {"role": "assistant", "content": final_response}) + elif ( + isinstance(_tail, dict) + and _tail.get("content") != final_response + and ( + _is_pure_tool_call_tail(_tail) + or (_recovered_from_stream and _assistant_row_missing_visible_text(_tail)) + ) + ): + # Pure tool-call turn or stream-recovered blank (#95514): fill its content + # rather than append a second row. + _fill_assistant_tail_content(agent, _tail, final_response) + + # Request is complete, so replace API-local voice/model/skill guidance with the + # clean user input before the durable snapshot; earlier flushes used the DB-only + # override as their messages were still needed (#48677 / #63766). + _apply_override = getattr(agent, "_apply_persist_user_message_override", None) + if callable(_apply_override): + _apply_override(messages) + return final_response + + +def _micro_compact_after_turn(agent, messages, final_response, logger) -> None: + """Post-turn micro-compaction: absorb the oldest uncompacted exchange into the + rolling summary before persist, amortizing compression across turns.""" + try: + _compressor = getattr(agent, "context_compressor", None) + # Strict `is True` + isinstance gates: plugin context engines and MagicMock + # compressors pass duck checks and would wipe the transcript. + if ( + _compressor + and getattr(_compressor, '_micro_compact_enabled', False) is True + and callable(getattr(_compressor, '_micro_compact', None)) + and final_response + # No checkpoint hook, so never run while compression.checkpoint_required + # is armed. + and getattr(agent, "compression_checkpoint_required", False) is not True + # Persistence-isolated agents (background review fork) must not + # micro-compact: it burns an aux-LLM call on a throwaway transcript and + # could archive_and_compact the CANONICAL session rows. + and not getattr(agent, "_persist_disabled", False) + ): + _before = len(messages) + _compacted = _compressor._micro_compact(messages) + # Defrag rewrites the newest MICRO marker in place and pops _db_persisted; + # the compressor flags us to invalidate the flush-scan cursor, else the + # rewritten row is identity-skipped (stale). + if getattr(_compressor, "_flush_scan_cursor_invalidated", False): + _compressor._flush_scan_cursor_invalidated = False + agent._db_flush_scan_prefix = None + if isinstance(_compacted, list) and _compacted: + messages[:] = _compacted + _after = len(messages) + if _before != _after: + logger.info("Micro-compaction: %d -> %d messages", _before, _after) + except Exception as _mc_err: + logger.info("Micro-compaction failed: %s", _mc_err) + + +def _log_turn_exit(agent, messages, final_response, api_call_count, _turn_exit_reason, interrupted, logger) -> None: + """Always INFO so agent.log captures WHY every turn ended; WARNING when the last + message is a tool result (the "just stops" scenario).""" + _last_msg_role = messages[-1].get("role") if messages else None + _last_tool_name = None + if _last_msg_role == "tool": + # Walk back to the assistant message with the tool call. + for _m in reversed(messages): + if _m.get("role") == "assistant" and _m.get("tool_calls"): + _tcs = _m["tool_calls"] + if _tcs and isinstance(_tcs[0], dict): + _last_tool_name = _tcs[-1].get("function", {}).get("name") + break + + _turn_tool_count = sum( + 1 for m in messages + if isinstance(m, dict) and m.get("role") == "assistant" and m.get("tool_calls") + ) + _diag_msg = ( + "Turn ended: reason=%s model=%s api_calls=%d/%d budget=%d/%d " + "tool_turns=%d last_msg_role=%s response_len=%d session=%s" + ) + _diag_args = ( + _turn_exit_reason, agent.model, api_call_count, agent.max_iterations, + agent.iteration_budget.used if agent.iteration_budget else 0, + agent.iteration_budget.max_total if agent.iteration_budget else 0, + _turn_tool_count, _last_msg_role, len(final_response) if final_response else 0, + agent.session_id or "none", + ) + if _last_msg_role == "tool" and not interrupted: + logger.warning( + "Turn ended with pending tool result (agent may appear stuck). " + + _diag_msg + " last_tool=%s", + *_diag_args, _last_tool_name, + ) + else: + logger.info(_diag_msg, *_diag_args) + + +def _append_file_mutation_footer(agent, final_response, logger): + """If ``write_file`` / ``patch`` calls failed and were never superseded by a + successful write to the same path, append an advisory so over-claiming is + surfaced.""" + try: + _failed = getattr(agent, "_turn_failed_file_mutations", None) or {} + if _failed and agent._file_mutation_verifier_enabled(): + footer = agent._format_file_mutation_failure_footer(_failed) + if footer: + final_response = final_response.rstrip() + "\n\n" + footer + except Exception as _ver_err: + logger.debug("file-mutation verifier footer failed: %s", _ver_err) + return final_response + + +def _explain_abnormal_exit(agent, final_response, _turn_exit_reason, preserved_verification_fallback, logger): + """Turn-completion explainer: on abnormal exits, surface one explanation from + ``_turn_exit_reason``. Only acts when no usable reply exists (empty, "(empty)", + or a short unpunctuated fragment); ``text_response(...)`` exits stay silent.""" + try: + if not agent._turn_completion_explainer_enabled(): + return final_response + _stripped = (final_response or "").strip() + _is_empty_terminal = _stripped == "" or _stripped == "(empty)" + # A short fragment not from a text_response exit and lacking sentence-ending + # punctuation is treated as a truncated partial (#34452). + _is_partial_fragment = ( + not _is_empty_terminal + and not preserved_verification_fallback + and not str(_turn_exit_reason).startswith("text_response") + and len(_stripped) <= 24 + and _stripped[-1:] not in _SENTENCE_END + ) + if ( + _is_empty_terminal + or _is_partial_fragment + or str(_turn_exit_reason) == "partial_stream_recovery" + ): + _explanation = agent._format_turn_completion_explanation( + _turn_exit_reason, + getattr(agent, "_last_persistence_error_cause", None), + ) + if _explanation: + # Replace the bare sentinel; keep a partial fragment and append why. + final_response = _explanation if _is_empty_terminal else _stripped + "\n\n" + _explanation + except Exception as _exp_err: + logger.debug("turn-completion explainer failed: %s", _exp_err) + return final_response + + +def _last_turn_reasoning(messages) -> Optional[Any]: + """Reasoning from the CURRENT turn only: stop at this turn's user message (#17055), + but take the most recent non-empty reasoning since many providers emit it on the + tool-call step and leave the final step with reasoning=None.""" + for msg in reversed(messages): + if msg.get("role") == "user": + return None # turn boundary — don't cross into prior turns + if msg.get("role") == "assistant" and msg.get("reasoning"): + return msg["reasoning"] + return None + + def finalize_turn( agent, *, @@ -133,234 +436,48 @@ def finalize_turn( """Run the post-loop finalization and return the turn ``result`` dict.""" from agent.conversation_loop import logger - budget_exhausted = ( - api_call_count >= agent.max_iterations - or agent.iteration_budget.remaining <= 0 - ) - budget_fallback_eligible = ( - budget_exhausted - and not interrupted - and not failed - and str(_turn_exit_reason) in {"unknown", "budget_exhausted"} - ) - continuation_budget_exhausted = ( - final_response is None - and bool(_pending_verification_response) - and budget_fallback_eligible + final_response, _turn_exit_reason, preserved_verification_fallback = _resolve_budget_fallback( + agent, + final_response=final_response, + api_call_count=api_call_count, + interrupted=interrupted, + failed=failed, + messages=messages, + _turn_exit_reason=_turn_exit_reason, + _pending_verification_response=_pending_verification_response, + _pending_verification_response_previewed=_pending_verification_response_previewed, + logger=logger, ) - iteration_limit_fallback = False - preserved_verification_fallback = False - if continuation_budget_exhausted: - # A verification gate withheld a composed answer, then the budget ran out: - # preserve it rather than make another fallible call. The explicit pending - # value is the provenance guard; unrelated error exits never enter here. - final_response = _pending_verification_response - # Previewed only if the reused candidate was actually streamed as interim. - if _pending_verification_response_previewed: - agent._response_was_previewed = True - _turn_exit_reason = f"max_iterations_reached({api_call_count}/{agent.max_iterations})" - iteration_limit_fallback = True - preserved_verification_fallback = True - elif final_response is None and budget_fallback_eligible: - # Budget exhausted: _handle_max_iterations makes one extra toolless request - # for a summary. - _turn_exit_reason = f"max_iterations_reached({api_call_count}/{agent.max_iterations})" - agent._emit_status( - f"⚠️ Iteration budget exhausted ({api_call_count}/{agent.max_iterations}) " - "— asking model to summarise" - ) - if not agent.quiet_mode: - agent._safe_print( - f"\n⚠️ Iteration budget exhausted ({api_call_count}/{agent.max_iterations}) " - "— requesting summary..." - ) - final_response = agent._handle_max_iterations(messages, api_call_count) - iteration_limit_fallback = True - - if iteration_limit_fallback: - # Kanban worker: signal the dispatcher the worker could not complete. Route - # via ``_record_task_failure(outcome="timed_out")`` (not ``kanban_block``) so - # it counts toward the consecutive-failure circuit breaker (#29747). - _kanban_task = os.environ.get("HERMES_KANBAN_TASK") - if _kanban_task: - _record_kanban_budget_exhausted( - _kanban_task, api_call_count, agent.max_iterations, logger, - ) - elif budget_exhausted: - # Bounded fallback: budget exhausted with no eligible fallback path. A kanban - # worker must still record a terminal outcome; the ``_end_run`` CAS makes it - # idempotent if another path already closed the run (#87096). - _kanban_task = os.environ.get("HERMES_KANBAN_TASK") - if _kanban_task: - _record_kanban_budget_exhausted( - _kanban_task, api_call_count, agent.max_iterations, logger, - ) - - # Determine if conversation completed successfully - normal_text_response = str(_turn_exit_reason).startswith("text_response(") completed = ( final_response is not None and not failed and ( api_call_count < agent.max_iterations - or normal_text_response + or str(_turn_exit_reason).startswith("text_response(") ) ) - # Roll back the preflight-seeded display count only when an interrupt wins - # before any provider response; compaction state (incl. ``-1``) stays with the - # real-usage path. Type-pinned guards keep MagicMock/SimpleNamespace doubles inert. - _preflight_snapshot = getattr( - agent, "_turn_preflight_display_snapshot", None + _rollback_interrupted_preflight_display(agent, interrupted) + + _cleanup_errors: List[str] = [] + # ``user_message`` may be a multimodal list of parts; the trajectory format wants + # a plain string. + _guarded_cleanup( + "save_trajectory", + lambda: agent._save_trajectory(messages, _summarize_user_message_for_log(user_message), completed), + _cleanup_errors, logger, ) - if ( - interrupted is True - and isinstance(_preflight_snapshot, int) - and not isinstance(_preflight_snapshot, bool) - and getattr(agent, "_turn_received_provider_response", False) is not True - and getattr(agent, "context_compressor", None) is not None - ): - _rollback_fn = getattr( - agent.context_compressor, - "rollback_interrupted_preflight_display_tokens", - None, - ) - if callable(_rollback_fn): - _rollback_fn(_preflight_snapshot) - - # Post-loop cleanup must never lose the response: trajectory save, teardown, - # and session persist are guarded independently and errors surface via - # ``cleanup_errors`` rather than killing the turn (#8049). - _cleanup_errors = [] - - # Save trajectory if enabled. ``user_message`` may be a multimodal - # list of parts; the trajectory format wants a plain string. + _guarded_cleanup( + "cleanup_task_resources", + lambda: agent._cleanup_task_resources(effective_task_id), + _cleanup_errors, logger, + ) + # Persist only after the transcript tail is shaped and scaffolding removed. try: - agent._save_trajectory(messages, _summarize_user_message_for_log(user_message), completed) - except Exception as _save_err: - _cleanup_errors.append(f"save_trajectory: {_save_err}") - logger.error("finalize_turn: _save_trajectory failed: %s", _save_err, exc_info=True) - - # Clean up VM and browser for this task after conversation completes - try: - agent._cleanup_task_resources(effective_task_id) - except Exception as _cleanup_err: - _cleanup_errors.append(f"cleanup_task_resources: {_cleanup_err}") - logger.error("finalize_turn: _cleanup_task_resources failed: %s", _cleanup_err, exc_info=True) - - # Persist only after private retry scaffolding is removed, or a later "continue" - # replays assistant("(empty)") / recovery nudges into the same empty-response loop. - try: - agent._drop_trailing_empty_response_scaffolding(messages) - - # Strip only the synthetic verification nudges before the tail-assistant - # check; the assistant candidate persists in state.db. (#65919) - _drop_verification_continuation_scaffolding(messages) - - # An empty terminal completion is not authoritative when the stream already - # delivered text; recover before persist so a blank tail isn't frozen (#95514). - _recovered_from_stream = False + final_response = _close_transcript_tail(agent, messages, final_response, interrupted, failed) if not interrupted and not failed: - _streamed = getattr(agent, "_current_streamed_assistant_text", "") or "" - if isinstance(_streamed, str): - _streamed = _streamed.strip() - else: - _streamed = "" - _final_visible = flatten_message_text(final_response).strip() if final_response else "" - if not _final_visible and _streamed: - final_response = _streamed - _recovered_from_stream = True - - # An interrupt can leave a tool result as the tail (no scaffolding flag rewinds - # it); close the sequence so strict providers don't see ``tool → user``. An - # explicit placeholder is used since final_response is usually empty (#48879). - if interrupted: - from agent.message_sanitization import close_interrupted_tool_sequence - close_interrupted_tool_sequence(messages, final_response) - - # Recovery ``break`` sites can return a final_response with no closing - # assistant row; enforce "delivered final_response ⇒ assistant row" here. - # Compare content, not role, so a matching verification candidate isn't dup'd. - if final_response and not interrupted: - try: - _tail = messages[-1] if messages else None - except Exception: - _tail = None - _tail_role = _tail.get("role") if isinstance(_tail, dict) else None - if _tail_role != "assistant": - # Tail is not an assistant row — append the final response - # so the durable turn closes with the answer (#43849/#44100). - append_message( - messages, - {"role": "assistant", "content": final_response}, - ) - elif ( - isinstance(_tail, dict) - and _tail.get("content") != final_response - and ( - _is_pure_tool_call_tail(_tail) - or ( - _recovered_from_stream - and _assistant_row_missing_visible_text(_tail) - ) - ) - ): - # Tail is an assistant row (pure tool-call turn or stream-recovered - # blank, #95514): fill its content rather than append a second row. - _fill_assistant_tail_content(agent, _tail, final_response) - - # Request is complete, so replace API-local voice/model/skill guidance with - # the clean user input before the durable snapshot; earlier flushes used the - # DB-only override as their messages were still needed (#48677 / #63766). - _apply_override = getattr(agent, "_apply_persist_user_message_override", None) - if callable(_apply_override): - _apply_override(messages) - - # ── Post-turn micro-compaction ──────────────────────────── - # Absorb the oldest uncompacted exchange into the rolling summary before - # persist, amortizing compression across turns instead of one big pause. - if not interrupted and not failed: - try: - _compressor = getattr(agent, "context_compressor", None) - # Strict `is True` + isinstance gates: plugin context engines and - # MagicMock compressors pass duck checks and would wipe the transcript. - if ( - _compressor - and getattr(_compressor, '_micro_compact_enabled', False) is True - and callable(getattr(_compressor, '_micro_compact', None)) - and final_response - # Micro-compaction has no checkpoint hook, so it must never run - # while compression.checkpoint_required is armed. - and getattr( - agent, "compression_checkpoint_required", False - ) is not True - # Persistence-isolated agents (background review fork) must not - # micro-compact: it burns an aux-LLM call on a throwaway transcript - # and could archive_and_compact the CANONICAL session rows. - and not getattr(agent, "_persist_disabled", False) - ): - _before = len(messages) - _compacted = _compressor._micro_compact(messages) - # Defrag rewrites the newest MICRO marker in place and pops - # _db_persisted; the compressor flags us to invalidate the flush- - # scan cursor, else the rewritten row is identity-skipped (stale). - if getattr( - _compressor, "_flush_scan_cursor_invalidated", False - ): - _compressor._flush_scan_cursor_invalidated = False - agent._db_flush_scan_prefix = None - if isinstance(_compacted, list) and _compacted: - messages[:] = _compacted - _after = len(messages) - if _before != _after: - logger.info( - "Micro-compaction: %d -> %d messages", - _before, _after, - ) - except Exception as _mc_err: - logger.info("Micro-compaction failed: %s", _mc_err) - + _micro_compact_after_turn(agent, messages, final_response, logger) agent._persist_session(messages, conversation_history) except Exception as _persist_err: _cleanup_errors.append(f"persist_session: {_persist_err}") @@ -373,146 +490,47 @@ def finalize_turn( except Exception: pass - # ── Turn-exit diagnostic log ───────────────────────────────────── - # Always INFO so agent.log captures WHY every turn ended; WARNING when the last - # message is a tool result (the "just stops" scenario). - _last_msg_role = messages[-1].get("role") if messages else None - _last_tool_name = None - if _last_msg_role == "tool": - # Walk back to find the assistant message with the tool call - for _m in reversed(messages): - if _m.get("role") == "assistant" and _m.get("tool_calls"): - _tcs = _m["tool_calls"] - if _tcs and isinstance(_tcs[0], dict): - _last_tool_name = _tcs[-1].get("function", {}).get("name") - break + _log_turn_exit(agent, messages, final_response, api_call_count, _turn_exit_reason, interrupted, logger) - _turn_tool_count = sum( - 1 for m in messages - if isinstance(m, dict) and m.get("role") == "assistant" and m.get("tool_calls") - ) - _resp_len = len(final_response) if final_response else 0 - _budget_used = agent.iteration_budget.used if agent.iteration_budget else 0 - _budget_max = agent.iteration_budget.max_total if agent.iteration_budget else 0 - - _diag_msg = ( - "Turn ended: reason=%s model=%s api_calls=%d/%d budget=%d/%d " - "tool_turns=%d last_msg_role=%s response_len=%d session=%s" - ) - _diag_args = ( - _turn_exit_reason, agent.model, api_call_count, agent.max_iterations, - _budget_used, _budget_max, - _turn_tool_count, _last_msg_role, _resp_len, - agent.session_id or "none", - ) - - if _last_msg_role == "tool" and not interrupted: - # Agent was mid-work — this is the "just stops" case. - logger.warning( - "Turn ended with pending tool result (agent may appear stuck). " - + _diag_msg + " last_tool=%s", - *_diag_args, _last_tool_name, - ) - else: - logger.info(_diag_msg, *_diag_args) - - # File-mutation verifier footer: if ``write_file`` / ``patch`` calls failed and - # were never superseded by a successful write to the same path, append an - # advisory so over-claiming is surfaced. Only on real, uninterrupted responses. + # Response transforms apply only to real, uninterrupted responses. if final_response and not interrupted: - try: - _failed = getattr(agent, "_turn_failed_file_mutations", None) or {} - if _failed and agent._file_mutation_verifier_enabled(): - footer = agent._format_file_mutation_failure_footer(_failed) - if footer: - final_response = final_response.rstrip() + "\n\n" + footer - except Exception as _ver_err: - logger.debug("file-mutation verifier footer failed: %s", _ver_err) - - # Turn-completion explainer: on abnormal exits, surface one explanation from - # ``_turn_exit_reason``. Only acts when no usable reply exists (empty, "(empty)", - # or a short unpunctuated fragment); ``text_response(...)`` exits stay silent. + final_response = _append_file_mutation_footer(agent, final_response, logger) if not interrupted: - try: - if agent._turn_completion_explainer_enabled(): - _stripped = (final_response or "").strip() - _is_empty_terminal = _stripped == "" or _stripped == "(empty)" - # A short fragment not from a text_response exit and lacking sentence- - # ending punctuation is treated as a truncated partial (#34452). - _is_partial_fragment = ( - not _is_empty_terminal - and not preserved_verification_fallback - and not str(_turn_exit_reason).startswith("text_response") - and len(_stripped) <= 24 - and _stripped[-1:] not in {".", "!", "?", "。", "！", "？", "`", ")"} - ) - _is_partial_stream_recovery = ( - str(_turn_exit_reason) == "partial_stream_recovery" - ) - if ( - _is_empty_terminal - or _is_partial_fragment - or _is_partial_stream_recovery - ): - _explanation = agent._format_turn_completion_explanation( - _turn_exit_reason, - getattr(agent, "_last_persistence_error_cause", None), - ) - if _explanation: - if _is_empty_terminal: - # Replace the bare "(empty)"/blank sentinel with - # the actionable explanation. - final_response = _explanation - else: - # Keep the partial fragment and append why it stopped. - final_response = ( - _stripped + "\n\n" + _explanation - ) - except Exception as _exp_err: - logger.debug("turn-completion explainer failed: %s", _exp_err) + final_response = _explain_abnormal_exit( + agent, final_response, _turn_exit_reason, preserved_verification_fallback, logger, + ) + _platform = getattr(agent, "platform", None) or "" _response_transformed = False _pre_transform_response = None - - # Plugin hook: transform_llm_output — fired once per turn after the tool loop. - # First hook to return a string wins; None/empty leaves the text unchanged. if final_response and not interrupted: - try: - from hermes_cli.lifecycle import invoke_hook as _invoke_hook - _transform_results = _invoke_hook( - "transform_llm_output", - response_text=final_response, - session_id=agent.session_id or "", - model=agent.model, - platform=getattr(agent, "platform", None) or "", - ) - for _hook_result in _transform_results: - if isinstance(_hook_result, str) and _hook_result: - _pre_transform_response = final_response - final_response = _hook_result - _response_transformed = True - break # First non-empty string wins - except Exception as exc: - logger.warning("transform_llm_output hook failed: %s", exc) - - # Plugin hook: post_llm_call — fired once per turn after the tool loop (e.g. sync - # conversation data to an external memory system). - if final_response and not interrupted: - try: - from hermes_cli.lifecycle import invoke_hook as _invoke_hook - _invoke_hook( - "post_llm_call", - session_id=agent.session_id, - task_id=effective_task_id, - turn_id=turn_id, - user_message=original_user_message, - assistant_response=final_response, - conversation_history=list(messages), - model=agent.model, - platform=getattr(agent, "platform", None) or "", - ) - except Exception as exc: - logger.warning("post_llm_call hook failed: %s", exc) + # Plugin hook: transform_llm_output — fired once per turn after the tool loop. + # First hook to return a string wins; None/empty leaves the text unchanged. + for _hook_result in _invoke_hook_safely( + "transform_llm_output", logger, + response_text=final_response, + session_id=agent.session_id or "", + model=agent.model, + platform=_platform, + ): + if isinstance(_hook_result, str) and _hook_result: + _pre_transform_response = final_response + final_response = _hook_result + _response_transformed = True + break + # Plugin hook: post_llm_call (e.g. sync conversation data to an external + # memory system). + _invoke_hook_safely( + "post_llm_call", logger, + session_id=agent.session_id, + task_id=effective_task_id, + turn_id=turn_id, + user_message=original_user_message, + assistant_response=final_response, + conversation_history=list(messages), + model=agent.model, + platform=_platform, + ) # Context engine observation hook (complements per-request select_context()): # notify the engine the turn finished with the finalized transcript. Fail-open. @@ -520,11 +538,10 @@ def finalize_turn( from agent.conversation_loop import _notify_context_engine_turn_complete # ``_last_turn_usage`` holds the last API response's canonical usage dict, or # ``None`` on turns that never reached a provider response — by contract. - _turn_usage = getattr(agent, "_last_turn_usage", None) _notify_context_engine_turn_complete( agent, messages, - usage=_turn_usage, + usage=getattr(agent, "_last_turn_usage", None), logger=logger, turn_id=turn_id, task_id=effective_task_id, @@ -536,27 +553,15 @@ def finalize_turn( except Exception as exc: logger.warning("on_turn_complete notification failed: %s", exc) - # Reasoning from the CURRENT turn only: stop at this turn's user message - # (#17055), but take the most recent non-empty reasoning since many providers - # emit it on the tool-call step and leave the final step with reasoning=None. - last_reasoning = None - for msg in reversed(messages): - if msg.get("role") == "user": - break # turn boundary — don't cross into prior turns - if msg.get("role") == "assistant" and msg.get("reasoning"): - last_reasoning = msg["reasoning"] - break - # Surrogate chokepoint: ``final_response`` may be RAW SDK content, and a lone UTF-16 # surrogate crashes downstream consumers (stdout, Telegram ``utf16_len``, JSON). # Scrub once where model text leaves the loop (#80366). if isinstance(final_response, str): final_response = _sanitize_surrogates(final_response) - # Build result with interrupt info if applicable result = { "final_response": final_response, - "last_reasoning": last_reasoning, + "last_reasoning": _last_turn_reasoning(messages), "messages": messages, "api_calls": api_call_count, "completed": completed, @@ -582,8 +587,8 @@ def finalize_turn( "estimated_cost_usd": agent.session_estimated_cost_usd, "cost_status": agent.session_cost_status, "cost_source": agent.session_cost_source, - # Requested service tier (from request_overrides.extra_body), for - # billing audits by callers like `hermes -z --usage-file`. + # Requested service tier (from request_overrides.extra_body), for billing + # audits by callers like `hermes -z --usage-file`. "service_tier": ( (getattr(agent, "request_overrides", {}) or {}).get("extra_body") or {} ).get("service_tier"), @@ -603,9 +608,7 @@ def finalize_turn( # Never clobber a failure_reason another path already stamped. if "failure_reason" not in result: _cause = getattr(agent, "_last_persistence_error_cause", None) - result["failure_reason"] = ( - "session_persistence_failed:" + (_cause or "unknown") - ) + result["failure_reason"] = "session_persistence_failed:" + (_cause or "unknown") # Surface post-loop cleanup failures so the caller can tell a clean turn from one # whose teardown raised; the response is returned either way (#8049). if _cleanup_errors: @@ -616,18 +619,13 @@ def finalize_turn( if _leftover_steer: result["pending_steer"] = _leftover_steer agent._response_was_previewed = False - - # Include interrupt message if one triggered the interrupt if interrupted and agent._interrupt_message: result["interrupt_message"] = agent._interrupt_message - - # Clear interrupt state after handling agent.clear_interrupt() - - # Clear stream callback so it doesn't leak into future calls + # Clear stream callback so it doesn't leak into future calls. agent._stream_callback = None - # Check skill trigger NOW — based on how many tool iterations THIS turn used. + # Skill trigger is checked NOW — based on how many tool iterations THIS turn used. _should_review_skills = False if (agent._skill_nudge_interval > 0 and agent._iters_since_skill >= agent._skill_nudge_interval @@ -653,8 +651,8 @@ def finalize_turn( and (_should_review_memory or _should_review_skills) ): try: - # _spawn_background_review clones the snapshot structurally so - # the fork's in-place sanitizers can't reach the live transcript. + # _spawn_background_review clones the snapshot structurally so the fork's + # in-place sanitizers can't reach the live transcript. agent._spawn_background_review( messages_snapshot=list(messages), review_memory=_should_review_memory, @@ -665,26 +663,19 @@ def finalize_turn( # Memory provider on_session_end()/shutdown_all() are NOT called here: # run_conversation() runs once per message; CLI/gateway own session-end cleanup. - - # Plugin hook: on_session_end — fired at the end of every run_conversation call. - try: - from hermes_cli.lifecycle import invoke_hook as _invoke_hook - _invoke_hook( - "on_session_end", - session_id=agent.session_id, - task_id=effective_task_id, - turn_id=turn_id, - completed=completed, - failed=failed, - interrupted=interrupted, - turn_exit_reason=_turn_exit_reason, - model=agent.model, - platform=getattr(agent, "platform", None) or "", - ) - except Exception as exc: - logger.warning("on_session_end hook failed: %s", exc) + _invoke_hook_safely( + "on_session_end", logger, + session_id=agent.session_id, + task_id=effective_task_id, + turn_id=turn_id, + completed=completed, + failed=failed, + interrupted=interrupted, + turn_exit_reason=_turn_exit_reason, + model=agent.model, + platform=_platform, + ) agent._turn_preflight_display_snapshot = None agent._turn_received_provider_response = False - return result diff --git a/tests/agent/test_micro_compaction.py b/tests/agent/test_micro_compaction.py index 4c56e585aa..86ed0ae1cb 100644 --- a/tests/agent/test_micro_compaction.py +++ b/tests/agent/test_micro_compaction.py @@ -724,7 +724,8 @@ class TestMicroCompaction: from agent import turn_finalizer - src = inspect.getsource(turn_finalizer.finalize_turn) + # The micro-compaction gate lives in the helper finalize_turn calls. + src = inspect.getsource(turn_finalizer._micro_compact_after_turn) micro_block = src.split("Post-turn micro-compaction", 1)[1] # Scope to the micro block only: stop at the persist call that follows. micro_block = micro_block.split("agent._persist_session", 1)[0] @@ -810,7 +811,8 @@ class TestDefragFlushCursorInvalidation: from agent import turn_finalizer - src = inspect.getsource(turn_finalizer.finalize_turn) + # The micro-compaction gate lives in the helper finalize_turn calls. + src = inspect.getsource(turn_finalizer._micro_compact_after_turn) micro_block = src.split("Post-turn micro-compaction", 1)[1] micro_block = micro_block.split("agent._persist_session", 1)[0] assert "_flush_scan_cursor_invalidated" in micro_block, (