TraderAI/traderai/agent.py

from __future__ import annotations

import asyncio
import json
import re
import shutil
import subprocess
import tempfile
import uuid
from collections.abc import AsyncIterator
from contextlib import nullcontext
from pathlib import Path
from typing import Any

import httpx
from tzlocal import get_localzone

from traderai.memory import DEFAULT_THREAD_ID, MemoryStore, iso_now, iso_now_in_zone, time_since
from traderai.tools import ToolRegistry
from traderai.version import __version__


SYSTEM_PROMPT = """You are TraderAI, a sharp Star Citizen marketplace copilot for UEX work.
Sound like a competent player who knows the game and the market. Be natural, direct, and helpful. Avoid corporate filler, robotic phrasing, and meta notes.
Use tools when the user asks about UEX data, open/current listings, active negotiations, unread notifications, messages, offers, or posting ads.
Use continual plan tools when the user asks for multi-day or recurring marketplace work, such as finding several parts, watching for deals, tracking candidates, or coordinating negotiations over time.
UEX credentials are configured server-side when available. Never ask the user to provide UEX_SECRET_KEY or UEX_BEARER_TOKEN in chat; call the authenticated UEX tool and only mention credential configuration if the tool returns an authentication error.
Use the specific UEX tool for the needed endpoint, such as get_uex_commodities_prices or get_uex_vehicles. Use fields, limit, and summary mode so tool results stay compact.
When the user asks for history, trends, changes over time, or past prices, prefer the summarize_uex_*_history tools when available; use search_uex_api_index(history_only=true) if you need to discover history endpoints.
When you need missing Star Citizen knowledge to answer accurately, use Star Citizen Wiki tools during your reasoning instead of guessing.
Use SCMDB tools when the user asks about Star Citizen missions/contracts, mission rewards, payouts, reputation gains, item rewards, blueprint rewards, or hauling mission cargo. Prefer SCMDB live data unless the user asks for PTU or a specific game version.
Use Star Citizen Wiki tools for general game knowledge, ships and vehicles, store availability, purchase locations, ship prices, manufacturers, locations, and page summaries from starcitizen.tools.
Use Wikelo ship project tools when the user asks for Wikelo ship requirements, Wikelo build materials, or what items are needed for a Wikelo ship project.
Use Cornerstone tools when the user asks where an item is sold, which shops carry an item, item store locations, in-game item base prices, or Universal Item Finder data.
When drafting UEX marketplace item posts that need images, use Cornerstone media tools or draft_marketplace_listing_with_cornerstone_image so the pending listing can include UEX image_data sourced from Cornerstone.
Prefer open and current UEX marketplace information. Do not use historical sale data, completed sale records, or sale/average-history information unless the user explicitly asks for historical sales.
Treat UEX marketplace prices as in-game aUEC/UEC credits, never real-world dollars, unless the user explicitly says otherwise.
For marketplace writes, draft the exact pending action and tell the user what will be sent; never claim it was sent until approval succeeds.
When drafting negotiation messages or marketplace replies, write like a real player would. Keep messages human, concise, and purposeful. Never include internal notes like "Tone note".
For continual plans, never invent an unknown parts checklist. If the required items cannot be derived from provided details or tools, create the plan in a needs-input state and say what item list is missing.
When a scheduled wake job fires, always write a concise Inbox-ready result that says what you checked, the key findings, and the suggested next action.
Keep prices, listing ids, slugs, users, and UEX status codes precise. If data is missing, say what you need next."""


class OllamaAgent:
    def __init__(
        self,
        base_url: str,
        model: str,
        tools: ToolRegistry,
        memory: MemoryStore | None = None,
        user_name: str | None = None,
        num_ctx: int | None = None,
        provider: str = "ollama",
        api_key: str | None = None,
        reasoning_effort: str = "medium",
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self.model = model
        self.tools = tools
        self.memory = memory
        self.user_name = user_name
        self.num_ctx = num_ctx
        self.provider = provider.strip().casefold() or "ollama"
        self.api_key = api_key
        self.reasoning_effort = reasoning_effort.strip().casefold() or "medium"
        self.thread_messages: dict[str, list[dict[str, Any]]] = {}

    async def health(self) -> dict[str, Any]:
        if self._is_openai_compatible_provider():
            return await self._openai_health()
        if self.provider == "codex":
            return await self._codex_health()
        try:
            async with httpx.AsyncClient(timeout=3) as client:
                response = await client.get(f"{self.base_url}/api/tags")
            response.raise_for_status()
            body = response.json()
        except (httpx.HTTPError, ValueError) as exc:
            return {
                "online": False,
                "model": self.model,
                "base_url": self.base_url,
                "message": f"Ollama is offline or unreachable at {self.base_url}. Open the Ollama tab and use the recommended action.",
                "detail": str(exc),
            }

        models = [model.get("name") or model.get("model") for model in body.get("models", [])]
        return {
            "online": True,
            "model": self.model,
            "base_url": self.base_url,
            "model_available": self.model in models,
            "models": models,
            "message": "Ollama is online.",
        }

    async def ensure_available(self) -> None:
        health = await self.health()
        if not health["online"]:
            raise OllamaUnavailable(health["message"])
        if health.get("model_available") is False:
            raise OllamaUnavailable(health["message"])

    async def chat(
        self,
        content: str,
        thread_id: str | None = DEFAULT_THREAD_ID,
        images: list[dict[str, Any]] | None = None,
    ) -> dict[str, Any]:
        await self.ensure_available()
        resolved_thread_id = self._thread_id(thread_id)
        messages = self._messages_for_thread(resolved_thread_id)
        previous_interaction = self.memory.last_interaction(resolved_thread_id) if self.memory else None
        normalized_images = self._normalize_images(images)
        prompt_text = self._prompt_text(content, len(normalized_images))
        memory_content = self._conversation_content(content, len(normalized_images))
        if self.memory:
            self.memory.add_conversation("user", memory_content, resolved_thread_id)
            await self._title_first_message(resolved_thread_id, prompt_text, previous_interaction)
        messages.append(self._user_message(prompt_text, normalized_images))
        last_tool_results: list[dict[str, Any]] = []
        image_scope = self.tools.chat_image_scope(normalized_images) if hasattr(self.tools, "chat_image_scope") else nullcontext()
        with image_scope:
            for _ in self._tool_rounds():
                try:
                    response = await self._chat_once(
                        prompt_text,
                        messages,
                        previous_interaction=previous_interaction,
                        thread_id=resolved_thread_id,
                    )
                except Exception as exc:
                    if not last_tool_results:
                        raise
                    answer = self._tool_result_fallback(
                        last_tool_results,
                        f"The {self._provider_label()} stopped after the tool call: {exc}",
                    )
                    messages.append({"role": "assistant", "content": answer})
                    if self.memory:
                        self.memory.add_conversation("assistant", answer, resolved_thread_id)
                    return {"message": answer, "pending_actions": self._pending_payloads(), "thread_id": resolved_thread_id}
                message = response.get("message") or {}
                tool_calls = message.get("tool_calls") or []
                if not tool_calls:
                    answer = message.get("content", "")
                    if not answer.strip():
                        answer = self._empty_response_fallback(last_tool_results)
                    messages.append({"role": "assistant", "content": answer})
                    if self.memory:
                        self.memory.add_conversation("assistant", answer, resolved_thread_id)
                    return {"message": answer, "pending_actions": self._pending_payloads(), "thread_id": resolved_thread_id}

                messages.append(message)
                for call in tool_calls:
                    name, arguments = self._extract_call(call)
                    result = await self.tools.execute(name, arguments)
                    last_tool_results.append({"tool": name, "result": result})
                    messages.append({"role": "tool", "tool_name": name, "tool_call_id": call.get("id"), "content": json.dumps(result)})
        fallback = self._tool_round_limit_message()
        messages.append({"role": "assistant", "content": fallback})
        if self.memory:
            self.memory.add_conversation("assistant", fallback, resolved_thread_id)
        return {"message": fallback, "pending_actions": self._pending_payloads(), "thread_id": resolved_thread_id}

    async def chat_events(
        self,
        content: str,
        thread_id: str | None = DEFAULT_THREAD_ID,
        images: list[dict[str, Any]] | None = None,
    ) -> AsyncIterator[dict[str, Any]]:
        health = await self.health()
        if not health["online"]:
            yield {"type": "warning", "message": health["message"]}
            yield {"type": "done", "pending_actions": self._pending_payloads()}
            return

        resolved_thread_id = self._thread_id(thread_id)
        messages = self._messages_for_thread(resolved_thread_id)
        previous_interaction = self.memory.last_interaction(resolved_thread_id) if self.memory else None
        normalized_images = self._normalize_images(images)
        prompt_text = self._prompt_text(content, len(normalized_images))
        memory_content = self._conversation_content(content, len(normalized_images))
        if self.memory:
            self.memory.add_conversation("user", memory_content, resolved_thread_id)
            await self._title_first_message(resolved_thread_id, prompt_text, previous_interaction)
        messages.append(self._user_message(prompt_text, normalized_images))
        yield {"type": "status", "message": "Thinking"}
        last_tool_results: list[dict[str, Any]] = []
        image_scope = self.tools.chat_image_scope(normalized_images) if hasattr(self.tools, "chat_image_scope") else nullcontext()
        with image_scope:
            for _ in self._tool_rounds():
                assistant_message: dict[str, Any] = {"role": "assistant", "content": ""}
                tool_calls: list[dict[str, Any]] = []

                try:
                    async for event in self._chat_stream_once(
                        prompt_text,
                        messages,
                        previous_interaction=previous_interaction,
                        thread_id=resolved_thread_id,
                    ):
                        if event.get("type") == "reasoning":
                            reasoning_chunk = event.get("content") or ""
                            if reasoning_chunk:
                                assistant_message["reasoning_content"] = assistant_message.get("reasoning_content", "") + reasoning_chunk
                                yield {"type": "reasoning", "content": reasoning_chunk}
                            continue
                        message = event.get("message") or {}
                        if message.get("reasoning_content"):
                            assistant_message["reasoning_content"] = message.get("reasoning_content")
                        chunk = message.get("content") or ""
                        if chunk:
                            assistant_message["content"] += chunk
                            yield {"type": "token", "content": chunk}
                        if message.get("tool_calls"):
                            tool_calls.extend(message["tool_calls"])
                        if event.get("done"):
                            metrics = self._stream_metrics(event)
                            if metrics:
                                yield {"type": "metrics", **metrics}
                except Exception as exc:
                    if not last_tool_results:
                        yield {"type": "warning", "message": f"Chat failed before any tool result was available: {exc}"}
                        yield {"type": "done", "pending_actions": self._pending_payloads(), "thread_id": resolved_thread_id}
                        return
                    fallback = self._tool_result_fallback(
                        last_tool_results,
                        f"The {self._provider_label()} stopped after the tool call: {exc}",
                    )
                    assistant_message["content"] = fallback
                    messages.append(assistant_message)
                    if self.memory:
                        self.memory.add_conversation("assistant", fallback, resolved_thread_id)
                    yield {"type": "token", "content": fallback}
                    yield {"type": "done", "pending_actions": self._pending_payloads(), "thread_id": resolved_thread_id}
                    return

                if not tool_calls:
                    if not assistant_message.get("content", "").strip():
                        fallback = self._empty_response_fallback(last_tool_results)
                        assistant_message["content"] = fallback
                        yield {"type": "token", "content": fallback}
                    messages.append(assistant_message)
                    if self.memory:
                        self.memory.add_conversation("assistant", assistant_message.get("content", ""), resolved_thread_id)
                    yield {"type": "done", "pending_actions": self._pending_payloads(), "thread_id": resolved_thread_id}
                    return

                assistant_message["tool_calls"] = tool_calls
                messages.append(assistant_message)
                for call in tool_calls:
                    name, arguments = self._extract_call(call)
                    yield {"type": "status", "message": self._tool_status(name)}
                    result = await self.tools.execute(name, arguments)
                    last_tool_results.append({"tool": name, "result": result})
                    messages.append({"role": "tool", "tool_name": name, "tool_call_id": call.get("id"), "content": json.dumps(result)})

                yield {"type": "status", "message": "Writing response"}
        fallback = self._tool_round_limit_message()
        messages.append({"role": "assistant", "content": fallback})
        if self.memory:
            self.memory.add_conversation("assistant", fallback, resolved_thread_id)
        yield {"type": "token", "content": fallback}
        yield {"type": "done", "pending_actions": self._pending_payloads(), "thread_id": resolved_thread_id}

    async def generate_wake_response(self, wake_message: str) -> str:
        await self.ensure_available()
        messages = self._messages_for_thread("wake")
        previous_interaction = self.memory.last_interaction("wake") if self.memory else None
        messages.append({"role": "user", "content": wake_message})
        last_tool_results: list[dict[str, Any]] = []
        for _ in self._tool_rounds():
            try:
                response = await self._chat_once(
                    wake_message,
                    messages,
                    previous_interaction=previous_interaction,
                    thread_id="wake",
                )
            except Exception as exc:
                if not last_tool_results:
                    raise
                content = self._tool_result_fallback(
                    last_tool_results,
                    f"The {self._provider_label()} stopped after the wake-job tool call: {exc}",
                )
                messages.append({"role": "assistant", "content": content})
                if self.memory:
                    self.memory.add_conversation("system", wake_message, "wake")
                    self.memory.add_conversation("assistant", content, "wake")
                return content
            message = response.get("message") or {}
            tool_calls = message.get("tool_calls") or []
            if not tool_calls:
                content = message.get("content", "")
                if not content.strip():
                    content = self._empty_response_fallback(last_tool_results)
                messages.append({"role": "assistant", "content": content})
                if self.memory:
                    self.memory.add_conversation("system", wake_message, "wake")
                    self.memory.add_conversation("assistant", content, "wake")
                return content

            messages.append(message)
            for call in tool_calls:
                name, arguments = self._extract_call(call)
                result = await self.tools.execute(name, arguments)
                last_tool_results.append({"tool": name, "result": result})
                messages.append({"role": "tool", "tool_name": name, "tool_call_id": call.get("id"), "content": json.dumps(result)})
        content = self._wake_tool_round_limit_message()
        messages.append({"role": "assistant", "content": content})
        if self.memory:
            self.memory.add_conversation("system", wake_message, "wake")
            self.memory.add_conversation("assistant", content, "wake")
        return content

    async def _chat_once(
        self,
        query: str = "",
        messages: list[dict[str, Any]] | None = None,
        previous_interaction: dict[str, Any] | None = None,
        thread_id: str | None = DEFAULT_THREAD_ID,
    ) -> dict[str, Any]:
        if self._is_openai_compatible_provider():
            return await self._openai_chat(
                query,
                messages,
                previous_interaction=previous_interaction,
                thread_id=thread_id,
            )
        if self.provider == "codex":
            return await self._codex_chat(
                query,
                messages,
                previous_interaction=previous_interaction,
                thread_id=thread_id,
            )
        return await self._ollama_chat(
            query,
            messages,
            previous_interaction=previous_interaction,
            thread_id=thread_id,
        )

    async def _chat_stream_once(
        self,
        query: str = "",
        messages: list[dict[str, Any]] | None = None,
        previous_interaction: dict[str, Any] | None = None,
        thread_id: str | None = DEFAULT_THREAD_ID,
    ) -> AsyncIterator[dict[str, Any]]:
        if self._is_openai_compatible_provider():
            async for event in self._openai_chat_stream(
                query,
                messages,
                previous_interaction=previous_interaction,
                thread_id=thread_id,
            ):
                yield event
            return
        if self.provider == "codex":
            async for event in self._codex_chat_stream(
                query,
                messages,
                previous_interaction=previous_interaction,
                thread_id=thread_id,
            ):
                yield event
            return
        async for event in self._ollama_chat_stream(
            query,
            messages,
            previous_interaction=previous_interaction,
            thread_id=thread_id,
        ):
            yield event

    async def _ollama_chat(
        self,
        query: str = "",
        messages: list[dict[str, Any]] | None = None,
        previous_interaction: dict[str, Any] | None = None,
        thread_id: str | None = DEFAULT_THREAD_ID,
    ) -> dict[str, Any]:
        async with httpx.AsyncClient(timeout=120) as client:
            response = await client.post(
                f"{self.base_url}/api/chat",
                json={
                    "model": self.model,
                    "messages": self._messages_with_context(
                        query,
                        messages or self._messages_for_thread(thread_id),
                        previous_interaction=previous_interaction,
                        thread_id=thread_id,
                    ),
                    "tools": self.tools.schemas,
                    "options": self._ollama_options(),
                    "stream": False,
                },
            )
        response.raise_for_status()
        return response.json()

    async def _ollama_chat_stream(
        self,
        query: str = "",
        messages: list[dict[str, Any]] | None = None,
        previous_interaction: dict[str, Any] | None = None,
        thread_id: str | None = DEFAULT_THREAD_ID,
    ) -> AsyncIterator[dict[str, Any]]:
        async with httpx.AsyncClient(timeout=120) as client:
            async with client.stream(
                "POST",
                f"{self.base_url}/api/chat",
                json={
                    "model": self.model,
                    "messages": self._messages_with_context(
                        query,
                        messages or self._messages_for_thread(thread_id),
                        previous_interaction=previous_interaction,
                        thread_id=thread_id,
                    ),
                    "tools": self.tools.schemas,
                    "options": self._ollama_options(),
                    "stream": True,
                },
            ) as response:
                response.raise_for_status()
                async for line in response.aiter_lines():
                    if line:
                        yield json.loads(line)

    async def _openai_chat(
        self,
        query: str = "",
        messages: list[dict[str, Any]] | None = None,
        previous_interaction: dict[str, Any] | None = None,
        thread_id: str | None = DEFAULT_THREAD_ID,
    ) -> dict[str, Any]:
        async with httpx.AsyncClient(timeout=120) as client:
            response = await client.post(
                f"{self.base_url}/chat/completions",
                headers=self._openai_headers(),
                json={
                    "model": self.model,
                    "messages": self._openai_messages(
                        query,
                        messages or self._messages_for_thread(thread_id),
                        previous_interaction=previous_interaction,
                        thread_id=thread_id,
                    ),
                    "tools": self.tools.schemas,
                    "stream": False,
                    **self._openai_request_options(stream=False),
                },
            )
        response.raise_for_status()
        body = response.json()
        choice = (body.get("choices") or [{}])[0]
        message = choice.get("message") or {}
        return {
            "message": {
                "role": message.get("role", "assistant"),
                "reasoning_content": message.get("reasoning_content") or "",
                "content": message.get("content") or "",
                "tool_calls": message.get("tool_calls") or [],
            }
        }

    async def _openai_chat_stream(
        self,
        query: str = "",
        messages: list[dict[str, Any]] | None = None,
        previous_interaction: dict[str, Any] | None = None,
        thread_id: str | None = DEFAULT_THREAD_ID,
    ) -> AsyncIterator[dict[str, Any]]:
        tool_calls: dict[int, dict[str, Any]] = {}
        async with httpx.AsyncClient(timeout=120) as client:
            async with client.stream(
                "POST",
                f"{self.base_url}/chat/completions",
                headers=self._openai_headers(),
                json={
                    "model": self.model,
                    "messages": self._openai_messages(
                        query,
                        messages or self._messages_for_thread(thread_id),
                        previous_interaction=previous_interaction,
                        thread_id=thread_id,
                    ),
                    "tools": self.tools.schemas,
                    "stream": True,
                    **self._openai_request_options(stream=True),
                },
            ) as response:
                response.raise_for_status()
                async for line in response.aiter_lines():
                    if not line or not line.startswith("data:"):
                        continue
                    payload = line.removeprefix("data:").strip()
                    if not payload:
                        continue
                    if payload == "[DONE]":
                        break
                    event = json.loads(payload)
                    if event.get("usage"):
                        metrics = self._cloud_usage_metrics(event["usage"])
                        if metrics:
                            yield {"type": "metrics", **metrics}
                    choice = (event.get("choices") or [{}])[0]
                    delta = choice.get("delta") or {}
                    reasoning_content = delta.get("reasoning_content") or ""
                    if reasoning_content:
                        yield {"type": "reasoning", "content": reasoning_content}
                    content = delta.get("content") or ""
                    if content:
                        yield {"message": {"role": "assistant", "content": content}}
                    for tool_call in delta.get("tool_calls") or []:
                        self._merge_openai_tool_call(tool_calls, tool_call)
                    finish_reason = choice.get("finish_reason")
                    if finish_reason:
                        message = choice.get("message") or {}
                        yield {
                            "message": {
                                "role": "assistant",
                                "reasoning_content": message.get("reasoning_content") or "",
                                "content": "",
                                "tool_calls": self._ordered_tool_calls(tool_calls),
                            },
                            "done": True,
                        }
                        return
        yield {
            "message": {
                "role": "assistant",
                "content": "",
                "tool_calls": self._ordered_tool_calls(tool_calls),
            },
            "done": True,
        }

    async def _codex_chat(
        self,
        query: str = "",
        messages: list[dict[str, Any]] | None = None,
        previous_interaction: dict[str, Any] | None = None,
        thread_id: str | None = DEFAULT_THREAD_ID,
    ) -> dict[str, Any]:
        result = await self._codex_cli_turn(
            query,
            messages or self._messages_for_thread(thread_id),
            previous_interaction=previous_interaction,
            thread_id=thread_id,
        )
        return self._codex_structured_response(result)

    async def _codex_chat_stream(
        self,
        query: str = "",
        messages: list[dict[str, Any]] | None = None,
        previous_interaction: dict[str, Any] | None = None,
        thread_id: str | None = DEFAULT_THREAD_ID,
    ) -> AsyncIterator[dict[str, Any]]:
        result = await self._codex_cli_turn(
            query,
            messages or self._messages_for_thread(thread_id),
            previous_interaction=previous_interaction,
            thread_id=thread_id,
        )
        response = self._codex_structured_response(result)
        message = response["message"]
        if message.get("content"):
            yield {"message": {"role": "assistant", "content": message["content"]}}
        yield {
            "message": {
                "role": "assistant",
                "content": "",
                "tool_calls": message.get("tool_calls") or [],
            },
            "done": True,
        }

    def _messages_with_context(
        self,
        query: str,
        messages: list[dict[str, Any]],
        previous_interaction: dict[str, Any] | None = None,
        thread_id: str | None = DEFAULT_THREAD_ID,
    ) -> list[dict[str, Any]]:
        attached_image_count = 0
        for message in reversed(messages):
            if message.get("role") != "user":
                continue
            attached_image_count = len(message.get("images") or [])
            break
        stable_context, volatile_context = self._runtime_context_parts(
            query,
            previous_interaction=previous_interaction,
            thread_id=thread_id,
            attached_image_count=attached_image_count,
        )
        contexts = [part for part in ("\n".join(stable_context), "\n".join(volatile_context)) if part]
        if not contexts:
            return messages
        return [messages[0], *({"role": "system", "content": context} for context in contexts), *messages[1:]]

    async def _openai_health(self) -> dict[str, Any]:
        return await self._cloud_health(self.provider if self._is_openai_compatible_provider() else "openai")

    async def _codex_health(self) -> dict[str, Any]:
        command = self._codex_command()
        if not command:
            return {
                "online": False,
                "model": self.model,
                "base_url": self.base_url,
                "provider": "codex",
                "model_available": False,
                "models": [],
                "message": "Codex CLI was not found on PATH.",
                "detail": "",
            }
        try:
            account, models = await self._codex_app_server_status()
        except Exception as exc:
            return {
                "online": False,
                "model": self.model,
                "base_url": command,
                "provider": "codex",
                "model_available": False,
                "models": [],
                "message": "Codex App Server is installed, but TraderAI could not connect to it.",
                "detail": str(exc),
            }
        logged_in = bool(account)
        detail = f"Logged in as {account.get('email')}" if isinstance(account, dict) and account.get("email") else ""
        return {
            "online": logged_in,
            "model": self.model,
            "base_url": command,
            "provider": "codex",
            "model_available": self.model in models if models else bool(self.model),
            "models": models,
            "message": "Codex App Server is online." if logged_in else "Codex CLI is installed, but not logged in with ChatGPT.",
            "detail": detail,
        }

    async def _cloud_health(self, provider: str) -> dict[str, Any]:
        if not self.api_key:
            return {
                "online": False,
                "model": self.model,
                "base_url": self.base_url,
                "provider": provider,
                "model_available": False,
                "models": [],
                "message": f"{self._provider_label()} is selected, but no API key is configured.",
                "detail": "",
            }
        try:
            async with httpx.AsyncClient(timeout=10) as client:
                response = await client.get(f"{self.base_url}/models", headers=self._openai_headers())
            response.raise_for_status()
            body = response.json()
        except (httpx.HTTPError, ValueError) as exc:
            return {
                "online": False,
                "model": self.model,
                "base_url": self.base_url,
                "provider": provider,
                "model_available": False,
                "models": [],
                "message": f"{self._provider_label()} is unreachable at {self.base_url} or rejected the API key.",
                "detail": str(exc),
            }
        models = sorted(item.get("id") for item in body.get("data", []) if item.get("id"))
        return {
            "online": True,
            "model": self.model,
            "base_url": self.base_url,
            "provider": provider,
            "model_available": self.model in models,
            "models": models,
            "message": f"{self._provider_label()} is online.",
        }

    def _openai_headers(self) -> dict[str, str]:
        return {
            "Authorization": f"Bearer {self.api_key or ''}",
            "Content-Type": "application/json",
        }

    def _openai_request_options(self, stream: bool) -> dict[str, Any]:
        if self.provider == "deepseek":
            options: dict[str, Any] = {}
            if self.reasoning_effort in {"none", "minimal"}:
                options["thinking"] = {"type": "disabled"}
            else:
                options["thinking"] = {"type": "enabled"}
                options["reasoning_effort"] = "max" if self.reasoning_effort in {"xhigh", "max"} else "high"
            if stream:
                options["stream_options"] = {"include_usage": True}
            return options
        return {"reasoning_effort": self.reasoning_effort}

    def _openai_messages(
        self,
        query: str,
        messages: list[dict[str, Any]],
        previous_interaction: dict[str, Any] | None = None,
        thread_id: str | None = DEFAULT_THREAD_ID,
    ) -> list[dict[str, Any]]:
        normalized: list[dict[str, Any]] = []
        for message in self._messages_with_context(
            query,
            messages,
            previous_interaction=previous_interaction,
            thread_id=thread_id,
        ):
            role = message.get("role")
            if role not in {"system", "user", "assistant", "tool"}:
                continue
            entry: dict[str, Any] = {"role": role, "content": message.get("content", "")}
            if role == "user" and message.get("images"):
                text_content = message.get("content", "")
                content_parts: list[dict[str, Any]] = []
                content_types = list(message.get("image_content_types") or [])
                if text_content:
                    content_parts.append({"type": "text", "text": text_content})
                for index, image_data in enumerate(message.get("images") or []):
                    content_type = content_types[index] if index < len(content_types) else "image/png"
                    content_parts.append(
                        {
                            "type": "image_url",
                            "image_url": {"url": f"data:{content_type};base64,{image_data}"},
                        }
                    )
                entry["content"] = content_parts
            if role == "assistant" and message.get("tool_calls"):
                entry["tool_calls"] = message["tool_calls"]
            if role == "assistant" and message.get("reasoning_content"):
                entry["reasoning_content"] = message["reasoning_content"]
            if role == "tool":
                entry["tool_call_id"] = message.get("tool_call_id") or message.get("tool_name") or "tool"
            normalized.append(entry)
        return normalized

    def _codex_tool_catalog(self) -> list[dict[str, Any]]:
        tools: list[dict[str, Any]] = []
        for schema in self.tools.schemas:
            if schema.get("type") != "function":
                continue
            function = schema.get("function") or {}
            tools.append(
                {
                    "name": function.get("name", ""),
                    "description": function.get("description", ""),
                    "parameters": function.get("parameters") or {"type": "object", "properties": {}},
                }
            )
        return tools

    def _provider_label(self) -> str:
        if self.provider == "openai":
            return "OpenAI model"
        if self.provider == "deepseek":
            return "DeepSeek model"
        if self.provider == "codex":
            return "Codex model"
        return "local model"

    def _is_openai_compatible_provider(self) -> bool:
        return self.provider in {"openai", "deepseek"}

    def _tool_rounds(self):
        if self.provider == "deepseek":
            while True:
                yield None
            return
        for _ in range(10):
            yield None

    def _tool_round_limit_message(self) -> str:
        return "I hit the tool-call limit while working on that. Try narrowing the request or approve any pending action first."

    def _wake_tool_round_limit_message(self) -> str:
        return "I hit the tool-call limit while running this scheduled wake job. Check the job prompt or pending approvals."

    @staticmethod
    def _merge_openai_tool_call(target: dict[int, dict[str, Any]], delta: dict[str, Any]) -> None:
        index = int(delta.get("index") or 0)
        current = target.setdefault(index, {"id": delta.get("id"), "type": "function", "function": {"name": "", "arguments": ""}})
        if delta.get("id"):
            current["id"] = delta["id"]
        function = delta.get("function") or {}
        current_function = current.setdefault("function", {"name": "", "arguments": ""})
        if function.get("name"):
            current_function["name"] += function["name"]
        if function.get("arguments"):
            current_function["arguments"] += function["arguments"]

    @staticmethod
    def _ordered_tool_calls(tool_calls: dict[int, dict[str, Any]]) -> list[dict[str, Any]]:
        return [tool_calls[index] for index in sorted(tool_calls)]

    async def _codex_cli_turn(
        self,
        query: str,
        messages: list[dict[str, Any]],
        previous_interaction: dict[str, Any] | None = None,
        thread_id: str | None = DEFAULT_THREAD_ID,
    ) -> dict[str, Any]:
        return await self._codex_app_server_turn(
            query,
            messages,
            previous_interaction=previous_interaction,
            thread_id=thread_id,
        )

    async def _codex_app_server_turn(
        self,
        query: str,
        messages: list[dict[str, Any]],
        previous_interaction: dict[str, Any] | None = None,
        thread_id: str | None = DEFAULT_THREAD_ID,
    ) -> dict[str, Any]:
        prompt = self._codex_cli_prompt(
            query,
            messages,
            previous_interaction=previous_interaction,
            thread_id=thread_id,
        )
        final_text = ""

        process = await self._start_codex_app_server()
        request_id = 1

        async def send_request(method: str, params: dict[str, Any] | None = None, timeout: int = 120) -> dict[str, Any]:
            nonlocal request_id
            current_id = request_id
            request_id += 1
            payload: dict[str, Any] = {"jsonrpc": "2.0", "id": current_id, "method": method}
            if params is not None:
                payload["params"] = params
            await self._codex_app_server_write(process, payload)
            while True:
                message = await self._codex_app_server_read(process, timeout=timeout)
                if message.get("id") == current_id:
                    if message.get("error"):
                        error = message["error"]
                        raise RuntimeError(error.get("message") or f"Codex App Server request failed: {error}")
                    return message.get("result") or {}
                await self._handle_codex_app_server_message(process, message)

        try:
            await send_request(
                "initialize",
                {
                    "clientInfo": {"name": "TraderAI", "version": __version__},
                    "capabilities": {"experimentalApi": True},
                },
                timeout=30,
            )
            await self._codex_app_server_write(process, {"jsonrpc": "2.0", "method": "initialized", "params": {}})
            thread = await send_request(
                "thread/start",
                {
                    "model": self.model,
                    "modelProvider": None,
                    "cwd": str(Path.cwd()),
                    "approvalPolicy": "never",
                    "sandbox": "read-only",
                    "baseInstructions": "You are TraderAI running through the local Codex App Server using ChatGPT OAuth.",
                    "developerInstructions": (
                        "Do not run shell commands, inspect files, or modify the workspace. "
                        "Answer only with JSON matching the requested output schema."
                    ),
                    "ephemeral": True,
                    "experimentalRawEvents": False,
                    "persistExtendedHistory": False,
                },
                timeout=30,
            )
            thread_id_value = ((thread.get("thread") or {}).get("id") or thread.get("threadId") or "").strip()
            if not thread_id_value:
                raise RuntimeError(f"Codex App Server did not return a thread id: {thread!r}")
            turn = await send_request(
                "turn/start",
                {
                    "threadId": thread_id_value,
                    "input": [{"type": "text", "text": prompt, "text_elements": []}],
                    "cwd": str(Path.cwd()),
                    "approvalPolicy": "never",
                    "sandboxPolicy": {"type": "readOnly", "access": {"type": "fullAccess"}},
                    "model": self.model,
                    "effort": self.reasoning_effort,
                    "summary": "none",
                    "outputSchema": self._codex_output_schema(),
                },
                timeout=60,
            )
            turn_id = ((turn.get("turn") or {}).get("id") or "").strip()
            if not turn_id:
                raise RuntimeError(f"Codex App Server did not return a turn id: {turn!r}")
            while True:
                message = await self._codex_app_server_read(process, timeout=240)
                method = message.get("method")
                params = message.get("params") or {}
                if method == "item/agentMessage/delta" and params.get("turnId") == turn_id:
                    final_text += params.get("delta") or ""
                elif method == "item/completed" and params.get("turnId") == turn_id:
                    item = params.get("item") or {}
                    if item.get("type") == "agentMessage":
                        final_text = item.get("text") or final_text
                elif method == "turn/completed" and (params.get("turn") or {}).get("id") == turn_id:
                    turn_status = (params.get("turn") or {}).get("status")
                    if turn_status != "completed":
                        error = (params.get("turn") or {}).get("error") or {}
                        raise RuntimeError(error.get("message") or f"Codex App Server turn ended with status {turn_status}.")
                    break
                elif method == "error":
                    error = params.get("message") or params.get("error") or params
                    raise RuntimeError(f"Codex App Server error: {error}")
                else:
                    await self._handle_codex_app_server_message(process, message)
        finally:
            await self._stop_codex_app_server(process)
        return self._parse_codex_app_server_text(final_text)

    def _codex_cli_prompt(
        self,
        query: str,
        messages: list[dict[str, Any]],
        previous_interaction: dict[str, Any] | None = None,
        thread_id: str | None = DEFAULT_THREAD_ID,
    ) -> str:
        conversation_lines: list[str] = []
        for message in self._messages_with_context(
            query,
            messages,
            previous_interaction=previous_interaction,
            thread_id=thread_id,
        ):
            role = message.get("role", "unknown")
            content = message.get("content", "")
            suffix = ""
            if role == "user" and message.get("images"):
                suffix = f" [attached images: {len(message.get('images') or [])}]"
            if role == "tool":
                suffix = f" [tool {message.get('tool_name') or ''}]"
            if role == "assistant" and message.get("tool_calls"):
                suffix = f" [tool calls: {json.dumps(message.get('tool_calls'), ensure_ascii=True)}]"
            conversation_lines.append(f"{role}{suffix}: {content}")
        tools_json = json.dumps(self._codex_tool_catalog(), ensure_ascii=True, indent=2)
        return (
            "You are TraderAI running through the local Codex App Server using ChatGPT OAuth.\n"
            "Do not run shell commands, inspect files, or modify the workspace.\n"
            "Your only job is to decide whether to answer directly or request exactly one TraderAI tool.\n\n"
            "Return JSON that matches the provided schema.\n"
            "- If you can answer now, set kind to final, put the user-facing reply in message, set tool_name to an empty string, and set arguments_json to '{}'.\n"
            "- If you need a tool, set kind to tool_call, set tool_name to the exact tool name, set message to an empty string, and set arguments_json to a valid JSON object string.\n"
            "- Never return more than one tool call at a time.\n"
            "- Prefer the TraderAI tools over guessing.\n\n"
            f"Available tools:\n{tools_json}\n\n"
            "Conversation transcript:\n"
            + "\n".join(conversation_lines)
        )

    def _codex_structured_response(self, result: dict[str, Any]) -> dict[str, Any]:
        if result.get("kind") == "tool_call":
            tool_name = str(result.get("tool_name") or "").strip()
            arguments_json = str(result.get("arguments_json") or "{}").strip() or "{}"
            return {
                "message": {
                    "role": "assistant",
                    "content": "",
                    "tool_calls": [
                        {
                            "id": f"codex-{uuid.uuid4()}",
                            "type": "function",
                            "function": {
                                "name": tool_name,
                                "arguments": arguments_json,
                            },
                        }
                    ],
                }
            }
        return {
            "message": {
                "role": "assistant",
                "content": str(result.get("message") or ""),
                "tool_calls": [],
            }
        }

    def _write_codex_schema(self) -> str:
        schema = self._codex_output_schema()
        with tempfile.NamedTemporaryFile("w", suffix="-traderai-codex-schema.json", delete=False, encoding="utf-8") as handle:
            json.dump(schema, handle, ensure_ascii=True)
            return handle.name

    @staticmethod
    def _codex_output_schema() -> dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "kind": {"type": "string", "enum": ["final", "tool_call"]},
                "message": {"type": "string"},
                "tool_name": {"type": "string"},
                "arguments_json": {"type": "string"},
            },
            "required": ["kind", "message", "tool_name", "arguments_json"],
            "additionalProperties": False,
        }

    def _parse_codex_app_server_text(self, final_text: str) -> dict[str, Any]:
        if not final_text.strip():
            raise RuntimeError("Codex App Server returned an empty response.")
        try:
            parsed = json.loads(final_text)
        except ValueError as exc:
            raise RuntimeError(f"Codex App Server returned non-JSON output: {final_text}") from exc
        if parsed.get("kind") not in {"final", "tool_call"}:
            raise RuntimeError(f"Codex App Server returned an invalid result kind: {parsed!r}")
        return parsed

    def _parse_codex_exec_output(self, output: dict[str, Any]) -> dict[str, Any]:
        events = output.get("events") or []
        final_text = ""
        error_text = ""
        for event in events:
            if event.get("type") == "item.completed":
                item = event.get("item") or {}
                if item.get("type") == "agent_message":
                    final_text = item.get("text") or final_text
            elif event.get("type") == "error":
                error_text = event.get("message") or error_text
            elif event.get("type") == "turn.failed":
                details = event.get("error") or {}
                error_text = details.get("message") or error_text
        if output.get("returncode") != 0 and not final_text:
            raise RuntimeError(error_text or output.get("stderr") or "Codex CLI failed.")
        try:
            parsed = json.loads(final_text)
        except ValueError as exc:
            raise RuntimeError(f"Codex CLI returned non-JSON output: {final_text}") from exc
        if parsed.get("kind") not in {"final", "tool_call"}:
            raise RuntimeError(f"Codex CLI returned an invalid result kind: {parsed!r}")
        return parsed

    def _codex_command(self, required: bool = False) -> str | None:
        configured = self.base_url.strip() if self.base_url else "codex"
        resolved = shutil.which(configured) or configured
        if required and not Path(resolved).exists() and shutil.which(resolved) is None:
            raise RuntimeError("Codex CLI was not found on PATH.")
        return resolved

    async def _codex_app_server_status(self) -> tuple[dict[str, Any] | None, list[str]]:
        process = await self._start_codex_app_server()
        request_id = 1

        async def send_request(method: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
            nonlocal request_id
            current_id = request_id
            request_id += 1
            payload: dict[str, Any] = {"jsonrpc": "2.0", "id": current_id, "method": method}
            if params is not None:
                payload["params"] = params
            await self._codex_app_server_write(process, payload)
            while True:
                message = await self._codex_app_server_read(process, timeout=30)
                if message.get("id") == current_id:
                    if message.get("error"):
                        error = message["error"]
                        raise RuntimeError(error.get("message") or f"Codex App Server request failed: {error}")
                    return message.get("result") or {}
                await self._handle_codex_app_server_message(process, message)

        try:
            await send_request(
                "initialize",
                {
                    "clientInfo": {"name": "TraderAI", "version": __version__},
                    "capabilities": {"experimentalApi": True},
                },
            )
            await self._codex_app_server_write(process, {"jsonrpc": "2.0", "method": "initialized", "params": {}})
            account_result = await send_request("account/read", {"refreshToken": False})
            models: list[str] = []
            cursor: str | None = None
            for _ in range(20):
                params: dict[str, Any] = {"limit": 50, "includeHidden": False}
                if cursor:
                    params["cursor"] = cursor
                page = await send_request("model/list", params)
                for item in page.get("data") or []:
                    model = item.get("id") or item.get("model")
                    if model:
                        models.append(model)
                cursor = page.get("nextCursor")
                if not cursor:
                    break
            return account_result.get("account"), sorted(set(models))
        finally:
            await self._stop_codex_app_server(process)

    async def _start_codex_app_server(self) -> asyncio.subprocess.Process:
        return await asyncio.create_subprocess_exec(
            self._codex_command(required=True),
            "app-server",
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

    async def _codex_app_server_write(self, process: asyncio.subprocess.Process, payload: dict[str, Any]) -> None:
        if process.stdin is None:
            raise RuntimeError("Codex App Server stdin is unavailable.")
        process.stdin.write((json.dumps(payload, ensure_ascii=True) + "\n").encode("utf-8"))
        await process.stdin.drain()

    async def _codex_app_server_read(self, process: asyncio.subprocess.Process, timeout: int) -> dict[str, Any]:
        if process.stdout is None:
            raise RuntimeError("Codex App Server stdout is unavailable.")
        try:
            line = await asyncio.wait_for(process.stdout.readline(), timeout=timeout)
        except TimeoutError as exc:
            raise RuntimeError("Codex App Server timed out.") from exc
        if not line:
            stderr = ""
            if process.stderr is not None:
                try:
                    stderr = (await asyncio.wait_for(process.stderr.read(), timeout=1)).decode("utf-8", errors="replace").strip()
                except TimeoutError:
                    stderr = ""
            raise RuntimeError(stderr or "Codex App Server exited without a response.")
        try:
            return json.loads(line.decode("utf-8", errors="replace"))
        except ValueError as exc:
            raise RuntimeError(f"Codex App Server returned invalid JSON-RPC: {line!r}") from exc

    async def _handle_codex_app_server_message(self, process: asyncio.subprocess.Process, message: dict[str, Any]) -> None:
        if "id" not in message or "method" not in message:
            return
        method = message.get("method")
        if method in {
            "item/commandExecution/requestApproval",
            "item/fileChange/requestApproval",
            "applyPatchApproval",
            "execCommandApproval",
        }:
            await self._codex_app_server_write(
                process,
                {
                    "jsonrpc": "2.0",
                    "id": message["id"],
                    "result": {
                        "decision": "deny",
                        "message": "TraderAI does not allow Codex to run commands or change files.",
                    },
                },
            )
            return
        await self._codex_app_server_write(
            process,
            {
                "jsonrpc": "2.0",
                "id": message["id"],
                "error": {"code": -32601, "message": f"TraderAI does not handle Codex App Server request {method}."},
            },
        )

    async def _stop_codex_app_server(self, process: asyncio.subprocess.Process) -> None:
        if process.returncode is not None:
            return
        process.terminate()
        try:
            await asyncio.wait_for(process.wait(), timeout=3)
        except TimeoutError:
            process.kill()
            await process.wait()

    async def _run_command(self, command: list[str], timeout: int = 120, stdin_text: str | None = None) -> dict[str, Any]:
        process = await asyncio.create_subprocess_exec(
            *command,
            stdin=asyncio.subprocess.PIPE if stdin_text is not None else None,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        try:
            payload = stdin_text.encode("utf-8") if stdin_text is not None else None
            stdout, stderr = await asyncio.wait_for(process.communicate(payload), timeout=timeout)
        except TimeoutError:
            process.kill()
            await process.communicate()
            raise RuntimeError(f"Command timed out: {' '.join(command[:3])}")
        stdout_text = stdout.decode("utf-8", errors="replace")
        stderr_text = stderr.decode("utf-8", errors="replace")
        events = []
        for line in stdout_text.splitlines():
            line = line.strip()
            if not line:
                continue
            try:
                events.append(json.loads(line))
            except ValueError:
                events.append({"type": "stdout", "text": line})
        return {
            "returncode": process.returncode,
            "stdout": stdout_text,
            "stderr": stderr_text,
            "events": events,
        }

    def _codex_model_cache(self) -> list[str]:
        cache_path = Path.home() / ".codex" / "models_cache.json"
        if not cache_path.exists():
            return []
        try:
            body = json.loads(cache_path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            return []
        models = []
        for item in body.get("models", []):
            slug = item.get("slug")
            if slug:
                models.append(slug)
        return sorted(set(models))

    def _runtime_context_parts(
        self,
        query: str,
        previous_interaction: dict[str, Any] | None = None,
        thread_id: str | None = DEFAULT_THREAD_ID,
        attached_image_count: int = 0,
    ) -> tuple[list[str], list[str]]:
        local_zone = get_localzone()
        stable_parts: list[str] = []
        volatile_parts = [
            f"Current local date/time: {iso_now()} UTC; {iso_now_in_zone(local_zone)} {local_zone}.",
        ]
        if attached_image_count:
            label = "image" if attached_image_count == 1 else "images"
            volatile_parts.append(
                f"Current user message includes {attached_image_count} pasted {label}. "
                "You can inspect them visually. If the user wants one reused in a marketplace listing draft, "
                "call draft_marketplace_listing or draft_marketplace_listing_with_cornerstone_image with "
                "use_attached_image=true and attached_image_index when needed."
            )
        uex = getattr(self.tools, "uex", None)
        if uex:
            auth_methods = []
            if uex.secret_key:
                auth_methods.append("secret key")
            if uex.bearer_token:
                auth_methods.append("bearer token")
            if auth_methods:
                stable_parts.append(
                    "UEX API authentication is configured server-side with "
                    + " and ".join(auth_methods)
                    + "; use authenticated UEX tools directly and do not ask for tokens."
                )
            else:
                stable_parts.append("UEX API authentication is not configured server-side.")
        if self.user_name:
            stable_parts.append(f"Known user name/handle: {self.user_name}.")

        if self.memory is None:
            return stable_parts, volatile_parts

        profile = self.memory.get_profile()
        if profile:
            identity = self._profile_identity(profile)
            if identity:
                stable_parts.append(identity)
            stable_parts.append(f"Known user profile JSON: {json.dumps(self._profile_for_prompt(profile), ensure_ascii=True)}.")

        last = previous_interaction if previous_interaction is not None else self.memory.last_interaction(thread_id)
        if last:
            volatile_parts.append(
                f"Previous interaction before this message: {last['created_at']} "
                f"({time_since(last['created_at'])}, role {last['role']})."
            )
        else:
            volatile_parts.append("Previous interaction before this message: none recorded.")

        memories = self.memory.recall(query, limit=6)
        if memories:
            memory_text = "\n".join(
                f"- [{item['kind']}, importance {item['importance']}] {item['content']}"
                for item in memories
            )
            volatile_parts.append(f"Relevant long-term memories:\n{memory_text}")

        return stable_parts, volatile_parts

    def _runtime_context(
        self,
        query: str,
        previous_interaction: dict[str, Any] | None = None,
        thread_id: str | None = DEFAULT_THREAD_ID,
        attached_image_count: int = 0,
    ) -> str:
        stable_parts, volatile_parts = self._runtime_context_parts(
            query,
            previous_interaction=previous_interaction,
            thread_id=thread_id,
            attached_image_count=attached_image_count,
        )
        return "\n".join(part for part in ("\n".join(stable_parts), "\n".join(volatile_parts)) if part)

    def _messages_for_thread(self, thread_id: str | None) -> list[dict[str, Any]]:
        resolved_thread_id = self._thread_id(thread_id)
        if resolved_thread_id not in self.thread_messages:
            messages: list[dict[str, Any]] = [{"role": "system", "content": SYSTEM_PROMPT}]
            if self.memory:
                self.memory.ensure_thread(resolved_thread_id)
                for item in self.memory.recent_conversation(limit=30, thread_id=resolved_thread_id):
                    role = item.get("role")
                    if role in {"user", "assistant"} and item.get("content"):
                        messages.append({"role": role, "content": item["content"]})
            self.thread_messages[resolved_thread_id] = messages
        return self.thread_messages[resolved_thread_id]

    async def _title_first_message(
        self,
        thread_id: str,
        first_message: str,
        previous_interaction: dict[str, Any] | None,
    ) -> None:
        if self.memory is None or previous_interaction is not None:
            return
        thread = self.memory.get_thread(thread_id)
        if not thread or thread.get("title") != "New chat":
            return
        title = await self._generate_chat_title(first_message)
        self.memory.rename_thread(thread_id, title or MemoryStore._thread_title(first_message))

    async def _generate_chat_title(self, first_message: str) -> str:
        prompt = (
            "Create a concise chat title for this first user message. "
            "Use 2 to 6 words. No quotes, no punctuation at the end, no preamble.\n\n"
            f"Message: {first_message[:800]}"
        )
        try:
            if self._is_openai_compatible_provider():
                async with httpx.AsyncClient(timeout=20) as client:
                    response = await client.post(
                        f"{self.base_url}/chat/completions",
                        headers=self._openai_headers(),
                        json={
                            "model": self.model,
                            "messages": [
                                {"role": "system", "content": "You write short chat titles."},
                                {"role": "user", "content": prompt},
                            ],
                            "stream": False,
                            **self._openai_request_options(stream=False),
                        },
                    )
                response.raise_for_status()
                choice = (response.json().get("choices") or [{}])[0]
                message = choice.get("message") or {}
                return self._clean_generated_title(message.get("content", ""))
            if self.provider == "codex":
                result = await self._codex_app_server_turn(
                    prompt,
                    [
                        {"role": "system", "content": "You write short chat titles."},
                        {"role": "user", "content": prompt},
                    ],
                    thread_id="title",
                )
                return self._clean_generated_title(result.get("message", ""))
            async with httpx.AsyncClient(timeout=20) as client:
                response = await client.post(
                    f"{self.base_url}/api/chat",
                    json={
                        "model": self.model,
                        "messages": [
                            {"role": "system", "content": "You write short chat titles."},
                            {"role": "user", "content": prompt},
                        ],
                        "options": self._ollama_options(),
                        "stream": False,
                    },
                )
            response.raise_for_status()
            message = response.json().get("message") or {}
            return self._clean_generated_title(message.get("content", ""))
        except Exception:
            return ""

    async def generate_plan_draft(
        self,
        title: str = "",
        objective: str = "",
        kind: str = "buying",
        constraints: dict[str, Any] | None = None,
        items: list[dict[str, Any]] | None = None,
    ) -> dict[str, Any]:
        clean_title = str(title or "").strip()
        clean_objective = str(objective or "").strip()
        clean_kind = str(kind or "buying").strip().casefold() or "buying"
        clean_constraints = dict(constraints or {})
        clean_items = self._normalize_plan_items(items or [])
        seed = {
            "title": clean_title,
            "objective": clean_objective,
            "kind": clean_kind,
            "constraints": clean_constraints,
            "items": clean_items,
        }
        prompt = self._plan_draft_prompt(seed)
        fallback = self._heuristic_plan_draft(seed)
        try:
            payload = await self._generate_plain_text(prompt, system_prompt="You draft structured continual plan JSON for TraderAI.")
            return self._normalize_plan_draft(payload, seed, fallback)
        except Exception:
            return fallback

    @staticmethod
    def _thread_id(thread_id: str | None) -> str:
        return (thread_id or DEFAULT_THREAD_ID).strip() or DEFAULT_THREAD_ID

    @staticmethod
    def _clean_generated_title(title: str) -> str:
        text = re.sub(r"[\r\n]+", " ", title).strip().strip('"').strip("'")
        text = re.sub(r"^(title|chat title)\s*:\s*", "", text, flags=re.IGNORECASE).strip()
        text = text.rstrip(".!?;:-").strip()
        if not text:
            return ""
        words = text.split()
        if len(words) > 8:
            text = " ".join(words[:8])
        return text[:64]

    async def _generate_plain_text(self, prompt: str, system_prompt: str) -> str:
        if self._is_openai_compatible_provider():
            async with httpx.AsyncClient(timeout=30) as client:
                response = await client.post(
                    f"{self.base_url}/chat/completions",
                    headers=self._openai_headers(),
                    json={
                        "model": self.model,
                        "messages": [
                            {"role": "system", "content": system_prompt},
                            {"role": "user", "content": prompt},
                        ],
                        "stream": False,
                        **self._openai_request_options(stream=False),
                    },
                )
            response.raise_for_status()
            choice = (response.json().get("choices") or [{}])[0]
            message = choice.get("message") or {}
            return str(message.get("content") or "")
        if self.provider == "codex":
            result = await self._codex_app_server_turn(
                prompt,
                [
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": prompt},
                ],
                thread_id="plan-draft",
            )
            return str(result.get("message") or "")
        async with httpx.AsyncClient(timeout=30) as client:
            response = await client.post(
                f"{self.base_url}/api/chat",
                json={
                    "model": self.model,
                    "messages": [
                        {"role": "system", "content": system_prompt},
                        {"role": "user", "content": prompt},
                    ],
                    "options": self._ollama_options(),
                    "stream": False,
                },
            )
        response.raise_for_status()
        message = response.json().get("message") or {}
        return str(message.get("content") or "")

    @classmethod
    def _normalize_plan_draft(
        cls,
        raw_text: str,
        seed: dict[str, Any],
        fallback: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        base = dict(fallback or cls._heuristic_plan_draft(seed))
        payload = cls._parse_json_object(raw_text)
        if not isinstance(payload, dict):
            return base

        title = str(payload.get("title") or seed.get("title") or base.get("title") or "").strip()
        objective = str(payload.get("objective") or seed.get("objective") or base.get("objective") or "").strip()
        kind = str(payload.get("kind") or seed.get("kind") or base.get("kind") or "buying").strip().casefold() or "buying"
        cadence = cls._normalize_plan_cadence(payload.get("cadence")) or base.get("cadence")
        constraints = cls._normalize_plan_constraints(payload.get("constraints"), seed.get("constraints") or {}, base.get("constraints") or {})
        items = cls._normalize_plan_items(payload.get("items") or seed.get("items") or base.get("items") or [])

        if kind == "buying" and not items:
            items = list(base.get("items") or [])
        if not constraints.get("instructions"):
            constraints["instructions"] = (base.get("constraints") or {}).get("instructions") or cls._default_plan_instructions(kind)
        if not constraints.get("message_tone"):
            constraints["message_tone"] = (base.get("constraints") or {}).get("message_tone") or "friendly and direct"

        return {
            "title": title or base.get("title") or "Continual plan",
            "objective": objective or base.get("objective") or title or "Continue this plan",
            "kind": kind,
            "cadence": cadence,
            "constraints": constraints,
            "items": items,
        }

    @classmethod
    def _heuristic_plan_draft(cls, seed: dict[str, Any]) -> dict[str, Any]:
        title = str(seed.get("title") or "").strip()
        objective = str(seed.get("objective") or "").strip()
        kind = str(seed.get("kind") or "buying").strip().casefold() or "buying"
        constraints = cls._normalize_plan_constraints(seed.get("constraints"), {}, {})
        items = cls._normalize_plan_items(seed.get("items") or [])

        if not items and kind == "buying":
            inferred_names = cls._infer_item_names(f"{title}\n{objective}")
            items = [{"item_name": name, "desired_quantity": 1, "max_unit_price": None} for name in inferred_names[:8]]

        if not constraints.get("message_tone"):
            constraints["message_tone"] = "friendly and direct"
        if not constraints.get("instructions"):
            constraints["instructions"] = cls._default_plan_instructions(kind)

        return {
            "title": title or "Continual plan",
            "objective": objective or title or "Continue this plan",
            "kind": kind,
            "cadence": cls._normalize_plan_cadence(seed.get("cadence")) or ("0 */6 * * *" if kind == "buying" else "0 */4 * * *"),
            "constraints": constraints,
            "items": items,
        }

    @staticmethod
    def _default_plan_instructions(kind: str) -> str:
        if kind == "custom":
            return "Check for meaningful updates, summarize what changed, and suggest the next move."
        return "Track the best active listings, avoid bad prices, and draft messages for approval when a strong candidate appears."

    @staticmethod
    def _plan_draft_prompt(seed: dict[str, Any]) -> str:
        return (
            "Draft a continual TraderAI plan as strict JSON.\n"
            "Return one JSON object only with keys: title, objective, kind, cadence, constraints, items.\n"
            "constraints may include message_tone, instructions, preferred_locations, excluded_sellers, max_unit_price.\n"
            "items must be an array of objects with item_name, desired_quantity, max_unit_price.\n"
            "If the request is vague, still fill cadence, message_tone, and instructions.\n"
            "Only include checklist items when they can be reasonably inferred from the request or existing draft.\n"
            "Do not wrap the JSON in markdown.\n\n"
            f"Current draft seed: {json.dumps(seed, ensure_ascii=True)}"
        )

    @staticmethod
    def _parse_json_object(raw_text: str) -> dict[str, Any] | None:
        text = str(raw_text or "").strip()
        if not text:
            return None
        try:
            parsed = json.loads(text)
            return parsed if isinstance(parsed, dict) else None
        except ValueError:
            pass
        start = text.find("{")
        end = text.rfind("}")
        if start == -1 or end <= start:
            return None
        try:
            parsed = json.loads(text[start : end + 1])
            return parsed if isinstance(parsed, dict) else None
        except ValueError:
            return None

    @classmethod
    def _normalize_plan_constraints(cls, value: Any, seed: dict[str, Any], fallback: dict[str, Any]) -> dict[str, Any]:
        merged: dict[str, Any] = {}
        for source in (fallback, seed, value if isinstance(value, dict) else {}):
            if not isinstance(source, dict):
                continue
            for key, item in source.items():
                if item in (None, "", [], {}):
                    continue
                if key in {"preferred_locations", "excluded_sellers"}:
                    if isinstance(item, list):
                        merged[key] = [str(entry).strip() for entry in item if str(entry).strip()]
                elif key == "max_unit_price":
                    try:
                        merged[key] = float(item)
                    except (TypeError, ValueError):
                        continue
                else:
                    merged[key] = str(item).strip()
        return merged

    @staticmethod
    def _normalize_plan_cadence(value: Any) -> str | None:
        text = str(value or "").strip()
        if not text:
            return None
        parts = text.split()
        return text if len(parts) == 5 else None

    @classmethod
    def _normalize_plan_items(cls, items: Any) -> list[dict[str, Any]]:
        if not isinstance(items, list):
            return []
        normalized: list[dict[str, Any]] = []
        for item in items:
            if not isinstance(item, dict):
                continue
            name = str(item.get("item_name") or item.get("name") or "").strip()
            if not name:
                continue
            normalized_item: dict[str, Any] = {"item_name": name}
            try:
                normalized_item["desired_quantity"] = max(1, int(item.get("desired_quantity") or item.get("quantity") or 1))
            except (TypeError, ValueError):
                normalized_item["desired_quantity"] = 1
            try:
                if item.get("max_unit_price") not in (None, ""):
                    normalized_item["max_unit_price"] = float(item.get("max_unit_price"))
                elif item.get("max_price") not in (None, ""):
                    normalized_item["max_unit_price"] = float(item.get("max_price"))
                else:
                    normalized_item["max_unit_price"] = None
            except (TypeError, ValueError):
                normalized_item["max_unit_price"] = None
            normalized.append(normalized_item)
        return normalized

    @staticmethod
    def _infer_item_names(text: str) -> list[str]:
        source = str(text or "")
        quoted = [match.strip() for match in re.findall(r'"([^"\n]{2,80})"|\'([^\'\n]{2,80})\'', source)]
        names = [next((part for part in group if part), "") for group in quoted]
        if names:
            return [name for name in names if name]

        lines = []
        for raw_line in source.splitlines():
            line = raw_line.strip(" -*\t")
            if not line:
                continue
            if any(token in line for token in [",", ";", "/"]):
                parts = re.split(r"[,;/]+", line)
                lines.extend(part.strip() for part in parts if part.strip())
            else:
                lines.append(line)

        stopwords = {
            "need", "needs", "want", "wants", "find", "draft", "deal", "deals", "parts", "part", "items",
            "watch", "track", "check", "buy", "buying", "for", "the", "and", "with", "from", "best", "cheapest",
        }
        inferred = []
        for line in lines:
            clean = re.sub(r"\s+", " ", line).strip().strip(".")
            if len(clean) < 3:
                continue
            lowered = clean.casefold()
            if lowered in stopwords:
                continue
            if any(phrase in lowered for phrase in ["find and draft", "check for", "continue this plan"]):
                continue
            inferred.append(clean[:120])
        deduped = []
        seen = set()
        for item in inferred:
            key = item.casefold()
            if key in seen:
                continue
            seen.add(key)
            deduped.append(item)
        return deduped

    def _pending_payloads(self) -> list[dict[str, Any]]:
        return [
            {
                "id": action.id,
                "label": action.label,
                "method": action.method,
                "endpoint": action.endpoint,
                "payload": self.tools._display_payload(action.payload) if hasattr(self.tools, "_display_payload") else action.payload,
                "metadata": action.metadata or {},
            }
            for action in self.tools.pending_actions.values()
        ]

    def _ollama_options(self) -> dict[str, Any]:
        if not self.num_ctx:
            return {}
        return {"num_ctx": self.num_ctx}

    @staticmethod
    def _empty_response_fallback(tool_results: list[dict[str, Any]]) -> str:
        if not tool_results:
            return "I did not get a usable response from the model. Please try again, or narrow the request a bit."
        return OllamaAgent._tool_result_fallback(
            tool_results,
            "I completed the tool call, but the model did not write a final answer.",
        )

    @staticmethod
    def _tool_result_fallback(tool_results: list[dict[str, Any]], reason: str) -> str:
        last = tool_results[-1]
        text = json.dumps(last, indent=2, ensure_ascii=True)
        if len(text) > 1800:
            text = text[:1800] + "\n..."
        return (
            f"{reason} "
            "Here is the last tool result so you are not left staring at a blank response:\n\n"
            f"```json\n{text}\n```"
        )

    @staticmethod
    def _tool_status(name: str) -> str:
        if name.startswith("get_uex_"):
            return f"Fetching UEX {name.removeprefix('get_uex_')}"
        if name.startswith("draft_uex_"):
            return f"Drafting UEX {name.removeprefix('draft_uex_')} for approval"
        if name.startswith("delete_uex_"):
            return f"Drafting UEX {name.removeprefix('delete_uex_')} delete for approval"
        labels = {
            "search_uex_api_index": "Searching UEX API index",
            "summarize_uex_commodity_price_history": "Summarizing commodity price history",
            "summarize_uex_marketplace_price_history": "Summarizing marketplace price history",
            "summarize_uex_currency_index_history": "Summarizing currency index history",
            "list_scmdb_versions": "Checking SCMDB versions",
            "search_scmdb_missions": "Searching SCMDB missions",
            "get_scmdb_mission_rewards": "Fetching SCMDB mission rewards",
            "search_scwiki_pages": "Searching Star Citizen Wiki",
            "get_scwiki_page": "Reading Star Citizen Wiki page",
            "search_scwiki_vehicles": "Searching Star Citizen Wiki vehicles",
            "get_scwiki_vehicle": "Fetching Star Citizen Wiki vehicle",
            "search_wikelo_ship_projects": "Searching Wikelo ship projects",
            "get_wikelo_ship_project": "Fetching Wikelo ship requirements",
            "search_cornerstone_items": "Searching Cornerstone items",
            "get_cornerstone_item_locations": "Fetching Cornerstone item locations",
            "get_cornerstone_item_media": "Fetching Cornerstone item media",
            "uex_api_catalog": "Checking UEX API catalog",
            "uex_get": "Fetching UEX data",
            "uex_draft_post": "Drafting UEX write for approval",
            "uex_draft_delete": "Drafting UEX delete for approval",
            "search_marketplace_listings": "Searching UEX listings",
            "get_marketplace_listing": "Fetching listing details",
            "list_marketplace_negotiations": "Checking negotiations",
            "get_negotiation_messages": "Reading negotiation messages",
            "draft_negotiation_message": "Drafting message for approval",
            "draft_marketplace_listing": "Drafting listing for approval",
            "draft_marketplace_listing_with_cornerstone_image": "Drafting listing with Cornerstone image",
            "check_uex_notifications": "Checking UEX notifications",
        }
        return labels.get(name, f"Running {name}")

    @staticmethod
    def _stream_metrics(event: dict[str, Any]) -> dict[str, Any]:
        prompt_tokens = int(event.get("prompt_eval_count") or 0)
        prompt_duration = int(event.get("prompt_eval_duration") or 0)
        output_tokens = int(event.get("eval_count") or 0)
        output_duration = int(event.get("eval_duration") or 0)

        def rate(tokens: int, duration_ns: int) -> float | None:
            if not tokens or not duration_ns:
                return None
            return tokens / (duration_ns / 1_000_000_000)

        return {
            "reading_tokens": prompt_tokens,
            "reading_tokens_per_second": rate(prompt_tokens, prompt_duration),
            "writing_tokens": output_tokens,
            "writing_tokens_per_second": rate(output_tokens, output_duration),
        }

    @staticmethod
    def _cloud_usage_metrics(usage: dict[str, Any]) -> dict[str, Any]:
        prompt_tokens = int(usage.get("prompt_tokens") or 0)
        completion_tokens = int(usage.get("completion_tokens") or 0)
        cache_hit_tokens = int(usage.get("prompt_cache_hit_tokens") or 0)
        cache_miss_tokens = int(usage.get("prompt_cache_miss_tokens") or 0)
        metrics = {
            "reading_tokens": prompt_tokens,
            "writing_tokens": completion_tokens,
        }
        if cache_hit_tokens or cache_miss_tokens:
            metrics["cache_hit_tokens"] = cache_hit_tokens
            metrics["cache_miss_tokens"] = cache_miss_tokens
        return metrics

    @staticmethod
    def _profile_identity(profile: dict[str, Any]) -> str:
        user = profile.get("uex_user")
        if not isinstance(user, dict):
            configured = profile.get("configured_name")
            return f"You are speaking with {configured}." if configured else ""

        username = user.get("username") or user.get("user_username")
        name = user.get("name")
        fields = []
        if username and name and username != name:
            fields.append(f"You are speaking with UEX user {username} ({name}).")
        elif username or name:
            fields.append(f"You are speaking with UEX user {username or name}.")

        details = []
        for key, label in [
            ("timezone", "timezone"),
            ("language", "preferred language"),
            ("specializations", "specializations"),
            ("languages", "languages"),
            ("archetypes", "archetypes"),
        ]:
            value = user.get(key)
            if value:
                details.append(f"{label}: {value}")
        if details:
            fields.append("UEX profile details: " + "; ".join(details) + ".")
        return " ".join(fields)

    @staticmethod
    def _profile_for_prompt(profile: dict[str, Any]) -> dict[str, Any]:
        user = profile.get("uex_user")
        if not isinstance(user, dict):
            return profile

        useful_user_fields = [
            "id",
            "name",
            "username",
            "avatar",
            "bio",
            "website_url",
            "timezone",
            "language",
            "day_availability",
            "time_availability",
            "specializations",
            "languages",
            "archetypes",
            "is_datarunner",
            "is_staff",
            "is_away_game",
            "date_rsi_verified",
            "date_twitch_verified",
        ]
        prompt_profile = dict(profile)
        prompt_profile["uex_user"] = {
            key: user[key]
            for key in useful_user_fields
            if key in user and user[key] not in (None, "")
        }
        return prompt_profile

    @staticmethod
    def _extract_call(call: dict[str, Any]) -> tuple[str, dict[str, Any]]:
        function = call.get("function") or {}
        name = function.get("name") or call.get("name")
        arguments = function.get("arguments") or call.get("arguments") or {}
        if isinstance(arguments, str):
            arguments = json.loads(arguments or "{}")
        return name, arguments

    @staticmethod
    def _normalize_images(images: list[dict[str, Any]] | None) -> list[dict[str, Any]]:
        normalized: list[dict[str, Any]] = []
        for image in images or []:
            if not isinstance(image, dict):
                continue
            image_data = str(image.get("image_data") or "").strip()
            if not image_data:
                continue
            normalized.append(
                {
                    "name": str(image.get("name") or "").strip() or "pasted-image.png",
                    "content_type": str(image.get("content_type") or "image/png").strip() or "image/png",
                    "image_data": image_data,
                }
            )
        return normalized

    @staticmethod
    def _prompt_text(content: str, image_count: int) -> str:
        text = content.strip()
        if text:
            return text
        return "Please analyze the attached image." if image_count == 1 else "Please analyze the attached images."

    @staticmethod
    def _conversation_content(content: str, image_count: int) -> str:
        text = content.strip()
        if not image_count:
            return text
        note = f"[Attached {image_count} pasted image{'s' if image_count != 1 else ''}]"
        return f"{text}\n\n{note}" if text else note

    @staticmethod
    def _user_message(content: str, images: list[dict[str, Any]]) -> dict[str, Any]:
        message: dict[str, Any] = {"role": "user", "content": content}
        if images:
            message["images"] = [image["image_data"] for image in images]
            message["image_content_types"] = [image["content_type"] for image in images]
        return message


class OllamaUnavailable(RuntimeError):
    pass