from __future__ import annotations
from backend.execution_context import ExecutionContext
from backend.orchestrator_agent_runtime import build_agent_runtime_helpers
from backend.orchestrator_hypothesis_runtime import build_hypothesis_runtime_helpers
def _load_main_globals() -> None:
    import backend.main as main
    globals().update({name: value for name, value in vars(main).items() if not name.startswith("__")})
async def run_orchestrator(goal: str, model: str, context: str, use_internet: bool, session: dict, websocket: WebSocket, fast_model: str = "", require_approval: bool = False, speed_mode: bool = False, use_outlook: bool = False, run_id: str = ""):
    _load_main_globals()
    _current_run_id.set(run_id)
    if use_outlook:
        session["use_outlook"] = True
    ensure_session_task_state(session, goal)
    agent_system = _compose_agent_system()
    await manager.send_message({"type": "info", "content": f"Starting AgentOrchestrator for goal: {goal}"}, websocket)
    await manager.send_message({
        "type": "status",
        "content": "Agent Orchestrator is doing its job - warming up the gears...",
    }, websocket)
    await manager.send_message({
        "type": "info",
        "content": "Agent Orchestrator is doing its job: loading memory, checking context, and choosing the leanest safe route before it commits to a plan.",
    }, websocket)
    run_settings = _load_settings()
    model = str(model or run_settings.get("main_model") or "").strip()
    fast_model = str(fast_model or run_settings.get("fast_model") or "").strip()
    if model:
        session["model"] = model
    nvidia_api_key = (run_settings.get("nvidia_api_key", "") or "").strip()
    use_google = bool(run_settings.get("google_enabled", False))
    _google_authorized = _google_get_credentials() is not None
    _gmail_available = use_google and GOOGLE_API_SUPPORT and _google_authorized
    if use_internet:
        await manager.send_message({"type": "info", "content": "Internet access: ENABLED"}, websocket)
    if speed_mode:
        await manager.send_message({"type": "info", "content": "Speed mode: ENABLED (faster planning/validation)"}, websocket)
    known_copilot_steps: list[dict] | None = None
    known_copilot_kind = ""
    def _as_int(value, default: int) -> int:
        try:
            return int(value)
        except Exception:
            return int(default)
    _policy_budget_mode = str(run_settings.get("policy_budget_mode", "budget") or "budget").strip().lower()
    _dynamic_budget_mode = _policy_budget_mode == "dynamic"
    _UNLIMITED_BUDGET_INT = 2_000_000_000
    _task_state_for_feedback = session.get("task_state") if isinstance(session.get("task_state"), dict) else None
    _policy_adjusted = apply_policy_feedback_adjustment(run_settings, task_state=_task_state_for_feedback)
    _planner_learning_block = _build_planner_learning_block(_task_state_for_feedback)
    run_budget = {
        "max_steps": _UNLIMITED_BUDGET_INT if _dynamic_budget_mode else max(1, _as_int(_policy_adjusted.get("policy_max_steps_per_run", run_settings.get("policy_max_steps_per_run", 12)), 12)),
        "max_tool_calls": _UNLIMITED_BUDGET_INT if _dynamic_budget_mode else max(1, _as_int(_policy_adjusted.get("policy_max_tool_calls_per_run", run_settings.get("policy_max_tool_calls_per_run", 30)), 30)),
        "max_validation_retries": max(0, _as_int(run_settings.get("policy_max_validation_retries_per_step", 1), 1)),
        "max_replan_rounds": max(1, _as_int(run_settings.get("policy_max_replan_rounds", 3), 3)),
        "max_total_tokens": _UNLIMITED_BUDGET_INT if _dynamic_budget_mode else max(1, _as_int(_policy_adjusted.get("policy_max_total_tokens_per_run", run_settings.get("policy_max_total_tokens_per_run", 120000)), 120000)),
        "max_runtime_seconds": _UNLIMITED_BUDGET_INT if _dynamic_budget_mode else max(10, _as_int(_policy_adjusted.get("policy_max_runtime_seconds", run_settings.get("policy_max_runtime_seconds", 900)), 900)),
        "steps_executed": 0,
        "tool_calls": 0,
        "tokens": 0,
        "started_at": time.monotonic(),
        "halted": False,
        "halt_reason": "",
        "dynamic_unlimited": _dynamic_budget_mode,
    }
    steps: list[dict] = []
    memory: list[dict] = []
    execution_context = ExecutionContext(
        session=session,
        run_budget=run_budget,
        get_steps=lambda: steps,
        get_memory=lambda: memory,
    )
    _hypothesis_runtime = initialize_hypothesis_runtime(
        run_id=run_id,
        goal=goal,
        task_state=session.get("task_state") if isinstance(session.get("task_state"), dict) else None,
        policy_mode=_policy_budget_mode,
        max_items=3,
    )
    session["hypothesis_runtime"] = _hypothesis_runtime
    session["hypothesis_summary"] = summarize_hypothesis_runtime(_hypothesis_runtime)
    log_system_event(
        "hypothesis",
        "Hypothesis runtime initialized",
        {
            "run_id": run_id,
            "active_hypothesis": str(_hypothesis_runtime.get("active_hypothesis_id") or ""),
            "policy_mode": _policy_budget_mode,
            "goal_preview": str(_hypothesis_runtime.get("goal_preview") or ""),
            "uitleg": (
                "Situation: new run started with initial working hypothesis.\n"
                "Signal: goal='{}', policy_mode='{}'.\n"
                "Decision: keep hypothesis tracking active during the run.\n"
                "Adjustment: update confidence and route per step based on tool outcomes."
            ).format(
                str(_hypothesis_runtime.get("goal_preview") or "unknown goal")[:120],
                _policy_budget_mode,
            ),
        },
    )
    _hypothesis_helpers = build_hypothesis_runtime_helpers(
        goal=goal,
        session=session,
        run_id=run_id,
        run_budget=run_budget,
        manager=manager,
        websocket=websocket,
        _hypothesis_runtime=_hypothesis_runtime,
        _policy_budget_mode=_policy_budget_mode,
        get_memory=execution_context.get_memory,
        get_steps=execution_context.get_steps,
        get_blocked_tools=lambda: blocked_tools,
        get_disabled_tools=lambda: disabled_tools,
        get_approval_tools=lambda: approval_tools,
        use_internet=use_internet,
    )
    _record_hypothesis_evidence = _hypothesis_helpers["record_hypothesis_evidence"]
    _apply_auto_remediation = _hypothesis_helpers["apply_auto_remediation"]
    _set_run_kpis_snapshot = _hypothesis_helpers["set_run_kpis_snapshot"]
    _persist_learning_feedback = _hypothesis_helpers["persist_learning_feedback"]
    _halt_run = _hypothesis_helpers["halt_run"]
    _check_runtime = _hypothesis_helpers["check_runtime"]
    _tool_policy_state = _hypothesis_helpers["tool_policy_state"]
    _authorize_tool_call = _hypothesis_helpers["authorize_tool_call"]
    auto_approved_tools = _hypothesis_helpers["auto_approved_tools"]
    if _dynamic_budget_mode:
        await manager.send_message({
            "type": "info",
            "content": "Run Policy mode: dynamic (unlimited run/calls/steps/tokens/runtime).",
        }, websocket)
    elif isinstance(_policy_adjusted.get("profile"), dict) and _policy_adjusted.get("profile"):
        await manager.send_message({
            "type": "info",
            "content": "Policy feedback active: safe run-budget tuning applied based on previous runs.",
        }, websocket)
    blocked_tools = {
        _normalize_tool_name(str(t).strip())
        for t in (run_settings.get("policy_blocked_tools") or [])
        if str(t).strip()
    }
    approval_tools = {
        _normalize_tool_name(str(t).strip())
        for t in (run_settings.get("policy_require_approval_tools") or [])
        if str(t).strip()
    }
    tool_enabled_map = (run_settings.get("tool_enabled_map") or {})
    disabled_tools = {
        _normalize_tool_name(str(t).strip())
        for t, enabled in tool_enabled_map.items()
        if str(t).strip() and not bool(enabled)
    }
    steps_auto_approved = False
    # ── Load past-session memory ───────────────────────────────────────────────
    loop = asyncio.get_event_loop()
    await manager.send_message({
        "type": "info",
        "content": "Preparing context: checking recent memory, durable notes, and user-provided context so the planner does not redo work unnecessarily.",
    }, websocket)
    memory_bundle = await loop.run_in_executor(None, _recall_memory_bundle, goal)
    past_memory = str(memory_bundle.get("summary", "") or "")
    deep_memory = str(memory_bundle.get("deep_memory", "") or "")
    memory_matches = list(memory_bundle.get("matches") or [])
    wiki_bundle = {"used": False, "matches": [], "summary": ""}
    if should_retrieve_wiki_context(goal, task_state=session.get("task_state"), user_context=context):
        wiki_bundle = await loop.run_in_executor(None, retrieve_wiki_context, goal)
    wiki_memory = str(wiki_bundle.get("summary", "") or "")
    if past_memory:
        await manager.send_message({
            "type": "memory_recall",
            "count": len(memory_matches),
            "content": past_memory,
            "matches": memory_matches,
            "citations": memory_bundle.get("citations", []),
        }, websocket)
    else:
        await manager.send_message({"type": "memory_recall", "count": 0, "content": "", "matches": [], "citations": []}, websocket)
    if deep_memory:
        await manager.send_message({
            "type": "info",
            "content": "Loaded detailed memory from a closely related past session.",
        }, websocket)
    if wiki_memory:
        await manager.send_message({
            "type": "info",
            "content": f"Loaded durable user context from {len(list(wiki_bundle.get('matches') or []))} page(s).",
        }, websocket)
    def sanitize_context_for_prompt(ctx: str) -> str:
        return re.sub(r"\[base64:[^;]+;[A-Za-z0-9+/=]+\]", "[image attached]", ctx)
    def extract_images(ctx: str) -> list[str]:
        images = re.findall(r"\[base64:[^;]+;([A-Za-z0-9+/=]+)\]", ctx)
        for target in _extract_attached_image_targets(ctx):
            source = str(target.get("source") or "").strip()
            if not source or source.lower().startswith("data:"):
                continue
            try:
                from backend.tool_general_runtime import _resolve_image_source
                resolved = _resolve_image_source(source)
                if isinstance(resolved, tuple):
                    raw_bytes, _mime, _name = resolved
                    images.append(base64.b64encode(raw_bytes).decode())
            except Exception:
                continue
        return images
    safe_context = sanitize_context_for_prompt(context)
    learned_facts = _update_user_profile_from_text(f"{goal}\n\n{safe_context}")
    if learned_facts:
        await manager.send_message({
            "type": "info",
            "content": f"User profile updated: {', '.join(learned_facts[:3])}{'…' if len(learned_facts) > 3 else ''}",
        }, websocket)
    # ── Suppress stale memory/context for live-data tasks (calendar, email, search) ──
    _task_state = (session or {}).get("task_state") or {}
    # Only suppress context when the *current* goal also appears to be a live-data
    # request.  If the user asks a general follow-up (e.g. "what is X?") after an
    # email session the previous task_state has requires_live_data=True, but we
    # must NOT wipe the conversation context — the LLM needs it to answer.
    _goal_is_live_data = bool(
        _EMAIL_MARKERS_RE.search(goal)
        or _CALENDAR_MARKERS_RE.search(goal)
        or _SEARCH_VERB_RE.search(goal)
    )
    if _task_state.get("requires_live_data") and _goal_is_live_data:
        past_memory = ""
        deep_memory = ""
        # Strip old conversation history that may contain stale API results;
        # keeps the LLM from answering with outdated data instead of calling the tool.
        safe_context = ""
    context_prompt = f"\n\nAdditional Context provided by user:\n{safe_context}\n" if safe_context else ""
    _planner_additional_context = ""
    _citation_instruction = (
        "\nWhen your answer draws on information from a past session listed above, "
        "cite it inline as [\U0001f4ce DATE: brief goal] where DATE and goal match the session reference.\n"
    ) if past_memory or deep_memory else ""
    memory_preamble = f"\n\n{past_memory}\n{_citation_instruction}" if past_memory else ""
    deep_memory_preamble = f"\n\n[Deep memory recall]\n{deep_memory[:6000]}\n{_citation_instruction}" if deep_memory else ""
    wiki_memory_preamble = f"\n\n{wiki_memory[:6000]}\n" if wiki_memory else ""
    context_prompt = deep_memory_preamble + memory_preamble + wiki_memory_preamble + context_prompt
    context_images = extract_images(context)

    _agent_helpers = build_agent_runtime_helpers(
        manager=manager,
        run_id=run_id,
        websocket=websocket,
        run_settings=run_settings,
        run_budget=run_budget,
        model=model,
        fast_model=fast_model,
        nvidia_api_key=nvidia_api_key,
        context_images=context_images,
        _halt_run=_halt_run,
        _check_runtime=_check_runtime,
    )
    agent_event = _agent_helpers["agent_event"]
    handoff = _agent_helpers["handoff"]
    _clip_agent_message = _agent_helpers["clip_agent_message"]
    _visible_agent_message = _agent_helpers["visible_agent_message"]
    _step_focus = _agent_helpers["step_focus"]
    _plan_preview = _agent_helpers["plan_preview"]
    _join_reason_labels = _agent_helpers["join_reason_labels"]
    _est_tokens = _agent_helpers["est_tokens"]
    track_tokens = _agent_helpers["track_tokens"]
    call_with_vision = _agent_helpers["call_with_vision"]
    call_fast = _agent_helpers["call_fast"]
    _stream_to_ws = _agent_helpers["stream_to_ws"]
    # ── Direct tool fast-path: 1 fast LLM call to classify intent ────────────
    # Instead of hard-coded regex patterns, use one cheap LLM call to check if
    # the request is a simple read-only email/calendar lookup.  If the LLM
    # confirms a match, execute the tool directly — saving 4-6 LLM calls
    # (planner, researcher, curator, executor, validator, report-out).
    _execution_mode = "full"
    _local_folder_reply = _maybe_answer_local_folder_request(goal)
    if _local_folder_reply:
        await agent_event("planner", "done", msg="Local folder lookup – direct answer")
        await manager.send_message({"type": "plan", "content": ["Direct folder listing"]}, websocket)
        await agent_event("reportout", "done", msg="Local folder listing delivered")
        execution_context.set_result(goal, _local_folder_reply, trace=[])
        _record_session_exchange(session, goal, _local_folder_reply)
        _save_session(session)
        await manager.send_message({"type": "final_report", "content": _local_folder_reply, "goal": goal, "session_id": session.get("id"), "run_id": run_id}, websocket)
        manager.active_runs.pop(run_id, None)
        return
    _attached_folder_pdf_targets = _select_attached_folder_pdf_summary_targets(goal, safe_context)
    if _attached_folder_pdf_targets and not known_copilot_kind and not _goal_requests_side_effect(goal):
        _extract_tool_name = "extract_source"
        if bool(_tool_policy_state(_extract_tool_name).get("allowed")):
            _remaining_tool_budget = execution_context.remaining_tool_calls()
            _attached_pdf_files = list(_attached_folder_pdf_targets.get("files") or [])
            if _remaining_tool_budget <= 0:
                await _halt_run(
                    "BUDGET_TOOL_CALLS",
                    f"Run stopped: tool-call budget exceeded ({run_budget['tool_calls']} / {run_budget['max_tool_calls']}).",
                )
                return
            _attached_pdf_files = _attached_pdf_files[:_remaining_tool_budget]
            _tool_trace: list[dict[str, object]] = []
            _summaries: list[dict[str, str]] = []
            _execution_mode = "simple"
            await manager.send_message({"type": "plan", "content": ["Extract and summarize attached folder PDFs"]}, websocket)
            await agent_event("planner", "done", msg="Attached folder PDF summary – direct answer")
            for step_index, target in enumerate(_attached_pdf_files):
                file_name = str(target.get("name") or "attached pdf").strip() or "attached pdf"
                source_path = str(target.get("full_path") or "").strip()
                if not source_path:
                    continue
                tool_args = {"source": source_path, "max_chars": 24000}
                execution_context.consume_tool_call()
                await manager.send_message(
                    {"type": "tool_call", "tool": _extract_tool_name, "args": tool_args, "step_index": step_index},
                    websocket,
                )
                _tool_result = await loop.run_in_executor(None, _execute_tool, _extract_tool_name, tool_args)
                _tool_envelope = build_tool_result_envelope(
                    f"{goal}\n\nFocus on {file_name}.",
                    f"{goal}\n\nFocus on {file_name}.",
                    _extract_tool_name,
                    str(_tool_result or ""),
                    str(_tool_result or ""),
                )
                _tool_details = dict(_tool_envelope.get("details") or {})
                _tool_result_for_context = str(_tool_envelope.get("context_result") or "")
                await manager.send_message(
                    {"type": "tool_result", "tool": _extract_tool_name, "result": _tool_result_ui_snippet(_extract_tool_name, _tool_result_for_context), "step_index": step_index},
                    websocket,
                )
                _append_workflow_context_artifact(
                    session,
                    tool_name=_extract_tool_name,
                    tool_args=tool_args,
                    tool_result=str(_tool_result or ""),
                    tool_details=_tool_details,
                    stage="direct",
                )
                _summary_reply = await _maybe_summarize_extracted_source_for_message(
                    f"{goal}\n\nFocus on {file_name}.",
                    str(_tool_result or ""),
                    settings=run_settings,
                    session=session,
                    preferred_model=model,
                    used_model=fast_model,
                )
                _fallback_summary = str(_tool_envelope.get("rendered_result") or "").strip()
                _summary_text = str(_summary_reply or _fallback_summary or "").strip()
                if _summary_text:
                    _summaries.append({"name": file_name, "summary": _summary_text})
                _tool_trace.append({
                    "tool_name": _extract_tool_name,
                    "tool_result": str(_tool_result or ""),
                    "tool_details": _tool_details,
                })
            if _summaries:
                _omitted_count = max(
                    int(_attached_folder_pdf_targets.get("omitted_count") or 0),
                    max(0, int(_attached_folder_pdf_targets.get("total_pdf_count") or 0) - len(_summaries)),
                )
                _attached_folder_reply = _render_attached_folder_pdf_summary_reply(
                    goal,
                    _summaries,
                    total_pdf_count=int(_attached_folder_pdf_targets.get("total_pdf_count") or len(_summaries)),
                    omitted_count=_omitted_count,
                )
                await agent_event("reportout", "done", msg="Attached folder PDF summary delivered")
                execution_context.set_result(goal, _attached_folder_reply, trace=_tool_trace)
                _record_session_exchange(session, goal, _attached_folder_reply)
                _save_session(session)
                await manager.send_message({"type": "final_report", "content": _attached_folder_reply, "goal": goal, "session_id": session.get("id"), "run_id": run_id}, websocket)
                manager.active_runs.pop(run_id, None)
                return
    _attached_context_target = _select_attached_context_summary_payload(goal, safe_context)
    if _attached_context_target and not known_copilot_kind and not _goal_requests_side_effect(goal):
        _execution_mode = "simple"
        _attached_names = [str(name or "").strip() for name in (_attached_context_target.get("names") or []) if str(name or "").strip()]
        _attached_payload = str(_attached_context_target.get("content") or "").strip()
        _attached_reply = ""
        _reply_candidates = _chat_followup_model_candidates(
            run_settings,
            session,
            preferred_model=model,
            used_model=fast_model,
        )
        if _reply_candidates and _attached_payload:
            _summary_prompt = (
                "Answer the user's request using only the attached file content below.\n"
                "Treat the attached content as the authoritative source for this answer.\n"
                "Be concise and factual. If the attached content is incomplete, say so.\n"
                "Do not mention workspace search, tool calls, or missing matches.\n\n"
                f"[USER REQUEST]\n{goal}\n\n"
                f"[ATTACHED FILE CONTENT]\n{_attached_payload[:18000]}\n"
            )
            try:
                _raw_reply, _ = await _call_text_model_candidates(
                    _summary_prompt,
                    _reply_candidates,
                    system=(
                        f"{_current_date_line()} "
                        "You summarize attached file content only. "
                        "Return a direct answer with no markdown boilerplate."
                    ),
                    num_ctx=16384,
                    thinking=False,
                    per_model_timeout=_CHAT_FOLLOWUP_PER_MODEL_TIMEOUT_SECONDS,
                )
                _clean_reply = re.sub(r'<think>[\s\S]*?</think>', '', str(_raw_reply or '')).strip()
                if _clean_reply and "TOOL_CALL" not in _clean_reply and '"tool"' not in _clean_reply:
                    _attached_reply = _clean_reply
            except Exception:
                _attached_reply = ""
        if not _attached_reply:
            _excerpt = _attached_payload[:1600].strip()
            if len(_attached_payload) > 1600:
                _excerpt = _excerpt.rstrip() + "…"
            if len(_attached_names) == 1:
                _attached_reply = (
                    f"Ik heb de inhoud van {_attached_names[0]} uit de meegestuurde context gehaald.\n\n{_excerpt}"
                    if _looks_dutch_text(goal)
                    else f"I used the attached content from {_attached_names[0]}.\n\n{_excerpt}"
                )
            else:
                _attached_reply = (
                    f"Ik heb de meegestuurde bestanden samengevat op basis van hun inhoud.\n\n{_excerpt}"
                    if _looks_dutch_text(goal)
                    else f"I summarized the attached files from their provided content.\n\n{_excerpt}"
                )
        await agent_event("planner", "done", msg="Attached file summary – direct answer")
        await manager.send_message({"type": "plan", "content": ["Summarize attached file context"]}, websocket)
        await agent_event("reportout", "done", msg="Attached file summary delivered")
        execution_context.set_result(goal, _attached_reply, trace=[])
        _record_session_exchange(session, goal, _attached_reply)
        _save_session(session)
        await manager.send_message({"type": "final_report", "content": _attached_reply, "goal": goal, "session_id": session.get("id"), "run_id": run_id}, websocket)
        manager.active_runs.pop(run_id, None)
        return
    # ── LLM Intent Classifier ────────────────────────────────────────────
    # One cheap LLM call replaces the chain of regex gates that used to
    # decide domain, action, and execution mode.  The result drives all
    # downstream routing: email/calendar fast-paths, direct acceleration,
    # and the complexity gate.
    _avail_integrations: list[str] = []
    if use_outlook and OUTLOOK_SUPPORT:
        _avail_integrations.append("Outlook (email + calendar)")
    if _gmail_available:
        _avail_integrations.append("Gmail / Google Calendar")
    if use_internet:
        _avail_integrations.append("Web search")
    _intent_cl: dict = _default_intent_classifier_payload()
    if not known_copilot_kind:
        _cl_prompt = (
            f"User request: {goal}\n"
            f"Available integrations: {', '.join(_avail_integrations) or 'none'}\n"
            f"Current date/time: {datetime.now().strftime('%Y-%m-%d %H:%M %A')}\n\n"
            "Classify this request. Respond with ONLY a JSON object:\n"
            "{\n"
            '  "domain": "email" | "calendar" | "search" | "code" | "general",\n'
            '  "subdomain": "filesystem" | "workspace" | "web" | "browser" | "mailbox" | "schedule" | "workspace" | "chat",\n'
            '  "goal_type": "short_snake_case_goal",\n'
            '  "action": "read" | "create" | "update" | "delete" | "send" | "none",\n'
            '  "confidence": 0.0-1.0,\n'
            '  "candidate_steps": ["short_step_id", "short_step_id"],\n'
            '  "parameters": {\n'
            '    "search_query": "optional search terms",\n'
            '    "target_summary": "which item to act on (event title, email subject, ordinal like nr 1)",\n'
            '    "new_summary": "new title/subject if renaming",\n'
            '    "new_start": "ISO datetime if time change requested",\n'
            '    "new_end": "ISO datetime if end time given or inferred",\n'
            '    "location": "new location if given",\n'
            '    "recipient": "email recipient if sending",\n'
            '    "subject": "email subject if sending",\n'
            '    "provider_scope": "default | outlook_only | gmail_only"\n'
            "  }\n"
            "}\n\n"
            "Rules:\n"
            "- domain='email' for inbox/search/send/reply/forward/delete mail.\n"
            "- domain='calendar' for agenda/appointments/events/meetings (read, create, update, delete).\n"
            "- domain='search' for web search, workspace search, file search.\n"
            "- domain='code' for code-related tasks (fix, build, implement, refactor).\n"
            "- domain='general' for everything else (chat, knowledge questions, weather, etc.).\n"
            "- Prefer subdomain='filesystem' for local folders/files/downloads/images, 'workspace' for repo/code/project search, 'web' for search-engine lookup, and 'browser' for explicit page navigation/inspection.\n"
            "- goal_type should describe the user's actual outcome, e.g. inspect_local_files, search_workspace, lookup_email, create_calendar_event.\n"
            "- candidate_steps should be 1-4 short planner step ids that describe the likely execution path.\n"
            "- action='read' for lookups/searches. action='none' for general chat.\n"
            "- action='update' for move/reschedule/change/edit requests.\n"
            "- action='send' for sending emails/messages.\n"
            "- Translate any relative dates to ISO using the current date.\n"
            "- Only include parameters that are explicitly mentioned or clearly implied.\n"
            "- Omit empty/null parameter fields.\n"
        )
        try:
            _cl_raw = await call_fast(_cl_prompt, system="You are a strict intent classifier. Output ONLY valid JSON.", num_ctx=2048)
            if _cl_raw and not _is_llm_error(_cl_raw):
                _cl_parsed = _parse_model_json(_cl_raw, expect="object")
                if isinstance(_cl_parsed, dict) and _cl_parsed.get("domain"):
                    _intent_cl = _normalize_intent_classifier_payload(_cl_parsed, goal)
        except Exception:
            pass  # On failure, _intent_cl stays at defaults — regex fallback below
    # Fallback: when LLM classifier failed or returned low-confidence defaults,
    # use regex heuristics as hint providers so routing still works.
    if _intent_cl["confidence"] < 0.1 and not known_copilot_kind:
        if _goal_requests_side_effect(goal):
            if _CALENDAR_MARKERS_RE.search(goal):
                _intent_cl["domain"] = "calendar"
                _intent_cl["action"] = _calendar_action_type(goal) or "update"
            elif _email_action_type(goal):
                _intent_cl["domain"] = "email"
                _intent_cl["action"] = _email_action_type(goal)
            else:
                _intent_cl["action"] = "update"
            _intent_cl["confidence"] = 0.3
        _intent_cl = _normalize_intent_classifier_payload(_intent_cl, goal)
    _stored_followup_task = session.get("task_state") if isinstance(session.get("task_state"), dict) else None
    if _should_preserve_live_lookup_followup(
        _stored_followup_task,
        goal,
        domain_hint=_intent_cl.get("domain", ""),
        domain_confidence=float(_intent_cl.get("confidence") or 0.0),
    ):
        _preserved_domain = _lookup_domain_for_task_state(_stored_followup_task)
        if _preserved_domain:
            _intent_cl["domain"] = _preserved_domain
            _intent_cl["action"] = "read"
            _intent_cl["confidence"] = max(float(_intent_cl.get("confidence") or 0.0), 0.95)
            _intent_cl = _normalize_intent_classifier_payload(_intent_cl, goal)
    _cl_is_action = _intent_cl["action"] not in ("read", "none")
    _cl_is_calendar = _intent_cl["domain"] == "calendar"
    _cl_is_email = _intent_cl["domain"] == "email"
    _defer_action_fast_paths_for_multi_goal = _should_defer_action_fast_paths_for_multi_goal(goal)
    _forced_extract_request = detect_extract_source_request(goal)
    if _forced_extract_request and not _cl_is_action:
        _execution_mode = "simple"
        await agent_event("planner", "done", msg="Source summary – single-step execution")
    _route_task_state = session.get("task_state") if isinstance(session.get("task_state"), dict) else None
    _route_email_action_request = bool(_cl_is_email and _cl_is_action)
    _route_calendar_action_request = bool(_cl_is_calendar and _cl_is_action)
    # Only pass the stored task_state to the protocol router when the LLM
    # classifier agrees the current goal is in the same domain.  When the user
    # switches to a general question (domain='general'), the stale email/calendar
    # task_state must not silently hijack the route to email_read/calendar_read.
    _cl_domain = _intent_cl.get("domain", "general")
    _route_task_state_for_protocol = scope_task_state_to_domain(_route_task_state, _cl_domain)
    _protocol_route = resolve_protocol_route(
        goal,
        task_state=_route_task_state_for_protocol,
        email_action_request=_route_email_action_request,
        calendar_action_request=_route_calendar_action_request,
        forced_extract=bool(_forced_extract_request),
    )
    session["intent_classifier"] = dict(_intent_cl)
    session["protocol_route"] = _protocol_route.to_dict()
    memory: list[dict] = []
    async def _execute_validated_graphs(
        graphs: list,
        *,
        allow_side_effects: bool,
        step_index_base: int,
    ) -> list[tuple[str, str]]:
        results: list[tuple[str, str]] = []
        next_step_index = step_index_base
        for graph in graphs:
            _store_task_graph_with_validation(
                session,
                graph,
                blocked_tools=blocked_tools,
                disabled_tools=disabled_tools,
                approval_tools=approval_tools,
                max_tasks=run_budget["max_tool_calls"],
                max_depth=run_budget["max_steps"],
                allow_side_effects=allow_side_effects,
            )
            graph_results = await _execute_task_graph(graph, websocket=websocket, run_id=run_id, step_index_base=next_step_index)
            results.extend(graph_results)
            next_step_index += len(graph.tasks)
        return results
    async def _finalize_action_clarification(
        content: str,
        *,
        trace: list[dict] | None = None,
        record_store: str = "",
    ) -> bool:
        compacted_trace = list(trace or [])
        if compacted_trace:
            execution_context.set_trace(compacted_trace)
            for _item in compacted_trace:
                if isinstance(_item, dict):
                    _record_hypothesis_evidence(
                        -1,
                        str(_item.get("tool_name") or _item.get("tool") or ""),
                        str(_item.get("tool_result") or _item.get("result") or ""),
                        dict(_item.get("tool_details") or {}),
                    )
                    _append_workflow_context_artifact(
                        session,
                        tool_name=str(_item.get("tool_name") or _item.get("tool") or ""),
                        tool_args={},
                        tool_result=str(_item.get("tool_result") or _item.get("result") or ""),
                        tool_details=dict(_item.get("tool_details") or {}),
                        stage="finalize",
                    )
            if record_store == "email":
                _store_session_email_records(session, compacted_trace)
            elif record_store == "calendar":
                _store_session_calendar_records(session, compacted_trace)
        execution_context.set_result(goal, content, trace=None)
        _set_run_kpis_snapshot(completed=True)
        _persist_learning_feedback()
        _record_session_exchange(session, goal, content)
        _save_session(session)
        await manager.send_message({"type": "final_report", "goal": goal, "content": content}, websocket)
        manager.active_runs.pop(run_id, None)
        return True
    async def _finalize_action_result(
        final_report: str,
        trace: list[dict],
        *,
        success: bool = True,
        record_store: str = "",
    ) -> bool:
        execution_context.set_trace(trace)
        for _item in trace:
            if isinstance(_item, dict):
                _record_hypothesis_evidence(
                    -1,
                    str(_item.get("tool_name") or _item.get("tool") or ""),
                    str(_item.get("tool_result") or _item.get("result") or ""),
                    dict(_item.get("tool_details") or {}),
                )
                _append_workflow_context_artifact(
                    session,
                    tool_name=str(_item.get("tool_name") or _item.get("tool") or ""),
                    tool_args={},
                    tool_result=str(_item.get("tool_result") or _item.get("result") or ""),
                    tool_details=dict(_item.get("tool_details") or {}),
                    stage="finalize",
                )
        if record_store == "email":
            _store_session_email_records(session, trace)
        elif record_store == "calendar":
            _store_session_calendar_records(session, trace)
        step_meta = trace[-1] if trace else {"tool_name": "", "tool_result": "", "tool_details": {}}
        execution_context.set_result(goal, final_report, trace=None, step_meta=step_meta)
        _set_run_kpis_snapshot(completed=success)
        _persist_learning_feedback()
        _record_session_exchange(session, goal, final_report)
        _store_session_continuity_summary(session, goal, session.get("task_state"), trace, final_report)
        _save_session(session)
        try:
            maybe_capture_session_wiki_draft(dict(session))
        except Exception as exc:
            print(f"[wiki] draft capture failed: {exc}")
        if success:
            await manager.send_message({"type": "success", "content": "Done!"}, websocket)
        await manager.send_message({"type": "final_report", "goal": goal, "content": final_report}, websocket)
        manager.active_runs.pop(run_id, None)
        return True
    async def _maybe_execute_email_read_followup_fast_path() -> bool:
        if _goal_requests_side_effect(goal):
            return False
        read_call = _build_email_followup_read_call(
            session,
            goal,
            use_outlook=bool(use_outlook and OUTLOOK_SUPPORT),
            gmail_available=bool(_gmail_available),
        )
        if not read_call:
            semantic_candidates = [candidate for candidate in (model, fast_model) if str(candidate or "").strip()]
            semantic_email_followup = await _classify_selected_email_followup_decision(
                session,
                goal,
                semantic_candidates,
            )
            if (
                semantic_email_followup.get("targets_selected_email")
                and semantic_email_followup.get("needs_full_email_read")
                and float(semantic_email_followup.get("confidence") or 0.0) >= 0.6
            ):
                selected_record = _default_selected_email_record(session)
                if selected_record:
                    read_call = _build_exact_email_read_call(
                        session,
                        selected_record.get("message_id"),
                        selected_record.get("provider"),
                        use_outlook=bool(use_outlook and OUTLOOK_SUPPORT),
                        gmail_available=bool(_gmail_available),
                        user_message=goal,
                        force_include_attachments=bool(semantic_email_followup.get("needs_attachment_analysis")),
                    )
        if not read_call:
            return False
        tool_name, tool_args, selected_record = read_call
        step_index = 0
        await manager.send_message(
            {"type": "tool_call", "tool": tool_name, "args": tool_args, "step_index": step_index},
            websocket,
        )
        loop = asyncio.get_event_loop()
        tool_result = await loop.run_in_executor(None, _execute_tool, tool_name, tool_args)
        await manager.send_message(
            {"type": "tool_result", "tool": tool_name, "result": _tool_result_ui_snippet(tool_name, str(tool_result)), "step_index": step_index},
            websocket,
        )
        _update_selected_email_record_from_read_result(session, selected_record, str(tool_result or ""))
        summary_reply = await _maybe_summarize_full_email_read_result(
            goal,
            tool_name,
            str(tool_result or ""),
            settings=run_settings,
            session=session,
            preferred_model=model,
            used_model=fast_model,
        )
        tool_envelope = build_tool_result_envelope(
            goal,
            goal,
            tool_name,
            str(tool_result or ""),
            summary_reply or str(tool_result or ""),
        )
        final_report = (
            str(tool_envelope.get("rendered_result") or "").strip()
            if bool(tool_envelope.get("was_composed"))
            else str(tool_result or "").strip()
        )
        if not final_report:
            final_report = "I opened the selected email." if not _looks_dutch_text(goal) else "Ik heb de geselecteerde e-mail geopend."
        if not _EMAIL_OFFER_ALREADY_RE.search(final_report):
            final_report = _append_email_followup_offer(final_report, session, dutch_hint=_looks_dutch_text(goal))
        trace = _build_compacted_tool_trace([{
            "tool_name": tool_name,
            "tool_result": str(tool_result or ""),
            "tool_details": {
                "kind": "email_read",
                "provider": str(selected_record.get("provider") or "").strip(),
                "message_id": str(selected_record.get("message_id") or "").strip(),
            },
        }])
        success = not str(tool_result or "").lower().startswith((
            "outlook get-email error:",
            "gmail get-email error:",
            "email not found",
            "no message id provided",
        ))
        return await _finalize_action_result(final_report, trace, success=success, record_store="")
    async def _maybe_summarize_extracted_source_result(
        user_message: str,
        tool_result: str,
        *,
        settings: dict,
        session: dict | None,
        preferred_model: str = "",
        used_model: str = "",
    ) -> str:
        if not str(tool_result or "").strip():
            return ""
        session_dict = session if isinstance(session, dict) else {}
        model_candidates = _chat_followup_model_candidates(
            settings,
            session_dict,
            preferred_model=preferred_model,
            used_model=used_model,
        )
        if not model_candidates:
            return ""
        envelope = build_tool_result_envelope(
            user_message,
            user_message,
            "extract_source",
            str(tool_result or ""),
            str(tool_result or ""),
        )
        source_content = str(envelope.get("context_result") or "")
        prompt = (
            "Summarize the extracted source below to answer the user's request.\n"
            "Focus on the document's substance, be concise and factual, and mention limitations only when they materially affect the answer.\n"
            "Do not mention extraction markers, console errors, or technical formatting.\n\n"
            f"[USER REQUEST]\n{user_message}\n\n"
            f"[EXTRACTED SOURCE CONTENT]\n{source_content[:18000]}\n"
        )
        try:
            raw_reply, _used_model = await _call_text_model_candidates(
                prompt,
                model_candidates,
                system=_compose_agent_system(),
                num_ctx=16384,
                thinking=False,
                per_model_timeout=_CHAT_FOLLOWUP_PER_MODEL_TIMEOUT_SECONDS,
            )
            if _is_llm_error(raw_reply):
                return ""
            reply = re.sub(r'<think>[\s\S]*?</think>', '', str(raw_reply or '')).strip()
            if not reply or "TOOL_CALL" in reply or '"tool"' in reply:
                return ""
            return reply
        except Exception:
            return ""
    async def _maybe_execute_extract_source_fast_path() -> bool:
        if not _forced_extract_request:
            return False
        if _goal_requests_side_effect(goal):
            return False
        if known_copilot_kind:
            return False
        tool_name = "extract_source"
        tool_args = {
            "source": str(_forced_extract_request.get("source") or "").strip(),
            "max_chars": int(_forced_extract_request.get("max_chars", 24000) or 24000),
        }
        if not tool_args["source"]:
            return False
        if not bool(_tool_policy_state(tool_name).get("allowed")):
            return False
        if not execution_context.has_tool_budget():
            await _halt_run(
                "BUDGET_TOOL_CALLS",
                f"Run stopped: tool-call budget exceeded ({run_budget['tool_calls']} / {run_budget['max_tool_calls']}).",
            )
            return True
        step_index = 0
        execution_context.consume_tool_call()
        await manager.send_message(
            {"type": "tool_call", "tool": tool_name, "args": tool_args, "step_index": step_index},
            websocket,
        )
        loop = asyncio.get_event_loop()
        tool_result = await loop.run_in_executor(None, _execute_tool, tool_name, tool_args)
        tool_envelope = build_tool_result_envelope(
            goal,
            goal,
            tool_name,
            str(tool_result or ""),
            str(tool_result or ""),
        )
        tool_details = dict(tool_envelope.get("details") or {})
        tool_result_for_context = str(tool_envelope.get("context_result") or "")
        await manager.send_message(
            {"type": "tool_result", "tool": tool_name, "result": _tool_result_ui_snippet(tool_name, tool_result_for_context), "step_index": step_index},
            websocket,
        )
        _append_workflow_context_artifact(
            session,
            tool_name=tool_name,
            tool_args=tool_args,
            tool_result=str(tool_result or ""),
            tool_details=tool_details,
            stage="direct",
        )
        summary_reply = await _maybe_summarize_extracted_source_result(
            goal,
            str(tool_result or ""),
            settings=run_settings,
            session=session,
            preferred_model=model,
            used_model=fast_model,
        )
        fallback_summary = _tool_result_ui_snippet(tool_name, tool_result_for_context).strip()
        composed_envelope = build_tool_result_envelope(
            goal,
            goal,
            tool_name,
            str(tool_result or ""),
            summary_reply or fallback_summary,
            tool_details,
        )
        final_report = str(composed_envelope.get("rendered_result") or "").strip()
        if not final_report:
            final_report = (
                "I extracted and summarized the requested source."
                if not _looks_dutch_text(goal)
                else "Ik heb de gevraagde bron geëxtraheerd en samengevat."
            )
        trace = _build_compacted_tool_trace([{
            "tool_name": tool_name,
            "tool_result": str(tool_result or ""),
            "tool_details": tool_details,
        }])
        result_lower = str(tool_result or "").strip().lower()
        success = not result_lower.startswith((
            "extract_source error:",
            "could not extract content",
            "source not found",
            "file not found",
            "could not read source",
        ))
        return await _finalize_action_result(final_report, trace, success=success, record_store="")
    async def _maybe_execute_email_attachment_summary_lookup_fast_path() -> bool:
        _outlook_enabled_for_run = bool((use_outlook or _settings_outlook_enabled(run_settings)) and OUTLOOK_SUPPORT)
        if _goal_requests_side_effect(goal):
            return False
        if known_copilot_kind:
            return False
        _explicit_attachment_summary_request = _is_explicit_email_attachment_summary_request(goal)
        if not _explicit_attachment_summary_request and not _email_attachment_focus_requested(goal):
            return False
        if not _explicit_attachment_summary_request and not _goal_requests_personal_summary(goal, "email"):
            return False
        if not (_outlook_enabled_for_run or _gmail_available):
            return False
        ensure_session_task_state(session, goal)
        task_state = session.get("task_state") if isinstance(session.get("task_state"), dict) else {}
        if str(task_state.get("task_type") or "") != "email_lookup":
            lowered_goal = str(goal or "").strip().lower()
            if not any(token in lowered_goal for token in ("email", "e-mail", "mail", "outlook", "gmail", "inbox", "postvak")):
                return False
            provider_scope = "default"
            if "outlook" in lowered_goal and _outlook_enabled_for_run:
                provider_scope = "outlook_only"
            elif "gmail" in lowered_goal and _gmail_available:
                provider_scope = "gmail_only"
            elif _outlook_enabled_for_run and not _gmail_available:
                provider_scope = "outlook_only"
            elif _gmail_available and not _outlook_enabled_for_run:
                provider_scope = "gmail_only"
            task_state = {
                "task_type": "email_lookup",
                "provider_scope": provider_scope,
                "folder_scope": "Inbox",
                "target_entity": {
                    "kind": "subject",
                    "text": _extract_email_subject(goal),
                },
                "target_reference": {},
            }
        target_entity = dict(task_state.get("target_entity") or {})
        target_reference = dict(task_state.get("target_reference") or {})
        raw_query = str(
            _extract_email_subject(goal)
            or target_entity.get("text")
            or target_reference.get("query")
            or ""
        ).strip()
        raw_query = re.split(
            r"\b(?:en|and)\b\s+(?:geef|give|vat|summarize|samenvat|analyseer|analyze|haal|extract|lees|read)\b",
            raw_query,
            maxsplit=1,
            flags=re.IGNORECASE,
        )[0].strip(" \t\r\n.,;:!?\"'")
        raw_query = re.split(
            r"(?:[.!?]\s*|\s+)(?:lees|read|open|toon|show|haal|extract|analyseer|analyze|vat|summarize|samenvat|geef|give)\b",
            raw_query,
            maxsplit=1,
            flags=re.IGNORECASE,
        )[0].strip(" \t\r\n.,;:!?\"'")
        query = _normalize_email_search_query(raw_query)
        if not query:
            return False
        provider_scope = str(task_state.get("provider_scope") or "default").strip().lower() or "default"
        folder_scope = str(task_state.get("folder_scope") or "Inbox").strip() or "Inbox"
        reference_kind = str(target_reference.get("kind") or "").strip()
        reference_ordinal = target_reference.get("ordinal")
        reference_temporal_hint = str(target_reference.get("temporal_hint") or "").strip()
        entity_kind = str(target_entity.get("kind") or "").strip().lower()
        search_calls: list[tuple[str, dict]] = []
        if provider_scope in {"outlook_only", "default", "cross_provider"} and _outlook_enabled_for_run:
            search_calls.append((
                "outlook_search_emails",
                {"query": query, "count": 25, "folder": folder_scope},
            ))
        if provider_scope in {"gmail_only", "default", "cross_provider"} and _gmail_available:
            gmail_query = f"subject:{query}" if entity_kind == "subject" else query
            search_calls.append((
                "gmail_search_emails",
                {"query": gmail_query, "count": 25, "folder": folder_scope},
            ))
        if not search_calls:
            return False
        loop = asyncio.get_event_loop()
        trace_items: list[dict] = []
        matched_record: dict | None = None
        all_records: list[dict] = []
        def _pick_match(records: list[dict]) -> dict | None:
            if not records:
                return None
            if entity_kind == "subject":
                normalized_query = _normalize_match_text(query)
                for record in records:
                    if _normalize_match_text(str(record.get("subject") or "")) == normalized_query:
                        return record
            matches = list(_select_email_records(
                records,
                goal,
                query,
                1,
                reference_kind=reference_kind,
                reference_ordinal=reference_ordinal,
                reference_temporal_hint=reference_temporal_hint,
            ))
            return matches[0] if matches else None
        for step_index, (tool_name, tool_args) in enumerate(search_calls):
            await manager.send_message(
                {"type": "tool_call", "tool": tool_name, "args": tool_args, "step_index": step_index},
                websocket,
            )
            tool_result = await loop.run_in_executor(None, _execute_tool, tool_name, tool_args)
            await manager.send_message(
                {"type": "tool_result", "tool": tool_name, "result": _tool_result_ui_snippet(tool_name, str(tool_result)), "step_index": step_index},
                websocket,
            )
            tool_details = build_tool_result_details(tool_name, str(tool_result or ""))
            trace_items.append({
                "tool_name": tool_name,
                "tool_result": str(tool_result or ""),
                "tool_details": tool_details,
            })
            records = extract_email_lookup_records(str(tool_result or ""))
            if not records:
                continue
            provider_hint = "gmail" if tool_name.startswith("gmail_") else "outlook"
            for record in records:
                record.setdefault("provider", provider_hint)
            all_records.extend(records)
            _merge_session_email_records(session, records)
            matched_record = _pick_match(records) or matched_record or _pick_match(all_records)
            if matched_record and str(matched_record.get("message_id") or "").strip():
                break
        if not matched_record:
            matched_record = _pick_match(all_records)
        if not matched_record:
            return False
        read_call = _build_exact_email_read_call(
            session,
            matched_record.get("message_id"),
            matched_record.get("provider"),
            use_outlook=_outlook_enabled_for_run,
            gmail_available=bool(_gmail_available),
            user_message=goal,
            force_include_attachments=True,
        )
        if not read_call:
            return False
        read_tool_name, read_tool_args, selected_record = read_call
        read_step_index = len(trace_items)
        await manager.send_message(
            {"type": "tool_call", "tool": read_tool_name, "args": read_tool_args, "step_index": read_step_index},
            websocket,
        )
        read_result = await loop.run_in_executor(None, _execute_tool, read_tool_name, read_tool_args)
        await manager.send_message(
            {"type": "tool_result", "tool": read_tool_name, "result": _tool_result_ui_snippet(read_tool_name, str(read_result)), "step_index": read_step_index},
            websocket,
        )
        _update_selected_email_record_from_read_result(session, selected_record, str(read_result or ""))
        read_details = build_tool_result_details(read_tool_name, str(read_result or ""))
        trace_items.append({
            "tool_name": read_tool_name,
            "tool_result": str(read_result or ""),
            "tool_details": read_details,
        })
        summary_reply = await _maybe_summarize_full_email_read_result(
            goal,
            read_tool_name,
            str(read_result or ""),
            settings=run_settings,
            session=session,
            preferred_model=model,
            used_model=fast_model,
        )
        read_envelope = build_tool_result_envelope(
            goal,
            goal,
            read_tool_name,
            str(read_result or ""),
            summary_reply or str(read_result or ""),
            read_details,
        )
        final_report = (
            str(read_envelope.get("rendered_result") or "").strip()
            if bool(read_envelope.get("was_composed"))
            else str(summary_reply or read_result or "").strip()
        )
        if not final_report:
            final_report = (
                "I opened the selected email and summarized the attachment."
                if not _looks_dutch_text(goal)
                else "Ik heb de geselecteerde e-mail geopend en de bijlage samengevat."
            )
        if not _EMAIL_OFFER_ALREADY_RE.search(final_report):
            final_report = _append_email_followup_offer(final_report, session, dutch_hint=_looks_dutch_text(goal))
        compacted_trace = _build_compacted_tool_trace(trace_items)
        success = not str(read_result or "").lower().startswith((
            "outlook get-email error:",
            "gmail get-email error:",
            "email not found",
            "no message id provided",
        ))
        return await _finalize_action_result(final_report, compacted_trace, success=success, record_store="email")
    async def _maybe_execute_email_action_fast_path() -> bool:
        if not (_cl_is_email and _cl_is_action):
            return False
        if known_copilot_kind:
            return False
        stored_action = ""
        if isinstance(session.get("task_state"), dict) and str(session["task_state"].get("task_type") or "") == "email_action":
            stored_action = str(session["task_state"].get("operation") or session["task_state"].get("action") or "").strip().lower()
        detected_action = detect_email_action(goal)
        action = str(_intent_cl.get("action") or "").strip().lower() or detected_action or stored_action
        if action not in {"send", "delete", "update"}:
            return False
        # Safety gate: LLM classification alone is insufficient for destructive
        # email actions.  Require explicit email keywords in the user text OR a
        # confirmed email task_state from the prior turn.
        if not detected_action and not stored_action:
            return False
        if not (use_outlook and OUTLOOK_SUPPORT) and not _gmail_available:
            return False
        dutch = _looks_dutch_text(goal)
        loop = asyncio.get_event_loop()
        intent = await _extract_email_intent_with_fallback(
            goal,
            model_candidates=[
                fast_model,
                model,
                str(run_settings.get("main_model") or ""),
                str(run_settings.get("fallback_model") or ""),
            ],
            prior_task_state=session.get("task_state") if isinstance(session.get("task_state"), dict) else None,
        )
        if intent.action in {"send", "update", "delete"}:
            action = intent.action
        if action == "send":
            explicit_provider = _detect_email_action_provider(
                goal,
                use_outlook=bool(use_outlook and OUTLOOK_SUPPORT),
                gmail_available=bool(_gmail_available),
            )
            forward_request = _is_email_forward_request(goal)
            send_plan = plan_email_action_execution(
                goal,
                route=_protocol_route,
                action="send",
                intent=intent,
                session_email_records=_session_email_records(session),
                use_outlook=bool(use_outlook and OUTLOOK_SUPPORT),
                gmail_available=bool(_gmail_available),
                settings=run_settings,
                dynamic_tools=_dynamic_tools,
                mcp_cache=_mcp_get_cache(),
                explicit_provider=explicit_provider,
                is_forward_request=forward_request,
                select_email_records=_select_email_records,
                resolve_forward_send_provider=_resolve_forward_send_provider,
                email_forward_subject=_email_forward_subject,
                build_forward_email_body=_build_forward_email_body,
                dutch=dutch,
            )
            if not send_plan.write_graphs:
                clarification = send_plan.clarification
                if not clarification and intent.missing_fields:
                    clarification = _email_intent_missing_clarification(intent, dutch=dutch)
                if not clarification:
                    return False
                return await _finalize_action_clarification(clarification)
            write_graph = send_plan.write_graphs[0]
            _send_task = write_graph.tasks[-1]
            _send_tool_name = _normalize_tool_name(_send_task.tool_name)
            _send_tool_args = dict(_send_task.args)
            _send_allowed, _send_tool_result, _send_policy = await _authorize_tool_call(
                0,
                _send_tool_name,
                _send_tool_args,
                approval_override=require_approval,
                prompt_on_approval=True,
            )
            if not _send_allowed:
                _policy_status = str(_send_policy.get("status") or "")
                _hitl_msg = (
                    "E-mail niet verzonden — niet goedgekeurd door gebruiker."
                    if dutch else
                    "Email not sent — not approved by user."
                ) if _policy_status == "approval_required" else str(_send_tool_result or "").strip()
                return await _finalize_action_result(_hitl_msg, [], success=False, record_store="")
            action_results = await _execute_validated_graphs([write_graph], allow_side_effects=True, step_index_base=0)
            if not action_results:
                return False
            tool_name, tool_result = action_results[-1]
            tool_args = dict(write_graph.tasks[-1].args)
            provider = str(send_plan.metadata.get("provider") or "")
            send_success = str(tool_result or "").strip().startswith(("Email sent", "Email sent successfully"))
            tool_details = dict(send_plan.tool_details[-1] if send_plan.tool_details else {})
            tool_details["status"] = "sent" if send_success else "failed"
            trace = _build_compacted_tool_trace([{
                "tool_name": tool_name,
                "tool_result": tool_result,
                "tool_details": tool_details,
            }])
            final_report = _email_send_action_report(provider, tool_args, tool_result, dutch=dutch)
            return await _finalize_action_result(final_report, trace, success=send_success, record_store="email")
        explicit_provider = _detect_email_action_provider(
            goal,
            use_outlook=bool(use_outlook and OUTLOOK_SUPPORT),
            gmail_available=bool(_gmail_available),
        )
        folder = _detect_email_folder(goal, {**(run_settings.get("outlook_folder_profiles") or {}), **(run_settings.get("google_folder_profiles") or {})})
        if action == "update" and folder == "Inbox":
            folder = "Drafts"
        requested_count = _email_requested_count(goal, default=1)
        selection_query = str(intent.selection_query or _extract_email_selection_query(goal, action)).strip()
        update_directives = email_intent_to_update_payload(intent) if action == "update" else {}
        if intent.provider and intent.provider in {"gmail", "outlook"}:
            explicit_provider = intent.provider
        return await _execute_email_record_action_plan(
            goal=goal,
            action=action,
            intent=intent,
            route=_protocol_route,
            session=session,
            run_settings=run_settings,
            dynamic_tools=_dynamic_tools,
            mcp_cache=_mcp_get_cache(),
            use_outlook=bool(use_outlook and OUTLOOK_SUPPORT),
            gmail_available=bool(_gmail_available),
            explicit_provider=explicit_provider,
            folder=folder,
            requested_count=requested_count,
            selection_query=selection_query,
            update_directives=update_directives,
            dutch=dutch,
            blocked_tools=blocked_tools,
            disabled_tools=disabled_tools,
            approval_tools=approval_tools,
            run_budget=run_budget,
            websocket=websocket,
            run_id=run_id,
            store_task_graph_with_validation=_store_task_graph_with_validation,
            execute_task_graph=_execute_task_graph,
            execute_validated_graphs=_execute_validated_graphs,
            build_compacted_tool_trace=_build_compacted_tool_trace,
            finalize_action_clarification=_finalize_action_clarification,
            finalize_action_result=_finalize_action_result,
        )
    if await _maybe_execute_extract_source_fast_path():
        return
    if await _maybe_execute_email_attachment_summary_lookup_fast_path():
        return
    if await _maybe_execute_email_read_followup_fast_path():
        return
    if not _defer_action_fast_paths_for_multi_goal and await _maybe_execute_email_action_fast_path():
        return
    async def _maybe_execute_calendar_action_fast_path() -> bool:
        if not (_cl_is_calendar and _cl_is_action):
            return False
        if not (use_outlook and OUTLOOK_SUPPORT) and not _gmail_available:
            return False
        action = _intent_cl.get("action", "none")
        # Map classifier verbs to the action verbs expected downstream
        _cal_action_map = {"update": "update", "create": "create", "delete": "delete", "send": "create"}
        action = _cal_action_map.get(action, "")
        if action not in {"create", "update", "delete"}:
            return False
        dutch = _looks_dutch_text(goal)
        provider_scope = _detect_calendar_provider_scope(goal, use_outlook=bool(use_outlook and OUTLOOK_SUPPORT), gmail_available=bool(_gmail_available))
        providers: list[str] = []
        if provider_scope == "outlook_only":
            providers = ["outlook"] if use_outlook and OUTLOOK_SUPPORT else []
        elif provider_scope == "gmail_only":
            providers = ["gmail"] if _gmail_available else []
        elif provider_scope == "cross_provider":
            if use_outlook and OUTLOOK_SUPPORT:
                providers.append("outlook")
            if _gmail_available:
                providers.append("gmail")
        else:
            if use_outlook and OUTLOOK_SUPPORT:
                providers = ["outlook"]
            elif _gmail_available:
                providers = ["gmail"]
        if not providers:
            return False
        reference = datetime.now()
        read_results: list[tuple[str, str]] = []
        action_results: list[tuple[str, str]] = []
        intent = await _extract_calendar_intent_with_fallback(
            goal,
            reference,
            model_candidates=[
                fast_model,
                model,
                str(run_settings.get("main_model") or ""),
                str(run_settings.get("fallback_model") or ""),
            ],
            classifier_params=_intent_cl.get("parameters"),
            classifier_action=action,
        )
        current_task_state = session.get("task_state") if isinstance(session.get("task_state"), dict) else None
        if str((current_task_state or {}).get("task_type") or "") == "calendar_action":
            current_frame = build_intent_frame(current_task_state)
            if current_frame is not None and current_frame.domain == "calendar":
                intent = merge_appointment_intents(intent, appointment_intent_from_frame(current_frame))
        if intent.action in {"create", "update", "delete"}:
            action = intent.action
        days_ahead = max(14, _detect_calendar_days_ahead(goal))
        requested_count = _calendar_requested_count(goal, default=1)
        selection_query = str(intent.selection_query or _extract_calendar_selection_query(goal, action)).strip()
        directives = appointment_intent_to_update_directives(intent) if action == "update" else {"selection_query": selection_query}
        return await _execute_calendar_action_plan(
            goal=goal,
            action=action,
            intent=intent,
            providers=providers,
            route=_protocol_route,
            session=session,
            run_settings=run_settings,
            dynamic_tools=_dynamic_tools,
            mcp_cache=_mcp_get_cache(),
            use_outlook=bool(use_outlook and OUTLOOK_SUPPORT),
            gmail_available=bool(_gmail_available),
            requested_count=requested_count,
            days_ahead=days_ahead,
            selection_query=selection_query,
            directives=directives,
            dutch=dutch,
            blocked_tools=blocked_tools,
            disabled_tools=disabled_tools,
            approval_tools=approval_tools,
            run_budget=run_budget,
            websocket=websocket,
            run_id=run_id,
            store_task_graph_with_validation=_store_task_graph_with_validation,
            execute_task_graph=_execute_task_graph,
            execute_validated_graphs=_execute_validated_graphs,
            build_compacted_tool_trace=_build_compacted_tool_trace,
            finalize_action_clarification=_finalize_action_clarification,
            finalize_action_result=_finalize_action_result,
        )
    if not _defer_action_fast_paths_for_multi_goal and await _maybe_execute_calendar_action_fast_path():
        return
    _outlook_on_early = use_outlook and OUTLOOK_SUPPORT
    # Multi-source guard: if the goal references BOTH email AND calendar,
    # skip direct acceleration — it requires cross-referencing multiple data
    # sources and the orchestrator handles that properly.
    _goal_mentions_email = bool(re.search(
        r"\b(email|emails|e-mail|mail|inbox|postvak|berichten|ontvangen|verzonden)\b",
        goal, re.IGNORECASE,
    ))
    _goal_mentions_calendar = bool(re.search(
        r"\b(agenda|calendar|afspraak|afspraken|events?|meetings?|vergadering)\b",
        goal, re.IGNORECASE,
    ))
    _multi_source_goal = _goal_mentions_email and _goal_mentions_calendar
    if (
        _execution_mode != "simple"
        and (_outlook_on_early or _gmail_available)
        and not _cl_is_action
        and not _multi_source_goal
        and _protocol_route.protocol != "code_change"
    ):
        _avail_sources = []
        if _outlook_on_early:
            _avail_sources.append("Outlook")
        if _gmail_available:
            _avail_sources.append("Gmail/Google Calendar")
        # Parse the classifier output
        _stored_task_state = scope_task_state_to_domain(
            session.get("task_state") if isinstance(session.get("task_state"), dict) else None,
            _protocol_route.domain,
        )
        _direct_plan = plan_direct_execution(
            goal,
            route=_protocol_route,
            stored_task_state=_stored_task_state,
            is_action=False,
            use_outlook=_outlook_on_early,
            gmail_available=_gmail_available,
            settings=run_settings,
            dynamic_tools=_dynamic_tools,
            mcp_cache=_mcp_get_cache(),
        )
        _direct_goal = _direct_plan.goal
        _direct_display_goal = _direct_plan.display_goal
        _direct_intent = _direct_plan.intent
        _direct_calls = _direct_plan.calls
        _classify_prompt = (
            f"User request: {goal}\n\n"
            f"Today's date: {datetime.now().strftime('%Y-%m-%d')} ({datetime.now().strftime('%A')})\n"
            f"Available sources: {', '.join(_avail_sources)}\n\n"
            "Is this a simple READ-ONLY lookup request that can be handled with a direct tool call? "
            "Write operations (send, reply, forward, create, delete, move, update, change, verplaats, wijzig, draft, compose) are NOT read-only.\n\n"
            "If YES, respond with ONLY this IntentFrame JSON object (no markdown, no explanation):\n"
            "{\n"
            '  "domain": "email|calendar|search",\n'
            '  "operation": "find_received|find_sent|calendar_read|search",\n'
            '  "provider_scope": "default|outlook_only|gmail_only|cross_provider",\n'
            '  "source_scope": "default|workspace|web|google_drive|onedrive",\n'
            '  "folder_scope": "Inbox|Sent|Drafts|Deleted",\n'
            '  "result_shape": "list|count",\n'
            '  "search_query": "",\n'
            '  "target_reference": {"query": "", "kind": "", "ordinal": null, "temporal_hint": ""},\n'
            '  "time_filter": "none|today|yesterday|this_week|last_week|this_month|last_month|custom_range",\n'
            '  "since": "YYYY-MM-DD or empty",\n'
            '  "until": "YYYY-MM-DD or empty",\n'
            '  "time_label": "",\n'
            '  "days_ahead": 7\n'
            "}\n\n"
            "Rules:\n"
            "- Keep it read-only only.\n"
            "- Preserve exact temporal constraints like '2 dagen geleden' using time_filter='custom_range' with since/until.\n"
            "- Use domain='email' for inbox/list/search mail requests.\n"
            "- Use domain='calendar' for agenda/event lookups.\n"
            "- Use domain='search' for workspace/web/cloud search.\n"
            "- search_query: extract ONLY the core topic keywords the user wants to find. "
            "Strip all prepositions (van, naar, from, to, over, about), auxiliary verbs (heb, heb ik, is er, are there), "
            "pronouns (ik, mijn, my, me), and filler words. Keep ONLY the essential noun/name.\n"
            "  Examples:\n"
            '  - "heb ik een reserveringsbevestiging" → search_query="reserveringsbevestiging"\n'
            '  - "emails over het project" → search_query="project"\n'
            '  - "mail van Jan" → search_query="Jan"\n'
            '  - "heb ik emails van github gekregen?" → search_query="github"\n'
            '  - "is er iets over de factuur" → search_query="factuur"\n'
            '  - "berichten van de belastingdienst" → search_query="belastingdienst"\n'
            "  - If the user asks for ALL mail (e.g. \"welke emails heb ik vandaag\"), leave search_query empty.\n\n"
            'If NOT a simple read-only request, respond with ONLY: {"domain": "none"}\n'
        )
        _classify_raw = await call_fast(
            _classify_prompt,
            system="You are a strict intent classifier. Output ONLY valid JSON, nothing else.",
            num_ctx=2048,
        )
        if _classify_raw and not _is_llm_error(_classify_raw):
            try:
                _parsed = _parse_model_json(_classify_raw, expect="object")
                if isinstance(_parsed, dict):
                    if str(_parsed.get("domain") or "").strip().lower() != "none":
                        _classified_frame = intent_frame_from_payload(_parsed)
                        if _classified_frame is not None:
                            _parsed = intent_frame_to_direct_intent(_classified_frame) or _parsed
                    _direct_plan = plan_direct_execution(
                        goal,
                        route=_protocol_route,
                        stored_task_state=_stored_task_state,
                        classified_intent=_parsed,
                        is_action=False,
                        use_outlook=_outlook_on_early,
                        gmail_available=_gmail_available,
                        settings=run_settings,
                        dynamic_tools=_dynamic_tools,
                        mcp_cache=_mcp_get_cache(),
                    )
                    _direct_goal = _direct_plan.goal
                    _direct_display_goal = _direct_plan.display_goal
                    _direct_intent = _direct_plan.intent
                    _direct_calls = _direct_plan.calls
            except (json.JSONDecodeError, ValueError):
                pass  # Classification failed — keep heuristic fallback if available
        _direct_task_graph = None
        _direct_task_graph_validation = None
        _graph_route = _direct_plan.graph_route
        _direct_task_graph = _direct_plan.task_graph
        if _direct_task_graph is not None:
            _direct_task_graph_validation = _store_task_graph_with_validation(
                session,
                _direct_task_graph,
                blocked_tools=blocked_tools,
                disabled_tools=disabled_tools,
                approval_tools=approval_tools,
                max_tasks=run_budget["max_tool_calls"],
                max_depth=run_budget["max_steps"],
                allow_side_effects=False,
            )
        if _direct_intent is not None and known_copilot_steps is None:
            _direct_calls = _apply_calendar_range_guard(_direct_calls, goal)
            # Respect policy: blocked / disabled / approval-required → fall through
            _direct_blocked = False
            if _direct_task_graph_validation and not _direct_task_graph_validation.valid:
                _direct_blocked = True
            for _direct_call in _direct_calls:
                _dt_name = _direct_call["tool"]
                if not bool(_tool_policy_state(_dt_name).get("allowed")):
                    _direct_blocked = True
                    break
            if _direct_calls and not _direct_blocked:
                _reused_lookup = _maybe_reuse_no_progress_lookup(session, _direct_goal, _stored_task_state, _direct_calls)
                if _reused_lookup:
                    final_report = str(_reused_lookup.get("final_report") or "").strip()
                    await manager.send_message(
                        {"type": "status", "content": "No-progress guard — reusing previous lookup result."},
                        websocket,
                    )
                    _record_session_exchange(session, _direct_goal, final_report)
                    execution_context.set_result(
                        _direct_goal,
                        final_report,
                        completed=True,
                        replace_steps=False,
                    )
                    if _stored_task_state:
                        session["task_state"] = _stored_task_state
                        session["goal"] = _direct_goal
                        session["display_goal"] = _direct_display_goal or _direct_goal
                    _store_session_continuity_summary(
                        session,
                        _direct_goal,
                        session.get("task_state") or _stored_task_state,
                        list(_reused_lookup.get("tool_trace") or []),
                        final_report,
                    )
                    _save_session(session)
                    await manager.send_message({"type": "success", "content": "Done!"}, websocket)
                    await manager.send_message(
                        {"type": "final_report", "goal": _direct_goal, "content": final_report}, websocket,
                    )
                    manager.active_runs.pop(run_id, None)
                    return
                await manager.send_message(
                    {"type": "status", "content": "Direct execution — fast path."}, websocket,
                )
                await manager.send_message({"type": "plan", "content": [_direct_goal]}, websocket)
                _direct_results: list[tuple[str, str]] = []
                if _direct_task_graph is not None:
                    _direct_results = await _execute_task_graph(_direct_task_graph, websocket=websocket, run_id=run_id, step_index_base=0)
                if _direct_results:
                    final_report = _format_direct_result(_direct_goal, _direct_results)
                    _domain = str(getattr(_graph_route, "domain", "") or "").strip().lower()
                    if _domain not in {"email", "calendar"}:
                        _tools_seen = {str(_tool_name or "").strip() for _tool_name, _ in _direct_results}
                        if _tools_seen & _EMAIL_LOOKUP_TOOL_NAMES:
                            _domain = "email"
                        elif _tools_seen & {"outlook_read_calendar", "google_read_calendar"}:
                            _domain = "calendar"
                    if _domain in {"email", "calendar"} and _goal_requests_personal_summary(goal, _domain):
                        _summary_norm = _normalize_tool_name("summarize_personal_items")
                        _summary_allowed = bool(_tool_policy_state(_summary_norm).get("allowed")) and run_budget["tool_calls"] < run_budget["max_tool_calls"]
                        if _summary_allowed:
                            _summary_records: list[dict] = []
                            for _tool_name, _tool_result in _direct_results:
                                if _domain == "email" and _tool_name in _EMAIL_LOOKUP_TOOL_NAMES:
                                    _summary_records.extend(extract_email_lookup_records(_tool_result))
                                if _domain == "calendar" and _tool_name in {"outlook_read_calendar", "google_read_calendar"}:
                                    _summary_records.extend(extract_calendar_lookup_records(_tool_result))
                            if _summary_records:
                                _summary_args = {
                                    "source_type": _domain,
                                    "items": _summary_records,
                                    "max_items": 6,
                                    "style": "brief",
                                }
                                execution_context.consume_tool_call()
                                await manager.send_message(
                                    {"type": "tool_call", "tool": "summarize_personal_items", "args": _summary_args, "step_index": len(_direct_results)},
                                    websocket,
                                )
                                _loop = asyncio.get_event_loop()
                                _summary_result = await _loop.run_in_executor(None, _execute_tool, "summarize_personal_items", _summary_args)
                                _summary_details = build_tool_result_details("summarize_personal_items", _summary_result)
                                _summary_compact = compact_tool_result_for_context("summarize_personal_items", _summary_result, _summary_details)
                                await manager.send_message(
                                    {"type": "tool_result", "tool": "summarize_personal_items", "result": _tool_result_ui_snippet("summarize_personal_items", _summary_compact), "step_index": len(_direct_results)},
                                    websocket,
                                )
                                _direct_results.append(("summarize_personal_items", _summary_result))
                                _parsed_summary = None
                                try:
                                    _parsed_summary = json.loads(_summary_result)
                                except Exception:
                                    _parsed_summary = None
                                if isinstance(_parsed_summary, dict):
                                    _summary_msg = str(_parsed_summary.get("message") or "").strip()
                                    if _summary_msg:
                                        final_report = _summary_msg
                    final_report = compose_direct_results(_direct_goal, _direct_results, final_report)
                    execution_context.set_trace(_build_compacted_tool_trace(_direct_results))
                    for _item in session["last_tool_trace"]:
                        if isinstance(_item, dict):
                            _append_workflow_context_artifact(
                                session,
                                tool_name=str(_item.get("tool_name") or _item.get("tool") or ""),
                                tool_args={},
                                tool_result=str(_item.get("tool_result") or _item.get("result") or ""),
                                tool_details=dict(_item.get("tool_details") or {}),
                                stage="direct",
                            )
                    _store_session_lookup_records_by_domain(session, _graph_route.domain, session["last_tool_trace"])
                    final_report = _append_email_followup_offer(final_report, session, dutch_hint=_looks_dutch_text(_direct_goal))
                    _direct_step_meta = session["last_tool_trace"][-1] if session["last_tool_trace"] else {"tool_name": "", "tool_result": "", "tool_details": {}}
                    execution_context.set_result(_direct_goal, final_report, trace=None, step_meta=_direct_step_meta)
                    _record_session_exchange(session, _direct_goal, final_report)
                    if _stored_task_state:
                        session["task_state"] = _stored_task_state
                        session["goal"] = _direct_goal
                        session["display_goal"] = _direct_display_goal or _direct_goal
                    session["last_lookup_signature"] = _lookup_signature(_direct_goal, session.get("task_state") or _stored_task_state, _direct_calls)
                    _store_session_continuity_summary(
                        session,
                        _direct_goal,
                        session.get("task_state") or _stored_task_state,
                        session["last_tool_trace"],
                        final_report,
                    )
                    _save_session(session)
                    await manager.send_message({"type": "success", "content": "Done!"}, websocket)
                    await manager.send_message(
                        {"type": "final_report", "goal": _direct_goal, "content": final_report}, websocket,
                    )
                    manager.active_runs.pop(run_id, None)
                    return
    # ── Internet: fetch any URLs already in context ─────────────────────────────
    if use_internet and safe_context:
        urls = re.findall(r'https?://[^\s"<>]+', safe_context)
        if urls:
            await manager.send_message({"type": "info", "content": f"Fetching {len(urls)} URL(s) from context..."}, websocket)
            loop = asyncio.get_event_loop()
            for url in urls[:3]:
                fetched = await loop.run_in_executor(None, _fetch_url, url)
                _web_page_block = f"[Web page: {url}]\n{fetched}"
                context_prompt += f"\n\n{_web_page_block}"
                _planner_additional_context += f"\n\n{_web_page_block}"
    # ── 0. Quick-answer gate ────────────────────────────────────────────────────
    _clarification_context = ""
    _execution_mode = "full"
    _wa_connected = _wa_is_connected()
    _onedrive_on = _is_onedrive_available()
    _skip_quick_gate = _protocol_route.protocol == "code_change"
    if _skip_quick_gate:
        await manager.send_message({
            "type": "info",
            "content": "Code-change request detected: skipping the quick complexity gate and going straight to planning.",
        }, websocket)
    if _forced_extract_request:
        _forced_source = str(_forced_extract_request.get("source") or "").strip()
        if _forced_source and Path(_forced_source).exists():
            _skip_quick_gate = True
    if not _skip_quick_gate and _is_explicit_email_attachment_summary_request(goal):
        _goal_lowered = str(goal or "").strip().lower()
        _outlook_enabled_for_run = bool((use_outlook or _settings_outlook_enabled(run_settings)) and OUTLOOK_SUPPORT)
        if (
            ("outlook" in _goal_lowered and _outlook_enabled_for_run)
            or ("gmail" in _goal_lowered and _gmail_available)
            or (_outlook_enabled_for_run and not _gmail_available)
            or (_gmail_available and not _outlook_enabled_for_run)
        ):
            _skip_quick_gate = True
    if known_copilot_steps is None and not _skip_quick_gate:
        await agent_event(
            "planner",
            "active",
            msg=_visible_agent_message(
                "Checking query complexity…",
                "Classifying whether this request can be answered directly, handled in one execution step, or needs a full multi-agent workflow.",
            ),
        )
        await manager.send_message({"type": "status", "content": "Checking how to handle this…"}, websocket)
        await manager.send_message({
            "type": "info",
            "content": "The orchestrator is first checking whether this can be answered directly, handled in one simple step, or whether a full workflow is actually worth the extra latency.",
        }, websocket)
        _available_tools = []
        if use_internet:  _available_tools.append("web search / internet")
        if use_outlook and OUTLOOK_SUPPORT: _available_tools.append("Outlook email & calendar")
        if _gmail_available: _available_tools.append("Gmail email & Google Calendar")
        if _wa_connected: _available_tools.append("WhatsApp messaging")
        if _gmail_available: _available_tools.append("Google Drive")
        if _onedrive_on: _available_tools.append("OneDrive")
        if PLAYWRIGHT_SUPPORT: _available_tools.append("Browser automation")
        _authorized_block = (
            f"\n\nAUTHORIZED TOOLS & CONTEXT (already granted — never ask about these):\n"
            f"- Available tools: {', '.join(_available_tools) if _available_tools else 'none'}\n"
            f"- Internet access: {'YES' if use_internet else 'NO'}\n"
            f"- Outlook email/calendar access: {'YES' if (use_outlook and OUTLOOK_SUPPORT) else 'NO'}\n"
            f"- Gmail / Google Calendar access: {'YES' if _gmail_available else 'NO'}\n"
            f"- WhatsApp messaging: {'YES' if _wa_connected else 'NO'}\n"
            f"- Google Drive: {'YES' if _gmail_available else 'NO'}\n"
            f"- OneDrive: {'YES' if _onedrive_on else 'NO'}\n"
            f"- Browser automation (Playwright): {'YES' if PLAYWRIGHT_SUPPORT else 'NO'}\n"
            f"- The user has already authorized whatever tools are listed above. Do NOT ask for permission to use them.\n"
            f"- Today's date/time is already known to you. Do NOT ask about it.\n"
            f"- The user's language, preferences and profile are in your system prompt. Do NOT ask about them."
        )
        for _round in range(3):
            if not await _check_runtime("complexity_gate"):
                manager.active_runs.pop(run_id, None)
                return
            if run_budget["halted"]:
                manager.active_runs.pop(run_id, None)
                return
            _quick_prompt = (
                f"Goal: {goal}{context_prompt}{_clarification_context}"
                f"{_authorized_block}\n\n"
                "Classify how to handle this goal:\n"
                "1. DIRECT  — You can answer fully from general/factual knowledge RIGHT NOW, no tools needed.\n"
                "   NEVER use DIRECT if the goal requires live/dynamic data: emails, calendar events, inbox,\n"
                "   current news, web pages, files, or any information only a tool can retrieve.\n"
                "   Reply: DIRECT: <complete answer>\n"
                "2. SIMPLE  — Needs AT MOST ONE tool call (e.g. one web search, one email fetch, one API call)\n"
                "   OR is a straightforward lookup/summary/translation/calculation with optional single search.\n"
                "   Reply: SIMPLE\n"
                "3. NEEDS_PLAN — Genuinely complex: requires multiple distinct steps, several tool calls,\n"
                "   coordination between different data sources, or iterative reasoning across many dimensions.\n"
                "   Reply: NEEDS_PLAN\n"
                "4. CLARIFY — ONLY if the goal is genuinely ambiguous in a way that CANNOT be resolved from context\n"
                "   (e.g. refers to something specific you cannot identify at all).\n"
                "   Reply: CLARIFY: <one concise question>\n"
                "   NEVER ask about: tools, permissions, current date/time, things you can infer.\n"
                "   When in doubt between SIMPLE and NEEDS_PLAN, choose NEEDS_PLAN.\n"
                "   When in doubt between NEEDS_PLAN and CLARIFY, choose NEEDS_PLAN."
            )
            _quick_raw = await call_fast(
                _quick_prompt,
                system=agent_system + "\nReply with exactly one of: 'DIRECT: ...', 'SIMPLE', 'NEEDS_PLAN', or 'CLARIFY: ...'. No extra text. No thinking tags.",
                num_ctx=8192,
            )
            _quick_raw = (_quick_raw or "").strip()
            if re.match(r"CLARIFY\s*:", _quick_raw, re.IGNORECASE):
                _clarify_question = re.sub(r"^CLARIFY\s*:\s*", "", _quick_raw, flags=re.IGNORECASE).strip()
                _clarify_override = _quick_clarification_override(
                    goal,
                    _clarify_question,
                    use_outlook=bool((use_outlook or _settings_outlook_enabled(run_settings)) and OUTLOOK_SUPPORT),
                    gmail_available=bool(_gmail_available),
                )
                if _clarify_override:
                    _quick_raw = _clarify_override
            _live_data_keywords = re.compile(
                r"\b(email|emails|mail|inbox|postvak|outlook|gmail|calendar|agenda|afspraak|"
                r"afspraken|nieuws|news|zoek|search|bestand|file|vandaag|today|gisteren|"
                r"yesterday|ontvangen|received|stuur|send|verstuur|whatsapp|bericht|"
                r"berichten|message|messages|drive|onedrive|browser|website|pagina|navigeer|"
                r"weather|weer|forecast|voorspelling|temperature|temperatuur|stock|koers|"
                r"price|prijs|live|current|actueel|real.time)\b",
                re.IGNORECASE,
            )
            _tools_active = (
                use_internet
                or (use_outlook and OUTLOOK_SUPPORT)
                or _gmail_available
                or _wa_connected
                or _onedrive_on
                or PLAYWRIGHT_SUPPORT
            )
            _action_goal = _cl_is_action
            if (
                re.match(r"DIRECT\s*:", _quick_raw, re.IGNORECASE)
                and _tools_active
                and _live_data_keywords.search(goal)
            ):
                _quick_raw = "SIMPLE"
            if _protocol_route.protocol == "code_change" and re.match(r"(?:DIRECT\s*:|SIMPLE\b)", _quick_raw, re.IGNORECASE):
                _quick_raw = "NEEDS_PLAN"
            if _action_goal and re.match(r"(?:DIRECT\s*:|SIMPLE\b)", _quick_raw, re.IGNORECASE):
                _quick_raw = "NEEDS_PLAN"
            _browser_navigation_goal = bool(
                PLAYWRIGHT_SUPPORT and _direct_accel_module._is_explicit_browser_navigation_request(goal)
            )
            if _browser_navigation_goal and re.match(r"(?:DIRECT\s*:|SIMPLE\b)", _quick_raw, re.IGNORECASE):
                _quick_raw = "NEEDS_PLAN"
            if re.match(r"DIRECT\s*:", _quick_raw, re.IGNORECASE):
                _direct_answer = re.sub(r"^DIRECT\s*:\s*", "", _quick_raw, flags=re.IGNORECASE).strip()
                await agent_event(
                    "planner",
                    "done",
                    msg=_visible_agent_message(
                        "Direct answer – no planning needed",
                        "The request does not require live data, tool use, or multi-step coordination, so it can be answered directly.",
                    ),
                )
                await manager.send_message({"type": "plan", "content": ["Direct answer"]}, websocket)
                await agent_event(
                    "reportout",
                    "done",
                    msg=_visible_agent_message(
                        "Direct answer delivered",
                        "Returning the direct answer immediately because no downstream execution or report synthesis was needed.",
                    ),
                )
                execution_context.set_result(goal, _direct_answer, trace=[])
                _record_session_exchange(session, goal, _direct_answer)
                _save_session(session)
                await manager.send_message({"type": "success", "content": "Done!"}, websocket)
                await manager.send_message({
                    "type": "final_report", "goal": goal, "content": _direct_answer,
                }, websocket)
                manager.active_runs.pop(run_id, None)
                return
            elif re.match(r"SIMPLE\b", _quick_raw, re.IGNORECASE):
                _execution_mode = "simple"
                await agent_event(
                    "planner",
                    "done",
                    msg=_visible_agent_message(
                        "Simple query – single-step execution",
                        "This request only needs one execution step, so the workflow skips full planning and moves straight to execution.",
                    ),
                )
                break
            elif re.match(r"CLARIFY\s*:", _quick_raw, re.IGNORECASE) and _round < 2:
                _question = re.sub(r"^CLARIFY\s*:\s*", "", _quick_raw, flags=re.IGNORECASE).strip()
                await manager.send_message({"type": "clarifying_question", "question": _question}, websocket)
                execution_context.set_result(goal, _question, verdict="CLARIFY", completed=False, trace=[])
                _set_run_kpis_snapshot(completed=False)
                _persist_learning_feedback()
                _record_session_exchange(session, goal, _question)
                _save_session(session)
                await manager.send_message({"type": "success", "content": "Awaiting your input."}, websocket)
                await manager.send_message({
                    "type": "final_report", "goal": goal, "content": _question,
                }, websocket)
                manager.active_runs.pop(run_id, None)
                return
            else:
                break
    if _clarification_context:
        context_prompt += _clarification_context
        _planner_additional_context += _clarification_context
    await agent_event("planner", "idle", msg="")
    if run_budget["halted"]:
        manager.active_runs.pop(run_id, None)
        return
    # ── Resolve weekday names to concrete dates so the LLM doesn't miscalculate ──
    _goal_lower = goal.lower()
    for _wdn, _wdi in _WEEKDAY_NAMES_TO_INDEX.items():
        if re.search(rf"\b{re.escape(_wdn)}\b", _goal_lower):
            _delta = (_wdi - datetime.now().weekday()) % 7
            if _delta == 0:
                _delta = 7
            _resolved = (datetime.now() + timedelta(days=_delta)).strftime("%Y-%m-%d")
            _weekday_note = f"Note: '{_wdn}' refers to {_resolved}."
            context_prompt += f"\n\n{_weekday_note}"
            _planner_additional_context += f"\n\n{_weekday_note}"
            break
    # ── 1. Planning ────────────────────────────────────────────────────────────────
    internet_note = (
        " For any step that would benefit from current/external information, set "
        "\"needs_search\": true and provide a concise \"search_hint\" query."
        if use_internet else ""
    )
    _outlook_on = use_outlook and OUTLOOK_SUPPORT
    _outlook_scope, _gmail_scope, _source_explicit = _detect_email_source_scope(
        goal,
        use_outlook=_outlook_on,
        gmail_available=_gmail_available,
    )
    _email_providers = []
    if _outlook_scope:
        _email_providers.append("Outlook")
    if _gmail_scope:
        _email_providers.append("Gmail")
    if _email_providers:
        _prov_str = " and ".join(_email_providers)
        email_note = (
            f" {_prov_str} email/calendar access is available in this run. "
            "Do NOT ask the user to grant access, authenticate, or paste emails. "
            f"If the goal mentions email, plan actionable steps that directly use {_prov_str} tools."
        )
        if _outlook_scope and _gmail_scope and not _source_explicit:
            email_note += (
                " When the user asks for 'email' generically, use BOTH Outlook and Gmail tools "
                "to cover all mailboxes. Combine and present the results together."
            )
        elif _source_explicit:
            email_note += (
                " The user explicitly named a provider. Stay on that provider only unless the user explicitly asks "
                "to compare, combine, or search across accounts."
            )
    else:
        email_note = " Email tools are disabled for this run. If email data is required, include a step that asks user input."
    email_goal = bool(re.search(r"\b(email|emails|mail|mailbox|outlook|gmail|google mail|inbox|postvak|e-mail|mails|sent|verzonden)\b", goal, re.IGNORECASE))
    email_count_goal = bool(re.search(r"\b(hoeveel|aantal|count|how many|number of)\b", goal, re.IGNORECASE))
    today_goal = bool(re.search(r"\b(vandaag|today)\b", goal, re.IGNORECASE))
    email_folder = _detect_email_folder(goal, {**(run_settings.get("outlook_folder_profiles") or {}), **(run_settings.get("google_folder_profiles") or {})})
    max_plan_steps = 3 if speed_mode else 5
    multi_goal_todo_steps = _build_multi_goal_todo_steps(goal, use_internet=use_internet, max_steps=max_plan_steps)
    multi_goal_mode = len(multi_goal_todo_steps) > 1
    if multi_goal_mode and known_copilot_steps is None:
        steps = multi_goal_todo_steps
        await manager.send_message({"type": "status", "content": "Multi-goal detected - executing as a to-do list…"}, websocket)
        await manager.send_message({"type": "plan", "content": [step["step"] for step in steps]}, websocket)
        await agent_event(
            "planner",
            "done",
            msg=_visible_agent_message(
                f"{len(steps)} to-do items planned",
                f"Split the goal into {len(steps)} independent to-do items: {_plan_preview(steps, 3)}",
            ),
        )
    elif _execution_mode == "simple":
        _simple_step = _forced_extract_request["display_goal"] if _forced_extract_request else goal
        _needs_search = use_internet or bool(_LIVE_DATA_QUERY_RE.search(goal))
        steps: list[dict] = [{"step": _simple_step, "needs_search": _needs_search, "search_hint": goal[:100] if _needs_search else ""}]
        await manager.send_message({"type": "status", "content": "Simple query – executing directly…"}, websocket)
        await manager.send_message({
            "type": "info",
            "content": "The planner decided a full plan would be overhead here, so the workflow is collapsing this into one direct execution step.",
        }, websocket)
        await manager.send_message({"type": "plan", "content": [_simple_step]}, websocket)
        await agent_event(
            "planner",
            "done",
            msg=_visible_agent_message(
                "1 step (simple mode)",
                f"Built a single execution step for the request: {_plan_preview(steps, 1)}",
            ),
        )
    elif known_copilot_steps is not None:
        steps = known_copilot_steps
        await manager.send_message({"type": "status", "content": "Known Co-Pilot routine detected - executing predefined steps..."}, websocket)
        await agent_event(
            "planner",
            "skip",
            msg=_visible_agent_message(
                f"Known Co-Pilot routine ({len(steps)} predefined steps)",
                f"Matched a predefined routine, so the planner is reusing {len(steps)} known steps instead of generating a new plan.",
            ),
        )
        await manager.send_message({"type": "plan", "content": [step["step"] for step in steps]}, websocket)
    else:
        await manager.send_message({"type": "status", "content": "Planning steps..."}, websocket)
        await manager.send_message({
            "type": "info",
            "content": "The planner is now decomposing the request into the minimum useful steps and deciding where research can be skipped without sacrificing answer quality.",
        }, websocket)
        if not manager.active_runs.get(run_id, False):
            return
        _plan_chat_hist = _build_chat_history_context(session, max_entries=6)
        _plan_context_sections = _build_planner_context_sections(
            recent_chat=_plan_chat_hist,
            past_memory=past_memory,
            deep_memory=deep_memory,
            wiki_memory=wiki_memory,
            user_context=safe_context,
            additional_context=_planner_additional_context,
        )
        _plan_context_block = f"\n\n{_plan_context_sections}" if _plan_context_sections else ""
        _plan_learning_block = f"\n\n{_planner_learning_block}" if _planner_learning_block else ""
        _routine_planner_guide = build_copilot_routine_planner_guide().strip()
        _routine_planner_block = (
            f"Available predefined Co-Pilot routines:\n{_routine_planner_guide}\n\n"
            if _routine_planner_guide else ""
        )
        # ── Tool-aware planner: build a compact capability digest ────
        _planner_catalog = build_runtime_tool_catalog(
            run_settings,
            dynamic_tools=_dynamic_tools,
            mcp_cache=_mcp_get_cache(),
            use_outlook=_outlook_on,
            gmail_available=_gmail_available,
        )
        _planner_tool_digest = _planner_capability_digest(_planner_catalog)
        _planner_tool_block = (
            f"\n\nAvailable tools & skills (use these in your step descriptions):\n"
            f"{_planner_tool_digest}\n"
            f"You may also create new skills at runtime using create_skill if no existing tool fits.\n"
            if _planner_tool_digest else ""
        )
        plan_prompt = (
            f"Goal: {goal}{_plan_context_block}{_plan_learning_block}{_planner_tool_block}\n\n"
            "First determine the user's true intent from the goal and any conversation "
            "context above, then plan accordingly.\n"
            f"Break this goal into the MINIMUM number of logical, actionable steps needed.{internet_note}{email_note}\n"
            "Each step MUST be concrete and tool-aware: mention which tool(s) to use.\n"
            "If no existing tool fits a step, plan to create a new skill with create_skill.\n"
            f"Simple/factual goals need 1-2 steps. Complex goals need at most {max_plan_steps} steps.\n"
            f"Use AT MOST {max_plan_steps} steps total — merge steps where possible.\n"
            f"{_routine_planner_block}"
            "If this clearly matches a predefined Co-Pilot routine, you MAY instead return ONLY this JSON object:\n"
            "  {\"routine_id\": \"morning_brief|follow_up\", \"steps\": []}\n"
            "Use routine_id only when that routine is the best exact fit, including paraphrased user requests.\n"
            "Otherwise return ONLY a valid JSON array where every element is an object with keys:\n"
            "  \"step\"         (string) — what to do (mention tool names)\n"
            "  \"needs_search\" (bool)   — whether web search is needed\n"
            "  \"search_hint\"  (string) — short search query if needs_search is true, else \"\"\n"
            "No markdown, no prose. Just the JSON object or JSON array."
        )
        await agent_event(
            "planner",
            "active",
            msg=_visible_agent_message(
                "Decomposing goal into steps…",
                "Breaking the goal into the smallest executable steps and mapping each step to the tools available in this run.",
            ),
        )
        # ── Deep-thinking gate: use thinking model for complex/creative goals ──
        _complex_goal_markers = re.compile(
            r"\b(maak|create|build|genereer|generate|ontwerp|design|herplan|reschedule|"
            r"reorgani[sz]e|optimalise|optimize|analyseer|analy[sz]e|dashboard|rapport|report|"
            r"html|pagina|page|website|overview|overzicht|combineer|combine|vergelijk|compare)\b",
            re.IGNORECASE,
        )
        _use_deep_thinking = bool(
            not speed_mode
            and _execution_mode == "needs_plan"
            and _complex_goal_markers.search(goal)
        )
        plan_call = call_with_vision if _use_deep_thinking else (call_fast if speed_mode else call_with_vision)
        plan_raw = await plan_call(
            plan_prompt,
            system=agent_system + "\n\nFor this task, output ONLY a valid JSON object or JSON array. No markdown, no prose.",
            num_ctx=_model_ctx(16384),
        )
        try:
            parsed_plan = _parse_planner_output(plan_raw)
            _planner_routine_id = str(parsed_plan.get("routine_id") or "").strip().lower()
            steps = [
                {
                    "step": item["step"],
                    "needs_search": bool(item.get("needs_search", False)) and use_internet,
                    "search_hint": item.get("search_hint", ""),
                }
                for item in list(parsed_plan.get("steps") or [])[:max_plan_steps]
            ]
            if _planner_routine_id and known_copilot_steps is None:
                _planner_routine = resolve_copilot_routine_by_id(
                    _planner_routine_id,
                    outlook_enabled=_outlook_on,
                    gmail_enabled=_gmail_available,
                    outlook_folder_profiles=run_settings.get("outlook_folder_profiles") or None,
                    google_folder_profiles=run_settings.get("google_folder_profiles") or None,
                )
                if _planner_routine is not None:
                    known_copilot_kind = str(_planner_routine.get("id") or "").strip().lower()
                    known_copilot_steps = list(_planner_routine.get("steps") or [])
                    steps = known_copilot_steps
                    await manager.send_message(
                        {"type": "status", "content": "Planner selected Co-Pilot routine - executing predefined steps..."},
                        websocket,
                    )
                    await agent_event(
                        "planner",
                        "skip",
                        msg=_visible_agent_message(
                            f"Planner selected Co-Pilot routine ({len(steps)} predefined steps)",
                            f"The planner recognized this as a routine and switched to {len(steps)} predefined steps.",
                        ),
                    )
            if not steps:
                raise ValueError("Planner did not return any executable steps")
            if (_outlook_on or _gmail_available) and email_goal:
                access_re = re.compile(r"request access|grant access|provide (the )?list of emails|cannot access|set up account access", re.IGNORECASE)
                ui_re = re.compile(r"open outlook|open gmail|navigate|click|apply (a )?filter|show only emails|inbox view|sent folder|open (the )?(inbox|sent)", re.IGNORECASE)
                date_re = re.compile(r"(determine|retrieve|get|obtain|check|find out|look up)\s+(the\s+)?(current\s+)?(today.?s?\s+)?(date|time|dag|datum)", re.IGNORECASE)
                _retrieve_label = " and ".join(_email_providers)
                _folder_label = email_folder
                rewritten = []
                for s in steps:
                    if date_re.search(s["step"]):
                        pass
                    elif access_re.search(s["step"]) or ui_re.search(s["step"]):
                        rewritten.append({"step": f"Retrieve emails from {_retrieve_label} {_folder_label} for the requested period.", "needs_search": False, "search_hint": ""})
                    else:
                        rewritten.append(s)
                if _source_explicit:
                    filtered_steps = []
                    for step_obj in rewritten:
                        step_text = step_obj["step"]
                        if _step_is_cross_source(step_text):
                            continue
                        if _gmail_scope and not _outlook_scope and _step_mentions_provider(step_text, "outlook"):
                            continue
                        if _outlook_scope and not _gmail_scope and _step_mentions_provider(step_text, "gmail"):
                            continue
                        filtered_steps.append(step_obj)
                    rewritten = filtered_steps or rewritten
                if not rewritten:
                    rewritten = [{"step": f"Retrieve emails from {_retrieve_label} {_folder_label} for today.", "needs_search": False, "search_hint": ""}]
                steps = rewritten
                if email_count_goal:
                    period = "today" if today_goal else "the requested period"
                    steps = [
                        {"step": f"Fetch emails from {_retrieve_label} {_folder_label} for {period}.", "needs_search": False, "search_hint": ""},
                        {"step": f"Count the fetched emails for {period} and report the exact total.", "needs_search": False, "search_hint": ""},
                    ]
        except Exception:
            steps = _build_planner_plan_b_steps(goal, use_internet=use_internet, max_steps=max_plan_steps)
            if not steps:
                await manager.send_message({"type": "error", "content": f"Failed to parse plan. Raw output: {plan_raw}"}, websocket)
                manager.active_runs.pop(run_id, None)
                return
            session["planner_fallback"] = {
                "reason": "parse_error",
                "raw_output": str(plan_raw or "")[:4000],
                "steps": [dict(step) for step in steps],
            }
            await agent_event(
                "planner",
                "error",
                msg=_visible_agent_message(
                    "Planner output invalid - using Plan B",
                    "The planner response was not usable JSON, so the workflow is falling back to a deterministic Plan B step list.",
                ),
            )
            await manager.send_message(
                {
                    "type": "info",
                    "content": "Planner output was invalid - continuing with Plan B.",
                },
                websocket,
            )
        await agent_event(
            "planner",
            "done",
            msg=_visible_agent_message(
                f"{len(steps)} steps planned",
                f"Planned {len(steps)} steps with tool-aware actions: {_plan_preview(steps, 3)}",
            ),
        )
        _first_search = any(s["needs_search"] for s in steps)
        await handoff(
            "planner",
            "researcher" if _first_search else "curator",
            summary=f"{len(steps)} steps: " + "; ".join(s["step"][:35] for s in steps[:3]) + ("…" if len(steps) > 3 else ""),
            question=(
                f"What external evidence is needed first for: {_plan_preview(steps, 2)}?"
                if _first_search else
                f"What context should execution keep in view first for: {_plan_preview(steps, 2)}?"
            ),
        )
        await manager.send_message({"type": "plan", "content": [s["step"] for s in steps]}, websocket)
    if len(steps) > run_budget["max_steps"]:
        await manager.send_message({
            "type": "info",
            "content": f"Plan truncated to budgeted max of {run_budget['max_steps']} steps.",
        }, websocket)
        steps = steps[:run_budget["max_steps"]]
    async def _push_todo_progress(
        event: str,
        *,
        step_index: int | None = None,
        goal_status: str | None = None,
        goal_evaluation: str | None = None,
        step_fields: dict | None = None,
    ) -> None:
        if step_index is not None and step_fields:
            _update_todo_progress_item(session, steps, step_index, **step_fields)
        await _send_todo_progress(
            session,
            steps,
            websocket,
            event=event,
            run_id=run_id,
            current_step_index=step_index,
            goal_status=goal_status,
            goal_evaluation=goal_evaluation,
        )
    await _push_todo_progress(
        "plan_created",
        goal_status="running",
        goal_evaluation="Execution plan ready.",
    )
    # ── 2. Execution pipeline ───────────────────────────────────────────────────
    async def researcher_agent(step_desc: str, search_hint: str, step_index: int) -> str:
        if run_budget["halted"] or (not await _check_runtime("researcher")):
            return ""
        query = search_hint.strip() or step_desc[:120]
        await agent_event(
            "researcher",
            "active",
            step_index,
            _visible_agent_message(
                f"Searching: {query[:60]}",
                f"Looking for external evidence not already present in memory or context. Query: {query[:90]}",
            ),
        )
        await manager.send_message({"type": "info", "content": f"Researcher: searching '{query}'"}, websocket)
        loop = asyncio.get_event_loop()
        raw_results = await loop.run_in_executor(None, _web_search, query, 6)
        _NO_RESULTS_MARKERS = ("No search results found", "Brave Search error:", "Brave Search auth error:")
        _search_failed = not raw_results or any(raw_results.startswith(m) for m in _NO_RESULTS_MARKERS)
        if _search_failed:
            await manager.send_message({
                "type": "info",
                "content": f"Researcher: web search returned no usable results for '{query}'."
            }, websocket)
            await agent_event(
                "researcher",
                "done",
                step_index,
                _visible_agent_message(
                    "No results found",
                    "Web search returned no usable hits or fetchable pages for this step, so execution will continue without external evidence.",
                ),
            )
            return ""
        await manager.send_message({
            "type": "info",
            "content": f"Researcher: {len(raw_results)} chars of raw search results. Synthesising…"
        }, websocket)
        urls_found = re.findall(r"URL:\s+(https?://[^\s]+)", raw_results)
        fetched_pages = []
        if urls_found[:2]:
            page_results = await asyncio.gather(
                *[loop.run_in_executor(None, _fetch_url, url, 3000) for url in urls_found[:2]]
            )
            for url, page in zip(urls_found[:2], page_results):
                if not page.startswith("Could not fetch"):
                    fetched_pages.append(f"[From {url}]\n{page}")
        full_raw = raw_results
        if fetched_pages:
            full_raw += "\n\n" + "\n\n".join(fetched_pages)
        synth_prompt = (
            f"Goal: {goal}\n"
            f"Step that needs this research: {step_desc}\n\n"
            f"Raw search results and page extracts:\n{full_raw[:20000]}\n\n"
            "Write a concise research brief (max 400 words) that directly supports "
            "executing the step above. Include key facts, figures, and URLs. "
            "Do NOT include irrelevant information. "
            "IMPORTANT: Only state facts that are explicitly present in the search results above. "
            "If the results do not contain the answer, say so — NEVER fabricate or guess."
        )
        brief = await call_with_vision(
            synth_prompt,
            system=(
                f"{_current_date_line()} "
                "You are a research analyst. Given raw web search results, extract and "
                "synthesise only the information that is directly relevant to the given step. "
                "Be factual, concise, and cite sources (URL) where possible. "
                "NEVER invent facts. If the search results do not contain the needed "
                "information, explicitly state that no relevant data was found."
            )
        )
        if _is_llm_error(brief):
            print(f"[researcher] synthesis failed for step {step_index + 1}: {brief[:220]}")
            await manager.send_message({
                "type": "info",
                "content": "Researcher: research synthesis was unavailable, so execution will continue without a curated external brief.",
            }, websocket)
            await agent_event(
                "researcher",
                "done",
                step_index,
                _visible_agent_message(
                    "Research synthesis unavailable",
                    "Web hits were found, but the research model could not turn them into a usable brief, so execution will continue without external synthesis.",
                ),
            )
            return ""
        await agent_event(
            "researcher",
            "done",
            step_index,
            _visible_agent_message(
                f"Brief ready ({len(brief)} chars)",
                "Condensed the search hits and fetched pages into a brief focused only on facts that help execute this step.",
            ),
        )
        await manager.send_message({
            "type": "research_ready", "step_index": step_index, "brief": brief.strip(),
        }, websocket)
        return brief.strip()
    async def curator_agent(step_desc: str, step_index: int, completed: list[dict], research_brief: str) -> str:
        if run_budget["halted"] or (not await _check_runtime("curator")):
            return ""
        _curation_sources: list[str] = []
        if research_brief:
            _curation_sources.append("research brief")
        await agent_event(
            "curator",
            "active",
            step_index,
            _visible_agent_message(
                "Filtering relevant context…",
                "Filtering prior context down to only the evidence that can help this step right now.",
            ),
        )
        sections = []
        # Include recent chat history so the curator can surface conversation context
        # (skip for live-data tasks to avoid stale API results leaking in)
        if not _task_state.get("requires_live_data"):
            _chat_hist_text = _build_chat_history_context(session, max_entries=8)
            if _chat_hist_text:
                sections.append(f"[Recent conversation]\n{_chat_hist_text}")
                _curation_sources.append("recent conversation")
        if research_brief:
            sections.append(f"[Research brief]\n{research_brief[:6000]}")
        if deep_memory:
            sections.append(f"[Deep memory recall]\n{deep_memory[:6000]}")
            _curation_sources.append("deep memory")
        if past_memory:
            sections.append(f"[Past session memory]\n{past_memory[:12000]}")
            _curation_sources.append("past sessions")
        if safe_context:
            sections.append(f"[User-provided context]\n{safe_context[:8000]}")
            _curation_sources.append("user context")
        if completed:
            recent = completed[-4:]
            prev = "\n\n".join(
                f"Step {completed.index(m)+1}: {m['step']}\nResult: {(m['result'] or '')[:1500].replace(chr(10),' ')}{'…' if len(m['result'] or '')>1500 else ''}"
                for m in recent
            )
            sections.append(f"[Previous results]\n{prev}")
            _curation_sources.append("previous results")
        if not sections:
            await agent_event(
                "curator",
                "skip",
                step_index,
                _visible_agent_message(
                    "Nothing to curate yet",
                    "No useful prior context, research, or memory was available, so the executor will work from the step itself.",
                ),
            )
            return ""
        if not past_memory and not safe_context and not completed and not sections:
            await agent_event(
                "curator",
                "skip",
                step_index,
                _visible_agent_message(
                    "Nothing to curate yet",
                    "There was nothing useful to filter for this step, so no curated context will be passed forward.",
                ),
            )
            return ""
        available = "\n\n".join(sections)
        plan_overview = "\n".join(f"{j+1}. {s['step']}" for j, s in enumerate(steps))
        curation_prompt = (
            f"Goal: {goal}\n\nPlan:\n{plan_overview}\n\n"
            f"Current step ({step_index+1}/{len(steps)}): \"{step_desc}\"\n\n"
            f"Available information:\n{available}\n\n"
            "Extract ONLY what is directly useful for this step (max 250 words). "
            "If nothing is relevant, reply with exactly: NONE"
        )
        result = await call_fast(
            curation_prompt,
            system=(
                "You are a context curator. Filter available information and return only "
                "snippets directly relevant to the current step. Be brief and specific."
            )
        )
        r = result.strip()
        curated_out = "" if r.upper() == "NONE" else r
        _curation_source_label = _join_reason_labels(_curation_sources)
        await agent_event(
            "curator",
            "done",
            step_index,
            _visible_agent_message(
                "No relevant context" if not curated_out else f"{len(curated_out)} chars curated",
                (
                    "Filtered available context, but nothing materially useful remained for this step."
                    if not curated_out else
                    f"Selected only the useful evidence from {_curation_source_label or 'available context'} and passed {len(curated_out)} chars forward."
                ),
            ),
        )
        return curated_out
    # ── Tool descriptions ──────────────────────────────────────────────────
    TOOLS_DESC = _build_tools_desc(
        use_outlook=_outlook_on, gmail_available=_gmail_available,
        whatsapp_available=_wa_is_connected(),
        google_drive_available=_gmail_available,
        onedrive_available=_is_onedrive_available(),
        browser_available=PLAYWRIGHT_SUPPORT,
    )
    # Native Ollama tool-calling schemas (only used for Ollama models)
    _use_native_tools = not (model.startswith(NVIDIA_PREFIX)
                             or model.startswith(OPENAI_PREFIX)
                             or model.startswith(GEMINI_PREFIX)
                             or model.startswith(LLAMACPP_PREFIX))
    _ollama_tools: list[dict] = []
    if _use_native_tools:
        _slack_available = bool((run_settings.get("slack_bot_token") or "").strip())
        _ollama_tools = build_ollama_tool_schemas(
            use_outlook=_outlook_on, gmail_available=_gmail_available,
            whatsapp_available=_wa_is_connected(),
            google_drive_available=_gmail_available,
            onedrive_available=_is_onedrive_available(),
            browser_available=PLAYWRIGHT_SUPPORT,
            slack_available=_slack_available,
            disabled_tools=disabled_tools,
            blocked_tools=blocked_tools,
            webhooks=run_settings.get("webhooks") or [],
        )
    async def _call_ollama_with_tools(
        messages: list[dict],
        tools: list[dict],
        num_ctx: int = 32768,
    ) -> dict:
        """Call Ollama /api/chat with native tool schemas.
        Returns the full response dict with message.content and/or
        message.tool_calls.
        """
        loop = asyncio.get_event_loop()
        def _do():
            _ollama_ensure_model(model)
            payload: dict = {
                "model": model, "messages": messages,
                "stream": False,
                "options": {"num_ctx": num_ctx, "temperature": 0.7},
                "keep_alive": "10m",
                "think": True,
            }
            if tools:
                payload["tools"] = tools
            try:
                res = requests.post(f"{OLLAMA_URL}/chat", json=payload, timeout=180)
                res.raise_for_status()
                return res.json()
            except Exception as e:
                return {"error": str(e)}
        return await loop.run_in_executor(None, _do)
    def _extract_native_tool_calls(response: dict) -> list[dict]:
        """Extract tool calls from an Ollama /api/chat response.
        Returns list of {"name": str, "arguments": dict} dicts,
        or empty list if no tool calls.
        """
        msg = response.get("message") or {}
        raw_calls = msg.get("tool_calls") or []
        result = []
        for tc in raw_calls:
            fn = tc.get("function") or {}
            name = fn.get("name", "")
            arguments = fn.get("arguments") or {}
            if name:
                result.append({"name": name, "arguments": arguments})
        return result

    async def _emit_native_response_thinking(response: dict) -> str:
        """Emit native-tool-call thinking blocks and return cleaned assistant text."""
        msg = response.get("message") or {}
        content = str(msg.get("content") or "")
        blocks: list[str] = []
        raw_thinking = msg.get("thinking", "")

        if isinstance(raw_thinking, list):
            blocks.extend(str(part) for part in raw_thinking if str(part or "").strip())
        elif str(raw_thinking or "").strip():
            blocks.append(str(raw_thinking))

        if not blocks:
            blocks.extend(re.findall(r'<think>([\s\S]*?)</think>', content))

        if run_settings.get("show_thinking", False) and blocks:
            await manager.send_message({"type": "stream_thinking_start", "run_id": run_id}, websocket)
            for block in blocks:
                if str(block or "").strip():
                    await manager.send_message({"type": "stream_thinking", "content": str(block), "run_id": run_id}, websocket)
            await manager.send_message({"type": "stream_thinking_end", "run_id": run_id}, websocket)
            await asyncio.sleep(0)

        return re.sub(r'<think>[\s\S]*?</think>', '', content).strip()

    # Sub-agent C — Executor
    async def executor_agent(
        step_desc: str,
        curated_context: str,
        research_brief: str,
        feedback: str = "",
        step_index: int = -1,
        needs_search: bool = False,
        completed_steps: list[dict] | None = None,
    ) -> tuple[str, bool, dict]:
        if run_budget["halted"] or (not await _check_runtime("executor")):
            return f"[Run halted: {run_budget['halt_reason'] or 'runtime budget exceeded'}]", False, {
                "tool_name": "",
                "tool_result": "",
                "deterministic_result": False,
            }
        parts = [f"Goal: {goal}\n"]
        # Inject recent conversation so the executor understands conversational context
        _exec_chat_hist = _build_chat_history_context(session, max_entries=8)
        if _exec_chat_hist:
            parts.append(f"Recent conversation:\n{_exec_chat_hist}\n")
        # Cross-step memory: give the executor direct access to prior step
        # results so it can use IDs, dates, and other precise data from
        # earlier steps without relying solely on the curator's summary.
        if completed_steps:
            _csm_lines: list[str] = []
            for _ci, _cs in enumerate(completed_steps[-4:], start=max(1, len(completed_steps) - 3)):
                _cs_tool = str(_cs.get("tool_name") or "").strip()
                _cs_result = str(_cs.get("tool_result") or _cs.get("result") or "").strip()
                if _cs_result:
                    _csm_lines.append(
                        f"Step {_ci}: {_cs.get('step', '')[:80]}\n"
                        f"Tool: {_cs_tool or 'none'}\n"
                        f"Result:\n{_cs_result[:2000]}"
                    )
            if _csm_lines:
                parts.append(f"Prior step results (use IDs and details from these):\n{'---'.join(_csm_lines)}\n")
        if curated_context:
            parts.append(f"Relevant prior context:\n{curated_context}\n")
        if research_brief:
            parts.append(f"Research brief:\n{research_brief}\n")
        elif needs_search:
            parts.append(
                "Note: Web search was attempted but returned no usable results. "
                "Use tools (e.g. web_search) to find the information yourself, or "
                "state clearly that the information could not be verified.\n"
            )
        if feedback:
            parts.append(
                f"⚠ Your previous attempt was rejected by quality control.\n"
                f"Feedback: {feedback}\n"
                f"Please correct the issues and provide an improved answer.\n"
            )
        parts.append(
            f"Task: {step_desc}\n\n"
            "Execute this task. Use tools if needed, then give your final answer.\n"
            "IMPORTANT: First determine the user's true intent from the conversation "
            "and all available context. Base your answer on provided data (search results, "
            "email/calendar/file content, tool output) — NOT on your training knowledge "
            "alone for factual claims about current events, dates, or real-world state. "
            "If no reliable data source confirms a fact, say so explicitly."
        )
        _SIDE_EFFECT_TOOLS = {
            "outlook_send_email", "outlook_update_email", "outlook_delete_email",
            "gmail_send_email", "gmail_update_email", "gmail_delete_email",
            "outlook_create_calendar_event", "outlook_update_calendar_event",
            "outlook_delete_calendar_event",
            "google_create_calendar_event", "google_update_calendar_event",
            "google_delete_calendar_event",
            "google_drive_upload", "onedrive_upload",
            "whatsapp_send_message",
            "slack_post_message",
            "write_file", "run_shell", "run_python",
            "schedule_goal",
        }
        tool_log:           list[str] = []
        last_clean_response: str      = ""
        latest_tool_name: str         = ""
        latest_tool_args: dict        = {}
        latest_tool_result: str       = ""
        latest_tool_details: dict     = {}
        used_side_effect_tool: bool   = False
        composed_from_tool: bool      = False
        tool_call_cache: dict[str, tuple[str, dict, str]] = {}
        current_prompt = "\n".join(parts)
        current_system = agent_system + TOOLS_DESC
        _extract_request = None
        if _forced_extract_request and _execution_mode == "simple" and step_index in {-1, 0}:
            _extract_request = dict(_forced_extract_request)
        if _extract_request and not feedback:
            _extract_tool = "extract_source"
            _extract_args = {
                "source": _extract_request["source"],
                "max_chars": int(_extract_request.get("max_chars", 24000) or 24000),
            }
            if bool(_tool_policy_state(_extract_tool).get("allowed")):
                if run_budget["tool_calls"] < run_budget["max_tool_calls"]:
                    execution_context.consume_tool_call()
                    await manager.send_message({
                        "type": "tool_call", "tool": _extract_tool, "args": _extract_args, "step_index": step_index,
                    }, websocket)
                    loop = asyncio.get_event_loop()
                    tool_result = await loop.run_in_executor(None, _execute_tool, _extract_tool, _extract_args)
                    tool_details = build_tool_result_details(_extract_tool, tool_result)
                    tool_result_for_context = compact_tool_result_for_context(_extract_tool, tool_result, tool_details)
                    latest_tool_name = _extract_tool
                    latest_tool_args = _extract_args
                    latest_tool_result = tool_result
                    latest_tool_details = tool_details
                    _append_workflow_context_artifact(
                        session,
                        tool_name=_extract_tool,
                        tool_args=_extract_args,
                        tool_result=tool_result,
                        tool_details=tool_details,
                        stage="executor",
                    )
                    await manager.send_message({
                        "type": "tool_result", "tool": _extract_tool,
                        "result": _tool_result_ui_snippet(_extract_tool, tool_result_for_context), "step_index": step_index,
                    }, websocket)
                    tool_log.append(f"Tool: {_extract_tool}\nArgs: {json.dumps(_extract_args)}\nResult:\n{tool_result_for_context}")
                    current_prompt = (
                        f"Goal: {goal}\n"
                        f"Task: {step_desc}\n\n"
                        f"The extract_source skill returned:\n{tool_result_for_context}\n\n"
                        "Write a concise answer in the same language as the user. "
                        "Summarize the source clearly, highlight the main points, and mention important limitations "
                        "such as missing transcript or OCR when they appear in the extracted content. "
                        "Do not call extract_source again unless the first result explicitly says the source could not be read because the source string is invalid."
                    )
                else:
                    await _halt_run(
                        "BUDGET_TOOL_CALLS",
                        f"Run stopped: tool-call budget exceeded ({run_budget['tool_calls']} / {run_budget['max_tool_calls']}).",
                    )
        async def _try_execute_tool_plan() -> bool:
            nonlocal last_clean_response, latest_tool_name, latest_tool_args, latest_tool_result
            nonlocal latest_tool_details, used_side_effect_tool, tool_log, _native_done
            try:
                await agent_event(
                    "executor",
                    "active",
                    step_index,
                    _visible_agent_message(
                        "Planning tool usage…",
                        "Working out the smallest reliable sequence of tool calls before executing anything.",
                    ),
                )
                runtime_catalog = build_runtime_tool_catalog(
                    run_settings,
                    dynamic_tools=_dynamic_tools,
                    mcp_cache=_mcp_get_cache(),
                    use_outlook=_outlook_on,
                    gmail_available=_gmail_available,
                )
                allowed_tool_names = {
                    _normalize_tool_name(str((schema.get("function") or {}).get("name") or "").strip())
                    for schema in _ollama_tools
                    if isinstance(schema, dict)
                }
                capability_summary = _tool_plan_capability_summary(runtime_catalog, allowed_tool_names)
                if not capability_summary:
                    return False
                plan_prompt = build_tool_plan_prompt(current_prompt, capability_summary)
                plan_messages: list[dict] = []
                if agent_system:
                    plan_messages.append({"role": "system", "content": agent_system + "\nReturn valid JSON only."})
                plan_messages.append({"role": "user", "content": plan_prompt})
                await manager.send_message({"type": "model_active", "model": model}, websocket)
                plan_response = await _call_ollama_with_tools(plan_messages, [], num_ctx=_model_ctx(32768))
                await manager.send_message({"type": "model_idle"}, websocket)
                if plan_response.get("error"):
                    return False
                plan_content = await _emit_native_response_thinking(plan_response)
                if not plan_content:
                    return False
                parsed_plan = _parse_model_json(plan_content, expect="object")
                plan = tool_plan_from_payload(parsed_plan)
                if plan is None:
                    return False
                plan_validation = validate_tool_plan(
                    plan,
                    catalog=runtime_catalog,
                    blocked_tools=blocked_tools,
                    disabled_tools=disabled_tools,
                    approval_tools=approval_tools,
                    allow_side_effects=True,
                    max_steps=max(1, min(8, run_budget["max_tool_calls"] - run_budget["tool_calls"])),
                )
                session["tool_plan"] = plan.to_dict()
                session["tool_plan_validation"] = plan_validation.to_dict()
                try:
                    plan_graph = tool_plan_to_task_graph(plan)
                    _store_task_graph_with_validation(
                        session,
                        plan_graph,
                        blocked_tools=blocked_tools,
                        disabled_tools=disabled_tools,
                        approval_tools=approval_tools,
                        max_tasks=max(1, len(plan.steps) + 2),
                        max_depth=8,
                        allow_side_effects=True,
                    )
                except Exception:
                    pass
                if not plan_validation.valid:
                    return False
                if plan.requires_clarification:
                    last_clean_response = plan.clarification_question or (
                        "I need one clarification before I can continue."
                        if not _looks_dutch(goal, step_desc)
                        else "Ik heb eerst een verduidelijking nodig voordat ik verder kan."
                    )
                    _native_done = True
                    return True
                plan_bindings: dict[str, dict[str, object]] = {}
                session["tool_plan_bindings"] = {}
                for plan_step in plan.steps:
                    t_name = _normalize_tool_name(plan_step.tool)
                    try:
                        t_args = _resolve_tool_plan_args(plan_step, plan_bindings)
                    except ValueError as exc:
                        latest_tool_name = t_name
                        latest_tool_result = f"Error: {exc}"
                        tool_log.append(f"Tool: {t_name}\nArgs: {{}}\nResult:\n{latest_tool_result}")
                        if plan_step.on_failure == "ask_user":
                            last_clean_response = str(exc)
                        _native_done = True
                        return True
                    _allowed, tool_result, _tool_policy = await _authorize_tool_call(
                        step_index,
                        t_name,
                        t_args,
                        enforce_approval_policy=not require_approval,
                        prompt_on_approval=not require_approval,
                    )
                    latest_tool_name = t_name
                    latest_tool_args = dict(t_args)
                    if tool_result:
                        latest_tool_result = tool_result
                        latest_tool_details = build_tool_result_details(t_name, tool_result)
                        await manager.send_message({"type": "tool_result", "tool": t_name, "result": tool_result, "step_index": step_index}, websocket)
                        tool_log.append(f"Tool: {t_name}\nArgs: {json.dumps(t_args)}\nResult:\n{tool_result}")
                        if plan_step.on_failure in {"abort", "ask_user"}:
                            if plan_step.on_failure == "ask_user":
                                last_clean_response = tool_result
                            _native_done = True
                            return True
                        continue
                    if run_budget["tool_calls"] >= run_budget["max_tool_calls"]:
                        await _halt_run("BUDGET_TOOL_CALLS", f"Run stopped: tool-call budget exceeded ({run_budget['tool_calls']} / {run_budget['max_tool_calls']}).")
                        _native_done = True
                        return True
                    execution_context.consume_tool_call()
                    await manager.send_message({"type": "tool_call", "tool": t_name, "args": t_args, "step_index": step_index}, websocket)
                    await track_tokens("tool", str(t_args), "")
                    cache_key = _tool_call_signature(t_name, t_args)
                    if t_name in _READONLY_TOOL_NAMES and cache_key in tool_call_cache:
                        tool_result, tool_details, tool_result_for_context = tool_call_cache[cache_key]
                    else:
                        loop = asyncio.get_event_loop()
                        tool_result = await loop.run_in_executor(None, _execute_tool, t_name, t_args)
                        tool_details = build_tool_result_details(t_name, tool_result)
                        tool_result_for_context = compact_tool_result_for_context(t_name, tool_result, tool_details)
                        if t_name in _READONLY_TOOL_NAMES:
                            tool_call_cache[cache_key] = (tool_result, tool_details, tool_result_for_context)
                    latest_tool_result = tool_result
                    latest_tool_details = tool_details
                    _append_workflow_context_artifact(
                        session,
                        tool_name=t_name,
                        tool_args=t_args,
                        tool_result=tool_result,
                        tool_details=tool_details,
                        stage="executor",
                    )
                    if t_name in _SIDE_EFFECT_TOOLS and not tool_result.startswith(("Error", "POLICY_DENIED", "HITL_REQUIRED")):
                        used_side_effect_tool = True
                    await manager.send_message({"type": "tool_result", "tool": t_name, "result": _tool_result_ui_snippet(t_name, tool_result_for_context), "step_index": step_index}, websocket)
                    tool_log.append(f"Tool: {t_name}\nArgs: {json.dumps(t_args)}\nResult:\n{tool_result_for_context}")
                    step_bindings = _extract_tool_plan_bindings(plan_step, tool_name=t_name, tool_result=tool_result, tool_details=tool_details)
                    plan_bindings[plan_step.step_id] = step_bindings
                    session["tool_plan_bindings"][plan_step.step_id] = dict(step_bindings)
                    if tool_result.startswith(("Error", "POLICY_DENIED", "HITL_REQUIRED", "TOOL_DISABLED")) and plan_step.on_failure in {"abort", "ask_user"}:
                        if plan_step.on_failure == "ask_user":
                            last_clean_response = tool_result
                        _native_done = True
                        return True
                _native_done = True
                return True
            except Exception:
                return False
        # ── Native Ollama tool-calling loop ──────────────────────────────────
        _native_done = False
        if _use_native_tools and _ollama_tools and not (_extract_request and not feedback):
            _tool_plan_handled = await _try_execute_tool_plan()
            if not _tool_plan_handled:
                _native_messages: list[dict] = []
                # Use agent_system WITHOUT TOOLS_DESC — native tool schemas are
                # passed via the 'tools' parameter; mixing both confuses the model.
                if agent_system:
                    _native_messages.append({"role": "system", "content": agent_system})
                _native_messages.append({"role": "user", "content": current_prompt})
                for _nturn in range(5):
                    if run_budget["halted"] or (not await _check_runtime("executor_native_turn")):
                        break
                    await manager.send_message({"type": "model_active", "model": model}, websocket)
                    _resp = await _call_ollama_with_tools(_native_messages, _ollama_tools, num_ctx=_model_ctx(32768))
                    await manager.send_message({"type": "model_idle"}, websocket)
                    if _resp.get("error"):
                        break  # fall back to text-based loop
                    _native_tc = _extract_native_tool_calls(_resp)
                    _msg_content = await _emit_native_response_thinking(_resp)
                    if not _native_tc:
                        # Side-effect step answered with prose — nudge model to call the tool
                        if (
                            _msg_content.strip()
                            and not tool_log
                            and _nturn == 0
                            and (_goal_requests_side_effect(step_desc) or _cl_is_action)
                            and _reply_claims_action(_msg_content)
                        ):
                            _native_messages.append({"role": "assistant", "content": _msg_content})
                            _native_messages.append({"role": "user", "content": (
                                "You described the action instead of performing it. "
                                "You MUST call the appropriate tool now. Do NOT narrate — invoke the function."
                            )})
                            continue
                        # Model answered without tool calls — stream the answer to the UI
                        if _msg_content.strip():
                            await manager.send_message({"type": "stream_start", "agent": "Executor"}, websocket)
                            # Strip think tags from one-shot native response
                            _nc_clean = re.sub(r'<think>[\s\S]*?</think>', '', _msg_content).strip()
                            _nc_out = _nc_clean or _msg_content
                            await manager.send_message({"type": "stream_token", "content": _nc_out}, websocket)
                            await manager.send_message({"type": "stream_end", "full": _nc_out}, websocket)
                            last_clean_response = _nc_out
                            _native_done = True
                        break
                    # Process each tool call from the native response
                    for _tc_item in _native_tc:
                        t_name = _normalize_tool_name(_tc_item["name"])
                        t_args = _tc_item.get("arguments") or {}
                        # Policy gates (same as text-based loop)
                        _allowed, tool_result, _tool_policy = await _authorize_tool_call(
                            step_index,
                            t_name,
                            t_args,
                            enforce_approval_policy=not require_approval,
                            prompt_on_approval=not require_approval,
                        )
                        if not _allowed:
                            latest_tool_name = t_name
                            latest_tool_result = tool_result
                            await manager.send_message({"type": "tool_result", "tool": t_name, "result": tool_result, "step_index": step_index}, websocket)
                            tool_log.append(f"Tool: {t_name}\nArgs: {json.dumps(t_args)}\nResult:\n{tool_result}")
                            _native_messages.append({"role": "assistant", "tool_calls": [{"type": "function", "function": {"name": _tc_item["name"], "arguments": t_args}}]})
                            _native_messages.append({"role": "tool", "content": tool_result})
                            continue
                        if run_budget["tool_calls"] >= run_budget["max_tool_calls"]:
                            await _halt_run("BUDGET_TOOL_CALLS", f"Run stopped: tool-call budget exceeded ({run_budget['tool_calls']} / {run_budget['max_tool_calls']}).")
                            break
                        execution_context.consume_tool_call()
                        await manager.send_message({"type": "tool_call", "tool": t_name, "args": t_args, "step_index": step_index}, websocket)
                        await track_tokens("tool", str(t_args), "")
                        cache_key = _tool_call_signature(t_name, t_args)
                        if t_name in _READONLY_TOOL_NAMES and cache_key in tool_call_cache:
                            tool_result, tool_details, tool_result_for_context = tool_call_cache[cache_key]
                        else:
                            loop = asyncio.get_event_loop()
                            tool_result = await loop.run_in_executor(None, _execute_tool, t_name, t_args)
                            tool_details = build_tool_result_details(t_name, tool_result)
                            tool_result_for_context = compact_tool_result_for_context(t_name, tool_result, tool_details)
                            if t_name in _READONLY_TOOL_NAMES:
                                tool_call_cache[cache_key] = (tool_result, tool_details, tool_result_for_context)
                        latest_tool_name = t_name
                        latest_tool_args = dict(t_args)
                        latest_tool_result = tool_result
                        latest_tool_details = tool_details
                        _append_workflow_context_artifact(
                            session,
                            tool_name=t_name,
                            tool_args=t_args,
                            tool_result=tool_result,
                            tool_details=tool_details,
                            stage="executor",
                        )
                        if t_name in _SIDE_EFFECT_TOOLS and not tool_result.startswith(("Error", "POLICY_DENIED", "HITL_REQUIRED")):
                            used_side_effect_tool = True
                        await manager.send_message({"type": "tool_result", "tool": t_name, "result": _tool_result_ui_snippet(t_name, tool_result_for_context), "step_index": step_index}, websocket)
                        tool_log.append(f"Tool: {t_name}\nArgs: {json.dumps(t_args)}\nResult:\n{tool_result_for_context}")
                        # Append assistant tool_call + tool result to messages
                        _native_messages.append({"role": "assistant", "tool_calls": [{"type": "function", "function": {"name": _tc_item["name"], "arguments": t_args}}]})
                        _native_messages.append({"role": "tool", "content": tool_result_for_context})
                    if run_budget["halted"]:
                        break
                # End native loop
        # ── Text-based fallback loop (non-Ollama, or native didn't complete) ──
        if not _native_done:
            await agent_event(
                "executor",
                "active",
                step_index,
                _visible_agent_message(
                    "Gathering evidence…",
                    "Falling back to the direct execution loop to gather the evidence and tool output needed for this step.",
                ),
            )
            for _turn in range(5):
                if run_budget["halted"] or (not await _check_runtime("executor_turn")):
                    break
                response = await _stream_to_ws(current_prompt, current_system)
                tc, cleaned_response, has_tool_marker = _extract_tool_call(response)
                if has_tool_marker:
                    await manager.send_message({"type": "stream_clear"}, websocket)
                if not has_tool_marker:
                    if (_goal_requests_side_effect(step_desc) or _cl_is_action) and not tool_log and _reply_claims_action(cleaned_response or response):
                        current_prompt = (
                            f"Task: {step_desc}\n\n"
                            "Your previous reply only described an intended action without actually doing it. "
                            "For action requests, you MUST either:\n"
                            "1. emit a TOOL_CALL that performs or gathers what is needed for the action, or\n"
                            "2. ask one short clarifying question if essential information is genuinely missing.\n\n"
                            "Do NOT answer with phrases like 'I will', 'I am going to', or 'first I will check' unless you also make the tool call."
                        )
                        continue
                    last_clean_response = response
                    break
                if not isinstance(tc, dict):
                    last_clean_response = cleaned_response.strip()
                    break
                t_name = _normalize_tool_name(tc.get("tool", ""))
                t_args = tc.get("args", {})
                if t_name in disabled_tools:
                    tool_result = (
                        f"TOOL_DISABLED: Skill '{t_name}' is disabled in Settings > Skills. "
                        "Do not create a copy skill. Advise the user to enable this skill."
                    )
                    latest_tool_name = t_name
                    latest_tool_result = tool_result
                    await manager.send_message({
                        "type": "tool_result", "tool": t_name,
                        "result": tool_result, "step_index": step_index,
                    }, websocket)
                    tool_log.append(f"Tool: {t_name}\nArgs: {json.dumps(t_args)}\nResult:\n{tool_result}")
                    current_prompt = (
                        f"Skill '{t_name}' returned:\n{tool_result}\n\n"
                        "Continue without disabled skills. "
                        "Tell the user to enable the skill in Settings > Skills when needed. "
                        "Never use create_skill/create_tool as a workaround for a disabled skill."
                    )
                    continue
                _allowed, tool_result, _tool_policy = await _authorize_tool_call(
                    step_index,
                    t_name,
                    t_args,
                    enforce_approval_policy=not require_approval,
                    prompt_on_approval=not require_approval,
                    disabled_message=(
                        f"TOOL_DISABLED: Skill '{t_name}' is disabled in Settings > Skills. "
                        "Do not create a copy skill. Advise the user to enable this skill."
                    ),
                    blocked_message=f"POLICY_DENIED: Skill '{t_name}' is blocked by policy for this run.",
                )
                if not _allowed:
                    latest_tool_name = t_name
                    latest_tool_result = tool_result
                    await manager.send_message({
                        "type": "tool_result", "tool": t_name,
                        "result": tool_result, "step_index": step_index,
                    }, websocket)
                    tool_log.append(f"Tool: {t_name}\nArgs: {json.dumps(t_args)}\nResult:\n{tool_result}")
                    current_prompt = (
                        f"Skill '{t_name}' returned:\n{tool_result}\n\n"
                        f"Continue task without this skill or request an alternative skill call."
                    )
                    continue
                if run_budget["tool_calls"] >= run_budget["max_tool_calls"]:
                    await _halt_run(
                        "BUDGET_TOOL_CALLS",
                        f"Run stopped: tool-call budget exceeded ({run_budget['tool_calls']} / {run_budget['max_tool_calls']}).",
                    )
                    break
                execution_context.consume_tool_call()
                await manager.send_message({
                    "type": "tool_call", "tool": t_name, "args": t_args, "step_index": step_index,
                }, websocket)
                await track_tokens("tool", str(t_args), "")
                cache_key = _tool_call_signature(t_name, t_args)
                if t_name in _READONLY_TOOL_NAMES and cache_key in tool_call_cache:
                    tool_result, tool_details, tool_result_for_context = tool_call_cache[cache_key]
                    await manager.send_message(
                        {"type": "status", "content": f"No-progress guard — reusing {t_name} result."},
                        websocket,
                    )
                else:
                    loop = asyncio.get_event_loop()
                    tool_result = await loop.run_in_executor(None, _execute_tool, t_name, t_args)
                    tool_details = build_tool_result_details(t_name, tool_result)
                    tool_result_for_context = compact_tool_result_for_context(t_name, tool_result, tool_details)
                    if t_name in _READONLY_TOOL_NAMES:
                        tool_call_cache[cache_key] = (tool_result, tool_details, tool_result_for_context)
                latest_tool_name = t_name
                latest_tool_args = dict(t_args)
                latest_tool_result = tool_result
                latest_tool_details = tool_details
                _append_workflow_context_artifact(
                    session,
                    tool_name=t_name,
                    tool_args=t_args,
                    tool_result=tool_result,
                    tool_details=tool_details,
                    stage="executor",
                )
                if t_name in _SIDE_EFFECT_TOOLS and not tool_result.startswith(("Error", "POLICY_DENIED", "HITL_REQUIRED")):
                    used_side_effect_tool = True
                await manager.send_message({
                    "type": "tool_result", "tool": t_name,
                    "result": _tool_result_ui_snippet(t_name, tool_result_for_context), "step_index": step_index,
                }, websocket)
                tool_log.append(f"Tool: {t_name}\nArgs: {json.dumps(t_args)}\nResult:\n{tool_result_for_context}")
                current_prompt = (
                    f"Skill '{t_name}' returned:\n{tool_result_for_context}\n\n"
                    f"Now write your complete answer to the task: {step_desc}\n"
                    f"If you need more information use another TOOL_CALL, otherwise write the full answer directly.\n"
                    f"IMPORTANT: Write your answer in plain, human-readable language. "
                    f"NEVER output code blocks, raw commands, curl, or scripts as an answer."
                )
                current_system = agent_system + TOOLS_DESC
            else:
                last_clean_response = ""
        if not last_clean_response:
            await agent_event(
                "executor",
                "active",
                step_index,
                _visible_agent_message(
                    "Synthesising tool results…",
                    "Combining the gathered tool output into one direct answer for this step.",
                ),
            )
            tools_summary = "\n\n---\n\n".join(tool_log) if tool_log else "(no tools were called)"
            synth_prompt = (
                f"Goal: {goal}\n"
                f"Task: {step_desc}\n\n"
                f"You gathered the following information using tools:\n\n{tools_summary}\n\n"
                "Using ONLY the information above, write a complete, direct answer to the task. "
                "Do NOT call any more tools. Do NOT output code blocks, commands, or scripts. "
                "Write the full answer in plain, human-readable language now."
            )
            last_clean_response = await _stream_to_ws(synth_prompt, agent_system)
        if latest_tool_result and len(re.sub(r"\s+", "", last_clean_response or "")) < 20:
            if latest_tool_name in {"outlook_get_inbox", "outlook_search_emails"}:
                if latest_tool_result.strip().lower().startswith("no emails"):
                    last_clean_response = "I found 0 emails for the requested period."
                else:
                    email_count = len(re.findall(r"(?m)^Mailbox:\s", latest_tool_result))
                    if email_count > 0:
                        last_clean_response = f"I found {email_count} emails for the requested period."
                    else:
                        last_clean_response = latest_tool_result[:3000]
            else:
                last_clean_response = latest_tool_result[:3000]
        if latest_tool_name in {
            "outlook_send_email",
            "outlook_update_email",
            "outlook_delete_email",
            "gmail_send_email",
            "gmail_update_email",
            "gmail_delete_email",
            "outlook_create_calendar_event",
            "outlook_update_calendar_event",
            "outlook_delete_calendar_event",
            "google_create_calendar_event",
            "google_update_calendar_event",
            "google_delete_calendar_event",
        } and latest_tool_result.startswith("Calendar event"):
            last_clean_response = latest_tool_result[:3000]
        _latest_domain = ""
        if latest_tool_name in _EMAIL_LOOKUP_TOOL_NAMES:
            _latest_domain = "email"
        elif latest_tool_name in {"outlook_read_calendar", "google_read_calendar"}:
            _latest_domain = "calendar"
        if latest_tool_name != "summarize_personal_items" and latest_tool_result and _latest_domain and _goal_requests_personal_summary(goal, _latest_domain):
            _summary_norm = _normalize_tool_name("summarize_personal_items")
            _summary_allowed = bool(_tool_policy_state(_summary_norm).get("allowed")) and run_budget["tool_calls"] < run_budget["max_tool_calls"]
            _summary_records: list[dict] = []
            if _latest_domain == "email":
                _summary_records = extract_email_lookup_records(latest_tool_result)
            elif _latest_domain == "calendar":
                _summary_records = extract_calendar_lookup_records(latest_tool_result)
            if _summary_allowed and _summary_records:
                _summary_args = {
                    "source_type": _latest_domain,
                    "items": _summary_records,
                    "max_items": 6,
                    "style": "brief",
                }
                execution_context.consume_tool_call()
                await manager.send_message({
                    "type": "tool_call", "tool": "summarize_personal_items", "args": _summary_args, "step_index": step_index,
                }, websocket)
                _loop = asyncio.get_event_loop()
                _summary_result = await _loop.run_in_executor(None, _execute_tool, "summarize_personal_items", _summary_args)
                _summary_details = build_tool_result_details("summarize_personal_items", _summary_result)
                _summary_compact = compact_tool_result_for_context("summarize_personal_items", _summary_result, _summary_details)
                await manager.send_message({
                    "type": "tool_result", "tool": "summarize_personal_items", "result": _tool_result_ui_snippet("summarize_personal_items", _summary_compact), "step_index": step_index,
                }, websocket)
                latest_tool_name = "summarize_personal_items"
                latest_tool_args = _summary_args
                latest_tool_result = _summary_result
                latest_tool_details = _summary_details
                _append_workflow_context_artifact(
                    session,
                    tool_name="summarize_personal_items",
                    tool_args=_summary_args,
                    tool_result=_summary_result,
                    tool_details=_summary_details,
                    stage="executor",
                )
                try:
                    _parsed_summary = json.loads(_summary_result)
                except Exception:
                    _parsed_summary = None
                if isinstance(_parsed_summary, dict):
                    _summary_msg = str(_parsed_summary.get("message") or "").strip()
                    if _summary_msg:
                        last_clean_response = _summary_msg
        if latest_tool_name and latest_tool_result:
            tool_envelope = build_tool_result_envelope(
                goal,
                step_desc,
                latest_tool_name,
                latest_tool_result,
                last_clean_response,
                latest_tool_details,
            )
            if str(tool_envelope.get("rendered_result") or "").strip():
                last_clean_response = str(tool_envelope.get("rendered_result") or "").strip()
            composed_from_tool = bool(tool_envelope.get("was_composed"))
        if last_clean_response:
            _code_blocks = re.findall(r"```[\s\S]*?```", last_clean_response)
            _code_chars = sum(len(b) for b in _code_blocks)
            _total_chars = len(last_clean_response.strip())
            _has_command_pattern = bool(re.search(
                r"(?m)^(?:curl |wget |bash |sh |pip |npm |docker |git clone|powershell|Invoke-)",
                last_clean_response
            ))
            if _total_chars > 0 and (
                (_code_chars / _total_chars > 0.5)
                or (_has_command_pattern and _code_chars / _total_chars > 0.3)
            ):
                await agent_event(
                    "executor",
                    "active",
                    step_index,
                    _visible_agent_message(
                        "Rewriting raw output…",
                        "Converting tool-heavy or code-heavy output into a plain user-facing answer.",
                    ),
                )
                _rewrite_prompt = (
                    f"Goal: {goal}\n"
                    f"Task: {step_desc}\n\n"
                    f"The previous answer contained raw code or commands instead of a proper response:\n"
                    f"{last_clean_response[:5000]}\n\n"
                    "Rewrite this as a clean, human-readable answer. "
                    "Extract the FACTS and DATA from the code/output and present them clearly. "
                    "NEVER include code blocks, commands, curl, bash, or scripts. "
                    "Write ONLY the substantive answer in plain language."
                )
                last_clean_response = await call_fast(
                    _rewrite_prompt, system=agent_system, num_ctx=_model_ctx(16384),
                )
            last_clean_response = _strip_tool_call_artifacts(last_clean_response)
            last_clean_response = _strip_filler_phrases(last_clean_response)
            if latest_tool_result:
                last_clean_response = _normalize_tool_json_reply(last_clean_response, latest_tool_result)
        return last_clean_response, used_side_effect_tool, {
            "tool_name": latest_tool_name,
            "tool_args": latest_tool_args,
            "tool_result": latest_tool_result[:3000],
            "tool_details": latest_tool_details,
            "deterministic_result": composed_from_tool,
        }
    # Sub-agent D — Validator
    async def validator_agent(step_desc: str, result: str) -> str:
        if run_budget["halted"] or (not await _check_runtime("validator")):
            return "ISSUE: Run halted by budget policy"
        validator_max_chars = 2400 if speed_mode else 9000
        def _build_validator_result_view(text: str, limit: int) -> tuple[str, bool]:
            if len(text) <= limit:
                return text, False
            head = max(500, int(limit * 0.7))
            tail = max(300, limit - head)
            marker = "\n\n[... middle omitted for validator context ...]\n\n"
            preview = text[:head] + marker + text[-tail:]
            return preview, True
        result_view, was_truncated = _build_validator_result_view(result, validator_max_chars)
        truncation_note = (
            "NOTE: The result excerpt below is intentionally truncated in the middle. "
            "Do NOT report ISSUE only because the excerpt boundary cuts a sentence. "
            "Report ISSUE only for real quality/completeness problems.\n\n"
            if was_truncated else ""
        )
        prompt = (
            f"Step: {step_desc}\n\n"
            f"{truncation_note}"
            f"Result produced:\n{result_view}\n\n"
            "Does this result fully and correctly address the step? "
            "Reply with one line: PASS if yes, or ISSUE: <brief reason> if not."
        )
        verdict = await call_fast(
            prompt,
            system="You are a quality control agent. Be strict but fair. Reply with exactly PASS or ISSUE: <reason>.",
            num_ctx=12288,
        )
        return verdict.strip()
    def _can_skip_validator_fast(result: str, had_side_effect: bool) -> bool:
        if had_side_effect:
            return True
        text = (result or "").strip()
        if len(text) < 120:
            return False
        low = text.lower()
        bad_markers = (
            "error", "traceback", "issue:", "tool_disabled",
            "policy_denied", "hitl_required", "run halted",
            "timed out", "cannot",
        )
        return not any(m in low for m in bad_markers)
    def _result_looks_failed(text: str) -> bool:
        cleaned = str(text or "").strip()
        if not cleaned:
            return True
        lowered = cleaned.lower()
        if lowered.startswith((
            "error",
            "tool_disabled",
            "policy_denied",
            "hitl_required",
            "[run halted",
            "[stream error",
            "stream error:",
        )):
            return True
        bad_markers = (
            "traceback",
            "failed to parse plan",
            "no completed steps were available",
            "could not complete",
            "unable to complete",
            "[stream error:",
            "stream error:",
            "error calling fast model",
            "error calling ollama",
            "client error:",
            "connection error",
            "read timed out",
        )
        return any(marker in lowered for marker in bad_markers)
    def _memory_contains_failure(mem: list[dict]) -> bool:
        for item in mem:
            verdict = str(item.get("verdict") or "").strip().upper()
            result = str(item.get("result") or "").strip()
            if verdict and not verdict.startswith(("PASS", "DONE")):
                return True
            if result and _result_looks_failed(result):
                return True
        return False
    def _should_attempt_executor_plan_b(
        step_result: str,
        verdict: str,
        step_meta: dict,
        *,
        had_side_effect: bool,
    ) -> bool:
        if had_side_effect or _cl_is_action or known_copilot_kind:
            return False
        if run_budget["halted"]:
            return False
        result_text = str(step_result or "").strip()
        tool_text = str((step_meta or {}).get("tool_result") or "").strip()
        combined = "\n".join(part for part in (result_text, tool_text) if part).lower()
        if any(marker in combined for marker in ("tool_disabled", "policy_denied", "hitl_required")):
            return False
        if str(verdict or "").strip().upper().startswith("ISSUE"):
            return True
        if _result_looks_failed(result_text):
            return True
        if not tool_text and len(re.sub(r"\s+", "", result_text)) < 20:
            return True
        return False
    async def _try_executor_plan_b(step_desc: str, step_index: int) -> tuple[str, bool, dict] | None:
        if not manager.active_runs.get(run_id, False):
            return None
        stored_task_state = scope_task_state_to_domain(
            session.get("task_state") if isinstance(session.get("task_state"), dict) else None,
            _protocol_route.domain,
        )
        classified_hint = None
        raw_classifier = session.get("intent_classifier")
        if len(steps) == 1 and step_index in {-1, 0} and isinstance(raw_classifier, dict):
            classified_hint = dict(raw_classifier)
        direct_plan = plan_direct_execution(
            step_desc,
            route=_protocol_route,
            stored_task_state=stored_task_state,
            classified_intent=classified_hint,
            is_action=False,
            use_outlook=bool(use_outlook and OUTLOOK_SUPPORT),
            gmail_available=bool(_gmail_available),
            settings=run_settings,
            dynamic_tools=_dynamic_tools,
            mcp_cache=_mcp_get_cache(),
        )
        direct_graph = direct_plan.task_graph
        if direct_graph is None and direct_plan.calls:
            direct_graph = task_graph_from_direct_calls(
                direct_plan.goal or step_desc,
                protocol=str(getattr(direct_plan.graph_route, "protocol", "read") or "read"),
                domain=str(getattr(direct_plan.graph_route, "domain", "general") or "general"),
                calls=[(call["tool"], dict(call.get("args") or {})) for call in direct_plan.calls],
                display_goal=direct_plan.display_goal or step_desc,
                metadata=dict(direct_plan.metadata or {}),
                side_effect=False,
            )
        if direct_graph is None:
            return None
        validation = _store_task_graph_with_validation(
            session,
            direct_graph,
            blocked_tools=blocked_tools,
            disabled_tools=disabled_tools,
            approval_tools=approval_tools,
            max_tasks=run_budget["max_tool_calls"],
            max_depth=max(1, min(run_budget["max_steps"], len(direct_graph.tasks) or 1)),
            allow_side_effects=False,
        )
        if not validation.valid:
            return None
        direct_results = await _execute_task_graph(
            direct_graph,
            websocket=websocket,
            run_id=run_id,
            step_index_base=step_index,
        )
        if not direct_results:
            return None
        rendered_results: list[str] = []
        latest_meta = {
            "tool_name": "",
            "tool_args": {},
            "tool_result": "",
            "tool_details": {},
            "deterministic_result": True,
            "plan_b_applied": True,
        }
        for task, (tool_name, tool_result) in zip(direct_graph.tasks, direct_results):
            tool_args = dict(getattr(task, "args", {}) or {})
            tool_envelope = build_tool_result_envelope(
                goal,
                step_desc,
                tool_name,
                tool_result,
                tool_result,
            )
            tool_details = dict(tool_envelope.get("details") or {})
            _append_workflow_context_artifact(
                session,
                tool_name=tool_name,
                tool_args=tool_args,
                tool_result=tool_result,
                tool_details=tool_details,
                stage="executor_plan_b",
            )
            _record_hypothesis_evidence(step_index, tool_name, tool_result, tool_details)
            rendered_results.append(str(tool_envelope.get("rendered_result") or "").strip())
            latest_meta = {
                "tool_name": tool_name,
                "tool_args": tool_args,
                "tool_result": str(tool_result or "")[:3000],
                "tool_details": tool_details,
                "deterministic_result": True,
                "plan_b_applied": True,
            }
        step_result = compose_direct_results(
            step_desc,
            direct_results,
            "\n\n---\n\n".join(text for text in rendered_results if text),
        ).strip()
        if not step_result:
            step_result = str(latest_meta.get("tool_result") or "").strip()
        if _result_looks_failed(step_result):
            return None
        return step_result, False, latest_meta
    async def _should_stop_after_step(step_desc: str, step_result: str, had_side_effect: bool, step_index: int) -> bool:
        if had_side_effect:
            return False
        if step_index >= len(steps) - 1:
            return False
        if multi_goal_mode:
            return False
        if not (step_result or "").strip():
            return False
        if known_copilot_kind:
            return False
        if _cl_is_action:
            return False
        completed_summary = "\n\n".join(
            f"Step {j+1}: {m['step']}\nResult: {(m['result'] or '')[:1000]}"
            for j, m in enumerate(memory)
        )
        remaining = "\n".join(f"- {s['step']}" for s in steps[step_index + 1:]) or "- none"
        stop_check = await call_fast(
            (
                f"Original user goal: {goal}\n\n"
                f"Completed results so far:\n{completed_summary}\n\n"
                f"Remaining planned steps:\n{remaining}\n\n"
                "Should the workflow STOP NOW because the user already has the final answer?\n"
                "Rules:\n"
                "- STOP only if the completed results already fully and completely answer the user's request with no gaps.\n"
                "- If the user explicitly named one provider/source/folder, results from that exact provider/source/folder are sufficient.\n"
                "- CONTINUE if any required part of the answer is missing, incomplete, or could be improved by remaining steps.\n"
                "- When in doubt, CONTINUE — it is better to do slightly more work than to leave the user with an incomplete answer.\n\n"
                "Reply with exactly one line: STOP or CONTINUE: <brief reason>."
            ),
            system="You are a strict workflow stop evaluator. Reply with exactly STOP or CONTINUE: <brief reason>.",
            num_ctx=4096,
        )
        return (stop_check or "").strip().upper().startswith("STOP")
    # Sub-agent E — Report Out
    async def _emit_thinking_events(raw: str) -> str:
        """Extract <think> blocks, send as WS events, return cleaned text."""
        if not run_settings.get("show_thinking", False):
            return re.sub(r'<think>[\s\S]*?</think>', '', raw).strip()
        blocks = re.findall(r'<think>([\s\S]*?)</think>', raw or "")
        if blocks:
            await manager.send_message({"type": "stream_thinking_start", "run_id": run_id}, websocket)
            for b in blocks:
                await manager.send_message({"type": "stream_thinking", "content": b, "run_id": run_id}, websocket)
            await manager.send_message({"type": "stream_thinking_end", "run_id": run_id}, websocket)
            await asyncio.sleep(0)
        return re.sub(r'<think>[\s\S]*?</think>', '', raw).strip()
    async def report_out_agent() -> str:
        if run_budget["halted"]:
            return f"Run stopped by policy: {run_budget['halt_reason']}."
        if not memory:
            return "No completed steps were available to build a final report."
        if multi_goal_mode:
            todo_report = _build_multi_goal_final_report(goal, steps, memory)
            if todo_report:
                return todo_report
        _want_thinking = run_settings.get("show_thinking", False)
        _report_routine_kind = known_copilot_kind
        if _report_routine_kind in {"morning_brief", "follow_up"}:
            _report_snippet_limit = 1200 if speed_mode else 1800
            raw_results = []
            for m in memory:
                result_text = (m.get("result", "") or "").strip()
                if result_text:
                    raw_results.append(result_text[:_report_snippet_limit])
            combined_results = "\n---\n".join(raw_results)
            if _report_routine_kind == "morning_brief":
                prompt = build_morning_brief_report_prompt(goal, combined_results)
                report_system = build_morning_brief_report_system(_current_date_line())
            else:
                prompt = build_followup_brief_report_prompt(goal, combined_results)
                report_system = build_followup_brief_report_system(_current_date_line())
            if speed_mode:
                report = await call_fast(prompt, system=report_system, num_ctx=_model_ctx(8192))
            else:
                report = await call_with_vision(prompt, system=report_system, num_ctx=_model_ctx(12288), thinking=_want_thinking)
            report = await _emit_thinking_events(report or "")
            # Background: enrich user.md from briefing data (#11)
            try:
                from backend.copilot_briefing import (
                    _collect_morning_brief_inputs,
                    enrich_user_profile_from_briefing,
                    extract_email_enrichments,
                )
                _b_emails, _b_events, _, _, _ = _collect_morning_brief_inputs(memory)
                _b_enrichments = extract_email_enrichments(memory)
                _b_suggestions = enrich_user_profile_from_briefing(_b_emails, _b_events, _b_enrichments)
                if _b_suggestions:
                    _user_path = Path("user.md")
                    if _user_path.exists():
                        _existing = _user_path.read_text(encoding="utf-8")
                        _new_lines = [s for s in _b_suggestions if s.strip().lstrip("- ").lower() not in _existing.lower()]
                        if _new_lines:
                            _user_path.write_text(
                                _existing.rstrip() + "\n" + "\n".join(_new_lines) + "\n",
                                encoding="utf-8",
                            )
            except Exception:
                pass
            return (report or "").strip()
        _report_snippet_limit = 2500 if speed_mode else 4000
        raw_results = []
        for m in memory:
            result_text = (m.get("result", "") or "").strip()
            if result_text:
                raw_results.append(result_text[:_report_snippet_limit])
        combined_results = "\n---\n".join(raw_results)
        prompt = (
            f"User's request: {goal}\n\n"
            f"Raw data collected:\n{combined_results}\n\n"
            "─── YOUR TASK ───\n"
            "Write the FINAL ANSWER the user will read.  This is the ONLY thing they see — "
            "they have NOT seen any intermediate steps or tool output.\n\n"
            "FORMAT (use exactly these headers):\n\n"
            "Answer\n"
            "Write the complete, direct answer here.  This is the ENTIRE deliverable.\n"
            "- Present the actual content: data, facts, numbers, email contents, names, dates, URLs, code, etc.\n"
            "- Structure with bullet points, numbered lists, or tables where it helps readability.\n"
            "- Be thorough — include every relevant detail from the raw data.  Do NOT summarise away specifics.\n"
            "- If the user asked for a list, give the full list.  If they asked for email content, show it.\n\n"
            "Notes\n"
            "(ONLY if there were real problems, incomplete data, or important caveats.  "
            "Omit this section entirely when everything is fine.)\n\n"
            "ABSOLUTE RULES:\n"
            "1. ZERO process narration.  Never say 'I searched…', 'I found…', 'The agent performed…', "
            "'Based on the research…', 'After analysing…'.  Just present the answer.\n"
            "2. BAD: 'I searched your inbox and found 5 emails.'  "
            "GOOD: '5 emails today:\n1. From john@ — Meeting tomorrow — …'\n"
            "3. BAD: 'A web search was performed for Python tutorials.'  "
            "GOOD: 'Top Python tutorials:\n1. …'\n"
            "4. Write as if YOU are the knowledgeable answer, not a report about an investigation.\n"
            "5. Write in the same language as the user's request.\n"
            "6. Do not invent information not present in the raw data.\n"
            "7. If the raw data references past sessions, include inline citations as "
            "[\U0001f4ce DATE: goal] for traceability."
        )
        report_system = (
            f"{_current_date_line()} "
            "You deliver clear, direct answers.  Your output is the final thing the user reads — "
            "give them the actual content they asked for, not a story about how it was found.  "
            "Never describe steps, process, or methodology.  Just the answer."
        )
        if speed_mode:
            report = await call_fast(prompt, system=report_system, num_ctx=_model_ctx(12288))
        else:
            report = await call_with_vision(prompt, system=report_system, num_ctx=_model_ctx(16384), thinking=_want_thinking)
        report = await _emit_thinking_events(report or "")
        return report
    # ── Execute each step through the pipeline ──────────────────────────────────
    research_cache: dict[int, str] = {}
    search_steps = [(idx, s) for idx, s in enumerate(steps) if s["needs_search"]]
    if search_steps:
        await manager.send_message({
            "type": "info",
            "content": f"⚡ Parallel research: pre-fetching {len(search_steps)} search(es) simultaneously…"
        }, websocket)
        await agent_event("researcher", "active", msg=f"Running {len(search_steps)} search(es) in parallel")
        async def _prefetch_research(idx: int, step_obj: dict) -> tuple[int, str]:
            brief = await researcher_agent(step_obj["step"], step_obj["search_hint"], idx)
            return idx, brief
        prefetch_results = await asyncio.gather(
            *[_prefetch_research(idx, s) for idx, s in search_steps],
            return_exceptions=True,
        )
        for item in prefetch_results:
            if isinstance(item, tuple):
                research_cache[item[0]] = item[1]
        await agent_event("researcher", "done",
                          msg=f"{len(research_cache)}/{len(search_steps)} briefs ready")
        await manager.send_message({
            "type": "info",
            "content": f"✓ Research complete: {len(research_cache)} brief(s) cached. Starting execution loop."
        }, websocket)
    # Phase B: sequential execution loop
    awaiting_user_input = False
    for i, step_obj in enumerate(steps):
        if run_budget["steps_executed"] >= run_budget["max_steps"]:
            await _halt_run(
                "BUDGET_STEPS",
                f"Run stopped: step budget exceeded ({run_budget['steps_executed']} / {run_budget['max_steps']}).",
            )
            break
        run_budget["steps_executed"] += 1
        if not await _wait_if_paused(run_id):
            await manager.send_message({"type": "info", "content": "Orchestration stopped by user."}, websocket)
            break
        if run_budget["halted"] or (not await _check_runtime("step_loop")):
            break
        step_desc   = step_obj["step"]
        needs_search = step_obj["needs_search"]
        search_hint  = step_obj["search_hint"]
        direct_calls = list(step_obj.get("direct_calls") or [])
        await manager.send_message({
            "type": "step_start", "step_index": i,
            "content": f"Step {i+1}/{len(steps)}: {step_desc}"
        }, websocket)
        await agent_event("all", "idle", i)
        _current_attempt = (_ensure_todo_progress_items(session, steps)[i] or {}).get("attempt") or 0
        await _push_todo_progress(
            "step_started",
            step_index=i,
            step_fields={
                "state": "active",
                "attempt": max(1, int(_current_attempt)),
                "evaluation": "",
                "approach": "Primary approach in progress.",
                "result_preview": "",
                "goal_met": False,
            },
        )
        if known_copilot_kind and direct_calls:
            await manager.send_message({"type": "status", "content": f"Step {i+1}: Executing…"}, websocket)
            await agent_event(
                "executor",
                "active",
                i,
                _visible_agent_message(
                    f"{step_desc[:60]}",
                    f"Executing routine step {i+1}: {_step_focus(step_desc, 120)} using the predefined direct-call graph.",
                ),
            )
            graph = task_graph_from_direct_calls(
                step_desc,
                protocol="read",
                domain=known_copilot_kind,
                calls=direct_calls,
                display_goal=step_desc,
                metadata={"goal": goal, "copilot": known_copilot_kind},
                side_effect=False,
                dependency_mode="independent",
            )
            validation = _store_task_graph_with_validation(
                session,
                graph,
                blocked_tools=blocked_tools,
                disabled_tools=disabled_tools,
                approval_tools=approval_tools,
                max_tasks=run_budget["max_tool_calls"],
                max_depth=max(1, len(direct_calls)),
                allow_side_effects=False,
            )
            if not validation.valid:
                step_result = "Unable to run Morning Brief fetch tasks: " + "; ".join(validation.errors or ["task graph invalid"])
                step_entries = [{
                    "step": step_desc,
                    "result": step_result,
                    "verdict": "ISSUE",
                    "tool_name": "",
                    "tool_result": "",
                    "tool_details": {},
                    "deterministic_result": False,
                }]
                verdict = "ISSUE"
                await agent_event("executor", "error", i, step_result[:80])
                await agent_event("validator", "done", i, "ISSUE")
            else:
                direct_results = await _execute_task_graph(
                    graph,
                    websocket=websocket,
                    run_id=run_id,
                    step_index_base=i,
                )
                step_entries = []
                rendered_results: list[str] = []
                for tool_name, tool_result in direct_results:
                    tool_envelope = build_tool_result_envelope(
                        goal,
                        step_desc,
                        tool_name,
                        tool_result,
                        tool_result,
                    )
                    tool_details = dict(tool_envelope.get("details") or {})
                    rendered_result = str(tool_envelope.get("rendered_result") or "").strip()
                    _record_hypothesis_evidence(i, tool_name, tool_result, tool_details)
                    rendered_results.append(rendered_result)
                    step_entries.append({
                        "step": step_desc,
                        "result": rendered_result,
                        "verdict": "PASS",
                        "tool_name": tool_name,
                        "tool_result": tool_result[:3000],
                        "tool_details": tool_details,
                        "deterministic_result": True,
                    })
                step_result = compose_direct_results(
                    step_desc,
                    direct_results,
                    "\n\n---\n\n".join(text for text in rendered_results if text),
                ).strip()
                verdict = "PASS"
                await agent_event(
                    "executor",
                    "done",
                    i,
                    _visible_agent_message(
                        f"{len(step_result)} chars produced",
                        f"Completed the routine step from the direct-call graph using {len(direct_results)} tool result(s).",
                    ),
                )
                await agent_event("validator", "done", i, "PASS (Morning Brief direct task graph)")
            memory.extend(step_entries)
            session["steps"].extend(step_entries)
            _save_session(session)
            await _push_todo_progress(
                "step_completed",
                step_index=i,
                step_fields={
                    "state": _todo_progress_state_from_verdict(verdict),
                    "verdict": verdict,
                    "evaluation": verdict if not str(verdict or "").upper().startswith("PASS") else "Step goal achieved.",
                    "approach": "Primary approach completed.",
                    "result_preview": step_result,
                    "goal_met": str(verdict or "").upper().startswith("PASS"),
                },
            )
            await manager.send_message({
                "type": "step_result", "step": step_desc, "result": step_result,
                "verdict": verdict, "memory": memory, "streamed": True,
            }, websocket)
            continue
        if require_approval and not steps_auto_approved:
            ap_event = asyncio.Event()
            manager.pending_approvals[run_id] = {"event": ap_event, "response": None, "note": ""}
            await manager.send_message({
                "type": "approval_needed", "step_index": i, "step": step_desc,
            }, websocket)
            await ap_event.wait()
            ap = dict(manager.pending_approvals.pop(run_id, {}) or {})
            resp = ap.get("response", "reject")
            if resp == "reject":
                note = ap.get("note") or "rejected by user"
                await manager.send_message({"type": "info", "content": f"Step {i+1} skipped: {note}"}, websocket)
                memory.append({"step": step_desc, "result": f"[Skipped: {note}]", "verdict": "SKIP"})
                session["steps"].append({"step": step_desc, "result": f"[Skipped: {note}]", "verdict": "SKIP"})
                await _push_todo_progress(
                    "step_completed",
                    step_index=i,
                    step_fields={
                        "state": "skipped",
                        "attempt": 1,
                        "verdict": "SKIP",
                        "evaluation": note,
                        "approach": "Skipped by user approval flow.",
                        "result_preview": f"[Skipped: {note}]",
                        "goal_met": False,
                    },
                )
                _save_session(session)
                continue
            if resp == "approve_always":
                steps_auto_approved = True
                await manager.send_message({"type": "info", "content": "Auto-approving all remaining steps in this workflow."}, websocket)
            if ap.get("note"):
                await manager.send_message({"type": "info", "content": f"Note from user: {ap['note']}"}, websocket)
        research_brief = ""
        if needs_search:
            research_brief = research_cache.get(i, "")
            if research_brief:
                await manager.send_message({"type": "info", "content": f"Step {i+1}: Using pre-fetched research brief."}, websocket)
                await agent_event(
                    "researcher",
                    "done",
                    i,
                    _visible_agent_message(
                        "Using cached brief",
                        "Reusing the research brief that was already gathered for this step instead of running another search.",
                    ),
                )
            else:
                await manager.send_message({"type": "status", "content": f"Step {i+1}: Researching… (cache miss)"}, websocket)
                research_brief = await researcher_agent(step_desc, search_hint, i)
            await handoff(
                "researcher",
                "curator",
                i,
                f"{research_brief[:80].replace(chr(10),' ')}…" if len(research_brief) > 80 else research_brief[:80],
                question=f"What evidence here is directly useful for: {_step_focus(step_desc, 90)}?",
            )
        else:
            await agent_event(
                "researcher",
                "skip",
                i,
                _visible_agent_message(
                    "No web search needed",
                    "This step can be executed from existing context and tools, so no external web evidence is needed.",
                ),
            )
        # Skip curator LLM call for simple single-step queries — pass any
        # user context directly to the executor without an LLM summarisation pass.
        if known_copilot_kind:
            curated = ""
            await manager.send_message({
                "type": "context_curated", "step_index": i, "briefing": curated,
            }, websocket)
            await agent_event(
                "curator",
                "skip",
                i,
                _visible_agent_message(
                    f"{known_copilot_kind} fast-path",
                    f"Skipping extra curation because the {known_copilot_kind} routine already has a fixed execution path.",
                ),
            )
        elif _execution_mode == "simple" and not memory:
            curated = safe_context[:4000] if safe_context else ""
            if past_memory and not curated:
                curated = past_memory[:2000]
            await manager.send_message({
                "type": "context_curated", "step_index": i, "briefing": curated,
            }, websocket)
            await agent_event(
                "curator",
                "skip" if not curated else "done",
                i,
                _visible_agent_message(
                    "Direct context pass (no LLM curation)" if curated else "Skipped (simple mode)",
                    (
                        "Passing the user-provided context straight into execution because this is a single-step request."
                        if curated else
                        "Skipping curation because the simple step already has all the context it needs."
                    ),
                ),
            )
        else:
            await manager.send_message({"type": "status", "content": f"Step {i+1}: Curating context…"}, websocket)
            curated = await curator_agent(step_desc, i, memory, research_brief)
            await manager.send_message({
                "type": "context_curated", "step_index": i, "briefing": curated,
            }, websocket)
        await handoff(
            "curator",
            "executor",
            i,
            f"Curated context ready ({len(curated)} chars)" if curated else "No prior context needed",
            question=f"What is the next concrete action to complete: {_step_focus(step_desc, 90)}?",
        )
        await manager.send_message({"type": "status", "content": f"Step {i+1}: Executing…"}, websocket)
        await agent_event(
            "executor",
            "active",
            i,
            _visible_agent_message(
                f"{step_desc[:60]}",
                f"Executing step {i+1}: {_step_focus(step_desc, 120)}. Tools will be chosen only when the current context is not enough.",
            ),
        )
        step_result, had_side_effect, step_meta = await executor_agent(step_desc, curated, research_brief, step_index=i, needs_search=needs_search, completed_steps=memory)
        _executor_tool_used = str(step_meta.get("tool_name") or "").strip()
        await agent_event(
            "executor",
            "done",
            i,
            _visible_agent_message(
                f"{len(step_result)} chars produced",
                (
                    f"Completed the step using {_executor_tool_used} and produced a user-facing result."
                    if _executor_tool_used else
                    "Completed the step without needing an additional tool call in the final pass."
                ),
            ),
        )
        await manager.send_message({
            "type": "step_draft", "step_index": i, "step": step_desc,
            "result": step_result, "attempt": 1,
        }, websocket)
        await handoff(
            "executor",
            "validator",
            i,
            f"Result ready ({len(step_result)} chars)",
            question=f"Did this result fully satisfy the step: {_step_focus(step_desc, 90)}?",
        )
        if known_copilot_kind:
            verdict = "PASS"
            await agent_event("validator", "done", i, f"PASS ({known_copilot_kind} fast-path)")
        elif _execution_mode == "simple":
            verdict = "PASS"
            await agent_event("validator", "done", i, "PASS (simple mode)")
        elif _can_skip_validator_fast(step_result, had_side_effect):
            verdict = "PASS"
            await agent_event("validator", "done", i, "PASS (fast-path, skip validator call)")
        else:
            await manager.send_message({"type": "status", "content": f"Step {i+1}: Validating…"}, websocket)
            await agent_event(
                "validator",
                "active",
                i,
                _visible_agent_message(
                    "Checking result quality…",
                    "Checking whether the step result actually satisfies the request and whether any concrete evidence is still missing.",
                ),
            )
            verdict = await validator_agent(step_desc, step_result)
        max_validation_retries = min((0 if speed_mode else 1), run_budget["max_validation_retries"])
        for retry_num in range(max_validation_retries):
            if not verdict.upper().startswith("ISSUE"):
                break
            await agent_event("validator", "error", i, verdict[:80])
            await manager.send_message({
                "type": "retry", "step_index": i, "attempt": retry_num + 1, "reason": verdict,
            }, websocket)
            await _push_todo_progress(
                "step_retrying",
                step_index=i,
                step_fields={
                    "state": "retrying",
                    "attempt": retry_num + 2,
                    "evaluation": verdict,
                    "approach": "Retrying with revised instructions.",
                    "goal_met": False,
                },
            )
            await manager.send_message({"type": "status", "content": f"Step {i+1}: Retrying (attempt {retry_num + 2})…"}, websocket)
            await handoff(
                "validator",
                "executor",
                i,
                f"Retry {retry_num+1}: {verdict[:60]}",
                question=f"How should the step be revised to fix: {_step_focus(verdict, 90)}?",
            )
            await agent_event(
                "executor",
                "active",
                i,
                _visible_agent_message(
                    f"Retry attempt {retry_num + 2}",
                    f"Retrying the step after validator feedback: {_step_focus(verdict, 120)}",
                ),
            )
            step_result, _, step_meta = await executor_agent(step_desc, curated, research_brief, feedback=verdict, step_index=i, needs_search=needs_search, completed_steps=memory)
            _retry_tool_used = str(step_meta.get("tool_name") or "").strip()
            await agent_event(
                "executor",
                "done",
                i,
                _visible_agent_message(
                    "Retry completed",
                    (
                        f"Retry completed with {_retry_tool_used}."
                        if _retry_tool_used else
                        "Retry completed and produced a revised answer for validation."
                    ),
                ),
            )
            await manager.send_message({
                "type": "step_draft", "step_index": i, "step": step_desc,
                "result": step_result, "attempt": retry_num + 2,
            }, websocket)
            await handoff(
                "executor",
                "validator",
                i,
                f"Revised result ({len(step_result)} chars)",
                question=f"Does the revised result now satisfy: {_step_focus(step_desc, 90)}?",
            )
            await manager.send_message({"type": "status", "content": f"Step {i+1}: Re-validating…"}, websocket)
            await agent_event(
                "validator",
                "active",
                i,
                _visible_agent_message(
                    f"Re-checking after retry {retry_num+1}…",
                    f"Re-checking the revised result after retry {retry_num + 1} to see whether the gap has been closed.",
                ),
            )
            verdict = await validator_agent(step_desc, step_result)
        if _should_attempt_executor_plan_b(step_result, verdict, step_meta, had_side_effect=had_side_effect):
            _plan_b_reason = verdict if str(verdict or "").strip() else "Executor result was insufficient."
            await manager.send_message({
                "type": "retry", "step_index": i, "attempt": max_validation_retries + 1, "reason": _plan_b_reason,
            }, websocket)
            await _push_todo_progress(
                "step_plan_b",
                step_index=i,
                step_fields={
                    "state": "plan_b",
                    "attempt": max_validation_retries + 2,
                    "evaluation": _plan_b_reason,
                    "approach": "Trying an alternative approach (Plan B).",
                    "goal_met": False,
                    "plan_b_applied": True,
                },
            )
            await manager.send_message({
                "type": "status", "content": f"Step {i+1}: Starting Plan B…"}, websocket)
            await handoff(
                "validator",
                "executor",
                i,
                f"Plan B: {_plan_b_reason[:60]}",
                question=f"What alternative path could still complete: {_step_focus(step_desc, 90)}?",
            )
            await agent_event(
                "executor",
                "active",
                i,
                _visible_agent_message(
                    "Plan B fallback",
                    f"Switching to an alternative execution path because validation found a gap: {_step_focus(_plan_b_reason, 120)}",
                ),
            )
            _plan_b_result = await _try_executor_plan_b(step_desc, i)
            if _plan_b_result is not None:
                step_result, had_side_effect, step_meta = _plan_b_result
                _plan_b_tool_used = str(step_meta.get("tool_name") or "").strip()
                await agent_event(
                    "executor",
                    "done",
                    i,
                    _visible_agent_message(
                        f"{len(step_result)} chars produced (Plan B)",
                        (
                            f"Plan B succeeded using {_plan_b_tool_used}."
                            if _plan_b_tool_used else
                            "Plan B produced an alternative result for the validator to check."
                        ),
                    ),
                )
                await manager.send_message({
                    "type": "step_draft", "step_index": i, "step": step_desc,
                    "result": step_result, "attempt": max_validation_retries + 2,
                }, websocket)
                await handoff(
                    "executor",
                    "validator",
                    i,
                    f"Plan B result ({len(step_result)} chars)",
                    question=f"Does Plan B fully resolve: {_step_focus(step_desc, 90)}?",
                )
                if _can_skip_validator_fast(step_result, had_side_effect):
                    verdict = "PASS"
                else:
                    await manager.send_message({"type": "status", "content": f"Step {i+1}: Validating Plan B…"}, websocket)
                    await agent_event(
                        "validator",
                        "active",
                        i,
                        _visible_agent_message(
                            "Checking Plan B result…",
                            "Checking whether the alternative Plan B result closes the missing gap.",
                        ),
                    )
                    verdict = await validator_agent(step_desc, step_result)
        await agent_event("validator", "done", i, verdict)
        _tool_used = str(step_meta.get("tool_name") or "").strip()
        if _tool_used:
            try:
                _was_quarantined = record_custom_tool_feedback(_tool_used, verdict, reason=str(step_result)[:260])
                if _was_quarantined:
                    await manager.send_message({
                        "type": "warning",
                        "content": f"Skill '{_tool_used}' has been automatically disabled after repeated failures. You can re-enable it in Settings > Skills.",
                    }, websocket)
            except Exception:
                pass
        memory.append({"step": step_desc, "result": step_result, "verdict": verdict, **step_meta})
        session["steps"].append({"step": step_desc, "result": step_result, "verdict": verdict, **step_meta})
        _record_hypothesis_evidence(
            i,
            str(step_meta.get("tool_name") or ""),
            str(step_meta.get("tool_result") or ""),
            dict(step_meta.get("tool_details") or {}),
        )
        _stop_after_remediation = await _apply_auto_remediation(i)
        _save_session(session)
        _step_passed = str(verdict or "").upper().startswith("PASS")
        await _push_todo_progress(
            "step_completed",
            step_index=i,
            step_fields={
                "state": _todo_progress_state_from_verdict(verdict),
                "verdict": verdict,
                "evaluation": verdict if not _step_passed else "Step goal achieved.",
                "approach": "Alternative approach succeeded." if step_meta.get("plan_b_applied") else "Primary approach completed.",
                "result_preview": step_result,
                "goal_met": _step_passed,
                "plan_b_applied": bool(step_meta.get("plan_b_applied")),
            },
        )
        await manager.send_message({
            "type": "step_result", "step": step_desc, "result": step_result,
            "verdict": verdict, "memory": memory, "streamed": True,
        }, websocket)
        if _step_result_requires_user_input(step_desc, step_result):
            awaiting_user_input = True
            await _push_todo_progress(
                "goal_evaluation",
                goal_status="awaiting_user",
                goal_evaluation="Waiting for user input before continuing.",
            )
            await manager.send_message({
                "type": "info",
                "content": "Stopping here: waiting for user input before continuing.",
            }, websocket)
            steps = steps[:i + 1]
            break
        if await _should_stop_after_step(step_desc, step_result, had_side_effect, i):
            await manager.send_message({
                "type": "info",
                "content": "Stopping early: the requested answer is already complete.",
            }, websocket)
            steps = steps[:i + 1]
            break
        if _stop_after_remediation and not multi_goal_mode:
            await manager.send_message({
                "type": "info",
                "content": "Stopping early: auto-remediation selected a final evidence-first response.",
            }, websocket)
            steps = steps[:i + 1]
            break
    # ── 3. Re-planning ──────────────────────────────────────────────────────────
    # Iterative replan loop: after each batch of new steps, re-check completion
    # and plan another batch if the goal still is not met. Bounded by
    # policy_max_replan_rounds (default 3) and stagnation detection.
    _max_replan_rounds = max(1, int(run_budget.get("max_replan_rounds", 3) or 3))
    _prior_gaps: list[str] = []
    _prior_replan_step_sigs: list[str] = []
    def _replan_step_signature(step_items: list[dict]) -> str:
        """Build a stable signature of planned steps to detect stagnation."""
        parts: list[str] = []
        for it in step_items[:5]:
            txt = str(it.get("step") or "").strip().lower()
            # keep first ~80 chars of each step for comparison
            parts.append(txt[:80])
        return " | ".join(parts)
    for _replan_round in range(_max_replan_rounds):
        if awaiting_user_input:
            break
        if _execution_mode == "simple" and memory:
            # SIMPLE mode vangnet: check if the single step actually achieved the goal.
            # If result is empty/error/clearly incomplete, upgrade to NEEDS_PLAN and re-plan.
            _simple_result = (memory[0].get("result") or "").strip() if memory else ""
            _simple_looks_failed = (
                not _simple_result
                or _simple_result.startswith(("Error", "TOOL_DISABLED", "POLICY_DENIED", "HITL_REQUIRED"))
                or len(_simple_result) < 20
            )
            if not _simple_looks_failed:
                break
            # Upgrade: allow re-planning to close the gap
            _execution_mode = "needs_plan"
            await manager.send_message(
                {"type": "info", "content": "Simple execution returned insufficient results — re-evaluating with full planning…"},
                websocket,
            )
            # Fall through to the normal re-planning logic below
        if known_copilot_kind:
            break
        # Skip re-planning when every step was an action/side-effect (e.g. calendar, email)
        _action_markers_re = (
            "uploaded", "sent", "saved", "created", "deleted", "scheduled",
            "posted", "moved", "wrote", "updated", "calendar", "event",
        )
        if memory and all(
            any(m in (mem_item.get("result") or "").lower()[:300] for m in _action_markers_re)
            for mem_item in memory if (mem_item.get("result") or "").strip()
        ):
            break
        if not manager.active_runs.get(run_id, False):
            break
        if run_budget["halted"] or (not await _check_runtime("replan")):
            break
        summary = "\n\n".join(
            f"Step {j+1}: {m['step']}\nResult: {(m['result'] or '')[:300].replace(chr(10), ' ')}"
            for j, m in enumerate(memory)
        )
        await agent_event("all", "idle")
        await agent_event(
            "planner",
            "active",
            msg=_visible_agent_message(
                "Re-evaluating goal completion…",
                "Checking whether the completed steps already satisfy the original goal or whether a specific gap still remains.",
            ),
        )
        _round_label = f" (round {_replan_round + 1}/{_max_replan_rounds})" if _max_replan_rounds > 1 else ""
        await manager.send_message({"type": "status", "content": f"Checking if goal is fully achieved{_round_label}…"}, websocket)
        _prior_gap_block = ""
        if _prior_gaps:
            _prior_gap_block = (
                "\n\nPrevious replan attempts already addressed these gaps "
                "(do NOT repeat — verify whether they are now resolved):\n- "
                + "\n- ".join(_prior_gaps[-3:])
                + "\n"
            )
        completion_check = await call_with_vision(
            f"Original goal: {goal}\n\nCompleted steps and results:\n{summary}{_prior_gap_block}\n\n"
            "Evaluate: Has the original goal been fully and satisfactorily achieved?\n"
            "Base your judgment on CONCRETE EVIDENCE in the results (tool confirmations, file paths, "
            "created IDs, saved content, etc.). If a required deliverable was mentioned in the goal but "
            "no result confirms it exists, reply INCOMPLETE.\n"
            "If the steps show that actions were successfully performed (e.g. event created, email sent, "
            "file written with confirmed path), consider the goal COMPLETE.\n"
            "Reply with exactly one of:\n"
            "  COMPLETE\n"
            "  INCOMPLETE: <brief explanation of exactly what concrete deliverable is still missing>",
            system=(
                "You are a goal completion evaluator. Be practical, concrete, and evidence-based. "
                "Only say COMPLETE when the results contain explicit confirmation of each required deliverable. "
                "Only say INCOMPLETE if a clearly required deliverable is genuinely missing or failed. "
                "Reply ONLY with COMPLETE or INCOMPLETE: <reason>."
            ),
        )
        if completion_check.strip().upper().startswith("COMPLETE"):
            await _push_todo_progress(
                "goal_evaluation",
                goal_status="complete",
                goal_evaluation="Current results satisfy the original goal.",
            )
            break
        gap = completion_check.strip()
        _prior_gaps.append(gap[:400])
        await _push_todo_progress(
            "goal_evaluation",
            goal_status="incomplete",
            goal_evaluation=gap,
        )
        await manager.send_message({"type": "info", "content": f"Re-planning{_round_label} — gap found: {gap}"}, websocket)
        # ── Tool-aware replan: refresh catalog for new steps ─────────
        _replan_catalog = build_runtime_tool_catalog(
            run_settings,
            dynamic_tools=_dynamic_tools,
            mcp_cache=_mcp_get_cache(),
            use_outlook=_outlook_on,
            gmail_available=_gmail_available,
        )
        _replan_tool_digest = _planner_capability_digest(_replan_catalog)
        _replan_tool_block = (
            f"\nAvailable tools & skills:\n{_replan_tool_digest}\n"
            f"You may create new skills with create_skill if no existing tool fits.\n"
            if _replan_tool_digest else ""
        )
        # Include what was already tried so the planner doesn't repeat failed approaches
        _prior_attempts_block = ""
        if _replan_round > 0 and _prior_gaps[:-1]:
            _prior_attempts_block = (
                "\nPrior replan rounds identified these gaps (do NOT repeat the same failed approach):\n- "
                + "\n- ".join(g[:200] for g in _prior_gaps[:-1])
                + "\n"
            )
        if _planner_learning_block:
            replan_prompt = (
                f"Original goal: {goal}\n\nAlready completed:\n{summary}\n\n"
                f"Gap identified: {gap}\n\n"
                f"{_planner_learning_block}\n\n"
                f"{_prior_attempts_block}"
                f"{_replan_tool_block}\n"
                f"List ONLY the additional steps needed to close this gap.{internet_note}\n"
                "Each step MUST be concrete and tool-aware: mention which tool(s) to use.\n"
                "If previous approaches failed, try a DIFFERENT strategy or tool.\n"
                "Return ONLY a valid JSON array of step objects with keys: \"step\", \"needs_search\", \"search_hint\"."
            )
        else:
            replan_prompt = (
                f"Original goal: {goal}\n\nAlready completed:\n{summary}\n\n"
                f"Gap identified: {gap}\n\n"
                f"{_prior_attempts_block}"
                f"{_replan_tool_block}\n"
                f"List ONLY the additional steps needed to close this gap.{internet_note}\n"
                "Each step MUST be concrete and tool-aware: mention which tool(s) to use.\n"
                "If previous approaches failed, try a DIFFERENT strategy or tool.\n"
                "Return ONLY a valid JSON array of step objects with keys: \"step\", \"needs_search\", \"search_hint\"."
            )
        replan_raw = await call_with_vision(
            replan_prompt,
            system=agent_system + "\n\nOutput ONLY a valid JSON array of objects. No prose.",
        )
        try:
            extra_items = _parse_model_json(replan_raw, expect="array")
            if not isinstance(extra_items, list) or not extra_items:
                break
            new_steps: list[dict] = []
            for item in extra_items:
                if isinstance(item, str):
                    new_steps.append({"step": item, "needs_search": False, "search_hint": ""})
                elif isinstance(item, dict) and "step" in item:
                    new_steps.append({
                        "step": item["step"],
                        "needs_search": bool(item.get("needs_search", False)) and use_internet,
                        "search_hint": item.get("search_hint", ""),
                    })
            if not new_steps:
                break
        except Exception:
            break
        # Stagnation detection: if this replan produces substantially the same
        # steps as a previous round, stop to avoid looping on the same failure.
        _this_sig = _replan_step_signature(new_steps)
        if _this_sig and _this_sig in _prior_replan_step_sigs:
            await manager.send_message({
                "type": "info",
                "content": "Stopping replan loop: planner is repeating previously attempted steps.",
            }, websocket)
            break
        _prior_replan_step_sigs.append(_this_sig)
        offset = len(steps)
        steps.extend(new_steps)
        await manager.send_message({
            "type": "plan_extended", "new_steps": [s["step"] for s in new_steps], "total": len(steps),
        }, websocket)
        await _push_todo_progress(
            "plan_extended",
            goal_status="running",
            goal_evaluation=gap,
        )
        for _extra_i, extra_step_obj in enumerate(new_steps):
            if run_budget["steps_executed"] >= run_budget["max_steps"]:
                await _halt_run(
                    "BUDGET_STEPS",
                    f"Run stopped: step budget exceeded ({run_budget['steps_executed']} / {run_budget['max_steps']}).",
                )
                break
            run_budget["steps_executed"] += 1
            if not await _wait_if_paused(run_id):
                break
            if run_budget["halted"] or (not await _check_runtime("replan_step_loop")):
                break
            i = offset + _extra_i
            step_desc    = extra_step_obj["step"]
            needs_search = extra_step_obj["needs_search"]
            search_hint  = extra_step_obj["search_hint"]
            await manager.send_message({
                "type": "step_start", "step_index": i,
                "content": f"Step {i+1}/{len(steps)}: {step_desc}",
            }, websocket)
            await agent_event("all", "idle", i)
            _current_attempt = (_ensure_todo_progress_items(session, steps)[i] or {}).get("attempt") or 0
            await _push_todo_progress(
                "step_started",
                step_index=i,
                step_fields={
                    "state": "active",
                    "attempt": max(1, int(_current_attempt)),
                    "evaluation": "",
                    "approach": "Primary approach in progress.",
                    "result_preview": "",
                    "goal_met": False,
                },
            )
            if require_approval and not steps_auto_approved:
                ap_event = asyncio.Event()
                manager.pending_approvals[run_id] = {"event": ap_event, "response": None, "note": ""}
                await manager.send_message({
                    "type": "approval_needed", "step_index": i, "step": step_desc,
                }, websocket)
                await ap_event.wait()
                ap = dict(manager.pending_approvals.pop(run_id, {}) or {})
                resp = ap.get("response", "reject")
                if resp == "reject":
                    note = ap.get("note") or "rejected by user"
                    await manager.send_message({"type": "info", "content": f"Step {i+1} skipped: {note}"}, websocket)
                    memory.append({"step": step_desc, "result": f"[Skipped: {note}]", "verdict": "SKIP"})
                    session["steps"].append({"step": step_desc, "result": f"[Skipped: {note}]", "verdict": "SKIP"})
                    await _push_todo_progress(
                        "step_completed",
                        step_index=i,
                        step_fields={
                            "state": "skipped",
                            "attempt": 1,
                            "verdict": "SKIP",
                            "evaluation": note,
                            "approach": "Skipped by user approval flow.",
                            "result_preview": f"[Skipped: {note}]",
                            "goal_met": False,
                        },
                    )
                    _save_session(session)
                    continue
                if resp == "approve_always":
                    steps_auto_approved = True
                    await manager.send_message({"type": "info", "content": "Auto-approving all remaining steps in this workflow."}, websocket)
                if ap.get("note"):
                    await manager.send_message({"type": "info", "content": f"Note from user: {ap['note']}"}, websocket)
            research_brief = ""
            if needs_search:
                await manager.send_message({"type": "status", "content": f"Step {i+1}: Researching…"}, websocket)
                research_brief = await researcher_agent(step_desc, search_hint, i)
                await handoff(
                    "researcher",
                    "curator",
                    i,
                    f"{research_brief[:80].replace(chr(10),' ')}…" if len(research_brief) > 80 else research_brief[:80],
                    question=f"What evidence here is directly useful for: {_step_focus(step_desc, 90)}?",
                )
            else:
                await agent_event(
                    "researcher",
                    "skip",
                    i,
                    _visible_agent_message(
                        "No web search needed",
                        "This extra step does not need fresh web evidence, so execution can continue from current context.",
                    ),
                )
            await manager.send_message({"type": "status", "content": f"Step {i+1}: Curating context…"}, websocket)
            curated = await curator_agent(step_desc, i, memory, research_brief)
            await manager.send_message({
                "type": "context_curated", "step_index": i, "briefing": curated,
            }, websocket)
            await handoff(
                "curator",
                "executor",
                i,
                f"Curated context ready ({len(curated)} chars)" if curated else "No prior context needed",
                question=f"What is the next concrete action to complete: {_step_focus(step_desc, 90)}?",
            )
            await manager.send_message({"type": "status", "content": f"Step {i+1}: Executing…"}, websocket)
            await agent_event(
                "executor",
                "active",
                i,
                _visible_agent_message(
                    f"{step_desc[:60]}",
                    f"Executing remedial step {i+1}: {_step_focus(step_desc, 120)}. The workflow will use tools only if needed to close the missing gap.",
                ),
            )
            step_result, had_side_effect, step_meta = await executor_agent(step_desc, curated, research_brief, step_index=i, needs_search=needs_search)
            _replan_tool_used = str(step_meta.get("tool_name") or "").strip()
            await agent_event(
                "executor",
                "done",
                i,
                _visible_agent_message(
                    f"{len(step_result)} chars produced",
                    (
                        f"Completed the remedial step using {_replan_tool_used}."
                        if _replan_tool_used else
                        "Completed the remedial step and produced a revised result."
                    ),
                ),
            )
            await manager.send_message({
                "type": "step_draft", "step_index": i, "step": step_desc,
                "result": step_result, "attempt": 1,
            }, websocket)
            await handoff(
                "executor",
                "validator",
                i,
                f"Result ready ({len(step_result)} chars)",
                question=f"Did this result fully satisfy the step: {_step_focus(step_desc, 90)}?",
            )
            if _can_skip_validator_fast(step_result, had_side_effect):
                verdict = "PASS"
                await agent_event("validator", "done", i, "PASS (fast-path, skip validator call)")
            else:
                await manager.send_message({"type": "status", "content": f"Step {i+1}: Validating…"}, websocket)
                await agent_event(
                    "validator",
                    "active",
                    i,
                    _visible_agent_message(
                        "Checking result quality…",
                        "Checking whether the remedial step actually closes the remaining goal gap.",
                    ),
                )
                verdict = await validator_agent(step_desc, step_result)
            max_validation_retries = min((0 if speed_mode else 1), run_budget["max_validation_retries"])
            for retry_num in range(max_validation_retries):
                if not verdict.upper().startswith("ISSUE"):
                    break
                await agent_event("validator", "error", i, verdict[:80])
                await manager.send_message({
                    "type": "retry", "step_index": i,
                    "attempt": retry_num + 1, "reason": verdict,
                }, websocket)
                await _push_todo_progress(
                    "step_retrying",
                    step_index=i,
                    step_fields={
                        "state": "retrying",
                        "attempt": retry_num + 2,
                        "evaluation": verdict,
                        "approach": "Retrying with revised instructions.",
                        "goal_met": False,
                    },
                )
                await manager.send_message({"type": "status", "content": f"Step {i+1}: Retrying (attempt {retry_num + 2})…"}, websocket)
                await handoff(
                    "validator",
                    "executor",
                    i,
                    f"Retry {retry_num+1}: {verdict[:60]}",
                    question=f"How should the step be revised to fix: {_step_focus(verdict, 90)}?",
                )
                await agent_event(
                    "executor",
                    "active",
                    i,
                    _visible_agent_message(
                        f"Retry attempt {retry_num + 2}",
                        f"Retrying the remedial step after validator feedback: {_step_focus(verdict, 120)}",
                    ),
                )
                step_result, _, step_meta = await executor_agent(step_desc, curated, research_brief, feedback=verdict, step_index=i, needs_search=needs_search, completed_steps=memory)
                _replan_retry_tool = str(step_meta.get("tool_name") or "").strip()
                await agent_event(
                    "executor",
                    "done",
                    i,
                    _visible_agent_message(
                        "Retry completed",
                        (
                            f"Retry completed with {_replan_retry_tool}."
                            if _replan_retry_tool else
                            "Retry completed and produced a revised result for validation."
                        ),
                    ),
                )
                await manager.send_message({
                    "type": "step_draft", "step_index": i, "step": step_desc,
                    "result": step_result, "attempt": retry_num + 2,
                }, websocket)
                await handoff(
                    "executor",
                    "validator",
                    i,
                    f"Revised result ({len(step_result)} chars)",
                    question=f"Does the revised result now satisfy: {_step_focus(step_desc, 90)}?",
                )
                await manager.send_message({"type": "status", "content": f"Step {i+1}: Re-validating…"}, websocket)
                await agent_event(
                    "validator",
                    "active",
                    i,
                    _visible_agent_message(
                        f"Re-checking after retry {retry_num+1}…",
                        f"Re-checking the revised remedial result after retry {retry_num + 1}.",
                    ),
                )
                verdict = await validator_agent(step_desc, step_result)
            if _should_attempt_executor_plan_b(step_result, verdict, step_meta, had_side_effect=had_side_effect):
                _plan_b_reason = verdict if str(verdict or "").strip() else "Executor result was insufficient."
                await manager.send_message({
                    "type": "retry", "step_index": i,
                    "attempt": max_validation_retries + 1, "reason": _plan_b_reason,
                }, websocket)
                await _push_todo_progress(
                    "step_plan_b",
                    step_index=i,
                    step_fields={
                        "state": "plan_b",
                        "attempt": max_validation_retries + 2,
                        "evaluation": _plan_b_reason,
                        "approach": "Trying an alternative approach (Plan B).",
                        "goal_met": False,
                        "plan_b_applied": True,
                    },
                )
                await manager.send_message({"type": "status", "content": f"Step {i+1}: Starting Plan B…"}, websocket)
                await handoff(
                    "validator",
                    "executor",
                    i,
                    f"Plan B: {_plan_b_reason[:60]}",
                    question=f"What alternative path could still complete: {_step_focus(step_desc, 90)}?",
                )
                await agent_event(
                    "executor",
                    "active",
                    i,
                    _visible_agent_message(
                        "Plan B fallback",
                        f"Trying an alternative remedial path because validation still reports: {_step_focus(_plan_b_reason, 120)}",
                    ),
                )
                _plan_b_result = await _try_executor_plan_b(step_desc, i)
                if _plan_b_result is not None:
                    step_result, had_side_effect, step_meta = _plan_b_result
                    _replan_plan_b_tool = str(step_meta.get("tool_name") or "").strip()
                    await agent_event(
                        "executor",
                        "done",
                        i,
                        _visible_agent_message(
                            f"{len(step_result)} chars produced (Plan B)",
                            (
                                f"Plan B succeeded using {_replan_plan_b_tool}."
                                if _replan_plan_b_tool else
                                "Plan B produced an alternative remedial result."
                            ),
                        ),
                    )
                    await manager.send_message({
                        "type": "step_draft", "step_index": i, "step": step_desc,
                        "result": step_result, "attempt": max_validation_retries + 2,
                    }, websocket)
                    await handoff(
                        "executor",
                        "validator",
                        i,
                        f"Plan B result ({len(step_result)} chars)",
                        question=f"Does Plan B fully resolve: {_step_focus(step_desc, 90)}?",
                    )
                    if _can_skip_validator_fast(step_result, had_side_effect):
                        verdict = "PASS"
                    else:
                        await manager.send_message({"type": "status", "content": f"Step {i+1}: Validating Plan B…"}, websocket)
                        await agent_event(
                            "validator",
                            "active",
                            i,
                            _visible_agent_message(
                                "Checking Plan B result…",
                                "Checking whether the alternative remedial result finally closes the goal gap.",
                            ),
                        )
                        verdict = await validator_agent(step_desc, step_result)
            await agent_event("validator", "done", i, verdict)
            _tool_used = str(step_meta.get("tool_name") or "").strip()
            if _tool_used:
                try:
                    _was_quarantined = record_custom_tool_feedback(_tool_used, verdict, reason=str(step_result)[:260])
                    if _was_quarantined:
                        await manager.send_message({
                            "type": "warning",
                            "content": f"Skill '{_tool_used}' has been automatically disabled after repeated failures. You can re-enable it in Settings > Skills.",
                        }, websocket)
                except Exception:
                    pass
            memory.append({"step": step_desc, "result": step_result, "verdict": verdict, **step_meta})
            session["steps"].append({"step": step_desc, "result": step_result, "verdict": verdict, **step_meta})
            _record_hypothesis_evidence(
                i,
                str(step_meta.get("tool_name") or ""),
                str(step_meta.get("tool_result") or ""),
                dict(step_meta.get("tool_details") or {}),
            )
            _stop_after_remediation = await _apply_auto_remediation(i)
            _save_session(session)
            _step_passed = str(verdict or "").upper().startswith("PASS")
            await _push_todo_progress(
                "step_completed",
                step_index=i,
                step_fields={
                    "state": _todo_progress_state_from_verdict(verdict),
                    "verdict": verdict,
                    "evaluation": verdict if not _step_passed else "Step goal achieved.",
                    "approach": "Alternative approach succeeded." if step_meta.get("plan_b_applied") else "Primary approach completed.",
                    "result_preview": step_result,
                    "goal_met": _step_passed,
                    "plan_b_applied": bool(step_meta.get("plan_b_applied")),
                },
            )
            await manager.send_message({
                "type": "step_result", "step": step_desc, "result": step_result,
                "verdict": verdict, "memory": memory, "streamed": True,
            }, websocket)
            if _step_result_requires_user_input(step_desc, step_result):
                awaiting_user_input = True
                await _push_todo_progress(
                    "goal_evaluation",
                    goal_status="awaiting_user",
                    goal_evaluation="Waiting for user input before continuing.",
                )
                await manager.send_message({
                    "type": "info",
                    "content": "Stopping here: waiting for user input before continuing.",
                }, websocket)
                steps = steps[:i + 1]
                break
            if _stop_after_remediation:
                await manager.send_message({
                    "type": "info",
                    "content": "Stopping early: auto-remediation selected a final evidence-first response.",
                }, websocket)
                steps = steps[:i + 1]
                break
    # ── Final report ────────────────────────────────────────────────────────────
    def _should_skip_report(mem: list[dict]) -> bool:
        """Decide whether the report agent can be skipped.
        Skip when the executor output is already a clean, user-ready answer
        and a separate synthesis pass would just repeat it with minor formatting.
        """
        if not mem:
            return True
        if _memory_contains_failure(mem):
            return False
        # Single-step runs: the executor result is already the answer.
        if len(mem) == 1:
            return True
        # All results are very short — nothing to synthesise.
        combined_len = sum(len((m.get("result") or "").strip()) for m in mem)
        if combined_len < 300:
            return True
        # Every step was a side-effect action (send email, write file, etc.)
        # where the result is just a confirmation message.
        _action_markers = (
            "uploaded", "sent", "saved", "created", "deleted", "scheduled",
            "posted", "moved", "wrote", "updated",
        )
        if all(
            any(marker in (m.get("result") or "").lower()[:200] for marker in _action_markers)
            for m in mem if (m.get("result") or "").strip()
        ):
            return True
        return False
    _final_goal_status = str(session.get("todo_progress_goal_status") or "").strip().lower()
    _final_goal_evaluation = str(session.get("todo_progress_goal_evaluation") or "").strip()
    if not _final_goal_status:
        if awaiting_user_input:
            _final_goal_status = "awaiting_user"
            _final_goal_evaluation = _final_goal_evaluation or "Waiting for user input before continuing."
        elif memory and all(str(item.get("verdict") or "").upper().startswith("PASS") for item in memory):
            _final_goal_status = "complete"
            _final_goal_evaluation = _final_goal_evaluation or "All planned step goals were achieved."
        elif memory:
            _final_goal_status = "partial"
            _final_goal_evaluation = _final_goal_evaluation or "Workflow finished with open issues or skipped items."
        else:
            _final_goal_status = "running"
    _failed_memory = _memory_contains_failure(memory)
    if not awaiting_user_input and _failed_memory:
        _final_goal_status = "partial"
        if not _final_goal_evaluation or _final_goal_evaluation == "Current results satisfy the original goal.":
            _final_goal_evaluation = "Workflow finished with open issues or incomplete results."
    await _push_todo_progress(
        "goal_final",
        goal_status=_final_goal_status,
        goal_evaluation=_final_goal_evaluation,
    )
    if manager.active_runs.get(run_id, False):
        if known_copilot_kind:
            # ── Layer 3: Build personal context ────────────────────────
            _user_ctx = ""
            try:
                _raw_user = _load_user_text()
                if _raw_user:
                    _user_ctx_parts = [build_user_context_block(_raw_user)]
                    _prio_block = build_priorities_context_block()
                    if _prio_block:
                        _user_ctx_parts.append(_prio_block)
                    _cp = load_contact_profiles()
                    if _cp:
                        _contacts_block = build_contacts_context_block(_cp)
                        if _contacts_block:
                            _user_ctx_parts.append(_contacts_block)
                    _user_ctx = "\n".join(p for p in _user_ctx_parts if p)
            except Exception:
                pass
            # ── Layer 1: Deep Content Analysis ─────────────────────────
            # Run LLM analysis over the fetched email bodies + calendar
            # events before composing the deterministic report.
            try:
                from backend.copilot_briefing import _collect_morning_brief_inputs
                _enrich_emails, _enrich_events, _, _, _ = _collect_morning_brief_inputs(memory)
                async def _run_email_analysis() -> str:
                    _analysis_prompt = build_copilot_email_analysis_prompt(_enrich_emails, _enrich_events)
                    _analysis_system = build_copilot_email_analysis_system(_current_date_line(), user_context=_user_ctx)
                    if speed_mode:
                        return await call_fast(_analysis_prompt, system=_analysis_system, num_ctx=12288)
                    return await call_with_vision(_analysis_prompt, system=_analysis_system, num_ctx=16384, thinking=False)

                async def _run_event_analysis() -> str:
                    _evt_prompt = build_copilot_event_analysis_prompt(_enrich_events, _enrich_emails)
                    _evt_system = build_copilot_event_analysis_system(_current_date_line(), user_context=_user_ctx)
                    if speed_mode:
                        return await call_fast(_evt_prompt, system=_evt_system, num_ctx=8192)
                    return await call_with_vision(_evt_prompt, system=_evt_system, num_ctx=12288, thinking=False)

                _parallel_analysis_tasks: list[tuple[str, object]] = []
                if _enrich_emails:
                    _parallel_analysis_tasks.append(("email", _run_email_analysis()))
                if _enrich_events:
                    _parallel_analysis_tasks.append(("event", _run_event_analysis()))
                if len(_parallel_analysis_tasks) == 2:
                    await agent_event("reportout", "active", msg="Analysing email content and calendar events in parallel…")
                elif _enrich_emails:
                    await agent_event("reportout", "active", msg="Analysing email content…")
                elif _enrich_events:
                    await agent_event("reportout", "active", msg="Analysing calendar events…")

                if _parallel_analysis_tasks:
                    _parallel_analysis_results = await asyncio.gather(
                        *(task for _, task in _parallel_analysis_tasks),
                        return_exceptions=True,
                    )
                    for (_analysis_kind, _task), _analysis_result in zip(_parallel_analysis_tasks, _parallel_analysis_results):
                        if isinstance(_analysis_result, Exception):
                            if _analysis_kind == "email":
                                await agent_event("reportout", "active", msg=f"Content analysis skipped: {_analysis_result}")
                            else:
                                await agent_event("reportout", "active", msg=f"Event analysis skipped: {_analysis_result}")
                            continue
                        if _analysis_kind == "email":
                            _analysis_parsed = parse_copilot_email_analysis(_analysis_result or "", len(_enrich_emails))
                            memory.append({
                                "tool_name": _COPILOT_EMAIL_ANALYSIS_ENTRY,
                                "tool_result": json.dumps(_analysis_parsed, ensure_ascii=False),
                                "result": "",
                            })
                            try:
                                _l1_for_contacts = extract_email_enrichments(memory)
                                update_contact_profiles_from_emails(_enrich_emails, _l1_for_contacts)
                            except Exception:
                                pass
                        else:
                            _evt_parsed = parse_copilot_event_analysis(_analysis_result or "", len(_enrich_events))
                            memory.append({
                                "tool_name": _COPILOT_EVENT_ANALYSIS_ENTRY,
                                "tool_result": json.dumps(_evt_parsed, ensure_ascii=False),
                                "result": "",
                            })
            except Exception as _parallel_enrich_err:
                await agent_event("reportout", "active", msg=f"Parallel brief analysis skipped: {_parallel_enrich_err}")
            # ── Layer 1c: Thread Context for Priority Emails ───────────
            # For high-priority emails, fetch the full conversation thread
            # and run LLM analysis to extract context, status, and suggestions.
            try:
                _l1c_enrichments = extract_email_enrichments(memory)
                if _l1c_enrichments and _enrich_emails:
                    _thread_candidates = identify_thread_candidates(_l1c_enrichments, _enrich_emails, max_threads=5)
                    if _thread_candidates:
                        await agent_event("reportout", "active", msg="Fetching email threads…")
                        _thread_data: list[dict] = []
                        for _tc in _thread_candidates:
                            try:
                                _tid = _tc.get("thread_id") or ""
                                _tprov = _tc.get("provider") or ""
                                if not _tid:
                                    continue
                                if _tprov == "gmail":
                                    _traw = _gmail_get_thread(_tid, body_length=400, max_messages=8)
                                elif _tprov == "outlook":
                                    _traw = _outlook_get_thread(_tid, body_length=400, max_messages=8)
                                else:
                                    continue
                                if _traw and not _traw.startswith("Error"):
                                    _thread_data.append({
                                        "email_index": _tc["email_index"],
                                        "subject": _tc.get("subject", ""),
                                        "thread_messages": _traw,
                                    })
                            except Exception:
                                pass
                        if _thread_data:
                            await agent_event("reportout", "active", msg="Analysing email threads…")
                            _tc_prompt = build_copilot_thread_analysis_prompt(_thread_data)
                            _tc_system = build_copilot_thread_analysis_system(_current_date_line(), user_context=_user_ctx)
                            _tc_indices = [d["email_index"] for d in _thread_data]
                            if speed_mode:
                                _tc_raw = await call_fast(_tc_prompt, system=_tc_system, num_ctx=8192)
                            else:
                                _tc_raw = await call_with_vision(_tc_prompt, system=_tc_system, num_ctx=12288, thinking=False)
                            _tc_parsed = parse_copilot_thread_analysis(_tc_raw or "", _tc_indices)
                            if _tc_parsed:
                                memory.append({
                                    "tool_name": _COPILOT_THREAD_ANALYSIS_ENTRY,
                                    "tool_result": json.dumps(_tc_parsed, ensure_ascii=False),
                                    "result": "",
                                })
            except Exception as _tc_err:
                await agent_event("reportout", "active", msg=f"Thread analysis skipped: {_tc_err}")
            # ── Layer 2: Cross-Item Intelligence ───────────────────────
            # Find topic clusters, time-pressure patterns, and a day
            # narrative across all emails + events.
            try:
                _l1_enrichments = extract_email_enrichments(memory)
                _l1b_event_enrichments = extract_event_enrichments(memory)
                if (_l1_enrichments or _l1b_event_enrichments) and (_enrich_emails or _enrich_events):
                    await agent_event("reportout", "active", msg="Finding cross-item patterns…")
                    _cross_prompt = build_copilot_cross_item_prompt(
                        _l1_enrichments, _enrich_emails, _enrich_events,
                        event_enrichments=_l1b_event_enrichments,
                    )
                    _cross_system = build_copilot_cross_item_system(_current_date_line(), user_context=_user_ctx)
                    if speed_mode:
                        _cross_raw = await call_fast(_cross_prompt, system=_cross_system, num_ctx=12288)
                    else:
                        _cross_raw = await call_with_vision(_cross_prompt, system=_cross_system, num_ctx=16384, thinking=False)
                    _cross_parsed = parse_copilot_cross_item_analysis(_cross_raw or "")
                    if _cross_parsed:
                        memory.append({
                            "tool_name": _COPILOT_CROSS_ITEM_ENTRY,
                            "tool_result": json.dumps(_cross_parsed, ensure_ascii=False),
                            "result": "",
                        })
            except Exception as _cross_err:
                await agent_event("reportout", "active", msg=f"Cross-item analysis skipped: {_cross_err}")
            final_report = compose_final_report(goal, memory, "")
            if final_report:
                await agent_event(
                    "reportout",
                    "done",
                    msg=_visible_agent_message(
                        f"{known_copilot_kind} composed from structured signals",
                        f"Built the final {known_copilot_kind} output directly from the validated structured signals already gathered in the workflow.",
                    ),
                )
            else:
                await agent_event(
                    "reportout",
                    "active",
                    msg=_visible_agent_message(
                        "Preparing final report…",
                        "Combining the validated step results into one concise user-facing answer and removing duplicate or unsupported details.",
                    ),
                )
                await handoff(
                    "validator",
                    "reportout",
                    summary="All steps complete; drafting final report",
                    question="How should the validated results be merged into one clear final answer?",
                )
                final_report = await report_out_agent()
                await agent_event(
                    "reportout",
                    "done",
                    msg=_visible_agent_message(
                        "Final report ready",
                        "Final report is ready after merging the validated results into one answer.",
                    ),
                )
        elif _should_skip_report(memory):
            # Use the executor output directly as the final report.
            if len(memory) == 1:
                final_report = (memory[0].get("result") or "").strip()
            else:
                final_report = "\n\n---\n\n".join(
                    (m.get("result") or "").strip() for m in memory if (m.get("result") or "").strip()
                )
            await agent_event(
                "reportout",
                "done",
                msg=_visible_agent_message(
                    "Report skipped — executor output is final",
                    "Skipping extra report synthesis because the executor output is already the final user-facing answer.",
                ),
            )
        else:
            await agent_event(
                "reportout",
                "active",
                msg=_visible_agent_message(
                    "Preparing final report…",
                    "Combining the validated step results into one concise user-facing answer and removing duplicate or unsupported details.",
                ),
            )
            await handoff(
                "validator",
                "reportout",
                summary="All steps complete; drafting final report",
                question="How should the validated results be merged into one clear final answer?",
            )
            final_report = await report_out_agent()
            await agent_event(
                "reportout",
                "done",
                msg=_visible_agent_message(
                    "Final report ready",
                    "Final report is ready after merging the validated results into one answer.",
                ),
            )
        final_report = compose_final_report(goal, memory, final_report)
        final_report += _format_sources_footer(memory, dutch=_looks_dutch_text(goal))
        execution_context.set_trace(_build_compacted_tool_trace(memory))
        _store_session_email_records(session, [
            {
                "tool_name": str(item.get("tool_name") or ""),
                "tool_result": str(item.get("tool_result") or ""),
            }
            for item in memory
            if str(item.get("tool_name") or "").strip()
        ])
        final_report = _append_email_followup_offer(final_report, session, dutch_hint=_looks_dutch_text(goal))
        session.pop("last_lookup_signature", None)
        _store_session_continuity_summary(
            session,
            goal,
            session.get("task_state"),
            session["last_tool_trace"],
            final_report,
        )
        _run_completed_successfully = _final_goal_status == "complete"
        execution_context.set_result(
            goal,
            final_report,
            completed=_run_completed_successfully,
            replace_steps=False,
        )
        _set_run_kpis_snapshot(completed=_run_completed_successfully)
        _persist_learning_feedback()
        # Seed chat_history so follow-up chat messages have conversation context
        _record_session_exchange(session, goal, final_report)
        # Collect output files created/modified during this run
        output_files = []
        try:
            run_start_ts = datetime.fromisoformat(session["created_at"]).timestamp()
            for f in OUTPUT_DIR.iterdir():
                if f.is_file() and f.stat().st_mtime >= run_start_ts:
                    output_files.append({"name": f.name, "size": f.stat().st_size})
        except Exception:
            pass
        if output_files:
            session["output_files"] = output_files
        _save_session(session)
        if awaiting_user_input:
            await manager.send_message({
                "type": "success",
                "content": "Awaiting your input.",
            }, websocket)
        elif _run_completed_successfully:
            await manager.send_message({
                "type": "success",
                "content": "Goal achieved!",
            }, websocket)
        else:
            await manager.send_message({
                "type": "info",
                "content": "Run finished with open issues.",
            }, websocket)
        await manager.send_message({
            "type": "final_report", "goal": goal, "content": final_report,
            "output_files": output_files,
        }, websocket)
    elif run_budget["halted"]:
        execution_context.set_result(
            goal,
            f"Run stopped by policy/budget: {run_budget['halt_reason']}",
            completed=False,
            replace_steps=False,
        )
        _set_run_kpis_snapshot(completed=False)
        _persist_learning_feedback()
        _save_session(session)
    else:
        execution_context.set_result(
            goal,
            "Run stopped by user.",
            completed=False,
            replace_steps=False,
        )
        _set_run_kpis_snapshot(completed=False)
        _persist_learning_feedback()
        _save_session(session)
        await manager.send_message({
            "type": "run_stopped", "reason_code": "USER_STOPPED",
            "content": "Workflow stopped by user.",
        }, websocket)
    manager.active_runs.pop(run_id, None)
    try:
        loop = asyncio.get_event_loop()
        await loop.run_in_executor(None, _ollama_unload_all)
    except Exception:
        pass
    try:
        convo_chunks = [f"Goal: {goal}"]
        if safe_context:
            convo_chunks.append(f"Context: {safe_context}")
        for st in (session.get("steps") or []):
            step_txt = (st.get("step") or "").strip()
            res_txt = (st.get("result") or "").strip()[:600]
            if step_txt:
                convo_chunks.append(f"Step: {step_txt}")
            if res_txt:
                convo_chunks.append(f"Result: {res_txt}")
        fr = (session.get("final_report") or "").strip()[:1500]
        if fr:
            convo_chunks.append(f"Final report: {fr}")
        convo_text = "\n".join(convo_chunks)
        threading.Thread(target=_enrich_user_profile_llm, args=(convo_text,), daemon=True).start()
    except Exception:
        pass
    # Post-run self-reflection (async, non-blocking)
    if session.get("completed"):
        try:
            threading.Thread(target=_run_self_reflection, args=(dict(session),), daemon=True).start()
        except Exception:
            pass
