OpenSpace/openspace/agents/turns/loop.py

"""Single-turn LLM/tool loop for GroundingAgent."""

from __future__ import annotations

import asyncio
import copy
import os
import time
from contextlib import nullcontext
from typing import Any

from openspace.agents.turns import compaction_controller
from openspace.agents.turns import events as turn_events
from openspace.agents.turns import model_call_controller
from openspace.agents.turns import stop_policy
from openspace.agents.turns import tool_turn_controller
from openspace.agents.turns.context import TurnControllerContext
from openspace.agents.turns.state import TurnState
from openspace.agents.turns.task_query import resolve_task_query
from openspace.services.memory.recall import start_relevant_memory_prefetch
from openspace.services.conversation.messages import extract_discovered_tool_names
from openspace.utils.logging import Logger

logger = Logger.get_logger(__name__)


def _env_bool(name: str, default: bool = False) -> bool:
    raw = os.environ.get(name)
    if raw is None:
        return default
    return raw.strip().lower() in {"1", "true", "yes", "y", "on"}


def _env_int(name: str, default: int) -> int:
    raw = os.environ.get(name)
    if raw is None:
        return default
    try:
        return int(raw)
    except (TypeError, ValueError):
        return default


def _maybe_build_bench_finalize_nudge(state: TurnState) -> dict[str, Any] | None:
    if not _env_bool("OPENSPACE_BENCH_FINALIZE_NUDGE_ENABLED"):
        return None

    max_nudges = max(0, _env_int("OPENSPACE_BENCH_FINALIZE_NUDGE_MAX", 1))
    if state.bench_finalize_nudge_count >= max_nudges:
        return None

    elapsed_s = time.monotonic() - state.started_at_monotonic
    after_sec = _env_int("OPENSPACE_BENCH_FINALIZE_NUDGE_AFTER_SEC", 0)
    after_iteration = _env_int("OPENSPACE_BENCH_FINALIZE_NUDGE_AFTER_ITERATION", 0)

    due_by_time = after_sec > 0 and elapsed_s >= after_sec
    due_by_iteration = (
        after_iteration > 0
        and state.current_iteration >= after_iteration
    )
    if not due_by_time and not due_by_iteration:
        return None

    state.bench_finalize_nudge_count += 1
    state.bench_finalize_nudge_iteration = state.current_iteration
    state.bench_finalize_nudge_monotonic = time.monotonic()
    return {
        "role": "user",
        "content": (
            "Terminal-Bench finalization checkpoint: the run budget is getting "
            "low. Immediately preserve the best current answer in the exact "
            "requested /app artifact path(s). If you have found a passing "
            "payload, script, data file, or command output, write it to the "
            "target artifact now. Run at most one quick verification against "
            "the visible checker or artifact parseability, then stop open-ended "
            "exploration and provide the final response."
        ),
        "_meta": {
            "type": "bench_finalize_nudge",
            "is_meta": True,
            "elapsed_s": round(elapsed_s, 3),
            "iteration": state.current_iteration,
        },
    }


def _bench_finalize_budget_exhausted(state: TurnState) -> tuple[bool, str]:
    if not _env_bool("OPENSPACE_BENCH_FINALIZE_NUDGE_ENABLED"):
        return False, ""
    if state.bench_finalize_nudge_count <= 0:
        return False, ""

    stop_after_iterations = _env_int(
        "OPENSPACE_BENCH_FINALIZE_STOP_AFTER_ITERATIONS",
        0,
    )
    if (
        stop_after_iterations > 0
        and state.bench_finalize_nudge_iteration is not None
        and state.current_iteration
        > state.bench_finalize_nudge_iteration + stop_after_iterations
    ):
        return (
            True,
            f"{stop_after_iterations} iterations after finalize nudge",
        )

    stop_after_sec = _env_int("OPENSPACE_BENCH_FINALIZE_STOP_AFTER_SEC", 0)
    if (
        stop_after_sec > 0
        and state.bench_finalize_nudge_monotonic is not None
        and time.monotonic() - state.bench_finalize_nudge_monotonic >= stop_after_sec
    ):
        return True, f"{stop_after_sec}s after finalize nudge"

    return False, ""


def _bench_checker_pass_budget_exhausted(state: TurnState) -> tuple[bool, str]:
    stop_after_iterations = _env_int(
        "OPENSPACE_BENCH_STOP_AFTER_CHECKER_PASS_ITERATIONS",
        0,
    )
    pass_iteration = getattr(state, "bench_visible_checker_pass_iteration", None)
    if (
        stop_after_iterations > 0
        and pass_iteration is not None
        and not bool(getattr(state, "bench_visible_checker_failed", False))
        and state.current_iteration > pass_iteration + stop_after_iterations
    ):
        return True, f"{stop_after_iterations} iterations after visible checker pass"
    return False, ""


def _append_lifecycle_hook_contexts(
    context: dict[str, Any],
    hook_result: Any,
) -> None:
    additional_contexts = getattr(hook_result, "additional_contexts", None)
    if additional_contexts:
        target = context.setdefault("hook_additional_contexts", [])
        if isinstance(target, list):
            target.extend(str(item) for item in additional_contexts if item)
    initial_user_message = getattr(hook_result, "initial_user_message", None)
    if initial_user_message:
        context["session_start_initial_user_message"] = str(initial_user_message)
    watch_paths = getattr(hook_result, "watch_paths", None)
    if watch_paths:
        target = context.setdefault("hook_watch_paths", [])
        if isinstance(target, list):
            target.extend(str(path) for path in watch_paths if path)


def _reserve_session_start_hooks(
    agent: Any,
    *,
    session_id: str | None,
    source: str,
) -> bool:
    key = f"{session_id or 'anonymous'}:{source}"
    seen = getattr(agent, "_session_start_hooks_seen", None)
    if not isinstance(seen, set):
        seen = set()
        setattr(agent, "_session_start_hooks_seen", seen)
    if key in seen:
        return False
    seen.add(key)
    return True


async def _emit_lifecycle_hook_messages(agent: Any, hook_result: Any) -> None:
    for message in getattr(hook_result, "messages", []) or []:
        await agent._emit_runtime_event(
            "hook_message",
            {"message": message},
        )
    for message in getattr(hook_result, "blocking_errors", []) or []:
        await agent._emit_runtime_event(
            "hook_message",
            {"message": message},
        )


async def run_grounding_turn(agent: Any, context: dict[str, Any]) -> dict[str, Any]:
    """OpenSpace external orchestration loop.

    Uses ``call_model()`` for LLM calls and ``run_tools()`` for tool execution.
    Implementation: ``query.ts`` ``queryLoop`` (L241-1729).

    Flow per iteration:
        1. abort check
        2. drain external messages (multi-agent inbox)
        3. auto_compact_if_needed (LLM summary compression)
        4. call_model (pure LLM, no tool execution)
        5. handle errors: PTL → compact retry, Fallback → model switch
        6. no tool_calls → stop hooks → break/continue
        7. run_tools (parallel/serial orchestration)
        8. enforce tool result budget
        9. prevent_continuation check
       10. iteration guidance injection → continue
    """
    self = agent
    instruction = context.get("instruction", "")
    if not instruction:
        logger.error("Grounding Agent: No instruction provided")
        return {"error": "No instruction provided", "status": "error"}

    self._current_instruction = instruction
    tool_retrieval_instruction = resolve_task_query(context, instruction)

    logger.info(f"Grounding Agent: Processing instruction at step {self.step}")
    agent_id = str(context.get("agent_id") or "primary")
    await self._emit_runtime_event(
        "agent_start",
        turn_events.agent_start_payload(
            self,
            context,
            agent_id=agent_id,
            instruction=instruction,
        ),
    )

    memory_mode = self._resolve_memory_mode(context.get("memory_mode"))
    context["memory_mode"] = memory_mode
    skills_disabled = self._skills_disabled_for_context(context)
    low_latency_profiler = context.get("low_latency_profiler")

    def _latency_span(name: str, **metadata: Any):
        if low_latency_profiler is None:
            return nullcontext()
        span = getattr(low_latency_profiler, "span", None)
        if not callable(span):
            return nullcontext()
        return span(name, **metadata)

    # ── Phase 0: Pre-loop setup (workspace, tools, messages) ──────

    with _latency_span("turn.workspace_scan"):
        workspace_info = await self._check_workspace_artifacts(context)
    if workspace_info["has_files"]:
        context["workspace_artifacts"] = workspace_info
        logger.info(f"Workspace has {len(workspace_info['files'])} existing files: {workspace_info['files']}")

    from openspace.services.runtime_support.low_latency import (
        SessionCapabilityState,
        build_active_tool_policy,
        classify_capability_profile,
    )

    session_capability_state_enabled = bool(
        context.get("session_capability_state_enabled", True)
    )
    session_capability_state = (
        SessionCapabilityState.from_mapping(
            context.get("session_capability_state")
        )
        if session_capability_state_enabled
        else SessionCapabilityState()
    )
    if (
        session_capability_state_enabled
        and session_capability_state.visible_skill_names
        and not context.get("sent_skill_names_by_agent")
    ):
        context["sent_skill_names_by_agent"] = {
            "primary": set(session_capability_state.visible_skill_names)
        }

    requested_profile = str(context.get("capability_profile") or "batch_full")
    classified_profile = classify_capability_profile(
        instruction,
        default_profile=requested_profile,
    )
    context["capability_profile"] = classified_profile.name
    if low_latency_profiler is not None:
        try:
            low_latency_profiler.profile = classified_profile.name
        except Exception:
            pass
    use_fast_tool_policy = bool(
        context.get("low_latency_enabled")
        and context.get("fast_tool_policy_enabled")
        and classified_profile.name == "interactive_fast"
        and context.get("disable_fast_auto_preselection", True)
    )
    active_tool_policy = None

    pre_filtered_tools = context.get("pre_filtered_tools")
    if pre_filtered_tools is not None:
        preselected_tools = list(pre_filtered_tools)
        tools = list(context.get("all_tools") or preselected_tools)
        tools = self._with_memory_mode_tools(tools, memory_mode)
        agent_definitions = self._resolve_agent_definitions(context, tools)
    elif use_fast_tool_policy:
        with _latency_span("turn.tool_universe", policy="hard_allowlist"):
            tools = await self._get_tools_without_auto_preselection()
        tools = self._with_memory_mode_tools(tools, memory_mode)
        agent_definitions = self._resolve_agent_definitions(context, tools)
        if self._should_append_agent_tools(context):
            tools = self._with_agent_tool(
                tools,
                context=context,
                agent_definitions=agent_definitions,
            )
        active_tool_policy = build_active_tool_policy(
            profile_name=classified_profile.name,
            instruction=tool_retrieval_instruction,
            tools=tools,
            hard_active_tool_limit=context.get("hard_active_tool_limit"),
        )
        preselected_tools = [
            tool
            for tool in tools
            if getattr(tool, "name", None) in active_tool_policy.active_tool_names
        ]
    else:
        with _latency_span("turn.tool_preselection"):
            preselected_tools = await self._get_available_tools(
                tool_retrieval_instruction
            )
        with _latency_span("turn.tool_universe"):
            tools = await self._get_tool_universe(preselected_tools)
        tools = self._with_memory_mode_tools(tools, memory_mode)
        agent_definitions = self._resolve_agent_definitions(context, tools)
        if self._should_append_agent_tools(context):
            tools = self._with_agent_tool(
                tools,
                context=context,
                agent_definitions=agent_definitions,
            )
    if skills_disabled:
        preselected_tools = self._without_skill_protocol_tools(preselected_tools)
        tools = self._without_skill_protocol_tools(tools)
    coordinator = context.get("coordinator_mode") or self._coordinator_mode
    coordinator_enabled = bool(
        pre_filtered_tools is None
        and coordinator is not None
        and coordinator.is_enabled(context)
    )
    context["coordinator_mode_enabled"] = coordinator_enabled
    if coordinator_enabled:
        context["coordinator_worker_tools"] = list(tools)
        context["coordinator_worker_tools_context"] = coordinator.get_worker_tools_context(
            tools,
            context=context,
        )
        notification_queue = context.get("coordinator_notification_queue")
        if not isinstance(notification_queue, asyncio.Queue):
            notification_queue = asyncio.Queue()
            context["coordinator_notification_queue"] = notification_queue
        context.setdefault("message_source", notification_queue)
        tools = coordinator.filter_coordinator_tools(tools)
        preselected_tools = coordinator.filter_coordinator_tools(preselected_tools)
    context["agent_definitions"] = agent_definitions
    self._last_tools = tools
    tool_set_signature = self._tool_set_signature(tools)
    configured_active_tool_names = context.get("active_tool_names")
    hard_active_tool_names = (
        {str(name) for name in configured_active_tool_names if str(name)}
        if configured_active_tool_names is not None
        else None
    )
    discovered_tool_names = set(context.get("discovered_tool_names") or ())
    if session_capability_state_enabled:
        discovered_tool_names.update(
            session_capability_state.discovered_tool_names
        )
    discovered_tool_names.update(
        extract_discovered_tool_names(context.get("conversation_history") or [])
    )
    if active_tool_policy is None:
        if hard_active_tool_names is None:
            discovered_tool_names.update(
                tool.name
                for tool in preselected_tools
                if getattr(tool, "is_deferred", False)
            )
        configured_deferred_tool_names = context.get("policy_deferred_tool_names")
        if configured_deferred_tool_names is not None:
            deferred_tool_names = sorted(
                str(name)
                for name in configured_deferred_tool_names
                if str(name) and str(name) not in discovered_tool_names
            )
        else:
            deferred_tool_names = self._deferred_tool_names(
                tools,
                discovered_tool_names=discovered_tool_names,
            )
        active_tool_names = hard_active_tool_names
    else:
        deferred_tool_names = sorted(
            name
            for name in active_tool_policy.deferred_tool_names
            if name not in discovered_tool_names
        )
        active_tool_names = set(active_tool_policy.active_tool_names)
        context["active_tool_policy_reason"] = active_tool_policy.reason
        context["policy_deferred_tool_names"] = set(
            active_tool_policy.deferred_tool_names
        )
    context["deferred_tool_names"] = deferred_tool_names
    context["discovered_tool_names"] = discovered_tool_names
    context["all_tools"] = tools
    if active_tool_names is not None:
        context["active_tool_names"] = set(active_tool_names)
    active_tools = self._build_active_tools(
        tools,
        discovered_tool_names=discovered_tool_names,
        active_tool_names=active_tool_names,
        deferred_tool_names=deferred_tool_names,
    )
    context["skill_tool_available"] = bool(
        not skills_disabled and self._has_skill_tool(active_tools)
    )
    context["discover_skills_tool_available"] = bool(
        not skills_disabled and self._has_discover_skills_tool(active_tools)
    )
    marker = getattr(low_latency_profiler, "mark", None)
    if callable(marker):
        marker(
            "turn.schema_build",
            active_schema_count=len(active_tools),
            all_tools_count=len(tools),
            deferred_tools_count=len(deferred_tool_names),
        )

    preselection_debug_info = None
    if self.grounding_client and pre_filtered_tools is None:
        preselection_debug_info = self.grounding_client.get_last_preselection_debug_info()

    retrieved_tools_list = self._build_retrieved_tools_list(tools, preselection_debug_info)

    if self._recording_manager:
        from openspace.recording import RecordingManager
        await RecordingManager.record_retrieved_tools(
            task_instruction=tool_retrieval_instruction,
            tools=tools,
            preselection_debug_info=preselection_debug_info,
        )

    max_iterations = context.get("max_iterations", self._max_iterations)
    abort_event: asyncio.Event | None = context.get("abort_event")
    if abort_event is not None and not isinstance(abort_event, asyncio.Event):
        abort_event = None
    message_source = context.get("message_source")
    async_rewake_queue = context.get("async_rewake_queue")
    if not isinstance(async_rewake_queue, asyncio.Queue):
        async_rewake_queue = asyncio.Queue()
        context["async_rewake_queue"] = async_rewake_queue

    if context.get("permission_mode") == "plan" and not context.get("plan_file_path"):
        try:
            from openspace.services.runtime_support.plan_mode import get_plan_file_path

            context["plan_file_path"] = str(
                get_plan_file_path(context.get("session_id"), context.get("agent_id", "primary"))
            )
        except Exception:
            pass

    with _latency_span("turn.message_build"):
        messages = self.construct_messages(context)

    try:
        tool_use_context = self._create_tool_use_context(
            tools=active_tools,
            messages=messages,
            context=context,
        )
    except Exception as e:
        logger.error(f"Grounding Agent: Execution failed: {e}")
        await self._emit_runtime_event(
            "agent_error",
            turn_events.agent_error_payload(
                self,
                context,
                agent_id=agent_id,
                error=e,
            ),
        )
        self.increment_step()
        return {
            "error": str(e),
            "status": "error",
            "instruction": instruction,
            "iteration": 0,
        }

    self._sync_tool_use_context_runtime(
        tool_use_context,
        messages=messages,
        tools=active_tools,
        all_tools=tools,
        current_iteration=0,
        max_iterations=max_iterations,
        model=getattr(self._llm_client, "model", "unknown"),
    )
    initial_context_modifier = context.get("initial_tool_use_context_modifier")
    if callable(initial_context_modifier):
        tool_use_context = initial_context_modifier(tool_use_context)
    active_tools = self._filter_tools_for_permission_mode(active_tools, tool_use_context)
    self._sync_tool_use_context_runtime(tool_use_context, tools=active_tools)
    self._bind_agent_tools_to_context(tools, tool_use_context)
    self._bind_skill_tools_to_context(tools, tool_use_context)

    try:
        from openspace.services.tooling.hooks import (
            run_session_end_hooks,
            run_session_start_hooks,
            run_user_prompt_submit_hooks,
        )

        raw_session_start_source = context.get("session_start_source")
        should_run_session_start = (
            raw_session_start_source is not None
            or not tool_use_context.session_id
        )
        if should_run_session_start:
            session_start_source = str(raw_session_start_source or "startup")
            if session_start_source not in {"startup", "resume", "clear", "compact"}:
                session_start_source = "startup"
            if _reserve_session_start_hooks(
                self,
                session_id=(
                    tool_use_context.session_id
                    or str(context.get("task_id") or id(context))
                ),
                source=session_start_source,
            ):
                session_start_result = await run_session_start_hooks(
                    tool_use_context.hook_registry,
                    source=session_start_source,  # type: ignore[arg-type]
                    context=tool_use_context,
                    session_id=tool_use_context.session_id,
                    agent_type=tool_use_context.agent_type,
                    model=tool_use_context.model,
                )
                _append_lifecycle_hook_contexts(context, session_start_result)
                await _emit_lifecycle_hook_messages(self, session_start_result)
                with _latency_span("turn.message_build.lifecycle"):
                    messages = self.construct_messages(context)
                self._sync_tool_use_context_runtime(
                    tool_use_context,
                    messages=messages,
                )

        prompt_submit_result = await run_user_prompt_submit_hooks(
            tool_use_context.hook_registry,
            str(instruction),
            context=tool_use_context,
            permission_mode=tool_use_context.permission_mode,
        )
        _append_lifecycle_hook_contexts(context, prompt_submit_result)
        await _emit_lifecycle_hook_messages(self, prompt_submit_result)
        if (
            prompt_submit_result.blocking_errors
            or prompt_submit_result.prevent_continuation
        ):
            reason = (
                prompt_submit_result.stop_reason
                or "UserPromptSubmit hook blocked execution"
            )
            await run_session_end_hooks(
                tool_use_context.hook_registry,
                reason,
                context=tool_use_context,
            )
            self.increment_step()
            messages = list(prompt_submit_result.blocking_errors)
            if not messages:
                messages.append(
                    {
                        "role": "system",
                        "content": reason,
                        "_meta": {
                            "type": "hook_blocking_error",
                            "hook_event": "UserPromptSubmit",
                        },
                    }
                )
            return {
                "status": "blocked",
                "response": reason,
                "stop_reason": "user_prompt_submit_hook_blocked",
                "instruction": instruction,
                "iteration": 0,
                "iterations": 0,
                "tool_executions": [],
                "messages": messages,
            }
    except Exception as e:
        logger.error(f"Grounding Agent: Lifecycle hooks failed: {e}")
        await self._emit_runtime_event(
            "agent_error",
            turn_events.agent_error_payload(
                self,
                context,
                agent_id=agent_id,
                error=e,
            ),
        )
        self.increment_step()
        return {
            "error": str(e),
            "status": "error",
            "instruction": instruction,
            "iteration": 0,
        }

    with _latency_span("turn.message_build"):
        messages = self.construct_messages(context)

    self._sync_tool_use_context_runtime(tool_use_context, messages=messages)
    self._append_agent_listing_delta(messages, tool_use_context)
    if self._skill_registry:
        from openspace.skill_engine.protocol import restore_skill_state_from_messages

        restored = restore_skill_state_from_messages(messages, tool_use_context)
        if any(restored.values()):
            logger.info("Restored skill protocol state from transcript: %s", restored)
    with _latency_span("turn.skill_listing"):
        self._append_skill_listing_delta(messages, tool_use_context)
    with _latency_span("turn.skill_discovery"):
        await self._append_skill_discovery_delta_async(
            messages,
            tool_use_context,
            query=tool_retrieval_instruction,
            source="turn0_prefetch",
        )

    from openspace.recording import RecordingManager
    await RecordingManager.record_conversation_setup(
        setup_messages=copy.deepcopy(messages),
        tools=active_tools,
    )

    # ── Phase 1: Main loop (OpenSpace queryLoop while(true)) ─────────────

    state = TurnState.from_agent_context(
        self,
        context,
        str(instruction),
        tool_use_context,
        max_iterations=max_iterations,
    )
    turn_context = TurnControllerContext(
        agent=self,
        context=context,
        tool_use_context=tool_use_context,
        state=state,
        abort_event=abort_event,
        instruction=str(instruction),
        agent_id=agent_id,
        low_latency_profiler=low_latency_profiler,
        latency_span=_latency_span,
    )

    # OpenSpace query.ts starts relevant memory prefetch once per user turn before
    # the loop.  The handle is only consumed after tools finish and only if
    # it has settled, so the recall side query never blocks the main turn.
    pending_memory_prefetch = start_relevant_memory_prefetch(
        messages,
        tool_use_context,
        llm_client=self._tool_retrieval_llm or self._llm_client,
        enabled=context.get("memory_recall_enabled"),
        model=context.get("memory_recall_model"),
    )

    try:
        while state.current_iteration < state.max_iterations:
            current_iteration = state.begin_iteration()
            logger.info(
                "Grounding Agent: Iteration %s/%s",
                current_iteration,
                state.max_iterations,
            )
            self._sync_tool_use_context_runtime(
                tool_use_context,
                messages=messages,
                tools=active_tools,
                all_tools=tools,
                current_iteration=current_iteration,
                max_iterations=state.max_iterations,
                model=state.effective_model,
            )

            await self._emit(
                "iteration_start",
                turn_events.iteration_start_payload(
                    current_iteration,
                    state.max_iterations,
                ),
            )
            await self._emit_runtime_event(
                "agent_progress",
                turn_events.agent_progress_payload(
                    self,
                    context,
                    agent_id=agent_id,
                    iteration=current_iteration,
                    max_iterations=state.max_iterations,
                ),
            )

            # ── 2a. Abort check (Implementation: abortController.signal.aborted) ─
            abort_stop_reason = stop_policy.abort_stop_reason(abort_event)
            if abort_stop_reason:
                logger.info("Agent aborted by external signal before call_model")
                from openspace.services.conversation.messages import build_user_interruption_message

                messages.append(build_user_interruption_message(tool_use=False))
                state.stop_reason_final = abort_stop_reason
                break

            bench_budget_exhausted, bench_budget_reason = (
                _bench_finalize_budget_exhausted(state)
            )
            if bench_budget_exhausted:
                state.stop_reason_final = "bench_finalize_budget"
                logger.info(
                    "Bench finalize budget exhausted: %s",
                    bench_budget_reason,
                )
                break

            checker_pass_budget_exhausted, checker_pass_budget_reason = (
                _bench_checker_pass_budget_exhausted(state)
            )
            if checker_pass_budget_exhausted:
                state.stop_reason_final = "bench_checker_pass_budget"
                logger.info(
                    "Bench checker-pass budget exhausted: %s",
                    checker_pass_budget_reason,
                )
                break

            # ── 2b. Drain external messages (Implementation: getCommandsByMaxPriority) ─
            rewake_injected = await self._drain_messages(async_rewake_queue)
            if rewake_injected:
                messages.extend(rewake_injected)
                self._sync_tool_use_context_runtime(
                    tool_use_context,
                    messages=messages,
                )

            if message_source:
                injected = await self._drain_messages(message_source)
                if injected:
                    messages.extend(injected)
                    self._sync_tool_use_context_runtime(
                        tool_use_context,
                        messages=messages,
                    )

            bench_finalize_nudge = _maybe_build_bench_finalize_nudge(state)
            if bench_finalize_nudge:
                messages.append(bench_finalize_nudge)
                self._sync_tool_use_context_runtime(
                    tool_use_context,
                    messages=messages,
                )
                logger.info(
                    "Bench finalize nudge %s injected at iteration %s",
                    state.bench_finalize_nudge_count,
                    current_iteration,
                )

            # ── 2b½. Time-based microcompact (Implementation: microcompactMessages
            #    → maybeTimeBasedMicrocompact, runs BEFORE autoCompact).
            #    Pre-processing layer: content-clear old tool results when
            #    gap since last assistant exceeds threshold. No LLM call.
            mc_decision = await compaction_controller.maybe_time_based_microcompact(
                messages,
                tool_use_context,
                query_source="main_thread",
            )
            if mc_decision.was_cleared:
                messages = mc_decision.messages
                self._sync_tool_use_context_runtime(
                    tool_use_context,
                    messages=messages,
                )

            await compaction_controller.emit_token_warning(
                messages,
                tool_use_context,
                model=state.effective_model,
            )

            # ── 2c. Auto compact (Implementation: deps.autocompact) ──────────────
            compact_decision = await compaction_controller.maybe_auto_compact(
                self,
                messages,
                tool_use_context,
                model=state.effective_model,
                tracking=state.compact_tracking,
                cwd=context.get("workspace_dir"),
            )
            if compact_decision.was_compacted:
                messages = compact_decision.messages
                self._sync_tool_use_context_runtime(
                    tool_use_context,
                    messages=messages,
                )
                if compact_decision.reset_max_output_recovery:
                    state.reset_max_output_recovery()

            refresh = await tool_turn_controller.refresh_tools_for_iteration(
                turn_context,
                messages=messages,
                tools=tools,
                active_tools=active_tools,
                pre_filtered_tools=pre_filtered_tools,
                use_fast_tool_policy=use_fast_tool_policy,
                tool_set_signature=tool_set_signature,
                preselection_debug_info=preselection_debug_info,
                retrieved_tools_list=retrieved_tools_list,
            )
            tools = refresh.tools
            active_tools = refresh.active_tools
            tool_set_signature = refresh.tool_set_signature
            preselection_debug_info = refresh.preselection_debug_info
            retrieved_tools_list = refresh.retrieved_tools_list

            preparation = await tool_turn_controller.prepare_tools_for_model_call(
                turn_context,
                messages=messages,
                tools=tools,
            )
            messages = preparation.messages
            active_tools = preparation.active_tools
            messages_input_snapshot = copy.deepcopy(messages)

            model_call = await model_call_controller.call_model_with_recovery(
                turn_context,
                messages=messages,
                active_tools=active_tools,
            )
            messages = model_call.messages
            if model_call.action == "retry":
                continue
            if model_call.action == "break":
                break
            response = await model_call_controller.handle_model_response(
                turn_context,
                model_response=model_call.model_response,
                messages=messages,
            )
            messages = response.messages
            if response.action == "continue":
                continue
            if response.action == "break":
                break

            tool_turn = await tool_turn_controller.execute_tool_turn(
                turn_context,
                messages=messages,
                tools=tools,
                active_tools=active_tools,
                model_response=model_call.model_response,
                messages_input_snapshot=messages_input_snapshot,
                pending_memory_prefetch=pending_memory_prefetch,
            )
            messages = tool_turn.messages
            tool_use_context = tool_turn.tool_use_context
            turn_context.tool_use_context = tool_use_context
            active_tools = tool_turn.active_tools
            if tool_turn.action == "break":
                break

            continue  # next iteration

        # ── Phase 4: Result assembly ──────────────────────────────────

        await self._emit(
            "status_update",
            turn_events.status_update_payload(
                total_iterations=state.current_iteration,
                total_tool_calls=len(state.all_tool_results),
            ),
        )

        result = await self._build_final_result(
            instruction=instruction,
            messages=messages,
            all_tool_results=state.all_tool_results,
            iterations=state.current_iteration,
            max_iterations=state.max_iterations,
            iteration_contexts=state.iteration_contexts,
            retrieved_tools_list=retrieved_tools_list,
            preselection_debug_info=preselection_debug_info,
            stop_reason=state.stop_reason_final,
        )

        result["permission_mode"] = tool_use_context.permission_mode
        result["pre_plan_mode"] = tool_use_context.pre_plan_mode
        result["plan_file_path"] = tool_use_context.plan_file_path
        result["plan_mode_exited_in_session"] = tool_use_context.plan_mode_exited_in_session
        if session_capability_state_enabled:
            result["session_capability_state"] = (
                self._build_session_capability_state_payload(
                    context,
                    tool_use_context,
                    active_tools=active_tools,
                    profile_name=context.get("capability_profile"),
                )
            )
        schema_cache_telemetry = self._summarize_tool_schema_cache_events(
            tool_use_context
        )
        if schema_cache_telemetry:
            result["tool_schema_cache_telemetry"] = schema_cache_telemetry
            marker = getattr(low_latency_profiler, "mark", None)
            if callable(marker):
                marker("turn.schema_cache", **schema_cache_telemetry)

        if self._recording_manager:
            await self._record_agent_execution(result, instruction)

        try:
            from openspace.services.tooling.hooks import run_session_end_hooks

            await run_session_end_hooks(
                tool_use_context.hook_registry,
                state.stop_reason_final or str(result.get("status") or "completed"),
                context=tool_use_context,
            )
        except Exception:
            logger.debug("SessionEnd hooks failed", exc_info=True)

        self.increment_step()

        logger.info(
            "Grounding Agent: Execution completed: %s (reason=%s)",
            result.get("status"),
            state.stop_reason_final,
        )
        await self._emit_runtime_event(
            "agent_complete",
            turn_events.agent_complete_payload(
                self,
                context,
                agent_id=agent_id,
                result=result,
                iterations=state.current_iteration,
                tool_calls=len(state.all_tool_results),
            ),
        )
        return result

    except Exception as e:
        logger.error(f"Grounding Agent: Execution failed: {e}")
        try:
            from openspace.services.tooling.hooks import (
                run_session_end_hooks,
                run_stop_failure_hooks,
            )

            await run_stop_failure_hooks(
                tool_use_context.hook_registry,
                e,
                error_details=str(e),
                last_assistant_message=self._extract_last_assistant_message(messages),
                context=tool_use_context,
            )
            await run_session_end_hooks(
                tool_use_context.hook_registry,
                "error",
                context=tool_use_context,
            )
        except Exception:
            logger.debug("Failure lifecycle hooks failed", exc_info=True)
        await self._emit_runtime_event(
            "agent_error",
            turn_events.agent_error_payload(
                self,
                context,
                agent_id=agent_id,
                error=e,
            ),
        )
        result = {
            "error": str(e),
            "status": "error",
            "instruction": instruction,
            "iteration": state.current_iteration,
        }
        self.increment_step()
        return result

    finally:
        try:
            tool_use_context.deactivate_all_skill_scopes()
        except Exception:
            logger.debug("Failed to deactivate skill scopes", exc_info=True)
        if pending_memory_prefetch is not None:
            pending_memory_prefetch.cancel()